multi: reset for ForEachNodeDirectedChannel

graph/db/graph.go

@@ -205,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

graph/db/graph_test.go

@@ -1490,7 +1490,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 			node, func(d *DirectedChannel) error {
 				delete(chanIndex, d.ChannelID)
 				return nil
-			},
+			}, func() {},
 		)
 		require.NoError(t, err)
 
@@ -1499,7 +1499,7 @@ func TestGraphTraversalCacheable(t *testing.T) {
 			node, func(d *DirectedChannel) error {
 				delete(chanIndex2, d.ChannelID)
 				return nil
-			},
+			}, func() {},
 		)
 		require.NoError(t, err)
 	}
@@ -4298,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)

graph/db/kv_store.go

@@ -655,9 +655,7 @@ 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 {
-
-	reset := func() {}
+	cb func(channel *DirectedChannel) error, reset func()) error {
 
 	return c.forEachNodeDirectedChannel(nil, nodePub, cb, reset)
 }
@@ -3999,7 +3997,7 @@ 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, func() {})
 }

graph/db/sql_store.go

@@ -907,9 +907,8 @@ 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 {
 
-	reset := func() {}
 	var ctx = context.TODO()
 
 	return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
@@ -2792,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()
 

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)

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]

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/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