Merge pull request #9826 from ziggie1984/refactor-payments-code-02
Refactor Payment PR 2
This commit is contained in:
10 files changed
+2824
-2724
No files matched your search
@@ -1,836 +0,0 @@
|
|||||||
package channeldb
|
|
||||||
|
|
||||||
import (
|
|
||||||
"bytes"
|
|
||||||
"encoding/binary"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"io"
|
|
||||||
"sync"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/lightningnetwork/lnd/kvdb"
|
|
||||||
"github.com/lightningnetwork/lnd/lntypes"
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
|
||||||
// paymentSeqBlockSize is the block size used when we batch allocate
|
|
||||||
// payment sequences for future payments.
|
|
||||||
paymentSeqBlockSize = 1000
|
|
||||||
|
|
||||||
// paymentProgressLogInterval is the interval we use limiting the
|
|
||||||
// logging output of payment processing.
|
|
||||||
paymentProgressLogInterval = 30 * time.Second
|
|
||||||
)
|
|
||||||
|
|
||||||
var (
|
|
||||||
// ErrAlreadyPaid signals we have already paid this payment hash.
|
|
||||||
ErrAlreadyPaid = errors.New("invoice is already paid")
|
|
||||||
|
|
||||||
// ErrPaymentInFlight signals that payment for this payment hash is
|
|
||||||
// already "in flight" on the network.
|
|
||||||
ErrPaymentInFlight = errors.New("payment is in transition")
|
|
||||||
|
|
||||||
// ErrPaymentExists is returned when we try to initialize an already
|
|
||||||
// existing payment that is not failed.
|
|
||||||
ErrPaymentExists = errors.New("payment already exists")
|
|
||||||
|
|
||||||
// ErrPaymentInternal is returned when performing the payment has a
|
|
||||||
// conflicting state, such as,
|
|
||||||
// - payment has StatusSucceeded but remaining amount is not zero.
|
|
||||||
// - payment has StatusInitiated but remaining amount is zero.
|
|
||||||
// - payment has StatusFailed but remaining amount is zero.
|
|
||||||
ErrPaymentInternal = errors.New("internal error")
|
|
||||||
|
|
||||||
// ErrPaymentNotInitiated is returned if the payment wasn't initiated.
|
|
||||||
ErrPaymentNotInitiated = errors.New("payment isn't initiated")
|
|
||||||
|
|
||||||
// ErrPaymentAlreadySucceeded is returned in the event we attempt to
|
|
||||||
// change the status of a payment already succeeded.
|
|
||||||
ErrPaymentAlreadySucceeded = errors.New("payment is already succeeded")
|
|
||||||
|
|
||||||
// ErrPaymentAlreadyFailed is returned in the event we attempt to alter
|
|
||||||
// a failed payment.
|
|
||||||
ErrPaymentAlreadyFailed = errors.New("payment has already failed")
|
|
||||||
|
|
||||||
// ErrUnknownPaymentStatus is returned when we do not recognize the
|
|
||||||
// existing state of a payment.
|
|
||||||
ErrUnknownPaymentStatus = errors.New("unknown payment status")
|
|
||||||
|
|
||||||
// ErrPaymentTerminal is returned if we attempt to alter a payment that
|
|
||||||
// already has reached a terminal condition.
|
|
||||||
ErrPaymentTerminal = errors.New("payment has reached terminal " +
|
|
||||||
"condition")
|
|
||||||
|
|
||||||
// ErrAttemptAlreadySettled is returned if we try to alter an already
|
|
||||||
// settled HTLC attempt.
|
|
||||||
ErrAttemptAlreadySettled = errors.New("attempt already settled")
|
|
||||||
|
|
||||||
// ErrAttemptAlreadyFailed is returned if we try to alter an already
|
|
||||||
// failed HTLC attempt.
|
|
||||||
ErrAttemptAlreadyFailed = errors.New("attempt already failed")
|
|
||||||
|
|
||||||
// ErrValueMismatch is returned if we try to register a non-MPP attempt
|
|
||||||
// with an amount that doesn't match the payment amount.
|
|
||||||
ErrValueMismatch = errors.New("attempted value doesn't match payment " +
|
|
||||||
"amount")
|
|
||||||
|
|
||||||
// ErrValueExceedsAmt is returned if we try to register an attempt that
|
|
||||||
// would take the total sent amount above the payment amount.
|
|
||||||
ErrValueExceedsAmt = errors.New("attempted value exceeds payment " +
|
|
||||||
"amount")
|
|
||||||
|
|
||||||
// ErrNonMPPayment is returned if we try to register an MPP attempt for
|
|
||||||
// a payment that already has a non-MPP attempt registered.
|
|
||||||
ErrNonMPPayment = errors.New("payment has non-MPP attempts")
|
|
||||||
|
|
||||||
// ErrMPPayment is returned if we try to register a non-MPP attempt for
|
|
||||||
// a payment that already has an MPP attempt registered.
|
|
||||||
ErrMPPayment = errors.New("payment has MPP attempts")
|
|
||||||
|
|
||||||
// ErrMPPRecordInBlindedPayment is returned if we try to register an
|
|
||||||
// attempt with an MPP record for a payment to a blinded path.
|
|
||||||
ErrMPPRecordInBlindedPayment = errors.New("blinded payment cannot " +
|
|
||||||
"contain MPP records")
|
|
||||||
|
|
||||||
// ErrBlindedPaymentTotalAmountMismatch is returned if we try to
|
|
||||||
// register an HTLC shard to a blinded route where the total amount
|
|
||||||
// doesn't match existing shards.
|
|
||||||
ErrBlindedPaymentTotalAmountMismatch = errors.New("blinded path " +
|
|
||||||
"total amount mismatch")
|
|
||||||
|
|
||||||
// ErrMPPPaymentAddrMismatch is returned if we try to register an MPP
|
|
||||||
// shard where the payment address doesn't match existing shards.
|
|
||||||
ErrMPPPaymentAddrMismatch = errors.New("payment address mismatch")
|
|
||||||
|
|
||||||
// ErrMPPTotalAmountMismatch is returned if we try to register an MPP
|
|
||||||
// shard where the total amount doesn't match existing shards.
|
|
||||||
ErrMPPTotalAmountMismatch = errors.New("mp payment total amount " +
|
|
||||||
"mismatch")
|
|
||||||
|
|
||||||
// ErrPaymentPendingSettled is returned when we try to add a new
|
|
||||||
// attempt to a payment that has at least one of its HTLCs settled.
|
|
||||||
ErrPaymentPendingSettled = errors.New("payment has settled htlcs")
|
|
||||||
|
|
||||||
// ErrPaymentPendingFailed is returned when we try to add a new attempt
|
|
||||||
// to a payment that already has a failure reason.
|
|
||||||
ErrPaymentPendingFailed = errors.New("payment has failure reason")
|
|
||||||
|
|
||||||
// ErrSentExceedsTotal is returned if the payment's current total sent
|
|
||||||
// amount exceed the total amount.
|
|
||||||
ErrSentExceedsTotal = errors.New("total sent exceeds total amount")
|
|
||||||
|
|
||||||
// errNoAttemptInfo is returned when no attempt info is stored yet.
|
|
||||||
errNoAttemptInfo = errors.New("unable to find attempt info for " +
|
|
||||||
"inflight payment")
|
|
||||||
|
|
||||||
// errNoSequenceNrIndex is returned when an attempt to lookup a payment
|
|
||||||
// index is made for a sequence number that is not indexed.
|
|
||||||
errNoSequenceNrIndex = errors.New("payment sequence number index " +
|
|
||||||
"does not exist")
|
|
||||||
)
|
|
||||||
|
|
||||||
// PaymentControl implements persistence for payments and payment attempts.
|
|
||||||
type PaymentControl struct {
|
|
||||||
paymentSeqMx sync.Mutex
|
|
||||||
currPaymentSeq uint64
|
|
||||||
storedPaymentSeq uint64
|
|
||||||
db *DB
|
|
||||||
}
|
|
||||||
|
|
||||||
// NewPaymentControl creates a new instance of the PaymentControl.
|
|
||||||
func NewPaymentControl(db *DB) *PaymentControl {
|
|
||||||
return &PaymentControl{
|
|
||||||
db: db,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// InitPayment checks or records the given PaymentCreationInfo with the DB,
|
|
||||||
// making sure it does not already exist as an in-flight payment. When this
|
|
||||||
// method returns successfully, the payment is guaranteed to be in the InFlight
|
|
||||||
// state.
|
|
||||||
func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash,
|
|
||||||
info *PaymentCreationInfo) error {
|
|
||||||
|
|
||||||
// Obtain a new sequence number for this payment. This is used
|
|
||||||
// to sort the payments in order of creation, and also acts as
|
|
||||||
// a unique identifier for each payment.
|
|
||||||
sequenceNum, err := p.nextPaymentSequence()
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
var b bytes.Buffer
|
|
||||||
if err := serializePaymentCreationInfo(&b, info); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
infoBytes := b.Bytes()
|
|
||||||
|
|
||||||
var updateErr error
|
|
||||||
err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
|
|
||||||
// Reset the update error, to avoid carrying over an error
|
|
||||||
// from a previous execution of the batched db transaction.
|
|
||||||
updateErr = nil
|
|
||||||
|
|
||||||
prefetchPayment(tx, paymentHash)
|
|
||||||
bucket, err := createPaymentBucket(tx, paymentHash)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get the existing status of this payment, if any.
|
|
||||||
paymentStatus, err := fetchPaymentStatus(bucket)
|
|
||||||
|
|
||||||
switch {
|
|
||||||
// If no error is returned, it means we already have this
|
|
||||||
// payment. We'll check the status to decide whether we allow
|
|
||||||
// retrying the payment or return a specific error.
|
|
||||||
case err == nil:
|
|
||||||
if err := paymentStatus.initializable(); err != nil {
|
|
||||||
updateErr = err
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Otherwise, if the error is not `ErrPaymentNotInitiated`,
|
|
||||||
// we'll return the error.
|
|
||||||
case !errors.Is(err, ErrPaymentNotInitiated):
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Before we set our new sequence number, we check whether this
|
|
||||||
// payment has a previously set sequence number and remove its
|
|
||||||
// index entry if it exists. This happens in the case where we
|
|
||||||
// have a previously attempted payment which was left in a state
|
|
||||||
// where we can retry.
|
|
||||||
seqBytes := bucket.Get(paymentSequenceKey)
|
|
||||||
if seqBytes != nil {
|
|
||||||
indexBucket := tx.ReadWriteBucket(paymentsIndexBucket)
|
|
||||||
if err := indexBucket.Delete(seqBytes); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Once we have obtained a sequence number, we add an entry
|
|
||||||
// to our index bucket which will map the sequence number to
|
|
||||||
// our payment identifier.
|
|
||||||
err = createPaymentIndexEntry(
|
|
||||||
tx, sequenceNum, info.PaymentIdentifier,
|
|
||||||
)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
err = bucket.Put(paymentSequenceKey, sequenceNum)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Add the payment info to the bucket, which contains the
|
|
||||||
// static information for this payment
|
|
||||||
err = bucket.Put(paymentCreationInfoKey, infoBytes)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// We'll delete any lingering HTLCs to start with, in case we
|
|
||||||
// are initializing a payment that was attempted earlier, but
|
|
||||||
// left in a state where we could retry.
|
|
||||||
err = bucket.DeleteNestedBucket(paymentHtlcsBucket)
|
|
||||||
if err != nil && err != kvdb.ErrBucketNotFound {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Also delete any lingering failure info now that we are
|
|
||||||
// re-attempting.
|
|
||||||
return bucket.Delete(paymentFailInfoKey)
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf("unable to init payment: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
return updateErr
|
|
||||||
}
|
|
||||||
|
|
||||||
// DeleteFailedAttempts deletes all failed htlcs for a payment if configured
|
|
||||||
// by the PaymentControl db.
|
|
||||||
func (p *PaymentControl) DeleteFailedAttempts(hash lntypes.Hash) error {
|
|
||||||
if !p.db.keepFailedPaymentAttempts {
|
|
||||||
const failedHtlcsOnly = true
|
|
||||||
err := p.db.DeletePayment(hash, failedHtlcsOnly)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// paymentIndexTypeHash is a payment index type which indicates that we have
|
|
||||||
// created an index of payment sequence number to payment hash.
|
|
||||||
type paymentIndexType uint8
|
|
||||||
|
|
||||||
// paymentIndexTypeHash is a payment index type which indicates that we have
|
|
||||||
// created an index of payment sequence number to payment hash.
|
|
||||||
const paymentIndexTypeHash paymentIndexType = 0
|
|
||||||
|
|
||||||
// createPaymentIndexEntry creates a payment hash typed index for a payment. The
|
|
||||||
// index produced contains a payment index type (which can be used in future to
|
|
||||||
// signal different payment index types) and the payment identifier.
|
|
||||||
func createPaymentIndexEntry(tx kvdb.RwTx, sequenceNumber []byte,
|
|
||||||
id lntypes.Hash) error {
|
|
||||||
|
|
||||||
var b bytes.Buffer
|
|
||||||
if err := WriteElements(&b, paymentIndexTypeHash, id[:]); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
indexes := tx.ReadWriteBucket(paymentsIndexBucket)
|
|
||||||
return indexes.Put(sequenceNumber, b.Bytes())
|
|
||||||
}
|
|
||||||
|
|
||||||
// deserializePaymentIndex deserializes a payment index entry. This function
|
|
||||||
// currently only supports deserialization of payment hash indexes, and will
|
|
||||||
// fail for other types.
|
|
||||||
func deserializePaymentIndex(r io.Reader) (lntypes.Hash, error) {
|
|
||||||
var (
|
|
||||||
indexType paymentIndexType
|
|
||||||
paymentHash []byte
|
|
||||||
)
|
|
||||||
|
|
||||||
if err := ReadElements(r, &indexType, &paymentHash); err != nil {
|
|
||||||
return lntypes.Hash{}, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// While we only have on payment index type, we do not need to use our
|
|
||||||
// index type to deserialize the index. However, we sanity check that
|
|
||||||
// this type is as expected, since we had to read it out anyway.
|
|
||||||
if indexType != paymentIndexTypeHash {
|
|
||||||
return lntypes.Hash{}, fmt.Errorf("unknown payment index "+
|
|
||||||
"type: %v", indexType)
|
|
||||||
}
|
|
||||||
|
|
||||||
hash, err := lntypes.MakeHash(paymentHash)
|
|
||||||
if err != nil {
|
|
||||||
return lntypes.Hash{}, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return hash, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// RegisterAttempt atomically records the provided HTLCAttemptInfo to the
|
|
||||||
// DB.
|
|
||||||
func (p *PaymentControl) RegisterAttempt(paymentHash lntypes.Hash,
|
|
||||||
attempt *HTLCAttemptInfo) (*MPPayment, error) {
|
|
||||||
|
|
||||||
// Serialize the information before opening the db transaction.
|
|
||||||
var a bytes.Buffer
|
|
||||||
err := serializeHTLCAttemptInfo(&a, attempt)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
htlcInfoBytes := a.Bytes()
|
|
||||||
|
|
||||||
htlcIDBytes := make([]byte, 8)
|
|
||||||
binary.BigEndian.PutUint64(htlcIDBytes, attempt.AttemptID)
|
|
||||||
|
|
||||||
var payment *MPPayment
|
|
||||||
err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
|
|
||||||
prefetchPayment(tx, paymentHash)
|
|
||||||
bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
payment, err = fetchPayment(bucket)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if registering a new attempt is allowed.
|
|
||||||
if err := payment.Registrable(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// If the final hop has encrypted data, then we know this is a
|
|
||||||
// blinded payment. In blinded payments, MPP records are not set
|
|
||||||
// for split payments and the recipient is responsible for using
|
|
||||||
// a consistent PathID across the various encrypted data
|
|
||||||
// payloads that we received from them for this payment. All we
|
|
||||||
// need to check is that the total amount field for each HTLC
|
|
||||||
// in the split payment is correct.
|
|
||||||
isBlinded := len(attempt.Route.FinalHop().EncryptedData) != 0
|
|
||||||
|
|
||||||
// Make sure any existing shards match the new one with regards
|
|
||||||
// to MPP options.
|
|
||||||
mpp := attempt.Route.FinalHop().MPP
|
|
||||||
|
|
||||||
// MPP records should not be set for attempts to blinded paths.
|
|
||||||
if isBlinded && mpp != nil {
|
|
||||||
return ErrMPPRecordInBlindedPayment
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, h := range payment.InFlightHTLCs() {
|
|
||||||
hMpp := h.Route.FinalHop().MPP
|
|
||||||
|
|
||||||
// If this is a blinded payment, then no existing HTLCs
|
|
||||||
// should have MPP records.
|
|
||||||
if isBlinded && hMpp != nil {
|
|
||||||
return ErrMPPRecordInBlindedPayment
|
|
||||||
}
|
|
||||||
|
|
||||||
// If this is a blinded payment, then we just need to
|
|
||||||
// check that the TotalAmtMsat field for this shard
|
|
||||||
// is equal to that of any other shard in the same
|
|
||||||
// payment.
|
|
||||||
if isBlinded {
|
|
||||||
if attempt.Route.FinalHop().TotalAmtMsat !=
|
|
||||||
h.Route.FinalHop().TotalAmtMsat {
|
|
||||||
|
|
||||||
//nolint:ll
|
|
||||||
return ErrBlindedPaymentTotalAmountMismatch
|
|
||||||
}
|
|
||||||
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
switch {
|
|
||||||
// We tried to register a non-MPP attempt for a MPP
|
|
||||||
// payment.
|
|
||||||
case mpp == nil && hMpp != nil:
|
|
||||||
return ErrMPPayment
|
|
||||||
|
|
||||||
// We tried to register a MPP shard for a non-MPP
|
|
||||||
// payment.
|
|
||||||
case mpp != nil && hMpp == nil:
|
|
||||||
return ErrNonMPPayment
|
|
||||||
|
|
||||||
// Non-MPP payment, nothing more to validate.
|
|
||||||
case mpp == nil:
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check that MPP options match.
|
|
||||||
if mpp.PaymentAddr() != hMpp.PaymentAddr() {
|
|
||||||
return ErrMPPPaymentAddrMismatch
|
|
||||||
}
|
|
||||||
|
|
||||||
if mpp.TotalMsat() != hMpp.TotalMsat() {
|
|
||||||
return ErrMPPTotalAmountMismatch
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// If this is a non-MPP attempt, it must match the total amount
|
|
||||||
// exactly. Note that a blinded payment is considered an MPP
|
|
||||||
// attempt.
|
|
||||||
amt := attempt.Route.ReceiverAmt()
|
|
||||||
if !isBlinded && mpp == nil && amt != payment.Info.Value {
|
|
||||||
return ErrValueMismatch
|
|
||||||
}
|
|
||||||
|
|
||||||
// Ensure we aren't sending more than the total payment amount.
|
|
||||||
sentAmt, _ := payment.SentAmt()
|
|
||||||
if sentAmt+amt > payment.Info.Value {
|
|
||||||
return fmt.Errorf("%w: attempted=%v, payment amount="+
|
|
||||||
"%v", ErrValueExceedsAmt, sentAmt+amt,
|
|
||||||
payment.Info.Value)
|
|
||||||
}
|
|
||||||
|
|
||||||
htlcsBucket, err := bucket.CreateBucketIfNotExists(
|
|
||||||
paymentHtlcsBucket,
|
|
||||||
)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
err = htlcsBucket.Put(
|
|
||||||
htlcBucketKey(htlcAttemptInfoKey, htlcIDBytes),
|
|
||||||
htlcInfoBytes,
|
|
||||||
)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Retrieve attempt info for the notification.
|
|
||||||
payment, err = fetchPayment(bucket)
|
|
||||||
return err
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return payment, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// SettleAttempt marks the given attempt settled with the preimage. If this is
|
|
||||||
// a multi shard payment, this might implicitly mean that the full payment
|
|
||||||
// succeeded.
|
|
||||||
//
|
|
||||||
// After invoking this method, InitPayment should always return an error to
|
|
||||||
// prevent us from making duplicate payments to the same payment hash. The
|
|
||||||
// provided preimage is atomically saved to the DB for record keeping.
|
|
||||||
func (p *PaymentControl) SettleAttempt(hash lntypes.Hash,
|
|
||||||
attemptID uint64, settleInfo *HTLCSettleInfo) (*MPPayment, error) {
|
|
||||||
|
|
||||||
var b bytes.Buffer
|
|
||||||
if err := serializeHTLCSettleInfo(&b, settleInfo); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
settleBytes := b.Bytes()
|
|
||||||
|
|
||||||
return p.updateHtlcKey(hash, attemptID, htlcSettleInfoKey, settleBytes)
|
|
||||||
}
|
|
||||||
|
|
||||||
// FailAttempt marks the given payment attempt failed.
|
|
||||||
func (p *PaymentControl) FailAttempt(hash lntypes.Hash,
|
|
||||||
attemptID uint64, failInfo *HTLCFailInfo) (*MPPayment, error) {
|
|
||||||
|
|
||||||
var b bytes.Buffer
|
|
||||||
if err := serializeHTLCFailInfo(&b, failInfo); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
failBytes := b.Bytes()
|
|
||||||
|
|
||||||
return p.updateHtlcKey(hash, attemptID, htlcFailInfoKey, failBytes)
|
|
||||||
}
|
|
||||||
|
|
||||||
// updateHtlcKey updates a database key for the specified htlc.
|
|
||||||
func (p *PaymentControl) updateHtlcKey(paymentHash lntypes.Hash,
|
|
||||||
attemptID uint64, key, value []byte) (*MPPayment, error) {
|
|
||||||
|
|
||||||
aid := make([]byte, 8)
|
|
||||||
binary.BigEndian.PutUint64(aid, attemptID)
|
|
||||||
|
|
||||||
var payment *MPPayment
|
|
||||||
err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
|
|
||||||
payment = nil
|
|
||||||
|
|
||||||
prefetchPayment(tx, paymentHash)
|
|
||||||
bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
p, err := fetchPayment(bucket)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// We can only update keys of in-flight payments. We allow
|
|
||||||
// updating keys even if the payment has reached a terminal
|
|
||||||
// condition, since the HTLC outcomes must still be updated.
|
|
||||||
if err := p.Status.updatable(); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
htlcsBucket := bucket.NestedReadWriteBucket(paymentHtlcsBucket)
|
|
||||||
if htlcsBucket == nil {
|
|
||||||
return fmt.Errorf("htlcs bucket not found")
|
|
||||||
}
|
|
||||||
|
|
||||||
if htlcsBucket.Get(htlcBucketKey(htlcAttemptInfoKey, aid)) == nil {
|
|
||||||
return fmt.Errorf("HTLC with ID %v not registered",
|
|
||||||
attemptID)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Make sure the shard is not already failed or settled.
|
|
||||||
if htlcsBucket.Get(htlcBucketKey(htlcFailInfoKey, aid)) != nil {
|
|
||||||
return ErrAttemptAlreadyFailed
|
|
||||||
}
|
|
||||||
|
|
||||||
if htlcsBucket.Get(htlcBucketKey(htlcSettleInfoKey, aid)) != nil {
|
|
||||||
return ErrAttemptAlreadySettled
|
|
||||||
}
|
|
||||||
|
|
||||||
// Add or update the key for this htlc.
|
|
||||||
err = htlcsBucket.Put(htlcBucketKey(key, aid), value)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Retrieve attempt info for the notification.
|
|
||||||
payment, err = fetchPayment(bucket)
|
|
||||||
return err
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return payment, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fail transitions a payment into the Failed state, and records the reason the
|
|
||||||
// payment failed. After invoking this method, InitPayment should return nil on
|
|
||||||
// its next call for this payment hash, allowing the switch to make a
|
|
||||||
// subsequent payment.
|
|
||||||
func (p *PaymentControl) Fail(paymentHash lntypes.Hash,
|
|
||||||
reason FailureReason) (*MPPayment, error) {
|
|
||||||
|
|
||||||
var (
|
|
||||||
updateErr error
|
|
||||||
payment *MPPayment
|
|
||||||
)
|
|
||||||
err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
|
|
||||||
// Reset the update error, to avoid carrying over an error
|
|
||||||
// from a previous execution of the batched db transaction.
|
|
||||||
updateErr = nil
|
|
||||||
payment = nil
|
|
||||||
|
|
||||||
prefetchPayment(tx, paymentHash)
|
|
||||||
bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
|
|
||||||
if err == ErrPaymentNotInitiated {
|
|
||||||
updateErr = ErrPaymentNotInitiated
|
|
||||||
return nil
|
|
||||||
} else if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// We mark the payment as failed as long as it is known. This
|
|
||||||
// lets the last attempt to fail with a terminal write its
|
|
||||||
// failure to the PaymentControl without synchronizing with
|
|
||||||
// other attempts.
|
|
||||||
_, err = fetchPaymentStatus(bucket)
|
|
||||||
if errors.Is(err, ErrPaymentNotInitiated) {
|
|
||||||
updateErr = ErrPaymentNotInitiated
|
|
||||||
return nil
|
|
||||||
} else if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Put the failure reason in the bucket for record keeping.
|
|
||||||
v := []byte{byte(reason)}
|
|
||||||
err = bucket.Put(paymentFailInfoKey, v)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Retrieve attempt info for the notification, if available.
|
|
||||||
payment, err = fetchPayment(bucket)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return payment, updateErr
|
|
||||||
}
|
|
||||||
|
|
||||||
// FetchPayment returns information about a payment from the database.
|
|
||||||
func (p *PaymentControl) FetchPayment(paymentHash lntypes.Hash) (
|
|
||||||
*MPPayment, error) {
|
|
||||||
|
|
||||||
var payment *MPPayment
|
|
||||||
err := kvdb.View(p.db, func(tx kvdb.RTx) error {
|
|
||||||
prefetchPayment(tx, paymentHash)
|
|
||||||
bucket, err := fetchPaymentBucket(tx, paymentHash)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
payment, err = fetchPayment(bucket)
|
|
||||||
|
|
||||||
return err
|
|
||||||
}, func() {
|
|
||||||
payment = nil
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return payment, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// prefetchPayment attempts to prefetch as much of the payment as possible to
|
|
||||||
// reduce DB roundtrips.
|
|
||||||
func prefetchPayment(tx kvdb.RTx, paymentHash lntypes.Hash) {
|
|
||||||
rb := kvdb.RootBucket(tx)
|
|
||||||
kvdb.Prefetch(
|
|
||||||
rb,
|
|
||||||
[]string{
|
|
||||||
// Prefetch all keys in the payment's bucket.
|
|
||||||
string(paymentsRootBucket),
|
|
||||||
string(paymentHash[:]),
|
|
||||||
},
|
|
||||||
[]string{
|
|
||||||
// Prefetch all keys in the payment's htlc bucket.
|
|
||||||
string(paymentsRootBucket),
|
|
||||||
string(paymentHash[:]),
|
|
||||||
string(paymentHtlcsBucket),
|
|
||||||
},
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
// createPaymentBucket creates or fetches the sub-bucket assigned to this
|
|
||||||
// payment hash.
|
|
||||||
func createPaymentBucket(tx kvdb.RwTx, paymentHash lntypes.Hash) (
|
|
||||||
kvdb.RwBucket, error) {
|
|
||||||
|
|
||||||
payments, err := tx.CreateTopLevelBucket(paymentsRootBucket)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return payments.CreateBucketIfNotExists(paymentHash[:])
|
|
||||||
}
|
|
||||||
|
|
||||||
// fetchPaymentBucket fetches the sub-bucket assigned to this payment hash. If
|
|
||||||
// the bucket does not exist, it returns ErrPaymentNotInitiated.
|
|
||||||
func fetchPaymentBucket(tx kvdb.RTx, paymentHash lntypes.Hash) (
|
|
||||||
kvdb.RBucket, error) {
|
|
||||||
|
|
||||||
payments := tx.ReadBucket(paymentsRootBucket)
|
|
||||||
if payments == nil {
|
|
||||||
return nil, ErrPaymentNotInitiated
|
|
||||||
}
|
|
||||||
|
|
||||||
bucket := payments.NestedReadBucket(paymentHash[:])
|
|
||||||
if bucket == nil {
|
|
||||||
return nil, ErrPaymentNotInitiated
|
|
||||||
}
|
|
||||||
|
|
||||||
return bucket, nil
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
// fetchPaymentBucketUpdate is identical to fetchPaymentBucket, but it returns a
|
|
||||||
// bucket that can be written to.
|
|
||||||
func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
|
|
||||||
kvdb.RwBucket, error) {
|
|
||||||
|
|
||||||
payments := tx.ReadWriteBucket(paymentsRootBucket)
|
|
||||||
if payments == nil {
|
|
||||||
return nil, ErrPaymentNotInitiated
|
|
||||||
}
|
|
||||||
|
|
||||||
bucket := payments.NestedReadWriteBucket(paymentHash[:])
|
|
||||||
if bucket == nil {
|
|
||||||
return nil, ErrPaymentNotInitiated
|
|
||||||
}
|
|
||||||
|
|
||||||
return bucket, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// nextPaymentSequence returns the next sequence number to store for a new
|
|
||||||
// payment.
|
|
||||||
func (p *PaymentControl) nextPaymentSequence() ([]byte, error) {
|
|
||||||
p.paymentSeqMx.Lock()
|
|
||||||
defer p.paymentSeqMx.Unlock()
|
|
||||||
|
|
||||||
// Set a new upper bound in the DB every 1000 payments to avoid
|
|
||||||
// conflicts on the sequence when using etcd.
|
|
||||||
if p.currPaymentSeq == p.storedPaymentSeq {
|
|
||||||
var currPaymentSeq, newUpperBound uint64
|
|
||||||
if err := kvdb.Update(p.db.Backend, func(tx kvdb.RwTx) error {
|
|
||||||
paymentsBucket, err := tx.CreateTopLevelBucket(
|
|
||||||
paymentsRootBucket,
|
|
||||||
)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
currPaymentSeq = paymentsBucket.Sequence()
|
|
||||||
newUpperBound = currPaymentSeq + paymentSeqBlockSize
|
|
||||||
return paymentsBucket.SetSequence(newUpperBound)
|
|
||||||
}, func() {}); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// We lazy initialize the cached currPaymentSeq here using the
|
|
||||||
// first nextPaymentSequence() call. This if statement will auto
|
|
||||||
// initialize our stored currPaymentSeq, since by default both
|
|
||||||
// this variable and storedPaymentSeq are zero which in turn
|
|
||||||
// will have us fetch the current values from the DB.
|
|
||||||
if p.currPaymentSeq == 0 {
|
|
||||||
p.currPaymentSeq = currPaymentSeq
|
|
||||||
}
|
|
||||||
|
|
||||||
p.storedPaymentSeq = newUpperBound
|
|
||||||
}
|
|
||||||
|
|
||||||
p.currPaymentSeq++
|
|
||||||
b := make([]byte, 8)
|
|
||||||
binary.BigEndian.PutUint64(b, p.currPaymentSeq)
|
|
||||||
|
|
||||||
return b, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// fetchPaymentStatus fetches the payment status of the payment. If the payment
|
|
||||||
// isn't found, it will return error `ErrPaymentNotInitiated`.
|
|
||||||
func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
|
|
||||||
// Creation info should be set for all payments, regardless of state.
|
|
||||||
// If not, it is unknown.
|
|
||||||
if bucket.Get(paymentCreationInfoKey) == nil {
|
|
||||||
return 0, ErrPaymentNotInitiated
|
|
||||||
}
|
|
||||||
|
|
||||||
payment, err := fetchPayment(bucket)
|
|
||||||
if err != nil {
|
|
||||||
return 0, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return payment.Status, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// FetchInFlightPayments returns all payments with status InFlight.
|
|
||||||
func (p *PaymentControl) FetchInFlightPayments() ([]*MPPayment, error) {
|
|
||||||
var (
|
|
||||||
inFlights []*MPPayment
|
|
||||||
start = time.Now()
|
|
||||||
lastLogTime = time.Now()
|
|
||||||
processedCount int
|
|
||||||
)
|
|
||||||
|
|
||||||
err := kvdb.View(p.db, func(tx kvdb.RTx) error {
|
|
||||||
payments := tx.ReadBucket(paymentsRootBucket)
|
|
||||||
if payments == nil {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
return payments.ForEach(func(k, _ []byte) error {
|
|
||||||
bucket := payments.NestedReadBucket(k)
|
|
||||||
if bucket == nil {
|
|
||||||
return fmt.Errorf("non bucket element")
|
|
||||||
}
|
|
||||||
|
|
||||||
p, err := fetchPayment(bucket)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
processedCount++
|
|
||||||
if time.Since(lastLogTime) >=
|
|
||||||
paymentProgressLogInterval {
|
|
||||||
|
|
||||||
log.Debugf("Scanning inflight payments "+
|
|
||||||
"(in progress), processed %d, last "+
|
|
||||||
"processed payment: %v", processedCount,
|
|
||||||
p.Info)
|
|
||||||
|
|
||||||
lastLogTime = time.Now()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Skip the payment if it's terminated.
|
|
||||||
if p.Terminated() {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
inFlights = append(inFlights, p)
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
}, func() {
|
|
||||||
inFlights = nil
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
elapsed := time.Since(start)
|
|
||||||
log.Debugf("Completed scanning for inflight payments: "+
|
|
||||||
"total_processed=%d, found_inflight=%d, elapsed=%v",
|
|
||||||
processedCount, len(inFlights),
|
|
||||||
elapsed.Round(time.Millisecond))
|
|
||||||
|
|
||||||
return inFlights, nil
|
|
||||||
}
|
|
||||||
File diff suppressed because it is too large.
Load diff
File diff suppressed because it is too large.
Load diff
File diff suppressed because it is too large.
Load diff
+21
-369
@@ -1,7 +1,6 @@
|
|||||||
package channeldb
|
package channeldb
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"math"
|
"math"
|
||||||
"reflect"
|
"reflect"
|
||||||
@@ -9,14 +8,9 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/btcsuite/btcd/btcec/v2"
|
"github.com/btcsuite/btcd/btcec/v2"
|
||||||
"github.com/btcsuite/btcwallet/walletdb"
|
|
||||||
"github.com/davecgh/go-spew/spew"
|
"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/record"
|
||||||
"github.com/lightningnetwork/lnd/routing/route"
|
"github.com/lightningnetwork/lnd/routing/route"
|
||||||
"github.com/lightningnetwork/lnd/tlv"
|
|
||||||
"github.com/stretchr/testify/require"
|
"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
|
// assertRouteEquals compares to routes for equality and returns an error if
|
||||||
// they are not equal.
|
// they are not equal.
|
||||||
func assertRouteEqual(a, b *route.Route) error {
|
func assertRouteEqual(a, b *route.Route) error {
|
||||||
@@ -200,53 +102,6 @@ func assertRouteEqual(a, b *route.Route) error {
|
|||||||
return nil
|
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
|
// TestQueryPayments tests retrieval of payments with forwards and reversed
|
||||||
// queries.
|
// queries.
|
||||||
func TestQueryPayments(t *testing.T) {
|
func TestQueryPayments(t *testing.T) {
|
||||||
@@ -502,9 +357,11 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
t.Fatalf("unable to init db: %v", err)
|
t.Fatalf("unable to init db: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
paymentDB := NewKVPaymentsDB(db)
|
||||||
|
|
||||||
// Make a preliminary query to make sure it's ok to
|
// Make a preliminary query to make sure it's ok to
|
||||||
// query when we have no payments.
|
// query when we have no payments.
|
||||||
resp, err := db.QueryPayments(tt.query)
|
resp, err := paymentDB.QueryPayments(tt.query)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
require.Len(t, resp.Payments, 0)
|
require.Len(t, resp.Payments, 0)
|
||||||
|
|
||||||
@@ -516,7 +373,6 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
// where we have duplicates in the nested duplicates
|
// where we have duplicates in the nested duplicates
|
||||||
// bucket.
|
// bucket.
|
||||||
nonDuplicatePayments := 6
|
nonDuplicatePayments := 6
|
||||||
pControl := NewPaymentControl(db)
|
|
||||||
|
|
||||||
for i := 0; i < nonDuplicatePayments; i++ {
|
for i := 0; i < nonDuplicatePayments; i++ {
|
||||||
// Generate a test payment.
|
// Generate a test payment.
|
||||||
@@ -530,7 +386,9 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
info.CreationTime = time.Unix(int64(i+1), 0)
|
info.CreationTime = time.Unix(int64(i+1), 0)
|
||||||
|
|
||||||
// Create a new payment entry in the database.
|
// Create a new payment entry in the database.
|
||||||
err = pControl.InitPayment(info.PaymentIdentifier, info)
|
err = paymentDB.InitPayment(
|
||||||
|
info.PaymentIdentifier, info,
|
||||||
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("unable to initialize "+
|
t.Fatalf("unable to initialize "+
|
||||||
"payment in database: %v", err)
|
"payment in database: %v", err)
|
||||||
@@ -538,7 +396,7 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
|
|
||||||
// Immediately delete the payment with index 2.
|
// Immediately delete the payment with index 2.
|
||||||
if i == 1 {
|
if i == 1 {
|
||||||
pmt, err := pControl.FetchPayment(
|
pmt, err := paymentDB.FetchPayment(
|
||||||
info.PaymentIdentifier,
|
info.PaymentIdentifier,
|
||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -552,13 +410,13 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
// to the parent payment + 1. Note that
|
// to the parent payment + 1. Note that
|
||||||
// duplicate payments will always be succeeded.
|
// duplicate payments will always be succeeded.
|
||||||
if i == (nonDuplicatePayments - 1) {
|
if i == (nonDuplicatePayments - 1) {
|
||||||
pmt, err := pControl.FetchPayment(
|
pmt, err := paymentDB.FetchPayment(
|
||||||
info.PaymentIdentifier,
|
info.PaymentIdentifier,
|
||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
appendDuplicatePayment(
|
appendDuplicatePayment(
|
||||||
t, pControl.db,
|
t, paymentDB.db,
|
||||||
info.PaymentIdentifier,
|
info.PaymentIdentifier,
|
||||||
pmt.SequenceNum+1,
|
pmt.SequenceNum+1,
|
||||||
preimg,
|
preimg,
|
||||||
@@ -567,19 +425,19 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Fetch all payments in the database.
|
// Fetch all payments in the database.
|
||||||
allPayments, err := db.FetchPayments()
|
allPayments, err := paymentDB.FetchPayments()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("payments could not be fetched from "+
|
t.Fatalf("payments could not be fetched from "+
|
||||||
"database: %v", err)
|
"database: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
if len(allPayments) != 6 {
|
if len(allPayments) != 6 {
|
||||||
t.Fatalf("Number of payments received does not "+
|
t.Fatalf("Number of payments received does "+
|
||||||
"match expected one. Got %v, want %v.",
|
"not match expected one. Got %v, "+
|
||||||
len(allPayments), 6)
|
"want %v.", len(allPayments), 6)
|
||||||
}
|
}
|
||||||
|
|
||||||
querySlice, err := db.QueryPayments(tt.query)
|
querySlice, err := paymentDB.QueryPayments(tt.query)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("unexpected error: %v", err)
|
t.Fatalf("unexpected error: %v", err)
|
||||||
}
|
}
|
||||||
@@ -587,7 +445,8 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
tt.lastIndex != querySlice.LastIndexOffset {
|
tt.lastIndex != querySlice.LastIndexOffset {
|
||||||
|
|
||||||
t.Errorf("First or last index does not match "+
|
t.Errorf("First or last index does not match "+
|
||||||
"expected index. Want (%d, %d), got (%d, %d).",
|
"expected index. Want (%d, %d), "+
|
||||||
|
"got (%d, %d).",
|
||||||
tt.firstIndex, tt.lastIndex,
|
tt.firstIndex, tt.lastIndex,
|
||||||
querySlice.FirstIndexOffset,
|
querySlice.FirstIndexOffset,
|
||||||
querySlice.LastIndexOffset)
|
querySlice.LastIndexOffset)
|
||||||
@@ -595,225 +454,18 @@ func TestQueryPayments(t *testing.T) {
|
|||||||
|
|
||||||
if len(querySlice.Payments) != len(tt.expectedSeqNrs) {
|
if len(querySlice.Payments) != len(tt.expectedSeqNrs) {
|
||||||
t.Errorf("expected: %v payments, got: %v",
|
t.Errorf("expected: %v payments, got: %v",
|
||||||
len(tt.expectedSeqNrs), len(querySlice.Payments))
|
len(tt.expectedSeqNrs),
|
||||||
|
len(querySlice.Payments))
|
||||||
}
|
}
|
||||||
|
|
||||||
for i, seqNr := range tt.expectedSeqNrs {
|
for i, seqNr := range tt.expectedSeqNrs {
|
||||||
q := querySlice.Payments[i]
|
q := querySlice.Payments[i]
|
||||||
if seqNr != q.SequenceNum {
|
if seqNr != q.SequenceNum {
|
||||||
t.Errorf("sequence numbers do not match, "+
|
t.Errorf("sequence numbers do not "+
|
||||||
"got %v, want %v", q.SequenceNum, seqNr)
|
"match, got %v, want %v",
|
||||||
|
q.SequenceNum, seqNr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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)
|
|
||||||
|
|
||||||
pControl := NewPaymentControl(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 = pControl.InitPayment(noDuplicates.PaymentIdentifier, noDuplicates)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
// Fetch the payment so we can get its sequence nr.
|
|
||||||
noDuplicatesPayment, err := pControl.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 = pControl.InitPayment(hasDuplicates.PaymentIdentifier, hasDuplicates)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
// Fetch the payment so we can get its sequence nr.
|
|
||||||
hasDuplicatesPayment, err := pControl.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)
|
|
||||||
}
|
|
||||||
@@ -924,6 +924,10 @@ type DatabaseInstances struct {
|
|||||||
// InvoiceDB is the database that stores information about invoices.
|
// InvoiceDB is the database that stores information about invoices.
|
||||||
InvoiceDB invoices.InvoiceDB
|
InvoiceDB invoices.InvoiceDB
|
||||||
|
|
||||||
|
// KVPaymentsDB is the database that stores all payment related
|
||||||
|
// information.
|
||||||
|
KVPaymentsDB *channeldb.KVPaymentsDB
|
||||||
|
|
||||||
// MacaroonDB is the database that stores macaroon root keys.
|
// MacaroonDB is the database that stores macaroon root keys.
|
||||||
MacaroonDB kvdb.Backend
|
MacaroonDB kvdb.Backend
|
||||||
|
|
||||||
@@ -1215,6 +1219,14 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
|
|||||||
return nil, nil, err
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Mount the payments DB which is only KV for now.
|
||||||
|
//
|
||||||
|
// TODO(ziggie): Add support for SQL payments DB.
|
||||||
|
kvPaymentsDB := channeldb.NewKVPaymentsDB(
|
||||||
|
dbs.ChanStateDB,
|
||||||
|
)
|
||||||
|
dbs.KVPaymentsDB = kvPaymentsDB
|
||||||
|
|
||||||
// Wrap the watchtower client DB and make sure we clean up.
|
// Wrap the watchtower client DB and make sure we clean up.
|
||||||
if cfg.WtClient.Active {
|
if cfg.WtClient.Active {
|
||||||
dbs.TowerClientDB, err = wtdb.OpenClientDB(
|
dbs.TowerClientDB, err = wtdb.OpenClientDB(
|
||||||
|
|||||||
@@ -151,7 +151,7 @@ func (s *controlTowerSubscriberImpl) Updates() <-chan interface{} {
|
|||||||
// controlTower is persistent implementation of ControlTower to restrict
|
// controlTower is persistent implementation of ControlTower to restrict
|
||||||
// double payment sending.
|
// double payment sending.
|
||||||
type controlTower struct {
|
type controlTower struct {
|
||||||
db *channeldb.PaymentControl
|
db *channeldb.KVPaymentsDB
|
||||||
|
|
||||||
// subscriberIndex is used to provide a unique id for each subscriber
|
// subscriberIndex is used to provide a unique id for each subscriber
|
||||||
// to all payments. This is used to easily remove the subscriber when
|
// to all payments. This is used to easily remove the subscriber when
|
||||||
@@ -168,7 +168,7 @@ type controlTower struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// NewControlTower creates a new instance of the controlTower.
|
// NewControlTower creates a new instance of the controlTower.
|
||||||
func NewControlTower(db *channeldb.PaymentControl) ControlTower {
|
func NewControlTower(db *channeldb.KVPaymentsDB) ControlTower {
|
||||||
return &controlTower{
|
return &controlTower{
|
||||||
db: db,
|
db: db,
|
||||||
subscribersAllPayments: make(
|
subscribersAllPayments: make(
|
||||||
|
|||||||
@@ -49,7 +49,7 @@ func TestControlTowerSubscribeUnknown(t *testing.T) {
|
|||||||
|
|
||||||
db := initDB(t, false)
|
db := initDB(t, false)
|
||||||
|
|
||||||
pControl := NewControlTower(channeldb.NewPaymentControl(db))
|
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db))
|
||||||
|
|
||||||
// Subscription should fail when the payment is not known.
|
// Subscription should fail when the payment is not known.
|
||||||
_, err := pControl.SubscribePayment(lntypes.Hash{1})
|
_, err := pControl.SubscribePayment(lntypes.Hash{1})
|
||||||
@@ -63,7 +63,7 @@ func TestControlTowerSubscribeSuccess(t *testing.T) {
|
|||||||
|
|
||||||
db := initDB(t, false)
|
db := initDB(t, false)
|
||||||
|
|
||||||
pControl := NewControlTower(channeldb.NewPaymentControl(db))
|
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db))
|
||||||
|
|
||||||
// Initiate a payment.
|
// Initiate a payment.
|
||||||
info, attempt, preimg, err := genInfo()
|
info, attempt, preimg, err := genInfo()
|
||||||
@@ -156,33 +156,33 @@ func TestControlTowerSubscribeSuccess(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPaymentControlSubscribeFail tests that payment updates for a
|
// TestKVPaymentsDBSubscribeFail tests that payment updates for a
|
||||||
// failed payment are properly sent to subscribers.
|
// failed payment are properly sent to subscribers.
|
||||||
func TestPaymentControlSubscribeFail(t *testing.T) {
|
func TestKVPaymentsDBSubscribeFail(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
t.Run("register attempt, keep failed payments", func(t *testing.T) {
|
t.Run("register attempt, keep failed payments", func(t *testing.T) {
|
||||||
testPaymentControlSubscribeFail(t, true, true)
|
testKVPaymentsDBSubscribeFail(t, true, true)
|
||||||
})
|
})
|
||||||
t.Run("register attempt, delete failed payments", func(t *testing.T) {
|
t.Run("register attempt, delete failed payments", func(t *testing.T) {
|
||||||
testPaymentControlSubscribeFail(t, true, false)
|
testKVPaymentsDBSubscribeFail(t, true, false)
|
||||||
})
|
})
|
||||||
t.Run("no register attempt, keep failed payments", func(t *testing.T) {
|
t.Run("no register attempt, keep failed payments", func(t *testing.T) {
|
||||||
testPaymentControlSubscribeFail(t, false, true)
|
testKVPaymentsDBSubscribeFail(t, false, true)
|
||||||
})
|
})
|
||||||
t.Run("no register attempt, delete failed payments", func(t *testing.T) {
|
t.Run("no register attempt, delete failed payments", func(t *testing.T) {
|
||||||
testPaymentControlSubscribeFail(t, false, false)
|
testKVPaymentsDBSubscribeFail(t, false, false)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPaymentControlSubscribeAllSuccess tests that multiple payments are
|
// TestKVPaymentsDBSubscribeAllSuccess tests that multiple payments are
|
||||||
// properly sent to subscribers of TrackPayments.
|
// properly sent to subscribers of TrackPayments.
|
||||||
func TestPaymentControlSubscribeAllSuccess(t *testing.T) {
|
func TestKVPaymentsDBSubscribeAllSuccess(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
db := initDB(t, true)
|
db := initDB(t, true)
|
||||||
|
|
||||||
pControl := NewControlTower(channeldb.NewPaymentControl(db))
|
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db))
|
||||||
|
|
||||||
// Initiate a payment.
|
// Initiate a payment.
|
||||||
info1, attempt1, preimg1, err := genInfo()
|
info1, attempt1, preimg1, err := genInfo()
|
||||||
@@ -288,14 +288,14 @@ func TestPaymentControlSubscribeAllSuccess(t *testing.T) {
|
|||||||
require.Equal(t, attempt2.Route, htlc2.Route, "unexpected htlc route.")
|
require.Equal(t, attempt2.Route, htlc2.Route, "unexpected htlc route.")
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPaymentControlSubscribeAllImmediate tests whether already inflight
|
// TestKVPaymentsDBSubscribeAllImmediate tests whether already inflight
|
||||||
// payments are reported at the start of the SubscribeAllPayments subscription.
|
// payments are reported at the start of the SubscribeAllPayments subscription.
|
||||||
func TestPaymentControlSubscribeAllImmediate(t *testing.T) {
|
func TestKVPaymentsDBSubscribeAllImmediate(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
db := initDB(t, true)
|
db := initDB(t, true)
|
||||||
|
|
||||||
pControl := NewControlTower(channeldb.NewPaymentControl(db))
|
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db))
|
||||||
|
|
||||||
// Initiate a payment.
|
// Initiate a payment.
|
||||||
info, attempt, _, err := genInfo()
|
info, attempt, _, err := genInfo()
|
||||||
@@ -325,14 +325,14 @@ func TestPaymentControlSubscribeAllImmediate(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestPaymentControlUnsubscribeSuccess tests that when unsubscribed, there are
|
// TestKVPaymentsDBUnsubscribeSuccess tests that when unsubscribed, there are
|
||||||
// no more notifications to that specific subscription.
|
// no more notifications to that specific subscription.
|
||||||
func TestPaymentControlUnsubscribeSuccess(t *testing.T) {
|
func TestKVPaymentsDBUnsubscribeSuccess(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
db := initDB(t, true)
|
db := initDB(t, true)
|
||||||
|
|
||||||
pControl := NewControlTower(channeldb.NewPaymentControl(db))
|
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db))
|
||||||
|
|
||||||
subscription1, err := pControl.SubscribeAllPayments()
|
subscription1, err := pControl.SubscribeAllPayments()
|
||||||
require.NoError(t, err, "expected subscribe to succeed, but got: %v")
|
require.NoError(t, err, "expected subscribe to succeed, but got: %v")
|
||||||
@@ -396,12 +396,12 @@ func TestPaymentControlUnsubscribeSuccess(t *testing.T) {
|
|||||||
require.Len(t, subscription2.Updates(), 0)
|
require.Len(t, subscription2.Updates(), 0)
|
||||||
}
|
}
|
||||||
|
|
||||||
func testPaymentControlSubscribeFail(t *testing.T, registerAttempt,
|
func testKVPaymentsDBSubscribeFail(t *testing.T, registerAttempt,
|
||||||
keepFailedPaymentAttempts bool) {
|
keepFailedPaymentAttempts bool) {
|
||||||
|
|
||||||
db := initDB(t, keepFailedPaymentAttempts)
|
db := initDB(t, keepFailedPaymentAttempts)
|
||||||
|
|
||||||
pControl := NewControlTower(channeldb.NewPaymentControl(db))
|
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db))
|
||||||
|
|
||||||
// Initiate a payment.
|
// Initiate a payment.
|
||||||
info, attempt, _, err := genInfo()
|
info, attempt, _, err := genInfo()
|
||||||
|
|||||||
+3
-3
@@ -7527,7 +7527,7 @@ func (r *rpcServer) ListPayments(ctx context.Context,
|
|||||||
query.MaxPayments = math.MaxUint64
|
query.MaxPayments = math.MaxUint64
|
||||||
}
|
}
|
||||||
|
|
||||||
paymentsQuerySlice, err := r.server.miscDB.QueryPayments(query)
|
paymentsQuerySlice, err := r.server.kvPaymentsDB.QueryPayments(query)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -7608,7 +7608,7 @@ func (r *rpcServer) DeletePayment(ctx context.Context,
|
|||||||
rpcsLog.Infof("[DeletePayment] payment_identifier=%v, "+
|
rpcsLog.Infof("[DeletePayment] payment_identifier=%v, "+
|
||||||
"failed_htlcs_only=%v", hash, req.FailedHtlcsOnly)
|
"failed_htlcs_only=%v", hash, req.FailedHtlcsOnly)
|
||||||
|
|
||||||
err = r.server.miscDB.DeletePayment(hash, req.FailedHtlcsOnly)
|
err = r.server.kvPaymentsDB.DeletePayment(hash, req.FailedHtlcsOnly)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -7648,7 +7648,7 @@ func (r *rpcServer) DeleteAllPayments(ctx context.Context,
|
|||||||
"failed_htlcs_only=%v", req.FailedPaymentsOnly,
|
"failed_htlcs_only=%v", req.FailedPaymentsOnly,
|
||||||
req.FailedHtlcsOnly)
|
req.FailedHtlcsOnly)
|
||||||
|
|
||||||
numDeletedPayments, err := r.server.miscDB.DeletePayments(
|
numDeletedPayments, err := r.server.kvPaymentsDB.DeletePayments(
|
||||||
req.FailedPaymentsOnly, req.FailedHtlcsOnly,
|
req.FailedPaymentsOnly, req.FailedHtlcsOnly,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -335,6 +335,12 @@ type server struct {
|
|||||||
|
|
||||||
invoicesDB invoices.InvoiceDB
|
invoicesDB invoices.InvoiceDB
|
||||||
|
|
||||||
|
// kvPaymentsDB is the DB that contains all functions for managing
|
||||||
|
// payments.
|
||||||
|
//
|
||||||
|
// TODO(ziggie): Replace with interface.
|
||||||
|
kvPaymentsDB *channeldb.KVPaymentsDB
|
||||||
|
|
||||||
aliasMgr *aliasmgr.Manager
|
aliasMgr *aliasmgr.Manager
|
||||||
|
|
||||||
htlcSwitch *htlcswitch.Switch
|
htlcSwitch *htlcswitch.Switch
|
||||||
@@ -678,6 +684,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
|
|||||||
addrSource: addrSource,
|
addrSource: addrSource,
|
||||||
miscDB: dbs.ChanStateDB,
|
miscDB: dbs.ChanStateDB,
|
||||||
invoicesDB: dbs.InvoiceDB,
|
invoicesDB: dbs.InvoiceDB,
|
||||||
|
kvPaymentsDB: dbs.KVPaymentsDB,
|
||||||
cc: cc,
|
cc: cc,
|
||||||
sigPool: lnwallet.NewSigPool(cfg.Workers.Sig, cc.Signer),
|
sigPool: lnwallet.NewSigPool(cfg.Workers.Sig, cc.Signer),
|
||||||
writePool: writePool,
|
writePool: writePool,
|
||||||
@@ -1127,9 +1134,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
|
|||||||
PathFindingConfig: pathFindingConfig,
|
PathFindingConfig: pathFindingConfig,
|
||||||
}
|
}
|
||||||
|
|
||||||
paymentControl := channeldb.NewPaymentControl(dbs.ChanStateDB)
|
s.controlTower = routing.NewControlTower(dbs.KVPaymentsDB)
|
||||||
|
|
||||||
s.controlTower = routing.NewControlTower(paymentControl)
|
|
||||||
|
|
||||||
strictPruning := cfg.Bitcoin.Node == "neutrino" ||
|
strictPruning := cfg.Bitcoin.Node == "neutrino" ||
|
||||||
cfg.Routing.StrictZombiePruning
|
cfg.Routing.StrictZombiePruning
|
||||||
|
|||||||
Reference in new issue
Block a user