payments: kv-to-sql migration (tests + wiring)

config_builder.go

@@ -51,6 +51,8 @@ import (
 	"github.com/lightningnetwork/lnd/macaroons"
 	"github.com/lightningnetwork/lnd/msgmux"
 	paymentsdb "github.com/lightningnetwork/lnd/payments/db"
+	paymentsmig1 "github.com/lightningnetwork/lnd/payments/db/migration1"
+	paymentsmig1sqlc "github.com/lightningnetwork/lnd/payments/db/migration1/sqlc"
 	"github.com/lightningnetwork/lnd/rpcperms"
 	"github.com/lightningnetwork/lnd/signal"
 	"github.com/lightningnetwork/lnd/sqldb"
@@ -76,6 +78,10 @@ const (
 	// graphMigration is the version number for the graph migration
 	// that migrates the KV graph to the native SQL schema.
 	graphMigration = 10
+
+	// paymentMigration is the version number for the payments migration
+	// that migrates KV payments to the native SQL schema.
+	paymentMigration = 12
 )
 
 // GrpcRegistrar is an interface that must be satisfied by an external subserver
@@ -1153,6 +1159,34 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 				return nil
 			}
 
+			//nolint:ll
+			paymentMig := func(tx *sqlc.Queries) error {
+				err := paymentsmig1.MigratePaymentsKVToSQL(
+					ctx,
+					dbs.ChanStateDB.Backend,
+					paymentsmig1sqlc.New(tx.GetTx()),
+					&paymentsmig1.SQLStoreConfig{
+						QueryCfg: queryCfg,
+						SkipMigrationValidation: cfg.DB.
+							SkipPaymentsMigrationValidation,
+					},
+				)
+				if err != nil {
+					return fmt.Errorf("failed to migrate "+
+						"payments to SQL: %w", err)
+				}
+
+				// Set the payments bucket tombstone to
+				// indicate that the migration has been
+				// completed.
+				d.logger.Debugf("Setting payments bucket " +
+					"tombstone")
+
+				return paymentsdb.SetPaymentsBucketTombstone(
+					dbs.ChanStateDB.Backend,
+				)
+			}
+
 			// Make sure we attach the custom migration function to
 			// the correct migration version.
 			for i := 0; i < len(migrations); i++ {
@@ -1162,11 +1196,17 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 					migrations[i].MigrationFn = invoiceMig
 
 					continue
+
 				case graphMigration:
 					migrations[i].MigrationFn = graphMig
 
 					continue
 
+				case paymentMigration:
+					migrations[i].MigrationFn = paymentMig
+
+					continue
+
 				default:
 				}
 
@@ -1265,6 +1305,27 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 			return nil, nil, err
 		}
 
