GlcoinCore/lnd
1
0
Fork 0
mirror of https://github.com/lightningnetwork/lnd.git synced 2026-08-13 12:32:48 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Merge pull request #10065 from ellemouton/asyncGraphCacheLoad

Browse source 
graph/db: async graph cache population
This commit is contained in:
ziggieXXX 2026-03-24 18:29:44 +01:00 • committed by GitHub
commit e4133bcb50
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 1668 additions and 754 deletions
Download patch file Download diff file Expand all files Collapse all files

3
config_builder.go
View file

@ -1056,6 +1056,9 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
chanGraphOpts := []graphdb.ChanGraphOption{
graphdb.WithUseGraphCache(!cfg.DB.NoGraphCache),
graphdb.WithAsyncGraphCachePopulation(
!cfg.DB.SyncGraphCacheLoad,
),
}
// We want to pre-allocate the channel graph cache according to what we

11
docs/release-notes/release-notes-0.21.0.md
View file

@ -169,6 +169,13 @@
## Performance Improvements
* Let the [channel graph cache be populated
asynchronously](https://github.com/lightningnetwork/lnd/pull/10065) on
startup. While the cache is being populated, the graph is still available for
queries, but all read queries will be served from the database until the cache
is fully populated. This new behaviour can be opted out of via the new
`--db.sync-graph-cache-load` option.
* [Replace the catch-all `FilterInvoices` SQL query with five focused,
index-friendly queries](https://github.com/lightningnetwork/lnd/pull/10601)
(`FetchPendingInvoices`, `FilterInvoicesBySettleIndex`,
@ -179,11 +186,11 @@
only the parameters it actually needs and uses a direct `ORDER BY`, allowing
the planner to perform efficient index range scans on the invoice table.
* [Fix full table scans on the HTLC settlement
* [Fix full table scans on the HTLC settlement
hot path](https://github.com/lightningnetwork/lnd/pull/10619).
Replace the catch-all `GetInvoice` query (which used `OR $1 IS NULL`
predicates that forced full table scans) with three dedicated queries
targeting uniquely-constrained columns. Also drop four redundant indexes
targeting uniquely-constrained columns. Also drop four redundant indexes
that duplicated UNIQUE constraints or were never used as query filters.
## Deprecations

240
graph/db/graph.go
View file

@ -15,6 +15,7 @@ import (
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/batch"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/graph/db/models"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/routing/route"
@ -25,18 +26,41 @@ import (
// busy shutting down.
var ErrChanGraphShuttingDown = fmt.Errorf("ChannelGraph shutting down")
// GraphCacheStatus describes the current state of the in-memory graph cache.
type GraphCacheStatus uint8
const (
// GraphCacheStatusDisabled indicates that the graph cache is disabled.
GraphCacheStatusDisabled GraphCacheStatus = iota
// GraphCacheStatusLoading indicates that the graph cache is still
// being populated from the DB and is not yet serving reads.
GraphCacheStatusLoading
// GraphCacheStatusLoaded indicates that the graph cache has
// completed its initial population and is serving reads.
GraphCacheStatusLoaded
// GraphCacheStatusFailed indicates that the initial population of
// the graph cache failed. Reads fall back to the database.
GraphCacheStatusFailed
)
// ChannelGraph is a layer above the graph's CRUD layer.
type ChannelGraph struct {
started atomic.Bool
stopped atomic.Bool
graphCache *GraphCache
opts *chanGraphOptions
cache *graphCacheState
db Store
*topologyManager
quit chan struct{}
wg sync.WaitGroup
quit chan struct{}
wg sync.WaitGroup
cancel fn.Option[context.CancelFunc]
}
// NewChannelGraph creates a new ChannelGraph instance with the given backend.
@ -49,6 +73,7 @@ func NewChannelGraph(v1Store Store,
}
g := &ChannelGraph{
opts: opts,
db: v1Store,
topologyManager: newTopologyManager(),
quit: make(chan struct{}),
@ -57,12 +82,29 @@ func NewChannelGraph(v1Store Store,
// The graph cache can be turned off (e.g. for mobile users) for a
// speed/memory usage tradeoff.
if opts.useGraphCache {
g.graphCache = NewGraphCache(opts.preAllocCacheNumNodes)
g.cache = newGraphCacheState(opts.preAllocCacheNumNodes)
}
return g, nil
}
// GraphCacheStatus returns the current state of the in-memory graph cache.
func (c *ChannelGraph) GraphCacheStatus() GraphCacheStatus {
switch {
case c.cache == nil:
return GraphCacheStatusDisabled
case c.cache.isLoaded():
return GraphCacheStatusLoaded
case c.cache.isFailed():
return GraphCacheStatusFailed
default:
return GraphCacheStatusLoading
}
}
// Start kicks off any goroutines required for the ChannelGraph to function.
// If the graph cache is enabled, then it will be populated with the contents of
// the database.
@ -73,9 +115,20 @@ func (c *ChannelGraph) Start() error {
log.Debugf("ChannelGraph starting")
defer log.Debug("ChannelGraph started")
ctx := context.TODO()
ctx, cancel := context.WithCancel(context.Background())
c.cancel = fn.Some(cancel)
if c.graphCache != nil {
if c.opts.asyncGraphCachePopulation {
c.wg.Add(1)
go func() {
defer c.wg.Done()
if err := c.populateCache(ctx); err != nil {
log.Criticalf("Could not populate the "+
"graph cache: %v", err)
}
}()
} else {
if err := c.populateCache(ctx); err != nil {
return fmt.Errorf("could not populate the graph "+
"cache: %w", err)
@ -97,6 +150,7 @@ func (c *ChannelGraph) Stop() error {
log.Debugf("ChannelGraph shutting down...")
defer log.Debug("ChannelGraph shutdown complete")
c.cancel.WhenSome(func(fn context.CancelFunc) { fn() })
close(c.quit)
c.wg.Wait()
@ -161,9 +215,22 @@ func (c *ChannelGraph) handleTopologySubscriptions(ctx context.Context) {
}
// populateCache loads the entire channel graph into the in-memory graph cache.
//
// NOTE: This should only be called if the graphCache has been constructed.
func (c *ChannelGraph) populateCache(ctx context.Context) error {
if c.cache == nil {
log.Info("In-memory channel graph cache disabled")
return nil
}
c.cache.beginPopulation()
loaded := false
defer func() {
c.cache.finishPopulation(loaded)
}()
cache := c.cache.graphCache
startTime := time.Now()
log.Info("Populating in-memory channel graph, this might take a " +
"while...")
@ -177,7 +244,7 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
func(node route.Vertex,
features *lnwire.FeatureVector) error {
c.graphCache.AddNodeFeatures(node, features)
cache.AddNodeFeatures(node, features)
return nil
}, func() {},
@ -194,7 +261,7 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
policy1,
policy2 *models.CachedEdgePolicy) error {
c.graphCache.AddChannel(info, policy1, policy2)
cache.AddChannel(info, policy1, policy2)
return nil
}, func() {},
@ -206,8 +273,10 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
}
}
loaded = true
log.Infof("Finished populating in-memory channel graph (took %v, %s)",
time.Since(startTime), c.graphCache.Stats())
time.Since(startTime), cache.Stats())
return nil
}
@ -226,8 +295,8 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(ctx context.Context,
node route.Vertex, cb func(channel *DirectedChannel) error,
reset func()) error {
if c.graphCache != nil {
return c.graphCache.ForEachChannel(node, cb)
if c.cache != nil && c.cache.isLoaded() {
return c.cache.graphCache.ForEachChannel(node, cb)
}
// TODO(elle): once the no-cache path needs to support
@ -247,8 +316,8 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(ctx context.Context,
func (c *ChannelGraph) FetchNodeFeatures(ctx context.Context,
node route.Vertex) (*lnwire.FeatureVector, error) {
if c.graphCache != nil {
return c.graphCache.GetFeatures(node), nil
if c.cache != nil && c.cache.isLoaded() {
return c.cache.graphCache.GetFeatures(node), nil
}
return c.db.FetchNodeFeatures(ctx, lnwire.GossipVersion1, node)
@ -261,7 +330,7 @@ func (c *ChannelGraph) FetchNodeFeatures(ctx context.Context,
func (c *ChannelGraph) GraphSession(ctx context.Context,
cb func(graph NodeTraverser) error, reset func()) error {
if c.graphCache != nil {
if c.cache != nil && c.cache.isLoaded() {
return cb(c)
}
@ -277,8 +346,8 @@ func (c *ChannelGraph) ForEachNodeCached(ctx context.Context,
cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
chans map[uint64]*DirectedChannel) error, reset func()) error {
if !withAddrs && c.graphCache != nil {
return c.graphCache.ForEachNode(
if !withAddrs && c.cache != nil && c.cache.isLoaded() {
return c.cache.graphCache.ForEachNode(
func(node route.Vertex,
channels map[uint64]*DirectedChannel) error {
@ -304,10 +373,12 @@ func (c *ChannelGraph) AddNode(ctx context.Context,
return err
}
if c.graphCache != nil {
c.graphCache.AddNodeFeatures(
node.PubKeyBytes, node.Features,
)
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
cache.AddNodeFeatures(
node.PubKeyBytes, node.Features,
)
})
}
select {
@ -333,8 +404,10 @@ func (c *ChannelGraph) AddChannelEdge(ctx context.Context,
return err
}
if c.graphCache != nil {
c.graphCache.AddChannel(models.NewCachedEdge(edge), nil, nil)
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
cache.AddChannel(models.NewCachedEdge(edge), nil, nil)
})
}
select {
@ -357,7 +430,7 @@ func (c *ChannelGraph) MarkEdgeLive(ctx context.Context,
return err
}
if c.graphCache != nil {
if c.cache != nil {
// We need to add the channel back into our graph cache,
// otherwise we won't use it for path finding.
infos, err := c.db.FetchChanInfos(ctx, v, []uint64{chanID})
@ -379,9 +452,12 @@ func (c *ChannelGraph) MarkEdgeLive(ctx context.Context,
policy2 = models.NewCachedPolicy(info.Policy2)
}
c.graphCache.AddChannel(
models.NewCachedEdge(info.Info), policy1, policy2,
)
c.cache.applyUpdate(func(cache *GraphCache) {
cache.AddChannel(
models.NewCachedEdge(info.Info),
policy1, policy2,
)
})
}
return nil
@ -406,13 +482,15 @@ func (c *ChannelGraph) DeleteChannelEdges(ctx context.Context,
return err
}
if c.graphCache != nil {
for _, info := range infos {
c.graphCache.RemoveChannel(
info.NodeKey1Bytes, info.NodeKey2Bytes,
info.ChannelID,
)
}
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
for _, info := range infos {
cache.RemoveChannel(
info.NodeKey1Bytes, info.NodeKey2Bytes,
info.ChannelID,
)
}
})
}
return err
@ -433,13 +511,15 @@ func (c *ChannelGraph) DisconnectBlockAtHeight(ctx context.Context,
return nil, err
}
if c.graphCache != nil {
for _, edge := range edges {
c.graphCache.RemoveChannel(
edge.NodeKey1Bytes, edge.NodeKey2Bytes,
edge.ChannelID,
)
}
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
for _, edge := range edges {
cache.RemoveChannel(
edge.NodeKey1Bytes, edge.NodeKey2Bytes,
edge.ChannelID,
)
}
})
}
return edges, nil
@ -464,20 +544,22 @@ func (c *ChannelGraph) PruneGraph(ctx context.Context,
return nil, err
}
if c.graphCache != nil {
for _, edge := range edges {
c.graphCache.RemoveChannel(
edge.NodeKey1Bytes, edge.NodeKey2Bytes,
edge.ChannelID,
)
}
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
for _, edge := range edges {
cache.RemoveChannel(
edge.NodeKey1Bytes, edge.NodeKey2Bytes,
edge.ChannelID,
)
}
for _, node := range nodes {
cache.RemoveNode(node)
}
})
for _, node := range nodes {
c.graphCache.RemoveNode(node)
if stats, ok := c.cache.stats(); ok {
log.Debugf("Pruned graph, cache now has %s", stats)
}
log.Debugf("Pruned graph, cache now has %s",
c.graphCache.Stats())
}
if len(edges) != 0 {
@ -507,10 +589,12 @@ func (c *ChannelGraph) PruneGraphNodes(ctx context.Context) error {
return err
}
if c.graphCache != nil {
for _, node := range nodes {
c.graphCache.RemoveNode(node)
}
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
for _, node := range nodes {
cache.RemoveNode(node)
}
})
}
return nil
@ -578,8 +662,10 @@ func (c *ChannelGraph) MarkEdgeZombie(ctx context.Context,
return err
}
if c.graphCache != nil {
c.graphCache.RemoveChannel(pubKey1, pubKey2, chanID)
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
cache.RemoveChannel(pubKey1, pubKey2, chanID)
})
}
return nil
@ -600,10 +686,12 @@ func (c *ChannelGraph) UpdateEdgePolicy(ctx context.Context,
return err
}
if c.graphCache != nil {
c.graphCache.UpdatePolicy(
models.NewCachedPolicy(edge), from, to,
)
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
cache.UpdatePolicy(
models.NewCachedPolicy(edge), from, to,
)
})
}
select {
@ -809,8 +897,8 @@ func NewVersionedGraph(c *ChannelGraph,
func (c *VersionedGraph) FetchNodeFeatures(ctx context.Context,
node route.Vertex) (*lnwire.FeatureVector, error) {
if c.graphCache != nil {
return c.graphCache.GetFeatures(node), nil
if c.cache != nil && c.cache.isLoaded() {
return c.cache.graphCache.GetFeatures(node), nil
}
return c.db.FetchNodeFeatures(ctx, c.v, node)
@ -826,8 +914,8 @@ func (c *VersionedGraph) ForEachNodeDirectedChannel(ctx context.Context,
node route.Vertex, cb func(channel *DirectedChannel) error,
reset func()) error {
if c.graphCache != nil {
return c.graphCache.ForEachChannel(node, cb)
if c.cache != nil && c.cache.isLoaded() {
return c.cache.graphCache.ForEachChannel(node, cb)
}
return c.db.ForEachNodeDirectedChannel(ctx, c.v, node, cb, reset)
@ -880,7 +968,7 @@ func (c *VersionedGraph) ChannelView(ctx context.Context) ([]EdgePoint,
func (c *VersionedGraph) GraphSession(ctx context.Context,
cb func(graph NodeTraverser) error, reset func()) error {
if c.graphCache != nil {
if c.cache != nil && c.cache.isLoaded() {
return cb(c)
}
@ -940,8 +1028,10 @@ func (c *VersionedGraph) DeleteNode(ctx context.Context,
return err
}
if c.graphCache != nil {
c.graphCache.RemoveNode(nodePub)
if c.cache != nil {
c.cache.applyUpdate(func(cache *GraphCache) {
cache.RemoveNode(nodePub)
})
}
return nil
@ -1085,7 +1175,13 @@ func MakeTestGraph(t testing.TB,
store := NewTestDB(t)
graph, err := NewChannelGraph(store, opts...)
// Default to synchronous cache population in tests so that the
// cache is fully loaded before the test proceeds.
allOpts := append(
[]ChanGraphOption{WithSyncGraphCachePopulation()}, opts...,
)
graph, err := NewChannelGraph(store, allOpts...)
require.NoError(t, err)
require.NoError(t, graph.Start())

105
graph/db/graph_cache_state.go Normal file
View file

@ -0,0 +1,105 @@
package graphdb
import (
"sync"
"sync/atomic"
)
// pendingUpdatesWarnThreshold is the number of buffered cache mutations at
// which a warning is logged. A large buffer indicates that cache population is
// taking a long time relative to the incoming gossip rate.
const pendingUpdatesWarnThreshold = 10_000
// graphCacheState tracks the in-memory graph cache together with its
// population state. The underlying GraphCache is independently thread-safe, so
// once reads are allowed to use it, they do not need to hold updateMtx.
type graphCacheState struct {
graphCache *GraphCache
loaded atomic.Bool
failed atomic.Bool
updateMtx sync.Mutex
loading bool
pendingUpdates []func(*GraphCache)
}
// newGraphCacheState constructs a graph cache state with a new cache instance.
func newGraphCacheState(preAllocNumNodes int) *graphCacheState {
return &graphCacheState{
graphCache: NewGraphCache(preAllocNumNodes),
}
}
// isLoaded reports whether the cache has finished its initial population and
// is safe to serve reads from.
func (s *graphCacheState) isLoaded() bool {
return s.loaded.Load()
}
// isFailed reports whether the cache population attempt has failed.
func (s *graphCacheState) isFailed() bool {
return s.failed.Load()
}
// stats returns the cache stats if the cache has finished its initial
// population.
func (s *graphCacheState) stats() (string, bool) {
if !s.isLoaded() {
return "", false
}
return s.graphCache.Stats(), true
}
// beginPopulation marks the cache as loading and starts buffering concurrent
// cache mutations until the population pass completes.
func (s *graphCacheState) beginPopulation() {
s.updateMtx.Lock()
defer s.updateMtx.Unlock()
s.loading = true
s.pendingUpdates = nil
}
// finishPopulation replays any buffered mutations and marks the cache as ready
// when the initial population completed successfully. If population failed,
// buffered mutations are discarded since the cache won't be used for reads.
func (s *graphCacheState) finishPopulation(loaded bool) {
s.updateMtx.Lock()
defer s.updateMtx.Unlock()
if loaded {
for _, update := range s.pendingUpdates {
update(s.graphCache)
}
s.loaded.Store(true)
} else {
s.failed.Store(true)
}
s.pendingUpdates = nil
s.loading = false
}
// applyUpdate applies a cache mutation immediately or buffers it when the
// cache is still being populated.
func (s *graphCacheState) applyUpdate(update func(cache *GraphCache)) {
s.updateMtx.Lock()
defer s.updateMtx.Unlock()
if s.loading {
s.pendingUpdates = append(s.pendingUpdates, update)
if len(s.pendingUpdates)%pendingUpdatesWarnThreshold == 0 {
log.Warnf("Graph cache has %d pending updates "+
"buffered during population",
len(s.pendingUpdates))
}
return
}
update(s.graphCache)
}

627
graph/db/graph_test.go
View file

@ -24,6 +24,7 @@ import (
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/graph/db/models"
"github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lntest/wait"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/routing/route"
"github.com/stretchr/testify/require"
@ -440,7 +441,9 @@ func testPartialNode(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
graph := NewVersionedGraph(
MakeTestGraph(t, WithSyncGraphCachePopulation()), v,
)
// To insert a partial node, we need to add a channel edge that has
// node keys for nodes we are not yet aware of.
@ -612,7 +615,9 @@ func testEdgeInsertionDeletion(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
graph := NewVersionedGraph(
MakeTestGraph(t, WithSyncGraphCachePopulation()), v,
)
// We'd like to test the insertion/deletion of edges, so we create two
// vertexes to connect.
@ -849,7 +854,7 @@ func TestDisconnectBlockAtHeight(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
graph := MakeTestGraph(t, WithSyncGraphCachePopulation())
sourceNode := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.SetSourceNode(ctx, sourceNode))
@ -1148,7 +1153,9 @@ func testEdgeInfoUpdates(t *testing.T, v lnwire.GossipVersion) {
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
graph := NewVersionedGraph(
MakeTestGraph(t, WithSyncGraphCachePopulation()), v,
)
// We'd like to test the update of edges inserted into the database, so
// we create two vertexes to connect.
@ -1166,7 +1173,7 @@ func testEdgeInfoUpdates(t *testing.T, v lnwire.GossipVersion) {
// is added, will fail.
err := graph.UpdateEdgePolicy(ctx, edge1)
require.ErrorIs(t, err, ErrEdgeNotFound)
require.Len(t, graph.graphCache.nodeChannels, 0)
require.Len(t, graph.cache.graphCache.nodeChannels, 0)
// Add the edge info.
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
@ -1329,8 +1336,9 @@ func assertNodeInCache(t *testing.T, g *ChannelGraph, n *models.Node,
expectedFeatures *lnwire.FeatureVector) {
// Let's check the internal view first.
nodeFeatures := g.cache.graphCache.nodeFeatures
require.Equal(
t, expectedFeatures, g.graphCache.nodeFeatures[n.PubKeyBytes],
t, expectedFeatures, nodeFeatures[n.PubKeyBytes],
)
// The external view should reflect this as well. Except when we expect
@ -1339,19 +1347,19 @@ func assertNodeInCache(t *testing.T, g *ChannelGraph, n *models.Node,
if expectedFeatures == nil {
expectedFeatures = lnwire.EmptyFeatureVector()
}
features := g.graphCache.GetFeatures(n.PubKeyBytes)
features := g.cache.graphCache.GetFeatures(n.PubKeyBytes)
require.Equal(t, expectedFeatures, features)
}
func assertNodeNotInCache(t *testing.T, g *ChannelGraph, n route.Vertex) {
_, ok := g.graphCache.nodeFeatures[n]
_, ok := g.cache.graphCache.nodeFeatures[n]
require.False(t, ok)
_, ok = g.graphCache.nodeChannels[n]
_, ok = g.cache.graphCache.nodeChannels[n]
require.False(t, ok)
// We should get the default features for this node.
features := g.graphCache.GetFeatures(n)
features := g.cache.graphCache.GetFeatures(n)
require.Equal(t, lnwire.EmptyFeatureVector(), features)
}
@ -1359,8 +1367,8 @@ func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
e *models.ChannelEdgeInfo) {
// Let's check the internal view first.
require.NotEmpty(t, g.graphCache.nodeChannels[e.NodeKey1Bytes])
require.NotEmpty(t, g.graphCache.nodeChannels[e.NodeKey2Bytes])
require.NotEmpty(t, g.cache.graphCache.nodeChannels[e.NodeKey1Bytes])
require.NotEmpty(t, g.cache.graphCache.nodeChannels[e.NodeKey2Bytes])
expectedNode1Channel := &DirectedChannel{
ChannelID: e.ChannelID,
@ -1370,12 +1378,13 @@ func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
OutPolicySet: false,
InPolicy: nil,
}
nodeChannels := g.cache.graphCache.nodeChannels
require.Contains(
t, g.graphCache.nodeChannels[e.NodeKey1Bytes], e.ChannelID,
t, nodeChannels[e.NodeKey1Bytes], e.ChannelID,
)
require.Equal(
t, expectedNode1Channel,
g.graphCache.nodeChannels[e.NodeKey1Bytes][e.ChannelID],
nodeChannels[e.NodeKey1Bytes][e.ChannelID],
)
expectedNode2Channel := &DirectedChannel{
@ -1387,16 +1396,16 @@ func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
InPolicy: nil,
}
require.Contains(
t, g.graphCache.nodeChannels[e.NodeKey2Bytes], e.ChannelID,
t, nodeChannels[e.NodeKey2Bytes], e.ChannelID,
)
require.Equal(
t, expectedNode2Channel,
g.graphCache.nodeChannels[e.NodeKey2Bytes][e.ChannelID],
nodeChannels[e.NodeKey2Bytes][e.ChannelID],
)
// The external view should reflect this as well.
var foundChannel *DirectedChannel
err := g.graphCache.ForEachChannel(
err := g.cache.graphCache.ForEachChannel(
e.NodeKey1Bytes, func(c *DirectedChannel) error {
if c.ChannelID == e.ChannelID {
foundChannel = c
@ -1409,7 +1418,7 @@ func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
require.NotNil(t, foundChannel)
require.Equal(t, expectedNode1Channel, foundChannel)
err = g.graphCache.ForEachChannel(
err = g.cache.graphCache.ForEachChannel(
e.NodeKey2Bytes, func(c *DirectedChannel) error {
if c.ChannelID == e.ChannelID {
foundChannel = c
@ -1426,7 +1435,7 @@ func assertEdgeWithNoPoliciesInCache(t *testing.T, g *ChannelGraph,
func assertNoEdge(t *testing.T, g *ChannelGraph, chanID uint64) {
// Make sure no channel in the cache has the given channel ID. If there
// are no channels at all, that is fine as well.
for _, channels := range g.graphCache.nodeChannels {
for _, channels := range g.cache.graphCache.nodeChannels {
for _, channel := range channels {
require.NotEqual(t, channel.ChannelID, chanID)
}
@ -1437,7 +1446,7 @@ func assertEdgeWithPolicyInCache(t *testing.T, g *ChannelGraph,
e *models.ChannelEdgeInfo, p *models.ChannelEdgePolicy, policy1 bool) {
// Check the internal state first.
c1, ok := g.graphCache.nodeChannels[e.NodeKey1Bytes][e.ChannelID]
c1, ok := g.cache.graphCache.nodeChannels[e.NodeKey1Bytes][e.ChannelID]
require.True(t, ok)
if policy1 {
@ -1450,7 +1459,7 @@ func assertEdgeWithPolicyInCache(t *testing.T, g *ChannelGraph,
)
}
c2, ok := g.graphCache.nodeChannels[e.NodeKey2Bytes][e.ChannelID]
c2, ok := g.cache.graphCache.nodeChannels[e.NodeKey2Bytes][e.ChannelID]
require.True(t, ok)
if policy1 {
@ -1468,14 +1477,14 @@ func assertEdgeWithPolicyInCache(t *testing.T, g *ChannelGraph,
c1Ext *DirectedChannel
c2Ext *DirectedChannel
)
require.NoError(t, g.graphCache.ForEachChannel(
require.NoError(t, g.cache.graphCache.ForEachChannel(
e.NodeKey1Bytes, func(c *DirectedChannel) error {
c1Ext = c
return nil
},
))
require.NoError(t, g.graphCache.ForEachChannel(
require.NoError(t, g.cache.graphCache.ForEachChannel(
e.NodeKey2Bytes, func(c *DirectedChannel) error {
c2Ext = c
@ -1739,11 +1748,16 @@ func testForEachSourceNodeChannel(t *testing.T, v lnwire.GossipVersion) {
require.Empty(t, expectedSrcChans)
}
// TestGraphTraversal tests that we can traverse the graph and find all
// nodes and channels that we expect to find.
func TestGraphTraversal(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
// If we turn the channel graph cache _off_, then iterate through the
// set of channels (to force the fall back), we should find all the
// channel as well as the nodes included.
graph := MakeTestGraph(t, WithUseGraphCache(false))
// We'd like to test some of the graph traversal capabilities within
// the DB, so we'll create a series of fake nodes to insert into the
@ -1760,10 +1774,6 @@ func TestGraphTraversal(t *testing.T) {
nodeIndex[node.PubKeyBytes] = struct{}{}
}
// If we turn the channel graph cache _off_, then iterate through the
// set of channels (to force the fall back), we should find all the
// channel as well as the nodes included.
graph.graphCache = nil
err := graph.ForEachNodeCached(ctx, lnwire.GossipVersion1, false,
func(_ context.Context, node route.Vertex, _ []net.Addr,
chans map[uint64]*DirectedChannel) error {
@ -1914,10 +1924,14 @@ func testGraphTraversalCacheable(t *testing.T, v lnwire.GossipVersion) {
require.Len(t, chanIndex2, 0)
}
// TestGraphCacheTraversal tests traversal of the graph via the graph cache.
func TestGraphCacheTraversal(t *testing.T) {
t.Parallel()
ctx := t.Context()
graph := MakeTestGraph(t)
// Explicitly enable the graph cache so that the
// ForEachNodeDirectedChannel call below will use the cache.
graph := MakeTestGraph(t, WithUseGraphCache(true))
// We'd like to test some of the graph traversal capabilities within
// the DB, so we'll create a series of fake nodes to insert into the
@ -1935,8 +1949,8 @@ func TestGraphCacheTraversal(t *testing.T) {
for _, node := range nodeList {
node := node
err := graph.graphCache.ForEachChannel(
node.PubKeyBytes, func(d *DirectedChannel) error {
err := graph.ForEachNodeDirectedChannel(
ctx, node.PubKeyBytes, func(d *DirectedChannel) error {
delete(chanIndex, d.ChannelID)
if !d.OutPolicySet || d.InPolicy == nil {
@ -1957,6 +1971,8 @@ func TestGraphCacheTraversal(t *testing.T) {
numNodeChans++
return nil
}, func() {
numNodeChans = 0
},
)
require.NoError(t, err)
@ -5022,7 +5038,7 @@ func BenchmarkForEachChannel(b *testing.B) {
}
}
// TestGraphCacheForEachNodeChannel tests that the forEachNodeDirectedChannel
// TestForEachNodeDirectedChannel tests that the ForEachNodeDirectedChannel
// method works as expected, and is able to handle nil self edges.
func testGraphCacheForEachNodeChannel(t *testing.T,
v lnwire.GossipVersion) {
@ -5030,11 +5046,12 @@ func testGraphCacheForEachNodeChannel(t *testing.T,
t.Parallel()
ctx := t.Context()
graph := NewVersionedGraph(MakeTestGraph(t), v)
// Unset the channel graph cache to simulate the user running with the
// option turned off.
graph.graphCache = nil
// option turned off. This forces the V1Store ForEachNodeDirectedChannel
// to be queried instead of the graph cache's ForEachChannel method.
graph := NewVersionedGraph(
MakeTestGraph(t, WithUseGraphCache(false)), v,
)
node1 := createTestVertex(t, v)
require.NoError(t, graph.AddNode(ctx, node1))
@ -5132,7 +5149,9 @@ func TestGraphLoading(t *testing.T) {
// Next, create the graph for the first time.
graphStore := NewTestDB(t)
graph, err := NewChannelGraph(graphStore)
graph, err := NewChannelGraph(
graphStore, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
@ -5148,7 +5167,9 @@ func TestGraphLoading(t *testing.T) {
// Recreate the graph. This should cause the graph cache to be
// populated.
graphReloaded, err := NewChannelGraph(graphStore)
graphReloaded, err := NewChannelGraph(
graphStore, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, graphReloaded.Start())
t.Cleanup(func() {
@ -5157,16 +5178,538 @@ func TestGraphLoading(t *testing.T) {
// Assert that the cache content is identical.
require.Equal(
t, graph.graphCache.nodeChannels,
graphReloaded.graphCache.nodeChannels,
t, graph.cache.graphCache.nodeChannels,
graphReloaded.cache.graphCache.nodeChannels,
)
require.Equal(
t, graph.graphCache.nodeFeatures,
graphReloaded.graphCache.nodeFeatures,
t, graph.cache.graphCache.nodeFeatures,
graphReloaded.cache.graphCache.nodeFeatures,
)
}
// TestAsyncGraphCache tests the behaviour of the ChannelGraph when the graph
// cache is populated asynchronously.
func TestAsyncGraphCache(t *testing.T) {
t.Parallel()
ctx := t.Context()
const (
numNodes = 100
numChannels = 3
)
// Next, create the graph for the first time.
graphStore := NewTestDB(t)
// The first time we spin up the graph, we Start is as normal and fill
// it with test data. This will ensure that the graph cache has
// something to load on the next Start.
graph, err := NewChannelGraph(graphStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
channels, nodes := fillTestGraph(
t, graph, numNodes, numChannels, lnwire.GossipVersion1,
)
assertGraphState := func() {
var (
numNodes int
chanIndex = make(map[uint64]struct{}, numChannels)
)
// We query the graph for all nodes and channels, and
// assert that we get the expected number of nodes and
// channels.
err := graph.ForEachNodeCached(
ctx, lnwire.GossipVersion1, false,
func(_ context.Context, node route.Vertex,
_ []net.Addr,
chans map[uint64]*DirectedChannel) error {
numNodes++
for chanID := range chans {
chanIndex[chanID] = struct{}{}
}
return nil
}, func() {
numNodes = 0
chanIndex = make(
map[uint64]struct{}, numChannels,
)
},
)
require.NoError(t, err)
require.Equal(t, len(nodes), numNodes)
require.Equal(t, len(channels), len(chanIndex))
}
assertGraphState()
// Now we stop the graph.
require.NoError(t, graph.Stop())
// Recreate it but don't start it yet.
graph, err = NewChannelGraph(graphStore)
require.NoError(t, err)
// Spin off a goroutine that starts to make queries to the ChannelGraph.
// We start this before we start the graph, so that we can ensure that
// the queries are made while the graph cache is being populated.
var (
wg sync.WaitGroup
numRuns = 10
)
for i := 0; i < numRuns; i++ {
wg.Add(1)
go func() {
defer wg.Done()
assertGraphState()
}()
}
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
wg.Wait()
// Wait for the cache to be fully populated.
err = wait.Predicate(func() bool {
return graph.cache.isLoaded()
}, wait.DefaultTimeout)
require.NoError(t, err)
// And then assert that all the expected nodes and channels are
// present in the graph cache.
for _, node := range nodes {
_, ok := graph.cache.graphCache.nodeChannels[node.PubKeyBytes]
require.True(t, ok)
}
}
type blockingCacheLoadStore struct {
Store
cacheLoadStarted chan struct{}
allowCacheLoad chan struct{}
blockOnce sync.Once
}
// ForEachChannelCacheable pauses the first cacheable channel iteration until
// the test allows it to continue.
func (s *blockingCacheLoadStore) ForEachChannelCacheable(ctx context.Context,
v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo,
*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
reset func()) error {
return s.Store.ForEachChannelCacheable(
ctx, v, func(info *models.CachedEdgeInfo,
policy1,
policy2 *models.CachedEdgePolicy) error {
s.blockOnce.Do(func() {
close(s.cacheLoadStarted)
<-s.allowCacheLoad
})
return cb(info, policy1, policy2)
}, reset,
)
}
type shutdownBlockingCacheLoadStore struct {
Store
cacheLoadStarted chan struct{}
blockOnce sync.Once
}
// ForEachChannelCacheable blocks until the context is canceled so tests can
// assert that Stop interrupts async cache population.
func (s *shutdownBlockingCacheLoadStore) ForEachChannelCacheable(
ctx context.Context, v lnwire.GossipVersion,
cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
*models.CachedEdgePolicy) error, reset func()) error {
return s.Store.ForEachChannelCacheable(
ctx, v, func(info *models.CachedEdgeInfo,
policy1,
policy2 *models.CachedEdgePolicy) error {
s.blockOnce.Do(func() {
close(s.cacheLoadStarted)
<-ctx.Done()
})
return ctx.Err()
}, reset,
)
}
type failingCacheLoadStore struct {
Store
cacheLoadAttempted chan struct{}
populateErr error
}
// ForEachChannelCacheable fails the initial cache population after signaling
// that the async load reached channel iteration.
func (s *failingCacheLoadStore) ForEachChannelCacheable(ctx context.Context,
v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo,
*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
reset func()) error {
close(s.cacheLoadAttempted)
return s.populateErr
}
// TestAsyncGraphCacheReplaysConcurrentWrites asserts that graph mutations that
// happen while the async cache population is running are replayed onto the
// cache before it becomes readable.
func TestAsyncGraphCacheReplaysConcurrentWrites(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
blockingStore := &blockingCacheLoadStore{
Store: store,
cacheLoadStarted: make(chan struct{}),
allowCacheLoad: make(chan struct{}),
}
graph, err := NewChannelGraph(blockingStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
<-blockingStore.cacheLoadStarted
updatedEdge := *edge1
updatedEdge.LastUpdate = nextUpdateTime()
updatedEdge.FeeBaseMSat++
require.NoError(t, graph.UpdateEdgePolicy(ctx, &updatedEdge))
close(blockingStore.allowCacheLoad)
err = wait.Predicate(func() bool {
return graph.cache.isLoaded()
}, wait.DefaultTimeout)
require.NoError(t, err)
var cachedFee lnwire.MilliSatoshi
err = graph.ForEachNodeDirectedChannel(
ctx, updatedEdge.ToNode,
func(channel *DirectedChannel) error {
if channel.ChannelID != updatedEdge.ChannelID {
return nil
}
require.NotNil(t, channel.InPolicy)
cachedFee = channel.InPolicy.FeeBaseMSat
return nil
}, func() {},
)
require.NoError(t, err)
require.Equal(t, updatedEdge.FeeBaseMSat, cachedFee)
}
// TestAsyncGraphCacheStopCancelsLoad asserts that Stop interrupts async cache
// population instead of waiting for the full load to finish.
func TestAsyncGraphCacheStopCancelsLoad(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
blockingStore := &shutdownBlockingCacheLoadStore{
Store: store,
cacheLoadStarted: make(chan struct{}),
}
graph, err := NewChannelGraph(blockingStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
<-blockingStore.cacheLoadStarted
stopErr := make(chan error, 1)
go func() {
stopErr <- graph.Stop()
}()
select {
case err := <-stopErr:
require.NoError(t, err)
case <-time.After(wait.DefaultTimeout):
t.Fatal("Stop did not cancel graph cache loading")
}
}
// TestAsyncGraphCachePopulationFailureFallsBackToDB asserts that cache
// population errors leave the cache unreadable while reads continue to succeed
// through the DB-backed path.
func TestAsyncGraphCachePopulationFailureFallsBackToDB(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
populateErr := errors.New("cache population failed")
failingStore := &failingCacheLoadStore{
Store: store,
cacheLoadAttempted: make(chan struct{}),
populateErr: populateErr,
}
graph, err := NewChannelGraph(failingStore)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
<-failingStore.cacheLoadAttempted
err = wait.Predicate(func() bool {
return graph.GraphCacheStatus() == GraphCacheStatusFailed
}, wait.DefaultTimeout)
require.NoError(t, err)
require.False(t, graph.cache.isLoaded())
var numChannels int
err = graph.ForEachNodeDirectedChannel(
ctx, edge1.ToNode,
func(channel *DirectedChannel) error {
if channel.ChannelID != edge1.ChannelID {
return nil
}
numChannels++
require.NotNil(t, channel.InPolicy)
require.Equal(t, edge1.FeeBaseMSat,
channel.InPolicy.FeeBaseMSat)
return nil
}, func() {},
)
require.NoError(t, err)
require.Equal(t, 1, numChannels)
}
// TestGraphCacheStatus asserts that the graph cache reports disabled, loading,
// loaded and failed states as expected.
func TestGraphCacheStatus(t *testing.T) {
t.Parallel()
ctx := t.Context()
store := NewTestDB(t)
disabledGraph, err := NewChannelGraph(
store, WithUseGraphCache(false),
)
require.NoError(t, err)
require.Equal(
t, GraphCacheStatusDisabled, disabledGraph.GraphCacheStatus(),
)
require.NoError(t, disabledGraph.Start())
require.Equal(
t, GraphCacheStatusDisabled, disabledGraph.GraphCacheStatus(),
)
require.NoError(t, disabledGraph.Stop())
setupGraph, err := NewChannelGraph(
store, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, setupGraph.Start())
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, setupGraph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, setupGraph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, setupGraph.UpdateEdgePolicy(ctx, edge2))
require.NoError(t, setupGraph.Stop())
blockingStore := &blockingCacheLoadStore{
Store: store,
cacheLoadStarted: make(chan struct{}),
allowCacheLoad: make(chan struct{}),
}
graph, err := NewChannelGraph(blockingStore)
require.NoError(t, err)
require.Equal(t, GraphCacheStatusLoading, graph.GraphCacheStatus())
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
<-blockingStore.cacheLoadStarted
require.Equal(t, GraphCacheStatusLoading, graph.GraphCacheStatus())
close(blockingStore.allowCacheLoad)
err = wait.Predicate(func() bool {
return graph.GraphCacheStatus() == GraphCacheStatusLoaded
}, wait.DefaultTimeout)
require.NoError(t, err)
require.NoError(t, graph.Stop())
// Assert the failed state by using a store that errors during cache
// population.
populateErr := errors.New("cache population failed")
failingStore := &failingCacheLoadStore{
Store: store,
cacheLoadAttempted: make(chan struct{}),
populateErr: populateErr,
}
failedGraph, err := NewChannelGraph(failingStore)
require.NoError(t, err)
require.NoError(t, failedGraph.Start())
t.Cleanup(func() {
require.NoError(t, failedGraph.Stop())
})
<-failingStore.cacheLoadAttempted
err = wait.Predicate(func() bool {
return failedGraph.GraphCacheStatus() == GraphCacheStatusFailed
}, wait.DefaultTimeout)
require.NoError(t, err)
}
// TestKVCacheableIteratorsRespectCancellation asserts that KV-backed cache
// iterators return when their context is canceled.
func TestKVCacheableIteratorsRespectCancellation(t *testing.T) {
t.Parallel()
if isSQLDB {
t.Skip("KV iterator cancellation is specific to KVStore")
}
ctx := t.Context()
store := NewTestDB(t)
kvStore, ok := store.(*KVStore)
require.True(t, ok)
graph, err := NewChannelGraph(
kvStore, WithSyncGraphCachePopulation(),
)
require.NoError(t, err)
require.NoError(t, graph.Start())
t.Cleanup(func() {
require.NoError(t, graph.Stop())
})
node1 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node1))
node2 := createTestVertex(t, lnwire.GossipVersion1)
require.NoError(t, graph.AddNode(ctx, node2))
edgeInfo, edge1, edge2 := createChannelEdge(
node1, node2, lnwire.GossipVersion1,
)
require.NoError(t, graph.AddChannelEdge(ctx, edgeInfo))
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge1))
require.NoError(t, graph.UpdateEdgePolicy(ctx, edge2))
canceledCtx, cancel := context.WithCancel(ctx)
cancel()
err = kvStore.ForEachNodeCacheable(
canceledCtx, lnwire.GossipVersion1,
func(route.Vertex, *lnwire.FeatureVector) error {
return nil
}, func() {},
)
require.ErrorIs(t, err, context.Canceled)
err = kvStore.ForEachChannelCacheable(
canceledCtx, lnwire.GossipVersion1,
func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
*models.CachedEdgePolicy) error {
return nil
}, func() {},
)
require.ErrorIs(t, err, context.Canceled)
}
// TestClosedScid tests that we can correctly insert a SCID into the index of
// closed short channel ids.
func TestClosedScid(t *testing.T) {

20
graph/db/kv_store.go
View file

@ -250,7 +250,7 @@ func (c channelMapKey) String() string {
// getChannelMap loads all channel edge policies from the database and stores
// them in a map.
func getChannelMap(edges kvdb.RBucket) (
func getChannelMap(ctx context.Context, edges kvdb.RBucket) (
map[channelMapKey]*models.ChannelEdgePolicy, error) {
// Create a map to store all channel edge policies.
@ -440,7 +440,9 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
// First, load all edges in memory indexed by node and channel
// id.
channelMap, err := getChannelMap(edges)
channelMap, err := getChannelMap(
context.Background(), edges,
)
if err != nil {
return err
}
@ -493,7 +495,7 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
//
// NOTE: this method is like ForEachChannel but fetches only the data required
// for the graph cache.
func (c *KVStore) ForEachChannelCacheable(_ context.Context,
func (c *KVStore) ForEachChannelCacheable(ctx context.Context,
v lnwire.GossipVersion, cb func(*models.CachedEdgeInfo,
*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
reset func()) error {
@ -510,7 +512,7 @@ func (c *KVStore) ForEachChannelCacheable(_ context.Context,
// First, load all edges in memory indexed by node and channel
// id.
channelMap, err := getChannelMap(edges)
channelMap, err := getChannelMap(ctx, edges)
if err != nil {
return err
}
@ -524,6 +526,10 @@ func (c *KVStore) ForEachChannelCacheable(_ context.Context,
// loaded above and invoke the callback.
return kvdb.ForAll(
edgeIndex, func(k, edgeInfoBytes []byte) error {
if err := ctx.Err(); err != nil {
return err
}
var chanID [8]byte
copy(chanID[:], k)
@ -895,7 +901,7 @@ func forEachNode(db kvdb.Backend,
// graph, executing the passed callback with each node encountered. If the
// callback returns an error, then the transaction is aborted and the iteration
// stops early.
func (c *KVStore) ForEachNodeCacheable(_ context.Context,
func (c *KVStore) ForEachNodeCacheable(ctx context.Context,
v lnwire.GossipVersion, cb func(route.Vertex,
*lnwire.FeatureVector) error, reset func()) error {
@ -912,6 +918,10 @@ func (c *KVStore) ForEachNodeCacheable(_ context.Context,
}
return nodes.ForEach(func(pubKey, nodeBytes []byte) error {
if err := ctx.Err(); err != nil {
return err
}
// If this is the source key, then we skip this
// iteration as the value for this key is a pubKey
// rather than raw node information.

30
graph/db/options.go
View file

@ -85,14 +85,21 @@ type chanGraphOptions struct {
// preAllocCacheNumNodes is the number of nodes we expect to be in the
// graph cache, so we can pre-allocate the map accordingly.
preAllocCacheNumNodes int
// asyncGraphCachePopulation indicates whether the graph cache
// should be populated asynchronously or if the Start method should
// block until the cache is fully populated. This is true by
// default.
asyncGraphCachePopulation bool
}
// defaultChanGraphOptions returns a new chanGraphOptions instance populated
// with default values.
func defaultChanGraphOptions() *chanGraphOptions {
return &chanGraphOptions{
useGraphCache: true,
preAllocCacheNumNodes: DefaultPreAllocCacheNumNodes,
useGraphCache: true,
asyncGraphCachePopulation: true,
preAllocCacheNumNodes: DefaultPreAllocCacheNumNodes,
}
}
@ -115,6 +122,25 @@ func WithPreAllocCacheNumNodes(n int) ChanGraphOption {
}
}
// WithAsyncGraphCachePopulation sets whether the graph cache should be
// populated asynchronously or if the Start method should block until the
// cache is fully populated.
func WithAsyncGraphCachePopulation(async bool) ChanGraphOption {
return func(o *chanGraphOptions) {
o.asyncGraphCachePopulation = async
}
}
// WithSyncGraphCachePopulation will cause the ChannelGraph to block
// until the graph cache is fully populated before returning from the Start
// method. This is useful for tests that need to ensure the graph cache is
// fully populated before proceeding with further operations.
func WithSyncGraphCachePopulation() ChanGraphOption {
return func(o *chanGraphOptions) {
o.asyncGraphCachePopulation = false
}
}
// StoreOptions holds parameters for tuning and customizing a graph DB.
type StoreOptions struct {
// RejectCacheSize is the maximum number of rejectCacheEntries to hold

2
lncfg/db.go
View file

@ -89,6 +89,8 @@ type DB struct {
NoGraphCache bool `long:"no-graph-cache" description:"Don't use the in-memory graph cache for path finding. Much slower but uses less RAM. Can only be used with a bolt database backend."`
SyncGraphCacheLoad bool `long:"sync-graph-cache-load" description:"Force synchronous loading of the graph cache. This will block the startup until the graph cache is fully loaded into memory. This is useful if any bugs appear with the new async loading feature of the graph cache."`
PruneRevocation bool `long:"prune-revocation" description:"Run the optional migration that prunes the revocation logs to save disk space."`
NoRevLogAmtData bool `long:"no-rev-log-amt-data" description:"If set, the to-local and to-remote output amounts of revoked commitment transactions will not be stored in the revocation log. Note that once this data is lost, a watchtower client will not be able to back up the revoked state."`

1331
lnrpc/lightning.pb.go
View file

File diff suppressed because it is too large Load diff

10
lnrpc/lightning.proto
View file

@ -2058,6 +2058,13 @@ message PeerEvent {
message GetInfoRequest {
}
enum GraphCacheStatus {
GRAPH_CACHE_STATUS_DISABLED = 0;
GRAPH_CACHE_STATUS_LOADING = 1;
GRAPH_CACHE_STATUS_LOADED = 2;
GRAPH_CACHE_STATUS_FAILED = 3;
}
message GetInfoResponse {
// The version of the LND software that the node is running.
string version = 14;
@ -2136,6 +2143,9 @@ message GetInfoResponse {
// Whether the wallet is fully synced to the best chain. This indicates the
// wallet's internal sync state with the backing chain source.
bool wallet_synced = 23;
// The current status of the in-memory graph cache.
GraphCacheStatus graph_cache_status = 24;
}
message GetDebugInfoRequest {

14
lnrpc/lightning.swagger.json
View file

@ -5598,6 +5598,10 @@
"wallet_synced": {
"type": "boolean",
"description": "Whether the wallet is fully synced to the best chain. This indicates the\nwallet's internal sync state with the backing chain source."
},
"graph_cache_status": {
"$ref": "#/definitions/lnrpcGraphCacheStatus",
"description": "The current status of the in-memory graph cache."
}
}
},
@ -5619,6 +5623,16 @@
}
}
},
"lnrpcGraphCacheStatus": {
"type": "string",
"enum": [
"GRAPH_CACHE_STATUS_DISABLED",
"GRAPH_CACHE_STATUS_LOADING",
"GRAPH_CACHE_STATUS_LOADED",
"GRAPH_CACHE_STATUS_FAILED"
],
"default": "GRAPH_CACHE_STATUS_DISABLED"
},
"lnrpcGraphTopologyUpdate": {
"type": "object",
"properties": {

23
rpcserver.go
View file

@ -3431,6 +3431,8 @@ func (r *rpcServer) GetInfo(_ context.Context,
isTestNet := chainreg.IsTestnet(&r.cfg.ActiveNetParams)
nodeColor := graphdb.EncodeHexColor(nodeAnn.RGBColor)
version := build.Version() + " commit=" + build.Commit
cacheStatus := r.server.graphDB.GraphCacheStatus()
graphCacheStatus := rpcGraphCacheStatus(cacheStatus)
return &lnrpc.GetInfoResponse{
IdentityPubkey: encodedIDPub,
@ -3454,9 +3456,30 @@ func (r *rpcServer) GetInfo(_ context.Context,
RequireHtlcInterceptor: r.cfg.RequireInterceptor,
StoreFinalHtlcResolutions: r.cfg.StoreFinalHtlcResolutions,
WalletSynced: syncInfo.isWalletSynced,
GraphCacheStatus: graphCacheStatus,
}, nil
}
// rpcGraphCacheStatus maps the graph DB cache status to the lnrpc enum used by
// GetInfo.
func rpcGraphCacheStatus(
status graphdb.GraphCacheStatus) lnrpc.GraphCacheStatus {
switch status {
case graphdb.GraphCacheStatusDisabled:
return lnrpc.GraphCacheStatus_GRAPH_CACHE_STATUS_DISABLED
case graphdb.GraphCacheStatusLoaded:
return lnrpc.GraphCacheStatus_GRAPH_CACHE_STATUS_LOADED
case graphdb.GraphCacheStatusFailed:
return lnrpc.GraphCacheStatus_GRAPH_CACHE_STATUS_FAILED
default:
return lnrpc.GraphCacheStatus_GRAPH_CACHE_STATUS_LOADING
}
}
// GetDebugInfo returns debug information concerning the state of the daemon
// and its subsystems. By default, this returns only the configuration. If the
// `include_log` flag is set in the request, the latest log entries from the

6
sample-lnd.conf
View file

@ -1501,6 +1501,12 @@
; less RAM. Can only be used with a bolt database backend.
; db.no-graph-cache=false
; Block the start-up of LND until the graph cache has been fully populated.
; If not set, the graph cache will be populated asynchronously and any read
; calls made before the cache is fully populated will fall back to the
; database.
; db.sync-graph-cache-load=false
; Specify whether the optional migration for pruning old revocation logs
; should be applied. This migration will only save disk space if there are open
; channels prior to lnd@v0.15.0.