15574102646

Committed 11 Jun 2025 01:44AM UTC coverage: 68.554% (+9.9%) from 58.637%

Build # 15574102646

Build Type

Pull #9652

github

Committed by

web-flow

Commit Message

Merge eb863e46a into 92a5d35cf

Pull Request Pull Request #9652: lnwallet/chancloser: fix flake in TestRbfCloseClosingNegotiationLocal

Run Details

11 of 12 new or added lines in 1 file covered. (91.67%)

7276 existing lines in 84 files now uncovered.

134508 of 196208 relevant lines covered (68.55%)

44569.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

77.49

/invoices/sql_migration.go

package invoices

import (
        "bytes"
        "context"
        "encoding/binary"
        "errors"
        "fmt"
        "reflect"
        "strconv"
        "time"

        "github.com/davecgh/go-spew/spew"
        "github.com/lightningnetwork/lnd/graph/db/models"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/sqldb"
        "github.com/lightningnetwork/lnd/sqldb/sqlc"
        "github.com/pmezard/go-difflib/difflib"
        "golang.org/x/time/rate"
)

var (
        // invoiceBucket is the name of the bucket within the database that
        // stores all data related to invoices no matter their final state.
        // Within the invoice bucket, each invoice is keyed by its invoice ID
        // which is a monotonically increasing uint32.
        invoiceBucket = []byte("invoices")

        // invoiceIndexBucket  is the name of the sub-bucket within the
        // invoiceBucket which indexes all invoices by their payment hash. The
        // payment hash is the sha256 of the invoice's payment preimage. This
        // index is used to detect duplicates, and also to provide a fast path
        // for looking up incoming HTLCs to determine if we're able to settle
        // them fully.
        //
        // maps: payHash => invoiceKey
        invoiceIndexBucket = []byte("paymenthashes")

        // numInvoicesKey is the name of key which houses the auto-incrementing
        // invoice ID which is essentially used as a primary key. With each
        // invoice inserted, the primary key is incremented by one. This key is
        // stored within the invoiceIndexBucket. Within the invoiceBucket
        // invoices are uniquely identified by the invoice ID.
        numInvoicesKey = []byte("nik")

        // addIndexBucket is an index bucket that we'll use to create a
        // monotonically increasing set of add indexes. Each time we add a new
        // invoice, this sequence number will be incremented and then populated
        // within the new invoice.
        //
        // In addition to this sequence number, we map:
        //
        //   addIndexNo => invoiceKey
        addIndexBucket = []byte("invoice-add-index")

        // ErrMigrationMismatch is returned when the migrated invoice does not
        // match the original invoice.
        ErrMigrationMismatch = fmt.Errorf("migrated invoice does not match " +
                "original invoice")
)

// createInvoiceHashIndex generates a hash index that contains payment hashes
// for each invoice in the database. Retrieving the payment hash for certain
// invoices, such as those created for spontaneous AMP payments, can be
// challenging because the hash is not directly derivable from the invoice's
// parameters and is stored separately in the `paymenthashes` bucket. This
// bucket maps payment hashes to invoice keys, but for migration purposes, we
// need the ability to query in the reverse direction. This function establishes
// a new index in the SQL database that maps each invoice key to its
// corresponding payment hash.
func createInvoiceHashIndex(ctx context.Context, db kvdb.Backend,
        tx *sqlc.Queries) error {

        return db.View(func(kvTx kvdb.RTx) error {
                invoices := kvTx.ReadBucket(invoiceBucket)
                if invoices == nil {
                        return ErrNoInvoicesCreated
                }

                invoiceIndex := invoices.NestedReadBucket(
                        invoiceIndexBucket,
                )
                if invoiceIndex == nil {
                        return ErrNoInvoicesCreated
                }

                addIndex := invoices.NestedReadBucket(addIndexBucket)
                if addIndex == nil {
                        return ErrNoInvoicesCreated
                }

                // First, iterate over all elements in the add index bucket and
                // insert the add index value for the corresponding invoice key
                // in the payment_hashes table.
                err := addIndex.ForEach(func(k, v []byte) error {
                        // The key is the add index, and the value is
                        // the invoice key.
                        addIndexNo := binary.BigEndian.Uint64(k)
                        invoiceKey := binary.BigEndian.Uint32(v)

                        return tx.InsertKVInvoiceKeyAndAddIndex(ctx,
                                sqlc.InsertKVInvoiceKeyAndAddIndexParams{
                                        ID:       int64(invoiceKey),
                                        AddIndex: int64(addIndexNo),
                                },
                        )
                })
                if err != nil {
                        return err
                }

                // Next, iterate over all hashes in the invoice index bucket and
                // set the hash to the corresponding the invoice key in the
                // payment_hashes table.
                return invoiceIndex.ForEach(func(k, v []byte) error {
                        // Skip the special numInvoicesKey as that does
                        // not point to a valid invoice.
                        if bytes.Equal(k, numInvoicesKey) {
                                return nil
                        }

                        // The key is the payment hash, and the value
                        // is the invoice key.
                        if len(k) != lntypes.HashSize {
                                return fmt.Errorf("invalid payment "+
                                        "hash length: expected %v, "+
                                        "got %v", lntypes.HashSize,
                                        len(k))
                        }

                        invoiceKey := binary.BigEndian.Uint32(v)

                        return tx.SetKVInvoicePaymentHash(ctx,
                                sqlc.SetKVInvoicePaymentHashParams{
                                        ID:   int64(invoiceKey),
                                        Hash: k,
                                },
                        )
                })
        }, func() {})
}

