graph/db: implement prefer-highest-version for node traversal

Update ForEachNodeDirectedChannel and FetchNodeFeatures on
ChannelGraph to iterate across all gossip versions (highest first)
instead of hardcoding v1. This ensures that channels announced via
v2 are preferred over v1 and that v2 features are used when
available.

Author: ellemouton

graph/db/graph.go

@@ -63,6 +63,14 @@ type ChannelGraph struct {
 	cancel fn.Option[context.CancelFunc]
 }
 
+// preferHighestNodeDirectedChanneler is implemented by stores that can stream
+// cross-version node-directed channel traversals directly.
+type preferHighestNodeDirectedChanneler interface {
+	ForEachNodeDirectedChannelPreferHighest(ctx context.Context,
+		node route.Vertex, cb func(channel *DirectedChannel) error,
+		reset func()) error
+}
+
 // NewChannelGraph creates a new ChannelGraph instance with the given backend.
 func NewChannelGraph(v1Store Store,
 	options ...ChanGraphOption) (*ChannelGraph, error) {
@@ -238,8 +246,9 @@ 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. Since AddNodeFeatures and
+		// AddChannel overwrite on key collision, v2 data naturally
+		// takes precedence when both versions exist.
 		err := c.db.ForEachNodeCacheable(ctx, v,
 			func(node route.Vertex,
 				features *lnwire.FeatureVector) error {
@@ -299,12 +308,59 @@ 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.
-	return c.db.ForEachNodeDirectedChannel(
-		ctx, gossipV1, node, cb, reset,
-	)
+	if db, ok := c.db.(preferHighestNodeDirectedChanneler); ok {
+		return db.ForEachNodeDirectedChannelPreferHighest(
+			ctx, node, cb, reset,
+		)
+	}
+
+	// Iterate across all gossip versions (highest first) so that
+	// channels announced via v2 are preferred over v1. We buffer
+	// results and deliver them at the end so that ExecTx retries
+	// within a single version don't corrupt the caller's state or
+	// lose channels from already-committed versions.
+	seen := make(map[uint64]struct{})
+	var allChannels []*DirectedChannel
+
+	for _, v := range []lnwire.GossipVersion{gossipV2, gossipV1} {
+		prevLen := len(allChannels)
+		err := c.db.ForEachNodeDirectedChannel(
+			ctx, v, node,
+			func(channel *DirectedChannel) error {
+				if _, ok := seen[channel.ChannelID]; ok {
+					return nil
+				}
+				seen[channel.ChannelID] = struct{}{}
+
+				allChannels = append(allChannels, channel)
+
+				return nil
+			},
+			func() {
+				// On ExecTx retry, undo this version's
+				// additions while keeping channels from
+				// earlier (already committed) versions.
+				for _, ch := range allChannels[prevLen:] {
+					delete(seen, ch.ChannelID)
+				}
+				allChannels = allChannels[:prevLen]
+			},
+		)
+		if err != nil &&
+			!errors.Is(err, ErrVersionNotSupportedForKVDB) {
+
+			return err
+		}
+	}
+
+	// Deliver the collected channels to the caller.
+	for _, ch := range allChannels {
+		if err := cb(ch); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 // FetchNodeFeatures returns the features of the given node. If no features are
@@ -320,7 +376,22 @@ func (c *ChannelGraph) FetchNodeFeatures(ctx context.Context,
 		return c.cache.graphCache.GetFeatures(node), nil
 	}
 
-	return c.db.FetchNodeFeatures(ctx, lnwire.GossipVersion1, node)
+	// Try v2 first, fall back to v1 if the v2 features are empty.
+	for _, v := range []lnwire.GossipVersion{gossipV2, gossipV1} {
+		features, err := c.db.FetchNodeFeatures(ctx, v, node)
+		if errors.Is(err, ErrVersionNotSupportedForKVDB) {
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+
+		if !features.IsEmpty() {
+			return features, nil
+		}
+	}
+
+	return lnwire.EmptyFeatureVector(), nil
 }
 
 // GraphSession will provide the call-back with access to a NodeTraverser

graph/db/graph_test.go

@@ -6234,49 +6234,56 @@ func TestPreferHighestAndGetVersions(t *testing.T) {
 		return
 	}
 
-	// Add v2 node announcements for the same pubkeys so we can create a
-	// versioned duplicate of the channel in the SQL store.
 	node1V2 := createNode(t, lnwire.GossipVersion2, node1Priv)
 	node2V2 := createNode(t, lnwire.GossipVersion2, node2Priv)
-
 	require.NoError(t, graph.AddNode(ctx, node1V2))
 	require.NoError(t, graph.AddNode(ctx, node2V2))
 
-	// Add a v2 copy of the same channel so the prefer-highest queries can
-	// choose between both versions for the same SCID/outpoint.
-	edgeInfoV2, _ := createEdge(
-		lnwire.GossipVersion2, 100, 1, 0, 1, node1V2, node2V2,
+	// Add a duplicate v1/v2 channel and verify prefer-highest chooses
+	// the v2 edge while GetVersions reports both versions.
+	dupV1, dupSCID := createEdge(
+		lnwire.GossipVersion1, 101, 1, 0, 2, node1V1, node2V1,
+	)
+	dupV2, _ := createEdge(
+		lnwire.GossipVersion2, 101, 1, 0, 2, node1V2, node2V2,
 	)
-	require.NoError(t, graph.AddChannelEdge(ctx, edgeInfoV2))
+	require.NoError(t, graph.AddChannelEdge(ctx, dupV1))
+	require.NoError(t, graph.AddChannelEdge(ctx, dupV2))
+
+	dupChanID := dupSCID.ToUint64()
+	dupOutpoint := dupV1.ChannelPoint
 
-	versions, err = store.GetVersionsBySCID(ctx, chanID)
+	info, _, _, err = store.FetchChannelEdgesByIDPreferHighest(
+		ctx, dupChanID,
+	)
+	require.NoError(t, err)
+	require.Equal(t, dupChanID, info.ChannelID)
+	require.Equal(t, lnwire.GossipVersion2, info.Version)
+
+	info, _, _, err = store.FetchChannelEdgesByOutpointPreferHighest(
+		ctx, &dupOutpoint,
+	)
+	require.NoError(t, err)
+	require.Equal(t, dupChanID, info.ChannelID)
+	require.Equal(t, lnwire.GossipVersion2, info.Version)
+
+	versions, err = store.GetVersionsBySCID(ctx, dupChanID)
 	require.NoError(t, err)
 	require.Equal(t, []lnwire.GossipVersion{
 		lnwire.GossipVersion1,
 		lnwire.GossipVersion2,
 	}, versions)
 
-	versions, err = store.GetVersionsByOutpoint(ctx, &op)
+	versions, err = store.GetVersionsByOutpoint(ctx, &dupOutpoint)
 	require.NoError(t, err)
 	require.Equal(t, []lnwire.GossipVersion{
 		lnwire.GossipVersion1,
 		lnwire.GossipVersion2,
 	}, versions)
-
-	// With no policies present for either version, prefer-highest should
-	// fall back to the higher gossip version.
-	info, _, _, err = store.FetchChannelEdgesByIDPreferHighest(ctx, chanID)
-	require.NoError(t, err)
-	require.Equal(t, lnwire.GossipVersion2, info.Version)
-
-	info, _, _, err = store.FetchChannelEdgesByOutpointPreferHighest(
-		ctx, &op,
-	)
-	require.NoError(t, err)
-	require.Equal(t, lnwire.GossipVersion2, info.Version)
 }
-// TestPreferHighestNodeTraversal verifies the no-cache traversal behavior of
-// ChannelGraph's ForEachNodeDirectedChannel and FetchNodeFeatures.
+// TestPreferHighestNodeTraversal verifies that ChannelGraph's
+// ForEachNodeDirectedChannel and FetchNodeFeatures correctly prefer v2 over v1
+// when the graph cache is disabled (exercising the no-cache code paths).
 func TestPreferHighestNodeTraversal(t *testing.T) {
 	t.Parallel()
 
@@ -6314,33 +6321,35 @@ func TestPreferHighestNodeTraversal(t *testing.T) {
 
 	features, err = graph.FetchNodeFeatures(ctx, nodeV2.PubKeyBytes)
 	require.NoError(t, err)
-	require.Empty(t, features.Features(),
-		"no-cache traversal still uses the v1 feature lookup path")
+	require.False(t, features.IsEmpty(),
+		"v2-only node should have features")
 
 	// Create a node with both v1 and v2 announcements.
 	privBoth, err := btcec.NewPrivateKey()
 	require.NoError(t, err)
 
 	nodeBothV1 := createNode(t, lnwire.GossipVersion1, privBoth)
-	nodeBothV1.Features = lnwire.NewFeatureVector(
+	v1Features := lnwire.NewFeatureVector(
 		lnwire.NewRawFeatureVector(lnwire.GossipQueriesRequired),
 		lnwire.Features,
 	)
+	nodeBothV1.Features = v1Features
 	require.NoError(t, graph.AddNode(ctx, nodeBothV1))
 
 	nodeBothV2 := createNode(t, lnwire.GossipVersion2, privBoth)
-	nodeBothV2.Features = lnwire.NewFeatureVector(
-		lnwire.NewRawFeatureVector(lnwire.DataLossProtectRequired),
+	v2Features := lnwire.NewFeatureVector(
+		lnwire.NewRawFeatureVector(lnwire.TLVOnionPayloadRequired),
 		lnwire.Features,
 	)
+	nodeBothV2.Features = v2Features
 	require.NoError(t, graph.AddNode(ctx, nodeBothV2))
 
 	features, err = graph.FetchNodeFeatures(
 		ctx, nodeBothV1.PubKeyBytes,
 	)
 	require.NoError(t, err)
-	require.Equal(t, nodeBothV1.Features, features,
-		"no-cache feature lookup should return the v1 features")
+	require.Equal(t, v2Features, features)
+	require.NotEqual(t, v1Features, features)
 
 	// --- ForEachNodeDirectedChannel ---
 
@@ -6569,3 +6578,97 @@ func TestPreferHighestForEachChannel(t *testing.T) {
 	require.Nil(t, gotShellPref.p1)
 	require.Nil(t, gotShellPref.p2)
 }
+
+// TestPreferHighestNodeDirectedChannelTraversal verifies that the no-cache
+// ChannelGraph.ForEachNodeDirectedChannel path streams one directed channel per
+// SCID while preferring the v2 advertisement when both versions exist.
+func TestPreferHighestNodeDirectedChannelTraversal(t *testing.T) {
+	t.Parallel()
+
+	if !isSQLDB {
+		t.Skip("prefer-highest requires SQL backend")
+	}
+
+	ctx := t.Context()
+	graph := MakeTestGraph(t, WithUseGraphCache(false))
+
+	localPriv, err := btcec.NewPrivateKey()
+	require.NoError(t, err)
+	peerBothPriv, err := btcec.NewPrivateKey()
+	require.NoError(t, err)
+	peerV1OnlyPriv, err := btcec.NewPrivateKey()
+	require.NoError(t, err)
+
+	localV1 := createNode(t, lnwire.GossipVersion1, localPriv)
+	localV2 := createNode(t, lnwire.GossipVersion2, localPriv)
+	peerBothV1 := createNode(t, lnwire.GossipVersion1, peerBothPriv)
+	peerBothV2 := createNode(t, lnwire.GossipVersion2, peerBothPriv)
+	peerV1Only := createNode(t, lnwire.GossipVersion1, peerV1OnlyPriv)
+
+	require.NoError(t, graph.AddNode(ctx, localV1))
+	require.NoError(t, graph.AddNode(ctx, localV2))
+	require.NoError(t, graph.AddNode(ctx, peerBothV1))
+	require.NoError(t, graph.AddNode(ctx, peerBothV2))
+	require.NoError(t, graph.AddNode(ctx, peerV1Only))
+
+	dupV1, _ := createEdge(
+		lnwire.GossipVersion1, 400, 0, 0, 10, localV1, peerBothV1,
+	)
+	dupV2, _ := createEdge(
+		lnwire.GossipVersion2, 400, 0, 0, 10, localV2, peerBothV2,
+	)
+	v1Only, _ := createEdge(
+		lnwire.GossipVersion1, 401, 0, 0, 11, localV1, peerV1Only,
+	)
+
+	require.NoError(t, graph.AddChannelEdge(ctx, dupV1))
+	require.NoError(t, graph.AddChannelEdge(ctx, dupV2))
+	require.NoError(t, graph.AddChannelEdge(ctx, v1Only))
+
+	addPolicies := func(edgeInfo *models.ChannelEdgeInfo,
+		version lnwire.GossipVersion,
+		fee lnwire.MilliSatoshi) {
+
+		policy1 := newEdgePolicy(version, edgeInfo.ChannelID, 1000, true)
+		policy1.ToNode = edgeInfo.NodeKey2Bytes
+		policy1.SigBytes = testSig.Serialize()
+		policy1.FeeBaseMSat = fee
+
+		policy2 := newEdgePolicy(version, edgeInfo.ChannelID, 1001, false)
+		policy2.ToNode = edgeInfo.NodeKey1Bytes
+		policy2.SigBytes = testSig.Serialize()
+		policy2.FeeBaseMSat = fee
+
+		require.NoError(t, graph.UpdateEdgePolicy(ctx, policy1))
+		require.NoError(t, graph.UpdateEdgePolicy(ctx, policy2))
+	}
+
+	addPolicies(dupV1, lnwire.GossipVersion1, 1111)
+	addPolicies(dupV2, lnwire.GossipVersion2, 2222)
+	addPolicies(v1Only, lnwire.GossipVersion1, 3333)
+
+	channelsByID := make(map[uint64]*DirectedChannel)
+	err = graph.ForEachNodeDirectedChannel(
+		ctx, localV1.PubKeyBytes,
+		func(channel *DirectedChannel) error {
+			channelsByID[channel.ChannelID] = channel.DeepCopy()
+			return nil
+		}, func() {
+			clear(channelsByID)
+		},
+	)
+	require.NoError(t, err)
+	require.Len(t, channelsByID, 2)
+
+	gotDup := channelsByID[dupV1.ChannelID]
+	require.NotNil(t, gotDup)
+	require.NotNil(t, gotDup.InPolicy)
+	require.Equal(t, lnwire.MilliSatoshi(2222), gotDup.InPolicy.FeeBaseMSat)
+
+	gotV1Only := channelsByID[v1Only.ChannelID]
+	require.NotNil(t, gotV1Only)
+	require.NotNil(t, gotV1Only.InPolicy)
+	require.Equal(
+		t, lnwire.MilliSatoshi(3333), gotV1Only.InPolicy.FeeBaseMSat,
+	)
+}