payments/migration1: add the payments mig code
What changed, and why it matters
This commit adds new code that migrates old-style payment records into a new SQL database format and checks that the migrated data matches the original. It is a data-integrity migration, not a typical security patch. The code does not obviously introduce remote attacks, but it changes how sensitive payment history is stored and interpreted. Any bug here could corrupt payment state, misclassify failed payments as successful (or vice versa), or lose duplicate-payment records. The commit itself does not claim to fix a security vulnerability.
Treat this as a high-risk migration feature rather than an active exploit fix. Reviewers should focus on correctness of state mapping (especially duplicate-payment terminal-state logic), SQL injection safety in generated sqlc queries, transaction atomicity guarantees, and rollback behavior if validation fails mid-migration. End users should ensure migrations run on backed-up nodes and monitor logs for validation warnings.
Security signals we found
New migration code handling payment state transitions and duplicate-payment resolution
Default-fail classification for duplicate payments lacking settle/fail data
Validation pass intended to detect data corruption or migration bugs
Direct SQL inserts of payment hashes, preimages, failure messages, and route details
No explicit security claim or CVE reference in commit message
Evidence from the diff
The commit introduces two new files implementing the KV-to-SQL payment migration for LND: sql_migration.go performs the migration, and migration_validation.go performs a structural and deep-comparison validation pass. It migrates payments, HTLC attempts, route hops, intents, custom records, MPP/AMP/blinded-route data, and duplicate payments. Duplicate payments without settle/fail info are marked failed with FailureReasonError to ensure a terminal state. Validation compares KV and SQL representations in batches and reports diffs on mismatch. The code is defensive (skips missing buckets, validates counts, logs inconsistencies) but is brand-new migration logic that has not yet been battle-tested.
Changed components
payments/db/migration1/sql_migration.gopayments/db/migration1/migration_validation.goLND payment database migration path (KV to SQL)Inspect captured patch +1372 / −0
diff --git a/payments/db/migration1/migration_validation.go b/payments/db/migration1/migration_validation.go
new file mode 100644
index 0000000..e97c3db
--- /dev/null
+++ b/payments/db/migration1/migration_validation.go
@@ -0,0 +1,555 @@
+package migration1
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "fmt"
+ "reflect"
+ "sort"
+ "time"
+
+ "github.com/davecgh/go-spew/spew"
+ "github.com/lightningnetwork/lnd/kvdb"
+ "github.com/lightningnetwork/lnd/lntypes"
+ "github.com/lightningnetwork/lnd/lnwire"
+ "github.com/lightningnetwork/lnd/payments/db/migration1/sqlc"
+ "github.com/lightningnetwork/lnd/record"
+ "github.com/pmezard/go-difflib/difflib"
+)
+
+// migratedPaymentRef is a reference to a migrated payment.
+type migratedPaymentRef struct {
+ Hash lntypes.Hash
+ PaymentID int64
+}
+
+// validateMigratedPaymentBatch performs a structural validation pass by
+// comparing key fields (hash, amount, fail reason, HTLC count) of KV payments
+// with their SQL counterparts. If a structural mismatch is detected, a full
+// deep comparison is performed to produce a detailed diff for debugging.
+func validateMigratedPaymentBatch(ctx context.Context,
+ kvBackend kvdb.Backend, sqlDB SQLQueries,
+ cfg *SQLStoreConfig, batch []migratedPaymentRef) error {
+
+ if len(batch) == 0 {
+ return nil
+ }
+
+ if cfg == nil || cfg.QueryCfg == nil {
+ return fmt.Errorf("missing SQL store config for validation")
+ }
+
+ paymentIDs := make([]int64, 0, len(batch))
+ for _, item := range batch {
+ paymentIDs = append(paymentIDs, item.PaymentID)
+ }
+
+ rows, err := sqlDB.FetchPaymentsByIDsMig(ctx, paymentIDs)
+ if err != nil {
+ return fmt.Errorf("fetch SQL payments: %w", err)
+ }
+ if len(rows) != len(paymentIDs) {
+ return fmt.Errorf("SQL payment batch mismatch: got=%d want=%d",
+ len(rows), len(paymentIDs))
+ }
+
+ // Perform the structural check by comparing key fields from the KV
+ // store with the SQL store.
+ err = kvBackend.View(func(kvTx kvdb.RTx) error {
+ paymentsBucket := kvTx.ReadBucket(paymentsRootBucket)
+ if paymentsBucket == nil {
+ return fmt.Errorf("no payments bucket")
+ }
+
+ for _, row := range rows {
+ hash := row.PaymentIdentifier
+ var paymentHash lntypes.Hash
+ copy(paymentHash[:], hash)
+
+ paymentBucket := paymentsBucket.NestedReadBucket(hash)
+ if paymentBucket == nil {
+ return fmt.Errorf("missing payment bucket %x",
+ hash[:8])
+ }
+
+ kvPayment, err := fetchPayment(paymentBucket)
+ if err != nil {
+ return fmt.Errorf("fetch KV payment %x: %w",
+ hash[:8], err)
+ }
+
+ err = structuralCompare(kvPayment, row)
+ if err != nil {
+ // On structural mismatch, perform a deep
+ // comparison to produce a detailed diff.
+ deepErr := deepComparePayment(
+ ctx, cfg, sqlDB, row.ID,
+ paymentHash, kvPayment,
+ )
+ if deepErr != nil {
+ return deepErr
+ }
+
+ // If deep comparison passes but structural
+ // failed, report the structural error as it
+ // indicates an unexpected inconsistency.
+ return err
+ }
+
+ err = compareDuplicatePayments(
+ ctx, paymentBucket, sqlDB, row.ID,
+ paymentHash,
+ )
+ if err != nil {
+ return err
+ }
+ }
+
+ return nil
+ }, func() {})
+ if err != nil {
+ return err
+ }
+
+ return nil
+}
+
+// structuralCompare performs a fast structural comparison between a KV payment
+// and a SQL payment row, checking key fields: payment identifier, amount,
+// failure reason, and HTLC attempt count.
+func structuralCompare(kvPayment *MPPayment,
+ sqlRow sqlc.FetchPaymentsByIDsMigRow) error {
+
+ // Compare payment identifier.
+ kvHash := kvPayment.Info.PaymentIdentifier[:]
+ if !bytes.Equal(kvHash, sqlRow.PaymentIdentifier) {
+ return fmt.Errorf("payment identifier mismatch: kv=%x sql=%x",
+ kvHash[:8], sqlRow.PaymentIdentifier[:8])
+ }
+
+ // Compare amount.
+ kvAmount := int64(kvPayment.Info.Value)
+ if kvAmount != sqlRow.AmountMsat {
+ return fmt.Errorf("amount mismatch for %x: kv=%d sql=%d",
+ sqlRow.PaymentIdentifier[:8], kvAmount,
+ sqlRow.AmountMsat)
+ }
+
+ // Compare failure reason.
+ var kvFailReason sql.NullInt32
+ if kvPayment.FailureReason != nil {
+ kvFailReason = sql.NullInt32{
+ Int32: int32(*kvPayment.FailureReason),
+ Valid: true,
+ }
+ }
+ if kvFailReason != sqlRow.FailReason {
+ return fmt.Errorf("fail reason mismatch for %x: kv=%v sql=%v",
+ sqlRow.PaymentIdentifier[:8], kvFailReason,
+ sqlRow.FailReason)
+ }
+
+ // Compare HTLC attempt count.
+ kvHTLCCount := int64(len(kvPayment.HTLCs))
+ if kvHTLCCount != sqlRow.HtlcAttemptCount {
+ return fmt.Errorf("HTLC count mismatch for %x: kv=%d sql=%d",
+ sqlRow.PaymentIdentifier[:8], kvHTLCCount,
+ sqlRow.HtlcAttemptCount)
+ }
+
+ return nil
+}
+
+// deepComparePayment performs a full deep comparison between a KV payment and
+// its SQL counterpart, producing a detailed diff on mismatch.
+func deepComparePayment(ctx context.Context, cfg *SQLStoreConfig,
+ sqlDB SQLQueries, paymentID int64, paymentHash lntypes.Hash,
+ kvPayment *MPPayment) error {
+
+ batchData, err := batchLoadPaymentDetailsData(
+ ctx, cfg.QueryCfg, sqlDB, []int64{paymentID},
+ )
+ if err != nil {
+ return fmt.Errorf("load payment data for deep compare %x: %w",
+ paymentHash[:8], err)
+ }
+
+ byIDRows, err := sqlDB.FetchPaymentsByIDs(ctx, []int64{paymentID})
+ if err != nil {
+ return fmt.Errorf("fetch payment by ID for deep compare "+
+ "%x: %w", paymentHash[:8], err)
+ }
+ if len(byIDRows) != 1 {
+ return fmt.Errorf("expected 1 payment for deep compare, "+
+ "got %d", len(byIDRows))
+ }
+
+ sqlPayment, err := buildPaymentFromBatchData(byIDRows[0], batchData)
+ if err != nil {
+ return fmt.Errorf("build SQL payment %x: %w",
+ paymentHash[:8], err)
+ }
+
+ normalizePaymentForCompare(kvPayment)
+ normalizePaymentForCompare(sqlPayment)
+
+ if !reflect.DeepEqual(kvPayment, sqlPayment) {
+ dumpCfg := spew.ConfigState{
+ DisablePointerAddresses: true,
+ DisableCapacities: true,
+ DisableMethods: true,
+ SortKeys: true,
+ }
+ diff := difflib.UnifiedDiff{
+ A: difflib.SplitLines(
+ dumpCfg.Sdump(kvPayment),
+ ),
+ B: difflib.SplitLines(
+ dumpCfg.Sdump(sqlPayment),
+ ),
+ FromFile: "kv",
+ ToFile: "sql",
+ Context: 3,
+ }
+ diffText, _ := difflib.GetUnifiedDiffString(diff)
+
+ return fmt.Errorf("payment mismatch %x\n%s",
+ paymentHash[:8], diffText)
+ }
+
+ return nil
+}
+
+// normalizePaymentForCompare normalizes fields that are expected to differ
+// between KV and SQL representations before deep comparison.
+func normalizePaymentForCompare(payment *MPPayment) {
+ if payment == nil {
+ return
+ }
+
+ // SequenceNum will not be equal because the kv db can have already
+ // payments deleted during its lifetime.
+ payment.SequenceNum = 0
+
+ // We normalize timestamps before deep-comparing KV vs SQL objects.
+ //
+ // - **Microseconds**: SQL backends typically persist timestamps at
+ // microsecond precision (e.g. Postgres), while KV (Go `time.Time`)
+ // can contain nanoseconds. Truncating avoids false mismatches caused
+ // solely by differing storage precision.
+ //
+ // - **Local timezone**: when reading from SQL, timestamps are typically
+ // materialized in the local timezone by the SQL layer (and/or
+ // converters). Converting both sides to `time.Local` ensures the
+ // comparison is consistent across KV and SQL representations.
+ trunc := func(t time.Time) time.Time {
+ if t.IsZero() {
+ return t
+ }
+
+ return time.Unix(0, t.UnixNano()).
+ In(time.Local).
+ Truncate(time.Microsecond)
+ }
+
+ // Normalize PaymentCreationInfo fields.
+ if payment.Info != nil {
+ payment.Info.CreationTime = trunc(
+ payment.Info.CreationTime,
+ )
+ if len(payment.Info.PaymentRequest) == 0 {
+ payment.Info.PaymentRequest = []byte{}
+ }
+ if len(payment.Info.FirstHopCustomRecords) == 0 {
+ payment.Info.FirstHopCustomRecords = lnwire.
+ CustomRecords{}
+ }
+ }
+
+ // Normalize HTLCAttemptInfo so nil is converted to an empty slice.
+ if len(payment.HTLCs) == 0 {
+ payment.HTLCs = []HTLCAttempt{}
+ }
+
+ // Normalize HTLC attempt ordering; SQL/KV may return attempts
+ // in different orders.
+ sort.SliceStable(payment.HTLCs, func(i, j int) bool {
+ return payment.HTLCs[i].AttemptID < payment.HTLCs[j].AttemptID
+ })
+
+ // Normalize HTLCAttemptInfo fields.
+ for i := range payment.HTLCs {
+ htlc := &payment.HTLCs[i]
+
+ htlc.AttemptTime = trunc(htlc.AttemptTime)
+ if htlc.Settle != nil {
+ htlc.Settle.SettleTime = trunc(
+ htlc.Settle.SettleTime,
+ )
+ }
+ if htlc.Failure != nil {
+ htlc.Failure.FailTime = trunc(
+ htlc.Failure.FailTime,
+ )
+ }
+
+ // Clear cached fields not persisted in storage.
+ htlc.onionBlob = [1366]byte{}
+ htlc.circuit = nil
+ htlc.cachedSessionKey = nil
+
+ if len(htlc.Route.FirstHopWireCustomRecords) == 0 {
+ htlc.Route.FirstHopWireCustomRecords =
+ lnwire.CustomRecords{}
+ }
+
+ for j := range htlc.Route.Hops {
+ if len(htlc.Route.Hops[j].CustomRecords) == 0 {
+ htlc.Route.Hops[j].CustomRecords =
+ record.CustomSet{}
+ }
+ }
+ }
+}
+
+// duplicateRecord is a record that represents a duplicate payment.
+type duplicateRecord struct {
+ AmountMsat int64
+ CreatedAt time.Time
+ FailReason sql.NullInt32
+ SettlePreimage []byte
+ SettleTime sql.NullTime
+}
+
+// compareDuplicatePayments validates migrated duplicate rows against KV data.
+func compareDuplicatePayments(ctx context.Context, paymentBucket kvdb.RBucket,
+ sqlDB SQLQueries, paymentID int64, hash lntypes.Hash) error {
+
+ // Fetch the duplicate payments from the KV store.
+ kvDuplicates, err := fetchDuplicateRecords(paymentBucket)
+ if err != nil {
+ return fmt.Errorf("fetch KV duplicates %x: %w",
+ hash[:8], err)
+ }
+
+ // Fetch the duplicate payments from the SQL store.
+ sqlDuplicates, err := sqlDB.FetchPaymentDuplicates(ctx, paymentID)
+ if err != nil {
+ return fmt.Errorf("fetch SQL duplicates %x: %w",
+ hash[:8], err)
+ }
+
+ if len(kvDuplicates) != len(sqlDuplicates) {
+ return fmt.Errorf("duplicate count mismatch %x: kv=%d "+
+ "sql=%d", hash[:8], len(kvDuplicates),
+ len(sqlDuplicates))
+ }
+
+ kvNormalized := normalizeDuplicateRecords(kvDuplicates)
+ sqlNormalized := normalizeDuplicateRecords(
+ dbDuplicatesToDuplicateRecords(sqlDuplicates),
+ )
+
+ sortDuplicates(kvNormalized)
+ sortDuplicates(sqlNormalized)
+
+ if !reflect.DeepEqual(kvNormalized, sqlNormalized) {
+ dumpCfg := spew.ConfigState{
+ DisablePointerAddresses: true,
+ DisableCapacities: true,
+ DisableMethods: true,
+ SortKeys: true,
+ }
+ diff := difflib.UnifiedDiff{
+ A: difflib.SplitLines(
+ dumpCfg.Sdump(kvNormalized),
+ ),
+ B: difflib.SplitLines(
+ dumpCfg.Sdump(sqlNormalized),
+ ),
+ FromFile: "kv",
+ ToFile: "sql",
+ Context: 3,
+ }
+ diffText, _ := difflib.GetUnifiedDiffString(diff)
+
+ return fmt.Errorf("duplicate mismatch %x\n%s",
+ hash[:8], diffText)
+ }
+
+ return nil
+}
+
+// fetchDuplicateRecords reads duplicate payment records from the KV bucket.
+func fetchDuplicateRecords(paymentBucket kvdb.RBucket) ([]duplicateRecord,
+ error) {
+
+ dupBucket := paymentBucket.NestedReadBucket(duplicatePaymentsBucket)
+ if dupBucket == nil {
+ return nil, nil
+ }
+
+ var duplicates []duplicateRecord
+ err := dupBucket.ForEach(func(seqBytes, _ []byte) error {
+ if len(seqBytes) != 8 {
+ return nil
+ }
+
+ subBucket := dupBucket.NestedReadBucket(seqBytes)
+ if subBucket == nil {
+ return nil
+ }
+
+ creationData := subBucket.Get(duplicatePaymentCreationInfoKey)
+ if creationData == nil {
+ return fmt.Errorf("missing duplicate creation info")
+ }
+
+ creationInfo, err := deserializeDuplicatePaymentCreationInfo(
+ bytes.NewReader(creationData),
+ )
+ if err != nil {
+ return fmt.Errorf("deserialize duplicate creation "+
+ "info: %w", err)
+ }
+
+ settleData := subBucket.Get(duplicatePaymentSettleInfoKey)
+ failReasonData := subBucket.Get(duplicatePaymentFailInfoKey)
+
+ if settleData != nil && len(failReasonData) > 0 {
+ return fmt.Errorf("duplicate has both settle and " +
+ "fail info")
+ }
+
+ var (
+ failReason sql.NullInt32
+ settlePreimage []byte
+ settleTime sql.NullTime
+ )
+
+ switch {
+ case settleData != nil:
+ settlePreimage, settleTime, err =
+ parseDuplicateSettleData(settleData)
+ if err != nil {
+ return err
+ }
+ case len(failReasonData) > 0:
+ failReason = sql.NullInt32{
+ Int32: int32(failReasonData[0]),
+ Valid: true,
+ }
+ default:
+ // If the duplicate has no settle or fail info, it is
+ // considered failed. Every duplicate payment must have
+ // either a settle or fail info in the sql database. So
+ // we set the fail reason to error to mimic the behavior
+ // for the kv store.
+ failReason = sql.NullInt32{
+ Int32: int32(FailureReasonError),
+ Valid: true,
+ }
+ }
+
+ duplicates = append(duplicates, duplicateRecord{
+ AmountMsat: int64(creationInfo.Value),
+ CreatedAt: normalizeTimeForSQL(
+ creationInfo.CreationTime,
+ ),
+ FailReason: failReason,
+ SettlePreimage: settlePreimage,
+ SettleTime: settleTime,
+ })
+
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+
+ return duplicates, nil
+}
+
+// dbDuplicatesToDuplicateRecords maps SQL duplicate rows into comparable
+// duplicate records.
+func dbDuplicatesToDuplicateRecords(
+ rows []sqlc.PaymentDuplicate) []duplicateRecord {
+
+ duplicates := make([]duplicateRecord, 0, len(rows))
+ for _, row := range rows {
+ duplicates = append(duplicates, duplicateRecord{
+ AmountMsat: row.AmountMsat,
+ CreatedAt: row.CreatedAt,
+ FailReason: row.FailReason,
+ SettlePreimage: row.SettlePreimage,
+ SettleTime: row.SettleTime,
+ })
+ }
+
+ return duplicates
+}
+
+// normalizeDuplicateRecords normalizes time precision and empty fields.
+func normalizeDuplicateRecords(records []duplicateRecord) []duplicateRecord {
+ if len(records) == 0 {
+ return []duplicateRecord{}
+ }
+
+ trunc := func(t time.Time) time.Time {
+ if t.IsZero() {
+ return t
+ }
+
+ return t.In(time.Local).Truncate(time.Microsecond)
+ }
+
+ for i := range records {
+ records[i].CreatedAt = trunc(records[i].CreatedAt)
+ if records[i].SettleTime.Valid {
+ records[i].SettleTime.Time = trunc(
+ records[i].SettleTime.Time,
+ )
+ }
+
+ if len(records[i].SettlePreimage) == 0 {
+ records[i].SettlePreimage = []byte{}
+ }
+ }
+
+ return records
+}
+
+// sortDuplicates orders records deterministically for deep comparison.
+func sortDuplicates(records []duplicateRecord) {
+ sort.SliceStable(records, func(i, j int) bool {
+ ai := records[i]
+ aj := records[j]
+
+ // Duplicates are "duplicates" because they share the same
+ // payment identifier. So ordering can be stable using
+ // timestamp + amount.
+ if !ai.CreatedAt.Equal(aj.CreatedAt) {
+ return ai.CreatedAt.Before(aj.CreatedAt)
+ }
+
+ return ai.AmountMsat < aj.AmountMsat
+ })
+}
+
+// validatePaymentCounts compares the number of migrated payments with the SQL
+// payment count to catch missing rows.
+func validatePaymentCounts(ctx context.Context, sqlDB SQLQueries,
+ expectedCount int64) error {
+
+ sqlCount, err := sqlDB.CountPayments(ctx)
+ if err != nil {
+ return fmt.Errorf("count SQL payments: %w", err)
+ }
+ if expectedCount != sqlCount {
+ return fmt.Errorf("payment count mismatch: kv=%d sql=%d",
+ expectedCount, sqlCount)
+ }
+
+ return nil
+}
diff --git a/payments/db/migration1/sql_migration.go b/payments/db/migration1/sql_migration.go
new file mode 100644
index 0000000..a7a02af
--- /dev/null
+++ b/payments/db/migration1/sql_migration.go
@@ -0,0 +1,817 @@
+package migration1
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "fmt"
+ "strconv"
+ "time"
+
+ "github.com/lightningnetwork/lnd/kvdb"
+ "github.com/lightningnetwork/lnd/lntypes"
+ "github.com/lightningnetwork/lnd/lnwire"
+ "github.com/lightningnetwork/lnd/payments/db/migration1/sqlc"
+ "golang.org/x/time/rate"
+)
+
+// MigrationStats tracks migration progress.
+type MigrationStats struct {
+ TotalPayments int64
+ SuccessfulPayments int64
+ FailedPayments int64
+ InFlightPayments int64
+ InitiatedPayments int64
+ TotalAttempts int64
+ SettledAttempts int64
+ FailedAttempts int64
+ InFlightAttempts int64
+ TotalHops int64
+ DuplicatePayments int64
+ DuplicateEntries int64
+ SkippedPayments int64
+ MigrationDuration time.Duration
+}
+
+// MigratePaymentsKVToSQL migrates payments from KV to SQL and validates
+// migrated data in batches. Callers are responsible for executing this within
+// a single SQL transaction if atomicity is required.
+func MigratePaymentsKVToSQL(ctx context.Context, kvBackend kvdb.Backend,
+ sqlDB SQLQueries, cfg *SQLStoreConfig) error {
+
+ if cfg == nil {
+ return fmt.Errorf("missing SQL store config for migration")
+ }
+
+ if cfg.QueryCfg == nil {
+ return fmt.Errorf("missing SQL store config for " +
+ "validation")
+ }
+
+ if cfg.QueryCfg.MaxBatchSize == 0 {
+ return fmt.Errorf("invalid max batch size for " +
+ "validation")
+ }
+
+ stats := &MigrationStats{}
+ startTime := time.Now()
+
+ log.Infof("Starting payment migration from KV to SQL...")
+
+ var (
+ validationBatch []migratedPaymentRef
+ validatedPayments int64
+
+ reportInterval = rate.Sometimes{Interval: 5 * time.Second}
+ validationInterval = rate.Sometimes{Interval: 5 * time.Second}
+ )
+
+ // Open the KV backend in read-only mode.
+ err := kvBackend.View(func(kvTx kvdb.RTx) error {
+ // In case we start with an empty database, there are no
+ // payments to migrate.
+ paymentsBucket := kvTx.ReadBucket(paymentsRootBucket)
+ if paymentsBucket == nil {
+ log.Infof("No payments bucket found - database is " +
+ "empty")
+
+ return nil
+ }
+
+ // The index bucket maps sequence number -> payment hash.
+ indexes := kvTx.ReadBucket(paymentsIndexBucket)
+ if indexes == nil {
+ return fmt.Errorf("index bucket does not exist")
+ }
+
+ // We iterate over all sequence numbers in the index bucket to
+ // make sure we have the correct order of payments. Otherwise,
+ // if we just loop over the payments bucket, we might get the
+ // payments not in the chronological order but rather the
+ // lexicographical order of the payment hashes.
+ return indexes.ForEach(func(seqKey, indexVal []byte) error {
+ // Progress reporting based on time + actual work done.
+ reportProgress := func() {
+ elapsed := time.Since(startTime)
+ if elapsed <= 0 {
+ return
+ }
+
+ if stats.TotalPayments == 0 {
+ return
+ }
+
+ paymentRate := float64(stats.TotalPayments) /
+ elapsed.Seconds()
+ attemptRate := float64(stats.TotalAttempts) /
+ elapsed.Seconds()
+
+ log.Infof("Progress: %d payments, %d "+
+ "attempts, %d hops | Rate: %.1f "+
+ "pmt/s, %.1f att/s | Elapsed: %v",
+ stats.TotalPayments,
+ stats.TotalAttempts, stats.TotalHops,
+ paymentRate, attemptRate,
+ elapsed.Round(time.Second),
+ )
+ }
+
+ reportInterval.Do(reportProgress)
+
+ r := bytes.NewReader(indexVal)
+ paymentHash, err := deserializePaymentIndex(r)
+ if err != nil {
+ return err
+ }
+
+ paymentBucket := paymentsBucket.NestedReadBucket(
+ paymentHash[:],
+ )
+ if paymentBucket == nil {
+ // We skip the entry in case this sequence
+ // number does not have a corresponding
+ // payment bucket. But aborting would not help
+ // either because it is just a db inconsistency.
+ log.Warnf("Missing bucket for payment %x",
+ paymentHash[:8])
+
+ stats.SkippedPayments++
+
+ return nil
+ }
+
+ // Every payment bucket should have a sequence number
+ // which is also important to check for duplicates.
+ seqBytes := paymentBucket.Get(paymentSequenceKey)
+ if seqBytes == nil {
+ return ErrNoSequenceNumber
+ }
+
+ // Skip duplicates. They are migrated into the
+ // payment_duplicates table when the primary payment is
+ // processed.
+ if !bytes.Equal(seqBytes, seqKey) {
+ return nil
+ }
+
+ // Fetch the payment from the kv store.
+ payment, err := fetchPayment(paymentBucket)
+ if err != nil {
+ return fmt.Errorf("fetch payment %x: %w",
+ paymentHash[:8], err)
+ }
+
+ // Migrate the payment to the SQL database.
+ paymentID, err := migratePayment(
+ ctx, payment, paymentHash, sqlDB, stats,
+ )
+ if err != nil {
+ return fmt.Errorf("migrate payment %x: %w",
+ paymentHash[:8], err)
+ }
+
+ // Check for duplicates.
+ dupBucket := paymentBucket.NestedReadBucket(
+ duplicatePaymentsBucket,
+ )
+ if dupBucket != nil {
+ err = migrateDuplicatePayments(
+ ctx, dupBucket, paymentHash,
+ paymentID, sqlDB, stats,
+ )
+ if err != nil {
+ return fmt.Errorf("migrate duplicates "+
+ "%x: %w", paymentHash[:8],
+ err)
+ }
+ }
+
+ // Add the payment to the validation batch.
+ validationBatch = append(
+ validationBatch, migratedPaymentRef{
+ Hash: paymentHash,
+ PaymentID: paymentID,
+ },
+ )
+ if uint32(len(validationBatch)) >=
+ cfg.QueryCfg.MaxBatchSize {
+
+ err := validateMigratedPaymentBatch(
+ ctx, kvBackend, sqlDB,
+ cfg,
+ validationBatch,
+ )
+ if err != nil {
+ return err
+ }
+
+ validatedPayments += int64(
+ len(validationBatch),
+ )
+
+ // Log validation progress periodically.
+ validationInterval.Do(func() {
+ log.Infof("Validated %d/%d "+
+ "payments",
+ validatedPayments,
+ stats.TotalPayments,
+ )
+ })
+
+ validationBatch = validationBatch[:0]
+ }
+
+ return nil
+ })
+ }, func() {})
+
+ if err != nil {
+ return fmt.Errorf("migrate payments: %w", err)
+ }
+
+ // Validate any remaining payments in the batch.
+ if len(validationBatch) > 0 {
+ if err := validateMigratedPaymentBatch(
+ ctx, kvBackend, sqlDB, cfg, validationBatch,
+ ); err != nil {
+ return err
+ }
+
+ validatedPayments += int64(len(validationBatch))
+ log.Infof("Validated %d/%d payments", validatedPayments,
+ stats.TotalPayments)
+ }
+
+ // Validate the total number of payments as an additional sanity check.
+ if err := validatePaymentCounts(
+ ctx, sqlDB, stats.TotalPayments,
+ ); err != nil {
+ return err
+ }
+
+ stats.MigrationDuration = time.Since(startTime)
+
+ printMigrationSummary(stats)
+
+ return nil
+}
+
+// normalizeTimeForSQL converts a timestamp into the representation we persist
+// and compare against in SQL:
+// - drops any monotonic clock reading (SQL can't store it),
+// - forces UTC for deterministic comparisons across environments.
+//
+// A zero time is returned unchanged.
+func normalizeTimeForSQL(t time.Time) time.Time {
+ if t.IsZero() {
+ return t
+ }
+
+ return time.Unix(0, t.UnixNano()).UTC()
+}
+
+// migratePayment migrates a single payment from KV to SQL.
+func migratePayment(ctx context.Context, payment *MPPayment, hash lntypes.Hash,
+ sqlDB SQLQueries, stats *MigrationStats) (int64, error) {
+
+ // Update migration stats based on payment status.
+ switch payment.Status {
+ case StatusSucceeded:
+ stats.SuccessfulPayments++
+
+ case StatusFailed:
+ stats.FailedPayments++
+
+ case StatusInFlight:
+ stats.InFlightPayments++
+
+ case StatusInitiated:
+ stats.InitiatedPayments++
+ }
+
+ // Prepare fail reason for SQL insert.
+ var failReason sql.NullInt32
+ if payment.FailureReason != nil {
+ failReason = sql.NullInt32{
+ Int32: int32(*payment.FailureReason),
+ Valid: true,
+ }
+ }
+
+ // Insert payment using migration query.
+ paymentID, err := sqlDB.InsertPaymentMig(
+ ctx, sqlc.InsertPaymentMigParams{
+ AmountMsat: int64(payment.Info.Value),
+ CreatedAt: normalizeTimeForSQL(
+ payment.Info.CreationTime,
+ ),
+ PaymentIdentifier: hash[:],
+ FailReason: failReason,
+ })
+ if err != nil {
+ return 0, fmt.Errorf("insert payment: %w", err)
+ }
+
+ // Insert payment intent.
+ //
+ // Only insert a row if we have an actual intent payload. For legacy
+ // hash-only/keysend-style payments, the intent may be absent.
+ if len(payment.Info.PaymentRequest) > 0 {
+ _, err = sqlDB.InsertPaymentIntent(
+ ctx, sqlc.InsertPaymentIntentParams{
+ PaymentID: paymentID,
+ IntentType: int16(PaymentIntentTypeBolt11),
+ IntentPayload: payment.Info.PaymentRequest,
+ },
+ )
+ if err != nil {
+ return 0, fmt.Errorf("insert intent: %w", err)
+ }
+ }
+
+ // Insert first hop custom records (payment level).
+ for key, value := range payment.Info.FirstHopCustomRecords {
+ err = sqlDB.InsertPaymentFirstHopCustomRecord(ctx,
+ sqlc.InsertPaymentFirstHopCustomRecordParams{
+ PaymentID: paymentID,
+ Key: int64(key),
+ Value: value,
+ },
+ )
+ if err != nil {
+ return 0, fmt.Errorf("insert custom record: %w", err)
+ }
+ }
+
+ // Migrate HTLC attempts.
+ for _, htlc := range payment.HTLCs {
+ err = migrateHTLCAttempt(
+ ctx, paymentID, hash, &htlc, sqlDB, stats,
+ )
+ if err != nil {
+ return 0, fmt.Errorf("migrate attempt %d: %w",
+ htlc.AttemptID, err)
+ }
+ }
+
+ stats.TotalPayments++
+
+ return paymentID, nil
+}
+
+// migrateHTLCAttempt migrates a single HTLC attempt.
+func migrateHTLCAttempt(ctx context.Context, paymentID int64,
+ parentPaymentHash lntypes.Hash, htlc *HTLCAttempt, sqlDB SQLQueries,
+ stats *MigrationStats) error {
+
+ // Validate that we have a payment hash for the attempt.
+ //
+ // NOTE: We always require an attempt payment hash. A missing hash is an
+ // unrecoverable inconsistency. All payments should have a payment hash
+ // (AMP,MPP,Legacy)
+ var paymentHash []byte
+ switch {
+ case htlc.Hash != nil:
+ paymentHash = (*htlc.Hash)[:]
+
+ default:
+ return fmt.Errorf("HTLC attempt %d missing payment hash "+
+ "(parent payment hash=%x)", htlc.AttemptID,
+ parentPaymentHash[:])
+ }
+
+ firstHopAmountMsat := int64(htlc.Route.FirstHopAmount.Val.Int())
+
+ // Get the session key bytes.
+ sessionKeyBytes := htlc.SessionKey().Serialize()
+
+ // Insert HTLC attempt.
+ _, err := sqlDB.InsertHtlcAttempt(ctx, sqlc.InsertHtlcAttemptParams{
+ PaymentID: paymentID,
+ AttemptIndex: int64(htlc.AttemptID),
+ SessionKey: sessionKeyBytes,
+ AttemptTime: normalizeTimeForSQL(htlc.AttemptTime),
+ PaymentHash: paymentHash,
+ FirstHopAmountMsat: firstHopAmountMsat,
+ RouteTotalTimeLock: int32(htlc.Route.TotalTimeLock),
+ RouteTotalAmount: int64(htlc.Route.TotalAmount),
+ RouteSourceKey: htlc.Route.SourcePubKey[:],
+ })
+ if err != nil {
+ return fmt.Errorf("insert HTLC attempt: %w", err)
+ }
+
+ // Insert the route-level first hop custom records.
+ for key, value := range htlc.Route.FirstHopWireCustomRecords {
+ err = sqlDB.InsertPaymentAttemptFirstHopCustomRecord(
+ ctx,
+ sqlc.InsertPaymentAttemptFirstHopCustomRecordParams{
+ HtlcAttemptIndex: int64(htlc.AttemptID),
+ Key: int64(key),
+ Value: value,
+ },
+ )
+ if err != nil {
+ return fmt.Errorf("insert attempt first hop custom "+
+ "record: %w", err)
+ }
+ }
+
+ // Insert route hops.
+ for hopIndex := range htlc.Route.Hops {
+ hop := htlc.Route.Hops[hopIndex]
+ err = migrateRouteHop(
+ ctx, int64(htlc.AttemptID), hopIndex, hop,
+ sqlDB, stats,
+ )
+ if err != nil {
+ return fmt.Errorf("migrate hop %d: %w", hopIndex, err)
+ }
+ }
+
+ // Handle attempt resolution (settle or fail).
+ switch {
+ case htlc.Settle != nil:
+ // Settled
+ err = sqlDB.SettleAttempt(ctx, sqlc.SettleAttemptParams{
+ AttemptIndex: int64(htlc.AttemptID),
+ ResolutionTime: normalizeTimeForSQL(
+ htlc.Settle.SettleTime,
+ ),
+ ResolutionType: int32(HTLCAttemptResolutionSettled),
+ SettlePreimage: htlc.Settle.Preimage[:],
+ })
+ if err != nil {
+ return fmt.Errorf("settle attempt: %w", err)
+ }
+
+ stats.SettledAttempts++
+
+ case htlc.Failure != nil:
+ var failureMsg bytes.Buffer
+ if htlc.Failure.Message != nil {
+ err := lnwire.EncodeFailureMessage(
+ &failureMsg, htlc.Failure.Message, 0,
+ )
+ if err != nil {
+ return fmt.Errorf("failed to encode "+
+ "failure message: %w", err)
+ }
+ }
+
+ err = sqlDB.FailAttempt(ctx, sqlc.FailAttemptParams{
+ AttemptIndex: int64(htlc.AttemptID),
+ ResolutionTime: normalizeTimeForSQL(
+ htlc.Failure.FailTime,
+ ),
+ ResolutionType: int32(HTLCAttemptResolutionFailed),
+ FailureSourceIndex: sql.NullInt32{
+ Int32: int32(htlc.Failure.FailureSourceIndex),
+ Valid: true,
+ },
+ HtlcFailReason: sql.NullInt32{
+ Int32: int32(htlc.Failure.Reason),
+ Valid: true,
+ },
+ FailureMsg: failureMsg.Bytes(),
+ })
+ if err != nil {
+ return fmt.Errorf("fail attempt: %w", err)
+ }
+
+ stats.FailedAttempts++
+
+ default:
+ // If the attempt is not settled or failed, it is in flight.
+ stats.InFlightAttempts++
+ }
+
+ stats.TotalAttempts++
+
+ return nil
+}
+
+// migrateRouteHop migrates a single route hop.
+func migrateRouteHop(ctx context.Context,
+ attemptID int64, hopIndex int, hop *Hop, sqlDB SQLQueries,
+ stats *MigrationStats) error {
+
+ // Convert channel ID to string representation of uint64.
+ // The SCID is stored as a decimal string to match the converter
+ // expectations (sql_converters.go:173).
+ scidStr := strconv.FormatUint(hop.ChannelID, 10)
+
+ // Insert route hop.
+ hopID, err := sqlDB.InsertRouteHop(ctx, sqlc.InsertRouteHopParams{
+ HtlcAttemptIndex: attemptID,
+ HopIndex: int32(hopIndex),
+ PubKey: hop.PubKeyBytes[:],
+ Scid: scidStr,
+ OutgoingTimeLock: int32(hop.OutgoingTimeLock),
+ AmtToForward: int64(hop.AmtToForward),
+ MetaData: hop.Metadata,
+ })
+ if err != nil {
+ return fmt.Errorf("insert hop: %w", err)
+ }
+
+ // Check for blinded route data (route blinding).
+ if len(hop.EncryptedData) > 0 || hop.BlindingPoint != nil ||
+ hop.TotalAmtMsat != 0 {
+
+ var blindingPoint []byte
+ if hop.BlindingPoint != nil {
+ blindingPoint = hop.BlindingPoint.SerializeCompressed()
+ }
+
+ var totalAmt sql.NullInt64
+ if hop.TotalAmtMsat != 0 {
+ totalAmt = sql.NullInt64{
+ Int64: int64(hop.TotalAmtMsat),
+ Valid: true,
+ }
+ }
+
+ err := sqlDB.InsertRouteHopBlinded(
+ ctx, sqlc.InsertRouteHopBlindedParams{
+ HopID: hopID,
+ EncryptedData: hop.EncryptedData,
+ BlindingPoint: blindingPoint,
+ BlindedPathTotalAmt: totalAmt,
+ },
+ )
+ if err != nil {
+ return fmt.Errorf("insert blinded hop: %w", err)
+ }
+ }
+
+ // Check for MPP record.
+ if hop.MPP != nil {
+ paymentAddr := hop.MPP.PaymentAddr()
+ err = sqlDB.InsertRouteHopMpp(ctx, sqlc.InsertRouteHopMppParams{
+ HopID: hopID,
+ PaymentAddr: paymentAddr[:],
+ TotalMsat: int64(hop.MPP.TotalMsat()),
+ })
+ if err != nil {
+ return fmt.Errorf("insert MPP: %w", err)
+ }
+ }
+
+ // Check for AMP record.
+ if hop.AMP != nil {
+ rootShare := hop.AMP.RootShare()
+ setID := hop.AMP.SetID()
+ err = sqlDB.InsertRouteHopAmp(ctx, sqlc.InsertRouteHopAmpParams{
+ HopID: hopID,
+ RootShare: rootShare[:],
+ SetID: setID[:],
+ ChildIndex: int32(hop.AMP.ChildIndex()),
+ })
+ if err != nil {
+ return fmt.Errorf("insert AMP: %w", err)
+ }
+ }
+
+ // Check for custom records.
+ if hop.CustomRecords != nil {
+ for tlvType, value := range hop.CustomRecords {
+ err = sqlDB.InsertPaymentHopCustomRecord(
+ ctx,
+ sqlc.InsertPaymentHopCustomRecordParams{
+ HopID: hopID,
+ Key: int64(tlvType),
+ Value: value,
+ },
+ )
+ if err != nil {
+ return fmt.Errorf("insert hop custom "+
+ "record: %w", err)
+ }
+ }
+ }
+
+ stats.TotalHops++
+
+ return nil
+}
+
+// migrateDuplicatePayments migrates duplicate payments into the dedicated
+// payment_duplicates table.
+func migrateDuplicatePayments(ctx context.Context, dupBucket kvdb.RBucket,
+ hash [32]byte, primaryPaymentID int64, sqlDB SQLQueries,
+ stats *MigrationStats) error {
+
+ duplicateCount := 0
+
+ err := dupBucket.ForEach(func(seqBytes, _ []byte) error {
+ // The duplicates bucket should only contain nested buckets
+ // keyed by 8-byte sequence numbers. Skip any unexpected keys
+ // (defensive check for corrupted or malformed data).
+ if len(seqBytes) != 8 {
+ log.Warnf("Skipping unexpected key in duplicates "+
+ "bucket for payment %x: key length %d, "+
+ "expected 8",
+ hash[:8], len(seqBytes),
+ )
+
+ return nil
+ }
+
+ seqNum := byteOrder.Uint64(seqBytes)
+ subBucket := dupBucket.NestedReadBucket(seqBytes)
+ if subBucket == nil {
+ return nil
+ }
+
+ duplicateCount++
+ log.Infof("Migrating duplicate payment seq=%d for "+
+ "payment %x", seqNum, hash[:8])
+
+ err := migrateSingleDuplicatePayment(
+ ctx, subBucket, hash, primaryPaymentID, seqNum,
+ sqlDB,
+ )
+ if err != nil {
+ return fmt.Errorf(
+ "migrate duplicate payment seq=%d: %w",
+ seqNum, err,
+ )
+ }
+
+ return nil
+ })
+
+ if duplicateCount > 0 {
+ stats.DuplicatePayments++
+ stats.DuplicateEntries += int64(duplicateCount)
+
+ log.Infof("Payment %x had %d duplicate(s) migrated", hash[:8],
+ duplicateCount)
+ }
+
+ return err
+}
+
+// migrateSingleDuplicatePayment inserts a duplicate payment record for the
+// given payment hash into payment_duplicates.
+func migrateSingleDuplicatePayment(ctx context.Context, dupBucket kvdb.RBucket,
+ hash [32]byte, primaryPaymentID int64, duplicateSeq uint64,
+ sqlDB SQLQueries) error {
+
+ creationData := dupBucket.Get(duplicatePaymentCreationInfoKey)
+ if creationData == nil {
+ return fmt.Errorf("duplicate payment seq=%d missing "+
+ "creation info (payment=%x)", duplicateSeq, hash[:8])
+ }
+
+ creationInfo, err := deserializeDuplicatePaymentCreationInfo(
+ bytes.NewReader(creationData),
+ )
+ if err != nil {
+ return fmt.Errorf("deserialize duplicate creation "+
+ "info: %w", err)
+ }
+
+ settleData := dupBucket.Get(duplicatePaymentSettleInfoKey)
+ failReasonData := dupBucket.Get(duplicatePaymentFailInfoKey)
+ attemptData := dupBucket.Get(duplicatePaymentAttemptInfoKey)
+
+ if settleData != nil && len(failReasonData) > 0 {
+ return fmt.Errorf("duplicate payment seq=%d has both "+
+ "settle and fail info (payment=%x)", duplicateSeq,
+ hash[:8])
+ }
+
+ var (
+ failReason sql.NullInt32
+ settlePreimage []byte
+ settleTime sql.NullTime
+ )
+
+ switch {
+ case settleData != nil:
+ settlePreimage, settleTime, err = parseDuplicateSettleData(
+ settleData,
+ )
+ if err != nil {
+ return err
+ }
+
+ case len(failReasonData) > 0:
+ failReason = sql.NullInt32{
+ Int32: int32(failReasonData[0]),
+ Valid: true,
+ }
+
+ default:
+ // If the duplicate payment has no settle or fail info,
+ // we mark it as failed during the migration. Duplicate
+ // payments were a bug in older versions of LND, so we can be
+ // sure if a duplicate payment has no failure reason or
+ // settlement data, the corresponding HTLC for this payment
+ // has been failed (resolved).
+ if attemptData == nil {
+ log.Warnf("Duplicate payment seq=%d has no "+
+ "attempt info and no resolution (payment=%x); "+
+ "marking failed", duplicateSeq, hash[:8])
+ } else {
+ log.Warnf("Duplicate payment seq=%d has attempt "+
+ "info but no resolution (payment=%x); "+
+ "marking failed", duplicateSeq, hash[:8])
+ }
+
+ failReason = sql.NullInt32{
+ Int32: int32(FailureReasonError),
+ Valid: true,
+ }
+ }
+
+ _, err = sqlDB.InsertPaymentDuplicateMig(
+ ctx, sqlc.InsertPaymentDuplicateMigParams{
+ PaymentID: primaryPaymentID,
+ AmountMsat: int64(creationInfo.Value),
+ CreatedAt: normalizeTimeForSQL(
+ creationInfo.CreationTime,
+ ),
+ FailReason: failReason,
+ SettlePreimage: settlePreimage,
+ SettleTime: settleTime,
+ },
+ )
+ if err != nil {
+ return fmt.Errorf("insert duplicate payment: %w", err)
+ }
+
+ return nil
+}
+
+// parseDuplicateSettleData extracts settle data from either legacy or modern
+// duplicate formats.
+func parseDuplicateSettleData(settleData []byte) ([]byte, sql.NullTime, error) {
+ if len(settleData) == lntypes.PreimageSize {
+ return append([]byte(nil), settleData...), sql.NullTime{}, nil
+ }
+
+ settleInfo, err := deserializeHTLCSettleInfo(
+ bytes.NewReader(settleData),
+ )
+ if err != nil {
+ return nil, sql.NullTime{},
+ fmt.Errorf("deserialize duplicate settle: %w", err)
+ }
+
+ settleTime := normalizeTimeForSQL(settleInfo.SettleTime)
+
+ return settleInfo.Preimage[:], sql.NullTime{
+ Time: settleTime,
+ Valid: !settleTime.IsZero(),
+ }, nil
+}
+
+// printMigrationSummary prints a summary of the migration.
+func printMigrationSummary(stats *MigrationStats) {
+ if stats.TotalPayments == 0 {
+ log.Infof("No payments migrated - database is empty")
+
+ return
+ }
+
+ log.Infof("========================================")
+ log.Infof(" Payment Migration Summary")
+ log.Infof("========================================")
+ log.Infof("Total Payments: %d", stats.TotalPayments)
+ log.Infof(" Successful: %d", stats.SuccessfulPayments)
+ log.Infof(" Failed: %d", stats.FailedPayments)
+ log.Infof(" In-Flight: %d", stats.InFlightPayments)
+ log.Infof(" Initiated: %d", stats.InitiatedPayments)
+ log.Infof("")
+ log.Infof("Total HTLC Attempts: %d", stats.TotalAttempts)
+ log.Infof(" Settled: %d", stats.SettledAttempts)
+ log.Infof(" Failed: %d", stats.FailedAttempts)
+ log.Infof(" In-Flight: %d", stats.InFlightAttempts)
+ log.Infof("")
+ log.Infof("Total Route Hops: %d", stats.TotalHops)
+
+ if stats.SkippedPayments > 0 {
+ log.Infof("")
+ log.Warnf("SKIPPED PAYMENTS:")
+ log.Warnf(" Indexed payments with missing buckets: %d",
+ stats.SkippedPayments)
+ log.Warnf(" These indicate minor DB inconsistencies.")
+ }
+
+ if stats.DuplicatePayments > 0 {
+ log.Infof("")
+ log.Warnf("DUPLICATE PAYMENTS DETECTED:")
+ log.Warnf(" Unique payment hashes with duplicates: %d",
+ stats.DuplicatePayments)
+ log.Warnf(" Total duplicate entries migrated: %d",
+ stats.DuplicateEntries)
+ log.Warnf(" These were caused by an old LND bug.")
+ }
+
+ log.Infof("")
+ log.Infof("Migration Duration: %v", stats.MigrationDuration)
+ log.Infof("========================================")
+}
Why this scored 30/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.