Merge pull request #10068 from ellemouton/graphResetForCallbacks

multi: let all V1Store `ForEach*` methods take a `reset` call-back

autopilot/agent.go

@@ -640,6 +640,9 @@ func (a *Agent) openChans(ctx context.Context, availableFunds btcutil.Amount,
 
 		nodes[nID] = struct{}{}
 		return nil
+	}, func() {
+		nodes = nil
+		addresses = nil
 	}); err != nil {
 		return fmt.Errorf("unable to get graph nodes: %w", err)
 	}

autopilot/graph.go

@@ -121,7 +121,7 @@ func (d *dbNode) ForEachChannel(ctx context.Context,
 //
 // NOTE: Part of the autopilot.ChannelGraph interface.
 func (d *databaseChannelGraph) ForEachNode(ctx context.Context,
-	cb func(context.Context, Node) error) error {
+	cb func(context.Context, Node) error, reset func()) error {
 
 	return d.db.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error {
 		// We'll skip over any node that doesn't have any advertised
@@ -136,7 +136,7 @@ func (d *databaseChannelGraph) ForEachNode(ctx context.Context,
 		}
 
 		return cb(ctx, node)
-	})
+	}, reset)
 }
 
 // databaseChannelGraphCached wraps a channeldb.ChannelGraph instance with the
@@ -217,7 +217,7 @@ func (nc dbNodeCached) ForEachChannel(ctx context.Context,
 //
 // NOTE: Part of the autopilot.ChannelGraph interface.
 func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context,
-	cb func(context.Context, Node) error) error {
+	cb func(context.Context, Node) error, reset func()) error {
 
 	return dc.db.ForEachNodeCached(ctx, func(n route.Vertex,
 		channels map[uint64]*graphdb.DirectedChannel) error {
@@ -231,7 +231,7 @@ func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context,
 			return cb(ctx, node)
 		}
 		return nil
-	})
+	}, reset)
 }
 
 // memNode is a purely in-memory implementation of the autopilot.Node

autopilot/interface.go

@@ -85,7 +85,8 @@ type ChannelGraph interface {
 	// ForEachNode is a higher-order function that should be called once
 	// for each connected node within the channel graph. If the passed
 	// callback returns an error, then execution should be terminated.
-	ForEachNode(context.Context, func(context.Context, Node) error) error
+	ForEachNode(context.Context, func(context.Context, Node) error,
+		func()) error
 }
 
 // NodeScore is a tuple mapping a NodeID to a score indicating the preference
@@ -228,12 +229,13 @@ 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(context.Context, func(graphdb.NodeRTx) error) error
+	ForEachNode(context.Context, func(graphdb.NodeRTx) error, func()) 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(ctx context.Context, cb func(node route.Vertex,
-		chans map[uint64]*graphdb.DirectedChannel) error) error
+		chans map[uint64]*graphdb.DirectedChannel) error,
+		reset func()) error
 }

autopilot/prefattach.go

@@ -105,6 +105,9 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph,
 		}
 
 		return nil
+	}, func() {
+		allChans = nil
+		clear(seenChans)
 	}); err != nil {
 		return nil, err
 	}
@@ -162,6 +165,9 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph,
 		log.Tracef("Counted %v channels for node %x", nodeChans, nID[:])
 
 		return nil
+	}, func() {
+		maxChans = 0
+		clear(nodeChanNum)
 	}); err != nil {
 		return nil, err
 	}

autopilot/prefattach_test.go

@@ -147,10 +147,13 @@ func TestPrefAttachmentSelectTwoVertexes(t *testing.T) {
 			// Get the score for all nodes found in the graph at
 			// this point.
 			nodes := make(map[NodeID]struct{})
-			err = graph.ForEachNode(ctx,
+			err = graph.ForEachNode(
+				ctx,
 				func(_ context.Context, n Node) error {
 					nodes[n.PubKey()] = struct{}{}
 					return nil
+				}, func() {
+					clear(nodes)
 				},
 			)
 			require.NoError(t1, err)
@@ -257,7 +260,12 @@ func TestPrefAttachmentSelectGreedyAllocation(t *testing.T) {
 					twoChans = twoChans || (numChans == 2)
 
 					return nil
-				})
+				}, func() {
+					numNodes = 0
+					twoChans = false
+					clear(nodes)
+				},
+			)
 			require.NoError(t1, err)
 
 			require.EqualValues(t1, 3, numNodes)
@@ -338,7 +346,7 @@ func TestPrefAttachmentSelectSkipNodes(t *testing.T) {
 					nodes[n.PubKey()] = struct{}{}
 
 					return nil
-				},
+				}, func() {},
 			)
 			require.NoError(t1, err)
 
@@ -593,7 +601,7 @@ func newMemChannelGraph() *memChannelGraph {
 //
 // NOTE: Part of the autopilot.ChannelGraph interface.
 func (m *memChannelGraph) ForEachNode(ctx context.Context,
-	cb func(context.Context, Node) error) error {
+	cb func(context.Context, Node) error, _ func()) error {
 
 	for _, node := range m.graph {
 		if err := cb(ctx, node); err != nil {
@@ -736,7 +744,8 @@ func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
 			policy2 *models.ChannelEdgePolicy) error {
 
 			return f(edge, policy1, policy2)
-		})
+		}, func() {},
+	)
 }
 
 func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) {

autopilot/simple_graph.go

@@ -60,6 +60,10 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) {
 				return nil
 			},
 		)
+	}, func() {
+		clear(adj)
+		clear(nodes)
+		nextIndex = 0
 	})
 	if err != nil {
 		return nil, err

discovery/bootstrapper.go

@@ -247,9 +247,9 @@ func (c *ChannelGraphBootstrapper) SampleNodeAddrs(_ context.Context,
 				})
 			}
 
-			c.tried[nID] = struct{}{}
-
 			return errFound
+		}, func() {
+			a = nil
 		})
 		if err != nil && !errors.Is(err, errFound) {
 			return nil, err
@@ -282,6 +282,14 @@ func (c *ChannelGraphBootstrapper) SampleNodeAddrs(_ context.Context,
 			continue
 		}
 
+		for _, addr := range sampleAddrs {
+			nID := autopilot.NodeID(
+				addr.IdentityKey.SerializeCompressed(),
+			)
+
+			c.tried[nID] = struct{}{}
+		}
+
 		addrs = append(addrs, sampleAddrs...)
 	}
 

discovery/gossiper.go

@@ -1758,6 +1758,9 @@ func (d *AuthenticatedGossiper) retransmitStaleAnns(ctx context.Context,
 		}
 
 		return nil
