15100227105

Committed 18 May 2025 09:28PM UTC coverage: 68.997% (+0.007%) from 68.99%

Build # 15100227105

Build Type

Pull #9822

github

Committed by

web-flow

Commit Message

Merge 40492099f into 3707b1fb7

Pull Request Pull Request #9822: Refactor Payments Code (1|?)

Run Details

414 of 501 new or added lines in 16 files covered. (82.63%)

49 existing lines in 19 files now uncovered.

133955 of 194145 relevant lines covered (69.0%)

22105.67 hits per line

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

73.12

/channeldb/payments.go

package channeldb

import (
        "bytes"
        "encoding/binary"
        "errors"
        "fmt"
        "io"
        "sort"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/btcsuite/btcd/wire"
        "github.com/lightningnetwork/lnd/kvdb"
        "github.com/lightningnetwork/lnd/lntypes"
        "github.com/lightningnetwork/lnd/lnwire"
        pymtpkg "github.com/lightningnetwork/lnd/payments"
        "github.com/lightningnetwork/lnd/record"
        "github.com/lightningnetwork/lnd/routing/route"
        "github.com/lightningnetwork/lnd/tlv"
)

var (
        // paymentsRootBucket is the name of the top-level bucket within the
        // database that stores all data related to payments. Within this
        // bucket, each payment hash its own sub-bucket keyed by its payment
        // hash.
        //
        // Bucket hierarchy:
        //
        // root-bucket
        //      |
        //      |-- <paymenthash>
        //      |        |--sequence-key: <sequence number>
        //      |        |--creation-info-key: <creation info>
        //      |        |--fail-info-key: <(optional) fail info>
        //      |        |
        //      |        |--payment-htlcs-bucket (shard-bucket)
        //      |        |        |
        //      |        |        |-- ai<htlc attempt ID>: <htlc attempt info>
        //      |        |        |-- si<htlc attempt ID>: <(optional) settle info>
        //      |        |        |-- fi<htlc attempt ID>: <(optional) fail info>
        //      |        |        |
        //      |        |       ...
        //      |        |
        //      |        |
        //      |        |--duplicate-bucket (only for old, completed payments)
        //      |                 |
        //      |                 |-- <seq-num>
        //      |                 |       |--sequence-key: <sequence number>
        //      |                 |       |--creation-info-key: <creation info>
        //      |                 |       |--ai: <attempt info>
        //      |                 |       |--si: <settle info>
        //      |                 |       |--fi: <fail info>
        //      |                 |
        //      |                 |-- <seq-num>
        //      |                 |       |
        //      |                ...     ...
        //      |
        //      |-- <paymenthash>
        //      |        |
        //      |       ...
        //     ...
        //
        paymentsRootBucket = []byte("payments-root-bucket")

        // paymentSequenceKey is a key used in the payment's sub-bucket to
        // store the sequence number of the payment.
        paymentSequenceKey = []byte("payment-sequence-key")

        // paymentCreationInfoKey is a key used in the payment's sub-bucket to
        // store the creation info of the payment.
        paymentCreationInfoKey = []byte("payment-creation-info")

        // paymentHtlcsBucket is a bucket where we'll store the information
        // about the HTLCs that were attempted for a payment.
        paymentHtlcsBucket = []byte("payment-htlcs-bucket")

        // htlcAttemptInfoKey is the key used as the prefix of an HTLC attempt
        // to store the info about the attempt that was done for the HTLC in
        // question. The HTLC attempt ID is concatenated at the end.
        htlcAttemptInfoKey = []byte("ai")

        // htlcSettleInfoKey is the key used as the prefix of an HTLC attempt
        // settle info, if any. The HTLC attempt ID is concatenated at the end.
        htlcSettleInfoKey = []byte("si")

        // htlcFailInfoKey is the key used as the prefix of an HTLC attempt
        // failure information, if any.The  HTLC attempt ID is concatenated at
        // the end.
        htlcFailInfoKey = []byte("fi")

        // paymentFailInfoKey is a key used in the payment's sub-bucket to
        // store information about the reason a payment failed.
        paymentFailInfoKey = []byte("payment-fail-info")

        // paymentsIndexBucket is the name of the top-level bucket within the
        // database that stores an index of payment sequence numbers to its
        // payment hash.
        // payments-sequence-index-bucket
        //         |--<sequence-number>: <payment hash>
        //         |--...
        //         |--<sequence-number>: <payment hash>
        paymentsIndexBucket = []byte("payments-index-bucket")
)

// htlcBucketKey creates a composite key from prefix and id where the result is
// simply the two concatenated.
func htlcBucketKey(prefix, id []byte) []byte {
        key := make([]byte, len(prefix)+len(id))
        copy(key, prefix)
        copy(key[len(prefix):], id)
        return key
}

