15145941929

Committed 20 May 2025 07:14PM UTC coverage: 69.008% (+0.01%) from 68.996%

Build # 15145941929

Build Type

Pull #9844

github

Committed by

web-flow

Commit Message

Merge daced8b33 into c52a6ddeb

Pull Request Pull Request #9844: Refactor Payment PR 3

Run Details

385 of 494 new or added lines in 5 files covered. (77.94%)

66 existing lines in 19 files now uncovered.

133966 of 194131 relevant lines covered (69.01%)

22018.93 hits per line

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

74.8

/channeldb/payments.go

package channeldb

import (
        "bytes"
        "context"
        "encoding/binary"
        "errors"
        "fmt"
        "io"
        "sort"
        "sync"
        "time"

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

var (
        // ErrNoSequenceNumber is returned if we look up a payment which does
        // not have a sequence number.
        ErrNoSequenceNumber = errors.New("sequence number not found")

        // ErrDuplicateNotFound is returned when we lookup a payment by its
        // index and cannot find a payment with a matching sequence number.
        ErrDuplicateNotFound = errors.New("duplicate payment not found")

        // ErrNoDuplicateBucket is returned when we expect to find duplicates
        // when looking up a payment from its index, but the payment does not
        // have any.
        ErrNoDuplicateBucket = errors.New("expected duplicate bucket")

        // ErrNoDuplicateNestedBucket is returned if we do not find duplicate
        // payments in their own sub-bucket.
        ErrNoDuplicateNestedBucket = errors.New("nested duplicate bucket not " +
                "found")
)

// Payment operations related errors.
var (
        // ErrAlreadyPaid signals we have already paid this payment hash.
        ErrAlreadyPaid = errors.New("invoice is already paid")

        // ErrPaymentInFlight signals that payment for this payment hash is
        // already "in flight" on the network.
        ErrPaymentInFlight = errors.New("payment is in transition")

        // ErrPaymentExists is returned when we try to initialize an already
        // existing payment that is not failed.
        ErrPaymentExists = errors.New("payment already exists")

        // ErrPaymentInternal is returned when performing the payment has a
        // conflicting state, such as,
        // - payment has StatusSucceeded but remaining amount is not zero.
        // - payment has StatusInitiated but remaining amount is zero.
        // - payment has StatusFailed but remaining amount is zero.
        ErrPaymentInternal = errors.New("internal error")

        // ErrPaymentNotInitiated is returned if the payment wasn't initiated.
        ErrPaymentNotInitiated = errors.New("payment isn't initiated")

        // ErrPaymentAlreadySucceeded is returned in the event we attempt to
        // change the status of a payment already succeeded.
        ErrPaymentAlreadySucceeded = errors.New("payment is already succeeded")

        // ErrPaymentAlreadyFailed is returned in the event we attempt to alter
        // a failed payment.
        ErrPaymentAlreadyFailed = errors.New("payment has already failed")

        // ErrUnknownPaymentStatus is returned when we do not recognize the
        // existing state of a payment.
        ErrUnknownPaymentStatus = errors.New("unknown payment status")

        // ErrPaymentTerminal is returned if we attempt to alter a payment that
        // already has reached a terminal condition.
        ErrPaymentTerminal = errors.New("payment has reached terminal " +
                "condition")

        // ErrAttemptAlreadySettled is returned if we try to alter an already
        // settled HTLC attempt.
        ErrAttemptAlreadySettled = errors.New("attempt already settled")

        // ErrAttemptAlreadyFailed is returned if we try to alter an already
        // failed HTLC attempt.
        ErrAttemptAlreadyFailed = errors.New("attempt already failed")

        // ErrValueMismatch is returned if we try to register a non-MPP attempt
        // with an amount that doesn't match the payment amount.
        ErrValueMismatch = errors.New("attempted value doesn't match payment " +
                "amount")

        // ErrValueExceedsAmt is returned if we try to register an attempt that
        // would take the total sent amount above the payment amount.
        ErrValueExceedsAmt = errors.New("attempted value exceeds payment " +
                "amount")

        // ErrNonMPPayment is returned if we try to register an MPP attempt for
        // a payment that already has a non-MPP attempt registered.
        ErrNonMPPayment = errors.New("payment has non-MPP attempts")

        // ErrMPPayment is returned if we try to register a non-MPP attempt for
        // a payment that already has an MPP attempt registered.
        ErrMPPayment = errors.New("payment has MPP attempts")

        // ErrMPPRecordInBlindedPayment is returned if we try to register an
        // attempt with an MPP record for a payment to a blinded path.
        ErrMPPRecordInBlindedPayment = errors.New("blinded payment cannot " +
                "contain MPP records")

        // ErrBlindedPaymentTotalAmountMismatch is returned if we try to
        // register an HTLC shard to a blinded route where the total amount
        // doesn't match existing shards.
        ErrBlindedPaymentTotalAmountMismatch = errors.New("blinded path " +
                "total amount mismatch")

        // ErrMPPPaymentAddrMismatch is returned if we try to register an MPP
        // shard where the payment address doesn't match existing shards.
        ErrMPPPaymentAddrMismatch = errors.New("payment address mismatch")

        // ErrMPPTotalAmountMismatch is returned if we try to register an MPP
        // shard where the total amount doesn't match existing shards.
        ErrMPPTotalAmountMismatch = errors.New("mp payment total amount " +
                "mismatch")

        // ErrPaymentPendingSettled is returned when we try to add a new
        // attempt to a payment that has at least one of its HTLCs settled.
        ErrPaymentPendingSettled = errors.New("payment has settled htlcs")

        // ErrPaymentPendingFailed is returned when we try to add a new attempt
        // to a payment that already has a failure reason.
        ErrPaymentPendingFailed = errors.New("payment has failure reason")

        // ErrSentExceedsTotal is returned if the payment's current total sent
        // amount exceed the total amount.
        ErrSentExceedsTotal = errors.New("total sent exceeds total amount")

        // errNoAttemptInfo is returned when no attempt info is stored yet.
        errNoAttemptInfo = errors.New("unable to find attempt info for " +
                "inflight payment")

        // errNoSequenceNrIndex is returned when an attempt to lookup a payment
        // index is made for a sequence number that is not indexed.
        errNoSequenceNrIndex = errors.New("payment sequence number index " +
                "does not exist")
)

// Payment operations related constants.
const (
        // paymentSeqBlockSize is the block size used when we batch allocate
        // payment sequences for future payments.
        paymentSeqBlockSize = 1000

        // paymentProgressLogInterval is the interval we use limiting the
        // logging output of payment processing.
        paymentProgressLogInterval = 30 * time.Second
)

// FailureReason encodes the reason a payment ultimately failed.
type FailureReason byte

const (
        // FailureReasonTimeout indicates that the payment did timeout before a
        // successful payment attempt was made.
        FailureReasonTimeout FailureReason = 0

        // FailureReasonNoRoute indicates no successful route to the
        // destination was found during path finding.
        FailureReasonNoRoute FailureReason = 1

        // FailureReasonError indicates that an unexpected error happened during
        // payment.
        FailureReasonError FailureReason = 2

        // FailureReasonPaymentDetails indicates that either the hash is unknown
        // or the final cltv delta or amount is incorrect.
        FailureReasonPaymentDetails FailureReason = 3

        // FailureReasonInsufficientBalance indicates that we didn't have enough
        // balance to complete the payment.
        FailureReasonInsufficientBalance FailureReason = 4

        // FailureReasonCanceled indicates that the payment was canceled by the
        // user.
        FailureReasonCanceled FailureReason = 5

        // TODO(joostjager): Add failure reasons for:
        // LocalLiquidityInsufficient, RemoteCapacityInsufficient.
)

// Error returns a human-readable error string for the FailureReason.
func (r FailureReason) Error() string {
        return r.String()
}

// String returns a human-readable FailureReason.
func (r FailureReason) String() string {
        switch r {
        case FailureReasonTimeout:
                return "timeout"
        case FailureReasonNoRoute:
                return "no_route"
        case FailureReasonError:
                return "error"
        case FailureReasonPaymentDetails:
                return "incorrect_payment_details"
        case FailureReasonInsufficientBalance:
                return "insufficient_balance"
        case FailureReasonCanceled:
                return "canceled"
        }

        return "unknown"
}

// KVPaymentsDB implements persistence for payments and payment attempts.
type KVPaymentsDB struct {
        paymentSeqMx     sync.Mutex
        currPaymentSeq   uint64
        storedPaymentSeq uint64
        db               *DB
}

// NewKVPaymentsDB creates a new instance of the KVPaymentsDB.
func NewKVPaymentsDB(db *DB) *KVPaymentsDB {
        return &KVPaymentsDB{
                db: db,
        }
}

// InitPayment checks or records the given PaymentCreationInfo with the DB,
// making sure it does not already exist as an in-flight payment. When this
// method returns successfully, the payment is guaranteed to be in the InFlight
// state.
func (p *KVPaymentsDB) InitPayment(paymentHash lntypes.Hash,
        info *PaymentCreationInfo) error {

        // Obtain a new sequence number for this payment. This is used
        // to sort the payments in order of creation, and also acts as
        // a unique identifier for each payment.
        sequenceNum, err := p.nextPaymentSequence()
        if err != nil {
                return err
        }

        var b bytes.Buffer
        if err := serializePaymentCreationInfo(&b, info); err != nil {
                return err
        }
        infoBytes := b.Bytes()

        var updateErr error
        err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
                // Reset the update error, to avoid carrying over an error
                // from a previous execution of the batched db transaction.
                updateErr = nil

                prefetchPayment(tx, paymentHash)
                bucket, err := createPaymentBucket(tx, paymentHash)
                if err != nil {
                        return err
                }

                // Get the existing status of this payment, if any.
                paymentStatus, err := fetchPaymentStatus(bucket)

                switch {
                // If no error is returned, it means we already have this
                // payment. We'll check the status to decide whether we allow
                // retrying the payment or return a specific error.
                case err == nil:
                        if err := paymentStatus.initializable(); err != nil {
                                updateErr = err
                                return nil
                        }

                // Otherwise, if the error is not `ErrPaymentNotInitiated`,
                // we'll return the error.
                case !errors.Is(err, ErrPaymentNotInitiated):
                        return err
                }

                // Before we set our new sequence number, we check whether this
                // payment has a previously set sequence number and remove its
                // index entry if it exists. This happens in the case where we
                // have a previously attempted payment which was left in a state
                // where we can retry.
                seqBytes := bucket.Get(paymentSequenceKey)
                if seqBytes != nil {
                        indexBucket := tx.ReadWriteBucket(paymentsIndexBucket)
                        if err := indexBucket.Delete(seqBytes); err != nil {
                                return err
                        }
                }

                // Once we have obtained a sequence number, we add an entry
                // to our index bucket which will map the sequence number to
                // our payment identifier.
                err = createPaymentIndexEntry(
                        tx, sequenceNum, info.PaymentIdentifier,
                )
                if err != nil {
                        return err
                }

                err = bucket.Put(paymentSequenceKey, sequenceNum)
                if err != nil {
                        return err
                }

                // Add the payment info to the bucket, which contains the
                // static information for this payment
                err = bucket.Put(paymentCreationInfoKey, infoBytes)
                if err != nil {
                        return err
                }

                // We'll delete any lingering HTLCs to start with, in case we
                // are initializing a payment that was attempted earlier, but
                // left in a state where we could retry.
                err = bucket.DeleteNestedBucket(paymentHtlcsBucket)
                if err != nil && !errors.Is(err, kvdb.ErrBucketNotFound) {
                        return err
                }

                // Also delete any lingering failure info now that we are
                // re-attempting.
                return bucket.Delete(paymentFailInfoKey)
        })
        if err != nil {
                return fmt.Errorf("unable to init payment: %w", err)
        }

        return updateErr
}

