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

@@ -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)
 }

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,
+	ForEachNodeCached(ctx context.Context, withAddrs bool,
-		chans map[uint64]*graphdb.DirectedChannel) error,
+		cb func(ctx context.Context, node route.Vertex,
+			addrs []net.Addr,
+			chans map[uint64]*graphdb.DirectedChannel) error,
 		reset func()) error
 }

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

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,
+func (c *ChannelGraph) ForEachNodeCached(ctx context.Context, withAddrs bool,
-	cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error,
+	cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
-	reset func()) error {
+		chans map[uint64]*DirectedChannel) error, reset func()) error {
 
-	if c.graphCache != nil {
+	if !withAddrs && c.graphCache != nil {
-		return c.graphCache.ForEachNode(cb)
+		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

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 {

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,
+	ForEachNodeCached(ctx context.Context, withAddrs bool,
-		chans map[uint64]*DirectedChannel) error, reset func()) error
+		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

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,
+func (c *KVStore) ForEachNodeCached(ctx context.Context, withAddrs bool,
-	cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error,
+	cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
-	reset func()) error {
+		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)
 }
 

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.
+// 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
-// NOTE: The callback contents MUST not be modified.
+// 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,
+func (s *SQLStore) ForEachNodeCached(ctx context.Context, withAddrs bool,
-	cb func(node route.Vertex, chans map[uint64]*DirectedChannel) error,
+	cb func(ctx context.Context, node route.Vertex, addrs []net.Addr,
-	reset func()) error {
+		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(

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.