// FetchPayments returns all sent payments found in the DB.
//
// nolint: dupl
//
// TODO(replace with QueryPayments)
func (d *DB) FetchPayments() ([]*pymtpkg.MPPayment, error) {
        var payments []*pymtpkg.MPPayment

        err := kvdb.View(d, func(tx kvdb.RTx) error {
                paymentsBucket := tx.ReadBucket(paymentsRootBucket)
                if paymentsBucket == nil {
                        return nil
                }

                return paymentsBucket.ForEach(func(k, v []byte) error {
                        bucket := paymentsBucket.NestedReadBucket(k)
                        if bucket == nil {
                                // We only expect sub-buckets to be found in
                                // this top-level bucket.
                                return fmt.Errorf("non bucket element in " +
                                        "payments bucket")
                        }

                        p, err := fetchPayment(bucket)
                        if err != nil {
                                return err
                        }

                        payments = append(payments, p)

                        // For older versions of lnd, duplicate payments to a
                        // payment has was possible. These will be found in a
                        // sub-bucket indexed by their sequence number if
                        // available.
                        duplicatePayments, err := fetchDuplicatePayments(bucket)
                        if err != nil {
                                return err
                        }

                        payments = append(payments, duplicatePayments...)
                        return nil
                })
        }, func() {
                payments = nil
        })
        if err != nil {
                return nil, err
        }

        // Before returning, sort the payments by their sequence number.
        sort.Slice(payments, func(i, j int) bool {
                return payments[i].SequenceNum < payments[j].SequenceNum
        })

        return payments, nil
}

func fetchCreationInfo(bucket kvdb.RBucket) (*pymtpkg.PaymentCreationInfo,
        error) {

        b := bucket.Get(paymentCreationInfoKey)
        if b == nil {
                return nil, fmt.Errorf("creation info not found")
        }

        r := bytes.NewReader(b)
        return deserializePaymentCreationInfo(r)
}

func fetchPayment(bucket kvdb.RBucket) (*pymtpkg.MPPayment, error) {
        seqBytes := bucket.Get(paymentSequenceKey)
        if seqBytes == nil {
                return nil, fmt.Errorf("sequence number not found")
        }

        sequenceNum := binary.BigEndian.Uint64(seqBytes)

        // Get the PaymentCreationInfo.
        creationInfo, err := fetchCreationInfo(bucket)
        if err != nil {
                return nil, err
        }

        var htlcs []pymtpkg.HTLCAttempt
        htlcsBucket := bucket.NestedReadBucket(paymentHtlcsBucket)
        if htlcsBucket != nil {
                // Get the payment attempts. This can be empty.
                htlcs, err = fetchHtlcAttempts(htlcsBucket)
                if err != nil {
                        return nil, err
                }
        }

        // Get failure reason if available.
        var failureReason *pymtpkg.FailureReason
        b := bucket.Get(paymentFailInfoKey)
        if b != nil {
                reason := pymtpkg.FailureReason(b[0])
                failureReason = &reason
        }

        // Create a new payment.
        payment := &pymtpkg.MPPayment{
                SequenceNum:   sequenceNum,
                Info:          creationInfo,
                HTLCs:         htlcs,
                FailureReason: failureReason,
        }

        // Set its state and status.
        if err := payment.SetState(); err != nil {
                return nil, err
        }

        return payment, nil
}

// fetchHtlcAttempts retrieves all htlc attempts made for the payment found in
// the given bucket.
func fetchHtlcAttempts(bucket kvdb.RBucket) ([]pymtpkg.HTLCAttempt, error) {
        htlcsMap := make(map[uint64]*pymtpkg.HTLCAttempt)

        attemptInfoCount := 0
        err := bucket.ForEach(func(k, v []byte) error {
                aid := byteOrder.Uint64(k[len(k)-8:])

                if _, ok := htlcsMap[aid]; !ok {
                        htlcsMap[aid] = &pymtpkg.HTLCAttempt{}
                }

                var err error
                switch {
                case bytes.HasPrefix(k, htlcAttemptInfoKey):
                        attemptInfo, err := readHtlcAttemptInfo(v)
                        if err != nil {
                                return err
                        }

                        attemptInfo.AttemptID = aid
                        htlcsMap[aid].HTLCAttemptInfo = *attemptInfo
                        attemptInfoCount++

                case bytes.HasPrefix(k, htlcSettleInfoKey):
                        htlcsMap[aid].Settle, err = readHtlcSettleInfo(v)
                        if err != nil {
                                return err
                        }

                case bytes.HasPrefix(k, htlcFailInfoKey):
                        htlcsMap[aid].Failure, err = readHtlcFailInfo(v)
                        if err != nil {
                                return err
                        }

                default:
                        return fmt.Errorf("unknown htlc attempt key")
                }

                return nil
        })
        if err != nil {
                return nil, err
        }

        // Sanity check that all htlcs have an attempt info.
        if attemptInfoCount != len(htlcsMap) {
                return nil, pymtpkg.ErrNoAttemptInfo
        }

        keys := make([]uint64, len(htlcsMap))
        i := 0
        for k := range htlcsMap {
                keys[i] = k
                i++
        }

        // Sort HTLC attempts by their attempt ID. This is needed because in the
        // DB we store the attempts with keys prefixed by their status which
        // changes order (groups them together by status).
        sort.Slice(keys, func(i, j int) bool {
                return keys[i] < keys[j]
        })

        htlcs := make([]pymtpkg.HTLCAttempt, len(htlcsMap))
        for i, key := range keys {
                htlcs[i] = *htlcsMap[key]
        }

        return htlcs, nil
}

