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.