lnd/payments/db/migration1/migration_validation.go

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/payments/db/migration1/lnwire"
	"github.com/lightningnetwork/lnd/payments/db/migration1/record"
	"github.com/lightningnetwork/lnd/payments/db/migration1/sqlc"
	"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 SQLMigrationQueries,
	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

		// For legacy payments, the HTLC Hash field may be nil in the
		// bbolt backend. During migration, the SQL code uses the
		// parent payment hash as fallback. To ensure the comparison
		// between bbolt and SQL data succeeds, we apply the same
		// fallback here.
		//
		// See also: patchLegacyPaymentHash in payment_lifecycle.go.
		if htlc.Hash == nil && payment.Info != nil {
			htlc.Hash = &payment.Info.PaymentIdentifier
		}

		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{}
			}

			// LegacyPayload was a hint used by the KV store to
			// determine how to serialize and deserialize the hop
			// payload (i.e. whether to use the legacy format or
			// TLV). The SQL store does not serialize hop data at
			// all — each field is stored natively in its own
			// column — so this flag has no meaning there and is
			// never persisted.
			htlc.Route.Hops[j].LegacyPayload = false
		}
	}
}

// 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
}