+	}, func() {
+		havePublicChannels = false
+		edgesToUpdate = nil
 	})
 	if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
 		return fmt.Errorf("unable to retrieve outgoing channels: %w",

discovery/gossiper_test.go

@@ -213,7 +213,7 @@ func (r *mockGraphSource) ForEachNode(
 
 func (r *mockGraphSource) ForAllOutgoingChannels(_ context.Context,
 	cb func(i *models.ChannelEdgeInfo,
-		c *models.ChannelEdgePolicy) error) error {
+		c *models.ChannelEdgePolicy) error, _ func()) error {
 
 	r.mu.Lock()
 	defer r.mu.Unlock()
@@ -3728,10 +3728,8 @@ out:
 		})
 
 		return nil
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
+	}, func() {})
+	require.NoError(t, err)
 
 	err = ctx.gossiper.PropagateChanPolicyUpdate(edgesToUpdate)
 	require.NoError(t, err, "unable to chan policies")

graph/builder.go

@@ -1276,10 +1276,11 @@ func (b *Builder) FetchLightningNode(ctx context.Context,
 //
 // NOTE: This method is part of the ChannelGraphSource interface.
 func (b *Builder) ForAllOutgoingChannels(ctx context.Context,
-	cb func(*models.ChannelEdgeInfo,
-		*models.ChannelEdgePolicy) error) error {
+	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy) error,
+	reset func()) error {
 
-	return b.cfg.Graph.ForEachNodeChannel(ctx, b.cfg.SelfNode,
+	return b.cfg.Graph.ForEachNodeChannel(
+		ctx, b.cfg.SelfNode,
 		func(c *models.ChannelEdgeInfo, e *models.ChannelEdgePolicy,
 			_ *models.ChannelEdgePolicy) error {
 
@@ -1289,7 +1290,7 @@ func (b *Builder) ForAllOutgoingChannels(ctx context.Context,
 			}
 
 			return cb(c, e)
-		},
+		}, reset,
 	)
 }
 

graph/db/graph.go

@@ -170,7 +170,7 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
 		c.graphCache.AddNodeFeatures(node, features)
 
 		return nil
-	})
+	}, func() {})
 	if err != nil {
 		return err
 	}
@@ -182,7 +182,8 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
 			c.graphCache.AddChannel(info, policy1, policy2)
 
 			return nil
-		})
+		}, func() {},
+	)
 	if err != nil {
 		return err
 	}
@@ -204,13 +205,13 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
 //
 // NOTE: this is part of the graphdb.NodeTraverser interface.
 func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex,
-	cb func(channel *DirectedChannel) error) error {
+	cb func(channel *DirectedChannel) error, reset func()) error {
 
 	if c.graphCache != nil {
 		return c.graphCache.ForEachChannel(node, cb)
 	}
 
-	return c.V1Store.ForEachNodeDirectedChannel(node, cb)
+	return c.V1Store.ForEachNodeDirectedChannel(node, cb, reset)
 }
 
 // FetchNodeFeatures returns the features of the given node. If no features are
@@ -233,12 +234,14 @@ func (c *ChannelGraph) FetchNodeFeatures(node route.Vertex) (
 // instance which can be used to perform queries against the channel graph. If
 // the graph cache is not enabled, then the call-back will be provided with
 // access to the graph via a consistent read-only transaction.
-func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error {
+func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error,
+	reset func()) error {
+
 	if c.graphCache != nil {
 		return cb(c)
 	}
 
-	return c.V1Store.GraphSession(cb)
+	return c.V1Store.GraphSession(cb, reset)
 }
 
 // ForEachNodeCached iterates through all the stored vertices/nodes in the
@@ -246,14 +249,14 @@ func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error) error {
 //
 // NOTE: The callback contents MUST not be modified.
 func (c *ChannelGraph) ForEachNodeCached(ctx context.Context,
-	cb func(node route.Vertex,
-		chans map[uint64]*DirectedChannel) error) error {
+	cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error,
+	reset func()) error {
 
 	if c.graphCache != nil {
 		return c.graphCache.ForEachNode(cb)
 	}
 
-	return c.V1Store.ForEachNodeCached(ctx, cb)
+	return c.V1Store.ForEachNodeCached(ctx, cb, reset)
 }
 
 // AddLightningNode adds a vertex/node to the graph database. If the node is not

graph/db/graph_test.go

@@ -935,7 +935,8 @@ func TestEdgePolicyCRUD(t *testing.T) {
 				)
 
 				return nil
-			})
+			}, func() {},
+		)
 		require.NoError(t, err)
 	}
 
@@ -1330,7 +1331,7 @@ func TestForEachSourceNodeChannel(t *testing.T) {
 		delete(expectedSrcChans, chanPoint)
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 	require.Empty(t, expectedSrcChans)
 }
@@ -1373,7 +1374,7 @@ func TestGraphTraversal(t *testing.T) {
 		}
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 
 	// Iterate through all the known channels within the graph DB, once
@@ -1385,7 +1386,7 @@ func TestGraphTraversal(t *testing.T) {
 
 		delete(chanIndex, ei.ChannelID)
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 	require.Len(t, chanIndex, 0)
 
@@ -1393,7 +1394,8 @@ func TestGraphTraversal(t *testing.T) {
 	// outgoing channels for a particular node.
 	numNodeChans := 0
 	firstNode, secondNode := nodeList[0], nodeList[1]
-	err = graph.ForEachNodeChannel(ctx, firstNode.PubKeyBytes,
+	err = graph.ForEachNodeChannel(
+		ctx, firstNode.PubKeyBytes,
 		func(_ *models.ChannelEdgeInfo, outEdge,
 			inEdge *models.ChannelEdgePolicy) error {
 
@@ -1424,7 +1426,8 @@ func TestGraphTraversal(t *testing.T) {
 			numNodeChans++
 
 			return nil
-		})
+		}, func() {},
+	)
 	require.NoError(t, err)
 	require.Equal(t, numChannels, numNodeChans)
 }
@@ -1451,7 +1454,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 		nodeMap[tx.Node().PubKeyBytes] = struct{}{}
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 	require.Len(t, nodeMap, numNodes)
 
@@ -1466,6 +1469,8 @@ func TestGraphTraversalCacheable(t *testing.T) {
 		nodes = append(nodes, node)
 
 		return nil
+	}, func() {
+		nodes = nil
 	})
 	require.NoError(t, err)
 	require.Len(t, nodeMap, 0)
@@ -1485,7 +1490,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 			node, func(d *DirectedChannel) error {
 				delete(chanIndex, d.ChannelID)
 				return nil
-			},
+			}, func() {},
 		)
 		require.NoError(t, err)
 
@@ -1494,7 +1499,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 			node, func(d *DirectedChannel) error {
 				delete(chanIndex2, d.ChannelID)
 				return nil
-			},
+			}, func() {},
 		)
 		require.NoError(t, err)
 	}
