graph/db: version cacheable iteration methods

Add gossip version parameters to ForEachNodeCacheable and
ForEachChannelCacheable in the Store interface and both implementations.

Thread the version through SQL helpers and update call sites/tests
accordingly.

graph/db/benchmark_test.go

@@ -669,7 +669,8 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 			name: "ForEachNodeCacheable",
 			fn: func(b testing.TB, store Store) {
 				err := store.ForEachNodeCacheable(
-					ctx, func(_ route.Vertex,
+					ctx, lnwire.GossipVersion1,
+					func(_ route.Vertex,
 						_ *lnwire.FeatureVector) error {
 
 						// Increment the counter to

graph/db/graph.go

@@ -163,19 +163,23 @@ func (c *ChannelGraph) populateCache(ctx context.Context) error {
 	log.Info("Populating in-memory channel graph, this might take a " +
 		"while...")
 
-	err := c.db.ForEachNodeCacheable(ctx, func(node route.Vertex,
-		features *lnwire.FeatureVector) error {
+	// 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 {
 
-		c.graphCache.AddNodeFeatures(node, features)
+			c.graphCache.AddNodeFeatures(node, features)
 
-		return nil
-	}, func() {})
+			return nil
+		}, func() {},
+	)
 	if err != nil {
 		return err
 	}
 
 	err = c.db.ForEachChannelCacheable(
-		func(info *models.CachedEdgeInfo,
+		lnwire.GossipVersion1, func(info *models.CachedEdgeInfo,
 			policy1, policy2 *models.CachedEdgePolicy) error {
 
 			c.graphCache.AddChannel(info, policy1, policy2)
@@ -606,10 +610,10 @@ func (c *ChannelGraph) ForEachNode(ctx context.Context,
 
 // ForEachNodeCacheable iterates through all stored vertices/nodes in the graph.
 func (c *ChannelGraph) ForEachNodeCacheable(ctx context.Context,
-	cb func(route.Vertex, *lnwire.FeatureVector) error,
-	reset func()) error {
+	v lnwire.GossipVersion, cb func(route.Vertex,
+		*lnwire.FeatureVector) error, reset func()) error {
 
-	return c.db.ForEachNodeCacheable(ctx, cb, reset)
+	return c.db.ForEachNodeCacheable(ctx, v, cb, reset)
 }
 
 // NodeUpdatesInHorizon returns all known lightning nodes with updates in the
@@ -642,14 +646,6 @@ func (c *ChannelGraph) ForEachChannel(ctx context.Context,
 	return c.db.ForEachChannel(ctx, v, cb, reset)
 }
 
-// ForEachChannelCacheable iterates through all channel edges for the cache.
-func (c *ChannelGraph) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-	*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
-	reset func()) error {
-
-	return c.db.ForEachChannelCacheable(cb, reset)
-}
-
 // DisabledChannelIDs returns the channel ids of disabled channels.
 func (c *ChannelGraph) DisabledChannelIDs() ([]uint64, error) {
 	return c.db.DisabledChannelIDs()
@@ -923,6 +919,22 @@ func (c *VersionedGraph) ForEachChannel(ctx context.Context,
 	return c.db.ForEachChannel(ctx, c.v, cb, reset)
 }
 
+// ForEachNodeCacheable iterates through all stored vertices/nodes in the graph.
+func (c *VersionedGraph) ForEachNodeCacheable(ctx context.Context,
+	cb func(route.Vertex, *lnwire.FeatureVector) error,
+	reset func()) error {
+
+	return c.db.ForEachNodeCacheable(ctx, c.v, cb, reset)
+}
+
+// ForEachChannelCacheable iterates through all channel edges for the cache.
+func (c *VersionedGraph) ForEachChannelCacheable(
+	cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
+		*models.CachedEdgePolicy) error, reset func()) error {
+
+	return c.db.ForEachChannelCacheable(c.v, cb, reset)
+}
+
 // IsPublicNode determines whether the node is seen as public in the graph.
 func (c *VersionedGraph) IsPublicNode(pubKey [33]byte) (bool, error) {
 	return c.db.IsPublicNode(c.v, pubKey)

graph/db/graph_test.go

@@ -1847,16 +1847,17 @@ func TestGraphTraversalCacheable(t *testing.T) {
 	// 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, func(node route.Vertex,
-		features *lnwire.FeatureVector) error {
+	err = graph.ForEachNodeCacheable(
+		ctx, lnwire.GossipVersion1, func(node route.Vertex,
+			features *lnwire.FeatureVector) error {
 
-		delete(nodeMap, node)
-		nodes = append(nodes, node)
+			delete(nodeMap, node)
+			nodes = append(nodes, node)
 
-		return nil
-	}, func() {
-		nodes = nil
-	})
+			return nil
+		}, func() {
+			nodes = nil
+		})
 	require.NoError(t, err)
 	require.Len(t, nodeMap, 0)
 
@@ -5062,15 +5063,16 @@ func BenchmarkForEachChannel(b *testing.B) {
 		)
 
 		var nodes []route.Vertex
-		err := graph.ForEachNodeCacheable(ctx, func(node route.Vertex,
-			vector *lnwire.FeatureVector) error {
+		err := graph.ForEachNodeCacheable(
+			ctx, lnwire.GossipVersion1, func(node route.Vertex,
+				vector *lnwire.FeatureVector) error {
 
-			nodes = append(nodes, node)
+				nodes = append(nodes, node)
 
-			return nil
-		}, func() {
-			nodes = nil
-		})
+				return nil
+			}, func() {
+				nodes = nil
+			})
 		require.NoError(b, err)
 
 		for _, n := range nodes {

graph/db/interfaces.go

@@ -103,8 +103,9 @@ type Store interface { //nolint:interfacebloat
 	// 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.
-	ForEachNodeCacheable(ctx context.Context, cb func(route.Vertex,
-		*lnwire.FeatureVector) error, reset func()) error
+	ForEachNodeCacheable(ctx context.Context, v lnwire.GossipVersion,
+		cb func(route.Vertex, *lnwire.FeatureVector) error,
+		reset func()) error
 
 	// LookupAlias attempts to return the alias as advertised by the target
 	// node.
@@ -182,9 +183,9 @@ type Store interface { //nolint:interfacebloat
 	//
 	// NOTE: this method is like ForEachChannel but fetches only the data
 	// required for the graph cache.
-	ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-		*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
-		reset func()) error
+	ForEachChannelCacheable(v lnwire.GossipVersion,
+		cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
+			*models.CachedEdgePolicy) error, reset func()) error
 
 	// DisabledChannelIDs returns the channel ids of disabled channels.
 	// A channel is disabled when two of the associated ChanelEdgePolicies

graph/db/kv_store.go

@@ -496,9 +496,13 @@ func forEachChannel(db kvdb.Backend, cb func(*models.ChannelEdgeInfo,
 //
 // NOTE: this method is like ForEachChannel but fetches only the data required
 // for the graph cache.
-func (c *KVStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-	*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
-	reset func()) error {
+func (c *KVStore) ForEachChannelCacheable(v lnwire.GossipVersion,
+	cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
+		*models.CachedEdgePolicy) error, reset func()) error {
+
+	if v != lnwire.GossipVersion1 {
+		return ErrVersionNotSupportedForKVDB
+	}
 
 	return c.db.View(func(tx kvdb.RTx) error {
 		edges := tx.ReadBucket(edgeBucket)
@@ -874,8 +878,12 @@ func forEachNode(db kvdb.Backend,
 // callback returns an error, then the transaction is aborted and the iteration
 // stops early.
 func (c *KVStore) ForEachNodeCacheable(_ context.Context,
-	cb func(route.Vertex, *lnwire.FeatureVector) error,
-	reset func()) error {
+	v lnwire.GossipVersion, cb func(route.Vertex,
+		*lnwire.FeatureVector) error, reset func()) error {
+
+	if v != lnwire.GossipVersion1 {
+		return ErrVersionNotSupportedForKVDB
+	}
 
 	traversal := func(tx kvdb.RTx) error {
 		// First grab the nodes bucket which stores the mapping from

graph/db/sql_store.go

@@ -1003,12 +1003,12 @@ func (s *SQLStore) ForEachNodeDirectedChannel(nodePub route.Vertex,
 // callback returns an error, then the transaction is aborted and the iteration
 // stops early.
 func (s *SQLStore) ForEachNodeCacheable(ctx context.Context,
-	cb func(route.Vertex, *lnwire.FeatureVector) error,
-	reset func()) error {
+	v lnwire.GossipVersion, cb func(route.Vertex,
+		*lnwire.FeatureVector) error, reset func()) error {
 
 	err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error {
 		return forEachNodeCacheable(
-			ctx, s.cfg.QueryCfg, db,
+			ctx, s.cfg.QueryCfg, db, v,
 			func(_ int64, nodePub route.Vertex,
 				features *lnwire.FeatureVector) error {
 
@@ -1547,12 +1547,16 @@ func (s *SQLStore) ForEachNodeCached(ctx context.Context, withAddrs bool,
 //
 // NOTE: this method is like ForEachChannel but fetches only the data
 // required for the graph cache.
-func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
-	*models.CachedEdgePolicy, *models.CachedEdgePolicy) error,
-	reset func()) error {
+func (s *SQLStore) ForEachChannelCacheable(v lnwire.GossipVersion,
+	cb func(*models.CachedEdgeInfo, *models.CachedEdgePolicy,
+		*models.CachedEdgePolicy) error, reset func()) error {
 
 	ctx := context.TODO()
 
+	if !isKnownGossipVersion(v) {
+		return fmt.Errorf("unsupported gossip version: %d", v)
+	}
+
 	handleChannel := func(_ context.Context,
 		row sqlc.ListChannelsWithPoliciesForCachePaginatedRow) error {
 
@@ -1596,7 +1600,7 @@ func (s *SQLStore) ForEachChannelCacheable(cb func(*models.CachedEdgeInfo,
 
 			return db.ListChannelsWithPoliciesForCachePaginated(
 				ctx, sqlc.ListChannelsWithPoliciesForCachePaginatedParams{
-					Version: int16(lnwire.GossipVersion1),
+					Version: int16(v),
 					ID:      lastID,
 					Limit:   limit,
 				},
@@ -3475,11 +3479,12 @@ func forEachNodeDirectedChannel(ctx context.Context, db SQLQueries,
 	return nil
 }
 
-// forEachNodeCacheable fetches all V1 node IDs and pub keys from the database,
+// forEachNodeCacheable fetches all node IDs and pub keys from the database,
 // and executes the provided callback for each node. It does so via pagination
 // along with batch loading of the node feature bits.
 func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig,
-	db SQLQueries, processNode func(nodeID int64, nodePub route.Vertex,
+	db SQLQueries, v lnwire.GossipVersion,
+	processNode func(nodeID int64, nodePub route.Vertex,
 		features *lnwire.FeatureVector) error) error {
 
 	handleNode := func(_ context.Context,
@@ -3504,7 +3509,7 @@ func forEachNodeCacheable(ctx context.Context, cfg *sqldb.QueryConfig,
 
 		return db.ListNodeIDsAndPubKeys(
 			ctx, sqlc.ListNodeIDsAndPubKeysParams{
-				Version: int16(lnwire.GossipVersion1),
+				Version: int16(v),
 				ID:      lastID,
 				Limit:   limit,
 			},

server.go

@@ -1128,7 +1128,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 		DefaultRoutingPolicy: cc.RoutingPolicy,
 		ForAllOutgoingChannels: func(ctx context.Context,
 			cb func(*models.ChannelEdgeInfo,
-			*models.ChannelEdgePolicy) error,
+				*models.ChannelEdgePolicy) error,
 			reset func()) error {
 
 			return s.v1Graph.ForEachNodeChannel(