From 0b1a1e3dc54a2ee60ad4729de75e622e3f8e56fa Mon Sep 17 00:00:00 2001 From: Elle Mouton Date: Thu, 12 Feb 2026 11:49:51 +0200 Subject: [PATCH] graph/db: version ForEachNodeDirectedChannel and cacheable traversal Add a gossip version parameter to ForEachNodeDirectedChannel on the Store interface and both DB implementations (KVStore, SQLStore). The NodeTraverser and routing.Graph interfaces remain unversioned since pathfinding operates on the merged cross-version cache view. The cache population in populateCache is updated in the same commit because it is logically atomic with the versioning changes: the graph cache is the unversioned, merged view used by pathfinding, so it must be populated with data from all gossip versions. Without this change, only v1 data would be loaded into the cache, making v2 nodes and channels invisible to pathfinding. --- graph/db/graph.go | 56 +++++++++++++++++++++++++----------------- graph/db/graph_test.go | 53 +++++++++++++++++++-------------------- graph/db/interfaces.go | 2 +- graph/db/kv_store.go | 14 ++++++++--- graph/db/sql_store.go | 23 ++++++++++------- routing/pathfind.go | 3 +-- 6 files changed, 87 insertions(+), 64 deletions(-) diff --git a/graph/db/graph.go b/graph/db/graph.go index ea41f8eca..91286b75f 100644 --- a/graph/db/graph.go +++ b/graph/db/graph.go @@ -163,32 +163,41 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error { log.Info("Populating in-memory channel graph, this might take a " + "while...") - // TODO(elle): the cache should be populated with data from across - // protocol versions. - err := c.db.ForEachNodeCacheable( - ctx, lnwire.GossipVersion1, func(node route.Vertex, - features *lnwire.FeatureVector) error { + for _, v := range []lnwire.GossipVersion{ + gossipV1, gossipV2, + } { + // TODO(elle): If we have both v1 and v2 entries for the same + // node/channel, prefer v2 when merging. + err := c.db.ForEachNodeCacheable(ctx, v, + func(node route.Vertex, + features *lnwire.FeatureVector) error { - c.graphCache.AddNodeFeatures(node, features) + c.graphCache.AddNodeFeatures(node, features) - return nil - }, func() {}, - ) - if err != nil { - return err - } + return nil + }, func() {}, + ) + if err != nil && + !errors.Is(err, ErrVersionNotSupportedForKVDB) { - err = c.db.ForEachChannelCacheable( - lnwire.GossipVersion1, func(info *models.CachedEdgeInfo, - policy1, policy2 *models.CachedEdgePolicy) error { + return err + } - c.graphCache.AddChannel(info, policy1, policy2) + err = c.db.ForEachChannelCacheable( + v, func(info *models.CachedEdgeInfo, + policy1, + policy2 *models.CachedEdgePolicy) error { - return nil - }, func() {}, - ) - if err != nil { - return err + c.graphCache.AddChannel(info, policy1, policy2) + + return nil + }, func() {}, + ) + if err != nil && + !errors.Is(err, ErrVersionNotSupportedForKVDB) { + + return err + } } log.Infof("Finished populating in-memory channel graph (took %v, %s)", @@ -214,7 +223,10 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(node route.Vertex, return c.graphCache.ForEachChannel(node, cb) } - return c.db.ForEachNodeDirectedChannel(node, cb, reset) + // TODO(elle): once the no-cache path needs to support + // pathfinding across gossip versions, this should iterate + // across all versions rather than defaulting to v1. + return c.db.ForEachNodeDirectedChannel(gossipV1, node, cb, reset) } // FetchNodeFeatures returns the features of the given node. If no features are diff --git a/graph/db/graph_test.go b/graph/db/graph_test.go index 792aaf85a..b5f25be2c 100644 --- a/graph/db/graph_test.go +++ b/graph/db/graph_test.go @@ -162,6 +162,10 @@ var versionedTests = []versionedTest{ name: "for each source node channel", test: testForEachSourceNodeChannel, }, + { + name: "graph traversal cacheable", + test: testGraphTraversalCacheable, + }, { name: "partial node", test: testPartialNode, @@ -1815,42 +1819,36 @@ func TestGraphTraversal(t *testing.T) { require.Equal(t, numChannels, numNodeChans) } -// TestGraphTraversalCacheable tests that the memory optimized node traversal is +// testGraphTraversalCacheable tests that the memory optimized node traversal is // working correctly. -func TestGraphTraversalCacheable(t *testing.T) { +func testGraphTraversalCacheable(t *testing.T, v lnwire.GossipVersion) { t.Parallel() ctx := t.Context() - graph := MakeTestGraph(t) + graph := NewVersionedGraph(MakeTestGraph(t), v) // We'd like to test some of the graph traversal capabilities within // the DB, so we'll create a series of fake nodes to insert into the // graph. And we'll create 5 channels between the first two nodes. const numNodes = 20 const numChannels = 5 - chanIndex, _ := fillTestGraph( - t, graph, numNodes, numChannels, lnwire.GossipVersion1, + chanIndex, nodeList := fillTestGraph( + t, graph.ChannelGraph, numNodes, numChannels, v, ) - // Create a map of all nodes with the iteration we know works (because - // it is tested in another test). + // Create a map of all nodes with the nodes we just inserted. nodeMap := make(map[route.Vertex]struct{}) - err := graph.ForEachNode(ctx, func(n *models.Node) error { - nodeMap[n.PubKeyBytes] = struct{}{} - - return nil - }, func() {}) - require.NoError(t, err) + for _, node := range nodeList { + nodeMap[node.PubKeyBytes] = struct{}{} + } require.Len(t, nodeMap, numNodes) // 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 []route.Vertex - err = graph.ForEachNodeCacheable( - ctx, lnwire.GossipVersion1, func(node route.Vertex, - features *lnwire.FeatureVector) error { - + err := graph.ForEachNodeCacheable(ctx, + func(node route.Vertex, features *lnwire.FeatureVector) error { delete(nodeMap, node) nodes = append(nodes, node) @@ -1872,7 +1870,7 @@ func TestGraphTraversalCacheable(t *testing.T) { for _, node := range nodes { // Query the ChannelGraph which uses the cache to iterate // through the channels for each node. - err = graph.ForEachNodeDirectedChannel( + err = graph.ChannelGraph.ForEachNodeDirectedChannel( node, func(d *DirectedChannel) error { delete(chanIndex, d.ChannelID) return nil @@ -1882,7 +1880,7 @@ func TestGraphTraversalCacheable(t *testing.T) { // Now skip the cache and query the DB directly. err = graph.db.ForEachNodeDirectedChannel( - node, func(d *DirectedChannel) error { + v, node, func(d *DirectedChannel) error { delete(chanIndex2, d.ChannelID) return nil }, func() {}, @@ -1957,12 +1955,12 @@ func fillTestGraph(t testing.TB, graph *ChannelGraph, numNodes, ctx := t.Context() nodes := make([]*models.Node, numNodes) - nodeIndex := map[string]struct{}{} + nodeIndex := map[route.Vertex]struct{}{} for i := 0; i < numNodes; i++ { node := createTestVertex(t, v) nodes[i] = node - nodeIndex[node.Alias.UnwrapOr("")] = struct{}{} + nodeIndex[node.PubKeyBytes] = struct{}{} } // Add each of the nodes into the graph, they should be inserted @@ -1973,10 +1971,12 @@ 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(n *models.Node) error { - delete(nodeIndex, n.Alias.UnwrapOr("")) - return nil - }, func() {}) + err := graph.ForEachNodeCacheable(ctx, v, + func(node route.Vertex, _ *lnwire.FeatureVector) error { + delete(nodeIndex, node) + + return nil + }, func() {}) require.NoError(t, err) require.Len(t, nodeIndex, 0) @@ -5135,7 +5135,8 @@ func TestGraphCacheForEachNodeChannel(t *testing.T) { getSingleChannel := func() *DirectedChannel { var ch *DirectedChannel - err := graph.ForEachNodeDirectedChannel(node1.PubKeyBytes, + err := graph.ForEachNodeDirectedChannel( + node1.PubKeyBytes, func(c *DirectedChannel) error { require.Nil(t, ch) ch = c diff --git a/graph/db/interfaces.go b/graph/db/interfaces.go index e7ddc6588..5a6d27d55 100644 --- a/graph/db/interfaces.go +++ b/graph/db/interfaces.go @@ -33,7 +33,7 @@ type NodeTraverser interface { type Store interface { //nolint:interfacebloat // ForEachNodeDirectedChannel calls the callback for every channel of // the given node. - ForEachNodeDirectedChannel(nodePub route.Vertex, + ForEachNodeDirectedChannel(v lnwire.GossipVersion, nodePub route.Vertex, cb func(channel *DirectedChannel) error, reset func()) error // FetchNodeFeatures returns the features of the given node. diff --git a/graph/db/kv_store.go b/graph/db/kv_store.go index 6cff4e5e9..3a143ee7e 100644 --- a/graph/db/kv_store.go +++ b/graph/db/kv_store.go @@ -669,8 +669,13 @@ func (c *KVStore) fetchNodeFeatures(tx kvdb.RTx, // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (c *KVStore) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, reset func()) error { +func (c *KVStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error { + + if v != lnwire.GossipVersion1 { + return ErrVersionNotSupportedForKVDB + } return c.forEachNodeDirectedChannel(nil, nodePub, cb, reset) } @@ -4471,8 +4476,9 @@ type nodeTraverserSession struct { // node. // // NOTE: Part of the NodeTraverser interface. -func (c *nodeTraverserSession) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, _ func()) error { +func (c *nodeTraverserSession) ForEachNodeDirectedChannel( + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + _ func()) error { return c.db.forEachNodeDirectedChannel(c.tx, nodePub, cb, func() {}) } diff --git a/graph/db/sql_store.go b/graph/db/sql_store.go index 03083b909..75f06f667 100644 --- a/graph/db/sql_store.go +++ b/graph/db/sql_store.go @@ -988,13 +988,14 @@ func (s *SQLStore) ForEachNode(ctx context.Context, // Unknown policies are passed into the callback as nil values. // // NOTE: this is part of the graphdb.NodeTraverser interface. -func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, reset func()) error { +func (s *SQLStore) ForEachNodeDirectedChannel(v lnwire.GossipVersion, + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + reset func()) error { var ctx = context.TODO() return s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { - return forEachNodeDirectedChannel(ctx, db, nodePub, cb) + return forEachNodeDirectedChannel(ctx, db, v, nodePub, cb) }, reset) } @@ -3347,12 +3348,15 @@ func newSQLNodeTraverser(db SQLQueries, // node. // // NOTE: Part of the NodeTraverser interface. -func (s *sqlNodeTraverser) ForEachNodeDirectedChannel(nodePub route.Vertex, - cb func(channel *DirectedChannel) error, _ func()) error { +func (s *sqlNodeTraverser) ForEachNodeDirectedChannel( + nodePub route.Vertex, cb func(channel *DirectedChannel) error, + _ func()) error { ctx := context.TODO() - return forEachNodeDirectedChannel(ctx, s.db, nodePub, cb) + return forEachNodeDirectedChannel( + ctx, s.db, lnwire.GossipVersion1, nodePub, cb, + ) } // FetchNodeFeatures returns the features of the given node. If the node is @@ -3372,7 +3376,8 @@ func (s *sqlNodeTraverser) FetchNodeFeatures(nodePub route.Vertex) ( // channel and its incoming policy. If the node is not found, no error is // returned. func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, - nodePub route.Vertex, cb func(channel *DirectedChannel) error) error { + v lnwire.GossipVersion, nodePub route.Vertex, + cb func(channel *DirectedChannel) error) error { toNodeCallback := func() route.Vertex { return nodePub @@ -3380,7 +3385,7 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, dbID, err := db.GetNodeIDByPubKey( ctx, sqlc.GetNodeIDByPubKeyParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), PubKey: nodePub[:], }, ) @@ -3392,7 +3397,7 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries, rows, err := db.ListChannelsByNodeID( ctx, sqlc.ListChannelsByNodeIDParams{ - Version: int16(lnwire.GossipVersion1), + Version: int16(v), NodeID1: dbID, }, ) diff --git a/routing/pathfind.go b/routing/pathfind.go index fab8015dd..ddaade60c 100644 --- a/routing/pathfind.go +++ b/routing/pathfind.go @@ -1461,8 +1461,7 @@ func processNodeForBlindedPath(g Graph, node route.Vertex, // Now, iterate over the node's channels in search for paths to this // node that can be used for blinded paths err = g.ForEachNodeDirectedChannel( - node, - func(channel *graphdb.DirectedChannel) error { + node, func(channel *graphdb.DirectedChannel) error { // Keep track of how many incoming channels this node // has. We only use a node as an introduction node if it // has channels other than the one that lead us to it.