@@ -1580,7 +1585,7 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes,
 	err := graph.ForEachNode(ctx, func(tx NodeRTx) error {
 		delete(nodeIndex, tx.Node().Alias)
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 	require.Len(t, nodeIndex, 0)
 
@@ -1669,22 +1674,19 @@ func assertPruneTip(t *testing.T, graph *ChannelGraph,
 
 func assertNumChans(t *testing.T, graph *ChannelGraph, n int) {
 	numChans := 0
-	if err := graph.ForEachChannel(context.Background(),
-		func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+	err := graph.ForEachChannel(
+		context.Background(), func(*models.ChannelEdgeInfo,
+			*models.ChannelEdgePolicy,
 			*models.ChannelEdgePolicy) error {
 
 			numChans++
 			return nil
+		}, func() {
+			numChans = 0
 		},
-	); err != nil {
-		_, _, line, _ := runtime.Caller(1)
-		t.Fatalf("line %v: unable to scan channels: %v", line, err)
-	}
-	if numChans != n {
-		_, _, line, _ := runtime.Caller(1)
-		t.Fatalf("line %v: expected %v chans instead have %v", line,
-			n, numChans)
-	}
+	)
+	require.NoError(t, err)
+	require.Equal(t, n, numChans)
 }
 
 func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {
@@ -1693,7 +1695,7 @@ func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) {
 		numNodes++
 
 		return nil
-	})
+	}, func() {})
 	if err != nil {
 		_, _, line, _ := runtime.Caller(1)
 		t.Fatalf("line %v: unable to scan nodes: %v", line, err)
@@ -3146,7 +3148,8 @@ func TestIncompleteChannelPolicies(t *testing.T) {
 		expectedOut bool) {
 
 		calls := 0
-		err := graph.ForEachNodeChannel(ctx, node.PubKeyBytes,
+		err := graph.ForEachNodeChannel(
+			ctx, node.PubKeyBytes,
 			func(_ *models.ChannelEdgeInfo, outEdge,
 				inEdge *models.ChannelEdgePolicy) error {
 
@@ -3169,14 +3172,10 @@ func TestIncompleteChannelPolicies(t *testing.T) {
 				calls++
 
 				return nil
-			})
-		if err != nil {
-			t.Fatalf("unable to scan channels: %v", err)
-		}
-
-		if calls != 1 {
-			t.Fatalf("Expected only one callback call")
-		}
+			}, func() {},
+		)
+		require.NoError(t, err)
+		require.Equal(t, 1, calls)
 	}
 
 	checkPolicies(node2, false, false)
@@ -4230,6 +4229,8 @@ func BenchmarkForEachChannel(b *testing.B) {
 			nodes = append(nodes, node)
 
 			return nil
+		}, func() {
+			nodes = nil
 		})
 		require.NoError(b, err)
 
@@ -4250,7 +4251,7 @@ func BenchmarkForEachChannel(b *testing.B) {
 				return nil
 			}
 
-			err := graph.ForEachNodeChannel(ctx, n, cb)
+			err := graph.ForEachNodeChannel(ctx, n, cb, func() {})
 			require.NoError(b, err)
 		}
 	}
@@ -4297,7 +4298,7 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) {
 				ch = c
 
 				return nil
-			},
+			}, func() {},
 		)
 		require.NoError(t, err)
 

graph/db/interfaces.go

@@ -40,7 +40,7 @@ type NodeTraverser interface {
 	// ForEachNodeDirectedChannel calls the callback for every channel of
 	// the given node.
 	ForEachNodeDirectedChannel(nodePub route.Vertex,
-		cb func(channel *DirectedChannel) error) error
+		cb func(channel *DirectedChannel) error, reset func()) error
 
 	// FetchNodeFeatures returns the features of the given node.
 	FetchNodeFeatures(nodePub route.Vertex) (*lnwire.FeatureVector, error)
@@ -72,7 +72,8 @@ type V1Store interface { //nolint:interfacebloat
 	// the channel and the channel peer's node information.
 	ForEachSourceNodeChannel(ctx context.Context,
 		cb func(chanPoint wire.OutPoint, havePolicy bool,
-			otherNode *models.LightningNode) error) error
+			otherNode *models.LightningNode) error,
+		reset func()) error
 
 	// ForEachNodeChannel iterates through all channels of the given node,
 	// executing the passed callback with an edge info structure and the
@@ -85,14 +86,14 @@ type V1Store interface { //nolint:interfacebloat
 	// Unknown policies are passed into the callback as nil values.
 	ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
 		cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-			*models.ChannelEdgePolicy) error) error
+			*models.ChannelEdgePolicy) error, reset func()) error
 
 	// ForEachNodeCached is similar to forEachNode, but it returns
 	// DirectedChannel data to the call-back.
 	//
 	// NOTE: The callback contents MUST not be modified.
 	ForEachNodeCached(ctx context.Context, cb func(node route.Vertex,
-		chans map[uint64]*DirectedChannel) error) error
+		chans map[uint64]*DirectedChannel) error, reset func()) error
 
 	// ForEachNode iterates through all the stored vertices/nodes in the
 	// graph, executing the passed callback with each node encountered. If
@@ -101,14 +102,15 @@ 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(ctx context.Context, cb func(tx NodeRTx) error) error
+	ForEachNode(ctx context.Context, cb func(tx NodeRTx) error,
+		reset func()) 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(ctx context.Context, cb func(route.Vertex,
-		*lnwire.FeatureVector) error) error
+		*lnwire.FeatureVector) error, reset func()) error
 
 	// LookupAlias attempts to return the alias as advertised by the target
 	// node.
@@ -147,7 +149,7 @@ type V1Store interface { //nolint:interfacebloat
 	// GraphSession will provide the call-back with access to a
 	// NodeTraverser instance which can be used to perform queries against
 	// the channel graph.
-	GraphSession(cb func(graph NodeTraverser) error) error
+	GraphSession(cb func(graph NodeTraverser) error, reset func()) error
 
 	// ForEachChannel iterates through all the channel edges stored within
 	// the graph and invokes the passed callback for each edge. The callback
@@ -159,8 +161,8 @@ type V1Store interface { //nolint:interfacebloat
 	// pointer for that particular channel edge routing policy will be
 	// passed into the callback.
 	ForEachChannel(ctx context.Context, cb func(*models.ChannelEdgeInfo,
-		*models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error
+		*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+		reset func()) error
 
 	// ForEachChannelCacheable iterates through all the channel edges stored
 	// within the graph and invokes the passed callback for each edge. The
@@ -175,8 +177,8 @@ type V1Store interface { //nolint:interfacebloat
 	// NOTE: this method is like ForEachChannel but fetches only the data
 	// required for the graph cache.
 	ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-		*models.CachedEdgePolicy,
-		*models.CachedEdgePolicy) error) error
+		*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
+		reset func()) error
 
 	// DisabledChannelIDs returns the channel ids of disabled channels.
 	// A channel is disabled when two of the associated ChanelEdgePolicies

graph/db/kv_store.go

@@ -406,9 +406,9 @@ func (c *KVStore) AddrsForNode(ctx context.Context,
 // callback.
 func (c *KVStore) ForEachChannel(_ context.Context,
 	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
+		*models.ChannelEdgePolicy) error, reset func()) error {
 
-	return forEachChannel(c.db, cb)
+	return forEachChannel(c.db, cb, reset)
 }
 
 // forEachChannel iterates through all the channel edges stored within the
@@ -421,7 +421,8 @@ func (c *KVStore) ForEachChannel(_ context.Context,
 // for that particular channel edge routing policy will be passed into the
 // callback.
 func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
-	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
+	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+	reset func()) error {
 
 	return db.View(func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
@@ -469,7 +470,7 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
 				return cb(&info, policy1, policy2)
 			},
 		)
-	}, func() {})
+	}, reset)
 }
 
 // ForEachChannelCacheable iterates through all the channel edges stored within