// readHtlcAttemptInfo reads the payment attempt info for this htlc.
func readHtlcAttemptInfo(b []byte) (*pymtpkg.HTLCAttemptInfo, error) {
        r := bytes.NewReader(b)
        return deserializeHTLCAttemptInfo(r)
}

// readHtlcSettleInfo reads the settle info for the htlc. If the htlc isn't
// settled, nil is returned.
func readHtlcSettleInfo(b []byte) (*pymtpkg.HTLCSettleInfo, error) {
        r := bytes.NewReader(b)
        return deserializeHTLCSettleInfo(r)
}

// readHtlcFailInfo reads the failure info for the htlc. If the htlc hasn't
// failed, nil is returned.
func readHtlcFailInfo(b []byte) (*pymtpkg.HTLCFailInfo, error) {
        r := bytes.NewReader(b)
        return deserializeHTLCFailInfo(r)
}

// fetchFailedHtlcKeys retrieves the bucket keys of all failed HTLCs of a
// payment bucket.
func fetchFailedHtlcKeys(bucket kvdb.RBucket) ([][]byte, error) {
        htlcsBucket := bucket.NestedReadBucket(paymentHtlcsBucket)

        var htlcs []pymtpkg.HTLCAttempt
        var err error
        if htlcsBucket != nil {
                htlcs, err = fetchHtlcAttempts(htlcsBucket)
                if err != nil {
                        return nil, err
                }
        }

        // Now iterate though them and save the bucket keys for the failed
        // HTLCs.
        var htlcKeys [][]byte
        for _, h := range htlcs {
                if h.Failure == nil {
                        continue
                }

                htlcKeyBytes := make([]byte, 8)
                binary.BigEndian.PutUint64(htlcKeyBytes, h.AttemptID)

                htlcKeys = append(htlcKeys, htlcKeyBytes)
        }

        return htlcKeys, nil
}

// QueryPayments is a query to the payments database which is restricted
// to a subset of payments by the payments query, containing an offset
// index and a maximum number of returned payments.
func (d *DB) QueryPayments(query pymtpkg.Query) (pymtpkg.Response, error) {
        var resp pymtpkg.Response

        if err := kvdb.View(d, func(tx kvdb.RTx) error {
                // Get the root payments bucket.
                paymentsBucket := tx.ReadBucket(paymentsRootBucket)
                if paymentsBucket == nil {
                        return nil
                }

                // Get the index bucket which maps sequence number -> payment
                // hash and duplicate bool. If we have a payments bucket, we
                // should have an indexes bucket as well.
                indexes := tx.ReadBucket(paymentsIndexBucket)
                if indexes == nil {
                        return fmt.Errorf("index bucket does not exist")
                }

                // accumulatePayments gets payments with the sequence number
                // and hash provided and adds them to our list of payments if
                // they meet the criteria of our query. It returns the number
                // of payments that were added.
                accumulatePayments := func(sequenceKey, hash []byte) (bool,
                        error) {

                        r := bytes.NewReader(hash)
                        paymentHash, err := deserializePaymentIndex(r)
                        if err != nil {
                                return false, err
                        }

                        payment, err := fetchPaymentWithSequenceNumber(
                                tx, paymentHash, sequenceKey,
                        )
                        if err != nil {
                                return false, err
                        }

                        // To keep compatibility with the old API, we only
                        // return non-succeeded payments if requested.
                        if payment.Status != pymtpkg.StatusSucceeded &&
                                !query.IncludeIncomplete {

                                return false, err
                        }

                        // Get the creation time in Unix seconds, this always
                        // rounds down the nanoseconds to full seconds.
                        createTime := payment.Info.CreationTime.Unix()

                        // Skip any payments that were created before the
                        // specified time.
                        if createTime < query.CreationDateStart {
                                return false, nil
                        }

                        // Skip any payments that were created after the
                        // specified time.
                        if query.CreationDateEnd != 0 &&
                                createTime > query.CreationDateEnd {

                                return false, nil
                        }

                        // At this point, we've exhausted the offset, so we'll
                        // begin collecting invoices found within the range.
                        resp.Payments = append(resp.Payments, payment)
                        return true, nil
                }

                // Create a paginator which reads from our sequence index bucket
                // with the parameters provided by the payments query.
                paginator := newPaginator(
                        indexes.ReadCursor(), query.Reversed, query.IndexOffset,
                        query.MaxPayments,
                )

                // Run a paginated query, adding payments to our response.
                if err := paginator.query(accumulatePayments); err != nil {
                        return err
                }

                // Counting the total number of payments is expensive, since we
                // literally have to traverse the cursor linearly, which can
                // take quite a while. So it's an optional query parameter.
                if query.CountTotal {
                        var (
                                totalPayments uint64
                                err           error
                        )
                        countFn := func(_, _ []byte) error {
                                totalPayments++

                                return nil
                        }

                        // In non-boltdb database backends, there's a faster
                        // ForAll query that allows for batch fetching items.
                        if fastBucket, ok := indexes.(kvdb.ExtendedRBucket); ok {
                                err = fastBucket.ForAll(countFn)
                        } else {
                                err = indexes.ForEach(countFn)
                        }
                        if err != nil {
                                return fmt.Errorf("error counting payments: %w",
                                        err)
                        }

                        resp.TotalCount = totalPayments
                }

                return nil
        }, func() {
                resp = pymtpkg.Response{}
        }); err != nil {
                return resp, err
        }

        // Need to swap the payments slice order if reversed order.
        if query.Reversed {
                for l, r := 0, len(resp.Payments)-1; l < r; l, r = l+1, r-1 {
                        resp.Payments[l], resp.Payments[r] =
                                resp.Payments[r], resp.Payments[l]
                }
        }

        // Set the first and last index of the returned payments so that the
        // caller can resume from this point later on.
        if len(resp.Payments) > 0 {
                resp.FirstIndexOffset = resp.Payments[0].SequenceNum
                resp.LastIndexOffset =
                        resp.Payments[len(resp.Payments)-1].SequenceNum
        }

        return resp, nil
}