// DeleteFailedAttempts deletes all failed htlcs for a payment if configured
// by the KVPaymentsDB db.
func (p *KVPaymentsDB) DeleteFailedAttempts(hash lntypes.Hash) error {
        if !p.db.keepFailedPaymentAttempts {
                const failedHtlcsOnly = true
                err := p.DeletePayment(hash, failedHtlcsOnly)
                if err != nil {
                        return err
                }
        }

        return nil
}

// paymentIndexTypeHash is a payment index type which indicates that we have
// created an index of payment sequence number to payment hash.
type paymentIndexType uint8

// paymentIndexTypeHash is a payment index type which indicates that we have
// created an index of payment sequence number to payment hash.
const paymentIndexTypeHash paymentIndexType = 0

// createPaymentIndexEntry creates a payment hash typed index for a payment. The
// index produced contains a payment index type (which can be used in future to
// signal different payment index types) and the payment identifier.
func createPaymentIndexEntry(tx kvdb.RwTx, sequenceNumber []byte,
        id lntypes.Hash) error {

        var b bytes.Buffer
        if err := WriteElements(&b, paymentIndexTypeHash, id[:]); err != nil {
                return err
        }

        indexes := tx.ReadWriteBucket(paymentsIndexBucket)

        return indexes.Put(sequenceNumber, b.Bytes())
}