@@ -485,7 +486,8 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
 // NOTE: this method is like ForEachChannel but fetches only the data required
 // for the graph cache.
 func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-	*models.CachedEdgePolicy, *models.CachedEdgePolicy) error) error {
+	*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
+	reset func()) error {
 
 	return c.db.View(func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
@@ -558,7 +560,7 @@ func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 				)
 			},
 		)
-	}, func() {})
+	}, reset)
 }
 
 // forEachNodeDirectedChannel iterates through all channels of a given node,
@@ -568,8 +570,13 @@ func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 // transaction may be provided. If none is provided, a new one will be created.
 //
 // Unknown policies are passed into the callback as nil values.
+//
+// NOTE: the reset param is only meaningful if the tx param is nil. If it is
+// not nil, the caller is expected to have passed in a reset to the parent
+// function's View/Update call which will then apply to the whole transaction.
 func (c *KVStore) forEachNodeDirectedChannel(tx kvdb.RTx,
-	node route.Vertex, cb func(channel *DirectedChannel) error) error {
+	node route.Vertex, cb func(channel *DirectedChannel) error,
+	reset func()) error {
 
 	// Fallback that uses the database.
 	toNodeCallback := func() route.Vertex {
@@ -612,7 +619,7 @@ func (c *KVStore) forEachNodeDirectedChannel(tx kvdb.RTx,
 		return cb(directedChannel)
 	}
 
-	return nodeTraversal(tx, node[:], c.db, dbCallback)
+	return nodeTraversal(tx, node[:], c.db, dbCallback, reset)
 }
 
 // fetchNodeFeatures returns the features of a given node. If no features are
@@ -648,9 +655,9 @@ func (c *KVStore) fetchNodeFeatures(tx kvdb.RTx,
 //
 // NOTE: this is part of the graphdb.NodeTraverser interface.
 func (c *KVStore) ForEachNodeDirectedChannel(nodePub route.Vertex,
-	cb func(channel *DirectedChannel) error) error {
+	cb func(channel *DirectedChannel) error, reset func()) error {
 
-	return c.forEachNodeDirectedChannel(nil, nodePub, cb)
+	return c.forEachNodeDirectedChannel(nil, nodePub, cb, reset)
 }
 
 // FetchNodeFeatures returns the features of the given node. If no features are
@@ -668,8 +675,8 @@ func (c *KVStore) FetchNodeFeatures(nodePub route.Vertex) (
 //
 // NOTE: The callback contents MUST not be modified.
 func (c *KVStore) ForEachNodeCached(_ context.Context,
-	cb func(node route.Vertex,
-		chans map[uint64]*DirectedChannel) error) error {
+	cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error,
+	reset func()) 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
@@ -723,13 +730,14 @@ func (c *KVStore) ForEachNodeCached(_ context.Context,
 				channels[e.ChannelID] = directedChannel
 
 				return nil
-			})
+			}, reset,
+		)
 		if err != nil {
 			return err
 		}
 
 		return cb(node.PubKeyBytes, channels)
-	})
+	}, reset)
 }
 
 // DisabledChannelIDs returns the channel ids of disabled channels.
@@ -791,13 +799,13 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) {
 // 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(_ context.Context,
-	cb func(tx NodeRTx) error) error {
+	cb func(tx NodeRTx) error, reset func()) error {
 
 	return forEachNode(c.db, func(tx kvdb.RTx,
 		node *models.LightningNode) error {
 
 		return cb(newChanGraphNodeTx(tx, c, node))
-	})
+	}, reset)
 }
 
 // forEachNode iterates through all the stored vertices/nodes in the graph,
@@ -808,7 +816,7 @@ func (c *KVStore) ForEachNode(_ context.Context,
 // TODO(roasbeef): add iterator interface to allow for memory efficient graph
 // traversal when graph gets mega.
 func forEachNode(db kvdb.Backend,
-	cb func(kvdb.RTx, *models.LightningNode) error) error {
+	cb func(kvdb.RTx, *models.LightningNode) error, reset func()) error {
 
 	traversal := func(tx kvdb.RTx) error {
 		// First grab the nodes bucket which stores the mapping from
@@ -838,7 +846,7 @@ func forEachNode(db kvdb.Backend,
 		})
 	}
 
-	return kvdb.View(db, traversal, func() {})
+	return kvdb.View(db, traversal, reset)
 }
 
 // ForEachNodeCacheable iterates through all the stored vertices/nodes in the
@@ -846,7 +854,8 @@ func forEachNode(db kvdb.Backend,
 // callback returns an error, then the transaction is aborted and the iteration
 // stops early.
 func (c *KVStore) ForEachNodeCacheable(_ context.Context,
-	cb func(route.Vertex, *lnwire.FeatureVector) error) error {
+	cb func(route.Vertex, *lnwire.FeatureVector) error,
+	reset func()) error {
 
 	traversal := func(tx kvdb.RTx) error {
 		// First grab the nodes bucket which stores the mapping from
@@ -878,7 +887,7 @@ func (c *KVStore) ForEachNodeCacheable(_ context.Context,
 		})
 	}
 
-	return kvdb.View(c.db, traversal, func() {})
+	return kvdb.View(c.db, traversal, reset)
 }
 
 // SourceNode returns the source node of the graph. The source node is treated
@@ -3018,6 +3027,8 @@ func (c *KVStore) isPublic(tx kvdb.RTx, nodePub route.Vertex,
 
 		// Otherwise, we'll continue our search.
 		return nil
+	}, func() {
+		nodeIsPublic = false
 	})
 	if err != nil && !errors.Is(err, errDone) {
 		return false, err
@@ -3157,9 +3168,13 @@ func (c *KVStore) HasLightningNode(_ context.Context,
 
 // nodeTraversal is used to traverse all channels of a node given by its
 // public key and passes channel information into the specified callback.
+//
+// NOTE: the reset param is only meaningful if the tx param is nil. If it is
+// not nil, the caller is expected to have passed in a reset to the parent
+// function's View/Update call which will then apply to the whole transaction.
 func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
 	cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
+		*models.ChannelEdgePolicy) error, reset func()) error {
 
 	traversal := func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
@@ -3230,7 +3245,7 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
 	// If no transaction was provided, then we'll create a new transaction
 	// to execute the transaction within.
 	if tx == nil {
-		return kvdb.View(db, traversal, func() {})
+		return kvdb.View(db, traversal, reset)
 	}
 
 	// Otherwise, we re-use the existing transaction to execute the graph
@@ -3248,14 +3263,16 @@ func nodeTraversal(tx kvdb.RTx, nodePub []byte, db kvdb.Backend,
 // Unknown policies are passed into the callback as nil values.
 func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex,
 	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
+		*models.ChannelEdgePolicy) error, reset func()) error {
 
-	return nodeTraversal(nil, nodePub[:], c.db, func(_ kvdb.RTx,
-		info *models.ChannelEdgeInfo, policy,
-		policy2 *models.ChannelEdgePolicy) error {
+	return nodeTraversal(
+		nil, nodePub[:], c.db, func(_ kvdb.RTx,
+			info *models.ChannelEdgeInfo, policy,
+			policy2 *models.ChannelEdgePolicy) error {
 
-		return cb(info, policy, policy2)
-	})
+			return cb(info, policy, policy2)
+		}, reset,
+	)
 }
 
 // ForEachSourceNodeChannel iterates through all channels of the source node,