// fetchPaymentWithSequenceNumber get the payment which matches the payment hash
// *and* sequence number provided from the database. This is required because
// we previously had more than one payment per hash, so we have multiple indexes
// pointing to a single payment; we want to retrieve the correct one.
func fetchPaymentWithSequenceNumber(tx kvdb.RTx, paymentHash lntypes.Hash,
        sequenceNumber []byte) (*pymtpkg.MPPayment, error) {

        // We can now lookup the payment keyed by its hash in
        // the payments root bucket.
        bucket, err := fetchPaymentBucket(tx, paymentHash)
        if err != nil {
                return nil, err
        }

        // A single payment hash can have multiple payments associated with it.
        // We lookup our sequence number first, to determine whether this is
        // the payment we are actually looking for.
        seqBytes := bucket.Get(paymentSequenceKey)
        if seqBytes == nil {
                return nil, pymtpkg.ErrNoSequenceNumber
        }

        // If this top level payment has the sequence number we are looking for,
        // return it.
        if bytes.Equal(seqBytes, sequenceNumber) {
                return fetchPayment(bucket)
        }

        // If we were not looking for the top level payment, we are looking for
        // one of our duplicate payments. We need to iterate through the seq
        // numbers in this bucket to find the correct payments. If we do not
        // find a duplicate payments bucket here, something is wrong.
        dup := bucket.NestedReadBucket(duplicatePaymentsBucket)
        if dup == nil {
                return nil, pymtpkg.ErrNoDuplicateBucket
        }

        var duplicatePayment *pymtpkg.MPPayment
        err = dup.ForEach(func(k, v []byte) error {
                subBucket := dup.NestedReadBucket(k)
                if subBucket == nil {
                        // We one bucket for each duplicate to be found.
                        return pymtpkg.ErrNoDuplicateNestedBucket
                }

                seqBytes := subBucket.Get(duplicatePaymentSequenceKey)
                if seqBytes == nil {
                        return err
                }

                // If this duplicate payment is not the sequence number we are
                // looking for, we can continue.
                if !bytes.Equal(seqBytes, sequenceNumber) {
                        return nil
                }

                duplicatePayment, err = fetchDuplicatePayment(subBucket)
                if err != nil {
                        return err
                }

                return nil
        })
        if err != nil {
                return nil, err
        }

        // If none of the duplicate payments matched our sequence number, we
        // failed to find the payment with this sequence number; something is
        // wrong.
        if duplicatePayment == nil {
                return nil, pymtpkg.ErrDuplicateNotFound
        }

        return duplicatePayment, nil
}

// DeletePayment deletes a payment from the DB given its payment hash. If
// failedHtlcsOnly is set, only failed HTLC attempts of the payment will be
// deleted.
func (d *DB) DeletePayment(paymentHash lntypes.Hash,
        failedHtlcsOnly bool) error {

        return kvdb.Update(d, func(tx kvdb.RwTx) error {
                payments := tx.ReadWriteBucket(paymentsRootBucket)
                if payments == nil {
                        return nil
                }

                bucket := payments.NestedReadWriteBucket(paymentHash[:])
                if bucket == nil {
                        return fmt.Errorf("non bucket element in payments " +
                                "bucket")
                }

                // If the status is InFlight, we cannot safely delete
                // the payment information, so we return early.
                paymentStatus, err := fetchPaymentStatus(bucket)
                if err != nil {
                        return err
                }

                // If the payment has inflight HTLCs, we cannot safely delete
                // the payment information, so we return an error.
                if err := paymentStatus.Removable(); err != nil {
                        return fmt.Errorf("payment '%v' has inflight HTLCs"+
                                "and therefore cannot be deleted: %w",
                                paymentHash.String(), err)
                }

                // Delete the failed HTLC attempts we found.
                if failedHtlcsOnly {
                        toDelete, err := fetchFailedHtlcKeys(bucket)
                        if err != nil {
                                return err
                        }

                        htlcsBucket := bucket.NestedReadWriteBucket(
                                paymentHtlcsBucket,
                        )

                        for _, htlcID := range toDelete {
                                err = htlcsBucket.Delete(
                                        htlcBucketKey(htlcAttemptInfoKey, htlcID),
                                )
                                if err != nil {
                                        return err
                                }

                                err = htlcsBucket.Delete(
                                        htlcBucketKey(htlcFailInfoKey, htlcID),
                                )
                                if err != nil {
                                        return err
                                }

                                err = htlcsBucket.Delete(
                                        htlcBucketKey(htlcSettleInfoKey, htlcID),
                                )
                                if err != nil {
                                        return err
                                }
                        }

                        return nil
                }

                seqNrs, err := fetchSequenceNumbers(bucket)
                if err != nil {
                        return err
                }

                if err := payments.DeleteNestedBucket(paymentHash[:]); err != nil {
                        return err
                }

                indexBucket := tx.ReadWriteBucket(paymentsIndexBucket)
                for _, k := range seqNrs {
                        if err := indexBucket.Delete(k); err != nil {
                                return err
                        }
                }

                return nil
        }, func() {})
}

