diff --git a/autopilot/graph.go b/autopilot/graph.go index be6401522..19bd3808a 100644 --- a/autopilot/graph.go +++ b/autopilot/graph.go @@ -101,25 +101,23 @@ func (d *databaseChannelGraph) 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. +// 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's pubkey, 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, + cb func(context.Context, NodeID, []*ChannelEdge) error, reset func()) error { + // The channel-scoring callers only need topology data here. Address + // filtering happens through ForEachNode before connecting to peers. return d.db.ForEachNodeCached( - ctx, true, func(ctx context.Context, node route.Vertex, - addrs []net.Addr, + ctx, func(ctx context.Context, node route.Vertex, 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 { + if len(chans) == 0 { return nil } @@ -134,10 +132,7 @@ func (d *databaseChannelGraph) ForEachNodesChannels(ctx context.Context, }) } - return cb(ctx, &dbNode{ - pub: node, - addrs: addrs, - }, edges) + return cb(ctx, NodeID(node), edges) }, reset, ) } @@ -196,8 +191,8 @@ func (nc dbNodeCached) Addrs() []net.Addr { func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context, cb func(context.Context, Node) error, reset func()) error { - return dc.db.ForEachNodeCached(ctx, false, func(ctx context.Context, - n route.Vertex, _ []net.Addr, + return dc.db.ForEachNodeCached(ctx, func(ctx context.Context, + n route.Vertex, channels map[uint64]*graphdb.DirectedChannel) error { if len(channels) > 0 { @@ -213,20 +208,24 @@ 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. +// 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's pubkey, 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, + cb func(context.Context, NodeID, []*ChannelEdge) error, reset func()) error { - return dc.db.ForEachNodeCached(ctx, false, func(ctx context.Context, - n route.Vertex, _ []net.Addr, + return dc.db.ForEachNodeCached(ctx, func(ctx context.Context, + n route.Vertex, channels map[uint64]*graphdb.DirectedChannel) error { + if len(channels) == 0 { + return nil + } + edges := make([]*ChannelEdge, 0, len(channels)) for cid, channel := range channels { edges = append(edges, &ChannelEdge{ @@ -236,18 +235,7 @@ func (dc *databaseChannelGraphCached) ForEachNodesChannels(ctx context.Context, }) } - if len(channels) > 0 { - node := dbNodeCached{ - node: n, - channels: channels, - } - - if err := cb(ctx, node, edges); err != nil { - return err - } - } - - return nil + return cb(ctx, NodeID(n), edges) }, reset) } diff --git a/autopilot/interface.go b/autopilot/interface.go index 215f92035..71813232a 100644 --- a/autopilot/interface.go +++ b/autopilot/interface.go @@ -84,10 +84,10 @@ type ChannelGraph interface { // 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 + // will be called with the context, the node's pubkey, and a slice of // ChannelEdge that connect to the node. ForEachNodesChannels(ctx context.Context, - cb func(context.Context, Node, []*ChannelEdge) error, + cb func(context.Context, NodeID, []*ChannelEdge) error, reset func()) error } @@ -237,9 +237,8 @@ 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, withAddrs bool, + ForEachNodeCached(ctx context.Context, cb func(ctx context.Context, node route.Vertex, - addrs []net.Addr, chans map[uint64]*graphdb.DirectedChannel) error, reset func()) error } diff --git a/autopilot/prefattach.go b/autopilot/prefattach.go index 267c13db3..7d36674dc 100644 --- a/autopilot/prefattach.go +++ b/autopilot/prefattach.go @@ -90,7 +90,7 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, seenChans = make(map[uint64]struct{}) ) err := g.ForEachNodesChannels( - ctx, func(_ context.Context, node Node, + ctx, func(_ context.Context, node NodeID, channels []*ChannelEdge) error { for _, e := range channels { @@ -121,7 +121,7 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, var maxChans int nodeChanNum := make(map[NodeID]int) err = g.ForEachNodesChannels( - ctx, func(ctx context.Context, node Node, + ctx, func(ctx context.Context, node NodeID, edges []*ChannelEdge) error { var nodeChans int @@ -154,17 +154,16 @@ func (p *PrefAttachment) NodeScores(ctx context.Context, g ChannelGraph, // If this node is not among our nodes to score, we can // return early. - nID := NodeID(node.PubKey()) - if _, ok := nodes[nID]; !ok { + if _, ok := nodes[node]; !ok { log.Tracef("Node %x not among nodes to score, "+ - "ignoring", nID[:]) + "ignoring", node[:]) return nil } // Otherwise we'll record the number of channels. - nodeChanNum[nID] = nodeChans + nodeChanNum[node] = nodeChans log.Tracef("Counted %v channels for node %x", nodeChans, - nID[:]) + node[:]) return nil }, func() { diff --git a/autopilot/prefattach_test.go b/autopilot/prefattach_test.go index efddfb81e..cdaec8746 100644 --- a/autopilot/prefattach_test.go +++ b/autopilot/prefattach_test.go @@ -246,11 +246,11 @@ func TestPrefAttachmentSelectGreedyAllocation(t *testing.T) { twoChans := false nodes := make(map[NodeID]struct{}) err = graph.ForEachNodesChannels( - ctx, func(_ context.Context, node Node, + ctx, func(_ context.Context, node NodeID, edges []*ChannelEdge) error { numNodes++ - nodes[node.PubKey()] = struct{}{} + nodes[node] = struct{}{} numChans := 0 for range edges { @@ -619,14 +619,15 @@ 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. +// 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's pubkey, 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 { + cb func(context.Context, NodeID, []*ChannelEdge) error, + _ func()) error { for _, node := range m.graph { edges := make([]*ChannelEdge, 0, len(node.chans)) @@ -634,7 +635,7 @@ func (m *memChannelGraph) ForEachNodesChannels(ctx context.Context, edges = append(edges, &node.chans[i]) } - if err := cb(ctx, node, edges); err != nil { + if err := cb(ctx, NewNodeID(node.pub), edges); err != nil { return err } } diff --git a/autopilot/simple_graph.go b/autopilot/simple_graph.go index 44f514903..d6072cd53 100644 --- a/autopilot/simple_graph.go +++ b/autopilot/simple_graph.go @@ -2,8 +2,6 @@ package autopilot import ( "context" - - "github.com/lightningnetwork/lnd/routing/route" ) // diameterCutoff is used to discard nodes in the diameter calculation. @@ -35,12 +33,11 @@ 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 route.Vertex) int { - key := NodeID(node) - nodeIndex, ok := nodes[key] + getNodeIndex := func(node NodeID) int { + nodeIndex, ok := nodes[node] if !ok { - nodes[key] = nextIndex + nodes[node] = nextIndex nodeIndex = nextIndex nextIndex++ } @@ -51,12 +48,12 @@ 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.ForEachNodesChannels(ctx, func(_ context.Context, - node Node, channels []*ChannelEdge) error { + node NodeID, channels []*ChannelEdge) error { - u := getNodeIndex(node.PubKey()) + u := getNodeIndex(node) for _, edge := range channels { - v := getNodeIndex(edge.Peer) + v := getNodeIndex(NodeID(edge.Peer)) adj[u] = append(adj[u], v) } diff --git a/docs/release-notes/release-notes-0.21.0.md b/docs/release-notes/release-notes-0.21.0.md index 54d2d632d..13d36c69f 100644 --- a/docs/release-notes/release-notes-0.21.0.md +++ b/docs/release-notes/release-notes-0.21.0.md @@ -265,6 +265,13 @@ is fully populated. This new behaviour can be opted out of via the new `--db.sync-graph-cache-load` option. +* Autopilot's graph-wide channel scoring traversal [no longer requests node + addresses](https://github.com/lightningnetwork/lnd/pull/10796) from the + graph backend, since the scoring code does not consume them. This removes + an unnecessary address batch-load on the SQL backend, and lets the kvdb + backend serve the traversal from the in-memory graph cache when it is + loaded. + * [Invoice pagination queries no longer use `OFFSET`](https://github.com/lightningnetwork/lnd/pull/10700). The five invoice filter queries previously used `LIMIT+OFFSET` for internal batching, diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index d0764446f..ff245d08b 100644 --- a/graph/db/benchmark_test.go +++ b/graph/db/benchmark_test.go @@ -5,7 +5,6 @@ import ( "database/sql" "errors" "fmt" - "net" "path" "sync" "testing" @@ -698,10 +697,9 @@ func BenchmarkGraphReadMethods(b *testing.B) { fn: func(b testing.TB, store Store) { //nolint:ll err := store.ForEachNodeCached( - ctx, lnwire.GossipVersion1, false, + ctx, lnwire.GossipVersion1, 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 9e72cad2a..a63b9d0e6 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -342,21 +342,21 @@ func (c *ChannelGraph) GraphSession(ctx context.Context, // // NOTE: The callback contents MUST not be modified. func (c *ChannelGraph) ForEachNodeCached(ctx context.Context, - v lnwire.GossipVersion, withAddrs bool, - cb func(ctx context.Context, node route.Vertex, addrs []net.Addr, + v lnwire.GossipVersion, + cb func(ctx context.Context, node route.Vertex, chans map[uint64]*DirectedChannel) error, reset func()) error { - if !withAddrs && c.cache != nil && c.cache.isLoaded() { + if c.cache != nil && c.cache.isLoaded() { return c.cache.graphCache.ForEachNode( func(node route.Vertex, channels map[uint64]*DirectedChannel) error { - return cb(ctx, node, nil, channels) + return cb(ctx, node, channels) }, ) } - return c.db.ForEachNodeCached(ctx, v, withAddrs, cb, reset) + return c.db.ForEachNodeCached(ctx, v, cb, reset) } // AddNode adds a vertex/node to the graph database. If the node is not @@ -919,12 +919,11 @@ func (c *VersionedGraph) ForEachNodeDirectedChannel(ctx context.Context, // ForEachNodeCached iterates through all stored vertices/nodes in the graph, // delegating to the embedded ChannelGraph. func (c *VersionedGraph) ForEachNodeCached(ctx context.Context, - withAddrs bool, cb func(ctx context.Context, node route.Vertex, - addrs []net.Addr, + cb func(ctx context.Context, node route.Vertex, chans map[uint64]*DirectedChannel) error, reset func()) error { - return c.ChannelGraph.ForEachNodeCached(ctx, c.v, withAddrs, cb, reset) + return c.ChannelGraph.ForEachNodeCached(ctx, c.v, cb, reset) } // ForEachNode iterates through all stored vertices/nodes in the graph. diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 6521a849e..3d0156ff0 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -449,24 +449,6 @@ func testNodeInsertionAndDeletion(t *testing.T, v lnwire.GossipVersion) { dbNode, err = graph.FetchNode(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 tests that partial/shell nodes are correctly created when @@ -1799,8 +1781,8 @@ func TestGraphTraversal(t *testing.T) { nodeIndex[node.PubKeyBytes] = struct{}{} } - err := graph.ForEachNodeCached(ctx, lnwire.GossipVersion1, false, - func(_ context.Context, node route.Vertex, _ []net.Addr, + err := graph.ForEachNodeCached(ctx, lnwire.GossipVersion1, + func(_ context.Context, node route.Vertex, chans map[uint64]*DirectedChannel) error { if _, ok := nodeIndex[node]; !ok { @@ -6026,9 +6008,8 @@ func TestAsyncGraphCache(t *testing.T) { // assert that we get the expected number of nodes and // channels. err := graph.ForEachNodeCached( - ctx, lnwire.GossipVersion1, false, + ctx, lnwire.GossipVersion1, func(_ context.Context, node route.Vertex, - _ []net.Addr, chans map[uint64]*DirectedChannel) error { numNodes++ diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index 023030029..c126be014 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -81,17 +81,11 @@ type Store interface { //nolint:interfacebloat *models.ChannelEdgePolicy) error, reset func()) error // ForEachNodeCached is similar to forEachNode, but it returns - // 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. + // DirectedChannel data to the call-back. // // NOTE: The callback contents MUST not be modified. ForEachNodeCached(ctx context.Context, v lnwire.GossipVersion, - withAddrs bool, cb func(ctx context.Context, node route.Vertex, - addrs []net.Addr, chans map[uint64]*DirectedChannel) error, reset func()) error diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index ed6d0e07e..47c478c36 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -703,8 +703,8 @@ func (c *KVStore) FetchNodeFeatures(_ context.Context, v lnwire.GossipVersion, // // NOTE: The callback contents MUST not be modified. func (c *KVStore) ForEachNodeCached(ctx context.Context, - v lnwire.GossipVersion, withAddrs bool, - cb func(ctx context.Context, node route.Vertex, addrs []net.Addr, + v lnwire.GossipVersion, + cb func(ctx context.Context, node route.Vertex, chans map[uint64]*DirectedChannel) error, reset func()) error { if v != lnwire.GossipVersion1 { @@ -769,12 +769,7 @@ func (c *KVStore) ForEachNodeCached(ctx context.Context, return err } - var addrs []net.Addr - if withAddrs { - addrs = node.Addresses - } - - return cb(ctx, node.PubKeyBytes, addrs, channels) + return cb(ctx, node.PubKeyBytes, channels) }, reset) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index f4824d75e..e705468a4 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -1719,20 +1719,16 @@ func (s *SQLStore) chanUpdatesInHorizonV2(ctx context.Context, } // ForEachNodeCached is similar to forEachNode, but it returns 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 -// result in an additional round-trip to the database, so it should only be used -// if the addresses are actually needed. +// data to the call-back. // // NOTE: part of the Store interface. func (s *SQLStore) ForEachNodeCached(ctx context.Context, - v lnwire.GossipVersion, withAddrs bool, - cb func(ctx context.Context, node route.Vertex, addrs []net.Addr, + v lnwire.GossipVersion, + cb func(ctx context.Context, node route.Vertex, 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 } @@ -1765,19 +1761,6 @@ 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( @@ -1866,7 +1849,6 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, return &nodeCachedBatchData{ features: nodeFeatures, - addrs: nodeAddrs, chanBatchData: channelBatchData, chanMap: nodeChannelMap, }, nil @@ -1910,15 +1892,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, channels[directedChan.ChannelID] = directedChan } - 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 cb(ctx, nodePub, channels) } return sqldb.ExecuteCollectAndBatchWithSharedDataQuery( diff --git a/itest/lnd_graph_migration_test.go b/itest/lnd_graph_migration_test.go index c8ae22991..5dc74913d 100644 --- a/itest/lnd_graph_migration_test.go +++ b/itest/lnd_graph_migration_test.go @@ -3,7 +3,6 @@ package itest import ( "context" "database/sql" - "net" graphdb "github.com/lightningnetwork/lnd/graph/db" "github.com/lightningnetwork/lnd/lntest" @@ -66,9 +65,9 @@ func testGraphMigration(ht *lntest.HarnessTest) { numNodes int edges = make(map[uint64]bool) ) - err := db.ForEachNodeCached(ctx, lnwire.GossipVersion1, false, + err := db.ForEachNodeCached(ctx, lnwire.GossipVersion1, func(_ context.Context, - _ route.Vertex, _ []net.Addr, + _ route.Vertex, chans map[uint64]*graphdb.DirectedChannel, ) error { diff --git a/rpcserver.go b/rpcserver.go index d4a9ce51b..491bd8a14 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -7439,8 +7439,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, false, func(ctx context.Context, - node route.Vertex, _ []net.Addr, + err := graph.ForEachNodeCached(ctx, func(ctx context.Context, + node route.Vertex, edges map[uint64]*graphdb.DirectedChannel) error { // Increment the total number of nodes with each iteration.