@@ -3264,7 +3281,7 @@ func (c *KVStore) ForEachNodeChannel(_ context.Context, nodePub route.Vertex,
 // peer's node information.
 func (c *KVStore) ForEachSourceNodeChannel(_ context.Context,
 	cb func(chanPoint wire.OutPoint, havePolicy bool,
-		otherNode *models.LightningNode) error) error {
+		otherNode *models.LightningNode) error, reset func()) error {
 
 	return kvdb.View(c.db, func(tx kvdb.RTx) error {
 		nodes := tx.ReadBucket(nodeBucket)
@@ -3292,9 +3309,9 @@ func (c *KVStore) ForEachSourceNodeChannel(_ context.Context,
 				return cb(
 					info.ChannelPoint, policy != nil, peer,
 				)
-			},
+			}, reset,
 		)
-	}, func() {})
+	}, reset)
 }
 
 // forEachNodeChannelTx iterates through all channels of the given node,
@@ -3310,12 +3327,14 @@ func (c *KVStore) ForEachSourceNodeChannel(_ context.Context,
 // should be passed as the first argument.  Otherwise, the first argument should
 // be nil and a fresh transaction will be created to execute the graph
 // traversal.
+//
+// NOTE: the reset function is only meaningful if the tx param is nil.
 func (c *KVStore) forEachNodeChannelTx(tx kvdb.RTx,
 	nodePub route.Vertex, cb func(kvdb.RTx, *models.ChannelEdgeInfo,
-		*models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
+		*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+	reset func()) error {
 
-	return nodeTraversal(tx, nodePub[:], c.db, cb)
+	return nodeTraversal(tx, nodePub[:], c.db, cb, reset)
 }
 
 // fetchOtherNode attempts to fetch the full LightningNode that's opposite of
@@ -3955,13 +3974,15 @@ func (c *KVStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) {
 
 // GraphSession will provide the call-back with access to a NodeTraverser
 // instance which can be used to perform queries against the channel graph.
-func (c *KVStore) GraphSession(cb func(graph NodeTraverser) error) error {
+func (c *KVStore) GraphSession(cb func(graph NodeTraverser) error,
+	reset func()) error {
+
 	return c.db.View(func(tx walletdb.ReadTx) error {
 		return cb(&nodeTraverserSession{
 			db: c,
 			tx: tx,
 		})
-	}, func() {})
+	}, reset)
 }
 
 // nodeTraverserSession implements the NodeTraverser interface but with a
@@ -3976,9 +3997,9 @@ type nodeTraverserSession struct {
 //
 // NOTE: Part of the NodeTraverser interface.
 func (c *nodeTraverserSession) ForEachNodeDirectedChannel(nodePub route.Vertex,
-	cb func(channel *DirectedChannel) error) error {
+	cb func(channel *DirectedChannel) error, _ func()) error {
 
-	return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb)
+	return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb, func() {})
 }
 
 // FetchNodeFeatures returns the features of the given node. If the node is
@@ -4902,11 +4923,16 @@ func (c *chanGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) {
 func (c *chanGraphNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo,
 	*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error {
 
-	return c.db.forEachNodeChannelTx(c.tx, c.node.PubKeyBytes,
+	return c.db.forEachNodeChannelTx(
+		c.tx, c.node.PubKeyBytes,
 		func(_ kvdb.RTx, info *models.ChannelEdgeInfo, policy1,
 			policy2 *models.ChannelEdgePolicy) error {
 
 			return f(info, policy1, policy2)
 		},
+		// NOTE: We don't need to reset anything here as the caller is
+		// expected to pass in the reset function to the ForEachNode
+		// method that constructed the chanGraphNodeTx.
+		func() {},
 	)
 }

graph/db/sql_migration.go

@@ -206,6 +206,9 @@ func migrateNodes(ctx context.Context, kvBackend kvdb.Backend,
 		return sqldb.CompareRecords(
 			node, migratedNode, fmt.Sprintf("node %x", pub),
 		)
+	}, func() {
+		// No reset is needed since if a retry occurs, the entire
+		// migration will be retried from the start.
 	})
 	if err != nil {
 		return fmt.Errorf("could not migrate nodes: %w", err)
@@ -384,6 +387,9 @@ func migrateChannelsAndPolicies(ctx context.Context, kvBackend kvdb.Backend,
 		}
 
 		return nil
+	}, func() {
+		// No reset is needed since if a retry occurs, the entire
+		// migration will be retried from the start.
 	})
 	if err != nil {
 		return fmt.Errorf("could not migrate channels and policies: %w",

graph/db/sql_migration_test.go

@@ -401,6 +401,8 @@ func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode {
 		nodes = append(nodes, node)
 
 		return nil
+	}, func() {
+		nodes = nil
 	})
 	require.NoError(t, err)
 
@@ -475,7 +477,7 @@ func fetchAllChannelsAndPolicies(t *testing.T, store V1Store) chanSet {
 		})
 
 		return nil
-	})
+	}, func() {})
 	require.NoError(t, err)
 
 	// Sort the channels by their channel ID to ensure a consistent order.

graph/db/sql_store.go