// DeletePayments deletes all completed and failed payments from the DB. If
// failedOnly is set, only failed payments will be considered for deletion. If
// failedHtlcsOnly is set, the payment itself won't be deleted, only failed HTLC
// attempts. The method returns the number of deleted payments, which is always
// 0 if failedHtlcsOnly is set.
func (d *DB) DeletePayments(failedOnly, failedHtlcsOnly bool) (int, error) {
        var numPayments int
        err := kvdb.Update(d, func(tx kvdb.RwTx) error {
                payments := tx.ReadWriteBucket(paymentsRootBucket)
                if payments == nil {
                        return nil
                }

                var (
                        // deleteBuckets is the set of payment buckets we need
                        // to delete.
                        deleteBuckets [][]byte

                        // deleteIndexes is the set of indexes pointing to these
                        // payments that need to be deleted.
                        deleteIndexes [][]byte

                        // deleteHtlcs maps a payment hash to the HTLC IDs we
                        // want to delete for that payment.
                        deleteHtlcs = make(map[lntypes.Hash][][]byte)
                )
                err := payments.ForEach(func(k, _ []byte) error {
                        bucket := payments.NestedReadBucket(k)
                        if bucket == nil {
                                // We only expect sub-buckets to be found in
                                // this top-level bucket.
                                return fmt.Errorf("non bucket element in " +
                                        "payments bucket")
                        }

                        // If the status is InFlight, we cannot safely delete
                        // the payment information, so we return early.
                        paymentStatus, err := fetchPaymentStatus(bucket)
                        if err != nil {
                                return err
                        }

                        // If the payment has inflight HTLCs, we cannot safely
                        // delete the payment information, so we return an nil
                        // to skip it.
                        if err := paymentStatus.Removable(); err != nil {
                                return nil
                        }

                        // If we requested to only delete failed payments, we
                        // can return if this one is not.
                        if failedOnly && paymentStatus != pymtpkg.StatusFailed {
                                return nil
                        }

                        // If we are only deleting failed HTLCs, fetch them.
                        if failedHtlcsOnly {
                                toDelete, err := fetchFailedHtlcKeys(bucket)
                                if err != nil {
                                        return err
                                }

                                hash, err := lntypes.MakeHash(k)
                                if err != nil {
                                        return err
                                }

                                deleteHtlcs[hash] = toDelete

                                // We return, we are only deleting attempts.
                                return nil
                        }

                        // Add the bucket to the set of buckets we can delete.
                        deleteBuckets = append(deleteBuckets, k)

                        // Get all the sequence number associated with the
                        // payment, including duplicates.
                        seqNrs, err := fetchSequenceNumbers(bucket)
                        if err != nil {
                                return err
                        }

                        deleteIndexes = append(deleteIndexes, seqNrs...)
                        numPayments++
                        return nil
                })
                if err != nil {
                        return err
                }

                // Delete the failed HTLC attempts we found.
                for hash, htlcIDs := range deleteHtlcs {
                        bucket := payments.NestedReadWriteBucket(hash[:])
                        htlcsBucket := bucket.NestedReadWriteBucket(
                                paymentHtlcsBucket,
                        )

                        for _, aid := range htlcIDs {
                                if err := htlcsBucket.Delete(
                                        htlcBucketKey(htlcAttemptInfoKey, aid),
                                ); err != nil {
                                        return err
                                }

                                if err := htlcsBucket.Delete(
                                        htlcBucketKey(htlcFailInfoKey, aid),
                                ); err != nil {
                                        return err
                                }

                                if err := htlcsBucket.Delete(
                                        htlcBucketKey(htlcSettleInfoKey, aid),
                                ); err != nil {
                                        return err
                                }
                        }
                }

                for _, k := range deleteBuckets {
                        if err := payments.DeleteNestedBucket(k); err != nil {
                                return err
                        }
                }

                // Get our index bucket and delete all indexes pointing to the
                // payments we are deleting.
                indexBucket := tx.ReadWriteBucket(paymentsIndexBucket)
                for _, k := range deleteIndexes {
                        if err := indexBucket.Delete(k); err != nil {
                                return err
                        }
                }

                return nil
        }, func() {
                numPayments = 0
        })
        if err != nil {
                return 0, err
        }

        return numPayments, nil
}

