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.