diff --git a/autopilot/graph.go b/autopilot/graph.go index f1506b5d7..5a1f8dceb 100644 --- a/autopilot/graph.go +++ b/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 { diff --git a/autopilot/interface.go b/autopilot/interface.go index a4efa2eec..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" ) @@ -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 } 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 b5c0b1d26..d7a578f7d 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) @@ -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) diff --git a/autopilot/simple_graph.go b/autopilot/simple_graph.go index 2f80c490a..44f514903 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 { @@ -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) diff --git a/graph/db/benchmark_test.go b/graph/db/benchmark_test.go index f1366459b..08dba16de 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" @@ -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 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..b9d972b3b 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 { @@ -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) diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index dbcd5d1e6..7f9b37008 100644 --- a/graph/db/interfaces.go +++ b/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 diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 7bac0ec13..49191d7c2 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) } @@ -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() {}, - ) -} 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 9d439f448..7a7987855 100644 --- a/graph/db/sql_store.go +++ b/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( 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 diff --git a/rpcserver.go b/rpcserver.go index 9b8bedf69..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) @@ -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.