graph/db: let ForEachNode take a context

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.

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

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

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

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 {
 

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

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 {
+func (s *SQLStore) ForEachNode(ctx context.Context,
-	var (
+	cb func(tx NodeRTx) error) error {
-		ctx          = context.TODO()
-		lastID int64 = 0
-	)
 
+	var lastID int64 = 0
 	handleNode := func(db SQLQueries, dbNode sqlc.Node) error {
 		node, err := buildNode(ctx, db, &dbNode)
 		if err != nil {

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)