[draft] graph/db+sqldb: graph store SQL implementation + migration

.github/workflows/main.yml

@@ -368,10 +368,14 @@ jobs:
             args: backend=bitcoind dbbackend=sqlite
           - name: bitcoind-sqlite-nativesql
             args: backend=bitcoind dbbackend=sqlite nativesql=true
+          - name: bitcoind-sqlite=nativesql-experiment
+            args: backend=bitcoind dbbackend=sqlite nativesql=true tags=test_native_sql
           - name: bitcoind-postgres
             args: backend=bitcoind dbbackend=postgres
           - name: bitcoind-postgres-nativesql
             args: backend=bitcoind dbbackend=postgres nativesql=true
+          - name: bitcoind-postgres-nativesql-experiment
+            args: backend=bitcoind dbbackend=postgres nativesql=true tags=test_native_sql
     steps:
       - name: git checkout
         uses: actions/checkout@v4

config_builder.go

@@ -1046,23 +1046,13 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 		)
 	}
 
-	graphStore, err := graphdb.NewKVStore(
+	kvGraphStore, err := graphdb.NewKVStore(
 		databaseBackends.GraphDB, graphDBOptions...,
 	)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	dbs.GraphDB, err = graphdb.NewChannelGraph(graphStore, chanGraphOpts...)
-	if err != nil {
-		cleanUp()
-
-		err = fmt.Errorf("unable to open graph DB: %w", err)
-		d.logger.Error(err)
-
-		return nil, nil, err
-	}
-
 	dbOptions := []channeldb.OptionModifier{
 		channeldb.OptionDryRunMigration(cfg.DryRunMigration),
 		channeldb.OptionKeepFailedPaymentAttempts(
@@ -1098,6 +1088,8 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 		return nil, nil, err
 	}
 
+	var graphStore graphdb.V1Store
+
 	// Instantiate a native SQL store if the flag is set.
 	if d.cfg.DB.UseNativeSQL {
 		migrations := sqldb.GetMigrations()
@@ -1110,7 +1102,7 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 		// migration's version (7), it will be skipped permanently,
 		// regardless of the flag.
 		if !d.cfg.DB.SkipNativeSQLMigration {
-			migrationFn := func(tx *sqlc.Queries) error {
+			invoiceMigFn := func(tx *sqlc.Queries) error {
 				err := invoices.MigrateInvoicesToSQL(
 					ctx, dbs.ChanStateDB.Backend,
 					dbs.ChanStateDB, tx,
@@ -1132,11 +1124,21 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 			// Make sure we attach the custom migration function to
 			// the correct migration version.
 			for i := 0; i < len(migrations); i++ {
-				if migrations[i].Version != invoiceMigration {
+				version := migrations[i].Version
+				if version == invoiceMigration {
+					migrations[i].MigrationFn = invoiceMigFn
+
+					continue
+				}
+
+				migFn, ok := getSQLMigration(
+					ctx, version, kvGraphStore,
+				)
+				if !ok {
 					continue
 				}
 
-				migrations[i].MigrationFn = migrationFn
+				migrations[i].MigrationFn = migFn
 			}
 		}
 
@@ -1156,17 +1158,27 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 		// the base DB and transaction executor for the native SQL
 		// invoice store.
 		baseDB := dbs.NativeSQLStore.GetBaseDB()
-		executor := sqldb.NewTransactionExecutor(
+		invoiceExecutor := sqldb.NewTransactionExecutor(
 			baseDB, func(tx *sql.Tx) invoices.SQLInvoiceQueries {
 				return baseDB.WithTx(tx)
 			},
 		)
 
 		sqlInvoiceDB := invoices.NewSQLStore(
-			executor, clock.NewDefaultClock(),
+			invoiceExecutor, clock.NewDefaultClock(),
 		)
 
 		dbs.InvoiceDB = sqlInvoiceDB
+
+		graphStore, err = d.getGraphStore(
+			baseDB, kvGraphStore, graphDBOptions...,
+		)
+		if err != nil {
+			err = fmt.Errorf("unable to get graph store: %w", err)
+			d.logger.Error(err)
+
+			return nil, nil, err
+		}
 	} else {
 		// Check if the invoice bucket tombstone is set. If it is, we
 		// need to return and ask the user switch back to using the
@@ -1188,6 +1200,18 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 		}
 
 		dbs.InvoiceDB = dbs.ChanStateDB
+
+		graphStore = kvGraphStore
+	}
+
+	dbs.GraphDB, err = graphdb.NewChannelGraph(graphStore, chanGraphOpts...)
+	if err != nil {
+		cleanUp()
+
+		err = fmt.Errorf("unable to open channel graph: %w", err)
+		d.logger.Error(err)
+
+		return nil, nil, err
 	}
 
 	// Wrap the watchtower client DB and make sure we clean up.

config_prod.go

@@ -0,0 +1,24 @@
+//go:build !test_native_sql
+
+package lnd
+
+import (
+	"context"
+
+	graphdb "github.com/lightningnetwork/lnd/graph/db"
+	"github.com/lightningnetwork/lnd/sqldb"
+	"github.com/lightningnetwork/lnd/sqldb/sqlc"
+)
+
+func getSQLMigration(ctx context.Context, version int,
+	kvGraphStore *graphdb.KVStore) (func(tx *sqlc.Queries) error, bool) {
+
+	return nil, false
+}
+
+func (d *DefaultDatabaseBuilder) getGraphStore(_ *sqldb.BaseDB,
+	kvGraphStore *graphdb.KVStore,
+	_ ...graphdb.StoreOptionModifier) (graphdb.V1Store, error) {
+
+	return kvGraphStore, nil
+}

config_test_native_sql.go

@@ -0,0 +1,53 @@
+//go:build test_native_sql
+
+package lnd
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	graphdb "github.com/lightningnetwork/lnd/graph/db"
+	"github.com/lightningnetwork/lnd/sqldb"
+	"github.com/lightningnetwork/lnd/sqldb/sqlc"
+)
+
+const graphSQLMigration = 9
+
+func getSQLMigration(ctx context.Context, version int,
+	kvGraphStore *graphdb.KVStore) (func(tx *sqlc.Queries) error, bool) {
+
+	if version != graphSQLMigration {
+		return nil, false
+	}
+
+	return func(tx *sqlc.Queries) error {
+		err := graphdb.MigrateGraphToSQL(
+			ctx, kvGraphStore, tx,
+		)
+		if err != nil {
+			return fmt.Errorf("failed to migrate "+
+				"graph to SQL: %w", err)
+		}
+
+		return nil
+	}, true
+}
+
+func (d *DefaultDatabaseBuilder) getGraphStore(baseDB *sqldb.BaseDB,
+	_ *graphdb.KVStore,
+	opts ...graphdb.StoreOptionModifier) (graphdb.V1Store, error) {
+
+	graphExecutor := sqldb.NewTransactionExecutor(
+		baseDB, func(tx *sql.Tx) graphdb.SQLQueries {
+			return baseDB.WithTx(tx)
+		},
+	)
+
+	return graphdb.NewSQLStore(
+		&graphdb.SQLStoreConfig{
+			ChainHash: *d.cfg.ActiveNetParams.GenesisHash,
+		},
+		graphExecutor, opts...,
+	)
+}

docs/release-notes/release-notes-0.20.0.md

@@ -86,6 +86,11 @@ circuit. The indices are only available for forwarding events saved after v0.20.
     * [4](https://github.com/lightningnetwork/lnd/pull/9931)
     * [5](https://github.com/lightningnetwork/lnd/pull/9935)
     * [6](https://github.com/lightningnetwork/lnd/pull/9936)
+    * [7](https://github.com/lightningnetwork/lnd/pull/9937)
+    * [8](https://github.com/lightningnetwork/lnd/pull/9938)
+    * [9](https://github.com/lightningnetwork/lnd/pull/9939)
+    * [10](https://github.com/lightningnetwork/lnd/pull/9971)
+    * [11](https://github.com/lightningnetwork/lnd/pull/9972)
 
 ## RPC Updates
 

graph/builder_test.go

@@ -892,15 +892,21 @@ func TestPruneChannelGraphDoubleDisabled(t *testing.T) {
 }
 
 func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) {
+	timestamp := time.Now()
+	nextTimeStamp := func() time.Time {
+		timestamp = timestamp.Add(time.Second)
+
+		return timestamp
+	}
+
 	// We'll create the following test graph so that only the last channel
 	// is pruned. We'll use a fresh timestamp to ensure they're not pruned
 	// according to that heuristic.
-	timestamp := time.Now()
 	testChannels := []*testChannel{
 		// Channel from self shouldn't be pruned.
 		symmetricTestChannel(
 			"self", "a", 100000, &testChannelPolicy{
-				LastUpdate: timestamp,
+				LastUpdate: nextTimeStamp(),
 				Disabled:   true,
 			}, 99,
 		),
@@ -918,7 +924,7 @@ func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) {
 			Node1: &testChannelEnd{
 				Alias: "a",
 				testChannelPolicy: &testChannelPolicy{
-					LastUpdate: timestamp,
+					LastUpdate: nextTimeStamp(),
 					Disabled:   true,
 				},
 			},
@@ -932,7 +938,7 @@ func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) {
 			Node1: &testChannelEnd{
 				Alias: "a",
 				testChannelPolicy: &testChannelPolicy{
-					LastUpdate: timestamp,
+					LastUpdate: nextTimeStamp(),
 					Disabled:   false,
 				},
 			},
@@ -946,14 +952,14 @@ func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) {
 			Node1: &testChannelEnd{
 				Alias: "a",
 				testChannelPolicy: &testChannelPolicy{
-					LastUpdate: timestamp,
+					LastUpdate: nextTimeStamp(),
 					Disabled:   true,
 				},
 			},
 			Node2: &testChannelEnd{
 				Alias: "b",
 				testChannelPolicy: &testChannelPolicy{
-					LastUpdate: timestamp,
+					LastUpdate: nextTimeStamp(),
 					Disabled:   false,
 				},
 			},
@@ -963,13 +969,13 @@ func testPruneChannelGraphDoubleDisabled(t *testing.T, assumeValid bool) {
 
 		// Both edges enabled.
 		symmetricTestChannel("c", "d", 100000, &testChannelPolicy{
-			LastUpdate: timestamp,
+			LastUpdate: nextTimeStamp(),
 			Disabled:   false,
 		}, 2),
 
 		// Both edges disabled, only one pruned.
 		symmetricTestChannel("e", "f", 100000, &testChannelPolicy{
-			LastUpdate: timestamp,
+			LastUpdate: nextTimeStamp(),
 			Disabled:   true,
 		}, 3),
 	}
@@ -1363,7 +1369,9 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
 	testAddrs = append(testAddrs, testAddr)
 
 	// Next, create a temporary graph database for usage within the test.
-	graph := graphdb.MakeTestGraph(t, graphdb.WithUseGraphCache(useCache))
+	graph := graphdb.MakeTestGraph(
+		t, graphdb.WithUseGraphCache(useCache),
+	)
 
 	aliasMap := make(map[string]route.Vertex)
 	privKeyMap := make(map[string]*btcec.PrivateKey)
@@ -1441,6 +1449,13 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
 			}
 
 			source = dbNode
+
+			// Set the selected source node
+			if err := graph.SetSourceNode(ctx, source); err != nil {
+				return nil, err
+			}
+
+			continue
 		}
 
 		// With the node fully parsed, add it as a vertex within the
@@ -1450,13 +1465,6 @@ func parseTestGraph(t *testing.T, useCache bool, path string) (
 		}
 	}
 
-	if source != nil {
-		// Set the selected source node
-		if err := graph.SetSourceNode(ctx, source); err != nil {
-			return nil, err
-		}
-	}
-
 	// With all the vertexes inserted, we can now insert the edges into the
 	// test graph.
 	for _, edge := range g.Edges {
@@ -1739,14 +1747,16 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 	testAddrs = append(testAddrs, testAddr)
 
 	// Next, create a temporary graph database for usage within the test.
-	graph := graphdb.MakeTestGraph(t, graphdb.WithUseGraphCache(useCache))
+	graph := graphdb.MakeTestGraph(
+		t, graphdb.WithUseGraphCache(useCache),
+	)
 
 	aliasMap := make(map[string]route.Vertex)
 	privKeyMap := make(map[string]*btcec.PrivateKey)
 
 	nodeIndex := byte(0)
-	addNodeWithAlias := func(alias string, features *lnwire.FeatureVector) (
-		*models.LightningNode, error) {
+	addNodeWithAlias := func(alias string,
+		features *lnwire.FeatureVector) error {
 
 		keyBytes := []byte{
 			0, 0, 0, 0, 0, 0, 0, 0,
@@ -1776,26 +1786,26 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 
 		// With the node fully parsed, add it as a vertex within the
 		// graph.
-		if err := graph.AddLightningNode(ctx, dbNode); err != nil {
-			return nil, err
+		if alias == source {
+			err = graph.SetSourceNode(ctx, dbNode)
+			require.NoError(t, err)
+		} else {
+			err := graph.AddLightningNode(ctx, dbNode)
+			require.NoError(t, err)
 		}
 
 		aliasMap[alias] = dbNode.PubKeyBytes
 		nodeIndex++
 
-		return dbNode, nil
+		return nil
 	}
 
 	// Add the source node.
-	dbNode, err := addNodeWithAlias(source, lnwire.EmptyFeatureVector())
+	err = addNodeWithAlias(source, lnwire.EmptyFeatureVector())
 	if err != nil {
 		return nil, err
 	}
 
-	if err = graph.SetSourceNode(ctx, dbNode); err != nil {
-		return nil, err
-	}
-
 	// Initialize variable that keeps track of the next channel id to assign
 	// if none is specified.
 	nextUnassignedChannelID := uint64(100000)
@@ -1813,7 +1823,7 @@ func createTestGraphFromChannels(t *testing.T, useCache bool,
 					features =
 						node.testChannelPolicy.Features
 				}
-				_, err := addNodeWithAlias(
+				err := addNodeWithAlias(
 					node.Alias, features,
 				)
 				if err != nil {