// deserializePaymentIndex deserializes a payment index entry. This function
// currently only supports deserialization of payment hash indexes, and will
// fail for other types.
func deserializePaymentIndex(r io.Reader) (lntypes.Hash, error) {
        var (
                indexType   paymentIndexType
                paymentHash []byte
        )

        if err := ReadElements(r, &indexType, &paymentHash); err != nil {
                return lntypes.Hash{}, err
        }

        // While we only have on payment index type, we do not need to use our
        // index type to deserialize the index. However, we sanity check that
        // this type is as expected, since we had to read it out anyway.
        if indexType != paymentIndexTypeHash {
                return lntypes.Hash{}, fmt.Errorf("unknown payment index "+
                        "type: %v", indexType)
        }

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

        return hash, nil
}

// RegisterAttempt atomically records the provided HTLCAttemptInfo to the
// DB.
func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
        attempt *HTLCAttemptInfo) (*MPPayment, error) {

        // Serialize the information before opening the db transaction.
        var a bytes.Buffer
        err := serializeHTLCAttemptInfo(&a, attempt)
        if err != nil {
                return nil, err
        }
        htlcInfoBytes := a.Bytes()

        htlcIDBytes := make([]byte, 8)
        binary.BigEndian.PutUint64(htlcIDBytes, attempt.AttemptID)

        var payment *MPPayment
        err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
                prefetchPayment(tx, paymentHash)
                bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
                if err != nil {
                        return err
                }

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

                // Check if registering a new attempt is allowed.
                if err := payment.Registrable(); err != nil {
                        return err
                }

                // If the final hop has encrypted data, then we know this is a
                // blinded payment. In blinded payments, MPP records are not set
                // for split payments and the recipient is responsible for using
                // a consistent PathID across the various encrypted data
                // payloads that we received from them for this payment. All we
                // need to check is that the total amount field for each HTLC
                // in the split payment is correct.
                isBlinded := len(attempt.Route.FinalHop().EncryptedData) != 0

                // Make sure any existing shards match the new one with regards
                // to MPP options.
                mpp := attempt.Route.FinalHop().MPP

                // MPP records should not be set for attempts to blinded paths.
                if isBlinded && mpp != nil {
                        return ErrMPPRecordInBlindedPayment
                }

                for _, h := range payment.InFlightHTLCs() {
                        hMpp := h.Route.FinalHop().MPP

                        // If this is a blinded payment, then no existing HTLCs
                        // should have MPP records.
                        if isBlinded && hMpp != nil {
                                return ErrMPPRecordInBlindedPayment
                        }

                        // If this is a blinded payment, then we just need to
                        // check that the TotalAmtMsat field for this shard
                        // is equal to that of any other shard in the same
                        // payment.
                        if isBlinded {
                                if attempt.Route.FinalHop().TotalAmtMsat !=
                                        h.Route.FinalHop().TotalAmtMsat {

                                        //nolint:ll
                                        return ErrBlindedPaymentTotalAmountMismatch
                                }

                                continue
                        }

                        switch {
                        // We tried to register a non-MPP attempt for a MPP
                        // payment.
                        case mpp == nil && hMpp != nil:
                                return ErrMPPayment

                        // We tried to register a MPP shard for a non-MPP
                        // payment.
                        case mpp != nil && hMpp == nil:
                                return ErrNonMPPayment

                        // Non-MPP payment, nothing more to validate.
                        case mpp == nil:
                                continue
                        }

                        // Check that MPP options match.
                        if mpp.PaymentAddr() != hMpp.PaymentAddr() {
                                return ErrMPPPaymentAddrMismatch
                        }

                        if mpp.TotalMsat() != hMpp.TotalMsat() {
                                return ErrMPPTotalAmountMismatch
                        }
                }

                // If this is a non-MPP attempt, it must match the total amount
                // exactly. Note that a blinded payment is considered an MPP
                // attempt.
                amt := attempt.Route.ReceiverAmt()
                if !isBlinded && mpp == nil && amt != payment.Info.Value {
                        return ErrValueMismatch
                }

                // Ensure we aren't sending more than the total payment amount.
                sentAmt, _ := payment.SentAmt()
                if sentAmt+amt > payment.Info.Value {
                        return fmt.Errorf("%w: attempted=%v, payment amount="+
                                "%v", ErrValueExceedsAmt, sentAmt+amt,
                                payment.Info.Value)
                }

                htlcsBucket, err := bucket.CreateBucketIfNotExists(
                        paymentHtlcsBucket,
                )
                if err != nil {
                        return err
                }

                err = htlcsBucket.Put(
                        htlcBucketKey(htlcAttemptInfoKey, htlcIDBytes),
                        htlcInfoBytes,
                )
                if err != nil {
                        return err
                }

                // Retrieve attempt info for the notification.
                payment, err = fetchPayment(bucket)

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

        return payment, err
}

