From aaa1ee0b746e2832dcc79ee095e55c344c6bc1c9 Mon Sep 17 00:00:00 2001 From: ziggie Date: Fri, 1 Aug 2025 20:19:33 +0200 Subject: [PATCH] channeldb: move kv related code We move all kv particular code which was in the payments.go file to the kv related file. Code which will be backend independant will remain in the payments.go file although only the kv backend is currently supported. --- channeldb/payments.go | 1328 --------------------------- channeldb/payments_kv_store.go | 1324 ++++++++++++++++++++++++++ channeldb/payments_kv_store_test.go | 355 +++++++ channeldb/payments_test.go | 358 -------- 4 files changed, 1679 insertions(+), 1686 deletions(-) diff --git a/channeldb/payments.go b/channeldb/payments.go index 51457671a..a23f891d0 100644 --- a/channeldb/payments.go +++ b/channeldb/payments.go @@ -1,126 +1,11 @@ package channeldb import ( - "bytes" - "encoding/binary" - "errors" "fmt" - "io" - "sort" "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 - // | - // |-- - // | |--sequence-key: - // | |--creation-info-key: - // | |--fail-info-key: <(optional) fail info> - // | | - // | |--payment-htlcs-bucket (shard-bucket) - // | | | - // | | |-- ai: - // | | |-- si: <(optional) settle info> - // | | |-- fi: <(optional) fail info> - // | | | - // | | ... - // | | - // | | - // | |--duplicate-bucket (only for old, completed payments) - // | | - // | |-- - // | | |--sequence-key: - // | | |--creation-info-key: - // | | |--ai: - // | | |--si: - // | | |--fi: - // | | - // | |-- - // | | | - // | ... ... - // | - // |-- - // | | - // | ... - // ... - // - 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 - // |--: - // |--... - // |--: - 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") ) // FailureReason encodes the reason a payment ultimately failed. @@ -208,253 +93,6 @@ func (p *PaymentCreationInfo) String() string { 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 (d *DB) FetchPayments() ([]*MPPayment, error) { - var payments []*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) (*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 { @@ -522,969 +160,3 @@ type PaymentsResponse struct { // 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 (d *DB) QueryPayments(query PaymentsQuery) (PaymentsResponse, error) { - var resp PaymentsResponse - - 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 != 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 (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 != 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 -} diff --git a/channeldb/payments_kv_store.go b/channeldb/payments_kv_store.go index 24124ba7a..198267bbc 100644 --- a/channeldb/payments_kv_store.go +++ b/channeldb/payments_kv_store.go @@ -6,11 +6,18 @@ import ( "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" ) const ( @@ -130,6 +137,110 @@ var ( "does not exist") ) +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 + // | + // |-- + // | |--sequence-key: + // | |--creation-info-key: + // | |--fail-info-key: <(optional) fail info> + // | | + // | |--payment-htlcs-bucket (shard-bucket) + // | | | + // | | |-- ai: + // | | |-- si: <(optional) settle info> + // | | |-- fi: <(optional) fail info> + // | | | + // | | ... + // | | + // | | + // | |--duplicate-bucket (only for old, completed payments) + // | | + // | |-- + // | | |--sequence-key: + // | | |--creation-info-key: + // | | |--ai: + // | | |--si: + // | | |--fi: + // | | + // | |-- + // | | | + // | ... ... + // | + // |-- + // | | + // | ... + // ... + // + 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 + // |--: + // |--... + // |--: + 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") +) + // KVPaymentsDB implements persistence for payments and payment attempts. type KVPaymentsDB struct { paymentSeqMx sync.Mutex @@ -834,3 +945,1216 @@ func (p *KVPaymentsDB) FetchInFlightPayments() ([]*MPPayment, error) { return inFlights, nil } + +// 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 (d *DB) FetchPayments() ([]*MPPayment, error) { + var payments []*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) (*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 +} + +// 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 PaymentsQuery) (PaymentsResponse, error) { + var resp PaymentsResponse + + 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 != 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 (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 != 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 +} diff --git a/channeldb/payments_kv_store_test.go b/channeldb/payments_kv_store_test.go index 6ed39f605..dfe36bce2 100644 --- a/channeldb/payments_kv_store_test.go +++ b/channeldb/payments_kv_store_test.go @@ -15,7 +15,10 @@ import ( "github.com/davecgh/go-spew/spew" "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" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -1427,3 +1430,355 @@ func assertPayments(t *testing.T, db *DB, payments []*payment) { // Check that each payment we want to assert exists in the database. require.Equal(t, payments, p) } + +func makeFakeInfo(t *testing.T) (*PaymentCreationInfo, *HTLCAttemptInfo) { + var preimg lntypes.Preimage + copy(preimg[:], rev[:]) + + hash := preimg.Hash() + + c := &PaymentCreationInfo{ + PaymentIdentifier: hash, + Value: 1000, + // Use single second precision to avoid false positive test + // failures due to the monotonic time component. + CreationTime: time.Unix(time.Now().Unix(), 0), + PaymentRequest: []byte("test"), + } + + a, err := NewHtlcAttempt( + 44, priv, testRoute, time.Unix(100, 0), &hash, + ) + require.NoError(t, err) + + return c, &a.HTLCAttemptInfo +} + +func TestSentPaymentSerialization(t *testing.T) { + t.Parallel() + + c, s := makeFakeInfo(t) + + var b bytes.Buffer + require.NoError(t, serializePaymentCreationInfo(&b, c), "serialize") + + // Assert the length of the serialized creation info is as expected, + // without any custom records. + baseLength := 32 + 8 + 8 + 4 + len(c.PaymentRequest) + require.Len(t, b.Bytes(), baseLength) + + newCreationInfo, err := deserializePaymentCreationInfo(&b) + require.NoError(t, err, "deserialize") + require.Equal(t, c, newCreationInfo) + + b.Reset() + + // Now we add some custom records to the creation info and serialize it + // again. + c.FirstHopCustomRecords = lnwire.CustomRecords{ + lnwire.MinCustomRecordsTlvType: []byte{1, 2, 3}, + } + require.NoError(t, serializePaymentCreationInfo(&b, c), "serialize") + + newCreationInfo, err = deserializePaymentCreationInfo(&b) + require.NoError(t, err, "deserialize") + require.Equal(t, c, newCreationInfo) + + b.Reset() + require.NoError(t, serializeHTLCAttemptInfo(&b, s), "serialize") + + newWireInfo, err := deserializeHTLCAttemptInfo(&b) + require.NoError(t, err, "deserialize") + + // First we verify all the records match up properly. + require.Equal(t, s.Route, newWireInfo.Route) + + // We now add the new fields and custom records to the route and + // serialize it again. + b.Reset() + s.Route.FirstHopAmount = tlv.NewRecordT[tlv.TlvType0]( + tlv.NewBigSizeT(lnwire.MilliSatoshi(1234)), + ) + s.Route.FirstHopWireCustomRecords = lnwire.CustomRecords{ + lnwire.MinCustomRecordsTlvType + 3: []byte{4, 5, 6}, + } + require.NoError(t, serializeHTLCAttemptInfo(&b, s), "serialize") + + newWireInfo, err = deserializeHTLCAttemptInfo(&b) + require.NoError(t, err, "deserialize") + require.Equal(t, s.Route, newWireInfo.Route) + + err = newWireInfo.attachOnionBlobAndCircuit() + require.NoError(t, err) + + // Clear routes to allow DeepEqual to compare the remaining fields. + newWireInfo.Route = route.Route{} + s.Route = route.Route{} + newWireInfo.AttemptID = s.AttemptID + + // Call session key method to set our cached session key so we can use + // DeepEqual, and assert that our key equals the original key. + require.Equal(t, s.cachedSessionKey, newWireInfo.SessionKey()) + + require.Equal(t, s, newWireInfo) +} + +// TestRouteSerialization tests serialization of a regular and blinded route. +func TestRouteSerialization(t *testing.T) { + t.Parallel() + + testSerializeRoute(t, testRoute) + testSerializeRoute(t, testBlindedRoute) +} + +func testSerializeRoute(t *testing.T, route route.Route) { + var b bytes.Buffer + err := SerializeRoute(&b, route) + require.NoError(t, err) + + r := bytes.NewReader(b.Bytes()) + route2, err := DeserializeRoute(r) + require.NoError(t, err) + + reflect.DeepEqual(route, route2) +} + +// deletePayment removes a payment with paymentHash from the payments database. +func deletePayment(t *testing.T, db *DB, paymentHash lntypes.Hash, seqNr uint64) { + t.Helper() + + err := kvdb.Update(db, func(tx kvdb.RwTx) error { + payments := tx.ReadWriteBucket(paymentsRootBucket) + + // Delete the payment bucket. + err := payments.DeleteNestedBucket(paymentHash[:]) + if err != nil { + return err + } + + key := make([]byte, 8) + byteOrder.PutUint64(key, seqNr) + + // Delete the index that references this payment. + indexes := tx.ReadWriteBucket(paymentsIndexBucket) + return indexes.Delete(key) + }, func() {}) + + if err != nil { + t.Fatalf("could not delete "+ + "payment: %v", err) + } +} + +// TestFetchPaymentWithSequenceNumber tests lookup of payments with their +// sequence number. It sets up one payment with no duplicates, and another with +// two duplicates in its duplicates bucket then uses these payments to test the +// case where a specific duplicate is not found and the duplicates bucket is not +// present when we expect it to be. +func TestFetchPaymentWithSequenceNumber(t *testing.T) { + db, err := MakeTestDB(t) + require.NoError(t, err) + + paymentDB := NewKVPaymentsDB(db) + + // Generate a test payment which does not have duplicates. + noDuplicates, _, _, err := genInfo(t) + require.NoError(t, err) + + // Create a new payment entry in the database. + err = paymentDB.InitPayment( + noDuplicates.PaymentIdentifier, noDuplicates, + ) + require.NoError(t, err) + + // Fetch the payment so we can get its sequence nr. + noDuplicatesPayment, err := paymentDB.FetchPayment( + noDuplicates.PaymentIdentifier, + ) + require.NoError(t, err) + + // Generate a test payment which we will add duplicates to. + hasDuplicates, _, preimg, err := genInfo(t) + require.NoError(t, err) + + // Create a new payment entry in the database. + err = paymentDB.InitPayment( + hasDuplicates.PaymentIdentifier, hasDuplicates, + ) + require.NoError(t, err) + + // Fetch the payment so we can get its sequence nr. + hasDuplicatesPayment, err := paymentDB.FetchPayment( + hasDuplicates.PaymentIdentifier, + ) + require.NoError(t, err) + + // We declare the sequence numbers used here so that we can reference + // them in tests. + var ( + duplicateOneSeqNr = hasDuplicatesPayment.SequenceNum + 1 + duplicateTwoSeqNr = hasDuplicatesPayment.SequenceNum + 2 + ) + + // Add two duplicates to our second payment. + appendDuplicatePayment( + t, db, hasDuplicates.PaymentIdentifier, duplicateOneSeqNr, preimg, + ) + appendDuplicatePayment( + t, db, hasDuplicates.PaymentIdentifier, duplicateTwoSeqNr, preimg, + ) + + tests := []struct { + name string + paymentHash lntypes.Hash + sequenceNumber uint64 + expectedErr error + }{ + { + name: "lookup payment without duplicates", + paymentHash: noDuplicates.PaymentIdentifier, + sequenceNumber: noDuplicatesPayment.SequenceNum, + expectedErr: nil, + }, + { + name: "lookup payment with duplicates", + paymentHash: hasDuplicates.PaymentIdentifier, + sequenceNumber: hasDuplicatesPayment.SequenceNum, + expectedErr: nil, + }, + { + name: "lookup first duplicate", + paymentHash: hasDuplicates.PaymentIdentifier, + sequenceNumber: duplicateOneSeqNr, + expectedErr: nil, + }, + { + name: "lookup second duplicate", + paymentHash: hasDuplicates.PaymentIdentifier, + sequenceNumber: duplicateTwoSeqNr, + expectedErr: nil, + }, + { + name: "lookup non-existent duplicate", + paymentHash: hasDuplicates.PaymentIdentifier, + sequenceNumber: 999999, + expectedErr: ErrDuplicateNotFound, + }, + { + name: "lookup duplicate, no duplicates bucket", + paymentHash: noDuplicates.PaymentIdentifier, + sequenceNumber: duplicateTwoSeqNr, + expectedErr: ErrNoDuplicateBucket, + }, + } + + for _, test := range tests { + test := test + + t.Run(test.name, func(t *testing.T) { + err := kvdb.Update( + db, func(tx walletdb.ReadWriteTx) error { + var seqNrBytes [8]byte + byteOrder.PutUint64( + seqNrBytes[:], test.sequenceNumber, + ) + + _, err := fetchPaymentWithSequenceNumber( + tx, test.paymentHash, seqNrBytes[:], + ) + return err + }, func() {}, + ) + require.Equal(t, test.expectedErr, err) + }) + } +} + +// appendDuplicatePayment adds a duplicate payment to an existing payment. Note +// that this function requires a unique sequence number. +// +// This code is *only* intended to replicate legacy duplicate payments in lnd, +// our current schema does not allow duplicates. +func appendDuplicatePayment(t *testing.T, db *DB, paymentHash lntypes.Hash, + seqNr uint64, preImg lntypes.Preimage) { + + err := kvdb.Update(db, func(tx walletdb.ReadWriteTx) error { + bucket, err := fetchPaymentBucketUpdate( + tx, paymentHash, + ) + if err != nil { + return err + } + + // Create the duplicates bucket if it is not + // present. + dup, err := bucket.CreateBucketIfNotExists( + duplicatePaymentsBucket, + ) + if err != nil { + return err + } + + var sequenceKey [8]byte + byteOrder.PutUint64(sequenceKey[:], seqNr) + + // Create duplicate payments for the two dup + // sequence numbers we've setup. + putDuplicatePayment(t, dup, sequenceKey[:], paymentHash, preImg) + + // Finally, once we have created our entry we add an index for + // it. + err = createPaymentIndexEntry(tx, sequenceKey[:], paymentHash) + require.NoError(t, err) + + return nil + }, func() {}) + require.NoError(t, err, "could not create payment") +} + +// putDuplicatePayment creates a duplicate payment in the duplicates bucket +// provided with the minimal information required for successful reading. +func putDuplicatePayment(t *testing.T, duplicateBucket kvdb.RwBucket, + sequenceKey []byte, paymentHash lntypes.Hash, + preImg lntypes.Preimage) { + + paymentBucket, err := duplicateBucket.CreateBucketIfNotExists( + sequenceKey, + ) + require.NoError(t, err) + + err = paymentBucket.Put(duplicatePaymentSequenceKey, sequenceKey) + require.NoError(t, err) + + // Generate fake information for the duplicate payment. + info, _, _, err := genInfo(t) + require.NoError(t, err) + + // Write the payment info to disk under the creation info key. This code + // is copied rather than using serializePaymentCreationInfo to ensure + // we always write in the legacy format used by duplicate payments. + var b bytes.Buffer + var scratch [8]byte + _, err = b.Write(paymentHash[:]) + require.NoError(t, err) + + byteOrder.PutUint64(scratch[:], uint64(info.Value)) + _, err = b.Write(scratch[:]) + require.NoError(t, err) + + err = serializeTime(&b, info.CreationTime) + require.NoError(t, err) + + byteOrder.PutUint32(scratch[:4], 0) + _, err = b.Write(scratch[:4]) + require.NoError(t, err) + + // Get the PaymentCreationInfo. + err = paymentBucket.Put(duplicatePaymentCreationInfoKey, b.Bytes()) + require.NoError(t, err) + + // Duolicate payments are only stored for successes, so add the + // preimage. + err = paymentBucket.Put(duplicatePaymentSettleInfoKey, preImg[:]) + require.NoError(t, err) +} diff --git a/channeldb/payments_test.go b/channeldb/payments_test.go index fdc68fbf9..83f85ba99 100644 --- a/channeldb/payments_test.go +++ b/channeldb/payments_test.go @@ -1,7 +1,6 @@ package channeldb import ( - "bytes" "fmt" "math" "reflect" @@ -9,14 +8,9 @@ import ( "time" "github.com/btcsuite/btcd/btcec/v2" - "github.com/btcsuite/btcwallet/walletdb" "github.com/davecgh/go-spew/spew" - "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" "github.com/stretchr/testify/require" ) @@ -97,98 +91,6 @@ var ( } ) -func makeFakeInfo(t *testing.T) (*PaymentCreationInfo, *HTLCAttemptInfo) { - var preimg lntypes.Preimage - copy(preimg[:], rev[:]) - - hash := preimg.Hash() - - c := &PaymentCreationInfo{ - PaymentIdentifier: hash, - Value: 1000, - // Use single second precision to avoid false positive test - // failures due to the monotonic time component. - CreationTime: time.Unix(time.Now().Unix(), 0), - PaymentRequest: []byte("test"), - } - - a, err := NewHtlcAttempt( - 44, priv, testRoute, time.Unix(100, 0), &hash, - ) - require.NoError(t, err) - - return c, &a.HTLCAttemptInfo -} - -func TestSentPaymentSerialization(t *testing.T) { - t.Parallel() - - c, s := makeFakeInfo(t) - - var b bytes.Buffer - require.NoError(t, serializePaymentCreationInfo(&b, c), "serialize") - - // Assert the length of the serialized creation info is as expected, - // without any custom records. - baseLength := 32 + 8 + 8 + 4 + len(c.PaymentRequest) - require.Len(t, b.Bytes(), baseLength) - - newCreationInfo, err := deserializePaymentCreationInfo(&b) - require.NoError(t, err, "deserialize") - require.Equal(t, c, newCreationInfo) - - b.Reset() - - // Now we add some custom records to the creation info and serialize it - // again. - c.FirstHopCustomRecords = lnwire.CustomRecords{ - lnwire.MinCustomRecordsTlvType: []byte{1, 2, 3}, - } - require.NoError(t, serializePaymentCreationInfo(&b, c), "serialize") - - newCreationInfo, err = deserializePaymentCreationInfo(&b) - require.NoError(t, err, "deserialize") - require.Equal(t, c, newCreationInfo) - - b.Reset() - require.NoError(t, serializeHTLCAttemptInfo(&b, s), "serialize") - - newWireInfo, err := deserializeHTLCAttemptInfo(&b) - require.NoError(t, err, "deserialize") - - // First we verify all the records match up properly. - require.Equal(t, s.Route, newWireInfo.Route) - - // We now add the new fields and custom records to the route and - // serialize it again. - b.Reset() - s.Route.FirstHopAmount = tlv.NewRecordT[tlv.TlvType0]( - tlv.NewBigSizeT(lnwire.MilliSatoshi(1234)), - ) - s.Route.FirstHopWireCustomRecords = lnwire.CustomRecords{ - lnwire.MinCustomRecordsTlvType + 3: []byte{4, 5, 6}, - } - require.NoError(t, serializeHTLCAttemptInfo(&b, s), "serialize") - - newWireInfo, err = deserializeHTLCAttemptInfo(&b) - require.NoError(t, err, "deserialize") - require.Equal(t, s.Route, newWireInfo.Route) - - err = newWireInfo.attachOnionBlobAndCircuit() - require.NoError(t, err) - - // Clear routes to allow DeepEqual to compare the remaining fields. - newWireInfo.Route = route.Route{} - s.Route = route.Route{} - newWireInfo.AttemptID = s.AttemptID - - // Call session key method to set our cached session key so we can use - // DeepEqual, and assert that our key equals the original key. - require.Equal(t, s.cachedSessionKey, newWireInfo.SessionKey()) - - require.Equal(t, s, newWireInfo) -} - // assertRouteEquals compares to routes for equality and returns an error if // they are not equal. func assertRouteEqual(a, b *route.Route) error { @@ -200,53 +102,6 @@ func assertRouteEqual(a, b *route.Route) error { return nil } -// TestRouteSerialization tests serialization of a regular and blinded route. -func TestRouteSerialization(t *testing.T) { - t.Parallel() - - testSerializeRoute(t, testRoute) - testSerializeRoute(t, testBlindedRoute) -} - -func testSerializeRoute(t *testing.T, route route.Route) { - var b bytes.Buffer - err := SerializeRoute(&b, route) - require.NoError(t, err) - - r := bytes.NewReader(b.Bytes()) - route2, err := DeserializeRoute(r) - require.NoError(t, err) - - reflect.DeepEqual(route, route2) -} - -// deletePayment removes a payment with paymentHash from the payments database. -func deletePayment(t *testing.T, db *DB, paymentHash lntypes.Hash, seqNr uint64) { - t.Helper() - - err := kvdb.Update(db, func(tx kvdb.RwTx) error { - payments := tx.ReadWriteBucket(paymentsRootBucket) - - // Delete the payment bucket. - err := payments.DeleteNestedBucket(paymentHash[:]) - if err != nil { - return err - } - - key := make([]byte, 8) - byteOrder.PutUint64(key, seqNr) - - // Delete the index that references this payment. - indexes := tx.ReadWriteBucket(paymentsIndexBucket) - return indexes.Delete(key) - }, func() {}) - - if err != nil { - t.Fatalf("could not delete "+ - "payment: %v", err) - } -} - // TestQueryPayments tests retrieval of payments with forwards and reversed // queries. func TestQueryPayments(t *testing.T) { @@ -610,216 +465,3 @@ func TestQueryPayments(t *testing.T) { }) } } - -// TestFetchPaymentWithSequenceNumber tests lookup of payments with their -// sequence number. It sets up one payment with no duplicates, and another with -// two duplicates in its duplicates bucket then uses these payments to test the -// case where a specific duplicate is not found and the duplicates bucket is not -// present when we expect it to be. -func TestFetchPaymentWithSequenceNumber(t *testing.T) { - db, err := MakeTestDB(t) - require.NoError(t, err) - - paymentDB := NewKVPaymentsDB(db) - - // Generate a test payment which does not have duplicates. - noDuplicates, _, _, err := genInfo(t) - require.NoError(t, err) - - // Create a new payment entry in the database. - err = paymentDB.InitPayment( - noDuplicates.PaymentIdentifier, noDuplicates, - ) - require.NoError(t, err) - - // Fetch the payment so we can get its sequence nr. - noDuplicatesPayment, err := paymentDB.FetchPayment( - noDuplicates.PaymentIdentifier, - ) - require.NoError(t, err) - - // Generate a test payment which we will add duplicates to. - hasDuplicates, _, preimg, err := genInfo(t) - require.NoError(t, err) - - // Create a new payment entry in the database. - err = paymentDB.InitPayment( - hasDuplicates.PaymentIdentifier, hasDuplicates, - ) - require.NoError(t, err) - - // Fetch the payment so we can get its sequence nr. - hasDuplicatesPayment, err := paymentDB.FetchPayment( - hasDuplicates.PaymentIdentifier, - ) - require.NoError(t, err) - - // We declare the sequence numbers used here so that we can reference - // them in tests. - var ( - duplicateOneSeqNr = hasDuplicatesPayment.SequenceNum + 1 - duplicateTwoSeqNr = hasDuplicatesPayment.SequenceNum + 2 - ) - - // Add two duplicates to our second payment. - appendDuplicatePayment( - t, db, hasDuplicates.PaymentIdentifier, duplicateOneSeqNr, preimg, - ) - appendDuplicatePayment( - t, db, hasDuplicates.PaymentIdentifier, duplicateTwoSeqNr, preimg, - ) - - tests := []struct { - name string - paymentHash lntypes.Hash - sequenceNumber uint64 - expectedErr error - }{ - { - name: "lookup payment without duplicates", - paymentHash: noDuplicates.PaymentIdentifier, - sequenceNumber: noDuplicatesPayment.SequenceNum, - expectedErr: nil, - }, - { - name: "lookup payment with duplicates", - paymentHash: hasDuplicates.PaymentIdentifier, - sequenceNumber: hasDuplicatesPayment.SequenceNum, - expectedErr: nil, - }, - { - name: "lookup first duplicate", - paymentHash: hasDuplicates.PaymentIdentifier, - sequenceNumber: duplicateOneSeqNr, - expectedErr: nil, - }, - { - name: "lookup second duplicate", - paymentHash: hasDuplicates.PaymentIdentifier, - sequenceNumber: duplicateTwoSeqNr, - expectedErr: nil, - }, - { - name: "lookup non-existent duplicate", - paymentHash: hasDuplicates.PaymentIdentifier, - sequenceNumber: 999999, - expectedErr: ErrDuplicateNotFound, - }, - { - name: "lookup duplicate, no duplicates bucket", - paymentHash: noDuplicates.PaymentIdentifier, - sequenceNumber: duplicateTwoSeqNr, - expectedErr: ErrNoDuplicateBucket, - }, - } - - for _, test := range tests { - test := test - - t.Run(test.name, func(t *testing.T) { - err := kvdb.Update( - db, func(tx walletdb.ReadWriteTx) error { - var seqNrBytes [8]byte - byteOrder.PutUint64( - seqNrBytes[:], test.sequenceNumber, - ) - - _, err := fetchPaymentWithSequenceNumber( - tx, test.paymentHash, seqNrBytes[:], - ) - return err - }, func() {}, - ) - require.Equal(t, test.expectedErr, err) - }) - } -} - -// appendDuplicatePayment adds a duplicate payment to an existing payment. Note -// that this function requires a unique sequence number. -// -// This code is *only* intended to replicate legacy duplicate payments in lnd, -// our current schema does not allow duplicates. -func appendDuplicatePayment(t *testing.T, db *DB, paymentHash lntypes.Hash, - seqNr uint64, preImg lntypes.Preimage) { - - err := kvdb.Update(db, func(tx walletdb.ReadWriteTx) error { - bucket, err := fetchPaymentBucketUpdate( - tx, paymentHash, - ) - if err != nil { - return err - } - - // Create the duplicates bucket if it is not - // present. - dup, err := bucket.CreateBucketIfNotExists( - duplicatePaymentsBucket, - ) - if err != nil { - return err - } - - var sequenceKey [8]byte - byteOrder.PutUint64(sequenceKey[:], seqNr) - - // Create duplicate payments for the two dup - // sequence numbers we've setup. - putDuplicatePayment(t, dup, sequenceKey[:], paymentHash, preImg) - - // Finally, once we have created our entry we add an index for - // it. - err = createPaymentIndexEntry(tx, sequenceKey[:], paymentHash) - require.NoError(t, err) - - return nil - }, func() {}) - require.NoError(t, err, "could not create payment") -} - -// putDuplicatePayment creates a duplicate payment in the duplicates bucket -// provided with the minimal information required for successful reading. -func putDuplicatePayment(t *testing.T, duplicateBucket kvdb.RwBucket, - sequenceKey []byte, paymentHash lntypes.Hash, - preImg lntypes.Preimage) { - - paymentBucket, err := duplicateBucket.CreateBucketIfNotExists( - sequenceKey, - ) - require.NoError(t, err) - - err = paymentBucket.Put(duplicatePaymentSequenceKey, sequenceKey) - require.NoError(t, err) - - // Generate fake information for the duplicate payment. - info, _, _, err := genInfo(t) - require.NoError(t, err) - - // Write the payment info to disk under the creation info key. This code - // is copied rather than using serializePaymentCreationInfo to ensure - // we always write in the legacy format used by duplicate payments. - var b bytes.Buffer - var scratch [8]byte - _, err = b.Write(paymentHash[:]) - require.NoError(t, err) - - byteOrder.PutUint64(scratch[:], uint64(info.Value)) - _, err = b.Write(scratch[:]) - require.NoError(t, err) - - err = serializeTime(&b, info.CreationTime) - require.NoError(t, err) - - byteOrder.PutUint32(scratch[:4], 0) - _, err = b.Write(scratch[:4]) - require.NoError(t, err) - - // Get the PaymentCreationInfo. - err = paymentBucket.Put(duplicatePaymentCreationInfoKey, b.Bytes()) - require.NoError(t, err) - - // Duolicate payments are only stored for successes, so add the - // preimage. - err = paymentBucket.Put(duplicatePaymentSettleInfoKey, preImg[:]) - require.NoError(t, err) -}