graph/db: cross-version graph Store

discovery/chan_series.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/lightningnetwork/lnd/fn/v2"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/netann"
@@ -115,7 +116,10 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 		// First, we'll query for all the set of channels that have an
 		// update that falls within the specified horizon.
 		chansInHorizon := c.graph.ChanUpdatesInHorizon(
-			context.TODO(), startTime, endTime,
+			context.TODO(), graphdb.ChanUpdateRange{
+				StartTime: fn.Some(startTime),
+				EndTime:   fn.Some(endTime),
+			},
 		)
 
 		for channel, err := range chansInHorizon {
@@ -181,7 +185,10 @@ func (c *ChanSeries) UpdatesInHorizon(chain chainhash.Hash,
 		// update within the horizon as well. We send these second to
 		// ensure that they follow any active channels they have.
 		nodeAnnsInHorizon := c.graph.NodeUpdatesInHorizon(
-			context.TODO(), startTime, endTime,
+			context.TODO(), graphdb.NodeUpdateRange{
+				StartTime: fn.Some(startTime),
+				EndTime:   fn.Some(endTime),
+			},
 			graphdb.WithIterPublicNodesOnly(),
 		)
 		for nodeAnn, err := range nodeAnnsInHorizon {

docs/release-notes/release-notes-0.21.0.md

@@ -267,6 +267,12 @@
     [4](https://github.com/lightningnetwork/lnd/pull/10542),
     [5](https://github.com/lightningnetwork/lnd/pull/10572),
     [6](https://github.com/lightningnetwork/lnd/pull/10582).
+* Make the [graph `Store` interface
+  cross-version](https://github.com/lightningnetwork/lnd/pull/10656) so that
+  query methods (`ForEachNode`, `ForEachChannel`, `NodeUpdatesInHorizon`,
+  `ChanUpdatesInHorizon`, `FilterKnownChanIDs`) work across gossip v1 and v2.
+  Add `PreferHighest` fetch helpers and `GetVersions` queries so callers can
+  retrieve channels without knowing which gossip version announced them.
 * Updated waiting proof persistence for gossip upgrades by introducing typed
   waiting proof keys and payloads, with a DB migration to rewrite legacy
   waiting proof records to the new key/value format

graph/builder.go

@@ -12,6 +12,7 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/lightningnetwork/lnd/batch"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/fn/v2"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/graph/db/models"
 	"github.com/lightningnetwork/lnd/lnutils"
@@ -648,7 +649,11 @@ func (b *Builder) pruneZombieChans() error {
 	startTime := time.Unix(0, 0)
 	endTime := time.Now().Add(-1 * chanExpiry)
 	oldEdgesIter := b.cfg.Graph.ChanUpdatesInHorizon(
-		context.TODO(), startTime, endTime,
+		context.TODO(), lnwire.GossipVersion1,
+		graphdb.ChanUpdateRange{
+			StartTime: fn.Some(startTime),
+			EndTime:   fn.Some(endTime),
+		},
 	)
 
 	for u, err := range oldEdgesIter {

graph/db/benchmark_test.go

@@ -372,7 +372,7 @@ func TestPopulateDBs(t *testing.T) {
 			numPolicies = 0
 		)
 		err := graph.ForEachChannel(
-			ctx, lnwire.GossipVersion1,
+			ctx,
 			func(info *models.ChannelEdgeInfo, policy,
 				policy2 *models.ChannelEdgePolicy) error {
 
@@ -500,7 +500,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 {
 
@@ -624,7 +624,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
@@ -640,12 +640,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
@@ -662,7 +661,13 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 			name: "NodeUpdatesInHorizon",
 			fn: func(b testing.TB, store Store) {
 				iter := store.NodeUpdatesInHorizon(
-					ctx, time.Unix(0, 0), time.Now(),
+					ctx, lnwire.GossipVersion1,
+					NodeUpdateRange{
+						StartTime: fn.Some(
+							time.Unix(0, 0),
+						),
+						EndTime: fn.Some(time.Now()),
+					},
 				)
 				_, err := fn.CollectErr(iter)
 				require.NoError(b, err)
@@ -713,7 +718,13 @@ func BenchmarkGraphReadMethods(b *testing.B) {
 			name: "ChanUpdatesInHorizon",
 			fn: func(b testing.TB, store Store) {
 				iter := store.ChanUpdatesInHorizon(
-					ctx, time.Unix(0, 0), time.Now(),
+					ctx, lnwire.GossipVersion1,
+					ChanUpdateRange{
+						StartTime: fn.Some(
+							time.Unix(0, 0),
+						),
+						EndTime: fn.Some(time.Now()),
+					},
 				)
 				_, err := fn.CollectErr(iter)
 				require.NoError(b, err)
@@ -817,7 +828,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 					)
 
 					err := store.ForEachNode(
-						ctx, lnwire.GossipVersion1,
+						ctx,
 						func(_ *models.Node) error {
 							numNodes++
 
@@ -828,7 +839,7 @@ func BenchmarkFindOptimalSQLQueryConfig(b *testing.B) {
 
 					//nolint:ll
 					err = store.ForEachChannel(
-						ctx, lnwire.GossipVersion1,
+						ctx,
 						func(_ *models.ChannelEdgeInfo,
 							_,
 							_ *models.ChannelEdgePolicy) error {