// SettleAttempt marks the given attempt settled with the preimage. If this is
// a multi shard payment, this might implicitly mean that the full payment
// succeeded.
//
// After invoking this method, InitPayment should always return an error to
// prevent us from making duplicate payments to the same payment hash. The
// provided preimage is atomically saved to the DB for record keeping.
func (p *KVPaymentsDB) SettleAttempt(hash lntypes.Hash,
        attemptID uint64, settleInfo *HTLCSettleInfo) (*MPPayment, error) {

        var b bytes.Buffer
        if err := serializeHTLCSettleInfo(&b, settleInfo); err != nil {
                return nil, err
        }
        settleBytes := b.Bytes()

        return p.updateHtlcKey(hash, attemptID, htlcSettleInfoKey, settleBytes)
}

// FailAttempt marks the given payment attempt failed.
func (p *KVPaymentsDB) FailAttempt(hash lntypes.Hash,
        attemptID uint64, failInfo *HTLCFailInfo) (*MPPayment, error) {

        var b bytes.Buffer
        if err := serializeHTLCFailInfo(&b, failInfo); err != nil {
                return nil, err
        }
        failBytes := b.Bytes()

        return p.updateHtlcKey(hash, attemptID, htlcFailInfoKey, failBytes)
}

// updateHtlcKey updates a database key for the specified htlc.
func (p *KVPaymentsDB) updateHtlcKey(paymentHash lntypes.Hash,
        attemptID uint64, key, value []byte) (*MPPayment, error) {

        aid := make([]byte, 8)
        binary.BigEndian.PutUint64(aid, attemptID)

        var payment *MPPayment
        err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
                payment = nil

                prefetchPayment(tx, paymentHash)
                bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
                if err != nil {
                        return err
                }

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

                // We can only update keys of in-flight payments. We allow
                // updating keys even if the payment has reached a terminal
                // condition, since the HTLC outcomes must still be updated.
                if err := p.Status.updatable(); err != nil {
                        return err
                }

                htlcsBucket := bucket.NestedReadWriteBucket(paymentHtlcsBucket)
                if htlcsBucket == nil {
                        return fmt.Errorf("htlcs bucket not found")
                }

                attemptInfo := htlcsBucket.Get(
                        htlcBucketKey(htlcAttemptInfoKey, aid),
                )
                if attemptInfo == nil {
                        return fmt.Errorf("HTLC with ID %v not registered",
                                attemptID)
                }

                failInfo := htlcsBucket.Get(
                        htlcBucketKey(htlcFailInfoKey, aid),
                )
                if failInfo != nil {
                        return ErrAttemptAlreadyFailed
                }

                settleInfo := htlcsBucket.Get(
                        htlcBucketKey(htlcSettleInfoKey, aid),
                )
                if settleInfo != nil {
                        return ErrAttemptAlreadySettled
                }

                // Add or update the key for this htlc.
                err = htlcsBucket.Put(htlcBucketKey(key, aid), value)
                if err != nil {
                        return err
                }

                // Retrieve attempt info for the notification.
                payment, err = fetchPayment(bucket)

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

        return payment, err
}

