Merge pull request #10965 from ziggie1984/sqlbase-migration-bulk

kvdb: add Postgres migration bulk support

kvdb/postgres/db.go

@@ -16,12 +16,10 @@ var sqliteCmdReplacements = sqlbase.SQLiteCmdReplacements{
 	"INTEGER PRIMARY KEY": "BIGSERIAL PRIMARY KEY",
 }
 
-// newPostgresBackend returns a db object initialized with the passed backend
-// config. If postgres connection cannot be established, then returns error.
-func newPostgresBackend(ctx context.Context, config *Config, prefix string) (
-	walletdb.DB, error) {
-
-	cfg := &sqlbase.Config{
+// newSQLBaseConfig builds the shared sqlbase config used by both the regular
+// and migration Postgres backends from the passed backend config and prefix.
+func newSQLBaseConfig(config *Config, prefix string) *sqlbase.Config {
+	return &sqlbase.Config{
 		DriverName:            "pgx",
 		Dsn:                   config.Dsn,
 		Timeout:               config.Timeout,
@@ -30,6 +28,22 @@ func newPostgresBackend(ctx context.Context, config *Config, prefix string) (
 		SQLiteCmdReplacements: sqliteCmdReplacements,
 		WithTxLevelLock:       config.WithGlobalLock,
 	}
-
-	return sqlbase.NewSqlBackend(ctx, cfg)
+}
+
+// newPostgresBackend returns a db object initialized with the passed backend
+// config. If postgres connection cannot be established, then returns error.
+func newPostgresBackend(ctx context.Context, config *Config, prefix string) (
+	walletdb.DB, error) {
+
+	return sqlbase.NewSqlBackend(ctx, newSQLBaseConfig(config, prefix))
+}
+
+// NewMigrationBackend returns a Postgres backend that explicitly exposes the
+// migration-only bulk KV interface.
+func NewMigrationBackend(ctx context.Context, config *Config, prefix string) (
+	sqlbase.MigrationBackend, error) {
+
+	return sqlbase.NewPostgresBackend(
+		ctx, newSQLBaseConfig(config, prefix),
+	)
 }

kvdb/postgres/db_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/btcsuite/btcwallet/walletdb"
 	"github.com/btcsuite/btcwallet/walletdb/walletdbtest"
+	"github.com/lightningnetwork/lnd/kvdb/sqlbase"
 	"github.com/stretchr/testify/require"
 )
 
@@ -20,6 +21,11 @@ func TestInterface(t *testing.T) {
 	f, err := NewFixture("")
 	require.NoError(t, err)
 
+	// The regular Postgres backend must not expose migration-only
+	// capabilities. Callers must opt in through NewMigrationBackend.
+	_, ok := f.Db.(sqlbase.MigrationBulkKVStore)
+	require.False(t, ok)
+
 	// dbType is the database type name for this driver.
 	const dbType = "postgres"
 

kvdb/postgres/fixture.go

@@ -59,7 +59,33 @@ func StartEmbeddedPostgres() (func() error, error) {
 
 // NewFixture returns a new postgres test database. The database name is
 // randomly generated.
-func NewFixture(dbName string) (*fixture, error) {
+func NewFixture(dbName string) (*fixture[walletdb.DB], error) {
+	return newFixture(dbName, prefix, false, newPostgresBackend)
+}
+
+// NewMigrationFixture returns a new postgres test database that explicitly
+// exposes the migration-only bulk KV interface.
+func NewMigrationFixture(dbName string) (
+	*fixture[sqlbase.MigrationBackend], error) {
+
+	return newFixture(dbName, prefix, false, NewMigrationBackend)
+}
+
+// NewMigrationFixtureWithLock is like NewMigrationFixture but enables the
+// global tx-level lock so the lock-guarded migration paths are exercised.
+func NewMigrationFixtureWithLock(dbName string) (
+	*fixture[sqlbase.MigrationBackend], error) {
+
+	return newFixture(dbName, prefix, true, NewMigrationBackend)
+}
+
+// newFixture creates a new postgres test database using the passed backend
+// constructor, allowing callers to select the regular or migration backend and
+// whether the global tx-level lock is enabled.
+func newFixture[T walletdb.DB](dbName, tablePrefix string,
+	withGlobalLock bool, openBackend func(context.Context, *Config,
+		string) (T, error)) (*fixture[T], error) {
+
 	if dbName == "" {
 		// Create random database name.
 		randBytes := make([]byte, 8)
@@ -87,35 +113,36 @@ func NewFixture(dbName string) (*fixture, error) {
 
 	// Open database
 	dsn := getTestDsn(dbName)
-	db, err := newPostgresBackend(
+	db, err := openBackend(
 		context.Background(),
 		&Config{
-			Dsn:     dsn,
-			Timeout: time.Minute,
+			Dsn:            dsn,
+			Timeout:        time.Minute,
+			WithGlobalLock: withGlobalLock,
 		},
-		prefix,
+		tablePrefix,
 	)
 	if err != nil {
 		return nil, err
 	}
 
-	return &fixture{
+	return &fixture[T]{
 		Dsn: dsn,
 		Db:  db,
 	}, nil
 }
 
-type fixture struct {
+type fixture[T walletdb.DB] struct {
 	Dsn string
-	Db  walletdb.DB
+	Db  T
 }
 
-func (b *fixture) DB() walletdb.DB {
+func (b *fixture[T]) DB() walletdb.DB {
 	return b.Db
 }
 
 // Dump returns the raw contents of the database.
-func (b *fixture) Dump() (map[string]interface{}, error) {
+func (b *fixture[T]) Dump() (map[string]interface{}, error) {
 	dbConn, err := sql.Open("pgx", b.Dsn)
 	if err != nil {
 		return nil, err

kvdb/postgres/migration_bulk_test.go

@@ -0,0 +1,302 @@
+//go:build kvdb_postgres
+
+package postgres
+
+import (
+	"math"
+	"testing"
+
+	"github.com/btcsuite/btcwallet/walletdb"
+	"github.com/lightningnetwork/lnd/kvdb/sqlbase"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMigrationBulkKVStorePostgres verifies explicit migration capability
+// opt-in, bucket sequence preservation, leaf COPY semantics, verification,
+// transaction closure, and target truncation.
+func TestMigrationBulkKVStorePostgres(t *testing.T) {
+	stop, err := StartEmbeddedPostgres()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, stop())
+	}()
+
+	f, err := NewMigrationFixture("")
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, f.Db.Close())
+	}()
+
+	ctx := t.Context()
+	store := f.Db
+
+	empty, err := store.CheckEmpty(ctx)
+	require.NoError(t, err)
+	require.True(t, empty)
+
+	tx, err := store.BeginBulk(ctx)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, tx.Rollback())
+	}()
+
+	// Both nil and non-nil empty bucket keys must follow walletdb's key
+	// semantics without poisoning the bulk transaction.
+	for _, key := range [][]byte{nil, {}} {
+		_, err := tx.InsertBucket(ctx, nil, key, 0)
+		require.ErrorIs(t, err, walletdb.ErrBucketNameRequired)
+	}
+
+	rootID, err := tx.InsertBucket(
+		ctx, nil, []byte("root"), math.MaxUint64,
+	)
+	require.NoError(t, err)
+
+	maxIntPlusOne := uint64(math.MaxInt64) + 1
+	nestedID, err := tx.InsertBucket(
+		ctx, &rootID, []byte("nested"), maxIntPlusOne,
+	)
+	require.NoError(t, err)
+
+	// Leaves must belong to a previously inserted bucket. In particular,
+	// the zero-value parent must not be treated as a top-level leaf.
+	err = tx.InsertLeaves(ctx, []sqlbase.MigrationBulkLeaf{{
+		Key:   []byte("top-level"),
+		Value: []byte("unsupported"),
+	}})
+	require.EqualError(t, err, "bulk leaf 0 has invalid parent id 0")
+
+	// As with regular walletdb writes, nil and non-nil empty leaf keys are
+	// invalid. The indexed error identifies the bad entry in a batch.
+	for _, key := range [][]byte{nil, {}} {
+		err = tx.InsertLeaves(ctx, []sqlbase.MigrationBulkLeaf{{
+			ParentID: rootID,
+			Key:      key,
+			Value:    []byte("value"),
+		}})
+		require.ErrorIs(t, err, walletdb.ErrKeyRequired)
+		require.ErrorContains(t, err, "bulk leaf 0")
+	}
+
+	require.NoError(t, tx.InsertLeaves(ctx, []sqlbase.MigrationBulkLeaf{
+		{
+			ParentID: rootID,
+			Key:      []byte("a"),
+			Value:    []byte("value"),
+		},
+		{
+			ParentID: nestedID,
+			Key:      []byte("empty"),
+			Value:    []byte{},
+		},
+		{
+			ParentID: nestedID,
+			Key:      []byte("nil"),
+			Value:    nil,
+		},
+	}))
+
+	require.NoError(t, tx.Commit())
+
+	_, err = tx.InsertBucket(ctx, nil, []byte("closed"), 0)
+	require.ErrorIs(t, err, walletdb.ErrTxClosed)
+	require.ErrorIs(t, tx.InsertLeaves(ctx, nil), walletdb.ErrTxClosed)
+
+	empty, err = store.CheckEmpty(ctx)
+	require.NoError(t, err)
+	require.False(t, empty)
+
+	verifier, err := store.BeginBulkVerify(ctx)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, verifier.Rollback())
+	}()
+
+	top, err := verifier.FetchTopLevel(ctx)
+	require.NoError(t, err)
+	require.Len(t, top, 1)
+	require.Equal(t, rootID, top[0].ID)
+	require.Nil(t, top[0].ParentID)
+	require.Equal(t, []byte("root"), top[0].Key)
+	require.True(t, top[0].IsBucket)
+	require.Equal(t, uint64(math.MaxUint64), top[0].Sequence)
+
+	rootChildren, err := verifier.FetchChildren(ctx, []int64{rootID})
+	require.NoError(t, err)
+	require.Len(t, rootChildren, 2)
+
+	require.Equal(t, []byte("a"), rootChildren[0].Key)
+	require.False(t, rootChildren[0].IsBucket)
+	require.Equal(t, []byte("value"), rootChildren[0].Value)
+	require.NotNil(t, rootChildren[0].ParentID)
+	require.Equal(t, rootID, *rootChildren[0].ParentID)
+
+	require.Equal(t, []byte("nested"), rootChildren[1].Key)
+	require.True(t, rootChildren[1].IsBucket)
+	require.Equal(t, nestedID, rootChildren[1].ID)
+	require.Equal(t, maxIntPlusOne, rootChildren[1].Sequence)
+
+	nestedChildren, err := verifier.FetchChildren(ctx, []int64{nestedID})
+	require.NoError(t, err)
+	require.Len(t, nestedChildren, 2)
+	require.Equal(t, []byte("empty"), nestedChildren[0].Key)
+	require.False(t, nestedChildren[0].IsBucket)
+	require.NotNil(t, nestedChildren[0].Value)
+	require.Empty(t, nestedChildren[0].Value)
+
+	require.Equal(t, []byte("nil"), nestedChildren[1].Key)
+	require.False(t, nestedChildren[1].IsBucket)
+	require.NotNil(t, nestedChildren[1].Value)
+	require.Empty(t, nestedChildren[1].Value)
+
+	noChildren, err := verifier.FetchChildren(ctx, nil)
+	require.NoError(t, err)
+	require.Nil(t, noChildren)
+
+	require.NoError(t, verifier.Rollback())
+
+	_, err = verifier.FetchTopLevel(ctx)
+	require.ErrorIs(t, err, walletdb.ErrTxClosed)
+	_, err = verifier.FetchChildren(ctx, nil)
+	require.ErrorIs(t, err, walletdb.ErrTxClosed)
+
+	require.NoError(t, store.TruncateTargetTable(ctx))
+	empty, err = store.CheckEmpty(ctx)
+	require.NoError(t, err)
+	require.True(t, empty)
+}
+
+// TestMigrationBulkKVStoreRollbackPostgres verifies that rollback discards a
+// bulk load, remains idempotent, and closes the transaction to further writes.
+func TestMigrationBulkKVStoreRollbackPostgres(t *testing.T) {
+	stop, err := StartEmbeddedPostgres()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, stop())
+	}()
+
+	f, err := NewMigrationFixture("")
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, f.Db.Close())
+	}()
+
+	ctx := t.Context()
+	store := f.Db
+
+	tx, err := store.BeginBulk(ctx)
+	require.NoError(t, err)
+
+	rootID, err := tx.InsertBucket(ctx, nil, []byte("root"), 0)
+	require.NoError(t, err)
+	require.NoError(t, tx.InsertLeaves(ctx, []sqlbase.MigrationBulkLeaf{{
+		ParentID: rootID,
+		Key:      []byte("leaf"),
+		Value:    []byte("value"),
+	}}))
+
+	require.NoError(t, tx.Rollback())
+	require.NoError(t, tx.Rollback())
+
+	_, err = tx.InsertBucket(ctx, nil, []byte("closed"), 0)
+	require.ErrorIs(t, err, walletdb.ErrTxClosed)
+	require.ErrorIs(t, tx.InsertLeaves(ctx, nil), walletdb.ErrTxClosed)
+
+	empty, err := store.CheckEmpty(ctx)
+	require.NoError(t, err)
+	require.True(t, empty)
+}
+
+// TestMigrationBulkKVStoreMixedCasePrefixPostgres verifies that Postgres's
+// unquoted SQL paths and the quoted COPY identifier resolve the same table
+// when the configured prefix contains uppercase characters.
+func TestMigrationBulkKVStoreMixedCasePrefixPostgres(t *testing.T) {
+	stop, err := StartEmbeddedPostgres()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, stop())
+	}()
+
+	f, err := newFixture(
+		"", "MixedCase", false, NewMigrationBackend,
+	)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, f.Db.Close())
+	}()
+
+	ctx := t.Context()
+	tx, err := f.Db.BeginBulk(ctx)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, tx.Rollback())
+	}()
+
+	rootID, err := tx.InsertBucket(ctx, nil, []byte("root"), 0)
+	require.NoError(t, err)
+	require.NoError(t, tx.InsertLeaves(ctx, []sqlbase.MigrationBulkLeaf{{
+		ParentID: rootID,
+		Key:      []byte("leaf"),
+		Value:    []byte("value"),
+	}}))
+	require.NoError(t, tx.Commit())
+
+	empty, err := f.Db.CheckEmpty(ctx)
+	require.NoError(t, err)
+	require.False(t, empty)
+}
+
+// TestMigrationBulkKVStoreGlobalLockPostgres runs a full bulk load and
+// verification cycle with the global tx-level lock enabled to exercise the
+// lock-guarded write and read paths without deadlocking.
+func TestMigrationBulkKVStoreGlobalLockPostgres(t *testing.T) {
+	stop, err := StartEmbeddedPostgres()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, stop())
+	}()
+
+	f, err := NewMigrationFixtureWithLock("")
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, f.Db.Close())
+	}()
+
+	ctx := t.Context()
+	store := f.Db
+
+	// Write path: BeginBulk takes the exclusive lock and Commit releases
+	// it.
+	tx, err := store.BeginBulk(ctx)
+	require.NoError(t, err)
+
+	rootID, err := tx.InsertBucket(ctx, nil, []byte("root"), 0)
+	require.NoError(t, err)
+	require.NoError(t, tx.InsertLeaves(ctx, []sqlbase.MigrationBulkLeaf{{
+		ParentID: rootID,
+		Key:      []byte("leaf"),
+		Value:    []byte("value"),
+	}}))
+	require.NoError(t, tx.Commit())
+
+	// Read path: CheckEmpty and the verifier take the shared read lock.
+	empty, err := store.CheckEmpty(ctx)
+	require.NoError(t, err)
+	require.False(t, empty)
+
+	verifier, err := store.BeginBulkVerify(ctx)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, verifier.Rollback())
+	}()
+
+	top, err := verifier.FetchTopLevel(ctx)
+	require.NoError(t, err)
+	require.Len(t, top, 1)
+	require.Equal(t, rootID, top[0].ID)
+
+	children, err := verifier.FetchChildren(ctx, []int64{rootID})
+	require.NoError(t, err)
+	require.Len(t, children, 1)
+	require.Equal(t, []byte("value"), children[0].Value)
+}