+		// Check if the payments bucket tombstone is set. If it is, we
+		// need to return and ask the user switch back to using the
+		// native SQL store.
+		ripPayments, err := paymentsdb.GetPaymentsBucketTombstone(
+			dbs.ChanStateDB.Backend,
+		)
+		if err != nil {
+			err = fmt.Errorf("unable to check payments bucket "+
+				"tombstone: %w", err)
+			d.logger.Error(err)
+
+			return nil, nil, err
+		}
+		if ripPayments {
+			err = fmt.Errorf("payments bucket tombstoned, please " +
+				"switch back to native SQL")
+			d.logger.Error(err)
+
+			return nil, nil, err
+		}
+
 		dbs.InvoiceDB = dbs.ChanStateDB
 
 		graphStore, err = graphdb.NewKVStore(

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

@@ -157,6 +157,10 @@
     db functions Part 2](https://github.com/lightningnetwork/lnd/pull/10308)
   * [Finalize SQL implementation for 
     payments db](https://github.com/lightningnetwork/lnd/pull/10373)
+  * [Add the KV-to-SQL payment
+    migration](https://github.com/lightningnetwork/lnd/pull/10485) with 
+    comprehensive tests and build tag "test_native_sql" gated wiring into the 
+    payment flow.
 
 
 ## Code Health

go.mod

@@ -141,7 +141,7 @@ require (
 	github.com/opencontainers/runc v1.1.14 // indirect
 	github.com/ory/dockertest/v3 v3.10.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.26.0 // indirect
 	github.com/prometheus/procfs v0.6.0 // indirect

lncfg/db.go

@@ -90,6 +90,8 @@ type DB struct {
 
 	SkipNativeSQLMigration bool `long:"skip-native-sql-migration" description:"If set to true, the KV to native SQL migration will be skipped. Note that this option is intended for users who experience non-resolvable migration errors. Enabling after there is a non-resolvable migration error that resulted in an incomplete migration will cause that partial migration to be abandoned and ignored and an empty database will be used instead. Since invoices are currently the only native SQL database used, our channels will still work but the invoice history will be forgotten. This option has no effect if native SQL is not in use (db.use-native-sql=false)."`
 
+	SkipPaymentsMigrationValidation bool `long:"skip-payments-migration-validation" description:"If set to true, the KV to native SQL payments migration will skip its in-migration validation step. This can be used if validation is too slow for large databases."`
+
 	NoGraphCache bool `long:"no-graph-cache" description:"Don't use the in-memory graph cache for path finding. Much slower but uses less RAM. Can only be used with a bolt database backend."`
 
 	PruneRevocation bool `long:"prune-revocation" description:"Run the optional migration that prunes the revocation logs to save disk space."`
@@ -130,8 +132,9 @@ func DefaultDB() *DB {
 			BusyTimeout:    defaultSqliteBusyTimeout,
 			QueryConfig:    *sqldb.DefaultSQLiteConfig(),
 		},
-		UseNativeSQL:           false,
-		SkipNativeSQLMigration: false,
+		UseNativeSQL:                    false,
+		SkipNativeSQLMigration:          false,
+		SkipPaymentsMigrationValidation: false,
 	}
 }
 

payments/db/kv_tombstone.go

@@ -0,0 +1,71 @@
+package paymentsdb
+
+import (
+	"fmt"
+
+	"github.com/lightningnetwork/lnd/kvdb"
+)
+
+var (
+	// paymentsBucketTombstone is the key used to mark the payments bucket
+	// as permanently closed after a successful migration.
+	paymentsBucketTombstone = []byte("payments-tombstone")
+)
+
+// SetPaymentsBucketTombstone sets the tombstone key in the payments bucket to
+// mark the bucket as permanently closed. This prevents it from being reopened
+// in the future.
+func SetPaymentsBucketTombstone(db kvdb.Backend) error {
+	return kvdb.Update(db, func(tx kvdb.RwTx) error {
+		// Access the top-level payments bucket.
+		payments := tx.ReadWriteBucket(paymentsRootBucket)
+
+		// In case the bucket doesn't exist, because we start
+		// immediately with the native SQL schema, we create it as well
+		// to make sure the user cannot switch back to the KV store.
+		if payments == nil {
+			var err error
+			payments, err = tx.CreateTopLevelBucket(
+				paymentsRootBucket,
+			)
+			if err != nil {
+				return fmt.Errorf("failed to create payments "+
+					"bucket: %w", err)
+			}
+		}
+
+		// Add the tombstone key to the payments bucket.
+		err := payments.Put(paymentsBucketTombstone, []byte("1"))
+		if err != nil {
+			return fmt.Errorf("failed to set tombstone: %w", err)
+		}
+
+		return nil
+	}, func() {})
+}
+
+// GetPaymentsBucketTombstone checks if the tombstone key exists in the payments
+// bucket. It returns true if the tombstone is present and false otherwise.
+func GetPaymentsBucketTombstone(db kvdb.Backend) (bool, error) {
+	var tombstoneExists bool
+
+	err := kvdb.View(db, func(tx kvdb.RTx) error {
+		// Access the top-level payments bucket.
+		payments := tx.ReadBucket(paymentsRootBucket)
+		if payments == nil {
+			tombstoneExists = false
+			return nil
+		}
+
+		// Check if the tombstone key exists.
+		tombstone := payments.Get(paymentsBucketTombstone)
+		tombstoneExists = tombstone != nil
+
+		return nil
+	}, func() {})
+	if err != nil {
+		return false, err
+	}
+
+	return tombstoneExists, nil
+}

payments/db/log.go

@@ -3,6 +3,7 @@ package paymentsdb
 import (
 	"github.com/btcsuite/btclog/v2"
 	"github.com/lightningnetwork/lnd/build"
+	paymentsmig1 "github.com/lightningnetwork/lnd/payments/db/migration1"
 )
 
 // log is a logger that is initialized with no output filters.  This
@@ -29,4 +30,5 @@ func DisableLog() {
 // using btclog.
 func UseLogger(logger btclog.Logger) {
 	log = logger
+	paymentsmig1.UseLogger(logger)
 }

payments/db/migration1/codec.go

@@ -0,0 +1,141 @@
+package migration1
+
+import (
+	"encoding/binary"
+	"errors"
+	"io"
+	"time"
+
+	"github.com/lightningnetwork/lnd/channeldb"
+)
+
+// Big endian is the preferred byte order, due to cursor scans over
+// integer keys iterating in order.
+var byteOrder = binary.BigEndian
+
+// UnknownElementType is an alias for channeldb.UnknownElementType.
+type UnknownElementType = channeldb.UnknownElementType
+
+// ReadElement deserializes a single element from the provided io.Reader.
+func ReadElement(r io.Reader, element interface{}) error {
+	err := channeldb.ReadElement(r, element)
+	switch {
+	// Known to channeldb codec.
+	case err == nil:
+		return nil
+
+	// Fail if error is not UnknownElementType.
+	default:
+		var unknownElementType UnknownElementType
+		if !errors.As(err, &unknownElementType) {
+			return err
+		}
+	}
+
+	// Process any paymentsdb-specific extensions to the codec.
+	switch e := element.(type) {
+	case *paymentIndexType:
+		if err := binary.Read(r, byteOrder, e); err != nil {
+			return err
+		}
+
+	// Type is still unknown to paymentsdb extensions, fail.
+	default:
+		return channeldb.NewUnknownElementType(
+			"ReadElement", element,
+		)
+	}
+
+	return nil
+}
+
+// WriteElement serializes a single element into the provided io.Writer.
+func WriteElement(w io.Writer, element interface{}) error {
+	err := channeldb.WriteElement(w, element)
+	switch {
+	// Known to channeldb codec.
+	case err == nil:
+		return nil
+
+	// Fail if error is not UnknownElementType.
+	default:
+		var unknownElementType UnknownElementType
+		if !errors.As(err, &unknownElementType) {
+			return err
+		}
+	}
+
+	// Process any paymentsdb-specific extensions to the codec.
+	switch e := element.(type) {
+	case paymentIndexType:
+		if err := binary.Write(w, byteOrder, e); err != nil {
+			return err
+		}
+
+	// Type is still unknown to paymentsdb extensions, fail.
+	default:
+		return channeldb.NewUnknownElementType(
+			"WriteElement", element,
+		)
+	}
+
+	return nil
+}
+
+// WriteElements serializes a variadic list of elements into the given
+// io.Writer.
+func WriteElements(w io.Writer, elements ...interface{}) error {
+	for _, element := range elements {
+		if err := WriteElement(w, element); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// ReadElements deserializes the provided io.Reader into a variadic list of
+// target elements.
+func ReadElements(r io.Reader, elements ...interface{}) error {
+	for _, element := range elements {
+		if err := ReadElement(r, element); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// deserializeTime deserializes time as unix nanoseconds.
+func deserializeTime(r io.Reader) (time.Time, error) {
+	var scratch [8]byte
+	if _, err := io.ReadFull(r, scratch[:]); err != nil {
+		return time.Time{}, err
+	}
+
+	// Convert to time.Time. Interpret unix nano time zero as a zero
+	// time.Time value.
+	unixNano := byteOrder.Uint64(scratch[:])
+	if unixNano == 0 {
+		return time.Time{}, nil
+	}
+
+	return time.Unix(0, int64(unixNano)), nil
+}
+
+// serializeTime serializes time as unix nanoseconds.
+func serializeTime(w io.Writer, t time.Time) error {
+	var scratch [8]byte
+
+	// Convert to unix nano seconds, but only if time is non-zero. Calling
+	// UnixNano() on a zero time yields an undefined result.
+	var unixNano int64
+	if !t.IsZero() {
+		unixNano = t.UnixNano()
+	}
+
+	byteOrder.PutUint64(scratch[:], uint64(unixNano))
+	_, err := w.Write(scratch[:])
+
+	return err
+}