// toInsertMigratedInvoiceParams creates the parameters for inserting a migrated
// invoice into the SQL database. The parameters are derived from the original
// invoice insert parameters.
func toInsertMigratedInvoiceParams(
        params sqlc.InsertInvoiceParams) sqlc.InsertMigratedInvoiceParams {

        return sqlc.InsertMigratedInvoiceParams{
                Hash:               params.Hash,
                Preimage:           params.Preimage,
                Memo:               params.Memo,
                AmountMsat:         params.AmountMsat,
                CltvDelta:          params.CltvDelta,
                Expiry:             params.Expiry,
                PaymentAddr:        params.PaymentAddr,
                PaymentRequest:     params.PaymentRequest,
                PaymentRequestHash: params.PaymentRequestHash,
                State:              params.State,
                AmountPaidMsat:     params.AmountPaidMsat,
                IsAmp:              params.IsAmp,
                IsHodl:             params.IsHodl,
                IsKeysend:          params.IsKeysend,
                CreatedAt:          params.CreatedAt,
        }
}

// MigrateSingleInvoice migrates a single invoice to the new SQL schema. Note
// that perfect equality between the old and new schemas is not achievable, as
// the invoice's add index cannot be mapped directly to its ID due to SQL’s
// auto-incrementing primary key. The ID returned from the insert will instead
// serve as the add index in the new schema.
func MigrateSingleInvoice(ctx context.Context, tx SQLInvoiceQueries,
        invoice *Invoice, paymentHash lntypes.Hash) error {

        insertInvoiceParams, err := makeInsertInvoiceParams(
                invoice, paymentHash,
        )
        if err != nil {
                return err
        }

        // Convert the insert invoice parameters to the migrated invoice insert
        // parameters.
        insertMigratedInvoiceParams := toInsertMigratedInvoiceParams(
                insertInvoiceParams,
        )

        // If the invoice is settled, we'll also set the timestamp and the index
        // at which it was settled.
        if invoice.State == ContractSettled {
                if invoice.SettleIndex == 0 {
                        return fmt.Errorf("settled invoice %s missing settle "+
                                "index", paymentHash)
                }

                if invoice.SettleDate.IsZero() {
                        return fmt.Errorf("settled invoice %s missing settle "+
                                "date", paymentHash)
                }

                insertMigratedInvoiceParams.SettleIndex = sqldb.SQLInt64(
                        invoice.SettleIndex,
                )
                insertMigratedInvoiceParams.SettledAt = sqldb.SQLTime(
                        invoice.SettleDate.UTC(),
                )
        }

        // First we need to insert the invoice itself so we can use the "add
        // index" which in this case is the auto incrementing primary key that
        // is returned from the insert.
        invoiceID, err := tx.InsertMigratedInvoice(
                ctx, insertMigratedInvoiceParams,
        )
        if err != nil {
                return fmt.Errorf("unable to insert invoice: %w", err)
        }

        // Insert the invoice's features.
        for feature := range invoice.Terms.Features.Features() {
                params := sqlc.InsertInvoiceFeatureParams{
                        InvoiceID: invoiceID,
                        Feature:   int32(feature),
                }

                err := tx.InsertInvoiceFeature(ctx, params)
                if err != nil {
                        return fmt.Errorf("unable to insert invoice "+
                                "feature(%v): %w", feature, err)
                }
        }

        sqlHtlcIDs := make(map[models.CircuitKey]int64)

        // Now insert the HTLCs of the invoice. We'll also keep track of the SQL
        // ID of each HTLC so we can use it when inserting the AMP sub invoices.
        for circuitKey, htlc := range invoice.Htlcs {
                htlcParams := sqlc.InsertInvoiceHTLCParams{
                        HtlcID: int64(circuitKey.HtlcID),
                        ChanID: strconv.FormatUint(
                                circuitKey.ChanID.ToUint64(), 10,
                        ),
                        AmountMsat:   int64(htlc.Amt),
                        AcceptHeight: int32(htlc.AcceptHeight),
                        AcceptTime:   htlc.AcceptTime.UTC(),
                        ExpiryHeight: int32(htlc.Expiry),
                        State:        int16(htlc.State),
                        InvoiceID:    invoiceID,
                }

                // Leave the MPP amount as NULL if the MPP total amount is zero.
                if htlc.MppTotalAmt != 0 {
                        htlcParams.TotalMppMsat = sqldb.SQLInt64(
                                int64(htlc.MppTotalAmt),
                        )
                }

                // Leave the resolve time as NULL if the HTLC is not resolved.
                if !htlc.ResolveTime.IsZero() {
                        htlcParams.ResolveTime = sqldb.SQLTime(
                                htlc.ResolveTime.UTC(),
                        )
                }

                sqlID, err := tx.InsertInvoiceHTLC(ctx, htlcParams)
                if err != nil {
                        return fmt.Errorf("unable to insert invoice htlc: %w",
                                err)
                }

                sqlHtlcIDs[circuitKey] = sqlID

                // Store custom records.
                for key, value := range htlc.CustomRecords {
                        err = tx.InsertInvoiceHTLCCustomRecord(
                                ctx, sqlc.InsertInvoiceHTLCCustomRecordParams{
                                        Key:    int64(key),
                                        Value:  value,
                                        HtlcID: sqlID,
                                },
                        )
                        if err != nil {
                                return err
                        }
                }
        }

        if !invoice.IsAMP() {
                return nil
        }

        for setID, ampState := range invoice.AMPState {
                // Find the earliest HTLC of the AMP invoice, which will
                // be used as the creation date of this sub invoice.
                var createdAt time.Time
                for circuitKey := range ampState.InvoiceKeys {
                        htlc := invoice.Htlcs[circuitKey]
                        if createdAt.IsZero() {
                                createdAt = htlc.AcceptTime.UTC()
                                continue
                        }

                        if createdAt.After(htlc.AcceptTime) {
                                createdAt = htlc.AcceptTime.UTC()
                        }
                }

                params := sqlc.InsertAMPSubInvoiceParams{
                        SetID:     setID[:],
                        State:     int16(ampState.State),
                        CreatedAt: createdAt,
                        InvoiceID: invoiceID,
                }

                if ampState.SettleIndex != 0 {
                        if ampState.SettleDate.IsZero() {
                                return fmt.Errorf("settled AMP sub invoice %x "+
                                        "missing settle date", setID)
                        }

                        params.SettledAt = sqldb.SQLTime(
                                ampState.SettleDate.UTC(),
                        )

                        params.SettleIndex = sqldb.SQLInt64(
                                ampState.SettleIndex,
                        )
                }

                err := tx.InsertAMPSubInvoice(ctx, params)
                if err != nil {
                        return fmt.Errorf("unable to insert AMP sub invoice: "+
                                "%w", err)
                }

                // Now we can add the AMP HTLCs to the database.
                for circuitKey := range ampState.InvoiceKeys {
                        htlc := invoice.Htlcs[circuitKey]
                        rootShare := htlc.AMP.Record.RootShare()

                        sqlHtlcID, ok := sqlHtlcIDs[circuitKey]
                        if !ok {
                                return fmt.Errorf("missing htlc for AMP htlc: "+
                                        "%v", circuitKey)
                        }

                        params := sqlc.InsertAMPSubInvoiceHTLCParams{
                                InvoiceID:  invoiceID,
                                SetID:      setID[:],
                                HtlcID:     sqlHtlcID,
                                RootShare:  rootShare[:],
                                ChildIndex: int64(htlc.AMP.Record.ChildIndex()),
                                Hash:       htlc.AMP.Hash[:],
                        }

                        if htlc.AMP.Preimage != nil {
                                params.Preimage = htlc.AMP.Preimage[:]
                        }

                        err = tx.InsertAMPSubInvoiceHTLC(ctx, params)
                        if err != nil {
                                return fmt.Errorf("unable to insert AMP sub "+
                                        "invoice: %w", err)
                        }
                }
        }

        return nil
}