// fetchSequenceNumbers fetches all the sequence numbers associated with a
// payment, including those belonging to any duplicate payments.
func fetchSequenceNumbers(paymentBucket kvdb.RBucket) ([][]byte, error) {
        seqNum := paymentBucket.Get(paymentSequenceKey)
        if seqNum == nil {
                return nil, errors.New("expected sequence number")
        }

        sequenceNumbers := [][]byte{seqNum}

        // Get the duplicate payments bucket, if it has no duplicates, just
        // return early with the payment sequence number.
        duplicates := paymentBucket.NestedReadBucket(duplicatePaymentsBucket)
        if duplicates == nil {
                return sequenceNumbers, nil
        }

        // If we do have duplicated, they are keyed by sequence number, so we
        // iterate through the duplicates bucket and add them to our set of
        // sequence numbers.
        if err := duplicates.ForEach(func(k, v []byte) error {
                sequenceNumbers = append(sequenceNumbers, k)
                return nil
        }); err != nil {
                return nil, err
        }

        return sequenceNumbers, nil
}

// nolint: dupl
func serializePaymentCreationInfo(w io.Writer,
        c *pymtpkg.PaymentCreationInfo) error {

        var scratch [8]byte

        if _, err := w.Write(c.PaymentIdentifier[:]); err != nil {
                return err
        }

        byteOrder.PutUint64(scratch[:], uint64(c.Value))
        if _, err := w.Write(scratch[:]); err != nil {
                return err
        }

        if err := serializeTime(w, c.CreationTime); err != nil {
                return err
        }

        byteOrder.PutUint32(scratch[:4], uint32(len(c.PaymentRequest)))
        if _, err := w.Write(scratch[:4]); err != nil {
                return err
        }

        if _, err := w.Write(c.PaymentRequest[:]); err != nil {
                return err
        }

        // Any remaining bytes are TLV encoded records. Currently, these are
        // only the custom records provided by the user to be sent to the first
        // hop. But this can easily be extended with further records by merging
        // the records into a single TLV stream.
        err := c.FirstHopCustomRecords.SerializeTo(w)
        if err != nil {
                return err
        }

        return nil
}

func deserializePaymentCreationInfo(r io.Reader) (*pymtpkg.PaymentCreationInfo,
        error) {

        var scratch [8]byte

        c := &pymtpkg.PaymentCreationInfo{}

        if _, err := io.ReadFull(r, c.PaymentIdentifier[:]); err != nil {
                return nil, err
        }

        if _, err := io.ReadFull(r, scratch[:]); err != nil {
                return nil, err
        }
        c.Value = lnwire.MilliSatoshi(byteOrder.Uint64(scratch[:]))

        creationTime, err := deserializeTime(r)
        if err != nil {
                return nil, err
        }
        c.CreationTime = creationTime

        if _, err := io.ReadFull(r, scratch[:4]); err != nil {
                return nil, err
        }

        reqLen := uint32(byteOrder.Uint32(scratch[:4]))
        payReq := make([]byte, reqLen)
        if reqLen > 0 {
                if _, err := io.ReadFull(r, payReq); err != nil {
                        return nil, err
                }
        }
        c.PaymentRequest = payReq

        // Any remaining bytes are TLV encoded records. Currently, these are
        // only the custom records provided by the user to be sent to the first
        // hop. But this can easily be extended with further records by merging
        // the records into a single TLV stream.
        c.FirstHopCustomRecords, err = lnwire.ParseCustomRecordsFrom(r)
        if err != nil {
                return nil, err
        }

        return c, nil
}

func serializeHTLCAttemptInfo(w io.Writer, a *pymtpkg.HTLCAttemptInfo) error {
        sessionKey := a.SessionKey().Serialize()

        // We nned to make sure the session key is 32 bytes, so we copy the
        // session key into a 32 byte array.
        var sessionKeyArray [32]byte
        copy(sessionKeyArray[:], sessionKey)
        if err := WriteElements(w, sessionKeyArray); err != nil {
                return err
        }

        if err := SerializeRoute(w, a.Route); err != nil {
                return err
        }

        if err := serializeTime(w, a.AttemptTime); err != nil {
                return err
        }

        // If the hash is nil we can just return.
        if a.Hash == nil {
                return nil
        }

        if _, err := w.Write(a.Hash[:]); err != nil {
                return err
        }

        // Merge the fixed/known records together with the custom records to
        // serialize them as a single blob. We can't do this in SerializeRoute
        // because we're in the middle of the byte stream there. We can only do
        // TLV serialization at the end of the stream, since EOF is allowed for
        // a stream if no more data is expected.
        producers := []tlv.RecordProducer{
                &a.Route.FirstHopAmount,
        }
        tlvData, err := lnwire.MergeAndEncode(
                producers, nil, a.Route.FirstHopWireCustomRecords,
        )
        if err != nil {
                return err
        }

        if _, err := w.Write(tlvData); err != nil {
                return err
        }

        return nil
}

