From 67c0f54e13581aca58ad36c3f376c134158efca4 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 7 Jul 2025 09:22:08 +0200 Subject: [PATCH] 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)