// Fail transitions a payment into the Failed state, and records the reason the
// payment failed. After invoking this method, InitPayment should return nil on
// its next call for this payment hash, allowing the switch to make a
// subsequent payment.
func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
        reason FailureReason) (*MPPayment, error) {

        var (
                updateErr error
                payment   *MPPayment
        )
        err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
                // Reset the update error, to avoid carrying over an error
                // from a previous execution of the batched db transaction.
                updateErr = nil
                payment = nil

                prefetchPayment(tx, paymentHash)
                bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
                if errors.Is(err, ErrPaymentNotInitiated) {
                        updateErr = ErrPaymentNotInitiated
                        return nil
                } else if err != nil {
                        return err
                }

                // We mark the payment as failed as long as it is known. This
                // lets the last attempt to fail with a terminal write its
                // failure to the KVPaymentsDB without synchronizing with
                // other attempts.
                _, err = fetchPaymentStatus(bucket)
                if errors.Is(err, ErrPaymentNotInitiated) {
                        updateErr = ErrPaymentNotInitiated
                        return nil
                } else if err != nil {
                        return err
                }

                // Put the failure reason in the bucket for record keeping.
                v := []byte{byte(reason)}
                err = bucket.Put(paymentFailInfoKey, v)
                if err != nil {
                        return err
                }

                // Retrieve attempt info for the notification, if available.
                payment, err = fetchPayment(bucket)
                if err != nil {
                        return err
                }

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

        return payment, updateErr
}