func deserializeHTLCAttemptInfo(r io.Reader) (*pymtpkg.HTLCAttemptInfo, error) {
        var sessionKey [32]byte
        err := ReadElements(r, &sessionKey)
        if err != nil {
                return nil, err
        }

        a := &pymtpkg.HTLCAttemptInfo{}
        a.SetSessionKey(sessionKey)

        a.Route, err = DeserializeRoute(r)
        if err != nil {

                return nil, err
        }

        a.AttemptTime, err = deserializeTime(r)
        if err != nil {
                return nil, err
        }

        hash := lntypes.Hash{}
        _, err = io.ReadFull(r, hash[:])

        switch {
        // Older payment attempts wouldn't have the hash set, in which case we
        // can just return.
        case err == io.EOF, err == io.ErrUnexpectedEOF:
                return a, nil

        case err != nil:
                return nil, err

        default:
        }

        a.Hash = &hash

        // Read any remaining data (if any) and parse it into the known records
        // and custom records.
        extraData, err := io.ReadAll(r)
        if err != nil {
                return nil, err
        }

        customRecords, _, _, err := lnwire.ParseAndExtractCustomRecords(
                extraData, &a.Route.FirstHopAmount,
        )
        if err != nil {
                return nil, err
        }

        a.Route.FirstHopWireCustomRecords = customRecords

        return a, nil
}

func serializeHop(w io.Writer, h *route.Hop) error {
        if err := WriteElements(w,
                h.PubKeyBytes[:],
                h.ChannelID,
                h.OutgoingTimeLock,
                h.AmtToForward,
        ); err != nil {
                return err
        }

        if err := binary.Write(w, byteOrder, h.LegacyPayload); err != nil {
                return err
        }

        // For legacy payloads, we don't need to write any TLV records, so
        // we'll write a zero indicating the our serialized TLV map has no
        // records.
        if h.LegacyPayload {
                return WriteElements(w, uint32(0))
        }

        // Gather all non-primitive TLV records so that they can be serialized
        // as a single blob.
        //
        // TODO(conner): add migration to unify all fields in a single TLV
        // blobs. The split approach will cause headaches down the road as more
        // fields are added, which we can avoid by having a single TLV stream
        // for all payload fields.
        var records []tlv.Record
        if h.MPP != nil {
                records = append(records, h.MPP.Record())
        }

        // Add blinding point and encrypted data if present.
        if h.EncryptedData != nil {
                records = append(records, record.NewEncryptedDataRecord(
                        &h.EncryptedData,
                ))
        }

        if h.BlindingPoint != nil {
                records = append(records, record.NewBlindingPointRecord(
                        &h.BlindingPoint,
                ))
        }

        if h.AMP != nil {
                records = append(records, h.AMP.Record())
        }

        if h.Metadata != nil {
                records = append(records, record.NewMetadataRecord(&h.Metadata))
        }

        if h.TotalAmtMsat != 0 {
                totalMsatInt := uint64(h.TotalAmtMsat)
                records = append(
                        records, record.NewTotalAmtMsatBlinded(&totalMsatInt),
                )
        }

        // Final sanity check to absolutely rule out custom records that are not
        // custom and write into the standard range.
        if err := h.CustomRecords.Validate(); err != nil {
                return err
        }

        // Convert custom records to tlv and add to the record list.
        // MapToRecords sorts the list, so adding it here will keep the list
        // canonical.
        tlvRecords := tlv.MapToRecords(h.CustomRecords)
        records = append(records, tlvRecords...)

        // Otherwise, we'll transform our slice of records into a map of the
        // raw bytes, then serialize them in-line with a length (number of
        // elements) prefix.
        mapRecords, err := tlv.RecordsToMap(records)
        if err != nil {
                return err
        }

        numRecords := uint32(len(mapRecords))
        if err := WriteElements(w, numRecords); err != nil {
                return err
        }

        for recordType, rawBytes := range mapRecords {
                if err := WriteElements(w, recordType); err != nil {
                        return err
                }

                if err := wire.WriteVarBytes(w, 0, rawBytes); err != nil {
                        return err
                }
        }

        return nil
}

// maxOnionPayloadSize is the largest Sphinx payload possible, so we don't need
// to read/write a TLV stream larger than this.
const maxOnionPayloadSize = 1300

