graph/db+server: let ForEachSourceNodeChannel take a context

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)

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,
+	ForEachSourceNodeChannel(ctx context.Context,
-		havePolicy bool, otherNode *models.LightningNode) error) error
+		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

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,
+func (c *KVStore) ForEachSourceNodeChannel(_ context.Context,
-	havePolicy bool, otherNode *models.LightningNode) error) error {
+	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)

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,
+func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
-	havePolicy bool, otherNode *models.LightningNode) error) error {
+	cb func(chanPoint wire.OutPoint, havePolicy bool,
-
+		otherNode *models.LightningNode) error) error {
-	var ctx = context.TODO()
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		nodeID, nodePub, err := s.getSourceNode(ctx, db, ProtocolV1)

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)