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

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

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

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.

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() {})
 }

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

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.