// FetchPayment returns information about a payment from the database.
func (p *KVPaymentsDB) FetchPayment(paymentHash lntypes.Hash) (
        *MPPayment, error) {

        var payment *MPPayment
        err := kvdb.View(p.db, func(tx kvdb.RTx) error {
                prefetchPayment(tx, paymentHash)
                bucket, err := fetchPaymentBucket(tx, paymentHash)
                if err != nil {
                        return err
                }

                payment, err = fetchPayment(bucket)

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

        return payment, nil
}

// prefetchPayment attempts to prefetch as much of the payment as possible to
// reduce DB roundtrips.
func prefetchPayment(tx kvdb.RTx, paymentHash lntypes.Hash) {
        rb := kvdb.RootBucket(tx)
        kvdb.Prefetch(
                rb,
                []string{
                        // Prefetch all keys in the payment's bucket.
                        string(paymentsRootBucket),
                        string(paymentHash[:]),
                },
                []string{
                        // Prefetch all keys in the payment's htlc bucket.
                        string(paymentsRootBucket),
                        string(paymentHash[:]),
                        string(paymentHtlcsBucket),
                },
        )
}

// createPaymentBucket creates or fetches the sub-bucket assigned to this
// payment hash.
func createPaymentBucket(tx kvdb.RwTx, paymentHash lntypes.Hash) (
        kvdb.RwBucket, error) {

        payments, err := tx.CreateTopLevelBucket(paymentsRootBucket)
        if err != nil {
                return nil, err
        }

        return payments.CreateBucketIfNotExists(paymentHash[:])
}

// fetchPaymentBucket fetches the sub-bucket assigned to this payment hash. If
// the bucket does not exist, it returns ErrPaymentNotInitiated.
func fetchPaymentBucket(tx kvdb.RTx, paymentHash lntypes.Hash) (
        kvdb.RBucket, error) {

        payments := tx.ReadBucket(paymentsRootBucket)
        if payments == nil {
                return nil, ErrPaymentNotInitiated
        }

        bucket := payments.NestedReadBucket(paymentHash[:])
        if bucket == nil {
                return nil, ErrPaymentNotInitiated
        }

        return bucket, nil
}

// fetchPaymentBucketUpdate is identical to fetchPaymentBucket, but it returns a
// bucket that can be written to.
func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
        kvdb.RwBucket, error) {

        payments := tx.ReadWriteBucket(paymentsRootBucket)
        if payments == nil {
                return nil, ErrPaymentNotInitiated
        }

        bucket := payments.NestedReadWriteBucket(paymentHash[:])
        if bucket == nil {
                return nil, ErrPaymentNotInitiated
        }

        return bucket, nil
}

// nextPaymentSequence returns the next sequence number to store for a new
// payment.
func (p *KVPaymentsDB) nextPaymentSequence() ([]byte, error) {
        p.paymentSeqMx.Lock()
        defer p.paymentSeqMx.Unlock()

        // Set a new upper bound in the DB every 1000 payments to avoid
        // conflicts on the sequence when using etcd.
        if p.currPaymentSeq == p.storedPaymentSeq {
                var currPaymentSeq, newUpperBound uint64
                if err := kvdb.Update(p.db.Backend, func(tx kvdb.RwTx) error {
                        paymentsBucket, err := tx.CreateTopLevelBucket(
                                paymentsRootBucket,
                        )
                        if err != nil {
                                return err
                        }

                        currPaymentSeq = paymentsBucket.Sequence()
                        newUpperBound = currPaymentSeq + paymentSeqBlockSize

                        return paymentsBucket.SetSequence(newUpperBound)
                }, func() {}); err != nil {
                        return nil, err
                }

                // We lazy initialize the cached currPaymentSeq here using the
                // first nextPaymentSequence() call. This if statement will auto
                // initialize our stored currPaymentSeq, since by default both
                // this variable and storedPaymentSeq are zero which in turn
                // will have us fetch the current values from the DB.
                if p.currPaymentSeq == 0 {
                        p.currPaymentSeq = currPaymentSeq
                }

                p.storedPaymentSeq = newUpperBound
        }

        p.currPaymentSeq++
        b := make([]byte, 8)
        binary.BigEndian.PutUint64(b, p.currPaymentSeq)

        return b, nil
}

// fetchPaymentStatus fetches the payment status of the payment. If the payment
// isn't found, it will return error `ErrPaymentNotInitiated`.
func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
        // Creation info should be set for all payments, regardless of state.
        // If not, it is unknown.
        if bucket.Get(paymentCreationInfoKey) == nil {
                return 0, ErrPaymentNotInitiated
        }

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

        return payment.Status, nil
}