@@ -727,7 +727,7 @@ func (s *SQLStore) updateEdgeCache(e *models.ChannelEdgePolicy,
 // NOTE: part of the V1Store interface.
 func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
 	cb func(chanPoint wire.OutPoint, havePolicy bool,
-		otherNode *models.LightningNode) error) error {
+		otherNode *models.LightningNode) error, reset func()) error {
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		nodeID, nodePub, err := s.getSourceNode(ctx, db, ProtocolV1)
@@ -773,7 +773,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
 				)
 			},
 		)
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // ForEachNode iterates through all the stored vertices/nodes in the graph,
@@ -785,7 +785,7 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context,
 //
 // NOTE: part of the V1Store interface.
 func (s *SQLStore) ForEachNode(ctx context.Context,
-	cb func(tx NodeRTx) error) error {
+	cb func(tx NodeRTx) error, reset func()) error {
 
 	var lastID int64 = 0
 	handleNode := func(db SQLQueries, dbNode sqlc.Node) error {
@@ -835,7 +835,7 @@ func (s *SQLStore) ForEachNode(ctx context.Context,
 		}
 
 		return nil
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // sqlGraphNodeTx is an implementation of the NodeRTx interface backed by the
@@ -907,13 +907,13 @@ func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) {
 //
 // NOTE: this is part of the graphdb.NodeTraverser interface.
 func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex,
-	cb func(channel *DirectedChannel) error) error {
+	cb func(channel *DirectedChannel) error, reset func()) error {
 
 	var ctx = context.TODO()
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		return forEachNodeDirectedChannel(ctx, db, nodePub, cb)
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // ForEachNodeCacheable iterates through all the stored vertices/nodes in the
@@ -923,7 +923,8 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex,
 //
 // NOTE: This is a part of the V1Store interface.
 func (s *SQLStore) ForEachNodeCacheable(ctx context.Context,
-	cb func(route.Vertex, *lnwire.FeatureVector) error) error {
+	cb func(route.Vertex, *lnwire.FeatureVector) error,
+	reset func()) error {
 
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		return forEachNodeCacheable(ctx, db, func(nodeID int64,
@@ -937,7 +938,7 @@ func (s *SQLStore) ForEachNodeCacheable(ctx context.Context,
 
 			return cb(nodePub, features)
 		})
-	}, sqldb.NoOpReset)
+	}, reset)
 	if err != nil {
 		return fmt.Errorf("unable to fetch nodes: %w", err)
 	}
@@ -957,7 +958,7 @@ func (s *SQLStore) ForEachNodeCacheable(ctx context.Context,
 // NOTE: part of the V1Store interface.
 func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
 	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
+		*models.ChannelEdgePolicy) error, reset func()) error {
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		dbNode, err := db.GetNodeByPubKey(
@@ -975,7 +976,7 @@ func (s *SQLStore) ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
 		return forEachNodeChannel(
 			ctx, db, s.cfg.ChainHash, dbNode.ID, cb,
 		)
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // ChanUpdatesInHorizon returns all the known channel edges which have at least
@@ -1104,8 +1105,8 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime,
 //
 // NOTE: part of the V1Store interface.
 func (s *SQLStore) ForEachNodeCached(ctx context.Context,
-	cb func(node route.Vertex,
-		chans map[uint64]*DirectedChannel) error) error {
+	cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error,
+	reset func()) error {
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		return forEachNodeCacheable(ctx, db, func(nodeID int64,
@@ -1218,7 +1219,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context,
 
 			return cb(nodePub, channels)
 		})
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // ForEachChannelCacheable iterates through all the channel edges stored
@@ -1234,8 +1235,8 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context,
 // NOTE: this method is like ForEachChannel but fetches only the data
 // required for the graph cache.
 func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-	*models.CachedEdgePolicy,
-	*models.CachedEdgePolicy) error) error {
+	*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
+	reset func()) error {
 
 	ctx := context.TODO()
 
@@ -1315,7 +1316,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 		}
 
 		return nil
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // ForEachChannel iterates through all the channel edges stored within the
@@ -1331,7 +1332,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 // NOTE: part of the V1Store interface.
 func (s *SQLStore) ForEachChannel(ctx context.Context,
 	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error) error {
+		*models.ChannelEdgePolicy) error, reset func()) error {
 
 	handleChannel := func(db SQLQueries,
 		row sqlc.ListChannelsWithPoliciesPaginatedRow) error {
@@ -1406,7 +1407,7 @@ func (s *SQLStore) ForEachChannel(ctx context.Context,
 		}
 
 		return nil
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // FilterChannelRange returns the channel ID's of all known channels which were
@@ -2754,12 +2755,14 @@ func (s *SQLStore) IsClosedScid(scid lnwire.ShortChannelID) (bool, error) {
 // instance which can be used to perform queries against the channel graph.
 //
 // NOTE: part of the V1Store interface.
-func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error) error {
+func (s *SQLStore) GraphSession(cb func(graph NodeTraverser) error,
+	reset func()) error {
+
 	var ctx = context.TODO()
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		return cb(newSQLNodeTraverser(db, s.cfg.ChainHash))
-	}, sqldb.NoOpReset)
+	}, reset)
 }
 
 // sqlNodeTraverser implements the NodeTraverser interface but with a backing
@@ -2788,7 +2791,7 @@ func newSQLNodeTraverser(db SQLQueries,
 //
 // NOTE: Part of the NodeTraverser interface.
 func (s *sqlNodeTraverser) ForEachNodeDirectedChannel(nodePub route.Vertex,
-	cb func(channel *DirectedChannel) error) error {
+	cb func(channel *DirectedChannel) error, _ func()) error {
 
 	ctx := context.TODO()
 

graph/interfaces.go

@@ -72,7 +72,7 @@ type ChannelGraphSource interface {
 	// star-graph.
 	ForAllOutgoingChannels(ctx context.Context,
 		cb func(c *models.ChannelEdgeInfo,
-			e *models.ChannelEdgePolicy) error) error
+			e *models.ChannelEdgePolicy) error, reset func()) error
 
 	// CurrentBlockHeight returns the block height from POV of the router
 	// subsystem.
@@ -260,7 +260,7 @@ type DB interface {
 	// Unknown policies are passed into the callback as nil values.
 	ForEachNodeChannel(ctx context.Context, nodePub route.Vertex,
 		cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-			*models.ChannelEdgePolicy) error) error
+			*models.ChannelEdgePolicy) error, reset func()) error
 
 	// AddEdgeProof sets the proof of an existing edge in the graph
 	// database.

itest/lnd_graph_migration_test.go

@@ -82,6 +82,9 @@ func testGraphMigration(ht *lntest.HarnessTest) {
 					return nil
 				},
 			)
+		}, func() {
+			clear(edges)
+			numNodes = 0
 		})
 		require.NoError(ht, err)
 		require.Equal(ht, expNumNodes, numNodes)

routing/bandwidth.go