kvdb/sqlbase/migration_bulk.go

@@ -0,0 +1,109 @@
+//go:build kvdb_postgres || (kvdb_sqlite && !(windows && (arm || 386)) && !(linux && (ppc64 || mips || mipsle || mips64)))
+
+package sqlbase
+
+import (
+	"context"
+
+	"github.com/btcsuite/btcwallet/walletdb"
+)
+
+// MigrationBackend combines the regular walletdb database operations with the
+// migration-only bulk capabilities guaranteed by a migration backend.
+type MigrationBackend interface {
+	walletdb.DB
+	MigrationBulkKVStore
+}
+
+// MigrationBulkKVStore exposes migration-only helpers for loading and verifying
+// the SQL KV schema directly. Normal application code should continue to use
+// the walletdb/kvdb bucket APIs.
+type MigrationBulkKVStore interface {
+	// CheckEmpty returns whether the underlying KV table has no rows.
+	CheckEmpty(ctx context.Context) (bool, error)
+
+	// TruncateTargetTable unconditionally and irreversibly removes every
+	// row from the underlying KV table. It is only intended for fresh-only
+	// migration recovery when the caller owns the whole target table.
+	TruncateTargetTable(ctx context.Context) error
+
+	// BeginBulk opens a destination write transaction for bulk loading.
+	// Callers must defer Rollback immediately after a successful open; the
+	// rollback is a no-op after Commit and releases locks/connections on
+	// all other exits.
+	BeginBulk(ctx context.Context) (MigrationBulkKVTx, error)
+
+	// BeginBulkVerify opens a read transaction for batched verification.
+	// Callers must defer Rollback immediately after a successful open so
+	// the read transaction lock is always released.
+	BeginBulkVerify(ctx context.Context) (MigrationBulkKVVerifier, error)
+}
+
+// MigrationBulkLeaf is a leaf key/value row to be inserted under ParentID.
+// ParentID must be a positive row ID returned by InsertBucket. Top-level leaves
+// are not supported by the migration bulk API.
+type MigrationBulkLeaf struct {
+	ParentID int64
+
+	// Key must be non-empty.
+	Key   []byte
+	Value []byte
+}
+
+// MigrationBulkKVTx is a migration-only transaction for bulk-loading SQL KV
+// data.
+type MigrationBulkKVTx interface {
+	// InsertBucket inserts a bucket row with a non-empty key. It returns
+	// the generated id. A nil parentID creates a top-level bucket.
+	InsertBucket(ctx context.Context, parentID *int64, key []byte,
+		seq uint64) (int64, error)
+
+	// InsertLeaves inserts leaf rows with non-empty keys. Implementations
+	// may choose COPY, multi-row INSERT, or another backend-specific
+	// strategy.
+	InsertLeaves(ctx context.Context, leaves []MigrationBulkLeaf) error
+
+	// Commit atomically commits the bulk load transaction.
+	Commit() error
+
+	// Rollback aborts the bulk load transaction.
+	Rollback() error
+}
+
+// MigrationBulkChild is a single SQL KV row returned by the verifier helpers.
+type MigrationBulkChild struct {
+	// ID is the SQL row id.
+	ID int64
+
+	// ParentID is nil for top-level rows.
+	ParentID *int64
+
+	// Key is the bucket key or leaf key.
+	Key []byte
+
+	// Value is the leaf value. It is nil for buckets.
+	Value []byte
+
+	// IsBucket identifies bucket rows explicitly. This avoids ambiguity
+	// between SQL NULL bucket markers and empty leaf values decoded as nil.
+	IsBucket bool
+
+	// Sequence is the bucket sequence number. It is zero for unset
+	// sequences and for leaf rows.
+	Sequence uint64
+}
+
+// MigrationBulkKVVerifier reads SQL KV rows in batches for migration
+// verification.
+type MigrationBulkKVVerifier interface {
+	// FetchTopLevel returns all top-level rows ordered by key.
+	FetchTopLevel(ctx context.Context) ([]MigrationBulkChild, error)
+
+	// FetchChildren returns direct children for the given bucket ids,
+	// ordered by parent id and key.
+	FetchChildren(ctx context.Context,
+		parentIDs []int64) ([]MigrationBulkChild, error)
+
+	// Rollback closes the verifier transaction.
+	Rollback() error
+}