// FetchInFlightPayments returns all payments with status InFlight.
func (p *KVPaymentsDB) FetchInFlightPayments() ([]*MPPayment, error) {
        var (
                inFlights      []*MPPayment
                start          = time.Now()
                lastLogTime    = time.Now()
                processedCount int
        )

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

                return payments.ForEach(func(k, _ []byte) error {
                        bucket := payments.NestedReadBucket(k)
                        if bucket == nil {
                                return fmt.Errorf("non bucket element")
                        }

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

                        processedCount++
                        if time.Since(lastLogTime) >=
                                paymentProgressLogInterval {

                                log.Debugf("Scanning inflight payments "+
                                        "(in progress), processed %d, last "+
                                        "processed payment: %v", processedCount,
                                        p.Info)

                                lastLogTime = time.Now()
                        }

                        // Skip the payment if it's terminated.
                        if p.Terminated() {
                                return nil
                        }

                        inFlights = append(inFlights, p)

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

        elapsed := time.Since(start)
        log.Debugf("Completed scanning for inflight payments: "+
                "total_processed=%d, found_inflight=%d, elapsed=%v",
                processedCount, len(inFlights),
                elapsed.Round(time.Millisecond))

        return inFlights, nil
}

// PaymentCreationInfo is the information necessary to have ready when
// initiating a payment, moving it into state InFlight.
type PaymentCreationInfo struct {
        // PaymentIdentifier is the hash this payment is paying to in case of
        // non-AMP payments, and the SetID for AMP payments.
        PaymentIdentifier lntypes.Hash

        // Value is the amount we are paying.
        Value lnwire.MilliSatoshi

        // CreationTime is the time when this payment was initiated.
        CreationTime time.Time

        // PaymentRequest is the full payment request, if any.
        PaymentRequest []byte

        // FirstHopCustomRecords are the TLV records that are to be sent to the
        // first hop of this payment. These records will be transmitted via the
        // wire message only and therefore do not affect the onion payload size.
        FirstHopCustomRecords lnwire.CustomRecords
}

// String returns a human-readable description of the payment creation info.
func (p *PaymentCreationInfo) String() string {
        return fmt.Sprintf("payment_id=%v, amount=%v, created_at=%v",
                p.PaymentIdentifier, p.Value, p.CreationTime)
}

// 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
func (p *KVPaymentsDB) FetchPayments() ([]*MPPayment, error) {
        var payments []*MPPayment

        err := kvdb.View(p.db, 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) (*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) (*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 []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 *FailureReason
        b := bucket.Get(paymentFailInfoKey)
        if b != nil {
                reason := FailureReason(b[0])
                failureReason = &reason
        }

        // Create a new payment.
        payment := &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) ([]HTLCAttempt, error) {
        htlcsMap := make(map[uint64]*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] = &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, 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([]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) (*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) (*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) (*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 []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
}

// PaymentsQuery represents a query to the payments database starting or ending
// at a certain offset index. The number of retrieved records can be limited.
type PaymentsQuery struct {
        // IndexOffset determines the starting point of the payments query and
        // is always exclusive. In normal order, the query starts at the next
        // higher (available) index compared to IndexOffset. In reversed order,
        // the query ends at the next lower (available) index compared to the
        // IndexOffset. In the case of a zero index_offset, the query will start
        // with the oldest payment when paginating forwards, or will end with
        // the most recent payment when paginating backwards.
        IndexOffset uint64

        // MaxPayments is the maximal number of payments returned in the
        // payments query.
        MaxPayments uint64

        // Reversed gives a meaning to the IndexOffset. If reversed is set to
        // true, the query will fetch payments with indices lower than the
        // IndexOffset, otherwise, it will return payments with indices greater
        // than the IndexOffset.
        Reversed bool

        // If IncludeIncomplete is true, then return payments that have not yet
        // fully completed. This means that pending payments, as well as failed
        // payments will show up if this field is set to true.
        IncludeIncomplete bool

        // CountTotal indicates that all payments currently present in the
        // payment index (complete and incomplete) should be counted.
        CountTotal bool

        // CreationDateStart, expressed in Unix seconds, if set, filters out
        // all payments with a creation date greater than or equal to it.
        CreationDateStart int64

        // CreationDateEnd, expressed in Unix seconds, if set, filters out all
        // payments with a creation date less than or equal to it.
        CreationDateEnd int64
}

// PaymentsResponse contains the result of a query to the payments database.
// It includes the set of payments that match the query and integers which
// represent the index of the first and last item returned in the series of
// payments. These integers allow callers to resume their query in the event
// that the query's response exceeds the max number of returnable events.
type PaymentsResponse struct {
        // Payments is the set of payments returned from the database for the
        // PaymentsQuery.
        Payments []*MPPayment

        // FirstIndexOffset is the index of the first element in the set of
        // returned MPPayments. Callers can use this to resume their query
        // in the event that the slice has too many events to fit into a single
        // response. The offset can be used to continue reverse pagination.
        FirstIndexOffset uint64

        // LastIndexOffset is the index of the last element in the set of
        // returned MPPayments. Callers can use this to resume their query
        // in the event that the slice has too many events to fit into a single
        // response. The offset can be used to continue forward pagination.
        LastIndexOffset uint64

        // TotalCount represents the total number of payments that are currently
        // stored in the payment database. This will only be set if the
        // CountTotal field in the query was set to true.
        TotalCount uint64
}

// 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 (p *KVPaymentsDB) QueryPayments(_ context.Context,
        query PaymentsQuery) (PaymentsResponse,
        error) {

        var resp PaymentsResponse

        if err := kvdb.View(p.db, 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 != 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 = PaymentsResponse{}
        }); 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) (*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, 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, ErrNoDuplicateBucket
        }

        var duplicatePayment *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 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, 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 (p *KVPaymentsDB) DeletePayment(paymentHash lntypes.Hash,
        failedHtlcsOnly bool) error {

        return kvdb.Update(p.db, 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 (p *KVPaymentsDB) DeletePayments(failedOnly, failedHtlcsOnly bool) (int,
        error) {

        var numPayments int
        err := kvdb.Update(p.db, 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 != 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 *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) (*PaymentCreationInfo,
        error) {

        var scratch [8]byte

        c := &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 *HTLCAttemptInfo) error {
        if err := WriteElements(w, a.sessionKey); 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) (*HTLCAttemptInfo, error) {
        a := &HTLCAttemptInfo{}
        err := ReadElements(r, &a.sessionKey)
        if err != nil {
                return nil, err
        }

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

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