migratekvdb: add Postgres bulk migration path

cmd_init_wallet.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcd/chaincfg/v2"
 	"github.com/btcsuite/btcwallet/wallet"
 	"github.com/jessevdk/go-flags"
 	"github.com/lightninglabs/protobuf-hex-display/jsonpb" // nolint

cmd_migrate_db.go

@@ -11,7 +11,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/btcsuite/btcd/btcutil"
+	"github.com/btcsuite/btcd/btcutil/v2"
 	"github.com/btcsuite/btclog/v2"
 	"github.com/jessevdk/go-flags"
 	"github.com/lightninglabs/lndinit/migratekvdb"
@@ -38,6 +38,21 @@ var (
 	defaultDataDir = filepath.Join(btcutil.AppDataDir("lnd", false), "data")
 )
 
+// destinationBackend keeps the regular KV backend together with the optional
+// migration-only bulk capability selected when opening the destination.
+type destinationBackend struct {
+	kvdb.Backend
+
+	bulk bulkMigrationRunner
+}
+
+// bulkMigrationRunner hides the build-tagged Postgres bulk migration APIs
+// from the command's build-independent wiring.
+type bulkMigrationRunner interface {
+	migrate(context.Context, *migratekvdb.Migrator, kvdb.Backend, bool) error
+	verify(context.Context, *migratekvdb.Migrator, kvdb.Backend) error
+}
+
 const (
 	// walletMetaBucket is the name of the meta bucket in the wallet db
 	// for the wallet ready marker.
@@ -99,6 +114,8 @@ type migrateDBCommand struct {
 	ForceNewMigration bool      `long:"force-new-migration" description:"Force a new migration from the beginning of the source DB so the resume state will be discarded"`
 	ForceVerifyDB     bool      `long:"force-verify-db" description:"Force a verification verifies two already marked (tombstoned and already migrated) dbs to make sure that the source db equals the content of the destination db"`
 	ChunkSize         uint64    `long:"chunk-size" description:"Chunk size for the migration in bytes"`
+	BulkWrites        bool      `long:"bulk-writes" description:"Enable the fresh-only Postgres bulk migration path. This uses lnd's SQL KV migration API. The bulk path does not resume mid-migration."`
+	ResetBulkTarget   bool      `long:"reset-bulk-target" description:"Allow recovery from an interrupted bulk migration to TRUNCATE a non-empty destination table, destroying all its rows, before restarting."`
 }
 
 func newMigrateDBCommand() *migrateDBCommand {
@@ -299,7 +316,9 @@ func (x *migrateDBCommand) Execute(_ []string) error {
 
 		// We open the destination DB as well to make sure both that
 		// both DBs are either marked or not.
-		destDb, err := openDestDb(ctx, x.Dest, prefix, x.Network)
+		destDb, err := openDestDb(
+			ctx, x.Dest, prefix, x.Network, x.BulkWrites,
+		)
 		if err != nil {
 			return fmt.Errorf("failed to open destination "+
 				"db with prefix `%s`: %w", prefix, err)
@@ -477,16 +496,33 @@ func (x *migrateDBCommand) Execute(_ []string) error {
 			return err
 		}
 
-		err = migrator.Migrate(ctx, srcDb, destDb)
+		// The destination backend was already validated to be postgres
+		// when --bulk-writes is set (see validateDBBackends), so we can
+		// use the flag directly here.
+		useBulk := x.BulkWrites
+
+		if useBulk {
+			err = destDb.bulk.migrate(
+				ctx, migrator, srcDb, x.ResetBulkTarget,
+			)
+		} else {
+			err = migrator.Migrate(ctx, srcDb, destDb)
+		}
 		if err != nil {
 			return err
 		}
 		logger.Infof("Migration of db with prefix %s completed", prefix)
 
 		// We migrated the DB successfully, now we verify the migration.
-		err = migrator.VerifyMigration(
-			ctx, srcDb, destDb, false,
-		)
+		if useBulk {
+			err = destDb.bulk.verify(
+				ctx, migrator, srcDb,
+			)
+		} else {
+			err = migrator.VerifyMigration(
+				ctx, srcDb, destDb, false,
+			)
+		}
 		if err != nil {
 			return err
 		}
@@ -582,6 +618,22 @@ func (x *migrateDBCommand) validateDBBackends() error {
 	// Destination must be sqlite or postgres.
 	switch x.Dest.Backend {
 	case lncfg.SqliteBackend, lncfg.PostgresBackend:
+		// The bulk migration path is only implemented for a postgres
+		// destination. Fail loudly if the operator explicitly requested
+		// it against a different backend rather than silently falling
+		// back to the standard path.
+		if x.BulkWrites && x.Dest.Backend != lncfg.PostgresBackend {
+			return fmt.Errorf("--bulk-writes requires a postgres "+
+				"destination, got: %s", x.Dest.Backend)
+		}
+		if x.ResetBulkTarget && !x.BulkWrites {
+			return fmt.Errorf("--reset-bulk-target requires --bulk-writes")
+		}
+		if x.ResetBulkTarget && x.ForceNewMigration {
+			return fmt.Errorf("--reset-bulk-target cannot be combined " +
+				"with --force-new-migration")
+		}
+
 		return nil
 	default:
 		return fmt.Errorf("destination database must be sqlite or "+
@@ -645,8 +697,8 @@ func openSourceDb(cfg *SourceDB, prefix, network string,
 }
 
 // openDestDb opens the different types of databases.
-func openDestDb(ctx context.Context, cfg *DestDB, prefix,
-	network string) (kvdb.Backend, error) {
+func openDestDb(ctx context.Context, cfg *DestDB, prefix, network string,
+	bulkWrites bool) (*destinationBackend, error) {
 
 	backend := cfg.Backend
 
@@ -661,19 +713,24 @@ func openDestDb(ctx context.Context, cfg *DestDB, prefix,
 
 	switch backend {
 	case kvdb.PostgresBackendName:
+		postgresCfg := &postgres.Config{
+			Dsn:            cfg.Postgres.Dsn,
+			Timeout:        time.Minute,
+			MaxConnections: 10,
+		}
+		logger.Infof("Opening postgres backend at `%s` with prefix `%s`",
+			cfg.Postgres.Dsn, prefix)
+
+		if bulkWrites {
+			return openPostgresBulkBackend(ctx, postgresCfg, prefix)
+		}
+
 		args = []interface{}{
 			ctx,
-			&postgres.Config{
-				Dsn:            cfg.Postgres.Dsn,
-				Timeout:        time.Minute,
-				MaxConnections: 10,
-			},
+			postgresCfg,
 			prefix,
 		}
 
-		logger.Infof("Opening postgres backend at `%s` with prefix `%s`",
-			cfg.Postgres.Dsn, prefix)
-
 	case kvdb.SqliteBackendName:
 		// Directories where the db files are located.
 		graphDir := lncfg.CleanAndExpandPath(
@@ -763,7 +820,12 @@ func openDestDb(ctx context.Context, cfg *DestDB, prefix,
 		return nil, fmt.Errorf("unknown backend: %v", backend)
 	}
 
-	return kvdb.Open(backend, args...)
+	db, err := kvdb.Open(backend, args...)
+	if err != nil {
+		return nil, err
+	}
+
+	return &destinationBackend{Backend: db}, nil
 }
 
 // checkMarkerPresent checks if a marker is present in the database.

cmd_migrate_db_bulk_postgres.go

@@ -0,0 +1,52 @@
+//go:build kvdb_postgres
+
+package main
+
+import (
+	"context"
+
+	"github.com/lightninglabs/lndinit/migratekvdb"
+	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/kvdb/postgres"
+	"github.com/lightningnetwork/lnd/kvdb/sqlbase"
+)
+
+// postgresBulkMigrationRunner contains the Postgres-only migration backend.
+// Keeping this type in a tagged file prevents its tagged lnd APIs from leaking
+// into the default build.
+type postgresBulkMigrationRunner struct {
+	target sqlbase.MigrationBackend
+}
+
+// migrate runs the Postgres-only bulk migration against the runner's target.
+func (p *postgresBulkMigrationRunner) migrate(ctx context.Context,
+	migrator *migratekvdb.Migrator, source kvdb.Backend,
+	resetTarget bool) error {
+
+	return migrator.MigrateBulk(ctx, source, p.target, resetTarget)
+}
+
+// verify compares the bulk-migrated target against the source database.
+func (p *postgresBulkMigrationRunner) verify(ctx context.Context,
+	migrator *migratekvdb.Migrator, source kvdb.Backend) error {
+
+	return migrator.VerifyMigrationBulk(ctx, source, p.target)
+}
+
+// openPostgresBulkBackend opens a Postgres destination with the migration-only
+// bulk capabilities enabled.
+func openPostgresBulkBackend(ctx context.Context, cfg *postgres.Config,
+	prefix string) (*destinationBackend, error) {
+
+	db, err := postgres.NewMigrationBackend(ctx, cfg, prefix)
+	if err != nil {
+		return nil, err
+	}
+
+	return &destinationBackend{
+		Backend: db,
+		bulk: &postgresBulkMigrationRunner{
+			target: db,
+		},
+	}, nil
+}

cmd_migrate_db_bulk_stub.go

@@ -0,0 +1,19 @@
+//go:build !kvdb_postgres
+
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/lightningnetwork/lnd/kvdb/postgres"
+)
+
+// openPostgresBulkBackend reports that bulk migration is unavailable in builds
+// without the kvdb_postgres tag.
+func openPostgresBulkBackend(context.Context, *postgres.Config,
+	string) (*destinationBackend, error) {
+
+	return nil, fmt.Errorf("postgres bulk migration support is not " +
+		"available; rebuild with -tags=kvdb_postgres")
+}

cmd_migrate_db_bulk_stub_test.go

@@ -0,0 +1,22 @@
+//go:build !kvdb_postgres
+
+package main
+
+import (
+	"context"
+	"testing"
+
+	"github.com/lightningnetwork/lnd/kvdb/postgres"
+	"github.com/stretchr/testify/require"
+)
+
+// TestOpenPostgresBulkBackendRequiresBuildTag verifies that a binary built
+// without Postgres support reports how to enable bulk migration.
+func TestOpenPostgresBulkBackendRequiresBuildTag(t *testing.T) {
+	t.Parallel()
+
+	_, err := openPostgresBulkBackend(
+		context.Background(), &postgres.Config{}, "test",
+	)
+	require.ErrorContains(t, err, "rebuild with -tags=kvdb_postgres")
+}