From 6295211d1ad7f06ef0f39d746d8723e608d46212 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:13:34 +0200 Subject: [PATCH 01/10] server: use errors.Is for error comparison --- server.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server.go b/server.go index 3626158fd..519471319 100644 --- a/server.go +++ b/server.go @@ -3604,7 +3604,7 @@ func (s *server) establishPersistentConnections() error { // attempt to connect to based on our set of previous connections. Set // the reconnection port to the default peer port. linkNodes, err := s.chanStateDB.LinkNodeDB().FetchAllLinkNodes() - if err != nil && err != channeldb.ErrLinkNodesNotFound { + if err != nil && !errors.Is(err, channeldb.ErrLinkNodesNotFound) { return fmt.Errorf("failed to fetch all link nodes: %w", err) } From 792970e50c2df59b4a64a72b5a29bd5c7c73ba38 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:16:53 +0200 Subject: [PATCH 02/10] server: create call-back helper forEachSrcNodeChan We do this so that in an upcoming commit, we can expand the param list of the call-back without adding another indent to the call-back code. --- server.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/server.go b/server.go index 519471319..2d39d182e 100644 --- a/server.go +++ b/server.go @@ -3622,7 +3622,7 @@ func (s *server) establishPersistentConnections() error { // that have been added via NodeAnnouncement messages. // TODO(roasbeef): instead iterate over link nodes and query graph for // each of the nodes. - err = s.graphDB.ForEachSourceNodeChannel(func(chanPoint wire.OutPoint, + forEachSrcNodeChan := func(chanPoint wire.OutPoint, havePolicy bool, channelPeer *models.LightningNode) error { // If the remote party has announced the channel to us, but we @@ -3666,6 +3666,7 @@ func (s *server) establishPersistentConnections() error { // addresses if Tor outbound support is enabled. case *tor.OnionAddr: if s.cfg.Tor.Active { + //nolint:ll addrSet[lnAddress.String()] = lnAddress } } @@ -3688,7 +3689,8 @@ func (s *server) establishPersistentConnections() error { nodeAddrsMap[pubStr] = n return nil - }) + } + err = s.graphDB.ForEachSourceNodeChannel(forEachSrcNodeChan) if err != nil { srvrLog.Errorf("Failed to iterate over source node channels: "+ "%v", err) From 67c0f54e13581aca58ad36c3f376c134158efca4 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:22:08 +0200 Subject: [PATCH 03/10] graph/db+server: let ForEachSourceNodeChannel take a context --- graph/db/graph_test.go | 2 +- graph/db/interfaces.go | 5 +++-- graph/db/kv_store.go | 5 +++-- graph/db/sql_store.go | 7 +++---- server.go | 6 +++--- 5 files changed, 13 insertions(+), 12 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index fefc125c4..33d576df5 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1311,7 +1311,7 @@ func TestForEachSourceNodeChannel(t *testing.T) { // Now, we'll use the ForEachSourceNodeChannel and assert that it // returns the expected data in the call-back. - err := graph.ForEachSourceNodeChannel(func(chanPoint wire.OutPoint, + err := graph.ForEachSourceNodeChannel(ctx, func(chanPoint wire.OutPoint, havePolicy bool, otherNode *models.LightningNode) error { require.Contains(t, expectedSrcChans, chanPoint) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 7140553b7..e04b5ff59 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -70,8 +70,9 @@ type V1Store interface { //nolint:interfacebloat // node, executing the passed callback on each. The call-back is // provided with the channel's outpoint, whether we have a policy for // the channel and the channel peer's node information. - ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint, - havePolicy bool, otherNode *models.LightningNode) error) error + ForEachSourceNodeChannel(ctx context.Context, + cb func(chanPoint wire.OutPoint, havePolicy bool, + otherNode *models.LightningNode) error) 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 42b9f9245..d685695f4 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3248,8 +3248,9 @@ func (c *KVStore) ForEachNodeChannel(nodePub route.Vertex, // executing the passed callback on each. The callback is provided with the // channel's outpoint, whether we have a policy for the channel and the channel // peer's node information. -func (c *KVStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint, - havePolicy bool, otherNode *models.LightningNode) error) error { +func (c *KVStore) ForEachSourceNodeChannel(_ context.Context, + cb func(chanPoint wire.OutPoint, havePolicy bool, + otherNode *models.LightningNode) error) 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 c39bc6bbd..302e08b9e 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -724,10 +724,9 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy, // peer's node information. // // NOTE: part of the V1Store interface. -func (s *SQLStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint, - havePolicy bool, otherNode *models.LightningNode) error) error { - - var ctx = context.TODO() +func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, + cb func(chanPoint wire.OutPoint, havePolicy bool, + otherNode *models.LightningNode) error) 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 2d39d182e..7ca8e8ea8 100644 --- a/server.go +++ b/server.go @@ -2607,7 +2607,7 @@ func (s *server) Start(ctx context.Context) error { return } - if err := s.establishPersistentConnections(); err != nil { + if err := s.establishPersistentConnections(ctx); err != nil { srvrLog.Errorf("Failed to establish persistent "+ "connections: %v", err) } @@ -3594,7 +3594,7 @@ type nodeAddresses struct { // to all our direct channel collaborators. In order to promote liveness of our // active channels, we instruct the connection manager to attempt to establish // and maintain persistent connections to all our direct channel counterparties. -func (s *server) establishPersistentConnections() error { +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. @@ -3690,7 +3690,7 @@ func (s *server) establishPersistentConnections() error { nodeAddrsMap[pubStr] = n return nil } - err = s.graphDB.ForEachSourceNodeChannel(forEachSrcNodeChan) + err = s.graphDB.ForEachSourceNodeChannel(ctx, forEachSrcNodeChan) if err != nil { srvrLog.Errorf("Failed to iterate over source node channels: "+ "%v", err) From f8d3a3f3ba7629dea31e7b8120ed6909282b455d Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:30:28 +0200 Subject: [PATCH 04/10] graph/db: let ForEachNodeChannel take a context --- autopilot/prefattach_test.go | 10 +++++----- graph/builder.go | 4 +++- graph/db/graph_test.go | 8 +++++--- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 2 +- graph/db/sql_store.go | 4 +--- graph/interfaces.go | 2 +- rpcserver.go | 4 ++-- server.go | 4 +++- 9 files changed, 22 insertions(+), 18 deletions(-) diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index ba79d9682..fd557a02c 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -731,12 +731,12 @@ func (t *testNodeTx) Node() *models.LightningNode { func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - return t.db.db.ForEachNodeChannel(t.node.PubKeyBytes, func( - edge *models.ChannelEdgeInfo, policy1, - policy2 *models.ChannelEdgePolicy) error { + return t.db.db.ForEachNodeChannel(context.Background(), + t.node.PubKeyBytes, func(edge *models.ChannelEdgeInfo, policy1, + policy2 *models.ChannelEdgePolicy) error { - return f(edge, policy1, policy2) - }) + return f(edge, policy1, policy2) + }) } func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) { diff --git a/graph/builder.go b/graph/builder.go index df4f3bfdf..d590fb896 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1278,7 +1278,9 @@ func (b *Builder) FetchLightningNode(ctx context.Context, func (b *Builder) ForAllOutgoingChannels(cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy) error) error { - return b.cfg.Graph.ForEachNodeChannel(b.cfg.SelfNode, + ctx := context.TODO() + + return b.cfg.Graph.ForEachNodeChannel(ctx, b.cfg.SelfNode, func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 33d576df5..b1206281d 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1332,6 +1332,7 @@ func TestForEachSourceNodeChannel(t *testing.T) { func TestGraphTraversal(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -1387,7 +1388,7 @@ func TestGraphTraversal(t *testing.T) { // outgoing channels for a particular node. numNodeChans := 0 firstNode, secondNode := nodeList[0], nodeList[1] - err = graph.ForEachNodeChannel(firstNode.PubKeyBytes, + err = graph.ForEachNodeChannel(ctx, firstNode.PubKeyBytes, func(_ *models.ChannelEdgeInfo, outEdge, inEdge *models.ChannelEdgePolicy) error { @@ -3138,7 +3139,7 @@ func TestIncompleteChannelPolicies(t *testing.T) { expectedOut bool) { calls := 0 - err := graph.ForEachNodeChannel(node.PubKeyBytes, + err := graph.ForEachNodeChannel(ctx, node.PubKeyBytes, func(_ *models.ChannelEdgeInfo, outEdge, inEdge *models.ChannelEdgePolicy) error { @@ -4204,6 +4205,7 @@ func TestBatchedUpdateEdgePolicy(t *testing.T) { // allocations and the total memory consumed by the full graph traversal. func BenchmarkForEachChannel(b *testing.B) { graph := MakeTestGraph(b) + ctx := context.Background() const numNodes = 100 const numChannels = 4 @@ -4244,7 +4246,7 @@ func BenchmarkForEachChannel(b *testing.B) { return nil } - err := graph.ForEachNodeChannel(n, cb) + err := graph.ForEachNodeChannel(ctx, n, cb) require.NoError(b, err) } } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index e04b5ff59..b43e3b004 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -83,7 +83,7 @@ type V1Store interface { //nolint:interfacebloat // to the caller. // // Unknown policies are passed into the callback as nil values. - ForEachNodeChannel(nodePub route.Vertex, + ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index d685695f4..8fb330e81 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -3232,7 +3232,7 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend, // halted with the error propagated back up to the caller. // // Unknown policies are passed into the callback as nil values. -func (c *KVStore) ForEachNodeChannel(nodePub route.Vertex, +func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 302e08b9e..bc9064252 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -958,12 +958,10 @@ func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex, // Unknown policies are passed into the callback as nil values. // // NOTE: part of the V1Store interface. -func (s *SQLStore) ForEachNodeChannel(nodePub route.Vertex, +func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - var ctx = context.TODO() - return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { dbNode, err := db.GetNodeByPubKey( ctx, sqlc.GetNodeByPubKeyParams{ diff --git a/graph/interfaces.go b/graph/interfaces.go index be226a495..9d576e4bc 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -257,7 +257,7 @@ type DB interface { // to the caller. // // Unknown policies are passed into the callback as nil values. - ForEachNodeChannel(nodePub route.Vertex, + ForEachNodeChannel(ctx context.Context, nodePub route.Vertex, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error diff --git a/rpcserver.go b/rpcserver.go index 87bda9633..5b39c47ca 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -7020,7 +7020,7 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context, channels []*lnrpc.ChannelEdge ) - err = graph.ForEachNodeChannel(node.PubKeyBytes, + err = graph.ForEachNodeChannel(ctx, node.PubKeyBytes, func(edge *models.ChannelEdgeInfo, c1, c2 *models.ChannelEdgePolicy) error { @@ -7702,7 +7702,7 @@ func (r *rpcServer) FeeReport(ctx context.Context, } var feeReports []*lnrpc.ChannelFeeReport - err = channelGraph.ForEachNodeChannel(selfNode.PubKeyBytes, + err = channelGraph.ForEachNodeChannel(ctx, selfNode.PubKeyBytes, func(chanInfo *models.ChannelEdgeInfo, edgePolicy, _ *models.ChannelEdgePolicy) error { diff --git a/server.go b/server.go index 7ca8e8ea8..729490e25 100644 --- a/server.go +++ b/server.go @@ -1255,7 +1255,9 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, ForAllOutgoingChannels: func(cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy) error) error { - return s.graphDB.ForEachNodeChannel(selfVertex, + ctx := context.TODO() + + return s.graphDB.ForEachNodeChannel(ctx, selfVertex, func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { From 046e106a17bf47700f5acaf4d1d8dded144d7d72 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:32:59 +0200 Subject: [PATCH 05/10] graph/db: let ForEachChannel take a context --- graph/db/graph_test.go | 34 ++++++++++++++++++++-------------- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 5 +++-- graph/db/sql_store.go | 7 +++---- rpcserver.go | 2 +- 5 files changed, 28 insertions(+), 22 deletions(-) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index b1206281d..b63651188 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -922,15 +922,20 @@ func TestEdgePolicyCRUD(t *testing.T) { // Use the ForEachChannel method to fetch the policies and // assert that the deserialized policies match the original // ones. - err := graph.ForEachChannel(func(info *models.ChannelEdgeInfo, - policy1 *models.ChannelEdgePolicy, - policy2 *models.ChannelEdgePolicy) error { + err := graph.ForEachChannel( + ctx, func(info *models.ChannelEdgeInfo, + policy1 *models.ChannelEdgePolicy, + policy2 *models.ChannelEdgePolicy) error { - require.NoError(t, compareEdgePolicies(edge1, policy1)) - require.NoError(t, compareEdgePolicies(edge2, policy2)) + require.NoError( + t, compareEdgePolicies(edge1, policy1), + ) + require.NoError( + t, compareEdgePolicies(edge2, policy2), + ) - return nil - }) + return nil + }) require.NoError(t, err) } @@ -1374,7 +1379,7 @@ func TestGraphTraversal(t *testing.T) { // Iterate through all the known channels within the graph DB, once // again if the map is empty that indicates that all edges have // properly been reached. - err = graph.ForEachChannel(func(ei *models.ChannelEdgeInfo, + err = graph.ForEachChannel(ctx, func(ei *models.ChannelEdgeInfo, _ *models.ChannelEdgePolicy, _ *models.ChannelEdgePolicy) error { @@ -1663,13 +1668,14 @@ func assertPruneTip(t *testing.T, graph *ChannelGraph, func assertNumChans(t *testing.T, graph *ChannelGraph, n int) { numChans := 0 - if err := graph.ForEachChannel(func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error { + if err := graph.ForEachChannel(context.Background(), + func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error { - numChans++ - return nil - }); err != nil { + numChans++ + return nil + }, + ); err != nil { _, _, line, _ := runtime.Caller(1) t.Fatalf("line %v: unable to scan channels: %v", line, err) } diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index b43e3b004..e46e578a9 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -158,7 +158,7 @@ type V1Store interface { //nolint:interfacebloat // NOTE: If an edge can't be found, or wasn't advertised, then a nil // pointer for that particular channel edge routing policy will be // passed into the callback. - ForEachChannel(cb func(*models.ChannelEdgeInfo, + ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 8fb330e81..02e6a4ab6 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -404,8 +404,9 @@ func (c *KVStore) AddrsForNode(ctx context.Context, // NOTE: If an edge can't be found, or wasn't advertised, then a nil pointer // for that particular channel edge routing policy will be passed into the // callback. -func (c *KVStore) ForEachChannel(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { +func (c *KVStore) ForEachChannel(_ context.Context, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error) error { return forEachChannel(c.db, cb) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index bc9064252..c8f80c9f8 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1333,10 +1333,9 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo, // callback. // // NOTE: part of the V1Store interface. -func (s *SQLStore) ForEachChannel(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - - ctx := context.TODO() +func (s *SQLStore) ForEachChannel(ctx context.Context, + cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, + *models.ChannelEdgePolicy) error) error { handleChannel := func(db SQLQueries, row sqlc.ListChannelsWithPoliciesPaginatedRow) error { diff --git a/rpcserver.go b/rpcserver.go index 5b39c47ca..a092dda8d 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6730,7 +6730,7 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, // Next, for each active channel we know of within the graph, create a // similar response which details both the edge information as well as // the routing policies of th nodes connecting the two edges. - err = graph.ForEachChannel(func(edgeInfo *models.ChannelEdgeInfo, + err = graph.ForEachChannel(ctx, func(edgeInfo *models.ChannelEdgeInfo, c1, c2 *models.ChannelEdgePolicy) error { // Do not include unannounced channels unless specifically From 85fda8b926836c0577390c46512fee1538531e84 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:35:03 +0200 Subject: [PATCH 06/10] routing+server: let ForAllOutgoingChannels take a context And so, remove a previously added context.TODO --- routing/localchans/manager.go | 7 ++++--- routing/localchans/manager_test.go | 5 +++-- server.go | 7 +++---- 3 files changed, 10 insertions(+), 9 deletions(-) diff --git a/routing/localchans/manager.go b/routing/localchans/manager.go index c76bad3a6..faba9fa8b 100644 --- a/routing/localchans/manager.go +++ b/routing/localchans/manager.go @@ -40,8 +40,9 @@ type Manager struct { // ForAllOutgoingChannels is required to iterate over all our local // channels. The ChannelEdgePolicy parameter may be nil. - ForAllOutgoingChannels func(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error + ForAllOutgoingChannels func(ctx context.Context, + cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy) error) error // FetchChannel is used to query local channel parameters. Optionally an // existing db tx can be supplied. @@ -151,7 +152,7 @@ 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(processChan) + err := r.ForAllOutgoingChannels(ctx, processChan) if err != nil { return nil, err } diff --git a/routing/localchans/manager_test.go b/routing/localchans/manager_test.go index d97b2552d..6e53f4a27 100644 --- a/routing/localchans/manager_test.go +++ b/routing/localchans/manager_test.go @@ -123,8 +123,9 @@ func TestManager(t *testing.T) { return nil } - forAllOutgoingChannels := func(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { + forAllOutgoingChannels := func(_ context.Context, + cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy) error) error { for _, c := range channelSet { if err := cb(c.edgeInfo, ¤tPolicy); err != nil { diff --git a/server.go b/server.go index 729490e25..5c0b2a77c 100644 --- a/server.go +++ b/server.go @@ -1252,10 +1252,9 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr, s.localChanMgr = &localchans.Manager{ SelfPub: nodeKeyDesc.PubKey, DefaultRoutingPolicy: cc.RoutingPolicy, - ForAllOutgoingChannels: func(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { - - ctx := context.TODO() + ForAllOutgoingChannels: func(ctx context.Context, + cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy) error) error { return s.graphDB.ForEachNodeChannel(ctx, selfVertex, func(c *models.ChannelEdgeInfo, From f05ef2db97a1f287190fe93010aaae41b653e116 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:45:00 +0200 Subject: [PATCH 07/10] graph/db: let ForEachNodeCacheable take a context --- graph/db/graph.go | 6 +++--- graph/db/graph_test.go | 5 +++-- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 4 ++-- graph/db/sql_store.go | 6 ++---- 5 files changed, 11 insertions(+), 12 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index 922adef53..86c14a6cf 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -75,7 +75,7 @@ func (c *ChannelGraph) Start() error { defer log.Debug("ChannelGraph started") if c.graphCache != nil { - if err := c.populateCache(); err != nil { + if err := c.populateCache(context.TODO()); err != nil { return fmt.Errorf("could not populate the graph "+ "cache: %w", err) } @@ -159,12 +159,12 @@ func (c *ChannelGraph) handleTopologySubscriptions() { // 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() error { +func (c *ChannelGraph) populateCache(ctx context.Context) error { startTime := time.Now() log.Info("Populating in-memory channel graph, this might take a " + "while...") - err := c.V1Store.ForEachNodeCacheable(func(node route.Vertex, + err := c.V1Store.ForEachNodeCacheable(ctx, func(node route.Vertex, features *lnwire.FeatureVector) error { c.graphCache.AddNodeFeatures(node, features) diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index b63651188..e3fd19456 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1433,6 +1433,7 @@ func TestGraphTraversal(t *testing.T) { // working correctly. func TestGraphTraversalCacheable(t *testing.T) { t.Parallel() + ctx := context.Background() graph := MakeTestGraph(t) @@ -1458,7 +1459,7 @@ func TestGraphTraversalCacheable(t *testing.T) { // iterating over each node, once again if the map is empty that // indicates that all edges have properly been reached. var nodes []route.Vertex - err = graph.ForEachNodeCacheable(func(node route.Vertex, + err = graph.ForEachNodeCacheable(ctx, func(node route.Vertex, features *lnwire.FeatureVector) error { delete(nodeMap, node) @@ -4226,7 +4227,7 @@ func BenchmarkForEachChannel(b *testing.B) { ) var nodes []route.Vertex - err := graph.ForEachNodeCacheable(func(node route.Vertex, + err := graph.ForEachNodeCacheable(ctx, func(node route.Vertex, vector *lnwire.FeatureVector) error { nodes = append(nodes, node) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index e46e578a9..8d2948478 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -107,7 +107,7 @@ type V1Store interface { //nolint:interfacebloat // in the 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. - ForEachNodeCacheable(cb func(route.Vertex, + ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex, *lnwire.FeatureVector) error) error // LookupAlias attempts to return the alias as advertised by the target diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 02e6a4ab6..9d92aaaca 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -842,8 +842,8 @@ 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(cb func(route.Vertex, - *lnwire.FeatureVector) error) error { +func (c *KVStore) ForEachNodeCacheable(_ context.Context, + cb func(route.Vertex, *lnwire.FeatureVector) error) 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 c8f80c9f8..825013413 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -923,10 +923,8 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, // stops early. // // NOTE: This is a part of the V1Store interface. -func (s *SQLStore) ForEachNodeCacheable(cb func(route.Vertex, - *lnwire.FeatureVector) error) error { - - ctx := context.TODO() +func (s *SQLStore) ForEachNodeCacheable(ctx context.Context, + cb func(route.Vertex, *lnwire.FeatureVector) error) error { err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodeCacheable(ctx, db, func(nodeID int64, From 1e804a3d5f0aa1e195c131d10d74e95028204c8c Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:46:23 +0200 Subject: [PATCH 08/10] graph/db: let ForEachNode take a context --- autopilot/graph.go | 2 +- autopilot/interface.go | 2 +- graph/db/graph_test.go | 6 +++--- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 4 +++- graph/db/sql_migration_test.go | 2 +- graph/db/sql_store.go | 8 +++----- rpcserver.go | 2 +- 8 files changed, 14 insertions(+), 14 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index d20c6316e..d50871591 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -123,7 +123,7 @@ func (d *dbNode) ForEachChannel(ctx context.Context, func (d *databaseChannelGraph) ForEachNode(ctx context.Context, cb func(context.Context, Node) error) error { - return d.db.ForEachNode(func(nodeTx graphdb.NodeRTx) error { + return d.db.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error { // We'll skip over any node that doesn't have any advertised // addresses. As we won't be able to reach them to actually // open any channels. diff --git a/autopilot/interface.go b/autopilot/interface.go index db25d9bb2..594f2ece5 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -228,7 +228,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(func(graphdb.NodeRTx) error) error + ForEachNode(context.Context, func(graphdb.NodeRTx) error) 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 e3fd19456..828d49697 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1447,7 +1447,7 @@ func TestGraphTraversalCacheable(t *testing.T) { // Create a map of all nodes with the iteration we know works (because // it is tested in another test). nodeMap := make(map[route.Vertex]struct{}) - err := graph.ForEachNode(func(tx NodeRTx) error { + err := graph.ForEachNode(ctx, func(tx NodeRTx) error { nodeMap[tx.Node().PubKeyBytes] = struct{}{} return nil @@ -1577,7 +1577,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, // Iterate over each node as returned by the graph, if all nodes are // reached, then the map created above should be empty. - err := graph.ForEachNode(func(tx NodeRTx) error { + err := graph.ForEachNode(ctx, func(tx NodeRTx) error { delete(nodeIndex, tx.Node().Alias) return nil }) @@ -1689,7 +1689,7 @@ func assertNumChans(t *testing.T, graph *ChannelGraph, n int) { func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) { numNodes := 0 - err := graph.ForEachNode(func(tx NodeRTx) error { + err := graph.ForEachNode(context.Background(), func(tx NodeRTx) error { numNodes++ return nil diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 8d2948478..fb5b88c99 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -101,7 +101,7 @@ 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(cb func(tx NodeRTx) error) error + ForEachNode(ctx context.Context, cb func(tx NodeRTx) error) 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 9d92aaaca..e19d80256 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -789,7 +789,9 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) { // early. Any operations performed on the NodeTx 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. -func (c *KVStore) ForEachNode(cb func(tx NodeRTx) error) error { +func (c *KVStore) ForEachNode(_ context.Context, + cb func(tx NodeRTx) error) 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 8e0df1c90..a8cd1c157 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -181,7 +181,7 @@ func assertInSync(t *testing.T, kvDB *KVStore, sqlDB *SQLStore, func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode { nodes := make([]*models.LightningNode, 0) - err := store.ForEachNode(func(tx NodeRTx) error { + err := store.ForEachNode(context.Background(), func(tx NodeRTx) error { node := tx.Node() // Call PubKey to ensure the objects cached pubkey is set so that diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 825013413..cf707579f 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -783,12 +783,10 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, // _MUST_ only be called from within the call-back. // // NOTE: part of the V1Store interface. -func (s *SQLStore) ForEachNode(cb func(tx NodeRTx) error) error { - var ( - ctx = context.TODO() - lastID int64 = 0 - ) +func (s *SQLStore) ForEachNode(ctx context.Context, + cb func(tx NodeRTx) error) error { + var lastID int64 = 0 handleNode := func(db SQLQueries, dbNode sqlc.Node) error { node, err := buildNode(ctx, db, &dbNode) if err != nil { diff --git a/rpcserver.go b/rpcserver.go index a092dda8d..9674d689f 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6716,7 +6716,7 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, // First iterate through all the known nodes (connected or unconnected // within the graph), collating their current state into the RPC // response. - err := graph.ForEachNode(func(nodeTx graphdb.NodeRTx) error { + err := graph.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error { lnNode := marshalNode(nodeTx.Node()) resp.Nodes = append(resp.Nodes, lnNode) From 3d5e852c8cb5d5f622a44db11bbddd2576f846b1 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:48:39 +0200 Subject: [PATCH 09/10] graph/db: let ForEachNodeCached take a context --- autopilot/graph.go | 2 +- autopilot/interface.go | 2 +- graph/db/graph.go | 7 ++++--- graph/db/graph_test.go | 2 +- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 5 +++-- graph/db/sql_store.go | 7 +++---- rpcserver.go | 2 +- 8 files changed, 15 insertions(+), 14 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index d50871591..3aa3ad055 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -219,7 +219,7 @@ func (nc dbNodeCached) ForEachChannel(ctx context.Context, func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, cb func(context.Context, Node) error) error { - return dc.db.ForEachNodeCached(func(n route.Vertex, + return dc.db.ForEachNodeCached(ctx, func(n route.Vertex, channels map[uint64]*graphdb.DirectedChannel) error { if len(channels) > 0 { diff --git a/autopilot/interface.go b/autopilot/interface.go index 594f2ece5..703f49020 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -234,6 +234,6 @@ type GraphSource interface { // channel graph cache if one is available. It is less consistent than // ForEachNode since any further calls are made across multiple // transactions. - ForEachNodeCached(cb func(node route.Vertex, + ForEachNodeCached(ctx context.Context, cb func(node route.Vertex, chans map[uint64]*graphdb.DirectedChannel) error) error } diff --git a/graph/db/graph.go b/graph/db/graph.go index 86c14a6cf..3168a6618 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -245,14 +245,15 @@ func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error { // graph, executing the passed callback with each node encountered. // // NOTE: The callback contents MUST not be modified. -func (c *ChannelGraph) ForEachNodeCached(cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { +func (c *ChannelGraph) ForEachNodeCached(ctx context.Context, + cb func(node route.Vertex, + chans map[uint64]*DirectedChannel) error) error { if c.graphCache != nil { return c.graphCache.ForEachNode(cb) } - return c.V1Store.ForEachNodeCached(cb) + return c.V1Store.ForEachNodeCached(ctx, cb) } // 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 828d49697..6edce2fb1 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1358,7 +1358,7 @@ func TestGraphTraversal(t *testing.T) { // 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(func(node route.Vertex, + err := graph.ForEachNodeCached(ctx, func(node route.Vertex, chans map[uint64]*DirectedChannel) error { if _, ok := nodeIndex[node]; !ok { diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index fb5b88c99..ecb07ec6d 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -91,7 +91,7 @@ type V1Store interface { //nolint:interfacebloat // DirectedChannel data to the call-back. // // NOTE: The callback contents MUST not be modified. - ForEachNodeCached(cb func(node route.Vertex, + ForEachNodeCached(ctx context.Context, cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error) error // ForEachNode iterates through all the stored vertices/nodes in the diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index e19d80256..9d1fc7cb7 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -667,8 +667,9 @@ func (c *KVStore) FetchNodeFeatures(nodePub route.Vertex) ( // data to the call-back. // // NOTE: The callback contents MUST not be modified. -func (c *KVStore) ForEachNodeCached(cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { +func (c *KVStore) ForEachNodeCached(_ context.Context, + cb func(node route.Vertex, + chans map[uint64]*DirectedChannel) error) 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 cf707579f..a77709b8e 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1102,10 +1102,9 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, // NOTE: The callback contents MUST not be modified. // // NOTE: part of the V1Store interface. -func (s *SQLStore) ForEachNodeCached(cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error) error { - - var ctx = context.TODO() +func (s *SQLStore) ForEachNodeCached(ctx context.Context, + cb func(node route.Vertex, + chans map[uint64]*DirectedChannel) error) 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 9674d689f..f974ea38b 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -7130,7 +7130,7 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, // network, tallying up the total number of nodes, and also gathering // each node so we can measure the graph diameter and degree stats // below. - err := graph.ForEachNodeCached(func(node route.Vertex, + err := graph.ForEachNodeCached(ctx, func(node route.Vertex, edges map[uint64]*graphdb.DirectedChannel) error { // Increment the total number of nodes with each iteration. From 793c1057bbb9f4d5113f46c861301b2b222749b4 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Tue, 8 Jul 2025 11:55:14 +0200 Subject: [PATCH 10/10] graph: remove one context.TODO By threading a context through to the Builder's ForAllOutgoingChannels method. --- discovery/gossiper.go | 2 +- discovery/gossiper_test.go | 7 ++++--- graph/builder.go | 7 +++---- graph/interfaces.go | 5 +++-- 4 files changed, 11 insertions(+), 10 deletions(-) diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 50834a632..70c1caf2e 100644 --- a/discovery/gossiper.go +++ b/discovery/gossiper.go @@ -1709,7 +1709,7 @@ func (d *AuthenticatedGossiper) retransmitStaleAnns(ctx context.Context, havePublicChannels bool edgesToUpdate []updateTuple ) - err := d.cfg.Graph.ForAllOutgoingChannels(func( + err := d.cfg.Graph.ForAllOutgoingChannels(ctx, func( info *models.ChannelEdgeInfo, edge *models.ChannelEdgePolicy) error { diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index a96ea0952..c10bdbbe1 100644 --- a/discovery/gossiper_test.go +++ b/discovery/gossiper_test.go @@ -211,8 +211,9 @@ func (r *mockGraphSource) ForEachNode( return nil } -func (r *mockGraphSource) ForAllOutgoingChannels(cb func( - i *models.ChannelEdgeInfo, c *models.ChannelEdgePolicy) error) error { +func (r *mockGraphSource) ForAllOutgoingChannels(_ context.Context, + cb func(i *models.ChannelEdgeInfo, + c *models.ChannelEdgePolicy) error) error { r.mu.Lock() defer r.mu.Unlock() @@ -3716,7 +3717,7 @@ out: // policy of all of them. const newTimeLockDelta = 100 var edgesToUpdate []EdgeWithInfo - err = ctx.router.ForAllOutgoingChannels(func( + err = ctx.router.ForAllOutgoingChannels(context.Background(), func( info *models.ChannelEdgeInfo, edge *models.ChannelEdgePolicy) error { diff --git a/graph/builder.go b/graph/builder.go index d590fb896..073716a83 100644 --- a/graph/builder.go +++ b/graph/builder.go @@ -1275,10 +1275,9 @@ func (b *Builder) FetchLightningNode(ctx context.Context, // the router. // // NOTE: This method is part of the ChannelGraphSource interface. -func (b *Builder) ForAllOutgoingChannels(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy) error) error { - - ctx := context.TODO() +func (b *Builder) ForAllOutgoingChannels(ctx context.Context, + cb func(*models.ChannelEdgeInfo, + *models.ChannelEdgePolicy) error) error { return b.cfg.Graph.ForEachNodeChannel(ctx, b.cfg.SelfNode, func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy, diff --git a/graph/interfaces.go b/graph/interfaces.go index 9d576e4bc..20b63dd15 100644 --- a/graph/interfaces.go +++ b/graph/interfaces.go @@ -70,8 +70,9 @@ type ChannelGraphSource interface { // ForAllOutgoingChannels is used to iterate over all channels // emanating from the "source" node which is the center of the // star-graph. - ForAllOutgoingChannels(cb func(c *models.ChannelEdgeInfo, - e *models.ChannelEdgePolicy) error) error + ForAllOutgoingChannels(ctx context.Context, + cb func(c *models.ChannelEdgeInfo, + e *models.ChannelEdgePolicy) error) error // CurrentBlockHeight returns the block height from POV of the router // subsystem.