Merge pull request #10127 from ellemouton/graphPerf8

[7] graph/db+autopilot: improve efficiency of autopilot methods that use the ForEachNode/ForEachChannel pattern

autopilot/graph.go

@@ -51,7 +51,8 @@ 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
 }
 
 // A compile time assertion to ensure dbNode meets the autopilot.Node
@@ -64,7 +65,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,47 +73,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
-}
-
-// 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
-		}
-
-		node, err := d.tx.FetchNode(ep.ToNode)
-		if err != nil {
-			return err
-		}
-
-		edge := ChannelEdge{
-			ChanID:   lnwire.NewShortChanIDFromInt(ep.ChannelID),
-			Capacity: ei.Capacity,
-			Peer: &dbNode{
-				tx: node,
-			},
-		}
-
-		return cb(ctx, edge)
-	})
+	return d.addrs
 }
 
 // ForEachNode is a higher-order function that should be called once for each
@@ -123,22 +84,64 @@ func (d *dbNode) ForEachChannel(ctx context.Context,
 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{
-			tx: nodeTx,
+			pub:   n.PubKeyBytes,
+			addrs: n.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 {
@@ -185,32 +188,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: dbNodeCached{
-				node: 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.
@@ -219,7 +196,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 {
@@ -230,6 +208,45 @@ func (dc *databaseChannelGraphCached) ForEachNode(ctx context.Context,
 
 			return cb(ctx, node)
 		}
+
+		return nil
+	}, 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)
 }
@@ -267,24 +284,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 {

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"
 )
@@ -31,13 +32,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
@@ -72,7 +66,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.
@@ -87,6 +81,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
@@ -227,15 +229,17 @@ 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
 	// 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
 }

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
 	}
 

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)
 
@@ -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)
@@ -535,18 +533,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
 }
@@ -612,6 +604,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)
@@ -692,14 +707,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)
 
@@ -726,38 +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() {},
-	)
-}
-
-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)

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 {
@@ -46,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 {
-		u := getNodeIndex(node)
+	err := g.ForEachNodesChannels(ctx, func(_ context.Context,
+		node Node, channels []*ChannelEdge) error {
 
-		return node.ForEachChannel(
-			ctx, func(_ context.Context,
-				edge ChannelEdge) error {
+		u := getNodeIndex(node.PubKey())
 
-				v := getNodeIndex(edge.Peer)
+		for _, edge := range channels {
+			v := getNodeIndex(edge.Peer)
+			adj[u] = append(adj[u], v)
+		}
 
-				adj[u] = append(adj[u], v)
-
-				return nil
-			},
-		)
+		return nil
 	}, func() {
 		clear(adj)
 		clear(nodes)

graph/db/benchmark_test.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"net"
 	"os"
 	"path"
 	"sync"
@@ -337,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
@@ -486,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()
@@ -657,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.
@@ -721,7 +720,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,
-	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

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 {
@@ -1450,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() {})
@@ -1582,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)
@@ -1691,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)

graph/db/interfaces.go

@@ -14,25 +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
-
-	// 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
-
-	// 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
 // 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.
@@ -89,20 +70,24 @@ 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
 	// 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

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)
 }
 
@@ -795,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)
 }
 
@@ -4872,67 +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
-}
-
-// 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.
-//
-// 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() {},
-	)
-}

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.

graph/db/sql_store.go

@@ -799,89 +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
-}
-
-// 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)
-}
-
-// 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
@@ -1075,17 +1010,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
 	}
@@ -1118,6 +1055,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(
@@ -1206,6 +1156,7 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context,
 
 			return &nodeCachedBatchData{
 				features:      nodeFeatures,
+				addrs:         nodeAddrs,
 				chanBatchData: channelBatchData,
 				chanMap:       nodeChannelMap,
 			}, nil
@@ -1249,7 +1200,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(

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

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