From 57d4ce0f9bee3e85ffd2111b76c8a1aceb0d76d0 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 09:50:42 +0200 Subject: [PATCH 01/11] graph/db: expose KVStore locations where a reset is needed This commit passes no-op reset callbacks to various places where they are needed in the KVStore. Upcoming commits will replace the no-ops by adding reset params to the methods that require it. --- graph/db/kv_store.go | 104 ++++++++++++++++++++++++++------------ graph/db/sql_migration.go | 6 +++ 2 files changed, 78 insertions(+), 32 deletions(-) diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 6ca9cdedb..92700376d 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -406,9 +406,11 @@ func (c *KVStore) AddrsForNode(ctx context.Context, // callback. func (c *KVStore) ForEachChannel(_ context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error) error { - return forEachChannel(c.db, cb) + reset := func() {} + + return forEachChannel(c.db, cb, reset) } // forEachChannel iterates through all the channel edges stored within the @@ -421,7 +423,8 @@ func (c *KVStore) ForEachChannel(_ context.Context, // for that particular channel edge routing policy will be passed into the // callback. func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, + reset func()) error { return db.View(func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) @@ -469,7 +472,7 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo, return cb(&info, policy1, policy2) }, ) - }, func() {}) + }, reset) } // ForEachChannelCacheable iterates through all the channel edges stored within @@ -487,6 +490,8 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo, func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, *models.CachedEdgePolicy) error) error { + reset := func() {} + return c.db.View(func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) if edges == nil { @@ -558,7 +563,7 @@ func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, ) }, ) - }, func() {}) + }, reset) } // forEachNodeDirectedChannel iterates through all channels of a given node, @@ -568,8 +573,13 @@ func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, // transaction may be provided. If none is provided, a new one will be created. // // Unknown policies are passed into the callback as nil values. +// +// NOTE: the reset param is only meaningful if the tx param is nil. If it is +// not nil, the caller is expected to have passed in a reset to the parent +// function's View/Update call which will then apply to the whole transaction. func (c *KVStore) forEachNodeDirectedChannel(tx kvdb.RTx, - node route.Vertex, cb func(channel *DirectedChannel) error) error { + node route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error { // Fallback that uses the database. toNodeCallback := func() route.Vertex { @@ -612,7 +622,7 @@ func (c *KVStore) forEachNodeDirectedChannel(tx kvdb.RTx, return cb(directedChannel) } - return nodeTraversal(tx, node[:], c.db, dbCallback) + return nodeTraversal(tx, node[:], c.db, dbCallback, reset) } // fetchNodeFeatures returns the features of a given node. If no features are @@ -650,7 +660,9 @@ func (c *KVStore) fetchNodeFeatures(tx kvdb.RTx, func (c *KVStore) ForEachNodeDirectedChannel(nodePub route.Vertex, cb func(channel *DirectedChannel) error) error { - return c.forEachNodeDirectedChannel(nil, nodePub, cb) + reset := func() {} + + return c.forEachNodeDirectedChannel(nil, nodePub, cb, reset) } // FetchNodeFeatures returns the features of the given node. If no features are @@ -669,7 +681,9 @@ func (c *KVStore) FetchNodeFeatures(nodePub route.Vertex) ( // NOTE: The callback contents MUST not be modified. func (c *KVStore) ForEachNodeCached(_ context.Context, cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { + chans map[uint64]*DirectedChannel) error) error { + + reset := func() {} // Otherwise call back to a version that uses the database directly. // We'll iterate over each node, then the set of channels for each @@ -723,13 +737,14 @@ func (c *KVStore) ForEachNodeCached(_ context.Context, channels[e.ChannelID] = directedChannel return nil - }) + }, reset, + ) if err != nil { return err } return cb(node.PubKeyBytes, channels) - }) + }, reset) } // DisabledChannelIDs returns the channel ids of disabled channels. @@ -793,11 +808,13 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) { func (c *KVStore) ForEachNode(_ context.Context, cb func(tx NodeRTx) error) error { + reset := func() {} + return forEachNode(c.db, func(tx kvdb.RTx, node *models.LightningNode) error { return cb(newChanGraphNodeTx(tx, c, node)) - }) + }, reset) } // forEachNode iterates through all the stored vertices/nodes in the graph, @@ -808,7 +825,7 @@ func (c *KVStore) ForEachNode(_ context.Context, // TODO(roasbeef): add iterator interface to allow for memory efficient graph // traversal when graph gets mega. func forEachNode(db kvdb.Backend, - cb func(kvdb.RTx, *models.LightningNode) error) error { + cb func(kvdb.RTx, *models.LightningNode) error, reset func()) error { traversal := func(tx kvdb.RTx) error { // First grab the nodes bucket which stores the mapping from @@ -838,7 +855,7 @@ func forEachNode(db kvdb.Backend, }) } - return kvdb.View(db, traversal, func() {}) + return kvdb.View(db, traversal, reset) } // ForEachNodeCacheable iterates through all the stored vertices/nodes in the @@ -848,6 +865,8 @@ func forEachNode(db kvdb.Backend, func (c *KVStore) ForEachNodeCacheable(_ context.Context, cb func(route.Vertex, *lnwire.FeatureVector) error) error { + reset := func() {} + traversal := func(tx kvdb.RTx) error { // First grab the nodes bucket which stores the mapping from // pubKey to node information. @@ -878,7 +897,7 @@ func (c *KVStore) ForEachNodeCacheable(_ context.Context, }) } - return kvdb.View(c.db, traversal, func() {}) + return kvdb.View(c.db, traversal, reset) } // SourceNode returns the source node of the graph. The source node is treated @@ -3018,6 +3037,8 @@ func (c *KVStore) isPublic(tx kvdb.RTx, nodePub route.Vertex, // Otherwise, we'll continue our search. return nil + }, func() { + nodeIsPublic = false }) if err != nil && !errors.Is(err, errDone) { return false, err @@ -3157,9 +3178,13 @@ func (c *KVStore) HasLightningNode(_ context.Context, // nodeTraversal is used to traverse all channels of a node given by its // public key and passes channel information into the specified callback. +// +// NOTE: the reset param is only meaningful if the tx param is nil. If it is +// not nil, the caller is expected to have passed in a reset to the parent +// function's View/Update call which will then apply to the whole transaction. func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error, reset func()) error { traversal := func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) @@ -3230,7 +3255,7 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, // If no transaction was provided, then we'll create a new transaction // to execute the transaction within. if tx == nil { - return kvdb.View(db, traversal, func() {}) + return kvdb.View(db, traversal, reset) } // Otherwise, we re-use the existing transaction to execute the graph @@ -3248,14 +3273,18 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, // Unknown policies are passed into the callback as nil values. func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error) error { - return nodeTraversal(nil, nodePub[:], c.db, func(_ kvdb.RTx, - info *models.ChannelEdgeInfo, policy, - policy2 *models.ChannelEdgePolicy) error { + reset := func() {} - return cb(info, policy, policy2) - }) + return nodeTraversal( + nil, nodePub[:], c.db, func(_ kvdb.RTx, + info *models.ChannelEdgeInfo, policy, + policy2 *models.ChannelEdgePolicy) error { + + return cb(info, policy, policy2) + }, reset, + ) } // ForEachSourceNodeChannel iterates through all channels of the source node, @@ -3264,7 +3293,9 @@ func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, // peer's node information. func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error { + otherNode *models.LightningNode) error) error { + + reset := func() {} return kvdb.View(c.db, func(tx kvdb.RTx) error { nodes := tx.ReadBucket(nodeBucket) @@ -3292,9 +3323,9 @@ func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, return cb( info.ChannelPoint, policy != nil, peer, ) - }, + }, reset, ) - }, func() {}) + }, reset) } // forEachNodeChannelTx iterates through all channels of the given node, @@ -3310,12 +3341,14 @@ func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, // should be passed as the first argument. Otherwise, the first argument should // be nil and a fresh transaction will be created to execute the graph // traversal. +// +// NOTE: the reset function is only meaningful if the tx param is nil. func (c *KVStore) forEachNodeChannelTx(tx kvdb.RTx, nodePub route.Vertex, cb func(kvdb.RTx, *models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, + reset func()) error { - return nodeTraversal(tx, nodePub[:], c.db, cb) + return nodeTraversal(tx, nodePub[:], c.db, cb, reset) } // fetchOtherNode attempts to fetch the full LightningNode that's opposite of @@ -3956,12 +3989,14 @@ func (c *KVStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { // GraphSession will provide the call-back with access to a NodeTraverser // instance which can be used to perform queries against the channel graph. func (c *KVStore) GraphSession(cb func(graph NodeTraverser) error) error { + reset := func() {} + return c.db.View(func(tx walletdb.ReadTx) error { return cb(&nodeTraverserSession{ db: c, tx: tx, }) - }, func() {}) + }, reset) } // nodeTraverserSession implements the NodeTraverser interface but with a @@ -3978,7 +4013,7 @@ type nodeTraverserSession struct { func (c *nodeTraverserSession) ForEachNodeDirectedChannel(nodePub route.Vertex, cb func(channel *DirectedChannel) error) error { - return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb) + return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb, func() {}) } // FetchNodeFeatures returns the features of the given node. If the node is @@ -4902,11 +4937,16 @@ func (c *chanGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { func (c *chanGraphNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - return c.db.forEachNodeChannelTx(c.tx, c.node.PubKeyBytes, + return c.db.forEachNodeChannelTx( + c.tx, c.node.PubKeyBytes, func(_ kvdb.RTx, info *models.ChannelEdgeInfo, policy1, policy2 *models.ChannelEdgePolicy) error { return f(info, policy1, policy2) }, + // NOTE: We don't need to reset anything here as the caller is + // expected to pass in the reset function to the ForEachNode + // method that constructed the chanGraphNodeTx. + func() {}, ) } diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index dfe5eb985..6cc6ab2ff 100644 --- a/graph/db/sql_migration.go +++ b/graph/db/sql_migration.go @@ -206,6 +206,9 @@ func migrateNodes(ctx context.Context, kvBackend kvdb.Backend, return sqldb.CompareRecords( node, migratedNode, fmt.Sprintf("node %x", pub), ) + }, func() { + // No reset is needed since if a retry occurs, the entire + // migration will be retried from the start. }) if err != nil { return fmt.Errorf("could not migrate nodes: %w", err) @@ -384,6 +387,9 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend, } return nil + }, func() { + // No reset is needed since if a retry occurs, the entire + // migration will be retried from the start. }) if err != nil { return fmt.Errorf("could not migrate channels and policies: %w", From e00f0a03fc227bed7e91a58f6cd410ebb1da51b7 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:11:30 +0200 Subject: [PATCH 02/11] graph/db: prep SQLStore for reset param This commit just surfaces the locations in the SQLStore where we will later pass reset params through. --- graph/db/sql_store.go | 34 +++++++++++++++++++++++++--------- 1 file changed, 25 insertions(+), 9 deletions(-) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index a42d4472d..98062cfe8 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -729,6 +729,8 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, otherNode *models.LightningNode) error) error { + reset := func() {} + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { nodeID, nodePub, err := s.getSourceNode(ctx, db, ProtocolV1) if err != nil { @@ -773,7 +775,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, ) }, ) - }, sqldb.NoOpReset) + }, reset) } // ForEachNode iterates through all the stored vertices/nodes in the graph, @@ -787,6 +789,8 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, func (s *SQLStore) ForEachNode(ctx context.Context, cb func(tx NodeRTx) error) error { + reset := func() {} + var lastID int64 = 0 handleNode := func(db SQLQueries, dbNode sqlc.Node) error { node, err := buildNode(ctx, db, &dbNode) @@ -835,7 +839,7 @@ func (s *SQLStore) ForEachNode(ctx context.Context, } return nil - }, sqldb.NoOpReset) + }, reset) } // sqlGraphNodeTx is an implementation of the NodeRTx interface backed by the @@ -909,11 +913,12 @@ func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, cb func(channel *DirectedChannel) error) error { + reset := func() {} var ctx = context.TODO() return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeDirectedChannel(ctx, db, nodePub, cb) - }, sqldb.NoOpReset) + }, reset) } // ForEachNodeCacheable iterates through all the stored vertices/nodes in the @@ -925,6 +930,8 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex, *lnwire.FeatureVector) error) error { + reset := func() {} + err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable(ctx, db, func(nodeID int64, nodePub route.Vertex) error { @@ -937,7 +944,7 @@ func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, return cb(nodePub, features) }) - }, sqldb.NoOpReset) + }, reset) if err != nil { return fmt.Errorf("unable to fetch nodes: %w", err) } @@ -959,6 +966,8 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { + reset := func() {} + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( ctx, sqlc.GetNodeByPubKeyParams{ @@ -975,7 +984,7 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, return forEachNodeChannel( ctx, db, s.cfg.ChainHash, dbNode.ID, cb, ) - }, sqldb.NoOpReset) + }, reset) } // ChanUpdatesInHorizon returns all the known channel edges which have at least @@ -1107,6 +1116,8 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error) error { + reset := func() {} + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable(ctx, db, func(nodeID int64, nodePub route.Vertex) error { @@ -1218,7 +1229,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, return cb(nodePub, channels) }) - }, sqldb.NoOpReset) + }, reset) } // ForEachChannelCacheable iterates through all the channel edges stored @@ -1237,6 +1248,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy, *models.CachedEdgePolicy) error) error { + reset := func() {} ctx := context.TODO() handleChannel := func(db SQLQueries, @@ -1315,7 +1327,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, } return nil - }, sqldb.NoOpReset) + }, reset) } // ForEachChannel iterates through all the channel edges stored within the @@ -1333,6 +1345,8 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { + reset := func() {} + handleChannel := func(db SQLQueries, row sqlc.ListChannelsWithPoliciesPaginatedRow) error { @@ -1406,7 +1420,7 @@ func (s *SQLStore) ForEachChannel(ctx context.Context, } return nil - }, sqldb.NoOpReset) + }, reset) } // FilterChannelRange returns the channel ID's of all known channels which were @@ -2757,9 +2771,11 @@ func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error) error { var ctx = context.TODO() + reset := func() {} + return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return cb(newSQLNodeTraverser(db, s.cfg.ChainHash)) - }, sqldb.NoOpReset) + }, reset) } // sqlNodeTraverser implements the NodeTraverser interface but with a backing From 88261834e58f0a37c8fe33ea92a430511c9e8d14 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:18:46 +0200 Subject: [PATCH 03/11] graph/db: expose reset in ForEachChannel Add a `reset` call-back param to the V1Store ForEachChannel method. --- graph/db/graph_test.go | 24 +++++++++++------------- graph/db/interfaces.go | 4 ++-- graph/db/kv_store.go | 4 +--- graph/db/sql_migration_test.go | 2 +- graph/db/sql_store.go | 4 +--- rpcserver.go | 2 ++ 6 files changed, 18 insertions(+), 22 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index bcafe4554..662eba77f 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -935,7 +935,8 @@ func TestEdgePolicyCRUD(t *testing.T) { ) return nil - }) + }, func() {}, + ) require.NoError(t, err) } @@ -1385,7 +1386,7 @@ func TestGraphTraversal(t *testing.T) { delete(chanIndex, ei.ChannelID) return nil - }) + }, func() {}) require.NoError(t, err) require.Len(t, chanIndex, 0) @@ -1669,22 +1670,19 @@ func assertPruneTip(t *testing.T, graph *ChannelGraph, func assertNumChans(t *testing.T, graph *ChannelGraph, n int) { numChans := 0 - if err := graph.ForEachChannel(context.Background(), - func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + err := graph.ForEachChannel( + context.Background(), func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error { numChans++ return nil + }, func() { + numChans = 0 }, - ); err != nil { - _, _, line, _ := runtime.Caller(1) - t.Fatalf("line %v: unable to scan channels: %v", line, err) - } - if numChans != n { - _, _, line, _ := runtime.Caller(1) - t.Fatalf("line %v: expected %v chans instead have %v", line, - n, numChans) - } + ) + require.NoError(t, err) + require.Equal(t, n, numChans) } func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) { diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index ecb07ec6d..388ba3ca6 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -159,8 +159,8 @@ type V1Store interface { //nolint:interfacebloat // pointer for that particular channel edge routing policy will be // passed into the callback. ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, + reset func()) error // ForEachChannelCacheable iterates through all the channel edges stored // within the graph and invokes the passed callback for each edge. The diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 92700376d..cd56bb41e 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -406,9 +406,7 @@ func (c *KVStore) AddrsForNode(ctx context.Context, // callback. func (c *KVStore) ForEachChannel(_ context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { - - reset := func() {} + *models.ChannelEdgePolicy) error, reset func()) error { return forEachChannel(c.db, cb, reset) } diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 886055fb2..336a4d1c4 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -475,7 +475,7 @@ func fetchAllChannelsAndPolicies(t *testing.T, store V1Store) chanSet { }) return nil - }) + }, func() {}) require.NoError(t, err) // Sort the channels by their channel ID to ensure a consistent order. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 98062cfe8..b241a3fa4 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1343,9 +1343,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { - - reset := func() {} + *models.ChannelEdgePolicy) error, reset func()) error { handleChannel := func(db SQLQueries, row sqlc.ListChannelsWithPoliciesPaginatedRow) error { diff --git a/rpcserver.go b/rpcserver.go index d4e7e9da0..258be4abd 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6768,6 +6768,8 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, resp.Edges = append(resp.Edges, edge) return nil + }, func() { + resp.Edges = nil }) if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) { return nil, err From 16126d77031eac0dabbe676e791bebeff3b14222 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:23:59 +0200 Subject: [PATCH 04/11] graph/db: expose reset in ForEachSourceNodeChannel --- graph/db/graph_test.go | 2 +- graph/db/interfaces.go | 3 ++- graph/db/kv_store.go | 4 +--- graph/db/sql_store.go | 4 +--- server.go | 18 ++++++++++++++---- 5 files changed, 19 insertions(+), 12 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 662eba77f..50d288cdc 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1331,7 +1331,7 @@ func TestForEachSourceNodeChannel(t *testing.T) { delete(expectedSrcChans, chanPoint) return nil - }) + }, func() {}) require.NoError(t, err) require.Empty(t, expectedSrcChans) } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 388ba3ca6..652e7bd34 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -72,7 +72,8 @@ type V1Store interface { //nolint:interfacebloat // the channel and the channel peer's node information. ForEachSourceNodeChannel(ctx context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error + otherNode *models.LightningNode) error, + reset func()) error // ForEachNodeChannel iterates through all channels of the given node, // executing the passed callback with an edge info structure and the diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index cd56bb41e..2a52848bf 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3291,9 +3291,7 @@ func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, // peer's node information. func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error { - - reset := func() {} + otherNode *models.LightningNode) error, reset func()) error { return kvdb.View(c.db, func(tx kvdb.RTx) error { nodes := tx.ReadBucket(nodeBucket) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index b241a3fa4..5a8c75a4f 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -727,9 +727,7 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy, // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error) error { - - reset := func() {} + otherNode *models.LightningNode) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { nodeID, nodePub, err := s.getSourceNode(ctx, db, ProtocolV1) diff --git a/server.go b/server.go index 4ad5c09e3..2c42dab6c 100644 --- a/server.go +++ b/server.go @@ -1254,7 +1254,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, DefaultRoutingPolicy: cc.RoutingPolicy, ForAllOutgoingChannels: func(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error) error { return s.graphDB.ForEachNodeChannel(ctx, selfVertex, func(c *models.ChannelEdgeInfo, @@ -3599,7 +3599,7 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { // nodeAddrsMap stores the combination of node public keys and addresses // that we'll attempt to reconnect to. PubKey strings are used as keys // since other PubKey forms can't be compared. - nodeAddrsMap := map[string]*nodeAddresses{} + nodeAddrsMap := make(map[string]*nodeAddresses) // Iterate through the list of LinkNodes to find addresses we should // attempt to connect to based on our set of previous connections. Set @@ -3623,6 +3623,7 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { // that have been added via NodeAnnouncement messages. // TODO(roasbeef): instead iterate over link nodes and query graph for // each of the nodes. + graphAddrs := make(map[string]*nodeAddresses) forEachSrcNodeChan := func(chanPoint wire.OutPoint, havePolicy bool, channelPeer *models.LightningNode) error { @@ -3688,10 +3689,14 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { return err } - nodeAddrsMap[pubStr] = n + graphAddrs[pubStr] = n return nil } - err = s.graphDB.ForEachSourceNodeChannel(ctx, forEachSrcNodeChan) + err = s.graphDB.ForEachSourceNodeChannel( + ctx, forEachSrcNodeChan, func() { + clear(graphAddrs) + }, + ) if err != nil { srvrLog.Errorf("Failed to iterate over source node channels: "+ "%v", err) @@ -3703,6 +3708,11 @@ func (s *server) establishPersistentConnections(ctx context.Context) error { } } + // Combine the addresses from the link nodes and the channel graph. + for pubStr, nodeAddr := range graphAddrs { + nodeAddrsMap[pubStr] = nodeAddr + } + srvrLog.Debugf("Establishing %v persistent connections on start", len(nodeAddrsMap)) From e5fbca8299fd6610cd7ee1fa82823a269a01b2c6 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:31:29 +0200 Subject: [PATCH 05/11] multi: let ForEachNodeChannel take a reset param --- autopilot/prefattach_test.go | 3 ++- discovery/gossiper.go | 3 +++ discovery/gossiper_test.go | 8 +++----- graph/builder.go | 9 +++++---- graph/db/graph_test.go | 23 +++++++++++------------ graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 14 ++++++-------- graph/db/sql_store.go | 4 +--- graph/interfaces.go | 4 ++-- routing/localchans/manager.go | 11 +++++++++-- routing/localchans/manager_test.go | 2 +- rpcserver.go | 12 ++++++++++-- server.go | 5 +++-- 13 files changed, 57 insertions(+), 43 deletions(-) diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index fd557a02c..282e32625 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -736,7 +736,8 @@ func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, policy2 *models.ChannelEdgePolicy) error { return f(edge, policy1, policy2) - }) + }, func() {}, + ) } func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) { diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 70c1caf2e..c4677bbf5 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -1758,6 +1758,9 @@ func (d *AuthenticatedGossiper) retransmitStaleAnns(ctx context.Context, } return nil + }, func() { + havePublicChannels = false + edgesToUpdate = nil }) if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) { return fmt.Errorf("unable to retrieve outgoing channels: %w", diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index c10bdbbe1..6b9a1d58f 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -213,7 +213,7 @@ func (r *mockGraphSource) ForEachNode( func (r *mockGraphSource) ForAllOutgoingChannels(_ context.Context, cb func(i *models.ChannelEdgeInfo, - c *models.ChannelEdgePolicy) error) error { + c *models.ChannelEdgePolicy) error, _ func()) error { r.mu.Lock() defer r.mu.Unlock() @@ -3728,10 +3728,8 @@ out: }) return nil - }) - if err != nil { - t.Fatal(err) - } + }, func() {}) + require.NoError(t, err) err = ctx.gossiper.PropagateChanPolicyUpdate(edgesToUpdate) require.NoError(t, err, "unable to chan policies") diff --git a/graph/builder.go b/graph/builder.go index 073716a83..5cfb79e4c 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1276,10 +1276,11 @@ func (b *Builder) FetchLightningNode(ctx context.Context, // // NOTE: This method is part of the ChannelGraphSource interface. func (b *Builder) ForAllOutgoingChannels(ctx context.Context, - cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy) error, + reset func()) error { - return b.cfg.Graph.ForEachNodeChannel(ctx, b.cfg.SelfNode, + return b.cfg.Graph.ForEachNodeChannel( + ctx, b.cfg.SelfNode, func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { @@ -1289,7 +1290,7 @@ func (b *Builder) ForAllOutgoingChannels(ctx context.Context, } return cb(c, e) - }, + }, reset, ) } diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 50d288cdc..8d62f65d4 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1394,7 +1394,8 @@ func TestGraphTraversal(t *testing.T) { // outgoing channels for a particular node. numNodeChans := 0 firstNode, secondNode := nodeList[0], nodeList[1] - err = graph.ForEachNodeChannel(ctx, firstNode.PubKeyBytes, + err = graph.ForEachNodeChannel( + ctx, firstNode.PubKeyBytes, func(_ *models.ChannelEdgeInfo, outEdge, inEdge *models.ChannelEdgePolicy) error { @@ -1425,7 +1426,8 @@ func TestGraphTraversal(t *testing.T) { numNodeChans++ return nil - }) + }, func() {}, + ) require.NoError(t, err) require.Equal(t, numChannels, numNodeChans) } @@ -3144,7 +3146,8 @@ func TestIncompleteChannelPolicies(t *testing.T) { expectedOut bool) { calls := 0 - err := graph.ForEachNodeChannel(ctx, node.PubKeyBytes, + err := graph.ForEachNodeChannel( + ctx, node.PubKeyBytes, func(_ *models.ChannelEdgeInfo, outEdge, inEdge *models.ChannelEdgePolicy) error { @@ -3167,14 +3170,10 @@ func TestIncompleteChannelPolicies(t *testing.T) { calls++ return nil - }) - if err != nil { - t.Fatalf("unable to scan channels: %v", err) - } - - if calls != 1 { - t.Fatalf("Expected only one callback call") - } + }, func() {}, + ) + require.NoError(t, err) + require.Equal(t, 1, calls) } checkPolicies(node2, false, false) @@ -4248,7 +4247,7 @@ func BenchmarkForEachChannel(b *testing.B) { return nil } - err := graph.ForEachNodeChannel(ctx, n, cb) + err := graph.ForEachNodeChannel(ctx, n, cb, func() {}) require.NoError(b, err) } } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 652e7bd34..8c4b0c855 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -86,7 +86,7 @@ type V1Store interface { //nolint:interfacebloat // Unknown policies are passed into the callback as nil values. ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error + *models.ChannelEdgePolicy) error, reset func()) error // ForEachNodeCached is similar to forEachNode, but it returns // DirectedChannel data to the call-back. diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 2a52848bf..19d412d83 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -406,7 +406,7 @@ func (c *KVStore) AddrsForNode(ctx context.Context, // callback. func (c *KVStore) ForEachChannel(_ context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error, reset func()) error { + *models.ChannelEdgePolicy) error, reset func()) error { return forEachChannel(c.db, cb, reset) } @@ -679,7 +679,7 @@ func (c *KVStore) FetchNodeFeatures(nodePub route.Vertex) ( // NOTE: The callback contents MUST not be modified. func (c *KVStore) ForEachNodeCached(_ context.Context, cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { + chans map[uint64]*DirectedChannel) error) error { reset := func() {} @@ -3182,7 +3182,7 @@ func (c *KVStore) HasLightningNode(_ context.Context, // function's View/Update call which will then apply to the whole transaction. func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error, reset func()) error { + *models.ChannelEdgePolicy) error, reset func()) error { traversal := func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) @@ -3271,9 +3271,7 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, // Unknown policies are passed into the callback as nil values. func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { - - reset := func() {} + *models.ChannelEdgePolicy) error, reset func()) error { return nodeTraversal( nil, nodePub[:], c.db, func(_ kvdb.RTx, @@ -3291,7 +3289,7 @@ func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, // peer's node information. func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, cb func(chanPoint wire.OutPoint, havePolicy bool, - otherNode *models.LightningNode) error, reset func()) error { + otherNode *models.LightningNode) error, reset func()) error { return kvdb.View(c.db, func(tx kvdb.RTx) error { nodes := tx.ReadBucket(nodeBucket) @@ -3341,7 +3339,7 @@ func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, // NOTE: the reset function is only meaningful if the tx param is nil. func (c *KVStore) forEachNodeChannelTx(tx kvdb.RTx, nodePub route.Vertex, cb func(kvdb.RTx, *models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, + *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error, reset func()) error { return nodeTraversal(tx, nodePub[:], c.db, cb, reset) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 5a8c75a4f..dc9365a17 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -962,9 +962,7 @@ func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { - - reset := func() {} + *models.ChannelEdgePolicy) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( diff --git a/graph/interfaces.go b/graph/interfaces.go index 20b63dd15..5494db61a 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -72,7 +72,7 @@ type ChannelGraphSource interface { // star-graph. ForAllOutgoingChannels(ctx context.Context, cb func(c *models.ChannelEdgeInfo, - e *models.ChannelEdgePolicy) error) error + e *models.ChannelEdgePolicy) error, reset func()) error // CurrentBlockHeight returns the block height from POV of the router // subsystem. @@ -260,7 +260,7 @@ type DB interface { // Unknown policies are passed into the callback as nil values. ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error + *models.ChannelEdgePolicy) error, reset func()) error // AddEdgeProof sets the proof of an existing edge in the graph // database. diff --git a/routing/localchans/manager.go b/routing/localchans/manager.go index faba9fa8b..b1d281187 100644 --- a/routing/localchans/manager.go +++ b/routing/localchans/manager.go @@ -42,7 +42,7 @@ type Manager struct { // channels. The ChannelEdgePolicy parameter may be nil. ForAllOutgoingChannels func(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error + *models.ChannelEdgePolicy) error, reset func()) error // FetchChannel is used to query local channel parameters. Optionally an // existing db tx can be supplied. @@ -152,7 +152,14 @@ func (r *Manager) UpdatePolicy(ctx context.Context, // Next, we'll loop over all the outgoing channels the router knows of. // If we have a filter then we'll only collect those channels, otherwise // we'll collect them all. - err := r.ForAllOutgoingChannels(ctx, processChan) + err := r.ForAllOutgoingChannels( + ctx, processChan, + func() { + failedUpdates = nil + edgesToUpdate = nil + clear(policiesToUpdate) + }, + ) if err != nil { return nil, err } diff --git a/routing/localchans/manager_test.go b/routing/localchans/manager_test.go index 6e53f4a27..09b4f0bc8 100644 --- a/routing/localchans/manager_test.go +++ b/routing/localchans/manager_test.go @@ -125,7 +125,7 @@ func TestManager(t *testing.T) { forAllOutgoingChannels := func(_ context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error, _ func()) error { for _, c := range channelSet { if err := cb(c.edgeInfo, ¤tPolicy); err != nil { diff --git a/rpcserver.go b/rpcserver.go index 258be4abd..0d003739f 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -7048,7 +7048,8 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context, channels []*lnrpc.ChannelEdge ) - err = graph.ForEachNodeChannel(ctx, node.PubKeyBytes, + err = graph.ForEachNodeChannel( + ctx, node.PubKeyBytes, func(edge *models.ChannelEdgeInfo, c1, c2 *models.ChannelEdgePolicy) error { @@ -7074,6 +7075,10 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context, } return nil + }, func() { + numChannels = 0 + totalCapacity = 0 + channels = nil }, ) if err != nil { @@ -7730,7 +7735,8 @@ func (r *rpcServer) FeeReport(ctx context.Context, } var feeReports []*lnrpc.ChannelFeeReport - err = channelGraph.ForEachNodeChannel(ctx, selfNode.PubKeyBytes, + err = channelGraph.ForEachNodeChannel( + ctx, selfNode.PubKeyBytes, func(chanInfo *models.ChannelEdgeInfo, edgePolicy, _ *models.ChannelEdgePolicy) error { @@ -7768,6 +7774,8 @@ func (r *rpcServer) FeeReport(ctx context.Context, }) return nil + }, func() { + feeReports = nil }, ) if err != nil { diff --git a/server.go b/server.go index 2c42dab6c..bfc1850d7 100644 --- a/server.go +++ b/server.go @@ -1254,7 +1254,8 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, DefaultRoutingPolicy: cc.RoutingPolicy, ForAllOutgoingChannels: func(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { + *models.ChannelEdgePolicy) error, + reset func()) error { return s.graphDB.ForEachNodeChannel(ctx, selfVertex, func(c *models.ChannelEdgeInfo, @@ -1264,7 +1265,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, // NOTE: The invoked callback here may // receive a nil channel policy. return cb(c, e) - }, + }, reset, ) }, PropagateChanPolicyUpdate: s.authGossiper.PropagateChanPolicyUpdate, From c32bf642d24ab410b0dba346af389ce33bde96b0 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:44:06 +0200 Subject: [PATCH 06/11] multi: pass reset to ForEachNodeCached --- autopilot/agent.go | 3 +++ autopilot/graph.go | 6 +++--- autopilot/interface.go | 6 ++++-- autopilot/prefattach.go | 6 ++++++ autopilot/prefattach_test.go | 16 ++++++++++++---- autopilot/simple_graph.go | 4 ++++ discovery/bootstrapper.go | 12 ++++++++++-- graph/db/graph.go | 6 +++--- graph/db/graph_test.go | 2 +- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 6 ++---- graph/db/sql_store.go | 6 ++---- rpcserver.go | 9 +++++++++ server.go | 2 +- 14 files changed, 61 insertions(+), 25 deletions(-) diff --git a/autopilot/agent.go b/autopilot/agent.go index 1b70580a3..30e76d3d9 100644 --- a/autopilot/agent.go +++ b/autopilot/agent.go @@ -640,6 +640,9 @@ func (a *Agent) openChans(ctx context.Context, availableFunds btcutil.Amount, nodes[nID] = struct{}{} return nil + }, func() { + nodes = nil + addresses = nil }); err != nil { return fmt.Errorf("unable to get graph nodes: %w", err) } diff --git a/autopilot/graph.go b/autopilot/graph.go index 3aa3ad055..fe74ad1fe 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -121,7 +121,7 @@ func (d *dbNode) ForEachChannel(ctx context.Context, // // NOTE: Part of the autopilot.ChannelGraph interface. func (d *databaseChannelGraph) ForEachNode(ctx context.Context, - cb func(context.Context, Node) error) error { + cb func(context.Context, Node) error, _ func()) error { return d.db.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error { // We'll skip over any node that doesn't have any advertised @@ -217,7 +217,7 @@ func (nc dbNodeCached) ForEachChannel(ctx context.Context, // // NOTE: Part of the autopilot.ChannelGraph interface. func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, - cb func(context.Context, Node) error) error { + cb func(context.Context, Node) error, reset func()) error { return dc.db.ForEachNodeCached(ctx, func(n route.Vertex, channels map[uint64]*graphdb.DirectedChannel) error { @@ -231,7 +231,7 @@ func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, return cb(ctx, node) } return nil - }) + }, reset) } // memNode is a purely in-memory implementation of the autopilot.Node diff --git a/autopilot/interface.go b/autopilot/interface.go index 703f49020..69eb4e6f1 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -85,7 +85,8 @@ type ChannelGraph interface { // ForEachNode is a higher-order function that should be called once // for each connected node within the channel graph. If the passed // callback returns an error, then execution should be terminated. - ForEachNode(context.Context, func(context.Context, Node) error) error + ForEachNode(context.Context, func(context.Context, Node) error, + func()) error } // NodeScore is a tuple mapping a NodeID to a score indicating the preference @@ -235,5 +236,6 @@ type GraphSource interface { // ForEachNode since any further calls are made across multiple // transactions. ForEachNodeCached(ctx context.Context, cb func(node route.Vertex, - chans map[uint64]*graphdb.DirectedChannel) error) error + chans map[uint64]*graphdb.DirectedChannel) error, + reset func()) error } diff --git a/autopilot/prefattach.go b/autopilot/prefattach.go index 76d814a46..e9a9c53b4 100644 --- a/autopilot/prefattach.go +++ b/autopilot/prefattach.go @@ -105,6 +105,9 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, } return nil + }, func() { + allChans = nil + clear(seenChans) }); err != nil { return nil, err } @@ -162,6 +165,9 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, log.Tracef("Counted %v channels for node %x", nodeChans, nID[:]) return nil + }, func() { + maxChans = 0 + clear(nodeChanNum) }); err != nil { return nil, err } diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index 282e32625..b5c0b1d26 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -147,10 +147,13 @@ func TestPrefAttachmentSelectTwoVertexes(t *testing.T) { // Get the score for all nodes found in the graph at // this point. nodes := make(map[NodeID]struct{}) - err = graph.ForEachNode(ctx, + err = graph.ForEachNode( + ctx, func(_ context.Context, n Node) error { nodes[n.PubKey()] = struct{}{} return nil + }, func() { + clear(nodes) }, ) require.NoError(t1, err) @@ -257,7 +260,12 @@ func TestPrefAttachmentSelectGreedyAllocation(t *testing.T) { twoChans = twoChans || (numChans == 2) return nil - }) + }, func() { + numNodes = 0 + twoChans = false + clear(nodes) + }, + ) require.NoError(t1, err) require.EqualValues(t1, 3, numNodes) @@ -338,7 +346,7 @@ func TestPrefAttachmentSelectSkipNodes(t *testing.T) { nodes[n.PubKey()] = struct{}{} return nil - }, + }, func() {}, ) require.NoError(t1, err) @@ -593,7 +601,7 @@ func newMemChannelGraph() *memChannelGraph { // // NOTE: Part of the autopilot.ChannelGraph interface. func (m *memChannelGraph) ForEachNode(ctx context.Context, - cb func(context.Context, Node) error) error { + cb func(context.Context, Node) error, _ func()) error { for _, node := range m.graph { if err := cb(ctx, node); err != nil { diff --git a/autopilot/simple_graph.go b/autopilot/simple_graph.go index f028db3c7..2f80c490a 100644 --- a/autopilot/simple_graph.go +++ b/autopilot/simple_graph.go @@ -60,6 +60,10 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) { return nil }, ) + }, func() { + clear(adj) + clear(nodes) + nextIndex = 0 }) if err != nil { return nil, err diff --git a/discovery/bootstrapper.go b/discovery/bootstrapper.go index dbfdbc7d2..1b6c981d4 100644 --- a/discovery/bootstrapper.go +++ b/discovery/bootstrapper.go @@ -247,9 +247,9 @@ func (c *ChannelGraphBootstrapper) SampleNodeAddrs(_ context.Context, }) } - c.tried[nID] = struct{}{} - return errFound + }, func() { + a = nil }) if err != nil && !errors.Is(err, errFound) { return nil, err @@ -282,6 +282,14 @@ func (c *ChannelGraphBootstrapper) SampleNodeAddrs(_ context.Context, continue } + for _, addr := range sampleAddrs { + nID := autopilot.NodeID( + addr.IdentityKey.SerializeCompressed(), + ) + + c.tried[nID] = struct{}{} + } + addrs = append(addrs, sampleAddrs...) } diff --git a/graph/db/graph.go b/graph/db/graph.go index 3168a6618..2ae944e9e 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -246,14 +246,14 @@ func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error { // // NOTE: The callback contents MUST not be modified. func (c *ChannelGraph) ForEachNodeCached(ctx context.Context, - cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { + cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error, + reset func()) error { if c.graphCache != nil { return c.graphCache.ForEachNode(cb) } - return c.V1Store.ForEachNodeCached(ctx, cb) + return c.V1Store.ForEachNodeCached(ctx, cb, reset) } // AddLightningNode adds a vertex/node to the graph database. If the node is not diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 8d62f65d4..20b097541 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1374,7 +1374,7 @@ func TestGraphTraversal(t *testing.T) { } return nil - }) + }, func() {}) require.NoError(t, err) // Iterate through all the known channels within the graph DB, once diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 8c4b0c855..e47935c11 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -93,7 +93,7 @@ type V1Store interface { //nolint:interfacebloat // // NOTE: The callback contents MUST not be modified. ForEachNodeCached(ctx context.Context, cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error + chans map[uint64]*DirectedChannel) error, reset func()) error // ForEachNode iterates through all the stored vertices/nodes in the // graph, executing the passed callback with each node encountered. If diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 19d412d83..45625437c 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -678,10 +678,8 @@ func (c *KVStore) FetchNodeFeatures(nodePub route.Vertex) ( // // NOTE: The callback contents MUST not be modified. func (c *KVStore) ForEachNodeCached(_ context.Context, - cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { - - reset := func() {} + cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error, + reset func()) error { // Otherwise call back to a version that uses the database directly. // We'll iterate over each node, then the set of channels for each diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index dc9365a17..ad7bfc0fe 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1109,10 +1109,8 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, // // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachNodeCached(ctx context.Context, - cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { - - reset := func() {} + cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error, + reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable(ctx, db, func(nodeID int64, diff --git a/rpcserver.go b/rpcserver.go index 0d003739f..39169a05b 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -7215,6 +7215,15 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, } return nil + }, func() { + numChannels = 0 + numNodes = 0 + maxChanOut = 0 + totalNetworkCapacity = 0 + minChannelSize = math.MaxInt64 + maxChannelSize = 0 + allChans = nil + clear(seenChans) }) if err != nil { return nil, err diff --git a/server.go b/server.go index bfc1850d7..7eb8c583d 100644 --- a/server.go +++ b/server.go @@ -1254,7 +1254,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, DefaultRoutingPolicy: cc.RoutingPolicy, ForAllOutgoingChannels: func(ctx context.Context, cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error, + *models.ChannelEdgePolicy) error, reset func()) error { return s.graphDB.ForEachNodeChannel(ctx, selfVertex, From 21ac7ef6f4e317ce88e0b14be237a3d74cb066f3 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:55:20 +0200 Subject: [PATCH 07/11] multi: add reset to ForEachNode --- autopilot/graph.go | 4 ++-- autopilot/interface.go | 2 +- graph/db/graph_test.go | 6 +++--- graph/db/interfaces.go | 3 ++- graph/db/kv_store.go | 4 +--- graph/db/sql_migration_test.go | 2 ++ graph/db/sql_store.go | 4 +--- itest/lnd_graph_migration_test.go | 3 +++ rpcserver.go | 2 ++ 9 files changed, 17 insertions(+), 13 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index fe74ad1fe..f1506b5d7 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -121,7 +121,7 @@ func (d *dbNode) ForEachChannel(ctx context.Context, // // NOTE: Part of the autopilot.ChannelGraph interface. func (d *databaseChannelGraph) ForEachNode(ctx context.Context, - cb func(context.Context, Node) error, _ func()) error { + cb func(context.Context, Node) error, reset func()) error { return d.db.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error { // We'll skip over any node that doesn't have any advertised @@ -136,7 +136,7 @@ func (d *databaseChannelGraph) ForEachNode(ctx context.Context, } return cb(ctx, node) - }) + }, reset) } // databaseChannelGraphCached wraps a channeldb.ChannelGraph instance with the diff --git a/autopilot/interface.go b/autopilot/interface.go index 69eb4e6f1..a4efa2eec 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -229,7 +229,7 @@ type GraphSource interface { // the callback returns an error, then the transaction is aborted and // the iteration stops early. Any operations performed on the NodeTx // passed to the call-back are executed under the same read transaction. - ForEachNode(context.Context, func(graphdb.NodeRTx) error) error + ForEachNode(context.Context, func(graphdb.NodeRTx) error, func()) error // ForEachNodeCached is similar to ForEachNode, but it utilizes the // channel graph cache if one is available. It is less consistent than diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 20b097541..3a7f9b9de 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1454,7 +1454,7 @@ func TestGraphTraversalCacheable(t *testing.T) { nodeMap[tx.Node().PubKeyBytes] = struct{}{} return nil - }) + }, func() {}) require.NoError(t, err) require.Len(t, nodeMap, numNodes) @@ -1583,7 +1583,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, err := graph.ForEachNode(ctx, func(tx NodeRTx) error { delete(nodeIndex, tx.Node().Alias) return nil - }) + }, func() {}) require.NoError(t, err) require.Len(t, nodeIndex, 0) @@ -1693,7 +1693,7 @@ func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) { numNodes++ return nil - }) + }, func() {}) if err != nil { _, _, line, _ := runtime.Caller(1) t.Fatalf("line %v: unable to scan nodes: %v", line, err) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index e47935c11..616aabc9f 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -102,7 +102,8 @@ type V1Store interface { //nolint:interfacebloat // passed to the call-back are executed under the same read transaction // and so, methods on the NodeTx object _MUST_ only be called from // within the call-back. - ForEachNode(ctx context.Context, cb func(tx NodeRTx) error) error + ForEachNode(ctx context.Context, cb func(tx NodeRTx) error, + reset func()) error // ForEachNodeCacheable iterates through all the stored vertices/nodes // in the graph, executing the passed callback with each node diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 45625437c..eff37ff21 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -802,9 +802,7 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) { // executed under the same read transaction and so, methods on the NodeTx object // _MUST_ only be called from within the call-back. func (c *KVStore) ForEachNode(_ context.Context, - cb func(tx NodeRTx) error) error { - - reset := func() {} + cb func(tx NodeRTx) error, reset func()) error { return forEachNode(c.db, func(tx kvdb.RTx, node *models.LightningNode) error { diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 336a4d1c4..9f32603fd 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -401,6 +401,8 @@ func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode { nodes = append(nodes, node) return nil + }, func() { + nodes = nil }) require.NoError(t, err) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index ad7bfc0fe..aef2f2464 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -785,9 +785,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, // // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachNode(ctx context.Context, - cb func(tx NodeRTx) error) error { - - reset := func() {} + cb func(tx NodeRTx) error, reset func()) error { var lastID int64 = 0 handleNode := func(db SQLQueries, dbNode sqlc.Node) error { diff --git a/itest/lnd_graph_migration_test.go b/itest/lnd_graph_migration_test.go index 412d23f57..d7417abe4 100644 --- a/itest/lnd_graph_migration_test.go +++ b/itest/lnd_graph_migration_test.go @@ -82,6 +82,9 @@ func testGraphMigration(ht *lntest.HarnessTest) { return nil }, ) + }, func() { + clear(edges) + numNodes = 0 }) require.NoError(ht, err) require.Equal(ht, expNumNodes, numNodes) diff --git a/rpcserver.go b/rpcserver.go index 39169a05b..8c134cbcd 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6745,6 +6745,8 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, resp.Nodes = append(resp.Nodes, lnNode) return nil + }, func() { + resp.Nodes = nil }) if err != nil { return nil, err From e17fd58a1d4dba7c0afd60749fa13a238a45c450 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 10:57:40 +0200 Subject: [PATCH 08/11] graph/db: pass reset to ForEachNodeCacheable --- graph/db/graph.go | 2 +- graph/db/graph_test.go | 4 ++++ graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 5 ++--- graph/db/sql_store.go | 5 ++--- 5 files changed, 10 insertions(+), 8 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index 2ae944e9e..ba6ad6ae0 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -170,7 +170,7 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { c.graphCache.AddNodeFeatures(node, features) return nil - }) + }, func() {}) if err != nil { return err } diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 3a7f9b9de..d2f1577b2 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1469,6 +1469,8 @@ func TestGraphTraversalCacheable(t *testing.T) { nodes = append(nodes, node) return nil + }, func() { + nodes = nil }) require.NoError(t, err) require.Len(t, nodeMap, 0) @@ -4227,6 +4229,8 @@ func BenchmarkForEachChannel(b *testing.B) { nodes = append(nodes, node) return nil + }, func() { + nodes = nil }) require.NoError(b, err) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 616aabc9f..e02081da1 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -110,7 +110,7 @@ type V1Store interface { //nolint:interfacebloat // encountered. If the callback returns an error, then the transaction // is aborted and the iteration stops early. ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex, - *lnwire.FeatureVector) error) error + *lnwire.FeatureVector) error, reset func()) error // LookupAlias attempts to return the alias as advertised by the target // node. diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index eff37ff21..906012bcc 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -857,9 +857,8 @@ func forEachNode(db kvdb.Backend, // callback returns an error, then the transaction is aborted and the iteration // stops early. func (c *KVStore) ForEachNodeCacheable(_ context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error) error { - - reset := func() {} + cb func(route.Vertex, *lnwire.FeatureVector) error, + reset func()) error { traversal := func(tx kvdb.RTx) error { // First grab the nodes bucket which stores the mapping from diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index aef2f2464..15a902c4e 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -924,9 +924,8 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, // // NOTE: This is a part of the V1Store interface. func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, - cb func(route.Vertex, *lnwire.FeatureVector) error) error { - - reset := func() {} + cb func(route.Vertex, *lnwire.FeatureVector) error, + reset func()) error { err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable(ctx, db, func(nodeID int64, From b3f3766fe6bdd7ff81257b24b1f041a62e814a37 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 11:02:15 +0200 Subject: [PATCH 09/11] multi: pass reset to GraphSession --- graph/db/graph.go | 6 ++++-- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 4 ++-- graph/db/sql_store.go | 6 +++--- routing/graph.go | 3 ++- routing/mock_graph_test.go | 4 ++-- routing/pathfind_test.go | 2 ++ routing/payment_session.go | 6 +++++- routing/payment_session_test.go | 4 ++-- 9 files changed, 23 insertions(+), 14 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index ba6ad6ae0..c0a4f52b6 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -233,12 +233,14 @@ func (c *ChannelGraph) FetchNodeFeatures(node route.Vertex) ( // instance which can be used to perform queries against the channel graph. If // the graph cache is not enabled, then the call-back will be provided with // access to the graph via a consistent read-only transaction. -func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error { +func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error, + reset func()) error { + if c.graphCache != nil { return cb(c) } - return c.V1Store.GraphSession(cb) + return c.V1Store.GraphSession(cb, reset) } // ForEachNodeCached iterates through all the stored vertices/nodes in the diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index e02081da1..f4cbb475d 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -149,7 +149,7 @@ type V1Store interface { //nolint:interfacebloat // GraphSession will provide the call-back with access to a // NodeTraverser instance which can be used to perform queries against // the channel graph. - GraphSession(cb func(graph NodeTraverser) error) error + GraphSession(cb func(graph NodeTraverser) error, reset func()) error // ForEachChannel iterates through all the channel edges stored within // the graph and invokes the passed callback for each edge. The callback diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 906012bcc..020d7d3d3 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3977,8 +3977,8 @@ func (c *KVStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { // GraphSession will provide the call-back with access to a NodeTraverser // instance which can be used to perform queries against the channel graph. -func (c *KVStore) GraphSession(cb func(graph NodeTraverser) error) error { - reset := func() {} +func (c *KVStore) GraphSession(cb func(graph NodeTraverser) error, + reset func()) error { return c.db.View(func(tx walletdb.ReadTx) error { return cb(&nodeTraverserSession{ diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 15a902c4e..58f29a75e 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -2757,10 +2757,10 @@ func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) { // instance which can be used to perform queries against the channel graph. // // NOTE: part of the V1Store interface. -func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error) error { - var ctx = context.TODO() +func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error, + reset func()) error { - reset := func() {} + var ctx = context.TODO() return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return cb(newSQLNodeTraverser(db, s.cfg.ChainHash)) diff --git a/routing/graph.go b/routing/graph.go index 115f2daf1..d5c6b2d7d 100644 --- a/routing/graph.go +++ b/routing/graph.go @@ -29,7 +29,8 @@ type GraphSessionFactory interface { // GraphSession will provide the call-back with access to a // graphdb.NodeTraverser instance which can be used to perform queries // against the channel graph. - GraphSession(cb func(graph graphdb.NodeTraverser) error) error + GraphSession(cb func(graph graphdb.NodeTraverser) error, + reset func()) error } // FetchAmountPairCapacity determines the maximal public capacity between two diff --git a/routing/mock_graph_test.go b/routing/mock_graph_test.go index cb5d07f3a..b5c8b55bc 100644 --- a/routing/mock_graph_test.go +++ b/routing/mock_graph_test.go @@ -232,8 +232,8 @@ func (m *mockGraph) FetchNodeFeatures(nodePub route.Vertex) ( // the channel graph. // // NOTE: Part of the GraphSessionFactory interface. -func (m *mockGraph) GraphSession( - cb func(graph graphdb.NodeTraverser) error) error { +func (m *mockGraph) GraphSession(cb func(graph graphdb.NodeTraverser) error, + _ func()) error { return cb(m) } diff --git a/routing/pathfind_test.go b/routing/pathfind_test.go index fd5f06656..2f746ac9e 100644 --- a/routing/pathfind_test.go +++ b/routing/pathfind_test.go @@ -3303,6 +3303,8 @@ func dbFindPath(graph *graphdb.ChannelGraph, ) return err + }, func() { + route = nil }) if err != nil { return nil, err diff --git a/routing/payment_session.go b/routing/payment_session.go index 607c10443..cf20251ed 100644 --- a/routing/payment_session.go +++ b/routing/payment_session.go @@ -343,7 +343,11 @@ func (p *paymentSession) RequestRoute(maxAmt, feeLimit lnwire.MilliSatoshi, } for { - err := p.graphSessFactory.GraphSession(findPath) + err := p.graphSessFactory.GraphSession( + findPath, func() { + path = nil + }, + ) // If there is an error, and it is not a path finding error, we // return it immediately. if err != nil && !lnutils.ErrorAs[*pathFindingError](err) { diff --git a/routing/payment_session_test.go b/routing/payment_session_test.go index 51eeda425..ffe9d996c 100644 --- a/routing/payment_session_test.go +++ b/routing/payment_session_test.go @@ -259,8 +259,8 @@ func (g *sessionGraph) sourceNode() route.Vertex { return route.Vertex{} } -func (g *sessionGraph) GraphSession( - cb func(graph graphdb.NodeTraverser) error) error { +func (g *sessionGraph) GraphSession(cb func(graph graphdb.NodeTraverser) error, + _ func()) error { return cb(g) } From 13d0bd5ef69ac63b3c90f3b19ad89e0c951ed92e Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 11:03:22 +0200 Subject: [PATCH 10/11] graph/db: let ForEachChannelCacheable take a reset param --- graph/db/graph.go | 3 ++- graph/db/interfaces.go | 4 ++-- graph/db/kv_store.go | 5 ++--- graph/db/sql_store.go | 5 ++--- 4 files changed, 8 insertions(+), 9 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index c0a4f52b6..e902896d7 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -182,7 +182,8 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { c.graphCache.AddChannel(info, policy1, policy2) return nil - }) + }, func() {}, + ) if err != nil { return err } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index f4cbb475d..756d7b1f2 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -177,8 +177,8 @@ type V1Store interface { //nolint:interfacebloat // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, - *models.CachedEdgePolicy) error) error + *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, + reset func()) error // DisabledChannelIDs returns the channel ids of disabled channels. // A channel is disabled when two of the associated ChanelEdgePolicies diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 020d7d3d3..8dab73cdf 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -486,9 +486,8 @@ 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(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, *models.CachedEdgePolicy) error) error { - - reset := func() {} + *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, + reset func()) error { return c.db.View(func(tx kvdb.RTx) error { edges := tx.ReadBucket(edgeBucket) diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 58f29a75e..33328c0ba 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1236,10 +1236,9 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, // NOTE: this method is like ForEachChannel but fetches only the data // required for the graph cache. func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, - *models.CachedEdgePolicy, - *models.CachedEdgePolicy) error) error { + *models.CachedEdgePolicy, *models.CachedEdgePolicy) error, + reset func()) error { - reset := func() {} ctx := context.TODO() handleChannel := func(db SQLQueries, From 83f6dc4983e43a91fdff8326244446649070d127 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 11:20:25 +0200 Subject: [PATCH 11/11] multi: reset for ForEachNodeDirectedChannel --- graph/db/graph.go | 4 ++-- graph/db/graph_test.go | 6 +++--- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 6 ++---- graph/db/sql_store.go | 5 ++--- routing/bandwidth.go | 10 ++++++---- routing/graph.go | 3 ++- routing/mock_graph_test.go | 2 +- routing/pathfind.go | 31 +++++++++++++++++++++++++------ routing/unified_edges.go | 22 ++++++++++++++++++---- 10 files changed, 62 insertions(+), 29 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index e902896d7..a50715774 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -205,13 +205,13 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { // // NOTE: this is part of the graphdb.NodeTraverser interface. func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex, - cb func(channel *DirectedChannel) error) error { + cb func(channel *DirectedChannel) error, reset func()) error { if c.graphCache != nil { return c.graphCache.ForEachChannel(node, cb) } - return c.V1Store.ForEachNodeDirectedChannel(node, cb) + return c.V1Store.ForEachNodeDirectedChannel(node, cb, reset) } // FetchNodeFeatures returns the features of the given node. If no features are diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index d2f1577b2..25ce1edf7 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1490,7 +1490,7 @@ func TestGraphTraversalCacheable(t *testing.T) { node, func(d *DirectedChannel) error { delete(chanIndex, d.ChannelID) return nil - }, + }, func() {}, ) require.NoError(t, err) @@ -1499,7 +1499,7 @@ func TestGraphTraversalCacheable(t *testing.T) { node, func(d *DirectedChannel) error { delete(chanIndex2, d.ChannelID) return nil - }, + }, func() {}, ) require.NoError(t, err) } @@ -4298,7 +4298,7 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { ch = c return nil - }, + }, func() {}, ) require.NoError(t, err) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 756d7b1f2..dbcd5d1e6 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -40,7 +40,7 @@ type NodeTraverser interface { // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error) error + cb func(channel *DirectedChannel) error, reset func()) error // FetchNodeFeatures returns the features of the given node. FetchNodeFeatures(nodePub route.Vertex) (*lnwire.FeatureVector, error) diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 8dab73cdf..7bac0ec13 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -655,9 +655,7 @@ func (c *KVStore) fetchNodeFeatures(tx kvdb.RTx, // // NOTE: this is part of the graphdb.NodeTraverser interface. func (c *KVStore) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error) error { - - reset := func() {} + cb func(channel *DirectedChannel) error, reset func()) error { return c.forEachNodeDirectedChannel(nil, nodePub, cb, reset) } @@ -3999,7 +3997,7 @@ type nodeTraverserSession struct { // // NOTE: Part of the NodeTraverser interface. func (c *nodeTraverserSession) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error) error { + cb func(channel *DirectedChannel) error, _ func()) error { return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb, func() {}) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 33328c0ba..cd5031f06 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -907,9 +907,8 @@ func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { // // NOTE: this is part of the graphdb.NodeTraverser interface. func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error) error { + cb func(channel *DirectedChannel) error, reset func()) error { - reset := func() {} var ctx = context.TODO() return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { @@ -2792,7 +2791,7 @@ func newSQLNodeTraverser(db SQLQueries, // // NOTE: Part of the NodeTraverser interface. func (s *sqlNodeTraverser) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error) error { + cb func(channel *DirectedChannel) error, _ func()) error { ctx := context.TODO() diff --git a/routing/bandwidth.go b/routing/bandwidth.go index f806b03d3..afe085c2e 100644 --- a/routing/bandwidth.go +++ b/routing/bandwidth.go @@ -63,16 +63,18 @@ func newBandwidthManager(graph Graph, sourceNode route.Vertex, // First, we'll collect the set of outbound edges from the target // source node and add them to our bandwidth manager's map of channels. - err := graph.ForEachNodeDirectedChannel(sourceNode, - func(channel *graphdb.DirectedChannel) error { + err := graph.ForEachNodeDirectedChannel( + sourceNode, func(channel *graphdb.DirectedChannel) error { shortID := lnwire.NewShortChanIDFromInt( channel.ChannelID, ) manager.localChans[shortID] = struct{}{} return nil - }) - + }, func() { + clear(manager.localChans) + }, + ) if err != nil { return nil, err } diff --git a/routing/graph.go b/routing/graph.go index d5c6b2d7d..4be34def9 100644 --- a/routing/graph.go +++ b/routing/graph.go @@ -15,7 +15,8 @@ type Graph interface { // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *graphdb.DirectedChannel) error) error + cb func(channel *graphdb.DirectedChannel) error, + reset func()) error // FetchNodeFeatures returns the features of the given node. FetchNodeFeatures(nodePub route.Vertex) (*lnwire.FeatureVector, error) diff --git a/routing/mock_graph_test.go b/routing/mock_graph_test.go index b5c8b55bc..d9b16f646 100644 --- a/routing/mock_graph_test.go +++ b/routing/mock_graph_test.go @@ -166,7 +166,7 @@ func (m *mockGraph) addChannel(id uint64, node1id, node2id byte, // // NOTE: Part of the Graph interface. func (m *mockGraph) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *graphdb.DirectedChannel) error) error { + cb func(channel *graphdb.DirectedChannel) error, _ func()) error { // Look up the mock node. node, ok := m.nodes[nodePub] diff --git a/routing/pathfind.go b/routing/pathfind.go index b22e7fd1c..fab8015dd 100644 --- a/routing/pathfind.go +++ b/routing/pathfind.go @@ -576,7 +576,12 @@ func getOutgoingBalance(node route.Vertex, outgoingChans map[uint64]struct{}, } // Iterate over all channels of the to node. - err := g.ForEachNodeDirectedChannel(node, cb) + err := g.ForEachNodeDirectedChannel( + node, cb, func() { + max = 0 + total = 0 + }, + ) if err != nil { return 0, 0, err } @@ -1276,7 +1281,12 @@ func findBlindedPaths(g Graph, target route.Vertex, // it later on when we call "processNodeForBlindedPath". visited[nextTarget] = true - err := g.ForEachNodeDirectedChannel(nextTarget, + var ( + incomingPathReset []blindedHop + nextTargetReset = nextTarget + ) + err := g.ForEachNodeDirectedChannel( + nextTarget, func(channel *graphdb.DirectedChannel) error { // This is not the right channel, continue to // the node's other channels. @@ -1286,18 +1296,21 @@ func findBlindedPaths(g Graph, target route.Vertex, // We found the channel in question. Prepend it // to the incoming path. - incomingPath = append([]blindedHop{ + incomingPathReset = append([]blindedHop{ { vertex: channel.OtherNode, channelID: channel.ChannelID, edgeCapacity: channel.Capacity, }, - }, incomingPath...) + }, incomingPathReset...) // Update the target node. - nextTarget = channel.OtherNode + nextTargetReset = channel.OtherNode return errChanFound + }, func() { + incomingPathReset = nil + nextTargetReset = nextTarget }, ) // We expect errChanFound to be returned if the channel in @@ -1308,6 +1321,8 @@ func findBlindedPaths(g Graph, target route.Vertex, return nil, fmt.Errorf("incoming channel %d is not "+ "seen as owned by node %v", chanID, nextTarget) } + nextTarget = nextTargetReset + incomingPath = append(incomingPathReset, incomingPath...) // Check that the user didn't accidentally add a channel that // is owned by a node in the node omission set. @@ -1445,7 +1460,8 @@ func processNodeForBlindedPath(g Graph, node route.Vertex, // Now, iterate over the node's channels in search for paths to this // node that can be used for blinded paths - err = g.ForEachNodeDirectedChannel(node, + err = g.ForEachNodeDirectedChannel( + node, func(channel *graphdb.DirectedChannel) error { // Keep track of how many incoming channels this node // has. We only use a node as an introduction node if it @@ -1488,6 +1504,9 @@ func processNodeForBlindedPath(g Graph, node route.Vertex, } return nil + }, func() { + hopSets = nil + chanCount = 0 }, ) if err != nil { diff --git a/routing/unified_edges.go b/routing/unified_edges.go index bf2f75dbf..f80e1cb1a 100644 --- a/routing/unified_edges.go +++ b/routing/unified_edges.go @@ -97,6 +97,7 @@ func (u *nodeEdgeUnifier) addPolicy(fromNode route.Vertex, // addGraphPolicies adds all policies that are known for the toNode in the // graph. func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error { + var channels []*graphdb.DirectedChannel cb := func(channel *graphdb.DirectedChannel) error { // If there is no edge policy for this candidate node, skip. // Note that we are searching backwards so this node would have @@ -108,6 +109,22 @@ func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error { return nil } + channels = append(channels, channel) + + return nil + } + + // Iterate over all channels of the to node. + err := g.ForEachNodeDirectedChannel( + u.toNode, cb, func() { + channels = nil + }, + ) + if err != nil { + return err + } + + for _, channel := range channels { // Add this policy to the corresponding edgeUnifier. We default // to the clear hop payload size function because // `addGraphPolicies` is only used for cleartext intermediate @@ -120,12 +137,9 @@ func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error { channel.OtherNode, channel.InPolicy, inboundFee, channel.Capacity, defaultHopPayloadSize, nil, ) - - return nil } - // Iterate over all channels of the to node. - return g.ForEachNodeDirectedChannel(u.toNode, cb) + return nil } // unifiedEdge is the individual channel data that is kept inside an edgeUnifier