12203658024

Committed 06 Dec 2024 05:54PM UTC coverage: 58.965% (+9.2%) from 49.807%

Build # 12203658024

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

506 of 695 new or added lines in 12 files covered. (72.81%)

67 existing lines in 19 files now uncovered.

133874 of 227038 relevant lines covered (58.97%)

19659.95 hits per line

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

73.72

/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 hashes in the invoice index bucket
                // and store each hash along with the invoice key in the
                // payment_hashes table.
                err := 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.InsertInvoicePaymentHashAndKey(ctx,
                                sqlc.InsertInvoicePaymentHashAndKeyParams{
                                        Hash: k,
                                        ID:   int64(invoiceKey),
                                },
                        )
                })
                if err != nil {
                        return err
                }

                // Next, iterate over all elements in the add index
                // bucket and update the add index value for the
                // corresponding row in the payment_hashes table.
                return 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.SetInvoicePaymentHashAddIndex(ctx,
                                sqlc.SetInvoicePaymentHashAddIndexParams{
                                        ID:       int64(invoiceKey),
                                        AddIndex: sqldb.SQLInt64(addIndexNo),
                                },
                        )
                })
        }, func() {})
}

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

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

                insertInvoiceParams.SettleIndex = sqldb.SQLInt64(
                        invoice.SettleIndex,
                )
                insertInvoiceParams.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.InsertInvoice(ctx, insertInvoiceParams)
        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)
        // 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
        }

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

                        break
                }

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

                offset = queryResult.LastIndexOffset
        }

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

        log.Infof("Migration of invoices from KV to SQL completed")

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

lightningnetwork / lnd / 12203658024

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