graph/db: v2 model and store support

discovery/gossiper.go

@@ -2197,9 +2197,12 @@ func (d *AuthenticatedGossiper) addNode(ctx context.Context,
 			err)
 	}
 
-	return d.cfg.Graph.AddNode(
-		ctx, models.NodeFromWireAnnouncement(msg), op...,
-	)
+	node, err := models.NodeFromWireAnnouncement(msg)
+	if err != nil {
+		return err
+	}
+
+	return d.cfg.Graph.AddNode(ctx, node, op...)
 }
 
 // isPremature decides whether a given network message has a block height+delta

docs/release-notes/release-notes-0.22.0.md

@@ -173,6 +173,19 @@
 
 ## Database
 
+* Make the [graph `Store` interface
+  cross-version](https://github.com/lightningnetwork/lnd/pull/11175) so that
+  `ForEachNode`, `ForEachChannel`, and `ForEachNodeDirectedChannel` work across
+  gossip v1 and v2. Add `Preferred` fetch helpers and `GetVersions` queries
+  so callers can retrieve channels without knowing which gossip version
+  announced them.
+
+* Add [v2 model and store
+  support](https://github.com/lightningnetwork/lnd/pull/11306) to the graph
+  database: wire conversion helpers for node announcements, channel auth
+  proofs, edge info, and edge policies; `VersionedGraph` zombie wrappers;
+  SQL queries for v2 node traversal; and v3-only onion address filtering.
+
 ## Code Health
 
 ## Tooling and Documentation
@@ -193,5 +206,7 @@
 * bitromortac
 * Boris Nagaev
 * Erick Cestari
+* Elle Mouton
 * Jared Tobin
 * Kevin Cai
+* ViktorT-11

graph/db/benchmark_test.go

@@ -347,8 +347,7 @@ func TestPopulateDBs(t *testing.T) {
 	// graph.
 	countNodes := func(graph *ChannelGraph) int {
 		numNodes := 0
-		v1Graph := NewVersionedGraph(graph, lnwire.GossipVersion1)
-		err := v1Graph.ForEachNode(
+		err := graph.ForEachNode(
 			ctx,
 			func(node *models.Node) error {
 				numNodes++
@@ -371,7 +370,7 @@ func TestPopulateDBs(t *testing.T) {
 			numPolicies = 0
 		)
 		err := graph.ForEachChannel(
-			ctx, lnwire.GossipVersion1,
+			ctx,
 			func(info *models.ChannelEdgeInfo, policy,
 				policy2 *models.ChannelEdgePolicy) error {
 
@@ -499,7 +498,7 @@ func syncGraph(t *testing.T, src, dest *ChannelGraph) {
 	}
 
 	var wgChans sync.WaitGroup
-	err = src.ForEachChannel(ctx, lnwire.GossipVersion1,
+	err = src.ForEachChannel(ctx,
 		func(info *models.ChannelEdgeInfo,
 			policy1, policy2 *models.ChannelEdgePolicy) error {
 
@@ -623,7 +622,7 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 			name: "ForEachNode",
 			fn: func(b testing.TB, store Store) {
 				err := store.ForEachNode(
-					ctx, lnwire.GossipVersion1,
+					ctx,
 					func(_ *models.Node) error {
 						// Increment the counter to
 						// ensure the callback is doing
@@ -639,12 +638,11 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 		{
 			name: "ForEachChannel",
 			fn: func(b testing.TB, store Store) {
-				//nolint:ll
-				err := store.ForEachChannel(
-					ctx, lnwire.GossipVersion1,
+				err := store.ForEachChannel(ctx,
 					func(_ *models.ChannelEdgeInfo,
+						_,
 						_ *models.ChannelEdgePolicy,
-						_ *models.ChannelEdgePolicy) error {
+					) error {
 
 						// Increment the counter to
 						// ensure the callback is doing
@@ -994,7 +992,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 					)
 
 					err := store.ForEachNode(
-						ctx, lnwire.GossipVersion1,
+						ctx,
 						func(_ *models.Node) error {
 							numNodes++
 
@@ -1005,7 +1003,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 
 					//nolint:ll
 					err = store.ForEachChannel(
-						ctx, lnwire.GossipVersion1,
+						ctx,
 						func(_ *models.ChannelEdgeInfo,
 							_,
 							_ *models.ChannelEdgePolicy) error {

graph/db/graph.go

@@ -214,6 +214,14 @@ func (c *ChannelGraph) handleTopologySubscriptions(ctx context.Context) {
 	}
 }
 
+// cacheableFeatures reports whether the cache should store a feature vector.
+// Empty v2 features must not overwrite v1 features for the same pubkey.
+func cacheableFeatures(v lnwire.GossipVersion,
+	features *lnwire.FeatureVector) bool {
+
+	return v != gossipV2 || !features.IsEmpty()
+}
+
 // populateCache loads the entire channel graph into the in-memory graph cache.
 func (c *ChannelGraph) populateCache(ctx context.Context) error {
 	if c.cache == nil {
@@ -238,12 +246,21 @@ func (c *ChannelGraph) populateCache(ctx context.Context) 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.
+		// We iterate v1 first, then v2. AddNodeFeatures overwrites on
+		// key collision, so v2 features take precedence when both
+		// versions exist, except for the entries that
+		// cacheableFeatures rejects. AddChannel ranks a colliding SCID
+		// itself, using the same rule as the preferred lookup tables,
+		// so the iteration order here does not decide which version a
+		// channel is cached from.
 		err := c.db.ForEachNodeCacheable(ctx, v,
 			func(node route.Vertex,
 				features *lnwire.FeatureVector) error {
 
+				if !cacheableFeatures(v, features) {
+					return nil
+				}
+
 				cache.AddNodeFeatures(node, features)
 
 				return nil
@@ -299,9 +316,8 @@ func (c *ChannelGraph) ForEachNodeDirectedChannel(ctx context.Context,
 		return c.cache.graphCache.ForEachChannel(node, cb)
 	}
 
-	// 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.
+	// The no-cache path only runs against the KV backend, which is
+	// v1-only.
 	return c.db.ForEachNodeDirectedChannel(
 		ctx, gossipV1, node, cb, reset,
 	)
@@ -320,13 +336,16 @@ func (c *ChannelGraph) FetchNodeFeatures(ctx context.Context,
 		return c.cache.graphCache.GetFeatures(node), nil
 	}
 
-	return c.db.FetchNodeFeatures(ctx, lnwire.GossipVersion1, node)
+	// The no-cache path only runs against the KV backend, which is
+	// v1-only.
+	return c.db.FetchNodeFeatures(ctx, gossipV1, node)
 }
 
 // GraphSession will provide the call-back with access to a NodeTraverser
 // instance which can be used to perform queries against the channel graph. If
 // the graph cache is not enabled, then the call-back will be provided with
-// access to the graph via a consistent read-only transaction.
+// access to the graph via a consistent read-only transaction; the no-cache
+// path only runs against the KV backend, which is v1-only.
 func (c *ChannelGraph) GraphSession(ctx context.Context,
 	cb func(graph NodeTraverser) error, reset func()) error {
 
@@ -373,7 +392,7 @@ func (c *ChannelGraph) AddNode(ctx context.Context,
 		return err
 	}
 
-	if c.cache != nil {
+	if c.cache != nil && cacheableFeatures(node.Version, node.Features) {
 		c.cache.applyUpdate(func(cache *GraphCache) {
 			cache.AddNodeFeatures(
 				node.PubKeyBytes, node.Features,
@@ -679,13 +698,22 @@ func (c *ChannelGraph) HasV1Node(ctx context.Context,
 	return c.db.HasV1Node(ctx, nodePub)
 }
 
-// ForEachChannel iterates through all channel edges stored within the graph.
+// ForEachChannel iterates through all channel edges stored within the graph
+// across all gossip versions.
 func (c *ChannelGraph) ForEachChannel(ctx context.Context,
-	v lnwire.GossipVersion, cb func(*models.ChannelEdgeInfo,
-		*models.ChannelEdgePolicy, *models.ChannelEdgePolicy) error,
+	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
+		*models.ChannelEdgePolicy) error,
 	reset func()) error {
 
-	return c.db.ForEachChannel(ctx, v, cb, reset)
+	return c.db.ForEachChannel(ctx, cb, reset)
+}
+
+// ForEachNode iterates through all stored vertices/nodes in the graph across
+// all gossip versions.
+func (c *ChannelGraph) ForEachNode(ctx context.Context,
+	cb func(*models.Node) error, reset func()) error {
+
+	return c.db.ForEachNode(ctx, cb, reset)
 }
 
 // DisabledChannelIDs returns the channel ids of disabled channels.
@@ -792,8 +820,7 @@ func (c *VersionedGraph) FilterKnownChanIDs(ctx context.Context,
 		// alive, and we let it be added to the set of IDs to
 		// query our peer for.
 		err := c.db.MarkEdgeLive(
-			ctx, info.Version,
-			info.ShortChannelID.ToUint64(),
+			ctx, c.v, info.ShortChannelID.ToUint64(),
 		)
 		// Since there is a chance that the edge could have been
 		// marked as "live" between the FilterKnownChanIDs call
@@ -817,26 +844,23 @@ func (c *ChannelGraph) FetchChanInfos(ctx context.Context,
 }
 
 // FetchChannelEdgesByOutpoint attempts to lookup directed edges by funding
-// outpoint.
+// outpoint, returning the highest available gossip version.
 func (c *ChannelGraph) FetchChannelEdgesByOutpoint(ctx context.Context,
 	op *wire.OutPoint) (
 	*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
 	*models.ChannelEdgePolicy, error) {
 
-	return c.db.FetchChannelEdgesByOutpoint(
-		ctx, lnwire.GossipVersion1, op,
-	)
+	return c.db.FetchChannelEdgesByOutpointPreferred(ctx, op)
 }
 
-// FetchChannelEdgesByID attempts to lookup directed edges by channel ID.
+// FetchChannelEdgesByID attempts to lookup directed edges by channel ID,
+// returning the highest available gossip version.
 func (c *ChannelGraph) FetchChannelEdgesByID(ctx context.Context,
 	chanID uint64) (
 	*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
 	*models.ChannelEdgePolicy, error) {
 
-	return c.db.FetchChannelEdgesByID(
-		ctx, lnwire.GossipVersion1, chanID,
-	)
+	return c.db.FetchChannelEdgesByIDPreferred(ctx, chanID)
 }
 
 // PutClosedScid stores a SCID for a closed channel in the database.
@@ -869,6 +893,14 @@ func (c *ChannelGraph) PruneTip(ctx context.Context) (*chainhash.Hash,
 
 // VersionedGraph is a wrapper around ChannelGraph that will call underlying
 // Store methods with a specific gossip version.
+//
+// NOTE: ForEachNode and ForEachChannel are the exception. They are promoted
+// from the embedded ChannelGraph and iterate every gossip version, yielding
+// the preferred record per pub key or SCID. Both callers that reach them
+// through this wrapper want exactly that: autopilot ranks every candidate node
+// for channel opening, and the discovery bootstrapper dials every known
+// address, so neither has a reason to ignore a v2 announcement. A caller that
+// does need one version must use a version-taking Store method instead.
 type VersionedGraph struct {
 	*ChannelGraph
 	v lnwire.GossipVersion
@@ -926,13 +958,6 @@ func (c *VersionedGraph) ForEachNodeCached(ctx context.Context,
 	return c.ChannelGraph.ForEachNodeCached(ctx, c.v, cb, reset)
 }
 
-// ForEachNode iterates through all stored vertices/nodes in the graph.
-func (c *VersionedGraph) ForEachNode(ctx context.Context,
-	cb func(*models.Node) error, reset func()) error {
-
-	return c.db.ForEachNode(ctx, c.v, cb, reset)
-}
-
 // NumZombies returns the current number of zombie channels in the graph.
 func (c *VersionedGraph) NumZombies(ctx context.Context) (uint64, error) {
 	return c.db.NumZombies(ctx, c.v)
@@ -964,24 +989,6 @@ func (c *VersionedGraph) ChannelView(ctx context.Context) ([]EdgePoint,
 	return c.db.ChannelView(ctx, c.v)
 }
 
-// GraphSession provides the callback with access to a NodeTraverser instance
-// for performing queries against the channel graph. If the graph cache is
-// enabled, the callback receives the VersionedGraph directly (which implements
-// NodeTraverser using the cache). Otherwise a read-only database session is
-// used.
-func (c *VersionedGraph) GraphSession(ctx context.Context,
-	cb func(graph NodeTraverser) error, reset func()) error {
-
-	if c.cache != nil && c.cache.isLoaded() {
-		return cb(c)
-	}
-
-	// TODO(elle): the underlying GraphSession currently creates a
-	// NodeTraverser that is hardcoded to GossipVersion1. This needs to be
-	// updated to pass the version through for v2 support.
-	return c.db.GraphSession(ctx, cb, reset)
-}
-
 // FetchNode attempts to look up a target node by its identity public key.
 func (c *VersionedGraph) FetchNode(ctx context.Context,
 	nodePub route.Vertex) (*models.Node, error) {
@@ -1079,6 +1086,23 @@ func (c *VersionedGraph) DeleteChannelEdges(ctx context.Context,
 	)
 }
 
+// MarkEdgeZombie marks a channel as a zombie for this version.
+func (c *VersionedGraph) MarkEdgeZombie(ctx context.Context, chanID uint64,
+	pubKey1, pubKey2 [33]byte) error {
+
+	return c.ChannelGraph.MarkEdgeZombie(
+		ctx, c.v, chanID, pubKey1, pubKey2,
+	)
+}
+
+// MarkEdgeLive clears an edge from our zombie index for this version, deeming
+// it as live.
+func (c *VersionedGraph) MarkEdgeLive(ctx context.Context,
+	chanID uint64) error {
+
+	return c.ChannelGraph.MarkEdgeLive(ctx, c.v, chanID)
+}
+
 // HasChannelEdge returns true if the database knows of a channel edge with the
 // passed channel ID and this graph's gossip version, and false otherwise. If it
 // is not found, then the zombie index is checked and its result is returned as
@@ -1106,14 +1130,6 @@ func (c *VersionedGraph) ForEachNodeChannel(ctx context.Context,
 	return c.db.ForEachNodeChannel(ctx, c.v, nodePub, cb, reset)
 }
 
-// ForEachChannel iterates through all channel edges stored within the graph.
-func (c *VersionedGraph) ForEachChannel(ctx context.Context,
-	cb func(*models.ChannelEdgeInfo, *models.ChannelEdgePolicy,
-		*models.ChannelEdgePolicy) error, reset func()) error {
-
-	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,