@@ -63,16 +63,18 @@ func newBandwidthManager(graph Graph, sourceNode route.Vertex,
 
 	// First, we'll collect the set of outbound edges from the target
 	// source node and add them to our bandwidth manager's map of channels.
-	err := graph.ForEachNodeDirectedChannel(sourceNode,
-		func(channel *graphdb.DirectedChannel) error {
+	err := graph.ForEachNodeDirectedChannel(
+		sourceNode, func(channel *graphdb.DirectedChannel) error {
 			shortID := lnwire.NewShortChanIDFromInt(
 				channel.ChannelID,
 			)
 			manager.localChans[shortID] = struct{}{}
 
 			return nil
-		})
-
+		}, func() {
+			clear(manager.localChans)
+		},
+	)
 	if err != nil {
 		return nil, err
 	}

routing/graph.go

@@ -15,7 +15,8 @@ type Graph interface {
 	// ForEachNodeDirectedChannel calls the callback for every channel of
 	// the given node.
 	ForEachNodeDirectedChannel(nodePub route.Vertex,
-		cb func(channel *graphdb.DirectedChannel) error) error
+		cb func(channel *graphdb.DirectedChannel) error,
+		reset func()) error
 
 	// FetchNodeFeatures returns the features of the given node.
 	FetchNodeFeatures(nodePub route.Vertex) (*lnwire.FeatureVector, error)
@@ -29,7 +30,8 @@ type GraphSessionFactory interface {
 	// GraphSession will provide the call-back with access to a
 	// graphdb.NodeTraverser instance which can be used to perform queries
 	// against the channel graph.
-	GraphSession(cb func(graph graphdb.NodeTraverser) error) error
+	GraphSession(cb func(graph graphdb.NodeTraverser) error,
+		reset func()) error
 }
 
 // FetchAmountPairCapacity determines the maximal public capacity between two

routing/localchans/manager.go

@@ -42,7 +42,7 @@ type Manager struct {
 	// channels. The ChannelEdgePolicy parameter may be nil.
 	ForAllOutgoingChannels func(ctx context.Context,
 		cb func(*models.ChannelEdgeInfo,
-			*models.ChannelEdgePolicy) error) error
+			*models.ChannelEdgePolicy) error, reset func()) error
 
 	// FetchChannel is used to query local channel parameters. Optionally an
 	// existing db tx can be supplied.
@@ -152,7 +152,14 @@ 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(ctx, processChan)
+	err := r.ForAllOutgoingChannels(
+		ctx, processChan,
+		func() {
+			failedUpdates = nil
+			edgesToUpdate = nil
+			clear(policiesToUpdate)
+		},
+	)
 	if err != nil {
 		return nil, err
 	}

routing/localchans/manager_test.go