func deserializeHop(r io.Reader) (*route.Hop, error) {
        h := &route.Hop{}

        var pub []byte
        if err := ReadElements(r, &pub); err != nil {
                return nil, err
        }
        copy(h.PubKeyBytes[:], pub)

        if err := ReadElements(r,
                &h.ChannelID, &h.OutgoingTimeLock, &h.AmtToForward,
        ); err != nil {
                return nil, err
        }

        // TODO(roasbeef): change field to allow LegacyPayload false to be the
        // legacy default?
        err := binary.Read(r, byteOrder, &h.LegacyPayload)
        if err != nil {
                return nil, err
        }

        var numElements uint32
        if err := ReadElements(r, &numElements); err != nil {
                return nil, err
        }

        // If there're no elements, then we can return early.
        if numElements == 0 {
                return h, nil
        }

        tlvMap := make(map[uint64][]byte)
        for i := uint32(0); i < numElements; i++ {
                var tlvType uint64
                if err := ReadElements(r, &tlvType); err != nil {
                        return nil, err
                }

                rawRecordBytes, err := wire.ReadVarBytes(
                        r, 0, maxOnionPayloadSize, "tlv",
                )
                if err != nil {
                        return nil, err
                }

                tlvMap[tlvType] = rawRecordBytes
        }

        // If the MPP type is present, remove it from the generic TLV map and
        // parse it back into a proper MPP struct.
        //
        // TODO(conner): add migration to unify all fields in a single TLV
        // blobs. The split approach will cause headaches down the road as more
        // fields are added, which we can avoid by having a single TLV stream
        // for all payload fields.
        mppType := uint64(record.MPPOnionType)
        if mppBytes, ok := tlvMap[mppType]; ok {
                delete(tlvMap, mppType)

                var (
                        mpp    = &record.MPP{}
                        mppRec = mpp.Record()
                        r      = bytes.NewReader(mppBytes)
                )
                err := mppRec.Decode(r, uint64(len(mppBytes)))
                if err != nil {
                        return nil, err
                }
                h.MPP = mpp
        }

        // If encrypted data or blinding key are present, remove them from
        // the TLV map and parse into proper types.
        encryptedDataType := uint64(record.EncryptedDataOnionType)
        if data, ok := tlvMap[encryptedDataType]; ok {
                delete(tlvMap, encryptedDataType)
                h.EncryptedData = data
        }

        blindingType := uint64(record.BlindingPointOnionType)
        if blindingPoint, ok := tlvMap[blindingType]; ok {
                delete(tlvMap, blindingType)

                h.BlindingPoint, err = btcec.ParsePubKey(blindingPoint)
                if err != nil {
                        return nil, fmt.Errorf("invalid blinding point: %w",
                                err)
                }
        }

        ampType := uint64(record.AMPOnionType)
        if ampBytes, ok := tlvMap[ampType]; ok {
                delete(tlvMap, ampType)

                var (
                        amp    = &record.AMP{}
                        ampRec = amp.Record()
                        r      = bytes.NewReader(ampBytes)
                )
                err := ampRec.Decode(r, uint64(len(ampBytes)))
                if err != nil {
                        return nil, err
                }
                h.AMP = amp
        }

        // If the metadata type is present, remove it from the tlv map and
        // populate directly on the hop.
        metadataType := uint64(record.MetadataOnionType)
        if metadata, ok := tlvMap[metadataType]; ok {
                delete(tlvMap, metadataType)

                h.Metadata = metadata
        }

        totalAmtMsatType := uint64(record.TotalAmtMsatBlindedType)
        if totalAmtMsat, ok := tlvMap[totalAmtMsatType]; ok {
                delete(tlvMap, totalAmtMsatType)

                var (
                        totalAmtMsatInt uint64
                        buf             [8]byte
                )
                if err := tlv.DTUint64(
                        bytes.NewReader(totalAmtMsat),
                        &totalAmtMsatInt,
                        &buf,
                        uint64(len(totalAmtMsat)),
                ); err != nil {
                        return nil, err
                }

                h.TotalAmtMsat = lnwire.MilliSatoshi(totalAmtMsatInt)
        }

        h.CustomRecords = tlvMap

        return h, nil
}

// SerializeRoute serializes a route.
func SerializeRoute(w io.Writer, r route.Route) error {
        if err := WriteElements(w,
                r.TotalTimeLock, r.TotalAmount, r.SourcePubKey[:],
        ); err != nil {
                return err
        }

        if err := WriteElements(w, uint32(len(r.Hops))); err != nil {
                return err
        }

        for _, h := range r.Hops {
                if err := serializeHop(w, h); err != nil {
                        return err
                }
        }

        // Any new/extra TLV data is encoded in serializeHTLCAttemptInfo!

        return nil
}

// DeserializeRoute deserializes a route.
func DeserializeRoute(r io.Reader) (route.Route, error) {
        rt := route.Route{}
        if err := ReadElements(r,
                &rt.TotalTimeLock, &rt.TotalAmount,
        ); err != nil {
                return rt, err
        }

        var pub []byte
        if err := ReadElements(r, &pub); err != nil {
                return rt, err
        }
        copy(rt.SourcePubKey[:], pub)

        var numHops uint32
        if err := ReadElements(r, &numHops); err != nil {
                return rt, err
        }

        var hops []*route.Hop
        for i := uint32(0); i < numHops; i++ {
                hop, err := deserializeHop(r)
                if err != nil {
                        return rt, err
                }
                hops = append(hops, hop)
        }
        rt.Hops = hops

        // Any new/extra TLV data is decoded in deserializeHTLCAttemptInfo!

        return rt, nil
}

lightningnetwork / lnd / 15100227105

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