// OverrideInvoiceTimeZone overrides the time zone of the invoice to the local
// time zone and chops off the nanosecond part for comparison. This is needed
// because KV database stores times as-is which as an unwanted side effect would
// fail migration due to time comparison expecting both the original and
// migrated invoices to be in the same local time zone and in microsecond
// precision. Note that PostgreSQL stores times in microsecond precision while
// SQLite can store times in nanosecond precision if using TEXT storage class.
func OverrideInvoiceTimeZone(invoice *Invoice) {
        fixTime := func(t time.Time) time.Time {
                return t.In(time.Local).Truncate(time.Microsecond)
        }

        invoice.CreationDate = fixTime(invoice.CreationDate)

        if !invoice.SettleDate.IsZero() {
                invoice.SettleDate = fixTime(invoice.SettleDate)
        }

        if invoice.IsAMP() {
                for setID, ampState := range invoice.AMPState {
                        if ampState.SettleDate.IsZero() {
                                continue
                        }

                        ampState.SettleDate = fixTime(ampState.SettleDate)
                        invoice.AMPState[setID] = ampState
                }
        }

        for _, htlc := range invoice.Htlcs {
                if !htlc.AcceptTime.IsZero() {
                        htlc.AcceptTime = fixTime(htlc.AcceptTime)
                }

                if !htlc.ResolveTime.IsZero() {
                        htlc.ResolveTime = fixTime(htlc.ResolveTime)
                }
        }
}

// MigrateInvoicesToSQL runs the migration of all invoices from the KV database
// to the SQL database. The migration is done in a single transaction to ensure
// that all invoices are migrated or none at all. This function can be run
// multiple times without causing any issues as it will check if the migration
// has already been performed.
func MigrateInvoicesToSQL(ctx context.Context, db kvdb.Backend,
        kvStore InvoiceDB, tx *sqlc.Queries, batchSize int) error {

        log.Infof("Starting migration of invoices from KV to SQL")

        offset := uint64(0)
        t0 := time.Now()

        // Create the hash index which we will use to look up invoice
        // payment hashes by their add index during migration.
        err := createInvoiceHashIndex(ctx, db, tx)
        if err != nil && !errors.Is(err, ErrNoInvoicesCreated) {
                log.Errorf("Unable to create invoice hash index: %v",
                        err)

                return err
        }
        log.Debugf("Created SQL invoice hash index in %v", time.Since(t0))

        s := rate.Sometimes{
                Interval: 30 * time.Second,
        }

        t0 = time.Now()
        chunk := 0
        total := 0

        // Now we can start migrating the invoices. We'll do this in
        // batches to reduce memory usage.
        for {
                query := InvoiceQuery{
                        IndexOffset:    offset,
                        NumMaxInvoices: uint64(batchSize),
                }

                queryResult, err := kvStore.QueryInvoices(ctx, query)
                if err != nil && !errors.Is(err, ErrNoInvoicesCreated) {
                        return fmt.Errorf("unable to query invoices: %w", err)
                }

                if len(queryResult.Invoices) == 0 {
                        log.Infof("All invoices migrated. Total: %d", total)
                        break
                }

                err = migrateInvoices(ctx, tx, queryResult.Invoices)
                if err != nil {
                        return err
                }

                offset = queryResult.LastIndexOffset
                resultCnt := len(queryResult.Invoices)
                total += resultCnt
                chunk += resultCnt

                s.Do(func() {
                        elapsed := time.Since(t0).Seconds()
                        ratePerSec := float64(chunk) / elapsed
                        log.Debugf("Migrated %d invoices (%.2f invoices/sec)",
                                total, ratePerSec)

                        t0 = time.Now()
                        chunk = 0
                })
        }

        // Clean up the hash index as it's no longer needed.
        err = tx.ClearKVInvoiceHashIndex(ctx)
        if err != nil {
                return fmt.Errorf("unable to clear invoice hash "+
                        "index: %w", err)
        }

        log.Infof("Migration of %d invoices from KV to SQL completed", total)

        return nil
}

