Merge pull request #9844 from ziggie1984/refactor-payments-code-03

Refactor Payment PR 3
This commit is contained in:
Yong authored and GitHub committed 2025-08-13 06:05:01 -04:00
commit c6a9116e3a
20 files changed
+538 -350

No files matched your search

-2
View File
@@ -352,7 +352,6 @@ type DB struct {
dbPath string dbPath string
clock clock.Clock clock clock.Clock
dryRun bool dryRun bool
keepFailedPaymentAttempts bool
storeFinalHtlcResolutions bool storeFinalHtlcResolutions bool
// noRevLogAmtData if true, means that commitment transaction amount // noRevLogAmtData if true, means that commitment transaction amount
@@ -413,7 +412,6 @@ func CreateWithBackend(backend kvdb.Backend, modifiers ...OptionModifier) (*DB,
}, },
clock: opts.clock, clock: opts.clock,
dryRun: opts.dryRun, dryRun: opts.dryRun,
keepFailedPaymentAttempts: opts.keepFailedPaymentAttempts,
storeFinalHtlcResolutions: opts.storeFinalHtlcResolutions, storeFinalHtlcResolutions: opts.storeFinalHtlcResolutions,
noRevLogAmtData: opts.NoRevLogAmtData, noRevLogAmtData: opts.NoRevLogAmtData,
} }
+2 -1
View File
@@ -11,6 +11,7 @@ import (
"github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/routing/route"
) )
@@ -74,7 +75,7 @@ func fetchDuplicatePaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
return StatusInFlight, nil return StatusInFlight, nil
} }
return 0, ErrPaymentNotInitiated return 0, paymentsdb.ErrPaymentNotInitiated
} }
func deserializeDuplicateHTLCAttemptInfo(r io.Reader) ( func deserializeDuplicateHTLCAttemptInfo(r io.Reader) (
+17 -13
View File
@@ -15,6 +15,7 @@ import (
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnutils" "github.com/lightningnetwork/lnd/lnutils"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/routing/route"
) )
@@ -349,12 +350,12 @@ func (m *MPPayment) Registrable() error {
// are settled HTLCs or the payment is failed. If we already have // are settled HTLCs or the payment is failed. If we already have
// settled HTLCs, we won't allow adding more HTLCs. // settled HTLCs, we won't allow adding more HTLCs.
if m.State.HasSettledHTLC { if m.State.HasSettledHTLC {
return ErrPaymentPendingSettled return paymentsdb.ErrPaymentPendingSettled
} }
// If the payment is already failed, we won't allow adding more HTLCs. // If the payment is already failed, we won't allow adding more HTLCs.
if m.State.PaymentFailed { if m.State.PaymentFailed {
return ErrPaymentPendingFailed return paymentsdb.ErrPaymentPendingFailed
} }
// Otherwise we can add more HTLCs. // Otherwise we can add more HTLCs.
@@ -371,8 +372,8 @@ func (m *MPPayment) setState() error {
// Sanity check we haven't sent a value larger than the payment amount. // Sanity check we haven't sent a value larger than the payment amount.
totalAmt := m.Info.Value totalAmt := m.Info.Value
if sentAmt > totalAmt { if sentAmt > totalAmt {
return fmt.Errorf("%w: sent=%v, total=%v", ErrSentExceedsTotal, return fmt.Errorf("%w: sent=%v, total=%v",
sentAmt, totalAmt) paymentsdb.ErrSentExceedsTotal, sentAmt, totalAmt)
} }
// Get any terminal info for this payment. // Get any terminal info for this payment.
@@ -451,7 +452,7 @@ func (m *MPPayment) NeedWaitAttempts() (bool, error) {
case StatusSucceeded: case StatusSucceeded:
return false, fmt.Errorf("%w: parts of the payment "+ return false, fmt.Errorf("%w: parts of the payment "+
"already succeeded but still have remaining "+ "already succeeded but still have remaining "+
"amount %v", ErrPaymentInternal, "amount %v", paymentsdb.ErrPaymentInternal,
m.State.RemainingAmt) m.State.RemainingAmt)
// The payment is failed and we have no inflight HTLCs, no need // The payment is failed and we have no inflight HTLCs, no need
@@ -462,7 +463,7 @@ func (m *MPPayment) NeedWaitAttempts() (bool, error) {
// Unknown payment status. // Unknown payment status.
default: default:
return false, fmt.Errorf("%w: %s", return false, fmt.Errorf("%w: %s",
ErrUnknownPaymentStatus, m.Status) paymentsdb.ErrUnknownPaymentStatus, m.Status)
} }
} }
@@ -472,7 +473,8 @@ func (m *MPPayment) NeedWaitAttempts() (bool, error) {
// When the payment is newly created, yet the payment has no remaining // When the payment is newly created, yet the payment has no remaining
// amount, return an error. // amount, return an error.
case StatusInitiated: case StatusInitiated:
return false, fmt.Errorf("%w: %v", ErrPaymentInternal, m.Status) return false, fmt.Errorf("%w: %v",
paymentsdb.ErrPaymentInternal, m.Status)
// If the payment is inflight, we must wait. // If the payment is inflight, we must wait.
// //
@@ -493,12 +495,13 @@ func (m *MPPayment) NeedWaitAttempts() (bool, error) {
// marked as failed with a reason, which means the remainingAmt must // marked as failed with a reason, which means the remainingAmt must
// not be zero because our sentAmt is zero. // not be zero because our sentAmt is zero.
case StatusFailed: case StatusFailed:
return false, fmt.Errorf("%w: %v", ErrPaymentInternal, m.Status) return false, fmt.Errorf("%w: %v",
paymentsdb.ErrPaymentInternal, m.Status)
// Unknown payment status. // Unknown payment status.
default: default:
return false, fmt.Errorf("%w: %s", ErrUnknownPaymentStatus, return false, fmt.Errorf("%w: %s",
m.Status) paymentsdb.ErrUnknownPaymentStatus, m.Status)
} }
} }
@@ -528,7 +531,8 @@ func (m *MPPayment) AllowMoreAttempts() (bool, error) {
// remainingAmt, return an error. // remainingAmt, return an error.
if m.Status == StatusInitiated { if m.Status == StatusInitiated {
return false, fmt.Errorf("%w: initiated payment has "+ return false, fmt.Errorf("%w: initiated payment has "+
"zero remainingAmt", ErrPaymentInternal) "zero remainingAmt",
paymentsdb.ErrPaymentInternal)
} }
// Otherwise, exit early since all other statuses with zero // Otherwise, exit early since all other statuses with zero
@@ -544,8 +548,8 @@ func (m *MPPayment) AllowMoreAttempts() (bool, error) {
// as the preimage is received. In this case, return an error state. // as the preimage is received. In this case, return an error state.
if m.Status == StatusSucceeded { if m.Status == StatusSucceeded {
return false, fmt.Errorf("%w: payment already succeeded but "+ return false, fmt.Errorf("%w: payment already succeeded but "+
"still have remaining amount %v", ErrPaymentInternal, "still have remaining amount %v",
m.State.RemainingAmt) paymentsdb.ErrPaymentInternal, m.State.RemainingAmt)
} }
// Now check if we can register a new HTLC. // Now check if we can register a new HTLC.
+15 -14
View File
@@ -8,6 +8,7 @@ import (
"github.com/btcsuite/btcd/btcec/v2" "github.com/btcsuite/btcd/btcec/v2"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/routing/route"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -65,35 +66,35 @@ func TestRegistrable(t *testing.T) {
// Test inflight status with settled HTLC but no failed // Test inflight status with settled HTLC but no failed
// payment. // payment.
status: StatusInFlight, status: StatusInFlight,
registryErr: ErrPaymentPendingSettled, registryErr: paymentsdb.ErrPaymentPendingSettled,
hasSettledHTLC: true, hasSettledHTLC: true,
}, },
{ {
// Test inflight status with no settled HTLC but failed // Test inflight status with no settled HTLC but failed
// payment. // payment.
status: StatusInFlight, status: StatusInFlight,
registryErr: ErrPaymentPendingFailed, registryErr: paymentsdb.ErrPaymentPendingFailed,
paymentFailed: true, paymentFailed: true,
}, },
{ {
// Test error state with settled HTLC and failed // Test error state with settled HTLC and failed
// payment. // payment.
status: 0, status: 0,
registryErr: ErrUnknownPaymentStatus, registryErr: paymentsdb.ErrUnknownPaymentStatus,
hasSettledHTLC: true, hasSettledHTLC: true,
paymentFailed: true, paymentFailed: true,
}, },
{ {
status: StatusSucceeded, status: StatusSucceeded,
registryErr: ErrPaymentAlreadySucceeded, registryErr: paymentsdb.ErrPaymentAlreadySucceeded,
}, },
{ {
status: StatusFailed, status: StatusFailed,
registryErr: ErrPaymentAlreadyFailed, registryErr: paymentsdb.ErrPaymentAlreadyFailed,
}, },
{ {
status: 0, status: 0,
registryErr: ErrUnknownPaymentStatus, registryErr: paymentsdb.ErrUnknownPaymentStatus,
}, },
} }
@@ -149,7 +150,7 @@ func TestPaymentSetState(t *testing.T) {
}, },
}, },
totalAmt: 1, totalAmt: 1,
errExpected: ErrSentExceedsTotal, errExpected: paymentsdb.ErrSentExceedsTotal,
}, },
{ {
// Test that when the htlc is failed, the fee is not // Test that when the htlc is failed, the fee is not
@@ -294,7 +295,7 @@ func TestNeedWaitAttempts(t *testing.T) {
status: StatusSucceeded, status: StatusSucceeded,
remainingAmt: 1000, remainingAmt: 1000,
needWait: false, needWait: false,
expectedErr: ErrPaymentInternal, expectedErr: paymentsdb.ErrPaymentInternal,
}, },
{ {
// Payment is in terminal state, no need to wait. // Payment is in terminal state, no need to wait.
@@ -309,7 +310,7 @@ func TestNeedWaitAttempts(t *testing.T) {
status: StatusInitiated, status: StatusInitiated,
remainingAmt: 0, remainingAmt: 0,
needWait: false, needWait: false,
expectedErr: ErrPaymentInternal, expectedErr: paymentsdb.ErrPaymentInternal,
}, },
{ {
// With zero remainingAmt we must wait for the results. // With zero remainingAmt we must wait for the results.
@@ -330,21 +331,21 @@ func TestNeedWaitAttempts(t *testing.T) {
status: StatusFailed, status: StatusFailed,
remainingAmt: 0, remainingAmt: 0,
needWait: false, needWait: false,
expectedErr: ErrPaymentInternal, expectedErr: paymentsdb.ErrPaymentInternal,
}, },
{ {
// Payment is in an unknown status, return an error. // Payment is in an unknown status, return an error.
status: 0, status: 0,
remainingAmt: 0, remainingAmt: 0,
needWait: false, needWait: false,
expectedErr: ErrUnknownPaymentStatus, expectedErr: paymentsdb.ErrUnknownPaymentStatus,
}, },
{ {
// Payment is in an unknown status, return an error. // Payment is in an unknown status, return an error.
status: 0, status: 0,
remainingAmt: 1000, remainingAmt: 1000,
needWait: false, needWait: false,
expectedErr: ErrUnknownPaymentStatus, expectedErr: paymentsdb.ErrUnknownPaymentStatus,
}, },
} }
@@ -397,7 +398,7 @@ func TestAllowMoreAttempts(t *testing.T) {
status: StatusInitiated, status: StatusInitiated,
remainingAmt: 0, remainingAmt: 0,
allowMore: false, allowMore: false,
expectedErr: ErrPaymentInternal, expectedErr: paymentsdb.ErrPaymentInternal,
}, },
{ {
// With zero remainingAmt we don't allow more HTLC // With zero remainingAmt we don't allow more HTLC
@@ -505,7 +506,7 @@ func TestAllowMoreAttempts(t *testing.T) {
remainingAmt: 1000, remainingAmt: 1000,
hasSettledHTLC: true, hasSettledHTLC: true,
allowMore: false, allowMore: false,
expectedErr: ErrPaymentInternal, expectedErr: paymentsdb.ErrPaymentInternal,
}, },
{ {
// With the payment failed with no inflight HTLCs, we // With the payment failed with no inflight HTLCs, we
-12
View File
@@ -68,10 +68,6 @@ type Options struct {
// database if set to true. // database if set to true.
dryRun bool dryRun bool
// keepFailedPaymentAttempts determines whether failed htlc attempts
// are kept on disk or removed to save space.
keepFailedPaymentAttempts bool
// storeFinalHtlcResolutions determines whether to persistently store // storeFinalHtlcResolutions determines whether to persistently store
// the final resolution of incoming htlcs. // the final resolution of incoming htlcs.
storeFinalHtlcResolutions bool storeFinalHtlcResolutions bool
@@ -120,14 +116,6 @@ func OptionDryRunMigration(dryRun bool) OptionModifier {
} }
} }
// OptionKeepFailedPaymentAttempts controls whether failed payment attempts are
// kept on disk after a payment settles.
func OptionKeepFailedPaymentAttempts(keepFailedPaymentAttempts bool) OptionModifier {
return func(o *Options) {
o.keepFailedPaymentAttempts = keepFailedPaymentAttempts
}
}
// OptionStoreFinalHtlcResolutions controls whether to persistently store the // OptionStoreFinalHtlcResolutions controls whether to persistently store the
// final resolution of incoming htlcs. // final resolution of incoming htlcs.
func OptionStoreFinalHtlcResolutions( func OptionStoreFinalHtlcResolutions(
+17 -10
View File
@@ -1,6 +1,10 @@
package channeldb package channeldb
import "fmt" import (
"fmt"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
)
// PaymentStatus represent current status of payment. // PaymentStatus represent current status of payment.
type PaymentStatus byte type PaymentStatus byte
@@ -58,24 +62,25 @@ func (ps PaymentStatus) initializable() error {
// again in case other goroutines have already been creating HTLCs for // again in case other goroutines have already been creating HTLCs for
// it. // it.
case StatusInitiated: case StatusInitiated:
return ErrPaymentExists return paymentsdb.ErrPaymentExists
// We already have an InFlight payment on the network. We will disallow // We already have an InFlight payment on the network. We will disallow
// any new payments. // any new payments.
case StatusInFlight: case StatusInFlight:
return ErrPaymentInFlight return paymentsdb.ErrPaymentInFlight
// The payment has been attempted and is succeeded so we won't allow // The payment has been attempted and is succeeded so we won't allow
// creating it again. // creating it again.
case StatusSucceeded: case StatusSucceeded:
return ErrAlreadyPaid return paymentsdb.ErrAlreadyPaid
// We allow retrying failed payments. // We allow retrying failed payments.
case StatusFailed: case StatusFailed:
return nil return nil
default: default:
return fmt.Errorf("%w: %v", ErrUnknownPaymentStatus, ps) return fmt.Errorf("%w: %v", paymentsdb.ErrUnknownPaymentStatus,
ps)
} }
} }
@@ -91,7 +96,7 @@ func (ps PaymentStatus) removable() error {
// There are still inflight HTLCs and the payment needs to wait for the // There are still inflight HTLCs and the payment needs to wait for the
// final outcomes. // final outcomes.
case StatusInFlight: case StatusInFlight:
return ErrPaymentInFlight return paymentsdb.ErrPaymentInFlight
// The payment has been attempted and is succeeded and is allowed to be // The payment has been attempted and is succeeded and is allowed to be
// removed. // removed.
@@ -103,7 +108,8 @@ func (ps PaymentStatus) removable() error {
return nil return nil
default: default:
return fmt.Errorf("%w: %v", ErrUnknownPaymentStatus, ps) return fmt.Errorf("%w: %v", paymentsdb.ErrUnknownPaymentStatus,
ps)
} }
} }
@@ -121,13 +127,14 @@ func (ps PaymentStatus) updatable() error {
// If the payment has a terminal condition, we won't allow any updates. // If the payment has a terminal condition, we won't allow any updates.
case StatusSucceeded: case StatusSucceeded:
return ErrPaymentAlreadySucceeded return paymentsdb.ErrPaymentAlreadySucceeded
case StatusFailed: case StatusFailed:
return ErrPaymentAlreadyFailed return paymentsdb.ErrPaymentAlreadyFailed
default: default:
return fmt.Errorf("%w: %v", ErrUnknownPaymentStatus, ps) return fmt.Errorf("%w: %v", paymentsdb.ErrUnknownPaymentStatus,
ps)
} }
} }
+10 -9
View File
@@ -5,6 +5,7 @@ import (
"testing" "testing"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -197,33 +198,33 @@ func TestPaymentStatusActions(t *testing.T) {
}{ }{
{ {
status: StatusInitiated, status: StatusInitiated,
initErr: ErrPaymentExists, initErr: paymentsdb.ErrPaymentExists,
updateErr: nil, updateErr: nil,
removeErr: nil, removeErr: nil,
}, },
{ {
status: StatusInFlight, status: StatusInFlight,
initErr: ErrPaymentInFlight, initErr: paymentsdb.ErrPaymentInFlight,
updateErr: nil, updateErr: nil,
removeErr: ErrPaymentInFlight, removeErr: paymentsdb.ErrPaymentInFlight,
}, },
{ {
status: StatusSucceeded, status: StatusSucceeded,
initErr: ErrAlreadyPaid, initErr: paymentsdb.ErrAlreadyPaid,
updateErr: ErrPaymentAlreadySucceeded, updateErr: paymentsdb.ErrPaymentAlreadySucceeded,
removeErr: nil, removeErr: nil,
}, },
{ {
status: StatusFailed, status: StatusFailed,
initErr: nil, initErr: nil,
updateErr: ErrPaymentAlreadyFailed, updateErr: paymentsdb.ErrPaymentAlreadyFailed,
removeErr: nil, removeErr: nil,
}, },
{ {
status: 0, status: 0,
initErr: ErrUnknownPaymentStatus, initErr: paymentsdb.ErrUnknownPaymentStatus,
updateErr: ErrUnknownPaymentStatus, updateErr: paymentsdb.ErrUnknownPaymentStatus,
removeErr: ErrUnknownPaymentStatus, removeErr: paymentsdb.ErrUnknownPaymentStatus,
}, },
} }
+102 -178
View File
@@ -2,6 +2,7 @@ package channeldb
import ( import (
"bytes" "bytes"
"context"
"encoding/binary" "encoding/binary"
"errors" "errors"
"fmt" "fmt"
@@ -15,6 +16,7 @@ import (
"github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"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/lightningnetwork/lnd/tlv"
@@ -30,113 +32,6 @@ const (
paymentProgressLogInterval = 30 * time.Second 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")
)
//nolint:ll //nolint:ll
var ( var (
// paymentsRootBucket is the name of the top-level bucket within the // paymentsRootBucket is the name of the top-level bucket within the
@@ -222,41 +117,70 @@ var (
paymentsIndexBucket = []byte("payments-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. // KVPaymentsDB implements persistence for payments and payment attempts.
type KVPaymentsDB struct { type KVPaymentsDB struct {
paymentSeqMx sync.Mutex // Sequence management for the kv store.
currPaymentSeq uint64 seqMu sync.Mutex
storedPaymentSeq uint64 currSeq uint64
db *DB storedSeq uint64
// db is the underlying database implementation.
db kvdb.Backend
keepFailedPaymentAttempts bool
} }
// NewKVPaymentsDB creates a new instance of the KVPaymentsDB. // defaultKVStoreOptions returns the default options for the KV store.
func NewKVPaymentsDB(db *DB) *KVPaymentsDB { func defaultKVStoreOptions() *paymentsdb.StoreOptions {
return &KVPaymentsDB{ return &paymentsdb.StoreOptions{
db: db, KeepFailedPaymentAttempts: false,
} }
} }
// NewKVPaymentsDB creates a new KVStore for payments.
func NewKVPaymentsDB(db kvdb.Backend,
options ...paymentsdb.OptionModifier) (*KVPaymentsDB, error) {
opts := defaultKVStoreOptions()
for _, applyOption := range options {
applyOption(opts)
}
if !opts.NoMigration {
if err := initKVStore(db); err != nil {
return nil, err
}
}
return &KVPaymentsDB{
db: db,
keepFailedPaymentAttempts: opts.KeepFailedPaymentAttempts,
}, nil
}
var paymentsTopLevelBuckets = [][]byte{
paymentsRootBucket,
paymentsIndexBucket,
}
// initKVStore creates and initializes the top-level buckets for the payment db.
func initKVStore(db kvdb.Backend) error {
err := kvdb.Update(db, func(tx kvdb.RwTx) error {
for _, tlb := range paymentsTopLevelBuckets {
if _, err := tx.CreateTopLevelBucket(tlb); err != nil {
return err
}
}
return nil
}, func() {})
if err != nil {
return fmt.Errorf("unable to create new payments db: %w", err)
}
return nil
}
// InitPayment checks or records the given PaymentCreationInfo with the 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 // 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 // method returns successfully, the payment is guaranteed to be in the InFlight
@@ -279,7 +203,7 @@ func (p *KVPaymentsDB) InitPayment(paymentHash lntypes.Hash,
infoBytes := b.Bytes() infoBytes := b.Bytes()
var updateErr error var updateErr error
err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error { err = kvdb.Batch(p.db, func(tx kvdb.RwTx) error {
// Reset the update error, to avoid carrying over an error // Reset the update error, to avoid carrying over an error
// from a previous execution of the batched db transaction. // from a previous execution of the batched db transaction.
updateErr = nil updateErr = nil
@@ -305,7 +229,7 @@ func (p *KVPaymentsDB) InitPayment(paymentHash lntypes.Hash,
// Otherwise, if the error is not `ErrPaymentNotInitiated`, // Otherwise, if the error is not `ErrPaymentNotInitiated`,
// we'll return the error. // we'll return the error.
case !errors.Is(err, ErrPaymentNotInitiated): case !errors.Is(err, paymentsdb.ErrPaymentNotInitiated):
return err return err
} }
@@ -366,7 +290,7 @@ func (p *KVPaymentsDB) InitPayment(paymentHash lntypes.Hash,
// DeleteFailedAttempts deletes all failed htlcs for a payment if configured // DeleteFailedAttempts deletes all failed htlcs for a payment if configured
// by the KVPaymentsDB db. // by the KVPaymentsDB db.
func (p *KVPaymentsDB) DeleteFailedAttempts(hash lntypes.Hash) error { func (p *KVPaymentsDB) DeleteFailedAttempts(hash lntypes.Hash) error {
if !p.db.keepFailedPaymentAttempts { if !p.keepFailedPaymentAttempts {
const failedHtlcsOnly = true const failedHtlcsOnly = true
err := p.DeletePayment(hash, failedHtlcsOnly) err := p.DeletePayment(hash, failedHtlcsOnly)
if err != nil { if err != nil {
@@ -447,7 +371,7 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
binary.BigEndian.PutUint64(htlcIDBytes, attempt.AttemptID) binary.BigEndian.PutUint64(htlcIDBytes, attempt.AttemptID)
var payment *MPPayment var payment *MPPayment
err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error { err = kvdb.Batch(p.db, func(tx kvdb.RwTx) error {
prefetchPayment(tx, paymentHash) prefetchPayment(tx, paymentHash)
bucket, err := fetchPaymentBucketUpdate(tx, paymentHash) bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
if err != nil { if err != nil {
@@ -479,7 +403,7 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
// MPP records should not be set for attempts to blinded paths. // MPP records should not be set for attempts to blinded paths.
if isBlinded && mpp != nil { if isBlinded && mpp != nil {
return ErrMPPRecordInBlindedPayment return paymentsdb.ErrMPPRecordInBlindedPayment
} }
for _, h := range payment.InFlightHTLCs() { for _, h := range payment.InFlightHTLCs() {
@@ -488,7 +412,7 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
// If this is a blinded payment, then no existing HTLCs // If this is a blinded payment, then no existing HTLCs
// should have MPP records. // should have MPP records.
if isBlinded && hMpp != nil { if isBlinded && hMpp != nil {
return ErrMPPRecordInBlindedPayment return paymentsdb.ErrMPPRecordInBlindedPayment
} }
// If this is a blinded payment, then we just need to // If this is a blinded payment, then we just need to
@@ -500,7 +424,7 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
h.Route.FinalHop().TotalAmtMsat { h.Route.FinalHop().TotalAmtMsat {
//nolint:ll //nolint:ll
return ErrBlindedPaymentTotalAmountMismatch return paymentsdb.ErrBlindedPaymentTotalAmountMismatch
} }
continue continue
@@ -510,12 +434,12 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
// We tried to register a non-MPP attempt for a MPP // We tried to register a non-MPP attempt for a MPP
// payment. // payment.
case mpp == nil && hMpp != nil: case mpp == nil && hMpp != nil:
return ErrMPPayment return paymentsdb.ErrMPPayment
// We tried to register a MPP shard for a non-MPP // We tried to register a MPP shard for a non-MPP
// payment. // payment.
case mpp != nil && hMpp == nil: case mpp != nil && hMpp == nil:
return ErrNonMPPayment return paymentsdb.ErrNonMPPayment
// Non-MPP payment, nothing more to validate. // Non-MPP payment, nothing more to validate.
case mpp == nil: case mpp == nil:
@@ -524,11 +448,11 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
// Check that MPP options match. // Check that MPP options match.
if mpp.PaymentAddr() != hMpp.PaymentAddr() { if mpp.PaymentAddr() != hMpp.PaymentAddr() {
return ErrMPPPaymentAddrMismatch return paymentsdb.ErrMPPPaymentAddrMismatch
} }
if mpp.TotalMsat() != hMpp.TotalMsat() { if mpp.TotalMsat() != hMpp.TotalMsat() {
return ErrMPPTotalAmountMismatch return paymentsdb.ErrMPPTotalAmountMismatch
} }
} }
@@ -537,15 +461,15 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
// attempt. // attempt.
amt := attempt.Route.ReceiverAmt() amt := attempt.Route.ReceiverAmt()
if !isBlinded && mpp == nil && amt != payment.Info.Value { if !isBlinded && mpp == nil && amt != payment.Info.Value {
return ErrValueMismatch return paymentsdb.ErrValueMismatch
} }
// Ensure we aren't sending more than the total payment amount. // Ensure we aren't sending more than the total payment amount.
sentAmt, _ := payment.SentAmt() sentAmt, _ := payment.SentAmt()
if sentAmt+amt > payment.Info.Value { if sentAmt+amt > payment.Info.Value {
return fmt.Errorf("%w: attempted=%v, payment amount="+ return fmt.Errorf("%w: attempted=%v, payment amount="+
"%v", ErrValueExceedsAmt, sentAmt+amt, "%v", paymentsdb.ErrValueExceedsAmt,
payment.Info.Value) sentAmt+amt, payment.Info.Value)
} }
htlcsBucket, err := bucket.CreateBucketIfNotExists( htlcsBucket, err := bucket.CreateBucketIfNotExists(
@@ -615,7 +539,7 @@ func (p *KVPaymentsDB) updateHtlcKey(paymentHash lntypes.Hash,
binary.BigEndian.PutUint64(aid, attemptID) binary.BigEndian.PutUint64(aid, attemptID)
var payment *MPPayment var payment *MPPayment
err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error { err := kvdb.Batch(p.db, func(tx kvdb.RwTx) error {
payment = nil payment = nil
prefetchPayment(tx, paymentHash) prefetchPayment(tx, paymentHash)
@@ -650,12 +574,12 @@ func (p *KVPaymentsDB) updateHtlcKey(paymentHash lntypes.Hash,
// Make sure the shard is not already failed or settled. // Make sure the shard is not already failed or settled.
failKey := htlcBucketKey(htlcFailInfoKey, aid) failKey := htlcBucketKey(htlcFailInfoKey, aid)
if htlcsBucket.Get(failKey) != nil { if htlcsBucket.Get(failKey) != nil {
return ErrAttemptAlreadyFailed return paymentsdb.ErrAttemptAlreadyFailed
} }
settleKey := htlcBucketKey(htlcSettleInfoKey, aid) settleKey := htlcBucketKey(htlcSettleInfoKey, aid)
if htlcsBucket.Get(settleKey) != nil { if htlcsBucket.Get(settleKey) != nil {
return ErrAttemptAlreadySettled return paymentsdb.ErrAttemptAlreadySettled
} }
// Add or update the key for this htlc. // Add or update the key for this htlc.
@@ -687,7 +611,7 @@ func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
updateErr error updateErr error
payment *MPPayment payment *MPPayment
) )
err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error { err := kvdb.Batch(p.db, func(tx kvdb.RwTx) error {
// Reset the update error, to avoid carrying over an error // Reset the update error, to avoid carrying over an error
// from a previous execution of the batched db transaction. // from a previous execution of the batched db transaction.
updateErr = nil updateErr = nil
@@ -695,8 +619,8 @@ func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
prefetchPayment(tx, paymentHash) prefetchPayment(tx, paymentHash)
bucket, err := fetchPaymentBucketUpdate(tx, paymentHash) bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
if errors.Is(err, ErrPaymentNotInitiated) { if errors.Is(err, paymentsdb.ErrPaymentNotInitiated) {
updateErr = ErrPaymentNotInitiated updateErr = paymentsdb.ErrPaymentNotInitiated
return nil return nil
} else if err != nil { } else if err != nil {
return err return err
@@ -707,8 +631,8 @@ func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
// failure to the KVPaymentsDB without synchronizing with // failure to the KVPaymentsDB without synchronizing with
// other attempts. // other attempts.
_, err = fetchPaymentStatus(bucket) _, err = fetchPaymentStatus(bucket)
if errors.Is(err, ErrPaymentNotInitiated) { if errors.Is(err, paymentsdb.ErrPaymentNotInitiated) {
updateErr = ErrPaymentNotInitiated updateErr = paymentsdb.ErrPaymentNotInitiated
return nil return nil
} else if err != nil { } else if err != nil {
return err return err
@@ -801,12 +725,12 @@ func fetchPaymentBucket(tx kvdb.RTx, paymentHash lntypes.Hash) (
payments := tx.ReadBucket(paymentsRootBucket) payments := tx.ReadBucket(paymentsRootBucket)
if payments == nil { if payments == nil {
return nil, ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
bucket := payments.NestedReadBucket(paymentHash[:]) bucket := payments.NestedReadBucket(paymentHash[:])
if bucket == nil { if bucket == nil {
return nil, ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
return bucket, nil return bucket, nil
@@ -819,12 +743,12 @@ func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
payments := tx.ReadWriteBucket(paymentsRootBucket) payments := tx.ReadWriteBucket(paymentsRootBucket)
if payments == nil { if payments == nil {
return nil, ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
bucket := payments.NestedReadWriteBucket(paymentHash[:]) bucket := payments.NestedReadWriteBucket(paymentHash[:])
if bucket == nil { if bucket == nil {
return nil, ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
return bucket, nil return bucket, nil
@@ -833,14 +757,14 @@ func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
// nextPaymentSequence returns the next sequence number to store for a new // nextPaymentSequence returns the next sequence number to store for a new
// payment. // payment.
func (p *KVPaymentsDB) nextPaymentSequence() ([]byte, error) { func (p *KVPaymentsDB) nextPaymentSequence() ([]byte, error) {
p.paymentSeqMx.Lock() p.seqMu.Lock()
defer p.paymentSeqMx.Unlock() defer p.seqMu.Unlock()
// Set a new upper bound in the DB every 1000 payments to avoid // Set a new upper bound in the DB every 1000 payments to avoid
// conflicts on the sequence when using etcd. // conflicts on the sequence when using etcd.
if p.currPaymentSeq == p.storedPaymentSeq { if p.currSeq == p.storedSeq {
var currPaymentSeq, newUpperBound uint64 var currPaymentSeq, newUpperBound uint64
if err := kvdb.Update(p.db.Backend, func(tx kvdb.RwTx) error { if err := kvdb.Update(p.db, func(tx kvdb.RwTx) error {
paymentsBucket, err := tx.CreateTopLevelBucket( paymentsBucket, err := tx.CreateTopLevelBucket(
paymentsRootBucket, paymentsRootBucket,
) )
@@ -861,16 +785,16 @@ func (p *KVPaymentsDB) nextPaymentSequence() ([]byte, error) {
// initialize our stored currPaymentSeq, since by default both // initialize our stored currPaymentSeq, since by default both
// this variable and storedPaymentSeq are zero which in turn // this variable and storedPaymentSeq are zero which in turn
// will have us fetch the current values from the DB. // will have us fetch the current values from the DB.
if p.currPaymentSeq == 0 { if p.currSeq == 0 {
p.currPaymentSeq = currPaymentSeq p.currSeq = currPaymentSeq
} }
p.storedPaymentSeq = newUpperBound p.storedSeq = newUpperBound
} }
p.currPaymentSeq++ p.currSeq++
b := make([]byte, 8) b := make([]byte, 8)
binary.BigEndian.PutUint64(b, p.currPaymentSeq) binary.BigEndian.PutUint64(b, p.currSeq)
return b, nil return b, nil
} }
@@ -881,7 +805,7 @@ func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
// Creation info should be set for all payments, regardless of state. // Creation info should be set for all payments, regardless of state.
// If not, it is unknown. // If not, it is unknown.
if bucket.Get(paymentCreationInfoKey) == nil { if bucket.Get(paymentCreationInfoKey) == nil {
return 0, ErrPaymentNotInitiated return 0, paymentsdb.ErrPaymentNotInitiated
} }
payment, err := fetchPayment(bucket) payment, err := fetchPayment(bucket)
@@ -1127,7 +1051,7 @@ func fetchHtlcAttempts(bucket kvdb.RBucket) ([]HTLCAttempt, error) {
// Sanity check that all htlcs have an attempt info. // Sanity check that all htlcs have an attempt info.
if attemptInfoCount != len(htlcsMap) { if attemptInfoCount != len(htlcsMap) {
return nil, errNoAttemptInfo return nil, paymentsdb.ErrNoAttemptInfo
} }
keys := make([]uint64, len(htlcsMap)) keys := make([]uint64, len(htlcsMap))
@@ -1206,8 +1130,8 @@ func fetchFailedHtlcKeys(bucket kvdb.RBucket) ([][]byte, error) {
// QueryPayments is a query to the payments database which is restricted // QueryPayments is a query to the payments database which is restricted
// to a subset of payments by the payments query, containing an offset // to a subset of payments by the payments query, containing an offset
// index and a maximum number of returned payments. // index and a maximum number of returned payments.
func (p *KVPaymentsDB) QueryPayments(query PaymentsQuery) (PaymentsResponse, func (p *KVPaymentsDB) QueryPayments(_ context.Context,
error) { query PaymentsQuery) (PaymentsResponse, error) {
var resp PaymentsResponse var resp PaymentsResponse
@@ -1366,7 +1290,7 @@ func fetchPaymentWithSequenceNumber(tx kvdb.RTx, paymentHash lntypes.Hash,
// the payment we are actually looking for. // the payment we are actually looking for.
seqBytes := bucket.Get(paymentSequenceKey) seqBytes := bucket.Get(paymentSequenceKey)
if seqBytes == nil { if seqBytes == nil {
return nil, ErrNoSequenceNumber return nil, paymentsdb.ErrNoSequenceNumber
} }
// If this top level payment has the sequence number we are looking for, // If this top level payment has the sequence number we are looking for,
@@ -1381,7 +1305,7 @@ func fetchPaymentWithSequenceNumber(tx kvdb.RTx, paymentHash lntypes.Hash,
// find a duplicate payments bucket here, something is wrong. // find a duplicate payments bucket here, something is wrong.
dup := bucket.NestedReadBucket(duplicatePaymentsBucket) dup := bucket.NestedReadBucket(duplicatePaymentsBucket)
if dup == nil { if dup == nil {
return nil, ErrNoDuplicateBucket return nil, paymentsdb.ErrNoDuplicateBucket
} }
var duplicatePayment *MPPayment var duplicatePayment *MPPayment
@@ -1389,7 +1313,7 @@ func fetchPaymentWithSequenceNumber(tx kvdb.RTx, paymentHash lntypes.Hash,
subBucket := dup.NestedReadBucket(k) subBucket := dup.NestedReadBucket(k)
if subBucket == nil { if subBucket == nil {
// We one bucket for each duplicate to be found. // We one bucket for each duplicate to be found.
return ErrNoDuplicateNestedBucket return paymentsdb.ErrNoDuplicateNestedBucket
} }
seqBytes := subBucket.Get(duplicatePaymentSequenceKey) seqBytes := subBucket.Get(duplicatePaymentSequenceKey)
@@ -1418,7 +1342,7 @@ func fetchPaymentWithSequenceNumber(tx kvdb.RTx, paymentHash lntypes.Hash,
// failed to find the payment with this sequence number; something is // failed to find the payment with this sequence number; something is
// wrong. // wrong.
if duplicatePayment == nil { if duplicatePayment == nil {
return nil, ErrDuplicateNotFound return nil, paymentsdb.ErrDuplicateNotFound
} }
return duplicatePayment, nil return duplicatePayment, nil
+56 -49
View File
@@ -16,6 +16,7 @@ import (
"github.com/lightningnetwork/lnd/kvdb" "github.com/lightningnetwork/lnd/kvdb"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"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/lightningnetwork/lnd/tlv"
@@ -66,7 +67,8 @@ func TestKVPaymentsDBSwitchFail(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
info, attempt, preimg, err := genInfo(t) info, attempt, preimg, err := genInfo(t)
require.NoError(t, err, "unable to generate htlc message") require.NoError(t, err, "unable to generate htlc message")
@@ -193,7 +195,7 @@ func TestKVPaymentsDBSwitchFail(t *testing.T) {
// Attempt a final payment, which should now fail since the prior // Attempt a final payment, which should now fail since the prior
// payment succeed. // payment succeed.
err = paymentDB.InitPayment(info.PaymentIdentifier, info) err = paymentDB.InitPayment(info.PaymentIdentifier, info)
if !errors.Is(err, ErrAlreadyPaid) { if !errors.Is(err, paymentsdb.ErrAlreadyPaid) {
t.Fatalf("unable to send htlc message: %v", err) t.Fatalf("unable to send htlc message: %v", err)
} }
} }
@@ -206,7 +208,8 @@ func TestKVPaymentsDBSwitchDoubleSend(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
info, attempt, preimg, err := genInfo(t) info, attempt, preimg, err := genInfo(t)
require.NoError(t, err, "unable to generate htlc message") require.NoError(t, err, "unable to generate htlc message")
@@ -228,7 +231,7 @@ func TestKVPaymentsDBSwitchDoubleSend(t *testing.T) {
// payment hash, should result in error indicating that payment has // payment hash, should result in error indicating that payment has
// already been sent. // already been sent.
err = paymentDB.InitPayment(info.PaymentIdentifier, info) err = paymentDB.InitPayment(info.PaymentIdentifier, info)
require.ErrorIs(t, err, ErrPaymentExists) require.ErrorIs(t, err, paymentsdb.ErrPaymentExists)
// Record an attempt. // Record an attempt.
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt)
@@ -246,7 +249,7 @@ func TestKVPaymentsDBSwitchDoubleSend(t *testing.T) {
// Sends base htlc message which initiate StatusInFlight. // Sends base htlc message which initiate StatusInFlight.
err = paymentDB.InitPayment(info.PaymentIdentifier, info) err = paymentDB.InitPayment(info.PaymentIdentifier, info)
if !errors.Is(err, ErrPaymentInFlight) { if !errors.Is(err, paymentsdb.ErrPaymentInFlight) {
t.Fatalf("payment control wrong behaviour: " + t.Fatalf("payment control wrong behaviour: " +
"double sending must trigger ErrPaymentInFlight error") "double sending must trigger ErrPaymentInFlight error")
} }
@@ -269,7 +272,7 @@ func TestKVPaymentsDBSwitchDoubleSend(t *testing.T) {
) )
err = paymentDB.InitPayment(info.PaymentIdentifier, info) err = paymentDB.InitPayment(info.PaymentIdentifier, info)
if !errors.Is(err, ErrAlreadyPaid) { if !errors.Is(err, paymentsdb.ErrAlreadyPaid) {
t.Fatalf("unable to send htlc message: %v", err) t.Fatalf("unable to send htlc message: %v", err)
} }
} }
@@ -282,7 +285,8 @@ func TestKVPaymentsDBSuccessesWithoutInFlight(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
info, _, preimg, err := genInfo(t) info, _, preimg, err := genInfo(t)
require.NoError(t, err, "unable to generate htlc message") require.NoError(t, err, "unable to generate htlc message")
@@ -294,9 +298,7 @@ func TestKVPaymentsDBSuccessesWithoutInFlight(t *testing.T) {
Preimage: preimg, Preimage: preimg,
}, },
) )
if err != ErrPaymentNotInitiated { require.ErrorIs(t, err, paymentsdb.ErrPaymentNotInitiated)
t.Fatalf("expected ErrPaymentNotInitiated, got %v", err)
}
} }
// TestKVPaymentsDBFailsWithoutInFlight checks that a strict payment // TestKVPaymentsDBFailsWithoutInFlight checks that a strict payment
@@ -307,16 +309,15 @@ func TestKVPaymentsDBFailsWithoutInFlight(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
info, _, _, err := genInfo(t) info, _, _, err := genInfo(t)
require.NoError(t, err, "unable to generate htlc message") require.NoError(t, err, "unable to generate htlc message")
// Calling Fail should return an error. // Calling Fail should return an error.
_, err = paymentDB.Fail(info.PaymentIdentifier, FailureReasonNoRoute) _, err = paymentDB.Fail(info.PaymentIdentifier, FailureReasonNoRoute)
if err != ErrPaymentNotInitiated { require.ErrorIs(t, err, paymentsdb.ErrPaymentNotInitiated)
t.Fatalf("expected ErrPaymentNotInitiated, got %v", err)
}
} }
// TestKVPaymentsDBDeleteNonInFlight checks that calling DeletePayments only // TestKVPaymentsDBDeleteNonInFlight checks that calling DeletePayments only
@@ -332,7 +333,8 @@ func TestKVPaymentsDBDeleteNonInFlight(t *testing.T) {
// start at 1, so 9999 is a safe bet for this test. // start at 1, so 9999 is a safe bet for this test.
var duplicateSeqNr = 9999 var duplicateSeqNr = 9999
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
payments := []struct { payments := []struct {
failed bool failed bool
@@ -550,7 +552,8 @@ func TestKVPaymentsDBDeletePayments(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
// Register three payments: // Register three payments:
// 1. A payment with two failed attempts. // 1. A payment with two failed attempts.
@@ -611,7 +614,8 @@ func TestKVPaymentsDBDeleteSinglePayment(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
// Register four payments: // Register four payments:
// All payments will have one failed HTLC attempt and one HTLC attempt // All payments will have one failed HTLC attempt and one HTLC attempt
@@ -717,7 +721,8 @@ func TestKVPaymentsDBMultiShard(t *testing.T) {
t.Fatalf("unable to init db: %v", err) t.Fatalf("unable to init db: %v", err)
} }
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
info, attempt, preimg, err := genInfo(t) info, attempt, preimg, err := genInfo(t)
if err != nil { if err != nil {
@@ -780,7 +785,7 @@ func TestKVPaymentsDBMultiShard(t *testing.T) {
b := *attempt b := *attempt
b.AttemptID = 3 b.AttemptID = 3
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b)
require.ErrorIs(t, err, ErrValueExceedsAmt) require.ErrorIs(t, err, paymentsdb.ErrValueExceedsAmt)
// Fail the second attempt. // Fail the second attempt.
a := attempts[1] a := attempts[1]
@@ -880,9 +885,13 @@ func TestKVPaymentsDBMultiShard(t *testing.T) {
b.AttemptID = 3 b.AttemptID = 3
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b)
if test.settleFirst { if test.settleFirst {
require.ErrorIs(t, err, ErrPaymentPendingSettled) require.ErrorIs(
t, err, paymentsdb.ErrPaymentPendingSettled,
)
} else { } else {
require.ErrorIs(t, err, ErrPaymentPendingFailed) require.ErrorIs(
t, err, paymentsdb.ErrPaymentPendingFailed,
)
} }
assertPaymentStatus( assertPaymentStatus(
@@ -951,21 +960,21 @@ func TestKVPaymentsDBMultiShard(t *testing.T) {
// settled. // settled.
case test.settleFirst && !test.settleLast: case test.settleFirst && !test.settleLast:
finalStatus = StatusSucceeded finalStatus = StatusSucceeded
registerErr = ErrPaymentAlreadySucceeded registerErr = paymentsdb.ErrPaymentAlreadySucceeded
case !test.settleFirst && test.settleLast: case !test.settleFirst && test.settleLast:
finalStatus = StatusSucceeded finalStatus = StatusSucceeded
registerErr = ErrPaymentAlreadySucceeded registerErr = paymentsdb.ErrPaymentAlreadySucceeded
// If both failed, we end up in a failed status. // If both failed, we end up in a failed status.
case !test.settleFirst && !test.settleLast: case !test.settleFirst && !test.settleLast:
finalStatus = StatusFailed finalStatus = StatusFailed
registerErr = ErrPaymentAlreadyFailed registerErr = paymentsdb.ErrPaymentAlreadyFailed
// Otherwise, the payment has a succeed status. // Otherwise, the payment has a succeed status.
case test.settleFirst && test.settleLast: case test.settleFirst && test.settleLast:
finalStatus = StatusSucceeded finalStatus = StatusSucceeded
registerErr = ErrPaymentAlreadySucceeded registerErr = paymentsdb.ErrPaymentAlreadySucceeded
} }
assertPaymentStatus( assertPaymentStatus(
@@ -994,7 +1003,8 @@ func TestKVPaymentsDBMPPRecordValidation(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
info, attempt, _, err := genInfo(t) info, attempt, _, err := genInfo(t)
require.NoError(t, err, "unable to generate htlc message") require.NoError(t, err, "unable to generate htlc message")
@@ -1021,27 +1031,21 @@ func TestKVPaymentsDBMPPRecordValidation(t *testing.T) {
b.AttemptID = 1 b.AttemptID = 1
b.Route.FinalHop().MPP = nil b.Route.FinalHop().MPP = nil
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b)
if err != ErrMPPayment { require.ErrorIs(t, err, paymentsdb.ErrMPPayment)
t.Fatalf("expected ErrMPPayment, got: %v", err)
}
// Try to register attempt one with a different payment address. // Try to register attempt one with a different payment address.
b.Route.FinalHop().MPP = record.NewMPP( b.Route.FinalHop().MPP = record.NewMPP(
info.Value, [32]byte{2}, info.Value, [32]byte{2},
) )
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b)
if err != ErrMPPPaymentAddrMismatch { require.ErrorIs(t, err, paymentsdb.ErrMPPPaymentAddrMismatch)
t.Fatalf("expected ErrMPPPaymentAddrMismatch, got: %v", err)
}
// Try registering one with a different total amount. // Try registering one with a different total amount.
b.Route.FinalHop().MPP = record.NewMPP( b.Route.FinalHop().MPP = record.NewMPP(
info.Value/2, [32]byte{1}, info.Value/2, [32]byte{1},
) )
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b)
if err != ErrMPPTotalAmountMismatch { require.ErrorIs(t, err, paymentsdb.ErrMPPTotalAmountMismatch)
t.Fatalf("expected ErrMPPTotalAmountMismatch, got: %v", err)
}
// Create and init a new payment. This time we'll check that we cannot // Create and init a new payment. This time we'll check that we cannot
// register an MPP attempt if we already registered a non-MPP one. // register an MPP attempt if we already registered a non-MPP one.
@@ -1063,9 +1067,7 @@ func TestKVPaymentsDBMPPRecordValidation(t *testing.T) {
) )
_, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b) _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, &b)
if err != ErrNonMPPayment { require.ErrorIs(t, err, paymentsdb.ErrNonMPPayment)
t.Fatalf("expected ErrNonMPPayment, got: %v", err)
}
} }
// TestDeleteFailedAttempts checks that DeleteFailedAttempts properly removes // TestDeleteFailedAttempts checks that DeleteFailedAttempts properly removes
@@ -1083,11 +1085,15 @@ func TestDeleteFailedAttempts(t *testing.T) {
func testDeleteFailedAttempts(t *testing.T, keepFailedPaymentAttempts bool) { func testDeleteFailedAttempts(t *testing.T, keepFailedPaymentAttempts bool) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err, "unable to init db") require.NoError(t, err, "unable to init db")
db.keepFailedPaymentAttempts = keepFailedPaymentAttempts
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(
db,
paymentsdb.WithKeepFailedPaymentAttempts(
keepFailedPaymentAttempts,
),
)
require.NoError(t, err)
// Register three payments: // Register three payments:
// All payments will have one failed HTLC attempt and one HTLC attempt // All payments will have one failed HTLC attempt and one HTLC attempt
@@ -1171,7 +1177,7 @@ func assertPaymentStatus(t *testing.T, p *KVPaymentsDB,
t.Helper() t.Helper()
payment, err := p.FetchPayment(hash) payment, err := p.FetchPayment(hash)
if errors.Is(err, ErrPaymentNotInitiated) { if errors.Is(err, paymentsdb.ErrPaymentNotInitiated) {
return return
} }
if err != nil { if err != nil {
@@ -1271,7 +1277,7 @@ func fetchPaymentIndexEntry(_ *testing.T, p *KVPaymentsDB,
indexValue := indexBucket.Get(key) indexValue := indexBucket.Get(key)
if indexValue == nil { if indexValue == nil {
return errNoSequenceNrIndex return paymentsdb.ErrNoSequenceNrIndex
} }
r := bytes.NewReader(indexValue) r := bytes.NewReader(indexValue)
@@ -1307,7 +1313,7 @@ func assertPaymentIndex(t *testing.T, p *KVPaymentsDB,
// exist. // exist.
func assertNoIndex(t *testing.T, p *KVPaymentsDB, seqNr uint64) { func assertNoIndex(t *testing.T, p *KVPaymentsDB, seqNr uint64) {
_, err := fetchPaymentIndexEntry(t, p, seqNr) _, err := fetchPaymentIndexEntry(t, p, seqNr)
require.Equal(t, errNoSequenceNrIndex, err) require.Equal(t, paymentsdb.ErrNoSequenceNrIndex, err)
} }
// payment is a helper structure that holds basic information on a test payment, // payment is a helper structure that holds basic information on a test payment,
@@ -1584,7 +1590,8 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) {
db, err := MakeTestDB(t) db, err := MakeTestDB(t)
require.NoError(t, err) require.NoError(t, err)
paymentDB := NewKVPaymentsDB(db) paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
// Generate a test payment which does not have duplicates. // Generate a test payment which does not have duplicates.
noDuplicates, _, _, err := genInfo(t) noDuplicates, _, _, err := genInfo(t)
@@ -1669,14 +1676,14 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) {
name: "lookup non-existent duplicate", name: "lookup non-existent duplicate",
paymentHash: hasDuplicates.PaymentIdentifier, paymentHash: hasDuplicates.PaymentIdentifier,
sequenceNumber: 999999, sequenceNumber: 999999,
expectedErr: ErrDuplicateNotFound, expectedErr: paymentsdb.ErrDuplicateNotFound,
}, },
{ {
name: "lookup duplicate, no duplicates " + name: "lookup duplicate, no duplicates " +
"bucket", "bucket",
paymentHash: noDuplicates.PaymentIdentifier, paymentHash: noDuplicates.PaymentIdentifier,
sequenceNumber: duplicateTwoSeqNr, sequenceNumber: duplicateTwoSeqNr,
expectedErr: ErrNoDuplicateBucket, expectedErr: paymentsdb.ErrNoDuplicateBucket,
}, },
} }
@@ -1710,8 +1717,8 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) {
// //
// This code is *only* intended to replicate legacy duplicate payments in lnd, // This code is *only* intended to replicate legacy duplicate payments in lnd,
// our current schema does not allow duplicates. // our current schema does not allow duplicates.
func appendDuplicatePayment(t *testing.T, db *DB, paymentHash lntypes.Hash, func appendDuplicatePayment(t *testing.T, db kvdb.Backend,
seqNr uint64, preImg lntypes.Preimage) { paymentHash lntypes.Hash, seqNr uint64, preImg lntypes.Preimage) {
err := kvdb.Update(db, func(tx walletdb.ReadWriteTx) error { err := kvdb.Update(db, func(tx walletdb.ReadWriteTx) error {
bucket, err := fetchPaymentBucketUpdate( bucket, err := fetchPaymentBucketUpdate(
+17 -14
View File
@@ -1,6 +1,7 @@
package channeldb package channeldb
import ( import (
"context"
"fmt" "fmt"
"math" "math"
"reflect" "reflect"
@@ -348,20 +349,21 @@ func TestQueryPayments(t *testing.T) {
} }
for _, tt := range tests { for _, tt := range tests {
tt := tt
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
t.Parallel() t.Parallel()
db, err := MakeTestDB(t) ctx := context.Background()
if err != nil {
t.Fatalf("unable to init db: %v", err)
}
paymentDB := NewKVPaymentsDB(db) db, err := MakeTestDB(t)
require.NoError(t, err)
// Initialize the payment database.
paymentDB, err := NewKVPaymentsDB(db)
require.NoError(t, err)
// 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 := paymentDB.QueryPayments(tt.query) resp, err := paymentDB.QueryPayments(ctx, tt.query)
require.NoError(t, err) require.NoError(t, err)
require.Len(t, resp.Payments, 0) require.Len(t, resp.Payments, 0)
@@ -389,10 +391,7 @@ func TestQueryPayments(t *testing.T) {
err = paymentDB.InitPayment( err = paymentDB.InitPayment(
info.PaymentIdentifier, info, info.PaymentIdentifier, info,
) )
if err != nil { require.NoError(t, err)
t.Fatalf("unable to initialize "+
"payment in database: %v", err)
}
// Immediately delete the payment with index 2. // Immediately delete the payment with index 2.
if i == 1 { if i == 1 {
@@ -401,8 +400,10 @@ func TestQueryPayments(t *testing.T) {
) )
require.NoError(t, err) require.NoError(t, err)
deletePayment(t, db, info.PaymentIdentifier, deletePayment(
pmt.SequenceNum) t, db, info.PaymentIdentifier,
pmt.SequenceNum,
)
} }
// If we are on the last payment entry, add a // If we are on the last payment entry, add a
@@ -437,7 +438,9 @@ func TestQueryPayments(t *testing.T) {
"want %v.", len(allPayments), 6) "want %v.", len(allPayments), 6)
} }
querySlice, err := paymentDB.QueryPayments(tt.query) querySlice, err := paymentDB.QueryPayments(
ctx, tt.query,
)
if err != nil { if err != nil {
t.Fatalf("unexpected error: %v", err) t.Fatalf("unexpected error: %v", err)
} }
+17 -4
View File
@@ -48,6 +48,7 @@ import (
"github.com/lightningnetwork/lnd/lnwallet/rpcwallet" "github.com/lightningnetwork/lnd/lnwallet/rpcwallet"
"github.com/lightningnetwork/lnd/macaroons" "github.com/lightningnetwork/lnd/macaroons"
"github.com/lightningnetwork/lnd/msgmux" "github.com/lightningnetwork/lnd/msgmux"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/rpcperms" "github.com/lightningnetwork/lnd/rpcperms"
"github.com/lightningnetwork/lnd/signal" "github.com/lightningnetwork/lnd/signal"
"github.com/lightningnetwork/lnd/sqldb" "github.com/lightningnetwork/lnd/sqldb"
@@ -1052,9 +1053,6 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
dbOptions := []channeldb.OptionModifier{ dbOptions := []channeldb.OptionModifier{
channeldb.OptionDryRunMigration(cfg.DryRunMigration), channeldb.OptionDryRunMigration(cfg.DryRunMigration),
channeldb.OptionKeepFailedPaymentAttempts(
cfg.KeepFailedPaymentAttempts,
),
channeldb.OptionStoreFinalHtlcResolutions( channeldb.OptionStoreFinalHtlcResolutions(
cfg.StoreFinalHtlcResolutions, cfg.StoreFinalHtlcResolutions,
), ),
@@ -1222,9 +1220,24 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
// Mount the payments DB which is only KV for now. // Mount the payments DB which is only KV for now.
// //
// TODO(ziggie): Add support for SQL payments DB. // TODO(ziggie): Add support for SQL payments DB.
kvPaymentsDB := channeldb.NewKVPaymentsDB( // Mount the payments DB for the KV store.
paymentsDBOptions := []paymentsdb.OptionModifier{
paymentsdb.WithKeepFailedPaymentAttempts(
cfg.KeepFailedPaymentAttempts,
),
}
kvPaymentsDB, err := channeldb.NewKVPaymentsDB(
dbs.ChanStateDB, dbs.ChanStateDB,
paymentsDBOptions...,
) )
if err != nil {
cleanUp()
err = fmt.Errorf("unable to open payments DB: %w", err)
d.logger.Error(err)
return nil, nil, err
}
dbs.KVPaymentsDB = kvPaymentsDB 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.
+8 -7
View File
@@ -22,6 +22,7 @@ import (
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/macaroons" "github.com/lightningnetwork/lnd/macaroons"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/routing" "github.com/lightningnetwork/lnd/routing"
"github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/routing/route"
"github.com/lightningnetwork/lnd/zpay32" "github.com/lightningnetwork/lnd/zpay32"
@@ -373,9 +374,9 @@ func (s *Server) SendPaymentV2(req *SendPaymentRequest,
payment.Identifier(), err) payment.Identifier(), err)
// Transform user errors to grpc code. // Transform user errors to grpc code.
if errors.Is(err, channeldb.ErrPaymentExists) || if errors.Is(err, paymentsdb.ErrPaymentExists) ||
errors.Is(err, channeldb.ErrPaymentInFlight) || errors.Is(err, paymentsdb.ErrPaymentInFlight) ||
errors.Is(err, channeldb.ErrAlreadyPaid) { errors.Is(err, paymentsdb.ErrAlreadyPaid) {
return status.Error( return status.Error(
codes.AlreadyExists, err.Error(), codes.AlreadyExists, err.Error(),
@@ -955,13 +956,13 @@ func (s *Server) SendToRouteV2(ctx context.Context,
// Transform user errors to grpc code. // Transform user errors to grpc code.
switch { switch {
case errors.Is(err, channeldb.ErrPaymentExists): case errors.Is(err, paymentsdb.ErrPaymentExists):
fallthrough fallthrough
case errors.Is(err, channeldb.ErrPaymentInFlight): case errors.Is(err, paymentsdb.ErrPaymentInFlight):
fallthrough fallthrough
case errors.Is(err, channeldb.ErrAlreadyPaid): case errors.Is(err, paymentsdb.ErrAlreadyPaid):
return nil, status.Error( return nil, status.Error(
codes.AlreadyExists, err.Error(), codes.AlreadyExists, err.Error(),
) )
@@ -1368,7 +1369,7 @@ func (s *Server) subscribePayment(identifier lntypes.Hash) (
sub, err := router.Tower.SubscribePayment(identifier) sub, err := router.Tower.SubscribePayment(identifier)
switch { switch {
case errors.Is(err, channeldb.ErrPaymentNotInitiated): case errors.Is(err, paymentsdb.ErrPaymentNotInitiated):
return nil, status.Error(codes.NotFound, err.Error()) return nil, status.Error(codes.NotFound, err.Error())
case err != nil: case err != nil:
+5
View File
@@ -46,6 +46,7 @@ import (
"github.com/lightningnetwork/lnd/monitoring" "github.com/lightningnetwork/lnd/monitoring"
"github.com/lightningnetwork/lnd/msgmux" "github.com/lightningnetwork/lnd/msgmux"
"github.com/lightningnetwork/lnd/netann" "github.com/lightningnetwork/lnd/netann"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/peer" "github.com/lightningnetwork/lnd/peer"
"github.com/lightningnetwork/lnd/peernotifier" "github.com/lightningnetwork/lnd/peernotifier"
"github.com/lightningnetwork/lnd/protofsm" "github.com/lightningnetwork/lnd/protofsm"
@@ -207,6 +208,10 @@ func SetupLoggers(root *build.SubLoggerManager, interceptor signal.Interceptor)
AddSubLogger(root, chainio.Subsystem, interceptor, chainio.UseLogger) AddSubLogger(root, chainio.Subsystem, interceptor, chainio.UseLogger)
AddSubLogger(root, msgmux.Subsystem, interceptor, msgmux.UseLogger) AddSubLogger(root, msgmux.Subsystem, interceptor, msgmux.UseLogger)
AddSubLogger(root, sqldb.Subsystem, interceptor, sqldb.UseLogger) AddSubLogger(root, sqldb.Subsystem, interceptor, sqldb.UseLogger)
AddSubLogger(
root, paymentsdb.Subsystem, interceptor, paymentsdb.UseLogger,
)
} }
// AddSubLogger is a helper method to conveniently create and register the // AddSubLogger is a helper method to conveniently create and register the
+133
View File
@@ -0,0 +1,133 @@
package paymentsdb
import "errors"
var (
// ErrAlreadyPaid signals we have already paid this payment hash.
ErrAlreadyPaid = errors.New("invoice is already paid")
// ErrPaymentInFlight signals that payment for this payment hash is
// already "in flight" on the network.
ErrPaymentInFlight = errors.New("payment is in transition")
// ErrPaymentExists is returned when we try to initialize an already
// existing payment that is not failed.
ErrPaymentExists = errors.New("payment already exists")
// ErrPaymentInternal is returned when performing the payment has a
// conflicting state, such as,
// - payment has StatusSucceeded but remaining amount is not zero.
// - payment has StatusInitiated but remaining amount is zero.
// - payment has StatusFailed but remaining amount is zero.
ErrPaymentInternal = errors.New("internal error")
// ErrPaymentNotInitiated is returned if the payment wasn't initiated.
ErrPaymentNotInitiated = errors.New("payment isn't initiated")
// ErrPaymentAlreadySucceeded is returned in the event we attempt to
// change the status of a payment already succeeded.
ErrPaymentAlreadySucceeded = errors.New("payment is already succeeded")
// ErrPaymentAlreadyFailed is returned in the event we attempt to alter
// a failed payment.
ErrPaymentAlreadyFailed = errors.New("payment has already failed")
// ErrUnknownPaymentStatus is returned when we do not recognize the
// existing state of a payment.
ErrUnknownPaymentStatus = errors.New("unknown payment status")
// ErrPaymentTerminal is returned if we attempt to alter a payment that
// already has reached a terminal condition.
ErrPaymentTerminal = errors.New("payment has reached terminal " +
"condition")
// ErrAttemptAlreadySettled is returned if we try to alter an already
// settled HTLC attempt.
ErrAttemptAlreadySettled = errors.New("attempt already settled")
// ErrAttemptAlreadyFailed is returned if we try to alter an already
// failed HTLC attempt.
ErrAttemptAlreadyFailed = errors.New("attempt already failed")
// ErrValueMismatch is returned if we try to register a non-MPP attempt
// with an amount that doesn't match the payment amount.
ErrValueMismatch = errors.New("attempted value doesn't match payment " +
"amount")
// ErrValueExceedsAmt is returned if we try to register an attempt that
// would take the total sent amount above the payment amount.
ErrValueExceedsAmt = errors.New("attempted value exceeds payment " +
"amount")
// ErrNonMPPayment is returned if we try to register an MPP attempt for
// a payment that already has a non-MPP attempt registered.
ErrNonMPPayment = errors.New("payment has non-MPP attempts")
// ErrMPPayment is returned if we try to register a non-MPP attempt for
// a payment that already has an MPP attempt registered.
ErrMPPayment = errors.New("payment has MPP attempts")
// ErrMPPRecordInBlindedPayment is returned if we try to register an
// attempt with an MPP record for a payment to a blinded path.
ErrMPPRecordInBlindedPayment = errors.New("blinded payment cannot " +
"contain MPP records")
// ErrBlindedPaymentTotalAmountMismatch is returned if we try to
// register an HTLC shard to a blinded route where the total amount
// doesn't match existing shards.
ErrBlindedPaymentTotalAmountMismatch = errors.New("blinded path " +
"total amount mismatch")
// ErrMPPPaymentAddrMismatch is returned if we try to register an MPP
// shard where the payment address doesn't match existing shards.
ErrMPPPaymentAddrMismatch = errors.New("payment address mismatch")
// ErrMPPTotalAmountMismatch is returned if we try to register an MPP
// shard where the total amount doesn't match existing shards.
ErrMPPTotalAmountMismatch = errors.New("mp payment total amount " +
"mismatch")
// ErrPaymentPendingSettled is returned when we try to add a new
// attempt to a payment that has at least one of its HTLCs settled.
ErrPaymentPendingSettled = errors.New("payment has settled htlcs")
// ErrPaymentPendingFailed is returned when we try to add a new attempt
// to a payment that already has a failure reason.
ErrPaymentPendingFailed = errors.New("payment has failure reason")
// ErrSentExceedsTotal is returned if the payment's current total sent
// amount exceed the total amount.
ErrSentExceedsTotal = errors.New("total sent exceeds total amount")
// ErrNoAttemptInfo is returned when no attempt info is stored yet.
ErrNoAttemptInfo = errors.New("unable to find attempt info for " +
"inflight payment")
)
// KV backend specific errors.
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")
// ErrNoSequenceNrIndex is returned when an attempt to lookup a payment
// index is made for a sequence number that is not indexed.
//
// NOTE: Only used for the kv backend.
ErrNoSequenceNrIndex = errors.New("payment sequence number index " +
"does not exist")
)
+34
View File
@@ -0,0 +1,34 @@
package paymentsdb
import (
"github.com/btcsuite/btclog/v2"
"github.com/lightningnetwork/lnd/build"
)
// log is a logger that is initialized with no output filters. This
// means the package will not perform any logging by default until the caller
// requests it.
//
//nolint:unused
var log btclog.Logger
// Subsystem defines the logging identifier for this subsystem.
const Subsystem = "PYDB"
// The default amount of logging is none.
func init() {
UseLogger(build.NewSubLogger(Subsystem, nil))
}
// DisableLog disables all library log output. Logging output is disabled
// by default until UseLogger is called.
func DisableLog() {
UseLogger(btclog.Disabled)
}
// UseLogger uses a specified Logger to output package logging info.
// This should be used in preference to SetLogWriter if the caller is also
// using btclog.
func UseLogger(logger btclog.Logger) {
log = logger
}
+30
View File
@@ -0,0 +1,30 @@
package paymentsdb
// StoreOptions holds parameters for the KVStore.
type StoreOptions struct {
// NoMigration allows to open the database in readonly mode
NoMigration bool
// KeepFailedPaymentAttempts is a flag that determines whether to keep
// failed payment attempts for a settled payment in the db.
KeepFailedPaymentAttempts bool
}
// OptionModifier is a function signature for modifying the default
// StoreOptions.
type OptionModifier func(*StoreOptions)
// WithKeepFailedPaymentAttempts sets the KeepFailedPaymentAttempts to n.
func WithKeepFailedPaymentAttempts(n bool) OptionModifier {
return func(o *StoreOptions) {
o.KeepFailedPaymentAttempts = n
}
}
// WithNoMigration allows the database to be opened in read only mode by
// disabling migrations.
func WithNoMigration(b bool) OptionModifier {
return func(o *StoreOptions) {
o.NoMigration = b
}
}
+52 -18
View File
@@ -13,6 +13,7 @@ import (
"github.com/davecgh/go-spew/spew" "github.com/davecgh/go-spew/spew"
"github.com/lightningnetwork/lnd/channeldb" "github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"github.com/lightningnetwork/lnd/routing/route" "github.com/lightningnetwork/lnd/routing/route"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -47,13 +48,19 @@ var (
func TestControlTowerSubscribeUnknown(t *testing.T) { func TestControlTowerSubscribeUnknown(t *testing.T) {
t.Parallel() t.Parallel()
db := initDB(t, false) db := initDB(t)
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db)) paymentDB, err := channeldb.NewKVPaymentsDB(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
// 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})
require.ErrorIs(t, err, channeldb.ErrPaymentNotInitiated) require.ErrorIs(t, err, paymentsdb.ErrPaymentNotInitiated)
} }
// TestControlTowerSubscribeSuccess tests that payment updates for a // TestControlTowerSubscribeSuccess tests that payment updates for a
@@ -61,9 +68,12 @@ func TestControlTowerSubscribeUnknown(t *testing.T) {
func TestControlTowerSubscribeSuccess(t *testing.T) { func TestControlTowerSubscribeSuccess(t *testing.T) {
t.Parallel() t.Parallel()
db := initDB(t, false) db := initDB(t)
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db)) paymentDB, err := channeldb.NewKVPaymentsDB(db)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
// Initiate a payment. // Initiate a payment.
info, attempt, preimg, err := genInfo() info, attempt, preimg, err := genInfo()
@@ -180,9 +190,15 @@ func TestKVPaymentsDBSubscribeFail(t *testing.T) {
func TestKVPaymentsDBSubscribeAllSuccess(t *testing.T) { func TestKVPaymentsDBSubscribeAllSuccess(t *testing.T) {
t.Parallel() t.Parallel()
db := initDB(t, true) db := initDB(t)
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db)) paymentDB, err := channeldb.NewKVPaymentsDB(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
// Initiate a payment. // Initiate a payment.
info1, attempt1, preimg1, err := genInfo() info1, attempt1, preimg1, err := genInfo()
@@ -293,9 +309,15 @@ func TestKVPaymentsDBSubscribeAllSuccess(t *testing.T) {
func TestKVPaymentsDBSubscribeAllImmediate(t *testing.T) { func TestKVPaymentsDBSubscribeAllImmediate(t *testing.T) {
t.Parallel() t.Parallel()
db := initDB(t, true) db := initDB(t)
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db)) paymentDB, err := channeldb.NewKVPaymentsDB(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
// Initiate a payment. // Initiate a payment.
info, attempt, _, err := genInfo() info, attempt, _, err := genInfo()
@@ -330,9 +352,15 @@ func TestKVPaymentsDBSubscribeAllImmediate(t *testing.T) {
func TestKVPaymentsDBUnsubscribeSuccess(t *testing.T) { func TestKVPaymentsDBUnsubscribeSuccess(t *testing.T) {
t.Parallel() t.Parallel()
db := initDB(t, true) db := initDB(t)
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db)) paymentDB, err := channeldb.NewKVPaymentsDB(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
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")
@@ -399,9 +427,17 @@ func TestKVPaymentsDBUnsubscribeSuccess(t *testing.T) {
func testKVPaymentsDBSubscribeFail(t *testing.T, registerAttempt, func testKVPaymentsDBSubscribeFail(t *testing.T, registerAttempt,
keepFailedPaymentAttempts bool) { keepFailedPaymentAttempts bool) {
db := initDB(t, keepFailedPaymentAttempts) db := initDB(t)
pControl := NewControlTower(channeldb.NewKVPaymentsDB(db)) paymentDB, err := channeldb.NewKVPaymentsDB(
db,
paymentsdb.WithKeepFailedPaymentAttempts(
keepFailedPaymentAttempts,
),
)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
// Initiate a payment. // Initiate a payment.
info, attempt, _, err := genInfo() info, attempt, _, err := genInfo()
@@ -517,11 +553,9 @@ func testKVPaymentsDBSubscribeFail(t *testing.T, registerAttempt,
} }
} }
func initDB(t *testing.T, keepFailedPaymentAttempts bool) *channeldb.DB { func initDB(t *testing.T) *channeldb.DB {
return channeldb.OpenForTesting( return channeldb.OpenForTesting(
t, t.TempDir(), channeldb.OptionKeepFailedPaymentAttempts( t, t.TempDir(),
keepFailedPaymentAttempts,
),
) )
} }
+17 -16
View File
@@ -13,6 +13,7 @@ import (
"github.com/lightningnetwork/lnd/htlcswitch" "github.com/lightningnetwork/lnd/htlcswitch"
"github.com/lightningnetwork/lnd/lntypes" "github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"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/routing/shards" "github.com/lightningnetwork/lnd/routing/shards"
@@ -308,7 +309,7 @@ func (m *mockControlTowerOld) InitPayment(phash lntypes.Hash,
// Don't allow re-init a successful payment. // Don't allow re-init a successful payment.
if _, ok := m.successful[phash]; ok { if _, ok := m.successful[phash]; ok {
return channeldb.ErrAlreadyPaid return paymentsdb.ErrAlreadyPaid
} }
_, failed := m.failed[phash] _, failed := m.failed[phash]
@@ -316,7 +317,7 @@ func (m *mockControlTowerOld) InitPayment(phash lntypes.Hash,
// If the payment is known, only allow re-init if failed. // If the payment is known, only allow re-init if failed.
if ok && !failed { if ok && !failed {
return channeldb.ErrPaymentInFlight return paymentsdb.ErrPaymentInFlight
} }
delete(m.failed, phash) delete(m.failed, phash)
@@ -330,7 +331,7 @@ func (m *mockControlTowerOld) InitPayment(phash lntypes.Hash,
func (m *mockControlTowerOld) DeleteFailedAttempts(phash lntypes.Hash) error { func (m *mockControlTowerOld) DeleteFailedAttempts(phash lntypes.Hash) error {
p, ok := m.payments[phash] p, ok := m.payments[phash]
if !ok { if !ok {
return channeldb.ErrPaymentNotInitiated return paymentsdb.ErrPaymentNotInitiated
} }
var inFlight bool var inFlight bool
@@ -347,7 +348,7 @@ func (m *mockControlTowerOld) DeleteFailedAttempts(phash lntypes.Hash) error {
} }
if inFlight { if inFlight {
return channeldb.ErrPaymentInFlight return paymentsdb.ErrPaymentInFlight
} }
return nil return nil
@@ -366,7 +367,7 @@ func (m *mockControlTowerOld) RegisterAttempt(phash lntypes.Hash,
// Lookup payment. // Lookup payment.
p, ok := m.payments[phash] p, ok := m.payments[phash]
if !ok { if !ok {
return channeldb.ErrPaymentNotInitiated return paymentsdb.ErrPaymentNotInitiated
} }
var inFlight bool var inFlight bool
@@ -387,15 +388,15 @@ func (m *mockControlTowerOld) RegisterAttempt(phash lntypes.Hash,
_, failed := m.failed[phash] _, failed := m.failed[phash]
if settled || failed { if settled || failed {
return channeldb.ErrPaymentTerminal return paymentsdb.ErrPaymentTerminal
} }
if settled && !inFlight { if settled && !inFlight {
return channeldb.ErrPaymentAlreadySucceeded return paymentsdb.ErrPaymentAlreadySucceeded
} }
if failed && !inFlight { if failed && !inFlight {
return channeldb.ErrPaymentAlreadyFailed return paymentsdb.ErrPaymentAlreadyFailed
} }
// Add attempt to payment. // Add attempt to payment.
@@ -421,7 +422,7 @@ func (m *mockControlTowerOld) SettleAttempt(phash lntypes.Hash,
// Only allow setting attempts if the payment is known. // Only allow setting attempts if the payment is known.
p, ok := m.payments[phash] p, ok := m.payments[phash]
if !ok { if !ok {
return nil, channeldb.ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
// Find the attempt with this pid, and set the settle info. // Find the attempt with this pid, and set the settle info.
@@ -431,10 +432,10 @@ func (m *mockControlTowerOld) SettleAttempt(phash lntypes.Hash,
} }
if a.Settle != nil { if a.Settle != nil {
return nil, channeldb.ErrAttemptAlreadySettled return nil, paymentsdb.ErrAttemptAlreadySettled
} }
if a.Failure != nil { if a.Failure != nil {
return nil, channeldb.ErrAttemptAlreadyFailed return nil, paymentsdb.ErrAttemptAlreadyFailed
} }
p.attempts[i].Settle = settleInfo p.attempts[i].Settle = settleInfo
@@ -462,7 +463,7 @@ func (m *mockControlTowerOld) FailAttempt(phash lntypes.Hash, pid uint64,
// Only allow failing attempts if the payment is known. // Only allow failing attempts if the payment is known.
p, ok := m.payments[phash] p, ok := m.payments[phash]
if !ok { if !ok {
return nil, channeldb.ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
// Find the attempt with this pid, and set the failure info. // Find the attempt with this pid, and set the failure info.
@@ -472,10 +473,10 @@ func (m *mockControlTowerOld) FailAttempt(phash lntypes.Hash, pid uint64,
} }
if a.Settle != nil { if a.Settle != nil {
return nil, channeldb.ErrAttemptAlreadySettled return nil, paymentsdb.ErrAttemptAlreadySettled
} }
if a.Failure != nil { if a.Failure != nil {
return nil, channeldb.ErrAttemptAlreadyFailed return nil, paymentsdb.ErrAttemptAlreadyFailed
} }
p.attempts[i].Failure = failInfo p.attempts[i].Failure = failInfo
@@ -499,7 +500,7 @@ func (m *mockControlTowerOld) FailPayment(phash lntypes.Hash,
// Payment must be known. // Payment must be known.
if _, ok := m.payments[phash]; !ok { if _, ok := m.payments[phash]; !ok {
return channeldb.ErrPaymentNotInitiated return paymentsdb.ErrPaymentNotInitiated
} }
m.failed[phash] = reason m.failed[phash] = reason
@@ -521,7 +522,7 @@ func (m *mockControlTowerOld) fetchPayment(phash lntypes.Hash) (
p, ok := m.payments[phash] p, ok := m.payments[phash]
if !ok { if !ok {
return nil, channeldb.ErrPaymentNotInitiated return nil, paymentsdb.ErrPaymentNotInitiated
} }
mp := &channeldb.MPPayment{ mp := &channeldb.MPPayment{
+3 -2
View File
@@ -24,6 +24,7 @@ import (
"github.com/lightningnetwork/lnd/lnutils" "github.com/lightningnetwork/lnd/lnutils"
"github.com/lightningnetwork/lnd/lnwallet" "github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
paymentsdb "github.com/lightningnetwork/lnd/payments/db"
"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/routing/shards" "github.com/lightningnetwork/lnd/routing/shards"
@@ -1133,8 +1134,8 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
switch { switch {
// If this is an MPP attempt and the hash is already registered with // If this is an MPP attempt and the hash is already registered with
// the database, we can go on to launch the shard. // the database, we can go on to launch the shard.
case mpp != nil && errors.Is(err, channeldb.ErrPaymentInFlight): case mpp != nil && errors.Is(err, paymentsdb.ErrPaymentInFlight):
case mpp != nil && errors.Is(err, channeldb.ErrPaymentExists): case mpp != nil && errors.Is(err, paymentsdb.ErrPaymentExists):
// Any other error is not tolerated. // Any other error is not tolerated.
case err != nil: case err != nil:
+3 -1
View File
@@ -7528,7 +7528,9 @@ func (r *rpcServer) ListPayments(ctx context.Context,
query.MaxPayments = math.MaxUint64 query.MaxPayments = math.MaxUint64
} }
paymentsQuerySlice, err := r.server.kvPaymentsDB.QueryPayments(query) paymentsQuerySlice, err := r.server.kvPaymentsDB.QueryPayments(
ctx, query,
)
if err != nil { if err != nil {
return nil, err return nil, err
} }