From 90179b651e300afee0c76a771bb92ab32b3bb0c9 Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 10 Feb 2025 16:29:00 +0200 Subject: [PATCH 1/2] graph/db: remove unnecessary AddNode method on GraphCache The AddNode method on the GraphCache calls `AddNodeFeatures` underneath and then iterates through all the node's persisted channels and adds them to the cache too via `AddChannel`. This is, however, not required since at the time the cache is populated in `NewChannelGraph`, the cache is populated will all persisted nodes and all persisted channels. Then, once any new channels come in, via `AddChannelEdge`, they are added to the cache via AddChannel. If any new nodes come in via `AddLightningNode`, then currently the cache's AddNode method is called which the both adds the node and again iterates through all persisted channels and re-adds them to the cache. This is definitely redundent since the initial cache population and updates via AddChannelEdge should keep the cache fresh in terms of channels. So we remove this for 2 reasons: 1) to remove the redundent DB calls and 2) this requires a kvdb.RTx to be passed in to the GraphCache calls which will make it hard to extract the cache out of the CRUD layer and be used more generally. The AddNode method made sense when the cache was first added in the code-base [here](https://github.com/lightningnetwork/lnd/commit/369c09be6152b76a24915050a8a5fe6bccf2b8f0#diff-ae36bdb6670644d20c4e43f3a0ed47f71886c2bcdf3cc2937de24315da5dc072R213) since then during graph cache population, nodes and channels would be added to the cache in a single DB transaction. This was, however, changed [later on](https://github.com/lightningnetwork/lnd/commit/352008a0c22ffd9e97b9481ffa44dbc50fca2ddc) to be done in 2 separate DB calls for efficiency reasons. --- graph/db/graph.go | 5 +---- graph/db/graph_cache.go | 17 ----------------- graph/db/graph_cache_test.go | 18 ++++++++---------- 3 files changed, 9 insertions(+), 31 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index 06bd0257d..326c66ab0 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -904,10 +904,7 @@ func (c *ChannelGraph) AddLightningNode(node *models.LightningNode, cNode := newGraphCacheNode( node.PubKeyBytes, node.Features, ) - err := c.graphCache.AddNode(tx, cNode) - if err != nil { - return err - } + c.graphCache.AddNodeFeatures(cNode) } return addLightningNode(tx, node) diff --git a/graph/db/graph_cache.go b/graph/db/graph_cache.go index a034e23af..67f34e86e 100644 --- a/graph/db/graph_cache.go +++ b/graph/db/graph_cache.go @@ -136,23 +136,6 @@ func (c *GraphCache) AddNodeFeatures(node GraphCacheNode) { c.mtx.Unlock() } -// AddNode adds a graph node, including all the (directed) channels of that -// node. -func (c *GraphCache) AddNode(tx kvdb.RTx, node GraphCacheNode) error { - c.AddNodeFeatures(node) - - return node.ForEachChannel( - tx, func(tx kvdb.RTx, info *models.ChannelEdgeInfo, - outPolicy *models.ChannelEdgePolicy, - inPolicy *models.ChannelEdgePolicy) error { - - c.AddChannel(info, outPolicy, inPolicy) - - return nil - }, - ) -} - // AddChannel adds a non-directed channel, meaning that the order of policy 1 // and policy 2 does not matter, the directionality is extracted from the info // and policy flags automatically. The policy will be set as the outgoing policy diff --git a/graph/db/graph_cache_test.go b/graph/db/graph_cache_test.go index 69abb2597..35048e7cd 100644 --- a/graph/db/graph_cache_test.go +++ b/graph/db/graph_cache_test.go @@ -88,18 +88,16 @@ func TestGraphCacheAddNode(t *testing.T) { node := &node{ pubKey: nodeA, features: lnwire.EmptyFeatureVector(), - edgeInfos: []*models.ChannelEdgeInfo{{ - ChannelID: 1000, - // Those are direction independent! - NodeKey1Bytes: pubKey1, - NodeKey2Bytes: pubKey2, - Capacity: 500, - }}, - outPolicies: []*models.ChannelEdgePolicy{outPolicy1}, - inPolicies: []*models.ChannelEdgePolicy{inPolicy1}, } cache := NewGraphCache(10) - require.NoError(t, cache.AddNode(nil, node)) + cache.AddNodeFeatures(node) + cache.AddChannel(&models.ChannelEdgeInfo{ + ChannelID: 1000, + // Those are direction independent! + NodeKey1Bytes: pubKey1, + NodeKey2Bytes: pubKey2, + Capacity: 500, + }, outPolicy1, inPolicy1) var fromChannels, toChannels []*DirectedChannel _ = cache.ForEachChannel(nodeA, func(c *DirectedChannel) error { From 5c2c00e414643b9344b0a16d1793441167e6264b Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Mon, 10 Feb 2025 17:08:08 +0200 Subject: [PATCH 2/2] graph/db: remove GraphCacheNode interface With the previous commit, the AddNode method was removed and since that was the only method making use of the ForEachChannel on the GraphCacheNode interface, we can remove that method. Since the only two methods left just expose the node's pub key and features, it really is not required anymore and so the entire thing can be removed along with the implementation of it. --- docs/release-notes/release-notes-0.19.0.md | 5 + graph/db/graph.go | 106 ++++++--------------- graph/db/graph_cache.go | 36 +------ graph/db/graph_cache_test.go | 40 +------- graph/db/graph_test.go | 81 +++++++--------- 5 files changed, 77 insertions(+), 191 deletions(-) diff --git a/docs/release-notes/release-notes-0.19.0.md b/docs/release-notes/release-notes-0.19.0.md index e70492d6f..2f9262958 100644 --- a/docs/release-notes/release-notes-0.19.0.md +++ b/docs/release-notes/release-notes-0.19.0.md @@ -257,6 +257,11 @@ The underlying functionality between those two options remain the same. * Log rotation can now use ZSTD +* [Remove redundant + iteration](https://github.com/lightningnetwork/lnd/pull/9496) over a node's + persisted channels when updating the graph cache with a new node or node + update. + ## Deprecations ### ⚠️ **Warning:** The following RPCs will be removed in release version **0.21**: diff --git a/graph/db/graph.go b/graph/db/graph.go index 326c66ab0..6759e18b8 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -231,13 +231,13 @@ func NewChannelGraph(db kvdb.Backend, options ...OptionModifier) (*ChannelGraph, log.Debugf("Populating in-memory channel graph, this might " + "take a while...") - err := g.ForEachNodeCacheable( - func(tx kvdb.RTx, node GraphCacheNode) error { - g.graphCache.AddNodeFeatures(node) + err := g.ForEachNodeCacheable(func(node route.Vertex, + features *lnwire.FeatureVector) error { - return nil - }, - ) + g.graphCache.AddNodeFeatures(node, features) + + return nil + }) if err != nil { return nil, err } @@ -772,8 +772,8 @@ func (c *ChannelGraph) forEachNode( // 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. -func (c *ChannelGraph) ForEachNodeCacheable(cb func(kvdb.RTx, - GraphCacheNode) error) error { +func (c *ChannelGraph) ForEachNodeCacheable(cb func(route.Vertex, + *lnwire.FeatureVector) error) error { traversal := func(tx kvdb.RTx) error { // First grab the nodes bucket which stores the mapping from @@ -792,7 +792,7 @@ func (c *ChannelGraph) ForEachNodeCacheable(cb func(kvdb.RTx, } nodeReader := bytes.NewReader(nodeBytes) - cacheableNode, err := deserializeLightningNodeCacheable( + node, features, err := deserializeLightningNodeCacheable( //nolint:ll nodeReader, ) if err != nil { @@ -801,7 +801,7 @@ func (c *ChannelGraph) ForEachNodeCacheable(cb func(kvdb.RTx, // Execute the callback, the transaction will abort if // this returns an error. - return cb(tx, cacheableNode) + return cb(node, features) }) } @@ -901,10 +901,9 @@ func (c *ChannelGraph) AddLightningNode(node *models.LightningNode, r := &batch.Request{ Update: func(tx kvdb.RwTx) error { if c.graphCache != nil { - cNode := newGraphCacheNode( + c.graphCache.AddNodeFeatures( node.PubKeyBytes, node.Features, ) - c.graphCache.AddNodeFeatures(cNode) } return addLightningNode(tx, node) @@ -3056,50 +3055,6 @@ func (c *ChannelGraph) fetchLightningNode(tx kvdb.RTx, return node, nil } -// graphCacheNode is a struct that wraps a LightningNode in a way that it can be -// cached in the graph cache. -type graphCacheNode struct { - pubKeyBytes route.Vertex - features *lnwire.FeatureVector -} - -// newGraphCacheNode returns a new cache optimized node. -func newGraphCacheNode(pubKey route.Vertex, - features *lnwire.FeatureVector) *graphCacheNode { - - return &graphCacheNode{ - pubKeyBytes: pubKey, - features: features, - } -} - -// PubKey returns the node's public identity key. -func (n *graphCacheNode) PubKey() route.Vertex { - return n.pubKeyBytes -} - -// Features returns the node's features. -func (n *graphCacheNode) Features() *lnwire.FeatureVector { - return n.features -} - -// ForEachChannel iterates through all channels of this node, executing the -// passed callback with an edge info structure and the policies of each end -// of the channel. The first edge policy is the outgoing edge *to* the -// connecting node, while the second is the incoming edge *from* the -// connecting node. If the callback returns an error, then the iteration is -// halted with the error propagated back up to the caller. -// -// Unknown policies are passed into the callback as nil values. -func (n *graphCacheNode) ForEachChannel(tx kvdb.RTx, - cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { - - return nodeTraversal(tx, n.pubKeyBytes[:], nil, cb) -} - -var _ GraphCacheNode = (*graphCacheNode)(nil) - // HasLightningNode determines if the graph has a vertex identified by the // target node identity public key. If the node exists in the database, a // timestamp of when the data for the node was lasted updated is returned along @@ -4059,60 +4014,59 @@ func fetchLightningNode(nodeBucket kvdb.RBucket, return deserializeLightningNode(nodeReader) } -func deserializeLightningNodeCacheable(r io.Reader) (*graphCacheNode, error) { - // Always populate a feature vector, even if we don't have a node - // announcement and short circuit below. - node := newGraphCacheNode( - route.Vertex{}, - lnwire.EmptyFeatureVector(), - ) +func deserializeLightningNodeCacheable(r io.Reader) (route.Vertex, + *lnwire.FeatureVector, error) { - var nodeScratch [8]byte + var ( + pubKey route.Vertex + features = lnwire.EmptyFeatureVector() + nodeScratch [8]byte + ) // Skip ahead: // - LastUpdate (8 bytes) if _, err := r.Read(nodeScratch[:]); err != nil { - return nil, err + return pubKey, nil, err } - if _, err := io.ReadFull(r, node.pubKeyBytes[:]); err != nil { - return nil, err + if _, err := io.ReadFull(r, pubKey[:]); err != nil { + return pubKey, nil, err } // Read the node announcement flag. if _, err := r.Read(nodeScratch[:2]); err != nil { - return nil, err + return pubKey, nil, err } hasNodeAnn := byteOrder.Uint16(nodeScratch[:2]) // The rest of the data is optional, and will only be there if we got a // node announcement for this node. if hasNodeAnn == 0 { - return node, nil + return pubKey, features, nil } // We did get a node announcement for this node, so we'll have the rest // of the data available. var rgb uint8 if err := binary.Read(r, byteOrder, &rgb); err != nil { - return nil, err + return pubKey, nil, err } if err := binary.Read(r, byteOrder, &rgb); err != nil { - return nil, err + return pubKey, nil, err } if err := binary.Read(r, byteOrder, &rgb); err != nil { - return nil, err + return pubKey, nil, err } if _, err := wire.ReadVarString(r, 0); err != nil { - return nil, err + return pubKey, nil, err } - if err := node.features.Decode(r); err != nil { - return nil, err + if err := features.Decode(r); err != nil { + return pubKey, nil, err } - return node, nil + return pubKey, features, nil } func deserializeLightningNode(r io.Reader) (models.LightningNode, error) { diff --git a/graph/db/graph_cache.go b/graph/db/graph_cache.go index 67f34e86e..cefa88a85 100644 --- a/graph/db/graph_cache.go +++ b/graph/db/graph_cache.go @@ -6,33 +6,10 @@ import ( "github.com/btcsuite/btcd/btcutil" "github.com/lightningnetwork/lnd/graph/db/models" - "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" ) -// GraphCacheNode is an interface for all the information the cache needs to know -// about a lightning node. -type GraphCacheNode interface { - // PubKey is the node's public identity key. - PubKey() route.Vertex - - // Features returns the node's p2p features. - Features() *lnwire.FeatureVector - - // ForEachChannel iterates through all channels of a given node, - // executing the passed callback with an edge info structure and the - // policies of each end of the channel. The first edge policy is the - // outgoing edge *to* the connecting node, while the second is the - // incoming edge *from* the connecting node. If the callback returns an - // error, then the iteration is halted with the error propagated back up - // to the caller. - ForEachChannel(kvdb.RTx, - func(kvdb.RTx, *models.ChannelEdgeInfo, - *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error -} - // DirectedChannel is a type that stores the channel information as seen from // one side of the channel. type DirectedChannel struct { @@ -124,16 +101,13 @@ func (c *GraphCache) Stats() string { } // AddNodeFeatures adds a graph node and its features to the cache. -func (c *GraphCache) AddNodeFeatures(node GraphCacheNode) { - nodePubKey := node.PubKey() +func (c *GraphCache) AddNodeFeatures(node route.Vertex, + features *lnwire.FeatureVector) { - // Only hold the lock for a short time. The `ForEachChannel()` below is - // possibly slow as it has to go to the backend, so we can unlock - // between the calls. And the AddChannel() method will acquire its own - // lock anyway. c.mtx.Lock() - c.nodeFeatures[nodePubKey] = node.Features() - c.mtx.Unlock() + defer c.mtx.Unlock() + + c.nodeFeatures[node] = features } // AddChannel adds a non-directed channel, meaning that the order of policy 1 diff --git a/graph/db/graph_cache_test.go b/graph/db/graph_cache_test.go index 35048e7cd..087fb81ac 100644 --- a/graph/db/graph_cache_test.go +++ b/graph/db/graph_cache_test.go @@ -5,7 +5,6 @@ import ( "testing" "github.com/lightningnetwork/lnd/graph/db/models" - "github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/routing/route" "github.com/stretchr/testify/require" @@ -25,39 +24,6 @@ var ( pubKey2, _ = route.NewVertexFromBytes(pubKey2Bytes) ) -type node struct { - pubKey route.Vertex - features *lnwire.FeatureVector - - edgeInfos []*models.ChannelEdgeInfo - outPolicies []*models.ChannelEdgePolicy - inPolicies []*models.ChannelEdgePolicy -} - -func (n *node) PubKey() route.Vertex { - return n.pubKey -} -func (n *node) Features() *lnwire.FeatureVector { - return n.features -} - -func (n *node) ForEachChannel(tx kvdb.RTx, - cb func(kvdb.RTx, *models.ChannelEdgeInfo, *models.ChannelEdgePolicy, - *models.ChannelEdgePolicy) error) error { - - for idx := range n.edgeInfos { - err := cb( - tx, n.edgeInfos[idx], n.outPolicies[idx], - n.inPolicies[idx], - ) - if err != nil { - return err - } - } - - return nil -} - // TestGraphCacheAddNode tests that a channel going from node A to node B can be // cached correctly, independent of the direction we add the channel as. func TestGraphCacheAddNode(t *testing.T) { @@ -85,12 +51,8 @@ func TestGraphCacheAddNode(t *testing.T) { ChannelFlags: lnwire.ChanUpdateChanFlags(channelFlagB), ToNode: nodeA, } - node := &node{ - pubKey: nodeA, - features: lnwire.EmptyFeatureVector(), - } cache := NewGraphCache(10) - cache.AddNodeFeatures(node) + cache.AddNodeFeatures(nodeA, lnwire.EmptyFeatureVector()) cache.AddChannel(&models.ChannelEdgeInfo{ ChannelID: 1000, // Those are direction independent! diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 3b3454b6b..eb658e2f1 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -1105,24 +1105,22 @@ func TestGraphTraversalCacheable(t *testing.T) { // Iterate through all the known channels within the graph DB by // iterating over each node, once again if the map is empty that // indicates that all edges have properly been reached. - var nodes []GraphCacheNode - err = graph.ForEachNodeCacheable( - func(tx kvdb.RTx, node GraphCacheNode) error { - delete(nodeMap, node.PubKey()) + var nodes []route.Vertex + err = graph.ForEachNodeCacheable(func(node route.Vertex, + features *lnwire.FeatureVector) error { - nodes = append(nodes, node) + delete(nodeMap, node) + nodes = append(nodes, node) - return nil - }, - ) + return nil + }) require.NoError(t, err) require.Len(t, nodeMap, 0) err = graph.db.View(func(tx kvdb.RTx) error { for _, node := range nodes { - err := node.ForEachChannel( - tx, func(tx kvdb.RTx, - info *models.ChannelEdgeInfo, + err := graph.ForEachNodeChannelTx(tx, node, + func(tx kvdb.RTx, info *models.ChannelEdgeInfo, policy *models.ChannelEdgePolicy, policy2 *models.ChannelEdgePolicy) error { //nolint:ll @@ -3883,44 +3881,37 @@ func BenchmarkForEachChannel(b *testing.B) { maxHTLCs lnwire.MilliSatoshi ) - var nodes []GraphCacheNode - err = graph.ForEachNodeCacheable( - func(tx kvdb.RTx, node GraphCacheNode) error { - nodes = append(nodes, node) + var nodes []route.Vertex + err = graph.ForEachNodeCacheable(func(node route.Vertex, + vector *lnwire.FeatureVector) error { - return nil - }, - ) - require.NoError(b, err) - - err = graph.db.View(func(tx kvdb.RTx) error { - for _, n := range nodes { - cb := func(tx kvdb.RTx, - info *models.ChannelEdgeInfo, - policy *models.ChannelEdgePolicy, - policy2 *models.ChannelEdgePolicy) error { //nolint:ll - - // We need to do something with - // the data here, otherwise the - // compiler is going to optimize - // this away, and we get bogus - // results. - totalCapacity += info.Capacity - maxHTLCs += policy.MaxHTLC - maxHTLCs += policy2.MaxHTLC - - return nil - } - - err := n.ForEachChannel(tx, cb) - if err != nil { - return err - } - } + nodes = append(nodes, node) return nil - }, func() {}) + }) require.NoError(b, err) + + for _, n := range nodes { + cb := func(tx kvdb.RTx, + info *models.ChannelEdgeInfo, + policy *models.ChannelEdgePolicy, + policy2 *models.ChannelEdgePolicy) error { //nolint:ll + + // We need to do something with + // the data here, otherwise the + // compiler is going to optimize + // this away, and we get bogus + // results. + totalCapacity += info.Capacity + maxHTLCs += policy.MaxHTLC + maxHTLCs += policy2.MaxHTLC + + return nil + } + + err := graph.ForEachNodeChannel(n, cb) + require.NoError(b, err) + } } }