From 83f6dc4983e43a91fdff8326244446649070d127 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Fri, 11 Jul 2025 11:20:25 +0200 Subject: [PATCH] multi: reset for ForEachNodeDirectedChannel --- graph/db/graph.go | 4 ++-- graph/db/graph_test.go | 6 +++--- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 6 ++---- graph/db/sql_store.go | 5 ++--- routing/bandwidth.go | 10 ++++++---- routing/graph.go | 3 ++- routing/mock_graph_test.go | 2 +- routing/pathfind.go | 31 +++++++++++++++++++++++++------ routing/unified_edges.go | 22 ++++++++++++++++++---- 10 files changed, 62 insertions(+), 29 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index e902896d7..a50715774 100644 --- a/graph/db/graph.go +++ b/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 diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index d2f1577b2..25ce1edf7 100644 --- a/graph/db/graph_test.go +++ b/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) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 756d7b1f2..dbcd5d1e6 100644 --- a/graph/db/interfaces.go +++ b/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) diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 8dab73cdf..7bac0ec13 100644 --- a/graph/db/kv_store.go +++ b/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() {}) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 33328c0ba..cd5031f06 100644 --- a/graph/db/sql_store.go +++ b/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() diff --git a/routing/bandwidth.go b/routing/bandwidth.go index f806b03d3..afe085c2e 100644 --- a/routing/bandwidth.go +++ b/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 } diff --git a/routing/graph.go b/routing/graph.go index d5c6b2d7d..4be34def9 100644 --- a/routing/graph.go +++ b/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) diff --git a/routing/mock_graph_test.go b/routing/mock_graph_test.go index b5c8b55bc..d9b16f646 100644 --- a/routing/mock_graph_test.go +++ b/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] diff --git a/routing/pathfind.go b/routing/pathfind.go index b22e7fd1c..fab8015dd 100644 --- a/routing/pathfind.go +++ b/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 { diff --git a/routing/unified_edges.go b/routing/unified_edges.go index bf2f75dbf..f80e1cb1a 100644 --- a/routing/unified_edges.go +++ b/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