12282842797

Committed 11 Dec 2024 06:42PM UTC coverage: 49.421% (-0.1%) from 49.54%

Build # 12282842797

Build Type

Pull #8831

github

Committed by

bhandras

Commit Message

docs: update release notes for 0.19.0

Pull Request Pull Request #8831: invoices: migrate KV invoices to native SQL for users of KV SQL backends

Run Details

2 of 496 new or added lines in 5 files covered. (0.4%)

88 existing lines in 16 files now uncovered.

100334 of 203019 relevant lines covered (49.42%)

2.05 hits per line

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

0.0

/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"
)

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")

        // paymentHashIndexBucket 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))

        total := 0
        // Now we can start migrating the invoices. We'll do this in
        // batches to reduce memory usage.
        for {
                t0 = time.Now()
                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")

                        break
                }

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

                offset = queryResult.LastIndexOffset
                total += len(queryResult.Invoices)
                log.Debugf("Migrated %d KV invoices to SQL in %v\n", total,
                        time.Since(t0))
        }

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

lightningnetwork / lnd / 12282842797

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