From a9747207f6a5aab2a3723027b16f41e6898d3a1b Mon Sep 17 00:00:00 2001 From: ziggie Date: Sun, 11 Jan 2026 00:02:52 +0100 Subject: [PATCH] payments/migration1: add the payments mig code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the KV→SQL payment migration and add an in-migration validation pass that deep-compares KV and SQL payment data in batches. Duplicate payments are migrated into the payment_duplicates table, and duplicates without attempt info or explicit resolution are marked failed to ensure terminal state. Validation checks those rows as well. --- .../db/migration1/migration_validation.go | 555 ++++++++++++ payments/db/migration1/sql_migration.go | 817 ++++++++++++++++++ 2 files changed, 1372 insertions(+) create mode 100644 payments/db/migration1/migration_validation.go create mode 100644 payments/db/migration1/sql_migration.go diff --git a/payments/db/migration1/migration_validation.go b/payments/db/migration1/migration_validation.go new file mode 100644 index 000000000..e97c3db40 --- /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 000000000..a7a02af5b --- /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("========================================") +}