@@ -125,7 +125,7 @@ func TestManager(t *testing.T) {
 
 	forAllOutgoingChannels := func(_ context.Context,
 		cb func(*models.ChannelEdgeInfo,
-			*models.ChannelEdgePolicy) error) error {
+			*models.ChannelEdgePolicy) error, _ func()) error {
 
 		for _, c := range channelSet {
 			if err := cb(c.edgeInfo, &currentPolicy); err != nil {

routing/mock_graph_test.go

@@ -166,7 +166,7 @@ func (m *mockGraph) addChannel(id uint64, node1id, node2id byte,
 //
 // NOTE: Part of the Graph interface.
 func (m *mockGraph) ForEachNodeDirectedChannel(nodePub route.Vertex,
-	cb func(channel *graphdb.DirectedChannel) error) error {
+	cb func(channel *graphdb.DirectedChannel) error, _ func()) error {
 
 	// Look up the mock node.
 	node, ok := m.nodes[nodePub]
@@ -232,8 +232,8 @@ func (m *mockGraph) FetchNodeFeatures(nodePub route.Vertex) (
 // the channel graph.
 //
 // NOTE: Part of the GraphSessionFactory interface.
-func (m *mockGraph) GraphSession(
-	cb func(graph graphdb.NodeTraverser) error) error {
+func (m *mockGraph) GraphSession(cb func(graph graphdb.NodeTraverser) error,
+	_ func()) error {
 
 	return cb(m)
 }

routing/pathfind.go

@@ -576,7 +576,12 @@ func getOutgoingBalance(node route.Vertex, outgoingChans map[uint64]struct{},
 	}
 
 	// Iterate over all channels of the to node.
-	err := g.ForEachNodeDirectedChannel(node, cb)
+	err := g.ForEachNodeDirectedChannel(
+		node, cb, func() {
+			max = 0
+			total = 0
+		},
+	)
 	if err != nil {
 		return 0, 0, err
 	}
@@ -1276,7 +1281,12 @@ func findBlindedPaths(g Graph, target route.Vertex,
 		// it later on when we call "processNodeForBlindedPath".
 		visited[nextTarget] = true
 
-		err := g.ForEachNodeDirectedChannel(nextTarget,
+		var (
+			incomingPathReset []blindedHop
+			nextTargetReset   = nextTarget
+		)
+		err := g.ForEachNodeDirectedChannel(
+			nextTarget,
 			func(channel *graphdb.DirectedChannel) error {
 				// This is not the right channel, continue to
 				// the node's other channels.
@@ -1286,18 +1296,21 @@ func findBlindedPaths(g Graph, target route.Vertex,
 
 				// We found the channel in question. Prepend it
 				// to the incoming path.
-				incomingPath = append([]blindedHop{
+				incomingPathReset = append([]blindedHop{
 					{
 						vertex:       channel.OtherNode,
 						channelID:    channel.ChannelID,
 						edgeCapacity: channel.Capacity,
 					},
-				}, incomingPath...)
+				}, incomingPathReset...)
 
 				// Update the target node.
-				nextTarget = channel.OtherNode
+				nextTargetReset = channel.OtherNode
 
 				return errChanFound
+			}, func() {
+				incomingPathReset = nil
+				nextTargetReset = nextTarget
 			},
 		)
 		// We expect errChanFound to be returned if the channel in
@@ -1308,6 +1321,8 @@ func findBlindedPaths(g Graph, target route.Vertex,
 			return nil, fmt.Errorf("incoming channel %d is not "+
 				"seen as owned by node %v", chanID, nextTarget)
 		}
+		nextTarget = nextTargetReset
+		incomingPath = append(incomingPathReset, incomingPath...)
 
 		// Check that the user didn't accidentally add a channel that
 		// is owned by a node in the node omission set.
@@ -1445,7 +1460,8 @@ func processNodeForBlindedPath(g Graph, node route.Vertex,
 
 	// Now, iterate over the node's channels in search for paths to this
 	// node that can be used for blinded paths
-	err = g.ForEachNodeDirectedChannel(node,
+	err = g.ForEachNodeDirectedChannel(
+		node,
 		func(channel *graphdb.DirectedChannel) error {
 			// Keep track of how many incoming channels this node
 			// has. We only use a node as an introduction node if it
@@ -1488,6 +1504,9 @@ func processNodeForBlindedPath(g Graph, node route.Vertex,
 			}
 
 			return nil
+		}, func() {
+			hopSets = nil
+			chanCount = 0
 		},
 	)
 	if err != nil {

routing/pathfind_test.go

@@ -3303,6 +3303,8 @@ func dbFindPath(graph *graphdb.ChannelGraph,
 		)
 
 		return err
+	}, func() {
+		route = nil
 	})
 	if err != nil {
 		return nil, err

routing/payment_session.go

@@ -343,7 +343,11 @@ func (p *paymentSession) RequestRoute(maxAmt, feeLimit lnwire.MilliSatoshi,
 	}
 
 	for {
-		err := p.graphSessFactory.GraphSession(findPath)
+		err := p.graphSessFactory.GraphSession(
+			findPath, func() {
+				path = nil
+			},
+		)
 		// If there is an error, and it is not a path finding error, we
 		// return it immediately.
 		if err != nil && !lnutils.ErrorAs[*pathFindingError](err) {

routing/payment_session_test.go

@@ -259,8 +259,8 @@ func (g *sessionGraph) sourceNode() route.Vertex {
 	return route.Vertex{}
 }
 
-func (g *sessionGraph) GraphSession(
-	cb func(graph graphdb.NodeTraverser) error) error {
+func (g *sessionGraph) GraphSession(cb func(graph graphdb.NodeTraverser) error,
+	_ func()) error {
 
 	return cb(g)
 }

routing/unified_edges.go

@@ -97,6 +97,7 @@ func (u *nodeEdgeUnifier) addPolicy(fromNode route.Vertex,
 // addGraphPolicies adds all policies that are known for the toNode in the
 // graph.
 func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error {
+	var channels []*graphdb.DirectedChannel
 	cb := func(channel *graphdb.DirectedChannel) error {
 		// If there is no edge policy for this candidate node, skip.
 		// Note that we are searching backwards so this node would have
@@ -108,6 +109,22 @@ func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error {
 			return nil
 		}
 
+		channels = append(channels, channel)
+
+		return nil
+	}
+
+	// Iterate over all channels of the to node.
+	err := g.ForEachNodeDirectedChannel(
+		u.toNode, cb, func() {
+			channels = nil
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	for _, channel := range channels {
 		// Add this policy to the corresponding edgeUnifier. We default
 		// to the clear hop payload size function because
 		// `addGraphPolicies` is only used for cleartext intermediate
@@ -120,12 +137,9 @@ func (u *nodeEdgeUnifier) addGraphPolicies(g Graph) error {
 			channel.OtherNode, channel.InPolicy, inboundFee,
 			channel.Capacity, defaultHopPayloadSize, nil,
 		)
-
-		return nil
 	}
 
-	// Iterate over all channels of the to node.
-	return g.ForEachNodeDirectedChannel(u.toNode, cb)
+	return nil
 }
 
 // unifiedEdge is the individual channel data that is kept inside an edgeUnifier

rpcserver.go

@@ -6745,6 +6745,8 @@ func (r *rpcServer) DescribeGraph(ctx context.Context,
 		resp.Nodes = append(resp.Nodes, lnNode)
 
 		return nil
+	}, func() {
+		resp.Nodes = nil
 	})
 	if err != nil {
 		return nil, err
@@ -6768,6 +6770,8 @@ func (r *rpcServer) DescribeGraph(ctx context.Context,
 		resp.Edges = append(resp.Edges, edge)
 
 		return nil
+	}, func() {
+		resp.Edges = nil
 	})
 	if err != nil && !errors.Is(err, graphdb.ErrGraphNoEdgesFound) {
 		return nil, err
@@ -7046,7 +7050,8 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
 		channels      []*lnrpc.ChannelEdge
 	)
 
-	err = graph.ForEachNodeChannel(ctx, node.PubKeyBytes,
+	err = graph.ForEachNodeChannel(
+		ctx, node.PubKeyBytes,
 		func(edge *models.ChannelEdgeInfo,
 			c1, c2 *models.ChannelEdgePolicy) error {
 
@@ -7072,6 +7077,10 @@ func (r *rpcServer) GetNodeInfo(ctx context.Context,
 			}
 
 			return nil
+		}, func() {
+			numChannels = 0
+			totalCapacity = 0
+			channels = nil
 		},
 	)
 	if err != nil {
@@ -7208,6 +7217,15 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context,
 		}
 
 		return nil
+	}, func() {
+		numChannels = 0
+		numNodes = 0
+		maxChanOut = 0
+		totalNetworkCapacity = 0
+		minChannelSize = math.MaxInt64
+		maxChannelSize = 0
+		allChans = nil
+		clear(seenChans)
 	})
 	if err != nil {
 		return nil, err
@@ -7728,7 +7746,8 @@ func (r *rpcServer) FeeReport(ctx context.Context,
 	}
 
 	var feeReports []*lnrpc.ChannelFeeReport
-	err = channelGraph.ForEachNodeChannel(ctx, selfNode.PubKeyBytes,
+	err = channelGraph.ForEachNodeChannel(
+		ctx, selfNode.PubKeyBytes,
 		func(chanInfo *models.ChannelEdgeInfo,
 			edgePolicy, _ *models.ChannelEdgePolicy) error {
 
@@ -7766,6 +7785,8 @@ func (r *rpcServer) FeeReport(ctx context.Context,
 			})
 
 			return nil
+		}, func() {
+			feeReports = nil
 		},
 	)
 	if err != nil {

server.go

@@ -1254,7 +1254,8 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 		DefaultRoutingPolicy: cc.RoutingPolicy,
 		ForAllOutgoingChannels: func(ctx context.Context,
 			cb func(*models.ChannelEdgeInfo,
-				*models.ChannelEdgePolicy) error) error {
+				*models.ChannelEdgePolicy) error,
+			reset func()) error {
 
 			return s.graphDB.ForEachNodeChannel(ctx, selfVertex,
 				func(c *models.ChannelEdgeInfo,
@@ -1264,7 +1265,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 					// NOTE: The invoked callback here may
 					// receive a nil channel policy.
 					return cb(c, e)
-				},
+				}, reset,
 			)
 		},
 		PropagateChanPolicyUpdate: s.authGossiper.PropagateChanPolicyUpdate,
@@ -3599,7 +3600,7 @@ 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.
-	nodeAddrsMap := map[string]*nodeAddresses{}
+	nodeAddrsMap := make(map[string]*nodeAddresses)
 
 	// Iterate through the list of LinkNodes to find addresses we should
 	// attempt to connect to based on our set of previous connections. Set
@@ -3623,6 +3624,7 @@ func (s *server) establishPersistentConnections(ctx context.Context) error {
 	// that have been added via NodeAnnouncement messages.
 	// TODO(roasbeef): instead iterate over link nodes and query graph for
 	// each of the nodes.
+	graphAddrs := make(map[string]*nodeAddresses)
 	forEachSrcNodeChan := func(chanPoint wire.OutPoint,
 		havePolicy bool, channelPeer *models.LightningNode) error {
 
@@ -3688,10 +3690,14 @@ func (s *server) establishPersistentConnections(ctx context.Context) error {
 			return err
 		}
 
-		nodeAddrsMap[pubStr] = n
+		graphAddrs[pubStr] = n
 		return nil
 	}
-	err = s.graphDB.ForEachSourceNodeChannel(ctx, forEachSrcNodeChan)
+	err = s.graphDB.ForEachSourceNodeChannel(
+		ctx, forEachSrcNodeChan, func() {
+			clear(graphAddrs)
+		},
+	)
 	if err != nil {
 		srvrLog.Errorf("Failed to iterate over source node channels: "+
 			"%v", err)
@@ -3703,6 +3709,11 @@ func (s *server) establishPersistentConnections(ctx context.Context) error {
 		}
 	}
 
+	// Combine the addresses from the link nodes and the channel graph.
+	for pubStr, nodeAddr := range graphAddrs {
+		nodeAddrsMap[pubStr] = nodeAddr
+	}
+
 	srvrLog.Debugf("Establishing %v persistent connections on start",
 		len(nodeAddrsMap))