From 70b8c8ec78b79fe4036a136b7dd579580fb4a215 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 12:01:10 +0200 Subject: [PATCH 1/7] autopilot: convert Peer to route.Vertex In preparation for removing the ForEachNode/ForEachChannel call pattern, we refactor the autopilot subserver by simplifying the `ChannelEdge` such that it no longer returns the `Node` type which has a `ForEachChannel` method on it. The aim is to completely remove the `ForEachChannel` usage. --- autopilot/graph.go | 8 ++------ autopilot/interface.go | 2 +- autopilot/prefattach_test.go | 20 +++++++------------- autopilot/simple_graph.go | 12 ++++++++---- 4 files changed, 18 insertions(+), 24 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index f1506b5d7..eba26c99a 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -106,9 +106,7 @@ func (d *dbNode) ForEachChannel(ctx context.Context, edge := ChannelEdge{ ChanID: lnwire.NewShortChanIDFromInt(ep.ChannelID), Capacity: ei.Capacity, - Peer: &dbNode{ - tx: node, - }, + Peer: node.Node().PubKeyBytes, } return cb(ctx, edge) @@ -198,9 +196,7 @@ func (nc dbNodeCached) ForEachChannel(ctx context.Context, edge := ChannelEdge{ ChanID: lnwire.NewShortChanIDFromInt(cid), Capacity: channel.Capacity, - Peer: dbNodeCached{ - node: channel.OtherNode, - }, + Peer: channel.OtherNode, } if err := cb(ctx, edge); err != nil { diff --git a/autopilot/interface.go b/autopilot/interface.go index a4efa2eec..5caef98c0 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -72,7 +72,7 @@ type ChannelEdge struct { // Peer is the peer that this channel creates an edge to in the channel // graph. - Peer Node + Peer route.Vertex } // ChannelGraph in an interface that represents a traversable channel graph. diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index b5c0b1d26..8ebb1419f 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -175,8 +175,8 @@ func TestPrefAttachmentSelectTwoVertexes(t *testing.T) { // The candidates should be amongst the two edges // created above. for nodeID, candidate := range candidates { - edge1Pub := edge1.Peer.PubKey() - edge2Pub := edge2.Peer.PubKey() + edge1Pub := edge1.Peer + edge2Pub := edge2.Peer switch { case bytes.Equal(nodeID[:], edge1Pub[:]): @@ -228,7 +228,7 @@ func TestPrefAttachmentSelectGreedyAllocation(t *testing.T) { ) require.NoError(t1, err) - peerPubBytes := edge1.Peer.PubKey() + peerPubBytes := edge1.Peer peerPub, err := btcec.ParsePubKey(peerPubBytes[:]) require.NoError(t1, err) @@ -535,18 +535,12 @@ func (d *testDBGraph) addRandChannel(node1, node2 *btcec.PublicKey, return &ChannelEdge{ ChanID: chanID, Capacity: capacity, - Peer: &dbNode{tx: &testNodeTx{ - db: d, - node: vertex1, - }}, + Peer: vertex1.PubKeyBytes, }, &ChannelEdge{ ChanID: chanID, Capacity: capacity, - Peer: &dbNode{tx: &testNodeTx{ - db: d, - node: vertex2, - }}, + Peer: vertex2.PubKeyBytes, }, nil } @@ -692,14 +686,14 @@ func (m *memChannelGraph) addRandChannel(node1, node2 *btcec.PublicKey, edge1 := ChannelEdge{ ChanID: randChanID(), Capacity: capacity, - Peer: vertex2, + Peer: vertex2.PubKey(), } vertex1.chans = append(vertex1.chans, edge1) edge2 := ChannelEdge{ ChanID: randChanID(), Capacity: capacity, - Peer: vertex1, + Peer: vertex1.PubKey(), } vertex2.chans = append(vertex2.chans, edge2) diff --git a/autopilot/simple_graph.go b/autopilot/simple_graph.go index 2f80c490a..32c4559c8 100644 --- a/autopilot/simple_graph.go +++ b/autopilot/simple_graph.go @@ -1,6 +1,10 @@ package autopilot -import "context" +import ( + "context" + + "github.com/lightningnetwork/lnd/routing/route" +) // diameterCutoff is used to discard nodes in the diameter calculation. // It is the multiplier for the eccentricity of the highest-degree node, @@ -31,8 +35,8 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) { // The returned index is then used to create a simplified adjacency list // where each node is identified by its index instead of its pubkey, and // also to create a mapping from node index to node pubkey. - getNodeIndex := func(node Node) int { - key := NodeID(node.PubKey()) + getNodeIndex := func(node route.Vertex) int { + key := NodeID(node) nodeIndex, ok := nodes[key] if !ok { @@ -47,7 +51,7 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) { // Iterate over each node and each channel and update the adj and the // node index. err := g.ForEachNode(ctx, func(ctx context.Context, node Node) error { - u := getNodeIndex(node) + u := getNodeIndex(node.PubKey()) return node.ForEachChannel( ctx, func(_ context.Context, From ae566744c49410f9684fa3c1811ac415b7bbb3ba Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 16:48:22 +0200 Subject: [PATCH 2/7] graph/db+autopilot: remove NodeRTx FetchNode method In this commit, we remove the need for the NodeRTx interface to have the FetchNode method on it. --- autopilot/graph.go | 7 +------ autopilot/prefattach_test.go | 12 ------------ graph/db/interfaces.go | 6 ------ graph/db/kv_store.go | 14 -------------- graph/db/sql_store.go | 17 ----------------- 5 files changed, 1 insertion(+), 55 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index eba26c99a..992f2752f 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -98,15 +98,10 @@ func (d *dbNode) ForEachChannel(ctx context.Context, return nil } - node, err := d.tx.FetchNode(ep.ToNode) - if err != nil { - return err - } - edge := ChannelEdge{ ChanID: lnwire.NewShortChanIDFromInt(ep.ChannelID), Capacity: ei.Capacity, - Peer: node.Node().PubKeyBytes, + Peer: ep.ToNode, } return cb(ctx, edge) diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index 8ebb1419f..58e5f8f12 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -742,16 +742,4 @@ func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, ) } -func (t *testNodeTx) FetchNode(pub route.Vertex) (graphdb.NodeRTx, error) { - node, err := t.db.db.FetchLightningNode(context.Background(), pub) - if err != nil { - return nil, err - } - - return &testNodeTx{ - db: t.db, - node: node, - }, nil -} - var _ graphdb.NodeRTx = (*testNodeTx)(nil) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index dbcd5d1e6..19b2e3104 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -25,12 +25,6 @@ type NodeRTx interface { // the same transaction used to fetch the node. ForEachChannel(func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error - - // FetchNode fetches the node with the given pub key under the same - // transaction used to fetch the current node. The returned node is also - // a NodeRTx and any operations on that NodeRTx will also be done under - // the same transaction. - FetchNode(node route.Vertex) (NodeRTx, error) } // NodeTraverser is an abstract read only interface that provides information diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 7bac0ec13..3d1129654 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -4902,20 +4902,6 @@ func (c *chanGraphNodeTx) Node() *models.LightningNode { return c.node } -// FetchNode fetches the node with the given pub key under the same transaction -// used to fetch the current node. The returned node is also a NodeRTx and any -// operations on that NodeRTx will also be done under the same transaction. -// -// NOTE: This is a part of the NodeRTx interface. -func (c *chanGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { - node, err := c.db.FetchLightningNodeTx(c.tx, nodePub) - if err != nil { - return nil, err - } - - return newChanGraphNodeTx(c.tx, c.db, node), nil -} - // ForEachChannel can be used to iterate over the node's channels under // the same transaction used to fetch the node. // diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 9d439f448..0b5e8be6a 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -865,23 +865,6 @@ func (s *sqlGraphNodeTx) ForEachChannel(cb func(*models.ChannelEdgeInfo, return forEachNodeChannel(ctx, s.db, s.cfg, s.id, cb) } -// FetchNode fetches the node with the given pub key under the same transaction -// used to fetch the current node. The returned node is also a NodeRTx and any -// operations on that NodeRTx will also be done under the same transaction. -// -// NOTE: This is a part of the NodeRTx interface. -func (s *sqlGraphNodeTx) FetchNode(nodePub route.Vertex) (NodeRTx, error) { - ctx := context.TODO() - - id, node, err := getNodeByPubKey(ctx, s.db, nodePub) - if err != nil { - return nil, fmt.Errorf("unable to fetch V1 node(%x): %w", - nodePub, err) - } - - return newSQLGraphNodeTx(s.db, s.cfg, id, node), nil -} - // ForEachNodeDirectedChannel iterates through all channels of a given node, // executing the passed callback on the directed edge representing the channel // and its incoming policy. If the callback returns an error, then the iteration From 5727bfa688261ed747f045295cb860173914dfe5 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 17:05:52 +0200 Subject: [PATCH 3/7] graph/db: let ForEachNodeCached maybe fetch node addresses Here we adjust the ForEachNodeCached graph DB method to pass in a node's addresses into the provided call-back if requested. This will allow us to improve the performance of node/channel iteration in the autopilot subserver. --- autopilot/graph.go | 4 +++- autopilot/interface.go | 6 ++++-- graph/db/benchmark_test.go | 5 ++++- graph/db/graph.go | 19 +++++++++++++------ graph/db/graph_test.go | 21 ++++++++++++++++++++- graph/db/interfaces.go | 13 ++++++++++--- graph/db/kv_store.go | 13 +++++++++---- graph/db/sql_store.go | 38 +++++++++++++++++++++++++++++++------- rpcserver.go | 3 ++- 9 files changed, 96 insertions(+), 26 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index 992f2752f..d9913fb0f 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -210,7 +210,8 @@ func (nc dbNodeCached) ForEachChannel(ctx context.Context, func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, cb func(context.Context, Node) error, reset func()) error { - return dc.db.ForEachNodeCached(ctx, func(n route.Vertex, + return dc.db.ForEachNodeCached(ctx, false, func(ctx context.Context, + n route.Vertex, _ []net.Addr, channels map[uint64]*graphdb.DirectedChannel) error { if len(channels) > 0 { @@ -221,6 +222,7 @@ func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, return cb(ctx, node) } + return nil }, reset) } diff --git a/autopilot/interface.go b/autopilot/interface.go index 5caef98c0..7a543c0d8 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -235,7 +235,9 @@ type GraphSource interface { // 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, + ForEachNodeCached(ctx context.Context, withAddrs bool, + cb func(ctx context.Context, node route.Vertex, + addrs []net.Addr, + chans map[uint64]*graphdb.DirectedChannel) error, reset func()) error } diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index f1366459b..e5fd6c256 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "fmt" + "net" "os" "path" "sync" @@ -721,7 +722,9 @@ func BenchmarkGraphReadMethods(b *testing.B) { fn: func(b testing.TB, store V1Store) { //nolint:ll err := store.ForEachNodeCached( - ctx, func(route.Vertex, + ctx, false, func(context.Context, + route.Vertex, + []net.Addr, map[uint64]*DirectedChannel) error { // Increment the counter to diff --git a/graph/db/graph.go b/graph/db/graph.go index a50715774..7ad87ed27 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "net" "sync" "sync/atomic" "testing" @@ -248,15 +249,21 @@ func (c *ChannelGraph) GraphSession(cb func(graph NodeTraverser) error, // graph, executing the passed callback with each node encountered. // // 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, - reset func()) error { +func (c *ChannelGraph) ForEachNodeCached(ctx context.Context, withAddrs bool, + cb func(ctx context.Context, node route.Vertex, addrs []net.Addr, + chans map[uint64]*DirectedChannel) error, reset func()) error { - if c.graphCache != nil { - return c.graphCache.ForEachNode(cb) + if !withAddrs && c.graphCache != nil { + return c.graphCache.ForEachNode( + func(node route.Vertex, + channels map[uint64]*DirectedChannel) error { + + return cb(ctx, node, nil, channels) + }, + ) } - return c.V1Store.ForEachNodeCached(ctx, cb, reset) + return c.V1Store.ForEachNodeCached(ctx, withAddrs, 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 3c60826e1..ef7ec4738 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -252,6 +252,24 @@ func TestNodeInsertionAndDeletion(t *testing.T) { dbNode, err = graph.FetchLightningNode(ctx, testPub) require.NoError(t, err) require.Equal(t, expAddrs, dbNode.Addresses) + + // Also check that the withAddr param of ForEachNodeCached correctly + // returns the addresses we expect for this node. + err = graph.ForEachNodeCached( + ctx, true, func(ctx context.Context, node route.Vertex, + addrs []net.Addr, + chans map[uint64]*DirectedChannel) error { + + if node != dbNode.PubKeyBytes { + return nil + } + + require.Equal(t, expAddrs, addrs) + + return nil + }, func() {}, + ) + require.NoError(t, err) } // TestPartialNode checks that we can add and retrieve a LightningNode where @@ -1359,7 +1377,8 @@ func TestGraphTraversal(t *testing.T) { // set of channels (to force the fall back), we should find all the // channel as well as the nodes included. graph.graphCache = nil - err := graph.ForEachNodeCached(ctx, func(node route.Vertex, + err := graph.ForEachNodeCached(ctx, false, func(_ context.Context, + node route.Vertex, _ []net.Addr, chans map[uint64]*DirectedChannel) error { if _, ok := nodeIndex[node]; !ok { diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 19b2e3104..3dadd144c 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -83,11 +83,18 @@ type V1Store interface { //nolint:interfacebloat *models.ChannelEdgePolicy) error, reset func()) error // ForEachNodeCached is similar to forEachNode, but it returns - // DirectedChannel data to the call-back. + // DirectedChannel data to the call-back. If withAddrs is true, then + // the call-back will also be provided with the addresses associated + // with the node. The address retrieval will likely result in an + // additional round-trip to the database, so it should only be used if + // the addresses are actually needed. // // NOTE: The callback contents MUST not be modified. - ForEachNodeCached(ctx context.Context, cb func(node route.Vertex, - chans map[uint64]*DirectedChannel) error, reset func()) error + ForEachNodeCached(ctx context.Context, withAddrs bool, + cb func(ctx context.Context, node route.Vertex, + addrs []net.Addr, + 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 diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 3d1129654..df761cfb2 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -674,9 +674,9 @@ func (c *KVStore) FetchNodeFeatures(nodePub route.Vertex) ( // data to the call-back. // // NOTE: The callback contents MUST not be modified. -func (c *KVStore) ForEachNodeCached(_ context.Context, - cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error, - reset func()) error { +func (c *KVStore) ForEachNodeCached(ctx context.Context, withAddrs bool, + cb func(ctx context.Context, node route.Vertex, addrs []net.Addr, + 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 @@ -736,7 +736,12 @@ func (c *KVStore) ForEachNodeCached(_ context.Context, return err } - return cb(node.PubKeyBytes, channels) + var addrs []net.Addr + if withAddrs { + addrs = node.Addresses + } + + return cb(ctx, node.PubKeyBytes, addrs, channels) }, reset) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 0b5e8be6a..dc1c1c460 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1058,17 +1058,19 @@ func (s *SQLStore) ChanUpdatesInHorizon(startTime, } // ForEachNodeCached is similar to forEachNode, but it returns DirectedChannel -// data to the call-back. -// -// NOTE: The callback contents MUST not be modified. +// data to the call-back. If withAddrs is true, then the call-back will also be +// provided with the addresses associated with the node. The address retrieval +// result in an additional round-trip to the database, so it should only be used +// if the addresses are actually needed. // // NOTE: part of the V1Store interface. -func (s *SQLStore) ForEachNodeCached(ctx context.Context, - cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error, - reset func()) error { +func (s *SQLStore) ForEachNodeCached(ctx context.Context, withAddrs bool, + cb func(ctx context.Context, node route.Vertex, addrs []net.Addr, + chans map[uint64]*DirectedChannel) error, reset func()) error { type nodeCachedBatchData struct { features map[int64][]int + addrs map[int64][]nodeAddress chanBatchData *batchChannelData chanMap map[int64][]sqlc.ListChannelsForNodeIDsRow } @@ -1101,6 +1103,19 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, "node features: %w", err) } + // Maybe fetch the node's addresses if requested. + var nodeAddrs map[int64][]nodeAddress + if withAddrs { + nodeAddrs, err = batchLoadNodeAddressesHelper( + ctx, s.cfg.QueryCfg, db, nodeIDs, + ) + if err != nil { + return nil, fmt.Errorf("unable to "+ + "batch load node "+ + "addresses: %w", err) + } + } + // Batch load ALL unique channels for ALL nodes in this // page. allChannels, err := db.ListChannelsForNodeIDs( @@ -1189,6 +1204,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, return &nodeCachedBatchData{ features: nodeFeatures, + addrs: nodeAddrs, chanBatchData: channelBatchData, chanMap: nodeChannelMap, }, nil @@ -1232,7 +1248,15 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, channels[directedChan.ChannelID] = directedChan } - return cb(nodePub, channels) + addrs, err := buildNodeAddresses( + batchData.addrs[nodeData.ID], + ) + if err != nil { + return fmt.Errorf("unable to build node "+ + "addresses: %w", err) + } + + return cb(ctx, nodePub, addrs, channels) } return sqldb.ExecuteCollectAndBatchWithSharedDataQuery( diff --git a/rpcserver.go b/rpcserver.go index 9b8bedf69..a4e83fe8f 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -7183,7 +7183,8 @@ func (r *rpcServer) GetNetworkInfo(ctx context.Context, // network, tallying up the total number of nodes, and also gathering // each node so we can measure the graph diameter and degree stats // below. - err := graph.ForEachNodeCached(ctx, func(node route.Vertex, + err := graph.ForEachNodeCached(ctx, false, func(ctx context.Context, + node route.Vertex, _ []net.Addr, edges map[uint64]*graphdb.DirectedChannel) error { // Increment the total number of nodes with each iteration. From ce7fe84da73f3edf32c16ddc39903776543493ae Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 17:18:35 +0200 Subject: [PATCH 4/7] autopilot: add ForEachNodesChannels method to ChannelGraph interface This is in preparation for removing the ForEachChannel method from the Node interface. --- autopilot/graph.go | 91 ++++++++++++++++++++++++++++++++++-- autopilot/interface.go | 8 ++++ autopilot/prefattach_test.go | 23 +++++++++ 3 files changed, 118 insertions(+), 4 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index d9913fb0f..68b964776 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -51,7 +51,9 @@ func ChannelGraphFromDatabase(db GraphSource) ChannelGraph { // channeldb.LightningNode. The wrapper method implement the autopilot.Node // interface. type dbNode struct { - tx graphdb.NodeRTx + tx graphdb.NodeRTx + pub [33]byte + addrs []net.Addr } // A compile time assertion to ensure dbNode meets the autopilot.Node @@ -64,7 +66,7 @@ var _ Node = (*dbNode)(nil) // // NOTE: Part of the autopilot.Node interface. func (d *dbNode) PubKey() [33]byte { - return d.tx.Node().PubKeyBytes + return d.pub } // Addrs returns a slice of publicly reachable public TCP addresses that the @@ -72,7 +74,7 @@ func (d *dbNode) PubKey() [33]byte { // // NOTE: Part of the autopilot.Node interface. func (d *dbNode) Addrs() []net.Addr { - return d.tx.Node().Addresses + return d.addrs } // ForEachChannel is a higher-order function that will be used to iterate @@ -125,13 +127,56 @@ func (d *databaseChannelGraph) ForEachNode(ctx context.Context, } node := &dbNode{ - tx: nodeTx, + tx: nodeTx, + pub: nodeTx.Node().PubKeyBytes, + addrs: nodeTx.Node().Addresses, } return cb(ctx, node) }, reset) } +// ForEachNodesChannels iterates through all connected nodes, and for each node, +// all the channels that connect to it. The passed callback will be called with +// the context, the Node itself, and a slice of ChannelEdge that connect to the +// node. +// +// NOTE: Part of the autopilot.ChannelGraph interface. +func (d *databaseChannelGraph) ForEachNodesChannels(ctx context.Context, + cb func(context.Context, Node, []*ChannelEdge) error, + reset func()) error { + + return d.db.ForEachNodeCached( + ctx, true, func(ctx context.Context, node route.Vertex, + addrs []net.Addr, + chans map[uint64]*graphdb.DirectedChannel) error { + + // We'll skip over any node that doesn't have any + // advertised addresses. As we won't be able to reach + // them to actually open any channels. + if len(addrs) == 0 { + return nil + } + + edges := make([]*ChannelEdge, 0, len(chans)) + for _, channel := range chans { + edges = append(edges, &ChannelEdge{ + ChanID: lnwire.NewShortChanIDFromInt( + channel.ChannelID, + ), + Capacity: channel.Capacity, + Peer: channel.OtherNode, + }) + } + + return cb(ctx, &dbNode{ + pub: node, + addrs: addrs, + }, edges) + }, reset, + ) +} + // databaseChannelGraphCached wraps a channeldb.ChannelGraph instance with the // necessary API to properly implement the autopilot.ChannelGraph interface. type databaseChannelGraphCached struct { @@ -227,6 +272,44 @@ func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, }, reset) } +// ForEachNodesChannels iterates through all connected nodes, and for each node, +// all the channels that connect to it. The passed callback will be called with +// the context, the Node itself, and a slice of ChannelEdge that connect to the +// node. +// +// NOTE: Part of the autopilot.ChannelGraph interface. +func (dc *databaseChannelGraphCached) ForEachNodesChannels(ctx context.Context, + cb func(context.Context, Node, []*ChannelEdge) error, + reset func()) error { + + return dc.db.ForEachNodeCached(ctx, false, func(ctx context.Context, + n route.Vertex, _ []net.Addr, + channels map[uint64]*graphdb.DirectedChannel) error { + + edges := make([]*ChannelEdge, 0, len(channels)) + for cid, channel := range channels { + edges = append(edges, &ChannelEdge{ + ChanID: lnwire.NewShortChanIDFromInt(cid), + Capacity: channel.Capacity, + Peer: channel.OtherNode, + }) + } + + if len(channels) > 0 { + node := dbNodeCached{ + node: n, + channels: channels, + } + + if err := cb(ctx, node, edges); err != nil { + return err + } + } + + return nil + }, reset) +} + // memNode is a purely in-memory implementation of the autopilot.Node // interface. type memNode struct { diff --git a/autopilot/interface.go b/autopilot/interface.go index 7a543c0d8..f35f319c5 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -87,6 +87,14 @@ type ChannelGraph interface { // callback returns an error, then execution should be terminated. ForEachNode(context.Context, func(context.Context, Node) error, func()) error + + // ForEachNodesChannels iterates through all connected nodes, and for + // each node, all the channels that connect to it. The passed callback + // will be called with the context, the Node itself, and a slice of + // ChannelEdge that connect to the node. + ForEachNodesChannels(ctx context.Context, + cb func(context.Context, Node, []*ChannelEdge) error, + reset func()) error } // NodeScore is a tuple mapping a NodeID to a score indicating the preference diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index 58e5f8f12..a9f67f79a 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -606,6 +606,29 @@ func (m *memChannelGraph) ForEachNode(ctx context.Context, return nil } +// ForEachNodesChannels iterates through all connected nodes, and for each node, +// all the channels that connect to it. The passed callback will be called with +// the context, the Node itself, and a slice of ChannelEdge that connect to the +// node. +// +// NOTE: Part of the autopilot.ChannelGraph interface. +func (m *memChannelGraph) ForEachNodesChannels(ctx context.Context, + cb func(context.Context, Node, []*ChannelEdge) error, _ func()) error { + + for _, node := range m.graph { + edges := make([]*ChannelEdge, 0, len(node.chans)) + for i := range node.chans { + edges = append(edges, &node.chans[i]) + } + + if err := cb(ctx, node, edges); err != nil { + return err + } + } + + return nil +} + // randChanID generates a new random channel ID. func randChanID() lnwire.ShortChannelID { id := atomic.AddUint64(&chanIDCounter, 1) From 699e3359549a32055a8325f1191475b430d3d3ab Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 17:22:56 +0200 Subject: [PATCH 5/7] autopilot: remove the ForEachChannel method from Node interface And instead make use of the new ForEachNodesChannels method which uses a much more efficient method for iterating through nodes&channels. --- autopilot/graph.go | 78 ---------------------- autopilot/interface.go | 7 -- autopilot/prefattach.go | 124 ++++++++++++++++++----------------- autopilot/prefattach_test.go | 22 +++---- autopilot/simple_graph.go | 19 +++--- 5 files changed, 82 insertions(+), 168 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index 68b964776..a2a2e02ff 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -10,7 +10,6 @@ import ( "github.com/btcsuite/btcd/btcec/v2/ecdsa" "github.com/btcsuite/btcd/btcutil" graphdb "github.com/lightningnetwork/lnd/graph/db" - "github.com/lightningnetwork/lnd/graph/db/models" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" ) @@ -51,7 +50,6 @@ func ChannelGraphFromDatabase(db GraphSource) ChannelGraph { // channeldb.LightningNode. The wrapper method implement the autopilot.Node // interface. type dbNode struct { - tx graphdb.NodeRTx pub [33]byte addrs []net.Addr } @@ -77,39 +75,6 @@ func (d *dbNode) Addrs() []net.Addr { return d.addrs } -// ForEachChannel is a higher-order function that will be used to iterate -// through all edges emanating from/to the target node. For each active -// channel, this function should be called with the populated ChannelEdge that -// describes the active channel. -// -// NOTE: Part of the autopilot.Node interface. -func (d *dbNode) ForEachChannel(ctx context.Context, - cb func(context.Context, ChannelEdge) error) error { - - return d.tx.ForEachChannel(func(ei *models.ChannelEdgeInfo, ep, - _ *models.ChannelEdgePolicy) error { - - // Skip channels for which no outgoing edge policy is available. - // - // TODO(joostjager): Ideally the case where channels have a nil - // policy should be supported, as autopilot is not looking at - // the policies. For now, it is not easily possible to get a - // reference to the other end LightningNode object without - // retrieving the policy. - if ep == nil { - return nil - } - - edge := ChannelEdge{ - ChanID: lnwire.NewShortChanIDFromInt(ep.ChannelID), - Capacity: ei.Capacity, - Peer: ep.ToNode, - } - - return cb(ctx, edge) - }) -} - // 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. @@ -127,7 +92,6 @@ func (d *databaseChannelGraph) ForEachNode(ctx context.Context, } node := &dbNode{ - tx: nodeTx, pub: nodeTx.Node().PubKeyBytes, addrs: nodeTx.Node().Addresses, } @@ -223,30 +187,6 @@ func (nc dbNodeCached) Addrs() []net.Addr { return []net.Addr{} } -// ForEachChannel is a higher-order function that will be used to iterate -// through all edges emanating from/to the target node. For each active -// channel, this function should be called with the populated ChannelEdge that -// describes the active channel. -// -// NOTE: Part of the autopilot.Node interface. -func (nc dbNodeCached) ForEachChannel(ctx context.Context, - cb func(context.Context, ChannelEdge) error) error { - - for cid, channel := range nc.channels { - edge := ChannelEdge{ - ChanID: lnwire.NewShortChanIDFromInt(cid), - Capacity: channel.Capacity, - Peer: channel.OtherNode, - } - - if err := cb(ctx, edge); err != nil { - return err - } - } - - return nil -} - // 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. @@ -343,24 +283,6 @@ func (m memNode) Addrs() []net.Addr { return m.addrs } -// ForEachChannel is a higher-order function that will be used to iterate -// through all edges emanating from/to the target node. For each active -// channel, this function should be called with the populated ChannelEdge that -// describes the active channel. -// -// NOTE: Part of the autopilot.Node interface. -func (m memNode) ForEachChannel(ctx context.Context, - cb func(context.Context, ChannelEdge) error) error { - - for _, channel := range m.chans { - if err := cb(ctx, channel); err != nil { - return err - } - } - - return nil -} - // Median returns the median value in the slice of Amounts. func Median(vals []btcutil.Amount) btcutil.Amount { sort.Slice(vals, func(i, j int) bool { diff --git a/autopilot/interface.go b/autopilot/interface.go index f35f319c5..1554b03ee 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -31,13 +31,6 @@ type Node interface { // Addrs returns a slice of publicly reachable public TCP addresses // that the peer is known to be listening on. Addrs() []net.Addr - - // ForEachChannel is a higher-order function that will be used to - // iterate through all edges emanating from/to the target node. For - // each active channel, this function should be called with the - // populated ChannelEdge that describes the active channel. - ForEachChannel(context.Context, func(context.Context, - ChannelEdge) error) error } // LocalChannel is a simple struct which contains relevant details of a diff --git a/autopilot/prefattach.go b/autopilot/prefattach.go index e9a9c53b4..267c13db3 100644 --- a/autopilot/prefattach.go +++ b/autopilot/prefattach.go @@ -89,26 +89,26 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, allChans []btcutil.Amount seenChans = make(map[uint64]struct{}) ) - if err := g.ForEachNode(ctx, func(ctx context.Context, n Node) error { - err := n.ForEachChannel(ctx, func(_ context.Context, - e ChannelEdge) error { + err := g.ForEachNodesChannels( + ctx, func(_ context.Context, node Node, + channels []*ChannelEdge) error { - if _, ok := seenChans[e.ChanID.ToUint64()]; ok { - return nil + for _, e := range channels { + if _, ok := seenChans[e.ChanID.ToUint64()]; ok { + continue + } + seenChans[e.ChanID.ToUint64()] = struct{}{} + allChans = append(allChans, e.Capacity) } - seenChans[e.ChanID.ToUint64()] = struct{}{} - allChans = append(allChans, e.Capacity) - return nil - }) - if err != nil { - return err - } - return nil - }, func() { - allChans = nil - clear(seenChans) - }); err != nil { + return nil + }, + func() { + allChans = nil + clear(seenChans) + }, + ) + if err != nil { return nil, err } @@ -120,55 +120,59 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, // the graph. var maxChans int nodeChanNum := make(map[NodeID]int) - if err := g.ForEachNode(ctx, func(ctx context.Context, n Node) error { - var nodeChans int - err := n.ForEachChannel(ctx, func(_ context.Context, - e ChannelEdge) error { + err = g.ForEachNodesChannels( + ctx, func(ctx context.Context, node Node, + edges []*ChannelEdge) error { - // Since connecting to nodes with a lot of small - // channels actually worsens our connectivity in the - // graph (we will potentially waste time trying to use - // these useless channels in path finding), we decrease - // the counter for such channels. - if e.Capacity < - medianChanSize/minMedianChanSizeFraction { + var nodeChans int + for _, e := range edges { + // Since connecting to nodes with a lot of small + // channels actually worsens our connectivity in + // the graph (we will potentially waste time + // trying to use these useless channels in path + // finding), we decrease the counter for such + // channels. + // + //nolint:ll + if e.Capacity < + medianChanSize/minMedianChanSizeFraction { - nodeChans-- + nodeChans-- + + continue + } + + // Larger channels we count. + nodeChans++ + } + + // We keep track of the highest-degree node we've seen, + // as this will be given the max score. + if nodeChans > maxChans { + maxChans = nodeChans + } + + // If this node is not among our nodes to score, we can + // return early. + nID := NodeID(node.PubKey()) + if _, ok := nodes[nID]; !ok { + log.Tracef("Node %x not among nodes to score, "+ + "ignoring", nID[:]) return nil } - // Larger channels we count. - nodeChans++ + // Otherwise we'll record the number of channels. + nodeChanNum[nID] = nodeChans + log.Tracef("Counted %v channels for node %x", nodeChans, + nID[:]) + return nil - }) - if err != nil { - return err - } - - // We keep track of the highest-degree node we've seen, as this - // will be given the max score. - if nodeChans > maxChans { - maxChans = nodeChans - } - - // If this node is not among our nodes to score, we can return - // early. - nID := NodeID(n.PubKey()) - if _, ok := nodes[nID]; !ok { - log.Tracef("Node %x not among nodes to score, "+ - "ignoring", nID[:]) - return nil - } - - // Otherwise we'll record the number of channels. - nodeChanNum[nID] = nodeChans - log.Tracef("Counted %v channels for node %x", nodeChans, nID[:]) - - return nil - }, func() { - maxChans = 0 - clear(nodeChanNum) - }); err != nil { + }, func() { + maxChans = 0 + clear(nodeChanNum) + }, + ) + if err != nil { return nil, err } diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index a9f67f79a..f0f351384 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -242,25 +242,23 @@ func TestPrefAttachmentSelectGreedyAllocation(t *testing.T) { numNodes := 0 twoChans := false nodes := make(map[NodeID]struct{}) - err = graph.ForEachNode( - ctx, func(ctx context.Context, n Node) error { + err = graph.ForEachNodesChannels( + ctx, func(_ context.Context, node Node, + edges []*ChannelEdge) error { + numNodes++ - nodes[n.PubKey()] = struct{}{} + nodes[node.PubKey()] = struct{}{} numChans := 0 - err := n.ForEachChannel(ctx, - func(_ context.Context, c ChannelEdge) error { //nolint:ll - numChans++ - return nil - }, - ) - if err != nil { - return err + + for range edges { + numChans++ } twoChans = twoChans || (numChans == 2) return nil - }, func() { + }, + func() { numNodes = 0 twoChans = false clear(nodes) diff --git a/autopilot/simple_graph.go b/autopilot/simple_graph.go index 32c4559c8..44f514903 100644 --- a/autopilot/simple_graph.go +++ b/autopilot/simple_graph.go @@ -50,20 +50,17 @@ func NewSimpleGraph(ctx context.Context, g ChannelGraph) (*SimpleGraph, error) { // Iterate over each node and each channel and update the adj and the // node index. - err := g.ForEachNode(ctx, func(ctx context.Context, node Node) error { + err := g.ForEachNodesChannels(ctx, func(_ context.Context, + node Node, channels []*ChannelEdge) error { + u := getNodeIndex(node.PubKey()) - return node.ForEachChannel( - ctx, func(_ context.Context, - edge ChannelEdge) error { + for _, edge := range channels { + v := getNodeIndex(edge.Peer) + adj[u] = append(adj[u], v) + } - v := getNodeIndex(edge.Peer) - - adj[u] = append(adj[u], v) - - return nil - }, - ) + return nil }, func() { clear(adj) clear(nodes) From 15e17e1270be7f920fb517026d53d6bc0115321a Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 17:26:07 +0200 Subject: [PATCH 6/7] graph/db: remove NodeRTx ForEachChannel method --- graph/db/interfaces.go | 5 ----- graph/db/kv_store.go | 21 --------------------- graph/db/sql_store.go | 12 ------------ itest/lnd_graph_migration_test.go | 20 ++++++++++---------- 4 files changed, 10 insertions(+), 48 deletions(-) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 3dadd144c..5ad086d54 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -20,11 +20,6 @@ import ( type NodeRTx interface { // Node returns the raw information of the node. Node() *models.LightningNode - - // ForEachChannel can be used to iterate over the node's channels under - // the same transaction used to fetch the node. - ForEachChannel(func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error } // NodeTraverser is an abstract read only interface that provides information diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index df761cfb2..cf49b0d60 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -4906,24 +4906,3 @@ func newChanGraphNodeTx(tx kvdb.RTx, db *KVStore, func (c *chanGraphNodeTx) Node() *models.LightningNode { return c.node } - -// ForEachChannel can be used to iterate over the node's channels under -// the same transaction used to fetch the node. -// -// NOTE: This is a part of the NodeRTx interface. -func (c *chanGraphNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - - 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_store.go b/graph/db/sql_store.go index dc1c1c460..f6430ba43 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -853,18 +853,6 @@ func (s *sqlGraphNodeTx) Node() *models.LightningNode { return s.node } -// ForEachChannel can be used to iterate over the node's channels under the same -// transaction used to fetch the node. -// -// NOTE: This is a part of the NodeRTx interface. -func (s *sqlGraphNodeTx) ForEachChannel(cb func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - - ctx := context.TODO() - - return forEachNodeChannel(ctx, s.db, s.cfg, s.id, cb) -} - // ForEachNodeDirectedChannel iterates through all channels of a given node, // executing the passed callback on the directed edge representing the channel // and its incoming policy. If the callback returns an error, then the iteration diff --git a/itest/lnd_graph_migration_test.go b/itest/lnd_graph_migration_test.go index fe3e6b175..07abc3bb0 100644 --- a/itest/lnd_graph_migration_test.go +++ b/itest/lnd_graph_migration_test.go @@ -3,12 +3,13 @@ package itest import ( "context" "database/sql" + "net" "github.com/lightningnetwork/lnd" graphdb "github.com/lightningnetwork/lnd/graph/db" - "github.com/lightningnetwork/lnd/graph/db/models" "github.com/lightningnetwork/lnd/lntest" "github.com/lightningnetwork/lnd/lntest/node" + "github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/sqldb" "github.com/stretchr/testify/require" ) @@ -69,19 +70,18 @@ func testGraphMigration(ht *lntest.HarnessTest) { numNodes int edges = make(map[uint64]bool) ) - err := db.ForEachNode(ctx, func(tx graphdb.NodeRTx) error { + err := db.ForEachNodeCached(ctx, false, func(_ context.Context, + _ route.Vertex, _ []net.Addr, + chans map[uint64]*graphdb.DirectedChannel) error { + numNodes++ // For each node, also count the number of edges. - return tx.ForEachChannel( - func(info *models.ChannelEdgeInfo, - _ *models.ChannelEdgePolicy, - _ *models.ChannelEdgePolicy) error { + for _, ch := range chans { + edges[ch.ChannelID] = true + } - edges[info.ChannelID] = true - return nil - }, - ) + return nil }, func() { clear(edges) numNodes = 0 From 8aa8c7cc422bd31e502c4aad47ddeb4757b15cc7 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Sun, 3 Aug 2025 17:33:04 +0200 Subject: [PATCH 7/7] multi: remove the NodeRTx interface type --- autopilot/graph.go | 9 +++---- autopilot/interface.go | 7 +++--- autopilot/prefattach_test.go | 23 ------------------ graph/db/benchmark_test.go | 28 ++++++++++------------ graph/db/graph_test.go | 17 ++++++------- graph/db/interfaces.go | 15 ++---------- graph/db/kv_store.go | 39 ++++-------------------------- graph/db/sql_migration_test.go | 27 +++++++++++---------- graph/db/sql_store.go | 44 ++++------------------------------ rpcserver.go | 4 ++-- 10 files changed, 58 insertions(+), 155 deletions(-) diff --git a/autopilot/graph.go b/autopilot/graph.go index a2a2e02ff..5a1f8dceb 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -10,6 +10,7 @@ import ( "github.com/btcsuite/btcd/btcec/v2/ecdsa" "github.com/btcsuite/btcd/btcutil" graphdb "github.com/lightningnetwork/lnd/graph/db" + "github.com/lightningnetwork/lnd/graph/db/models" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" ) @@ -83,17 +84,17 @@ func (d *dbNode) Addrs() []net.Addr { func (d *databaseChannelGraph) ForEachNode(ctx context.Context, cb func(context.Context, Node) error, reset func()) error { - return d.db.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error { + return d.db.ForEachNode(ctx, func(n *models.LightningNode) error { // We'll skip over any node that doesn't have any advertised // addresses. As we won't be able to reach them to actually // open any channels. - if len(nodeTx.Node().Addresses) == 0 { + if len(n.Addresses) == 0 { return nil } node := &dbNode{ - pub: nodeTx.Node().PubKeyBytes, - addrs: nodeTx.Node().Addresses, + pub: n.PubKeyBytes, + addrs: n.Addresses, } return cb(ctx, node) diff --git a/autopilot/interface.go b/autopilot/interface.go index 1554b03ee..9d9d0d1ca 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -8,6 +8,7 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/btcsuite/btcd/wire" graphdb "github.com/lightningnetwork/lnd/graph/db" + "github.com/lightningnetwork/lnd/graph/db/models" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" ) @@ -228,9 +229,9 @@ type GraphSource interface { // ForEachNode 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. 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, func()) error + // the iteration stops early. + ForEachNode(context.Context, func(*models.LightningNode) error, + func()) error // ForEachNodeCached is similar to ForEachNode, but it utilizes the // channel graph cache if one is available. It is less consistent than diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index f0f351384..d7a578f7d 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -741,26 +741,3 @@ func (m *memChannelGraph) addRandNode() (*btcec.PublicKey, error) { return newPub, nil } - -type testNodeTx struct { - db *testDBGraph - node *models.LightningNode -} - -func (t *testNodeTx) Node() *models.LightningNode { - return t.node -} - -func (t *testNodeTx) ForEachChannel(f func(*models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error) error { - - return t.db.db.ForEachNodeChannel(context.Background(), - t.node.PubKeyBytes, func(edge *models.ChannelEdgeInfo, policy1, - policy2 *models.ChannelEdgePolicy) error { - - return f(edge, policy1, policy2) - }, func() {}, - ) -} - -var _ graphdb.NodeRTx = (*testNodeTx)(nil) diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index e5fd6c256..08dba16de 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -338,12 +338,15 @@ func TestPopulateDBs(t *testing.T) { // graph. countNodes := func(graph *ChannelGraph) int { numNodes := 0 - err := graph.ForEachNode(ctx, func(tx NodeRTx) error { - numNodes++ - return nil - }, func() { - numNodes = 0 - }) + err := graph.ForEachNode( + ctx, func(node *models.LightningNode) error { + numNodes++ + + return nil + }, func() { + numNodes = 0 + }, + ) require.NoError(t, err) return numNodes @@ -487,18 +490,12 @@ func syncGraph(t *testing.T, src, dest *ChannelGraph) { } var wgNodes sync.WaitGroup - err := src.ForEachNode(ctx, func(tx NodeRTx) error { + err := src.ForEachNode(ctx, func(node *models.LightningNode) error { wgNodes.Add(1) go func() { defer wgNodes.Done() - // NOTE: even though the transaction (tx) may have - // already been aborted, it is still ok to use the - // Node() result since that is a static object that - // is not affected by the transaction state. - err := dest.AddLightningNode( - ctx, tx.Node(), batch.LazyAdd(), - ) + err := dest.AddLightningNode(ctx, node, batch.LazyAdd()) require.NoError(t, err) mu.Lock() @@ -658,7 +655,8 @@ func BenchmarkGraphReadMethods(b *testing.B) { name: "ForEachNode", fn: func(b testing.TB, store V1Store) { err := store.ForEachNode( - ctx, func(_ NodeRTx) error { + ctx, + func(_ *models.LightningNode) error { // Increment the counter to // ensure the callback is doing // something. diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index ef7ec4738..b9d972b3b 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1469,8 +1469,8 @@ func TestGraphTraversalCacheable(t *testing.T) { // Create a map of all nodes with the iteration we know works (because // it is tested in another test). nodeMap := make(map[route.Vertex]struct{}) - err := graph.ForEachNode(ctx, func(tx NodeRTx) error { - nodeMap[tx.Node().PubKeyBytes] = struct{}{} + err := graph.ForEachNode(ctx, func(n *models.LightningNode) error { + nodeMap[n.PubKeyBytes] = struct{}{} return nil }, func() {}) @@ -1601,8 +1601,8 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, // Iterate over each node as returned by the graph, if all nodes are // reached, then the map created above should be empty. - err := graph.ForEachNode(ctx, func(tx NodeRTx) error { - delete(nodeIndex, tx.Node().Alias) + err := graph.ForEachNode(ctx, func(n *models.LightningNode) error { + delete(nodeIndex, n.Alias) return nil }, func() {}) require.NoError(t, err) @@ -1710,11 +1710,12 @@ func assertNumChans(t *testing.T, graph *ChannelGraph, n int) { func assertNumNodes(t *testing.T, graph *ChannelGraph, n int) { numNodes := 0 - err := graph.ForEachNode(context.Background(), func(tx NodeRTx) error { - numNodes++ + err := graph.ForEachNode(context.Background(), + func(_ *models.LightningNode) error { + numNodes++ - return nil - }, func() {}) + return nil + }, func() {}) if err != nil { _, _, line, _ := runtime.Caller(1) t.Fatalf("line %v: unable to scan nodes: %v", line, err) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 5ad086d54..7f9b37008 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -14,14 +14,6 @@ import ( "github.com/lightningnetwork/lnd/routing/route" ) -// NodeRTx represents transaction object with an underlying node associated that -// can be used to make further queries to the graph under the same transaction. -// This is useful for consistency during graph traversal and queries. -type NodeRTx interface { - // Node returns the raw information of the node. - Node() *models.LightningNode -} - // NodeTraverser is an abstract read only interface that provides information // about nodes and their edges. The interface is about providing fast read-only // access to the graph and so if a cache is available, it should be used. @@ -94,11 +86,8 @@ type V1Store interface { //nolint:interfacebloat // ForEachNode 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. Any operations performed on the NodeTx - // passed to the call-back are executed under the same read transaction - // and so, methods on the NodeTx object _MUST_ only be called from - // within the call-back. - ForEachNode(ctx context.Context, cb func(tx NodeRTx) error, + // the iteration stops early. + ForEachNode(ctx context.Context, cb func(*models.LightningNode) error, reset func()) error // ForEachNodeCacheable iterates through all the stored vertices/nodes diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index cf49b0d60..49191d7c2 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -800,16 +800,16 @@ func (c *KVStore) DisabledChannelIDs() ([]uint64, error) { // ForEachNode 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. Any operations performed on the NodeTx passed to the call-back are -// executed under the same read transaction and so, methods on the NodeTx object -// _MUST_ only be called from within the call-back. +// early. +// +// NOTE: this is part of the V1Store interface. func (c *KVStore) ForEachNode(_ context.Context, - cb func(tx NodeRTx) error, reset func()) error { + cb func(*models.LightningNode) error, reset func()) error { return forEachNode(c.db, func(tx kvdb.RTx, node *models.LightningNode) error { - return cb(newChanGraphNodeTx(tx, c, node)) + return cb(node) }, reset) } @@ -4877,32 +4877,3 @@ func deserializeChanEdgePolicyRaw(r io.Reader) (*models.ChannelEdgePolicy, return edge, nil } - -// chanGraphNodeTx is an implementation of the NodeRTx interface backed by the -// KVStore and a kvdb.RTx. -type chanGraphNodeTx struct { - tx kvdb.RTx - db *KVStore - node *models.LightningNode -} - -// A compile-time constraint to ensure chanGraphNodeTx implements the NodeRTx -// interface. -var _ NodeRTx = (*chanGraphNodeTx)(nil) - -func newChanGraphNodeTx(tx kvdb.RTx, db *KVStore, - node *models.LightningNode) *chanGraphNodeTx { - - return &chanGraphNodeTx{ - tx: tx, - db: db, - node: node, - } -} - -// Node returns the raw information of the node. -// -// NOTE: This is a part of the NodeRTx interface. -func (c *chanGraphNodeTx) Node() *models.LightningNode { - return c.node -} diff --git a/graph/db/sql_migration_test.go b/graph/db/sql_migration_test.go index 7d6b0e3b0..d134b2fb1 100644 --- a/graph/db/sql_migration_test.go +++ b/graph/db/sql_migration_test.go @@ -395,23 +395,24 @@ func assertInSync(t *testing.T, kvDB *KVStore, sqlDB *SQLStore, func fetchAllNodes(t *testing.T, store V1Store) []*models.LightningNode { nodes := make([]*models.LightningNode, 0) - err := store.ForEachNode(context.Background(), func(tx NodeRTx) error { - node := tx.Node() + err := store.ForEachNode(context.Background(), + func(node *models.LightningNode) error { - // Call PubKey to ensure the objects cached pubkey is set so that - // the objects can be compared as a whole. - _, err := node.PubKey() - require.NoError(t, err) + // Call PubKey to ensure the objects cached pubkey is set so that + // the objects can be compared as a whole. + _, err := node.PubKey() + require.NoError(t, err) - // Sort the addresses to ensure a consistent order. - sortAddrs(node.Addresses) + // Sort the addresses to ensure a consistent order. + sortAddrs(node.Addresses) - nodes = append(nodes, node) + nodes = append(nodes, node) - return nil - }, func() { - nodes = nil - }) + return nil + }, func() { + nodes = nil + }, + ) require.NoError(t, err) // Sort the nodes by their public key to ensure a consistent order. diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index f6430ba43..7a7987855 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -799,60 +799,24 @@ func (s *SQLStore) ForEachSourceNodeChannel(ctx context.Context, // ForEachNode 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. Any operations performed on the NodeTx passed to the call-back are -// executed under the same read transaction and so, methods on the NodeTx object -// _MUST_ only be called from within the call-back. +// early. // // NOTE: part of the V1Store interface. func (s *SQLStore) ForEachNode(ctx context.Context, - cb func(tx NodeRTx) error, reset func()) error { + cb func(node *models.LightningNode) error, reset func()) error { return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { return forEachNodePaginated( ctx, s.cfg.QueryCfg, db, - ProtocolV1, - func(ctx context.Context, dbNodeID int64, + ProtocolV1, func(_ context.Context, _ int64, node *models.LightningNode) error { - return cb(newSQLGraphNodeTx( - db, s.cfg, dbNodeID, node, - )) + return cb(node) }, ) }, reset) } -// sqlGraphNodeTx is an implementation of the NodeRTx interface backed by the -// SQLStore and a SQL transaction. -type sqlGraphNodeTx struct { - db SQLQueries - id int64 - node *models.LightningNode - cfg *SQLStoreConfig -} - -// A compile-time constraint to ensure sqlGraphNodeTx implements the NodeRTx -// interface. -var _ NodeRTx = (*sqlGraphNodeTx)(nil) - -func newSQLGraphNodeTx(db SQLQueries, cfg *SQLStoreConfig, - id int64, node *models.LightningNode) *sqlGraphNodeTx { - - return &sqlGraphNodeTx{ - db: db, - cfg: cfg, - id: id, - node: node, - } -} - -// Node returns the raw information of the node. -// -// NOTE: This is a part of the NodeRTx interface. -func (s *sqlGraphNodeTx) Node() *models.LightningNode { - return s.node -} - // ForEachNodeDirectedChannel iterates through all channels of a given node, // executing the passed callback on the directed edge representing the channel // and its incoming policy. If the callback returns an error, then the iteration diff --git a/rpcserver.go b/rpcserver.go index a4e83fe8f..08f9cf565 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -6757,8 +6757,8 @@ func (r *rpcServer) DescribeGraph(ctx context.Context, // First iterate through all the known nodes (connected or unconnected // within the graph), collating their current state into the RPC // response. - err := graph.ForEachNode(ctx, func(nodeTx graphdb.NodeRTx) error { - lnNode := marshalNode(nodeTx.Node()) + err := graph.ForEachNode(ctx, func(node *models.LightningNode) error { + lnNode := marshalNode(node) resp.Nodes = append(resp.Nodes, lnNode)