func migrateInvoices(ctx context.Context, tx *sqlc.Queries,
        invoices []Invoice) error {

        for i, invoice := range invoices {
                var paymentHash lntypes.Hash
                if invoice.Terms.PaymentPreimage != nil {
                        paymentHash = invoice.Terms.PaymentPreimage.Hash()
                } else {
                        paymentHashBytes, err :=
                                tx.GetKVInvoicePaymentHashByAddIndex(
                                        ctx, int64(invoice.AddIndex),
                                )
                        if err != nil {
                                // This would be an unexpected inconsistency
                                // in the kv database. We can't do much here
                                // so we'll notify the user and continue.
                                log.Warnf("Cannot migrate invoice, unable to "+
                                        "fetch payment hash (add_index=%v): %v",
                                        invoice.AddIndex, err)

                                continue
                        }

                        copy(paymentHash[:], paymentHashBytes)
                }

                err := MigrateSingleInvoice(ctx, tx, &invoices[i], paymentHash)
                if err != nil {
                        return fmt.Errorf("unable to migrate invoice(%v): %w",
                                paymentHash, err)
                }

                migratedInvoice, err := fetchInvoice(
                        ctx, tx, InvoiceRefByHash(paymentHash),
                )
                if err != nil {
                        return fmt.Errorf("unable to fetch migrated "+
                                "invoice(%v): %w", paymentHash, err)
                }

                // Override the time zone for comparison. Note that we need to
                // override both invoices as the original invoice is coming from
                // KV database, it was stored as a binary serialized Go
                // time.Time value which has nanosecond precision but might have
                // been created in a different time zone. The migrated invoice
                // is stored in SQL in UTC and selected in the local time zone,
                // however in PostgreSQL it has microsecond precision while in
                // SQLite it has nanosecond precision if using TEXT storage
                // class.
                OverrideInvoiceTimeZone(&invoice)
                OverrideInvoiceTimeZone(migratedInvoice)

                // Override the add index before checking for equality.
                migratedInvoice.AddIndex = invoice.AddIndex

                if !reflect.DeepEqual(invoice, *migratedInvoice) {
                        diff := difflib.UnifiedDiff{
                                A: difflib.SplitLines(
                                        spew.Sdump(invoice),
                                ),
                                B: difflib.SplitLines(
                                        spew.Sdump(migratedInvoice),
                                ),
                                FromFile: "Expected",
                                FromDate: "",
                                ToFile:   "Actual",
                                ToDate:   "",
                                Context:  3,
                        }
                        diffText, _ := difflib.GetUnifiedDiffString(diff)

                        return fmt.Errorf("%w: %v.\n%v", ErrMigrationMismatch,
                                paymentHash, diffText)
                }
        }

        return nil
}

