13211764208

Committed 08 Feb 2025 03:08AM UTC coverage: 49.288% (-9.5%) from 58.815%

Build # 13211764208

Build Type

Pull #9489

github

Committed by

calvinrzachman

Commit Message

itest: verify switchrpc server enforces send then track

We prevent the rpc server from allowing onion dispatches for
attempt IDs which have already been tracked by rpc clients.

This helps protect the client from leaking a duplicate onion
attempt. NOTE: This is not the only method for solving this
issue! The issue could be addressed via careful client side
programming which accounts for the uncertainty and async
nature of dispatching onions to a remote process via RPC.
This would require some lnd ChannelRouter changes for how
we intend to use these RPCs though.

Pull Request Pull Request #9489: multi: add BuildOnion, SendOnion, and TrackOnion RPCs

Run Details

474 of 990 new or added lines in 11 files covered. (47.88%)

27321 existing lines in 435 files now uncovered.

101192 of 205306 relevant lines covered (49.29%)

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

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

        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	// invoiceIndexBucket 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,
UNCOV 72	tx *sqlc.Queries) error {	×
UNCOV 73		×
UNCOV 74	return db.View(func(kvTx kvdb.RTx) error {	×
UNCOV 75	invoices := kvTx.ReadBucket(invoiceBucket)	×
UNCOV 76	if invoices == nil {	×
77	return ErrNoInvoicesCreated	×
78	}	×
79
UNCOV 80	invoiceIndex := invoices.NestedReadBucket(	×
UNCOV 81	invoiceIndexBucket,	×
UNCOV 82	)	×
UNCOV 83	if invoiceIndex == nil {	×
UNCOV 84	return ErrNoInvoicesCreated	×
UNCOV 85	}	×
86
UNCOV 87	addIndex := invoices.NestedReadBucket(addIndexBucket)	×
UNCOV 88	if addIndex == nil {	×
89	return ErrNoInvoicesCreated	×
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.
UNCOV 95	err := addIndex.ForEach(func(k, v []byte) error {	×
UNCOV 96	// The key is the add index, and the value is	×
UNCOV 97	// the invoice key.	×
UNCOV 98	addIndexNo := binary.BigEndian.Uint64(k)	×
UNCOV 99	invoiceKey := binary.BigEndian.Uint32(v)	×
UNCOV 100		×
UNCOV 101	return tx.InsertKVInvoiceKeyAndAddIndex(ctx,	×
UNCOV 102	sqlc.InsertKVInvoiceKeyAndAddIndexParams{	×
UNCOV 103	ID: int64(invoiceKey),	×
UNCOV 104	AddIndex: int64(addIndexNo),	×
UNCOV 105	},	×
UNCOV 106	)	×
UNCOV 107	})	×
UNCOV 108	if err != nil {	×
109	return err	×
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.
UNCOV 115	return invoiceIndex.ForEach(func(k, v []byte) error {	×
UNCOV 116	// Skip the special numInvoicesKey as that does	×
UNCOV 117	// not point to a valid invoice.	×
UNCOV 118	if bytes.Equal(k, numInvoicesKey) {	×
UNCOV 119	return nil	×
UNCOV 120	}	×
121
122	// The key is the payment hash, and the value
123	// is the invoice key.
UNCOV 124	if len(k) != lntypes.HashSize {	×
125	return fmt.Errorf("invalid payment "+	×
126	"hash length: expected %v, "+	×
127	"got %v", lntypes.HashSize,	×
128	len(k))	×
129	}	×
130
UNCOV 131	invoiceKey := binary.BigEndian.Uint32(v)	×
UNCOV 132		×
UNCOV 133	return tx.SetKVInvoicePaymentHash(ctx,	×
UNCOV 134	sqlc.SetKVInvoicePaymentHashParams{	×
UNCOV 135	ID: int64(invoiceKey),	×
UNCOV 136	Hash: k,	×
UNCOV 137	},	×
UNCOV 138	)	×
139	})
UNCOV 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(
UNCOV 147	params sqlc.InsertInvoiceParams) sqlc.InsertMigratedInvoiceParams {	×
UNCOV 148		×
UNCOV 149	return sqlc.InsertMigratedInvoiceParams{	×
UNCOV 150	Hash: params.Hash,	×
UNCOV 151	Preimage: params.Preimage,	×
UNCOV 152	Memo: params.Memo,	×
UNCOV 153	AmountMsat: params.AmountMsat,	×
UNCOV 154	CltvDelta: params.CltvDelta,	×
UNCOV 155	Expiry: params.Expiry,	×
UNCOV 156	PaymentAddr: params.PaymentAddr,	×
UNCOV 157	PaymentRequest: params.PaymentRequest,	×
UNCOV 158	PaymentRequestHash: params.PaymentRequestHash,	×
UNCOV 159	State: params.State,	×
UNCOV 160	AmountPaidMsat: params.AmountPaidMsat,	×
UNCOV 161	IsAmp: params.IsAmp,	×
UNCOV 162	IsHodl: params.IsHodl,	×
UNCOV 163	IsKeysend: params.IsKeysend,	×
UNCOV 164	CreatedAt: params.CreatedAt,	×
UNCOV 165	}	×
UNCOV 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,
UNCOV 174	invoice *Invoice, paymentHash lntypes.Hash) error {	×
UNCOV 175		×
UNCOV 176	insertInvoiceParams, err := makeInsertInvoiceParams(	×
UNCOV 177	invoice, paymentHash,	×
UNCOV 178	)	×
UNCOV 179	if err != nil {	×
180	return err	×
181	}	×
182
183	// Convert the insert invoice parameters to the migrated invoice insert
184	// parameters.
UNCOV 185	insertMigratedInvoiceParams := toInsertMigratedInvoiceParams(	×
UNCOV 186	insertInvoiceParams,	×
UNCOV 187	)	×
UNCOV 188		×
UNCOV 189	// If the invoice is settled, we'll also set the timestamp and the index	×
UNCOV 190	// at which it was settled.	×
UNCOV 191	if invoice.State == ContractSettled {	×
UNCOV 192	if invoice.SettleIndex == 0 {	×
193	return fmt.Errorf("settled invoice %s missing settle "+	×
194	"index", paymentHash)	×
195	}	×
196
UNCOV 197	if invoice.SettleDate.IsZero() {	×
198	return fmt.Errorf("settled invoice %s missing settle "+	×
199	"date", paymentHash)	×
200	}	×
201
UNCOV 202	insertMigratedInvoiceParams.SettleIndex = sqldb.SQLInt64(	×
UNCOV 203	invoice.SettleIndex,	×
UNCOV 204	)	×
UNCOV 205	insertMigratedInvoiceParams.SettledAt = sqldb.SQLTime(	×
UNCOV 206	invoice.SettleDate.UTC(),	×
UNCOV 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.
UNCOV 213	invoiceID, err := tx.InsertMigratedInvoice(	×
UNCOV 214	ctx, insertMigratedInvoiceParams,	×
UNCOV 215	)	×
UNCOV 216	if err != nil {	×
217	return fmt.Errorf("unable to insert invoice: %w", err)	×
218	}	×
219
220	// Insert the invoice's features.
UNCOV 221	for feature := range invoice.Terms.Features.Features() {	×
UNCOV 222	params := sqlc.InsertInvoiceFeatureParams{	×
UNCOV 223	InvoiceID: invoiceID,	×
UNCOV 224	Feature: int32(feature),	×
UNCOV 225	}	×
UNCOV 226		×
UNCOV 227	err := tx.InsertInvoiceFeature(ctx, params)	×
UNCOV 228	if err != nil {	×
229	return fmt.Errorf("unable to insert invoice "+	×
230	"feature(%v): %w", feature, err)	×
231	}	×
232	}
233
UNCOV 234	sqlHtlcIDs := make(map[models.CircuitKey]int64)	×
UNCOV 235		×
UNCOV 236	// Now insert the HTLCs of the invoice. We'll also keep track of the SQL	×
UNCOV 237	// ID of each HTLC so we can use it when inserting the AMP sub invoices.	×
UNCOV 238	for circuitKey, htlc := range invoice.Htlcs {	×
UNCOV 239	htlcParams := sqlc.InsertInvoiceHTLCParams{	×
UNCOV 240	HtlcID: int64(circuitKey.HtlcID),	×
UNCOV 241	ChanID: strconv.FormatUint(	×
UNCOV 242	circuitKey.ChanID.ToUint64(), 10,	×
UNCOV 243	),	×
UNCOV 244	AmountMsat: int64(htlc.Amt),	×
UNCOV 245	AcceptHeight: int32(htlc.AcceptHeight),	×
UNCOV 246	AcceptTime: htlc.AcceptTime.UTC(),	×
UNCOV 247	ExpiryHeight: int32(htlc.Expiry),	×
UNCOV 248	State: int16(htlc.State),	×
UNCOV 249	InvoiceID: invoiceID,	×
UNCOV 250	}	×
UNCOV 251		×
UNCOV 252	// Leave the MPP amount as NULL if the MPP total amount is zero.	×
UNCOV 253	if htlc.MppTotalAmt != 0 {	×
UNCOV 254	htlcParams.TotalMppMsat = sqldb.SQLInt64(	×
UNCOV 255	int64(htlc.MppTotalAmt),	×
UNCOV 256	)	×
UNCOV 257	}	×
258
259	// Leave the resolve time as NULL if the HTLC is not resolved.
UNCOV 260	if !htlc.ResolveTime.IsZero() {	×
UNCOV 261	htlcParams.ResolveTime = sqldb.SQLTime(	×
UNCOV 262	htlc.ResolveTime.UTC(),	×
UNCOV 263	)	×
UNCOV 264	}	×
265
UNCOV 266	sqlID, err := tx.InsertInvoiceHTLC(ctx, htlcParams)	×
UNCOV 267	if err != nil {	×
268	return fmt.Errorf("unable to insert invoice htlc: %w",	×
269	err)	×
270	}	×
271
UNCOV 272	sqlHtlcIDs[circuitKey] = sqlID	×
UNCOV 273		×
UNCOV 274	// Store custom records.	×
UNCOV 275	for key, value := range htlc.CustomRecords {	×
UNCOV 276	err = tx.InsertInvoiceHTLCCustomRecord(	×
UNCOV 277	ctx, sqlc.InsertInvoiceHTLCCustomRecordParams{	×
UNCOV 278	Key: int64(key),	×
UNCOV 279	Value: value,	×
UNCOV 280	HtlcID: sqlID,	×
UNCOV 281	},	×
UNCOV 282	)	×
UNCOV 283	if err != nil {	×
284	return err	×
285	}	×
286	}
287	}
288
UNCOV 289	if !invoice.IsAMP() {	×
UNCOV 290	return nil	×
UNCOV 291	}	×
292
UNCOV 293	for setID, ampState := range invoice.AMPState {	×
UNCOV 294	// Find the earliest HTLC of the AMP invoice, which will	×
UNCOV 295	// be used as the creation date of this sub invoice.	×
UNCOV 296	var createdAt time.Time	×
UNCOV 297	for circuitKey := range ampState.InvoiceKeys {	×
UNCOV 298	htlc := invoice.Htlcs[circuitKey]	×
UNCOV 299	if createdAt.IsZero() {	×
UNCOV 300	createdAt = htlc.AcceptTime.UTC()	×
UNCOV 301	continue	×
302	}
303
UNCOV 304	if createdAt.After(htlc.AcceptTime) {	×
UNCOV 305	createdAt = htlc.AcceptTime.UTC()	×
UNCOV 306	}	×
307	}
308
UNCOV 309	params := sqlc.InsertAMPSubInvoiceParams{	×
UNCOV 310	SetID: setID[:],	×
UNCOV 311	State: int16(ampState.State),	×
UNCOV 312	CreatedAt: createdAt,	×
UNCOV 313	InvoiceID: invoiceID,	×
UNCOV 314	}	×
UNCOV 315		×
UNCOV 316	if ampState.SettleIndex != 0 {	×
UNCOV 317	if ampState.SettleDate.IsZero() {	×
318	return fmt.Errorf("settled AMP sub invoice %x "+	×
319	"missing settle date", setID)	×
320	}	×
321
UNCOV 322	params.SettledAt = sqldb.SQLTime(	×
UNCOV 323	ampState.SettleDate.UTC(),	×
UNCOV 324	)	×
UNCOV 325		×
UNCOV 326	params.SettleIndex = sqldb.SQLInt64(	×
UNCOV 327	ampState.SettleIndex,	×
UNCOV 328	)	×
329	}
330
UNCOV 331	err := tx.InsertAMPSubInvoice(ctx, params)	×
UNCOV 332	if err != nil {	×
333	return fmt.Errorf("unable to insert AMP sub invoice: "+	×
334	"%w", err)	×
335	}	×
336
337	// Now we can add the AMP HTLCs to the database.
UNCOV 338	for circuitKey := range ampState.InvoiceKeys {	×
UNCOV 339	htlc := invoice.Htlcs[circuitKey]	×
UNCOV 340	rootShare := htlc.AMP.Record.RootShare()	×
UNCOV 341		×
UNCOV 342	sqlHtlcID, ok := sqlHtlcIDs[circuitKey]	×
UNCOV 343	if !ok {	×
344	return fmt.Errorf("missing htlc for AMP htlc: "+	×
345	"%v", circuitKey)	×
346	}	×
347
UNCOV 348	params := sqlc.InsertAMPSubInvoiceHTLCParams{	×
UNCOV 349	InvoiceID: invoiceID,	×
UNCOV 350	SetID: setID[:],	×
UNCOV 351	HtlcID: sqlHtlcID,	×
UNCOV 352	RootShare: rootShare[:],	×
UNCOV 353	ChildIndex: int64(htlc.AMP.Record.ChildIndex()),	×
UNCOV 354	Hash: htlc.AMP.Hash[:],	×
UNCOV 355	}	×
UNCOV 356		×
UNCOV 357	if htlc.AMP.Preimage != nil {	×
UNCOV 358	params.Preimage = htlc.AMP.Preimage[:]	×
UNCOV 359	}	×
360
UNCOV 361	err = tx.InsertAMPSubInvoiceHTLC(ctx, params)	×
UNCOV 362	if err != nil {	×
363	return fmt.Errorf("unable to insert AMP sub "+	×
364	"invoice: %w", err)	×
365	}	×
366	}
367	}
368
UNCOV 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.
UNCOV 379	func OverrideInvoiceTimeZone(invoice *Invoice) {	×
UNCOV 380	fixTime := func(t time.Time) time.Time {	×
UNCOV 381	return t.In(time.Local).Truncate(time.Microsecond)	×
UNCOV 382	}	×
383
UNCOV 384	invoice.CreationDate = fixTime(invoice.CreationDate)	×
UNCOV 385		×
UNCOV 386	if !invoice.SettleDate.IsZero() {	×
UNCOV 387	invoice.SettleDate = fixTime(invoice.SettleDate)	×
UNCOV 388	}	×
389
UNCOV 390	if invoice.IsAMP() {	×
UNCOV 391	for setID, ampState := range invoice.AMPState {	×
UNCOV 392	if ampState.SettleDate.IsZero() {	×
UNCOV 393	continue	×
394	}
395
UNCOV 396	ampState.SettleDate = fixTime(ampState.SettleDate)	×
UNCOV 397	invoice.AMPState[setID] = ampState	×
398	}
399	}
400
UNCOV 401	for _, htlc := range invoice.Htlcs {	×
UNCOV 402	if !htlc.AcceptTime.IsZero() {	×
UNCOV 403	htlc.AcceptTime = fixTime(htlc.AcceptTime)	×
UNCOV 404	}	×
405
UNCOV 406	if !htlc.ResolveTime.IsZero() {	×
UNCOV 407	htlc.ResolveTime = fixTime(htlc.ResolveTime)	×
UNCOV 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,
UNCOV 418	kvStore InvoiceDB, tx *sqlc.Queries, batchSize int) error {	×
UNCOV 419		×
UNCOV 420	log.Infof("Starting migration of invoices from KV to SQL")	×
UNCOV 421		×
UNCOV 422	offset := uint64(0)	×
UNCOV 423	t0 := time.Now()	×
UNCOV 424		×
UNCOV 425	// Create the hash index which we will use to look up invoice	×
UNCOV 426	// payment hashes by their add index during migration.	×
UNCOV 427	err := createInvoiceHashIndex(ctx, db, tx)	×
UNCOV 428	if err != nil && !errors.Is(err, ErrNoInvoicesCreated) {	×
429	log.Errorf("Unable to create invoice hash index: %v",	×
430	err)	×
431		×
432	return err	×
433	}	×
UNCOV 434	log.Debugf("Created SQL invoice hash index in %v", time.Since(t0))	×
UNCOV 435		×
UNCOV 436	total := 0	×
UNCOV 437	// Now we can start migrating the invoices. We'll do this in	×
UNCOV 438	// batches to reduce memory usage.	×
UNCOV 439	for {	×
UNCOV 440	t0 = time.Now()	×
UNCOV 441	query := InvoiceQuery{	×
UNCOV 442	IndexOffset: offset,	×
UNCOV 443	NumMaxInvoices: uint64(batchSize),	×
UNCOV 444	}	×
UNCOV 445		×
UNCOV 446	queryResult, err := kvStore.QueryInvoices(ctx, query)	×
UNCOV 447	if err != nil && !errors.Is(err, ErrNoInvoicesCreated) {	×
448	return fmt.Errorf("unable to query invoices: "+	×
449	"%w", err)	×
450	}	×
451
UNCOV 452	if len(queryResult.Invoices) == 0 {	×
UNCOV 453	log.Infof("All invoices migrated")	×
UNCOV 454		×
UNCOV 455	break	×
456	}
457
UNCOV 458	err = migrateInvoices(ctx, tx, queryResult.Invoices)	×
UNCOV 459	if err != nil {	×
460	return err	×
461	}	×
462
UNCOV 463	offset = queryResult.LastIndexOffset	×
UNCOV 464	total += len(queryResult.Invoices)	×
UNCOV 465	log.Debugf("Migrated %d KV invoices to SQL in %v\n", total,	×
UNCOV 466	time.Since(t0))	×
467	}
468
469	// Clean up the hash index as it's no longer needed.
UNCOV 470	err = tx.ClearKVInvoiceHashIndex(ctx)	×
UNCOV 471	if err != nil {	×
472	return fmt.Errorf("unable to clear invoice hash "+	×
473	"index: %w", err)	×
474	}	×
475
UNCOV 476	log.Infof("Migration of %d invoices from KV to SQL completed", total)	×
UNCOV 477		×
UNCOV 478	return nil	×
479	}
480
481	func migrateInvoices(ctx context.Context, tx *sqlc.Queries,
UNCOV 482	invoices []Invoice) error {	×
UNCOV 483		×
UNCOV 484	for i, invoice := range invoices {	×
UNCOV 485	var paymentHash lntypes.Hash	×
UNCOV 486	if invoice.Terms.PaymentPreimage != nil {	×
487	paymentHash = invoice.Terms.PaymentPreimage.Hash()	×
UNCOV 488	} else {	×
UNCOV 489	paymentHashBytes, err :=	×
UNCOV 490	tx.GetKVInvoicePaymentHashByAddIndex(	×
UNCOV 491	ctx, int64(invoice.AddIndex),	×
UNCOV 492	)	×
UNCOV 493	if err != nil {	×
494	// This would be an unexpected inconsistency	×
495	// in the kv database. We can't do much here	×
496	// so we'll notify the user and continue.	×
497	log.Warnf("Cannot migrate invoice, unable to "+	×
498	"fetch payment hash (add_index=%v): %v",	×
499	invoice.AddIndex, err)	×
500		×
501	continue	×
502	}
503
UNCOV 504	copy(paymentHash[:], paymentHashBytes)	×
505	}
506
UNCOV 507	err := MigrateSingleInvoice(ctx, tx, &invoices[i], paymentHash)	×
UNCOV 508	if err != nil {	×
509	return fmt.Errorf("unable to migrate invoice(%v): %w",	×
510	paymentHash, err)	×
511	}	×
512
UNCOV 513	migratedInvoice, err := fetchInvoice(	×
UNCOV 514	ctx, tx, InvoiceRefByHash(paymentHash),	×
UNCOV 515	)	×
UNCOV 516	if err != nil {	×
517	return fmt.Errorf("unable to fetch migrated "+	×
518	"invoice(%v): %w", paymentHash, err)	×
519	}	×
520
521	// Override the time zone for comparison. Note that we need to
522	// override both invoices as the original invoice is coming from
523	// KV database, it was stored as a binary serialized Go
524	// time.Time value which has nanosecond precision but might have
525	// been created in a different time zone. The migrated invoice
526	// is stored in SQL in UTC and selected in the local time zone,
527	// however in PostgreSQL it has microsecond precision while in
528	// SQLite it has nanosecond precision if using TEXT storage
529	// class.
UNCOV 530	OverrideInvoiceTimeZone(&invoice)	×
UNCOV 531	OverrideInvoiceTimeZone(migratedInvoice)	×
UNCOV 532		×
UNCOV 533	// Override the add index before checking for equality.	×
UNCOV 534	migratedInvoice.AddIndex = invoice.AddIndex	×
UNCOV 535		×
UNCOV 536	if !reflect.DeepEqual(invoice, *migratedInvoice) {	×
537	diff := difflib.UnifiedDiff{	×
538	A: difflib.SplitLines(	×
539	spew.Sdump(invoice),	×
540	),	×
541	B: difflib.SplitLines(	×
542	spew.Sdump(migratedInvoice),	×
543	),	×
544	FromFile: "Expected",	×
545	FromDate: "",	×
546	ToFile: "Actual",	×
547	ToDate: "",	×
548	Context: 3,	×
549	}	×
550	diffText, _ := difflib.GetUnifiedDiffString(diff)	×
551		×
552	return fmt.Errorf("%w: %v.\n%v", ErrMigrationMismatch,	×
553	paymentHash, diffText)	×
554	}	×
555	}
556
UNCOV 557	return nil	×
558	}

lightningnetwork / lnd / 13211764208

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