kvdb/sqlbase/migration_bulk_postgres.go

@@ -0,0 +1,428 @@
+//go:build kvdb_postgres
+
+package sqlbase
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/btcsuite/btcwallet/walletdb"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/stdlib"
+)
+
+// postgresDB adds Postgres-only capabilities to the shared SQL backend.
+type postgresDB struct {
+	*db
+}
+
+var (
+	_ walletdb.DB          = (*postgresDB)(nil)
+	_ MigrationBulkKVStore = (*postgresDB)(nil)
+)
+
+// bulkLeafCols is the leaf-row projection of the shared KV table schema
+// defined in schema.go. The id column is database-generated. Sequence is
+// walletdb bucket metadata copied separately by InsertBucket, so leaf rows do
+// not include either column in the COPY operation.
+var bulkLeafCols = []string{"parent_id", "key", "value"}
+
+// NewPostgresBackend returns a shared SQL backend with Postgres-only
+// capabilities, including migration bulk loading.
+func NewPostgresBackend(ctx context.Context, cfg *Config) (
+	MigrationBackend, error) {
+
+	db, err := NewSqlBackend(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &postgresDB{db: db}, nil
+}
+
+// CheckEmpty returns whether the underlying KV table has no rows.
+func (p *postgresDB) CheckEmpty(ctx context.Context) (bool, error) {
+	locker := p.bulkLocker(true)
+	locker.Lock()
+	defer locker.Unlock()
+
+	var count int64
+	err := p.db.db.QueryRowContext(
+		ctx, "SELECT COUNT(*) FROM "+p.table,
+	).Scan(&count)
+	if err != nil {
+		return false, err
+	}
+
+	return count == 0, nil
+}
+
+// TruncateTargetTable unconditionally and irreversibly removes every row from
+// the underlying KV table. It is only intended for fresh migration recovery
+// where the caller owns the whole target table.
+func (p *postgresDB) TruncateTargetTable(ctx context.Context) error {
+	locker := p.bulkLocker(false)
+	locker.Lock()
+	defer locker.Unlock()
+
+	_, err := p.db.db.ExecContext(ctx, "TRUNCATE TABLE "+p.table)
+
+	return err
+}
+
+// BeginBulk opens a write transaction for bulk loading. It uses a dedicated
+// *sql.Conn so InsertLeaves can reach the underlying pgx connection and COPY
+// into the same transaction. Callers must defer Rollback immediately after a
+// successful open so the lock and connection are released on all exits.
+func (p *postgresDB) BeginBulk(ctx context.Context) (MigrationBulkKVTx, error) {
+	locker := p.bulkLocker(false)
+	locker.Lock()
+
+	conn, err := p.db.db.Conn(ctx)
+	if err != nil {
+		locker.Unlock()
+		return nil, err
+	}
+
+	tx, err := conn.BeginTx(ctx, &sql.TxOptions{
+		Isolation: sql.LevelSerializable,
+	})
+	if err != nil {
+		locker.Unlock()
+		_ = conn.Close()
+		return nil, err
+	}
+
+	return &postgresBulkKVTx{
+		db:     p.db,
+		conn:   conn,
+		tx:     tx,
+		locker: locker,
+		active: true,
+	}, nil
+}
+
+// BeginBulkVerify opens a read-only transaction for batched verification.
+// Callers must defer Rollback immediately after a successful open so the read
+// transaction lock is always released.
+func (p *postgresDB) BeginBulkVerify(
+	ctx context.Context) (MigrationBulkKVVerifier, error) {
+
+	locker := p.bulkLocker(true)
+	locker.Lock()
+
+	tx, err := p.db.db.BeginTx(ctx, &sql.TxOptions{
+		ReadOnly:  true,
+		Isolation: sql.LevelRepeatableRead,
+	})
+	if err != nil {
+		locker.Unlock()
+		return nil, err
+	}
+
+	return &postgresBulkKVVerifier{
+		db:     p.db,
+		tx:     tx,
+		locker: locker,
+		active: true,
+	}, nil
+}
+
+// bulkLocker returns the same optional global lock used by regular sqlbase
+// transactions so migration-only transactions respect WithTxLevelLock.
+func (p *postgresDB) bulkLocker(readOnly bool) sync.Locker {
+	if !p.cfg.WithTxLevelLock {
+		return newNoopLocker()
+	}
+	if readOnly {
+		return p.lock.RLocker()
+	}
+
+	return &p.lock
+}
+
+// postgresBulkKVTx is a migration-only Postgres transaction for loading the SQL
+// KV table directly.
+type postgresBulkKVTx struct {
+	db     *db
+	conn   *sql.Conn
+	tx     *sql.Tx
+	locker sync.Locker
+	active bool
+}
+
+// InsertBucket inserts a bucket row and returns its generated id.
+func (p *postgresBulkKVTx) InsertBucket(ctx context.Context,
+	parentID *int64, key []byte, seq uint64) (int64, error) {
+
+	if !p.active {
+		return 0, walletdb.ErrTxClosed
+	}
+	if len(key) == 0 {
+		return 0, walletdb.ErrBucketNameRequired
+	}
+
+	keyCopy := cloneBulkBytes(key)
+
+	var id int64
+	if seq != 0 {
+		err := p.tx.QueryRowContext(
+			ctx, "INSERT INTO "+p.db.table+
+				" (parent_id, key, sequence) "+
+				"VALUES ($1,$2,$3) RETURNING id",
+			parentID, keyCopy, int64(seq),
+		).Scan(&id)
+
+		return id, err
+	}
+
+	err := p.tx.QueryRowContext(
+		ctx, "INSERT INTO "+p.db.table+" (parent_id, key) "+
+			"VALUES ($1,$2) RETURNING id",
+		parentID, keyCopy,
+	).Scan(&id)
+
+	return id, err
+}
+
+// InsertLeaves inserts leaf rows with Postgres COPY.
+func (p *postgresBulkKVTx) InsertLeaves(ctx context.Context,
+	leaves []MigrationBulkLeaf) error {
+
+	if !p.active {
+		return walletdb.ErrTxClosed
+	}
+	if len(leaves) == 0 {
+		return nil
+	}
+
+	rows := make([][]any, len(leaves))
+	for i := range leaves {
+		if leaves[i].ParentID <= 0 {
+			return fmt.Errorf(
+				"bulk leaf %d has invalid parent id %d", i,
+				leaves[i].ParentID,
+			)
+		}
+		if len(leaves[i].Key) == 0 {
+			return fmt.Errorf(
+				"bulk leaf %d: %w", i, walletdb.ErrKeyRequired,
+			)
+		}
+
+		value := cloneBulkBytes(leaves[i].Value)
+		if value == nil {
+			value = []byte{}
+		}
+
+		rows[i] = []any{
+			leaves[i].ParentID,
+			cloneBulkBytes(leaves[i].Key),
+			value,
+		}
+	}
+
+	var copied int64
+	err := p.conn.Raw(func(driverConn any) error {
+		pgxConn, ok := driverConn.(*stdlib.Conn)
+		if !ok {
+			return fmt.Errorf("driver conn is %T, not "+
+				"pgx/v5/stdlib.Conn", driverConn)
+		}
+
+		var copyErr error
+		// The shared schema and normal SQL paths use unquoted
+		// identifiers, which Postgres folds to lowercase. CopyFrom
+		// quotes its identifier, so fold it explicitly to resolve the
+		// same physical table.
+		copied, copyErr = pgxConn.Conn().CopyFrom(
+			ctx, pgx.Identifier{strings.ToLower(p.db.table)},
+			bulkLeafCols,
+			pgx.CopyFromRows(rows),
+		)
+
+		return copyErr
+	})
+	if err != nil {
+		return err
+	}
+	if copied != int64(len(leaves)) {
+		return fmt.Errorf("bulk leaf copy count mismatch: got=%d "+
+			"want=%d", copied, len(leaves))
+	}
+
+	return nil
+}
+
+// Commit commits the bulk transaction and releases its dedicated connection.
+func (p *postgresBulkKVTx) Commit() error {
+	if !p.active {
+		return walletdb.ErrTxClosed
+	}
+
+	err := p.tx.Commit()
+	p.active = false
+	p.locker.Unlock()
+	closeErr := p.conn.Close()
+	if err != nil {
+		if closeErr != nil {
+			log.Warnf(
+				"Could not close bulk migration connection: %v",
+				closeErr,
+			)
+		}
+
+		return err
+	}
+	if closeErr != nil {
+		log.Warnf("Could not close bulk migration connection after "+
+			"commit: %v", closeErr)
+	}
+
+	return nil
+}
+
+// Rollback rolls back the bulk transaction and releases its dedicated
+// connection. It is idempotent for already-closed transactions.
+func (p *postgresBulkKVTx) Rollback() error {
+	if !p.active {
+		return nil
+	}
+
+	err := p.tx.Rollback()
+	p.active = false
+	p.locker.Unlock()
+	closeErr := p.conn.Close()
+	if err != nil && !errors.Is(err, sql.ErrTxDone) {
+		return err
+	}
+
+	return closeErr
+}
+
+// postgresBulkKVVerifier is a read-only Postgres transaction for batched SQL KV
+// verification.
+type postgresBulkKVVerifier struct {
+	db     *db
+	tx     *sql.Tx
+	locker sync.Locker
+	active bool
+}
+
+// FetchTopLevel returns all top-level rows ordered by key.
+func (p *postgresBulkKVVerifier) FetchTopLevel(
+	ctx context.Context) ([]MigrationBulkChild, error) {
+
+	if !p.active {
+		return nil, walletdb.ErrTxClosed
+	}
+
+	// The table name is constructed internally from the configured prefix.
+	//nolint:gosec
+	rows, err := p.tx.QueryContext(ctx, "SELECT id, parent_id, key, "+
+		"value, sequence, CASE WHEN value IS NULL THEN 1 ELSE 0 END "+
+		"FROM "+p.db.table+" WHERE parent_id IS NULL ORDER BY key")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	return scanBulkChildren(rows)
+}
+
+// FetchChildren returns direct children for parentIDs ordered by parent id and
+// key.
+func (p *postgresBulkKVVerifier) FetchChildren(ctx context.Context,
+	parentIDs []int64) ([]MigrationBulkChild, error) {
+
+	if !p.active {
+		return nil, walletdb.ErrTxClosed
+	}
+	if len(parentIDs) == 0 {
+		return nil, nil
+	}
+
+	// parentIDs is passed as a native []int64; the pgx stdlib driver
+	// encodes it as a Postgres bigint array for the ANY($1) match.
+	//
+	// The table name is constructed internally from the configured prefix.
+	//nolint:gosec
+	rows, err := p.tx.QueryContext(ctx, "SELECT id, parent_id, key, "+
+		"value, sequence, CASE WHEN value IS NULL THEN 1 ELSE 0 END "+
+		"FROM "+p.db.table+" WHERE parent_id = ANY($1) "+
+		"ORDER BY parent_id, key", parentIDs)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	return scanBulkChildren(rows)
+}
+
+// Rollback closes the verifier read transaction.
+func (p *postgresBulkKVVerifier) Rollback() error {
+	if !p.active {
+		return nil
+	}
+
+	err := p.tx.Rollback()
+	p.active = false
+	p.locker.Unlock()
+	if err != nil && !errors.Is(err, sql.ErrTxDone) {
+		return err
+	}
+
+	return nil
+}
+
+// scanBulkChildren scans verifier rows and preserves an explicit IsBucket flag
+// so empty leaf values are not confused with SQL NULL bucket markers.
+func scanBulkChildren(rows *sql.Rows) ([]MigrationBulkChild, error) {
+	var children []MigrationBulkChild
+	for rows.Next() {
+		var (
+			child      MigrationBulkChild
+			parentID   sql.NullInt64
+			sequence   sql.NullInt64
+			bucketFlag int
+		)
+		if err := rows.Scan(
+			&child.ID, &parentID, &child.Key, &child.Value,
+			&sequence, &bucketFlag,
+		); err != nil {
+			return nil, err
+		}
+		if parentID.Valid {
+			id := parentID.Int64
+			child.ParentID = &id
+		}
+		if sequence.Valid {
+			child.Sequence = uint64(sequence.Int64)
+		}
+		child.IsBucket = bucketFlag == 1
+		if !child.IsBucket && child.Value == nil {
+			child.Value = []byte{}
+		}
+
+		children = append(children, child)
+	}
+
+	return children, rows.Err()
+}
+
+// cloneBulkBytes copies driver-owned byte slices before they are buffered or
+// returned to callers.
+func cloneBulkBytes(b []byte) []byte {
+	if b == nil {
+		return nil
+	}
+
+	out := make([]byte, len(b))
+	copy(out, b)
+
+	return out
+}

kvdb/sqlite/db_test.go

@@ -26,6 +26,11 @@ func TestInterface(t *testing.T) {
 	sqlDB, err := NewSqliteBackend(ctx, cfg, dir, "tmp.db", "table")
 	require.NoError(t, err)
 
+	// The regular SQLite backend must not expose migration-only
+	// capabilities. Migration backends must be explicitly selected.
+	_, ok := sqlDB.(sqlbase.MigrationBulkKVStore)
+	require.False(t, ok)
+
 	t.Cleanup(func() {
 		require.NoError(t, sqlDB.Close())
 	})