1	package invoices
2
3	import (
4	"bytes"
5	"context"
6	"encoding/binary"
7	"errors"
8	"fmt"
9	"reflect"
10	"strconv"
11	"time"
12
13	"github.com/davecgh/go-spew/spew"
14	"github.com/lightningnetwork/lnd/graph/db/models"
15	"github.com/lightningnetwork/lnd/kvdb"
16	"github.com/lightningnetwork/lnd/lntypes"
17	"github.com/lightningnetwork/lnd/sqldb"
18	"github.com/lightningnetwork/lnd/sqldb/sqlc"
19	"github.com/pmezard/go-difflib/difflib"
20	"golang.org/x/time/rate"
21	)
22
23	var (
24	// invoiceBucket is the name of the bucket within the database that
25	// stores all data related to invoices no matter their final state.
26	// Within the invoice bucket, each invoice is keyed by its invoice ID
27	// which is a monotonically increasing uint32.
28	invoiceBucket = []byte("invoices")
29
30	// invoiceIndexBucket is the name of the sub-bucket within the
31	// invoiceBucket which indexes all invoices by their payment hash. The
32	// payment hash is the sha256 of the invoice's payment preimage. This
33	// index is used to detect duplicates, and also to provide a fast path
34	// for looking up incoming HTLCs to determine if we're able to settle
35	// them fully.
36	//
37	// maps: payHash => invoiceKey
38	invoiceIndexBucket = []byte("paymenthashes")
39
40	// numInvoicesKey is the name of key which houses the auto-incrementing
41	// invoice ID which is essentially used as a primary key. With each
42	// invoice inserted, the primary key is incremented by one. This key is
43	// stored within the invoiceIndexBucket. Within the invoiceBucket
44	// invoices are uniquely identified by the invoice ID.
45	numInvoicesKey = []byte("nik")
46
47	// addIndexBucket is an index bucket that we'll use to create a
48	// monotonically increasing set of add indexes. Each time we add a new
49	// invoice, this sequence number will be incremented and then populated
50	// within the new invoice.
51	//
52	// In addition to this sequence number, we map:
53	//
54	// addIndexNo => invoiceKey
55	addIndexBucket = []byte("invoice-add-index")
56
57	// ErrMigrationMismatch is returned when the migrated invoice does not
58	// match the original invoice.
59	ErrMigrationMismatch = fmt.Errorf("migrated invoice does not match " +
60	"original invoice")
61	)
62
63	// createInvoiceHashIndex generates a hash index that contains payment hashes
64	// for each invoice in the database. Retrieving the payment hash for certain
65	// invoices, such as those created for spontaneous AMP payments, can be
66	// challenging because the hash is not directly derivable from the invoice's
67	// parameters and is stored separately in the `paymenthashes` bucket. This
68	// bucket maps payment hashes to invoice keys, but for migration purposes, we
69	// need the ability to query in the reverse direction. This function establishes
70	// a new index in the SQL database that maps each invoice key to its
71	// corresponding payment hash.
72	func createInvoiceHashIndex(ctx context.Context, db kvdb.Backend,
73	tx *sqlc.Queries) error {	8✔
74		8✔
75	return db.View(func(kvTx kvdb.RTx) error {	16✔
76	invoices := kvTx.ReadBucket(invoiceBucket)	8✔
77	if invoices == nil {	8✔
78	return ErrNoInvoicesCreated	×
UNCOV 79	}	×
80
81	invoiceIndex := invoices.NestedReadBucket(	8✔
82	invoiceIndexBucket,	8✔
83	)	8✔
84	if invoiceIndex == nil {	12✔
85	return ErrNoInvoicesCreated	4✔
86	}	4✔
87
88	addIndex := invoices.NestedReadBucket(addIndexBucket)	4✔
89	if addIndex == nil {	4✔
90	return ErrNoInvoicesCreated	×
UNCOV 91	}	×
92
93	// First, iterate over all elements in the add index bucket and
94	// insert the add index value for the corresponding invoice key
95	// in the payment_hashes table.
96	err := addIndex.ForEach(func(k, v []byte) error {	12✔
97	// The key is the add index, and the value is	8✔
98	// the invoice key.	8✔
99	addIndexNo := binary.BigEndian.Uint64(k)	8✔
100	invoiceKey := binary.BigEndian.Uint32(v)	8✔
101		8✔
102	return tx.InsertKVInvoiceKeyAndAddIndex(ctx,	8✔
103	sqlc.InsertKVInvoiceKeyAndAddIndexParams{	8✔
104	ID: int64(invoiceKey),	8✔
105	AddIndex: int64(addIndexNo),	8✔
106	},	8✔
107	)	8✔
108	})	8✔
109	if err != nil {	4✔
110	return err	×
UNCOV 111	}	×
112
113	// Next, iterate over all hashes in the invoice index bucket and
114	// set the hash to the corresponding the invoice key in the
115	// payment_hashes table.
116	return invoiceIndex.ForEach(func(k, v []byte) error {	16✔
117	// Skip the special numInvoicesKey as that does	12✔
118	// not point to a valid invoice.	12✔
119	if bytes.Equal(k, numInvoicesKey) {	16✔
120	return nil	4✔
121	}	4✔
122
123	// The key is the payment hash, and the value
124	// is the invoice key.
125	if len(k) != lntypes.HashSize {	8✔
126	return fmt.Errorf("invalid payment "+	×
127	"hash length: expected %v, "+	×
128	"got %v", lntypes.HashSize,	×
129	len(k))	×
UNCOV 130	}	×
131
132	invoiceKey := binary.BigEndian.Uint32(v)	8✔
133		8✔
134	return tx.SetKVInvoicePaymentHash(ctx,	8✔
135	sqlc.SetKVInvoicePaymentHashParams{	8✔
136	ID: int64(invoiceKey),	8✔
137	Hash: k,	8✔
138	},	8✔
139	)	8✔
140	})
141	}, func() {})	8✔
142	}
143
144	// toInsertMigratedInvoiceParams creates the parameters for inserting a migrated
145	// invoice into the SQL database. The parameters are derived from the original
146	// invoice insert parameters.
147	func toInsertMigratedInvoiceParams(
148	params sqlc.InsertInvoiceParams) sqlc.InsertMigratedInvoiceParams {	40,008✔
149		40,008✔
150	return sqlc.InsertMigratedInvoiceParams{	40,008✔
151	Hash: params.Hash,	40,008✔
152	Preimage: params.Preimage,	40,008✔
153	Memo: params.Memo,	40,008✔
154	AmountMsat: params.AmountMsat,	40,008✔
155	CltvDelta: params.CltvDelta,	40,008✔
156	Expiry: params.Expiry,	40,008✔
157	PaymentAddr: params.PaymentAddr,	40,008✔
158	PaymentRequest: params.PaymentRequest,	40,008✔
159	PaymentRequestHash: params.PaymentRequestHash,	40,008✔
160	State: params.State,	40,008✔
161	AmountPaidMsat: params.AmountPaidMsat,	40,008✔
162	IsAmp: params.IsAmp,	40,008✔
163	IsHodl: params.IsHodl,	40,008✔
164	IsKeysend: params.IsKeysend,	40,008✔
165	CreatedAt: params.CreatedAt,	40,008✔
166	}	40,008✔
167	}	40,008✔
168
169	// MigrateSingleInvoice migrates a single invoice to the new SQL schema. Note
170	// that perfect equality between the old and new schemas is not achievable, as
171	// the invoice's add index cannot be mapped directly to its ID due to SQL’s
172	// auto-incrementing primary key. The ID returned from the insert will instead
173	// serve as the add index in the new schema.
174	func MigrateSingleInvoice(ctx context.Context, tx SQLInvoiceQueries,
175	invoice *Invoice, paymentHash lntypes.Hash) error {	40,008✔
176		40,008✔
177	insertInvoiceParams, err := makeInsertInvoiceParams(	40,008✔
178	invoice, paymentHash,	40,008✔
179	)	40,008✔
180	if err != nil {	40,008✔
181	return err	×
UNCOV 182	}	×
183
184	// Convert the insert invoice parameters to the migrated invoice insert
185	// parameters.
186	insertMigratedInvoiceParams := toInsertMigratedInvoiceParams(	40,008✔
187	insertInvoiceParams,	40,008✔
188	)	40,008✔
189		40,008✔
190	// If the invoice is settled, we'll also set the timestamp and the index	40,008✔
191	// at which it was settled.	40,008✔
192	if invoice.State == ContractSettled {	50,804✔
193	if invoice.SettleIndex == 0 {	10,796✔
194	return fmt.Errorf("settled invoice %s missing settle "+	×
195	"index", paymentHash)	×
UNCOV 196	}	×
197
198	if invoice.SettleDate.IsZero() {	10,796✔
199	return fmt.Errorf("settled invoice %s missing settle "+	×
200	"date", paymentHash)	×
UNCOV 201	}	×
202
203	insertMigratedInvoiceParams.SettleIndex = sqldb.SQLInt64(	10,796✔
204	invoice.SettleIndex,	10,796✔
205	)	10,796✔
206	insertMigratedInvoiceParams.SettledAt = sqldb.SQLTime(	10,796✔
207	invoice.SettleDate.UTC(),	10,796✔
208	)	10,796✔
209	}
210
211	// First we need to insert the invoice itself so we can use the "add
212	// index" which in this case is the auto incrementing primary key that
213	// is returned from the insert.
214	invoiceID, err := tx.InsertMigratedInvoice(	40,008✔
215	ctx, insertMigratedInvoiceParams,	40,008✔
216	)	40,008✔
217	if err != nil {	40,008✔
218	return fmt.Errorf("unable to insert invoice: %w", err)	×
UNCOV 219	}	×
220
221	// Insert the invoice's features.
222	for feature := range invoice.Terms.Features.Features() {	69,232✔
223	params := sqlc.InsertInvoiceFeatureParams{	29,224✔
224	InvoiceID: invoiceID,	29,224✔
225	Feature: int32(feature),	29,224✔
226	}	29,224✔
227		29,224✔
228	err := tx.InsertInvoiceFeature(ctx, params)	29,224✔
229	if err != nil {	29,224✔
230	return fmt.Errorf("unable to insert invoice "+	×
231	"feature(%v): %w", feature, err)	×
UNCOV 232	}	×
233	}
234
235	sqlHtlcIDs := make(map[models.CircuitKey]int64)	40,008✔
236		40,008✔
237	// Now insert the HTLCs of the invoice. We'll also keep track of the SQL	40,008✔
238	// ID of each HTLC so we can use it when inserting the AMP sub invoices.	40,008✔
239	for circuitKey, htlc := range invoice.Htlcs {	240,818✔
240	htlcParams := sqlc.InsertInvoiceHTLCParams{	200,810✔
241	HtlcID: int64(circuitKey.HtlcID),	200,810✔
242	ChanID: strconv.FormatUint(	200,810✔
243	circuitKey.ChanID.ToUint64(), 10,	200,810✔
244	),	200,810✔
245	AmountMsat: int64(htlc.Amt),	200,810✔
246	AcceptHeight: int32(htlc.AcceptHeight),	200,810✔
247	AcceptTime: htlc.AcceptTime.UTC(),	200,810✔
248	ExpiryHeight: int32(htlc.Expiry),	200,810✔
249	State: int16(htlc.State),	200,810✔
250	InvoiceID: invoiceID,	200,810✔
251	}	200,810✔
252		200,810✔
253	// Leave the MPP amount as NULL if the MPP total amount is zero.	200,810✔
254	if htlc.MppTotalAmt != 0 {	233,097✔
255	htlcParams.TotalMppMsat = sqldb.SQLInt64(	32,287✔
256	int64(htlc.MppTotalAmt),	32,287✔
257	)	32,287✔
258	}	32,287✔
259
260	// Leave the resolve time as NULL if the HTLC is not resolved.
261	if !htlc.ResolveTime.IsZero() {	305,992✔
262	htlcParams.ResolveTime = sqldb.SQLTime(	105,182✔
263	htlc.ResolveTime.UTC(),	105,182✔
264	)	105,182✔
265	}	105,182✔
266
267	sqlID, err := tx.InsertInvoiceHTLC(ctx, htlcParams)	200,810✔
268	if err != nil {	200,810✔
269	return fmt.Errorf("unable to insert invoice htlc: %w",	×
270	err)	×
UNCOV 271	}	×
272
273	sqlHtlcIDs[circuitKey] = sqlID	200,810✔
274		200,810✔
275	// Store custom records.	200,810✔
276	for key, value := range htlc.CustomRecords {	636,788✔
277	err = tx.InsertInvoiceHTLCCustomRecord(	435,978✔
278	ctx, sqlc.InsertInvoiceHTLCCustomRecordParams{	435,978✔
279	Key: int64(key),	435,978✔
280	Value: value,	435,978✔
281	HtlcID: sqlID,	435,978✔
282	},	435,978✔
283	)	435,978✔
284	if err != nil {	435,978✔
285	return err	×
UNCOV 286	}	×
287	}
288	}
289
290	if !invoice.IsAMP() {	60,008✔
291	return nil	20,000✔
292	}	20,000✔
293
294	for setID, ampState := range invoice.AMPState {	76,553✔
295	// Find the earliest HTLC of the AMP invoice, which will	56,545✔
296	// be used as the creation date of this sub invoice.	56,545✔
297	var createdAt time.Time	56,545✔
298	for circuitKey := range ampState.InvoiceKeys {	217,157✔
299	htlc := invoice.Htlcs[circuitKey]	160,612✔
300	if createdAt.IsZero() {	217,157✔
301	createdAt = htlc.AcceptTime.UTC()	56,545✔
302	continue	56,545✔
303	}
304
305	if createdAt.After(htlc.AcceptTime) {	143,225✔
306	createdAt = htlc.AcceptTime.UTC()	39,158✔
307	}	39,158✔
308	}
309
310	params := sqlc.InsertAMPSubInvoiceParams{	56,545✔
311	SetID: setID[:],	56,545✔
312	State: int16(ampState.State),	56,545✔
313	CreatedAt: createdAt,	56,545✔
314	InvoiceID: invoiceID,	56,545✔
315	}	56,545✔
316		56,545✔
317	if ampState.SettleIndex != 0 {	71,818✔
318	if ampState.SettleDate.IsZero() {	15,273✔
319	return fmt.Errorf("settled AMP sub invoice %x "+	×
320	"missing settle date", setID)	×
UNCOV 321	}	×
322
323	params.SettledAt = sqldb.SQLTime(	15,273✔
324	ampState.SettleDate.UTC(),	15,273✔
325	)	15,273✔
326		15,273✔
327	params.SettleIndex = sqldb.SQLInt64(	15,273✔
328	ampState.SettleIndex,	15,273✔
329	)	15,273✔
330	}
331
332	err := tx.InsertAMPSubInvoice(ctx, params)	56,545✔
333	if err != nil {	56,545✔
334	return fmt.Errorf("unable to insert AMP sub invoice: "+	×
335	"%w", err)	×
UNCOV 336	}	×
337
338	// Now we can add the AMP HTLCs to the database.
339	for circuitKey := range ampState.InvoiceKeys {	217,157✔
340	htlc := invoice.Htlcs[circuitKey]	160,612✔
341	rootShare := htlc.AMP.Record.RootShare()	160,612✔
342		160,612✔
343	sqlHtlcID, ok := sqlHtlcIDs[circuitKey]	160,612✔
344	if !ok {	160,612✔
345	return fmt.Errorf("missing htlc for AMP htlc: "+	×
346	"%v", circuitKey)	×
UNCOV 347	}	×
348
349	params := sqlc.InsertAMPSubInvoiceHTLCParams{	160,612✔
350	InvoiceID: invoiceID,	160,612✔
351	SetID: setID[:],	160,612✔
352	HtlcID: sqlHtlcID,	160,612✔
353	RootShare: rootShare[:],	160,612✔
354	ChildIndex: int64(htlc.AMP.Record.ChildIndex()),	160,612✔
355	Hash: htlc.AMP.Hash[:],	160,612✔
356	}	160,612✔
357		160,612✔
358	if htlc.AMP.Preimage != nil {	321,224✔
359	params.Preimage = htlc.AMP.Preimage[:]	160,612✔
360	}	160,612✔
361
362	err = tx.InsertAMPSubInvoiceHTLC(ctx, params)	160,612✔
363	if err != nil {	160,612✔
364	return fmt.Errorf("unable to insert AMP sub "+	×
365	"invoice: %w", err)	×
UNCOV 366	}	×
367	}
368	}
369
370	return nil	20,008✔
371	}
372
373	// OverrideInvoiceTimeZone overrides the time zone of the invoice to the local
374	// time zone and chops off the nanosecond part for comparison. This is needed
375	// because KV database stores times as-is which as an unwanted side effect would
376	// fail migration due to time comparison expecting both the original and
377	// migrated invoices to be in the same local time zone and in microsecond
378	// precision. Note that PostgreSQL stores times in microsecond precision while
379	// SQLite can store times in nanosecond precision if using TEXT storage class.
380	func OverrideInvoiceTimeZone(invoice *Invoice) {	80,032✔
381	fixTime := func(t time.Time) time.Time {	824,242✔
382	return t.In(time.Local).Truncate(time.Microsecond)	744,210✔
383	}	744,210✔
384
385	invoice.CreationDate = fixTime(invoice.CreationDate)	80,032✔
386		80,032✔
387	if !invoice.SettleDate.IsZero() {	101,624✔
388	invoice.SettleDate = fixTime(invoice.SettleDate)	21,592✔
389	}	21,592✔
390
391	if invoice.IsAMP() {	120,064✔
392	for setID, ampState := range invoice.AMPState {	153,130✔
393	if ampState.SettleDate.IsZero() {	195,642✔
394	continue	82,544✔
395	}
396
397	ampState.SettleDate = fixTime(ampState.SettleDate)	30,554✔
398	invoice.AMPState[setID] = ampState	30,554✔
399	}
400	}
401
402	for _, htlc := range invoice.Htlcs {	481,676✔
403	if !htlc.AcceptTime.IsZero() {	803,288✔
404	htlc.AcceptTime = fixTime(htlc.AcceptTime)	401,644✔
405	}	401,644✔
406
407	if !htlc.ResolveTime.IsZero() {	612,032✔
408	htlc.ResolveTime = fixTime(htlc.ResolveTime)	210,388✔
409	}	210,388✔
410	}
411	}
412
413	// MigrateInvoicesToSQL runs the migration of all invoices from the KV database
414	// to the SQL database. The migration is done in a single transaction to ensure
415	// that all invoices are migrated or none at all. This function can be run
416	// multiple times without causing any issues as it will check if the migration
417	// has already been performed.
418	func MigrateInvoicesToSQL(ctx context.Context, db kvdb.Backend,
419	kvStore InvoiceDB, tx *sqlc.Queries, batchSize int) error {	8✔
420		8✔
421	log.Infof("Starting migration of invoices from KV to SQL")	8✔
422		8✔
423	offset := uint64(0)	8✔
424	t0 := time.Now()	8✔
425		8✔
426	// Create the hash index which we will use to look up invoice	8✔
427	// payment hashes by their add index during migration.	8✔
428	err := createInvoiceHashIndex(ctx, db, tx)	8✔
429	if err != nil && !errors.Is(err, ErrNoInvoicesCreated) {	8✔
430	log.Errorf("Unable to create invoice hash index: %v",	×
431	err)	×
432		×
433	return err	×
434	}	×
435	log.Debugf("Created SQL invoice hash index in %v", time.Since(t0))	8✔
436		8✔
437	s := rate.Sometimes{	8✔
438	Interval: 30 * time.Second,	8✔
439	}	8✔
440		8✔
441	t0 = time.Now()	8✔
442	chunk := 0	8✔
443	total := 0	8✔
444		8✔
445	// Now we can start migrating the invoices. We'll do this in	8✔
446	// batches to reduce memory usage.	8✔
447	for {	20✔
448	query := InvoiceQuery{	12✔
449	IndexOffset: offset,	12✔
450	NumMaxInvoices: uint64(batchSize),	12✔
451	}	12✔
452		12✔
453	queryResult, err := kvStore.QueryInvoices(ctx, query)	12✔
454	if err != nil && !errors.Is(err, ErrNoInvoicesCreated) {	12✔
455	return fmt.Errorf("unable to query invoices: %w", err)	×
UNCOV 456	}	×
457
458	if len(queryResult.Invoices) == 0 {	20✔
459	log.Infof("All invoices migrated. Total: %d", total)	8✔
460	break	8✔
461	}
462
463	err = migrateInvoices(ctx, tx, queryResult.Invoices)	4✔
464	if err != nil {	4✔
465	return err	×
466	}	×
467
468	offset = queryResult.LastIndexOffset	4✔
469	resultCnt := len(queryResult.Invoices)	4✔
470	total += resultCnt	4✔
471	chunk += resultCnt	4✔
472		4✔
473	s.Do(func() {	8✔
474	elapsed := time.Since(t0).Seconds()	4✔
475	ratePerSec := float64(chunk) / elapsed	4✔
476	log.Debugf("Migrated %d invoices (%.2f invoices/sec)",	4✔
477	total, ratePerSec)	4✔
478		4✔
479	t0 = time.Now()	4✔
480	chunk = 0	4✔
481	})	4✔
482	}
483
484	// Clean up the hash index as it's no longer needed.
485	err = tx.ClearKVInvoiceHashIndex(ctx)	8✔
486	if err != nil {	8✔
487	return fmt.Errorf("unable to clear invoice hash "+	×
488	"index: %w", err)	×
489	}	×
490
491	log.Infof("Migration of %d invoices from KV to SQL completed", total)	8✔
492		8✔
493	return nil	8✔
494	}
495
496	func migrateInvoices(ctx context.Context, tx *sqlc.Queries,
497	invoices []Invoice) error {	4✔
498		4✔
499	for i, invoice := range invoices {	12✔
500	var paymentHash lntypes.Hash	8✔
501	if invoice.Terms.PaymentPreimage != nil {	8✔
UNCOV 502	paymentHash = invoice.Terms.PaymentPreimage.Hash()	×
503	} else {	8✔
504	paymentHashBytes, err :=	8✔
505	tx.GetKVInvoicePaymentHashByAddIndex(	8✔
506	ctx, int64(invoice.AddIndex),	8✔
507	)	8✔
508	if err != nil {	8✔
509	// This would be an unexpected inconsistency	×
510	// in the kv database. We can't do much here	×
511	// so we'll notify the user and continue.	×
UNCOV 512	log.Warnf("Cannot migrate invoice, unable to "+	×
513	"fetch payment hash (add_index=%v): %v",	×
514	invoice.AddIndex, err)	×
515		×
516	continue	×
517	}
518
519	copy(paymentHash[:], paymentHashBytes)	8✔
520	}
521
522	err := MigrateSingleInvoice(ctx, tx, &invoices[i], paymentHash)	8✔
523	if err != nil {	8✔
UNCOV 524	return fmt.Errorf("unable to migrate invoice(%v): %w",	×
UNCOV 525	paymentHash, err)	×
UNCOV 526	}	×
527
528	migratedInvoice, err := fetchInvoice(	8✔
529	ctx, tx, InvoiceRefByHash(paymentHash),	8✔
530	)	8✔
531	if err != nil {	8✔
532	return fmt.Errorf("unable to fetch migrated "+	×
533	"invoice(%v): %w", paymentHash, err)	×
534	}	×
535
536	// Override the time zone for comparison. Note that we need to
537	// override both invoices as the original invoice is coming from
538	// KV database, it was stored as a binary serialized Go
539	// time.Time value which has nanosecond precision but might have
540	// been created in a different time zone. The migrated invoice
541	// is stored in SQL in UTC and selected in the local time zone,
542	// however in PostgreSQL it has microsecond precision while in
543	// SQLite it has nanosecond precision if using TEXT storage
544	// class.
545	OverrideInvoiceTimeZone(&invoice)	8✔
546	OverrideInvoiceTimeZone(migratedInvoice)	8✔
547		8✔
548	// Override the add index before checking for equality.	8✔
549	migratedInvoice.AddIndex = invoice.AddIndex	8✔
550		8✔
551	if !reflect.DeepEqual(invoice, *migratedInvoice) {	8✔
552	diff := difflib.UnifiedDiff{	×
553	A: difflib.SplitLines(	×
554	spew.Sdump(invoice),	×
UNCOV 555	),	×
UNCOV 556	B: difflib.SplitLines(	×
557	spew.Sdump(migratedInvoice),	×
UNCOV 558	),	×
UNCOV 559	FromFile: "Expected",	×
UNCOV 560	FromDate: "",	×
UNCOV 561	ToFile: "Actual",	×
UNCOV 562	ToDate: "",	×
UNCOV 563	Context: 3,	×
UNCOV 564	}	×
UNCOV 565	diffText, _ := difflib.GetUnifiedDiffString(diff)	×
UNCOV 566		×
UNCOV 567	return fmt.Errorf("%w: %v.\n%v", ErrMigrationMismatch,	×
UNCOV 568	paymentHash, diffText)	×
UNCOV 569	}	×
570	}
571
572	return nil	4✔
573	}

lightningnetwork / lnd / 15574102646

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous