diff --git a/autopilot/agent.go b/autopilot/agent.go index 1b70580a3..30e76d3d9 100644 --- a/autopilot/agent.go +++ b/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) } diff --git a/autopilot/graph.go b/autopilot/graph.go index 3aa3ad055..f1506b5d7 100644 --- a/autopilot/graph.go +++ b/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 diff --git a/autopilot/interface.go b/autopilot/interface.go index 703f49020..a4efa2eec 100644 --- a/autopilot/interface.go +++ b/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 } diff --git a/autopilot/prefattach.go b/autopilot/prefattach.go index 76d814a46..e9a9c53b4 100644 --- a/autopilot/prefattach.go +++ b/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 } diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index fd557a02c..b5c0b1d26 100644 --- a/autopilot/prefattach_test.go +++ b/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) { diff --git a/autopilot/simple_graph.go b/autopilot/simple_graph.go index f028db3c7..2f80c490a 100644 --- a/autopilot/simple_graph.go +++ b/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 diff --git a/discovery/bootstrapper.go b/discovery/bootstrapper.go index dbfdbc7d2..1b6c981d4 100644 --- a/discovery/bootstrapper.go +++ b/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...) } diff --git a/discovery/gossiper.go b/discovery/gossiper.go index 70c1caf2e..c4677bbf5 100644 --- a/discovery/gossiper.go +++ b/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", diff --git a/discovery/gossiper_test.go b/discovery/gossiper_test.go index c10bdbbe1..6b9a1d58f 100644 --- a/discovery/gossiper_test.go +++ b/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") diff --git a/graph/builder.go b/graph/builder.go index 073716a83..5cfb79e4c 100644 --- a/graph/builder.go +++ b/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, ) } diff --git a/graph/db/graph.go b/graph/db/graph.go index 3168a6618..a50715774 100644 --- a/graph/db/graph.go +++ b/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 diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index bcafe4554..25ce1edf7 100644 --- a/graph/db/graph_test.go +++ b/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) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index ecb07ec6d..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) @@ -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 diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 6ca9cdedb..7bac0ec13 100644 --- a/graph/db/kv_store.go +++ b/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() {}, ) } diff --git a/graph/db/sql_migration.go b/graph/db/sql_migration.go index dfe5eb985..6cc6ab2ff 100644 --- a/graph/db/sql_migration.go +++ b/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", diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 886055fb2..9f32603fd 100644 --- a/graph/db/sql_migration_test.go +++ b/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. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index a42d4472d..cd5031f06 100644 --- a/graph/db/sql_store.go +++ b/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() diff --git a/graph/interfaces.go b/graph/interfaces.go index 20b63dd15..5494db61a 100644 --- a/graph/interfaces.go +++ b/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. diff --git a/itest/lnd_graph_migration_test.go b/itest/lnd_graph_migration_test.go index 412d23f57..d7417abe4 100644 --- a/itest/lnd_graph_migration_test.go +++ b/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) 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 115f2daf1..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) @@ -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 diff --git a/routing/localchans/manager.go b/routing/localchans/manager.go index faba9fa8b..b1d281187 100644 --- a/routing/localchans/manager.go +++ b/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 } diff --git a/routing/localchans/manager_test.go b/routing/localchans/manager_test.go index 6e53f4a27..09b4f0bc8 100644 --- a/routing/localchans/manager_test.go +++ b/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, ¤tPolicy); err != nil { diff --git a/routing/mock_graph_test.go b/routing/mock_graph_test.go index cb5d07f3a..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] @@ -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) } 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/pathfind_test.go b/routing/pathfind_test.go index fd5f06656..2f746ac9e 100644 --- a/routing/pathfind_test.go +++ b/routing/pathfind_test.go @@ -3303,6 +3303,8 @@ func dbFindPath(graph *graphdb.ChannelGraph, ) return err + }, func() { + route = nil }) if err != nil { return nil, err diff --git a/routing/payment_session.go b/routing/payment_session.go index 607c10443..cf20251ed 100644 --- a/routing/payment_session.go +++ b/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) { diff --git a/routing/payment_session_test.go b/routing/payment_session_test.go index 51eeda425..ffe9d996c 100644 --- a/routing/payment_session_test.go +++ b/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) } 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 diff --git a/rpcserver.go b/rpcserver.go index d4e7e9da0..8c134cbcd 100644 --- a/rpcserver.go +++ b/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 { diff --git a/server.go b/server.go index 4ad5c09e3..7eb8c583d 100644 --- a/server.go +++ b/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))