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