Repository: lightningnetwork/lnd
Pull request #10043 (2025-07-07 07:59:38)

multi: add context.Context param to more graphdb.V1Store methods
Following on from https://github.com/lightningnetwork/lnd/pull/9959...

In preparation for having the ChannelGraph perform some remote RPC calls, we need all its methods to take a context parameter so that those calls are cancellable. This also lets us remove some of the context.TODO() calls that we currently have in the SQLStore method implementations. More will be done in follow up PRs.

This PR threads contexts through to the following methods:

- ForEachSourceNodeChannel
- ForEachNodeChannel
- ForEachChannel
- ForAllOutgoingChannels
- ForEachNodeCacheable
- ForEachNode
- ForEachNodeCached

Wherever we can we thread through existing contexts from call sites. In some places we just use a context.TODO for now where context threading will be more complicated. Removing these context.TODOs can happen in later PRs that are dedicated to adding contexts to the subsystems in question.


Patch:
diff --git a/autopilot/graph.go b/autopilot/graph.go
index d20c6316e..3aa3ad055 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.
@@ -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 db25d9bb2..703f49020 100644
--- a/autopilot/interface.go
+++ b/autopilot/interface.go
@@ -228,12 +228,12 @@ 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
 	// 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/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/graph/builder.go b/graph/builder.go
index df4f3bfdf..073716a83 100644
--- a/graph/builder.go
+++ b/graph/builder.go
@@ -1275,10 +1275,11 @@ 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 {
+func (b *Builder) ForAllOutgoingChannels(ctx context.Context,
+	cb func(*models.ChannelEdgeInfo,
+		*models.ChannelEdgePolicy) error) error {
 
-	return b.cfg.Graph.ForEachNodeChannel(b.cfg.SelfNode,
+	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.go b/graph/db/graph.go
index 922adef53..3168a6618 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)
@@ -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/interfaces.go b/graph/db/interfaces.go
index 7140553b7..ecb07ec6d 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
@@ -82,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
 
@@ -90,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
@@ -100,13 +101,13 @@ 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
 	// 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
@@ -157,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 42b9f9245..9d1fc7cb7 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)
 }
@@ -666,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
@@ -788,7 +790,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 {
 
@@ -841,8 +845,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
@@ -3232,7 +3236,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 {
 
@@ -3248,8 +3252,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..a77709b8e 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)
@@ -784,12 +783,10 @@ func (s *SQLStore) ForEachSourceNodeChannel(cb func(chanPoint wire.OutPoint,
 // _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 {
@@ -924,10 +921,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,
@@ -959,12 +954,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{
@@ -1109,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,
@@ -1336,10 +1328,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/graph/interfaces.go b/graph/interfaces.go
index be226a495..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.
@@ -257,7 +258,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/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/rpcserver.go b/rpcserver.go
index 87bda9633..f974ea38b 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)
@@ -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
@@ -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 {
 
@@ -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.
@@ -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 3626158fd..5c0b2a77c 100644
--- a/server.go
+++ b/server.go
@@ -1252,10 +1252,11 @@ 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 {
+		ForAllOutgoingChannels: func(ctx context.Context,
+			cb func(*models.ChannelEdgeInfo,
+				*models.ChannelEdgePolicy) error) error {
 
-			return s.graphDB.ForEachNodeChannel(selfVertex,
+			return s.graphDB.ForEachNodeChannel(ctx, selfVertex,
 				func(c *models.ChannelEdgeInfo,
 					e *models.ChannelEdgePolicy,
 					_ *models.ChannelEdgePolicy) error {
@@ -2607,7 +2608,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 +3595,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.
@@ -3604,7 +3605,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)
 	}
 
@@ -3622,7 +3623,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 +3667,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 +3690,8 @@ func (s *server) establishPersistentConnections() error {
 
 		nodeAddrsMap[pubStr] = n
 		return nil
-	})
+	}
+	err = s.graphDB.ForEachSourceNodeChannel(ctx, forEachSrcNodeChan)
 	if err != nil {
 		srvrLog.Errorf("Failed to iterate over source node channels: "+
 			"%v", err)


Test patch:
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/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/db/graph_test.go b/graph/db/graph_test.go
index fefc125c4..6edce2fb1 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)
 	}
 
@@ -1311,7 +1316,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)
@@ -1332,6 +1337,7 @@ func TestForEachSourceNodeChannel(t *testing.T) {
 
 func TestGraphTraversal(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -1352,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 {
@@ -1373,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 {
 
@@ -1387,7 +1393,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 {
 
@@ -1427,6 +1433,7 @@ func TestGraphTraversal(t *testing.T) {
 // working correctly.
 func TestGraphTraversalCacheable(t *testing.T) {
 	t.Parallel()
+	ctx := context.Background()
 
 	graph := MakeTestGraph(t)
 
@@ -1440,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
@@ -1452,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)
@@ -1570,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
 	})
@@ -1662,13 +1669,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)
 	}
@@ -1681,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
@@ -3138,7 +3146,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 +4212,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
@@ -4218,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)
@@ -4244,7 +4253,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/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/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, &currentPolicy); err != nil {