payments/migration1: add the payments mig code
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.
This commit is contained in:
2 files changed
+1372
No files matched your search
@@ -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
|
||||
}
|
||||
Reference in new issue
Block a user