16139624707

Committed 08 Jul 2025 09:37AM UTC coverage: 67.518% (+9.7%) from 57.787%

Build # 16139624707

Build Type

Pull #10036

github

Committed by

web-flow

Commit Message

Merge cb959bddb into b815109b8

Pull Request Pull Request #10036: [graph mig 1]: graph/db: migrate graph nodes from kvdb to SQL

Run Details

0 of 204 new or added lines in 3 files covered. (0.0%)

27 existing lines in 7 files now uncovered.

135164 of 200190 relevant lines covered (67.52%)

21804.67 hits per line

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

60.76

/invoices/sql_migration.go

package invoices

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

        "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"
        "golang.org/x/time/rate"
)

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

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

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

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

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

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

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

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

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

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

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

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

                        invoiceKey := binary.BigEndian.Uint32(v)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

                sqlHtlcIDs[circuitKey] = sqlID

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

        if !invoice.IsAMP() {
                return nil
        }

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

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

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

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

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

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

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

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

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

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

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

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

        return nil
}

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

        invoice.CreationDate = fixTime(invoice.CreationDate)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        return nil
}

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

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

                                continue
                        }

                        copy(paymentHash[:], paymentHashBytes)
                }

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

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

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

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

                err = sqldb.CompareRecords(invoice, *migratedInvoice, "invoice")
                if err != nil {
                        return err
                }
        }

        return nil
}

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

lightningnetwork / lnd / 16139624707

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