Merge pull request #10153 from ziggie1984/refactor-payments-code-05

Refactor Payment PR 5
This commit is contained in:
Yong authored and GitHub committed 2025-08-20 21:18:04 +08:00
commit 8f489d0f12
13 files changed
+1854 -1682

No files matched your search

+4 -4
View File
@@ -925,9 +925,9 @@ type DatabaseInstances struct {
// InvoiceDB is the database that stores information about invoices.
InvoiceDB invoices.InvoiceDB
// KVPaymentsDB is the database that stores all payment related
// PaymentsDB is the database that stores all payment related
// information.
KVPaymentsDB *paymentsdb.KVPaymentsDB
PaymentsDB paymentsdb.DB
// MacaroonDB is the database that stores macaroon root keys.
MacaroonDB kvdb.Backend
@@ -1225,7 +1225,7 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
cfg.KeepFailedPaymentAttempts,
),
}
kvPaymentsDB, err := paymentsdb.NewKVPaymentsDB(
kvPaymentsDB, err := paymentsdb.NewKVStore(
dbs.ChanStateDB,
paymentsDBOptions...,
)
@@ -1237,7 +1237,7 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
return nil, nil, err
}
dbs.KVPaymentsDB = kvPaymentsDB
dbs.PaymentsDB = kvPaymentsDB
// Wrap the watchtower client DB and make sure we clean up.
if cfg.WtClient.Active {
+123
View File
@@ -0,0 +1,123 @@
package paymentsdb
import (
"context"
"github.com/lightningnetwork/lnd/lntypes"
)
// DB represents the interface to the underlying payments database.
type DB interface {
PaymentReader
PaymentWriter
}
// PaymentReader represents the interface to read operations from the payments
// database.
type PaymentReader interface {
// QueryPayments queries the payments database and should support
// pagination.
QueryPayments(ctx context.Context, query Query) (Response, error)
// FetchPayment fetches the payment corresponding to the given payment
// hash.
FetchPayment(paymentHash lntypes.Hash) (*MPPayment, error)
// FetchInFlightPayments returns all payments with status InFlight.
FetchInFlightPayments() ([]*MPPayment, error)
}
// PaymentWriter represents the interface to write operations to the payments
// database.
type PaymentWriter interface {
// DeletePayment deletes a payment from the DB given its payment hash.
DeletePayment(paymentHash lntypes.Hash, failedAttemptsOnly bool) error
// DeletePayments deletes all payments from the DB given the specified
// flags.
DeletePayments(failedOnly, failedAttemptsOnly bool) (int, error)
PaymentControl
}
// PaymentControl represents the interface to control the payment lifecycle and
// its database operations. This interface represents the control flow of how
// a payment should be handled in the database. They are not just writing
// operations but they inherently represent the flow of a payment. The methods
// are called in the following order.
//
// 1. InitPayment.
// 2. RegisterAttempt (a payment can have multiple attempts).
// 3. SettleAttempt or FailAttempt (attempts can also fail as long as the
// sending amount will be eventually settled).
// 4. Payment succeeds or "Fail" is called.
// 5. DeleteFailedAttempts is called which will delete all failed attempts
// for a payment to clean up the database.
type PaymentControl interface {
// InitPayment checks that no other payment with the same payment hash
// exists in the database before creating a new payment. However, it
// should allow the user making a subsequent payment if the payment is
// in a Failed state.
InitPayment(lntypes.Hash, *PaymentCreationInfo) error
// RegisterAttempt atomically records the provided HTLCAttemptInfo.
RegisterAttempt(lntypes.Hash, *HTLCAttemptInfo) (*MPPayment, error)
// SettleAttempt marks the given attempt settled with the preimage. If
// this is a multi shard payment, this might implicitly mean 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.
SettleAttempt(lntypes.Hash, uint64, *HTLCSettleInfo) (*MPPayment, error)
// FailAttempt marks the given payment attempt failed.
FailAttempt(lntypes.Hash, uint64, *HTLCFailInfo) (*MPPayment, error)
// Fail transitions a payment into the Failed state, and records
// the ultimate reason the payment failed. Note that this should only
// be called when all active attempts are already failed. After
// invoking this method, InitPayment should return nil on its next call
// for this payment hash, allowing the user to make a subsequent
// payment.
Fail(lntypes.Hash, FailureReason) (*MPPayment, error)
// DeleteFailedAttempts removes all failed HTLCs from the db. It should
// be called for a given payment whenever all inflight htlcs are
// completed, and the payment has reached a final terminal state.
DeleteFailedAttempts(lntypes.Hash) error
}
// DBMPPayment is an interface that represents the payment state during a
// payment lifecycle.
type DBMPPayment interface {
// GetState returns the current state of the payment.
GetState() *MPPaymentState
// Terminated returns true if the payment is in a final state.
Terminated() bool
// GetStatus returns the current status of the payment.
GetStatus() PaymentStatus
// NeedWaitAttempts specifies whether the payment needs to wait for the
// outcome of an attempt.
NeedWaitAttempts() (bool, error)
// GetHTLCs returns all HTLCs of this payment.
GetHTLCs() []HTLCAttempt
// InFlightHTLCs returns all HTLCs that are in flight.
InFlightHTLCs() []HTLCAttempt
// AllowMoreAttempts is used to decide whether we can safely attempt
// more HTLCs for a given payment state. Return an error if the payment
// is in an unexpected state.
AllowMoreAttempts() (bool, error)
// TerminalInfo returns the settled HTLC attempt or the payment's
// failure reason.
TerminalInfo() (*HTLCAttempt, *FailureReason)
}
+111 -22
View File
@@ -7,6 +7,7 @@ import (
"errors"
"fmt"
"io"
"math"
"sort"
"sync"
"time"
@@ -117,8 +118,8 @@ var (
paymentsIndexBucket = []byte("payments-index-bucket")
)
// KVPaymentsDB implements persistence for payments and payment attempts.
type KVPaymentsDB struct {
// KVStore implements persistence for payments and payment attempts.
type KVStore struct {
// Sequence management for the kv store.
seqMu sync.Mutex
currSeq uint64
@@ -139,9 +140,9 @@ func defaultKVStoreOptions() *StoreOptions {
}
}
// NewKVPaymentsDB creates a new KVStore for payments.
func NewKVPaymentsDB(db kvdb.Backend,
options ...OptionModifier) (*KVPaymentsDB, error) {
// NewKVStore creates a new KVStore for payments.
func NewKVStore(db kvdb.Backend,
options ...OptionModifier) (*KVStore, error) {
opts := defaultKVStoreOptions()
for _, applyOption := range options {
@@ -154,7 +155,7 @@ func NewKVPaymentsDB(db kvdb.Backend,
}
}
return &KVPaymentsDB{
return &KVStore{
db: db,
keepFailedPaymentAttempts: opts.KeepFailedPaymentAttempts,
}, nil
@@ -189,7 +190,7 @@ func initKVStore(db kvdb.Backend) error {
// making sure it does not already exist as an in-flight payment. When this
// method returns successfully, the payment is guaranteed to be in the InFlight
// state.
func (p *KVPaymentsDB) InitPayment(paymentHash lntypes.Hash,
func (p *KVStore) InitPayment(paymentHash lntypes.Hash,
info *PaymentCreationInfo) error {
// Obtain a new sequence number for this payment. This is used
@@ -292,8 +293,8 @@ func (p *KVPaymentsDB) InitPayment(paymentHash lntypes.Hash,
}
// DeleteFailedAttempts deletes all failed htlcs for a payment if configured
// by the KVPaymentsDB db.
func (p *KVPaymentsDB) DeleteFailedAttempts(hash lntypes.Hash) error {
// by the KVStore db.
func (p *KVStore) DeleteFailedAttempts(hash lntypes.Hash) error {
if !p.keepFailedPaymentAttempts {
const failedHtlcsOnly = true
err := p.DeletePayment(hash, failedHtlcsOnly)
@@ -360,7 +361,7 @@ func deserializePaymentIndex(r io.Reader) (lntypes.Hash, error) {
// RegisterAttempt atomically records the provided HTLCAttemptInfo to the
// DB.
func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
func (p *KVStore) RegisterAttempt(paymentHash lntypes.Hash,
attempt *HTLCAttemptInfo) (*MPPayment, error) {
// Serialize the information before opening the db transaction.
@@ -510,7 +511,7 @@ func (p *KVPaymentsDB) RegisterAttempt(paymentHash lntypes.Hash,
// After invoking this method, InitPayment should always return an error to
// prevent us from making duplicate payments to the same payment hash. The
// provided preimage is atomically saved to the DB for record keeping.
func (p *KVPaymentsDB) SettleAttempt(hash lntypes.Hash,
func (p *KVStore) SettleAttempt(hash lntypes.Hash,
attemptID uint64, settleInfo *HTLCSettleInfo) (*MPPayment, error) {
var b bytes.Buffer
@@ -523,7 +524,7 @@ func (p *KVPaymentsDB) SettleAttempt(hash lntypes.Hash,
}
// FailAttempt marks the given payment attempt failed.
func (p *KVPaymentsDB) FailAttempt(hash lntypes.Hash,
func (p *KVStore) FailAttempt(hash lntypes.Hash,
attemptID uint64, failInfo *HTLCFailInfo) (*MPPayment, error) {
var b bytes.Buffer
@@ -536,7 +537,7 @@ func (p *KVPaymentsDB) FailAttempt(hash lntypes.Hash,
}
// updateHtlcKey updates a database key for the specified htlc.
func (p *KVPaymentsDB) updateHtlcKey(paymentHash lntypes.Hash,
func (p *KVStore) updateHtlcKey(paymentHash lntypes.Hash,
attemptID uint64, key, value []byte) (*MPPayment, error) {
aid := make([]byte, 8)
@@ -608,7 +609,7 @@ func (p *KVPaymentsDB) updateHtlcKey(paymentHash lntypes.Hash,
// payment failed. After invoking this method, InitPayment should return nil on
// its next call for this payment hash, allowing the switch to make a
// subsequent payment.
func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
func (p *KVStore) Fail(paymentHash lntypes.Hash,
reason FailureReason) (*MPPayment, error) {
var (
@@ -632,7 +633,7 @@ func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
// We mark the payment as failed as long as it is known. This
// lets the last attempt to fail with a terminal write its
// failure to the KVPaymentsDB without synchronizing with
// failure to the KVStore without synchronizing with
// other attempts.
_, err = fetchPaymentStatus(bucket)
if errors.Is(err, ErrPaymentNotInitiated) {
@@ -665,7 +666,7 @@ func (p *KVPaymentsDB) Fail(paymentHash lntypes.Hash,
}
// FetchPayment returns information about a payment from the database.
func (p *KVPaymentsDB) FetchPayment(paymentHash lntypes.Hash) (
func (p *KVStore) FetchPayment(paymentHash lntypes.Hash) (
*MPPayment, error) {
var payment *MPPayment
@@ -760,7 +761,7 @@ func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
// nextPaymentSequence returns the next sequence number to store for a new
// payment.
func (p *KVPaymentsDB) nextPaymentSequence() ([]byte, error) {
func (p *KVStore) nextPaymentSequence() ([]byte, error) {
p.seqMu.Lock()
defer p.seqMu.Unlock()
@@ -821,7 +822,7 @@ func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
}
// FetchInFlightPayments returns all payments with status InFlight.
func (p *KVPaymentsDB) FetchInFlightPayments() ([]*MPPayment, error) {
func (p *KVStore) FetchInFlightPayments() ([]*MPPayment, error) {
var (
inFlights []*MPPayment
start = time.Now()
@@ -894,7 +895,7 @@ func htlcBucketKey(prefix, id []byte) []byte {
}
// FetchPayments returns all sent payments found in the DB.
func (p *KVPaymentsDB) FetchPayments() ([]*MPPayment, error) {
func (p *KVStore) FetchPayments() ([]*MPPayment, error) {
var payments []*MPPayment
err := kvdb.View(p.db, func(tx kvdb.RTx) error {
@@ -1134,7 +1135,7 @@ func fetchFailedHtlcKeys(bucket kvdb.RBucket) ([][]byte, error) {
// QueryPayments is a query to the payments database which is restricted
// to a subset of payments by the payments query, containing an offset
// index and a maximum number of returned payments.
func (p *KVPaymentsDB) QueryPayments(_ context.Context,
func (p *KVStore) QueryPayments(_ context.Context,
query Query) (Response, error) {
var resp Response
@@ -1355,7 +1356,7 @@ func fetchPaymentWithSequenceNumber(tx kvdb.RTx, paymentHash lntypes.Hash,
// DeletePayment deletes a payment from the DB given its payment hash. If
// failedHtlcsOnly is set, only failed HTLC attempts of the payment will be
// deleted.
func (p *KVPaymentsDB) DeletePayment(paymentHash lntypes.Hash,
func (p *KVStore) DeletePayment(paymentHash lntypes.Hash,
failedHtlcsOnly bool) error {
return kvdb.Update(p.db, func(tx kvdb.RwTx) error {
@@ -1452,7 +1453,7 @@ func (p *KVPaymentsDB) DeletePayment(paymentHash lntypes.Hash,
// failedHtlcsOnly is set, the payment itself won't be deleted, only failed HTLC
// attempts. The method returns the number of deleted payments, which is always
// 0 if failedHtlcsOnly is set.
func (p *KVPaymentsDB) DeletePayments(failedOnly,
func (p *KVStore) DeletePayments(failedOnly,
failedHtlcsOnly bool) (int, error) {
var numPayments int
@@ -2107,3 +2108,91 @@ func DeserializeRoute(r io.Reader) (route.Route, error) {
return rt, nil
}
// serializeHTLCSettleInfo serializes the details of a settled htlc.
func serializeHTLCSettleInfo(w io.Writer, s *HTLCSettleInfo) error {
if _, err := w.Write(s.Preimage[:]); err != nil {
return err
}
if err := serializeTime(w, s.SettleTime); err != nil {
return err
}
return nil
}
// deserializeHTLCSettleInfo deserializes the details of a settled htlc.
func deserializeHTLCSettleInfo(r io.Reader) (*HTLCSettleInfo, error) {
s := &HTLCSettleInfo{}
if _, err := io.ReadFull(r, s.Preimage[:]); err != nil {
return nil, err
}
var err error
s.SettleTime, err = deserializeTime(r)
if err != nil {
return nil, err
}
return s, nil
}
// serializeHTLCFailInfo serializes the details of a failed htlc including the
// wire failure.
func serializeHTLCFailInfo(w io.Writer, f *HTLCFailInfo) error {
if err := serializeTime(w, f.FailTime); err != nil {
return err
}
// Write failure. If there is no failure message, write an empty
// byte slice.
var messageBytes bytes.Buffer
if f.Message != nil {
err := lnwire.EncodeFailureMessage(&messageBytes, f.Message, 0)
if err != nil {
return err
}
}
if err := wire.WriteVarBytes(w, 0, messageBytes.Bytes()); err != nil {
return err
}
return WriteElements(w, byte(f.Reason), f.FailureSourceIndex)
}
// deserializeHTLCFailInfo deserializes the details of a failed htlc including
// the wire failure.
func deserializeHTLCFailInfo(r io.Reader) (*HTLCFailInfo, error) {
f := &HTLCFailInfo{}
var err error
f.FailTime, err = deserializeTime(r)
if err != nil {
return nil, err
}
// Read failure.
failureBytes, err := wire.ReadVarBytes(
r, 0, math.MaxUint16, "failure",
)
if err != nil {
return nil, err
}
if len(failureBytes) > 0 {
f.Message, err = lnwire.DecodeFailureMessage(
bytes.NewReader(failureBytes), 0,
)
if err != nil {
return nil, err
}
}
var reason byte
err = ReadElements(r, &reason, &f.FailureSourceIndex)
if err != nil {
return nil, err
}
f.Reason = HTLCFailReason(reason)
return f, nil
}
+432 -1114
View File
File diff suppressed because it is too large. Load diff
-91
View File
@@ -4,12 +4,9 @@ import (
"bytes"
"errors"
"fmt"
"io"
"math"
"time"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/wire"
"github.com/davecgh/go-spew/spew"
sphinx "github.com/lightningnetwork/lightning-onion"
"github.com/lightningnetwork/lnd/lntypes"
@@ -650,94 +647,6 @@ func (m *MPPayment) AllowMoreAttempts() (bool, error) {
return true, nil
}
// serializeHTLCSettleInfo serializes the details of a settled htlc.
func serializeHTLCSettleInfo(w io.Writer, s *HTLCSettleInfo) error {
if _, err := w.Write(s.Preimage[:]); err != nil {
return err
}
if err := serializeTime(w, s.SettleTime); err != nil {
return err
}
return nil
}
// deserializeHTLCSettleInfo deserializes the details of a settled htlc.
func deserializeHTLCSettleInfo(r io.Reader) (*HTLCSettleInfo, error) {
s := &HTLCSettleInfo{}
if _, err := io.ReadFull(r, s.Preimage[:]); err != nil {
return nil, err
}
var err error
s.SettleTime, err = deserializeTime(r)
if err != nil {
return nil, err
}
return s, nil
}
// serializeHTLCFailInfo serializes the details of a failed htlc including the
// wire failure.
func serializeHTLCFailInfo(w io.Writer, f *HTLCFailInfo) error {
if err := serializeTime(w, f.FailTime); err != nil {
return err
}
// Write failure. If there is no failure message, write an empty
// byte slice.
var messageBytes bytes.Buffer
if f.Message != nil {
err := lnwire.EncodeFailureMessage(&messageBytes, f.Message, 0)
if err != nil {
return err
}
}
if err := wire.WriteVarBytes(w, 0, messageBytes.Bytes()); err != nil {
return err
}
return WriteElements(w, byte(f.Reason), f.FailureSourceIndex)
}
// deserializeHTLCFailInfo deserializes the details of a failed htlc including
// the wire failure.
func deserializeHTLCFailInfo(r io.Reader) (*HTLCFailInfo, error) {
f := &HTLCFailInfo{}
var err error
f.FailTime, err = deserializeTime(r)
if err != nil {
return nil, err
}
// Read failure.
failureBytes, err := wire.ReadVarBytes(
r, 0, math.MaxUint16, "failure",
)
if err != nil {
return nil, err
}
if len(failureBytes) > 0 {
f.Message, err = lnwire.DecodeFailureMessage(
bytes.NewReader(failureBytes), 0,
)
if err != nil {
return nil, err
}
}
var reason byte
err = ReadElements(r, &reason, &f.FailureSourceIndex)
if err != nil {
return nil, err
}
f.Reason = HTLCFailReason(reason)
return f, nil
}
// generateSphinxPacket generates then encodes a sphinx packet which encodes
// the onion route specified by the passed layer 3 route. The blob returned
// from this function can immediately be included within an HTLC add packet to
+1112 -376
View File
File diff suppressed because it is too large. Load diff
+19 -2
View File
@@ -8,7 +8,24 @@ import (
)
// NewTestDB is a helper function that creates an BBolt database for testing.
func NewTestDB(t *testing.T, opts ...OptionModifier) *KVPaymentsDB {
func NewTestDB(t *testing.T, opts ...OptionModifier) DB {
backend, backendCleanup, err := kvdb.GetTestBackend(
t.TempDir(), "paymentsDB",
)
require.NoError(t, err)
t.Cleanup(backendCleanup)
paymentDB, err := NewKVStore(backend, opts...)
require.NoError(t, err)
return paymentDB
}
// NewKVTestDB is a helper function that creates an BBolt database for testing
// and there is no need to convert the interface to the KVStore because for
// some unit tests we still need access to the kvdb interface.
func NewKVTestDB(t *testing.T, opts ...OptionModifier) *KVStore {
backend, backendCleanup, err := kvdb.GetTestBackend(
t.TempDir(), "kvPaymentDB",
)
@@ -16,7 +33,7 @@ func NewTestDB(t *testing.T, opts ...OptionModifier) *KVPaymentsDB {
t.Cleanup(backendCleanup)
paymentDB, err := NewKVPaymentsDB(backend, opts...)
paymentDB, err := NewKVStore(backend, opts...)
require.NoError(t, err)
return paymentDB
+16 -38
View File
@@ -9,46 +9,16 @@ import (
"github.com/lightningnetwork/lnd/queue"
)
// DBMPPayment is an interface derived from channeldb.MPPayment that is used by
// the payment lifecycle.
type DBMPPayment interface {
// GetState returns the current state of the payment.
GetState() *paymentsdb.MPPaymentState
// Terminated returns true if the payment is in a final state.
Terminated() bool
// GetStatus returns the current status of the payment.
GetStatus() paymentsdb.PaymentStatus
// NeedWaitAttempts specifies whether the payment needs to wait for the
// outcome of an attempt.
NeedWaitAttempts() (bool, error)
// GetHTLCs returns all HTLCs of this payment.
GetHTLCs() []paymentsdb.HTLCAttempt
// InFlightHTLCs returns all HTLCs that are in flight.
InFlightHTLCs() []paymentsdb.HTLCAttempt
// AllowMoreAttempts is used to decide whether we can safely attempt
// more HTLCs for a given payment state. Return an error if the payment
// is in an unexpected state.
AllowMoreAttempts() (bool, error)
// TerminalInfo returns the settled HTLC attempt or the payment's
// failure reason.
TerminalInfo() (*paymentsdb.HTLCAttempt, *paymentsdb.FailureReason)
}
// ControlTower tracks all outgoing payments made, whose primary purpose is to
// prevent duplicate payments to the same payment hash. In production, a
// persistent implementation is preferred so that tracking can survive across
// restarts. Payments are transitioned through various payment states, and the
// ControlTower interface provides access to driving the state transitions.
type ControlTower interface {
// This method checks that no succeeded payment exist for this payment
// hash.
// InitPayment initializes a new payment with the given payment hash and
// also notifies subscribers of the payment creation.
//
// NOTE: Subscribers should be notified by the new state of the payment.
InitPayment(lntypes.Hash, *paymentsdb.PaymentCreationInfo) error
// DeleteFailedAttempts removes all failed HTLCs from the db. It should
@@ -57,6 +27,8 @@ type ControlTower interface {
DeleteFailedAttempts(lntypes.Hash) error
// RegisterAttempt atomically records the provided HTLCAttemptInfo.
//
// NOTE: Subscribers should be notified by the new state of the payment.
RegisterAttempt(lntypes.Hash, *paymentsdb.HTLCAttemptInfo) error
// SettleAttempt marks the given attempt settled with the preimage. If
@@ -67,16 +39,20 @@ type ControlTower interface {
// 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.
//
// NOTE: Subscribers should be notified by the new state of the payment.
SettleAttempt(lntypes.Hash, uint64, *paymentsdb.HTLCSettleInfo) (
*paymentsdb.HTLCAttempt, error)
// FailAttempt marks the given payment attempt failed.
//
// NOTE: Subscribers should be notified by the new state of the payment.
FailAttempt(lntypes.Hash, uint64, *paymentsdb.HTLCFailInfo) (
*paymentsdb.HTLCAttempt, error)
// FetchPayment fetches the payment corresponding to the given payment
// hash.
FetchPayment(paymentHash lntypes.Hash) (DBMPPayment, error)
FetchPayment(paymentHash lntypes.Hash) (paymentsdb.DBMPPayment, error)
// FailPayment transitions a payment into the Failed state, and records
// the ultimate reason the payment failed. Note that this should only
@@ -84,6 +60,8 @@ type ControlTower interface {
// invoking this method, InitPayment should return nil on its next call
// for this payment hash, allowing the user to make a subsequent
// payment.
//
// NOTE: Subscribers should be notified by the new state of the payment.
FailPayment(lntypes.Hash, paymentsdb.FailureReason) error
// FetchInFlightPayments returns all payments with status InFlight.
@@ -151,7 +129,7 @@ func (s *controlTowerSubscriberImpl) Updates() <-chan interface{} {
// controlTower is persistent implementation of ControlTower to restrict
// double payment sending.
type controlTower struct {
db *paymentsdb.KVPaymentsDB
db paymentsdb.DB
// subscriberIndex is used to provide a unique id for each subscriber
// to all payments. This is used to easily remove the subscriber when
@@ -168,7 +146,7 @@ type controlTower struct {
}
// NewControlTower creates a new instance of the controlTower.
func NewControlTower(db *paymentsdb.KVPaymentsDB) ControlTower {
func NewControlTower(db paymentsdb.DB) ControlTower {
return &controlTower{
db: db,
subscribersAllPayments: make(
@@ -273,7 +251,7 @@ func (p *controlTower) FailAttempt(paymentHash lntypes.Hash,
// FetchPayment fetches the payment corresponding to the given payment hash.
func (p *controlTower) FetchPayment(paymentHash lntypes.Hash) (
DBMPPayment, error) {
paymentsdb.DBMPPayment, error) {
return p.db.FetchPayment(paymentHash)
}
+19 -19
View File
@@ -50,7 +50,7 @@ func TestControlTowerSubscribeUnknown(t *testing.T) {
db := initDB(t)
paymentDB, err := paymentsdb.NewKVPaymentsDB(
paymentDB, err := paymentsdb.NewKVStore(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
@@ -70,7 +70,7 @@ func TestControlTowerSubscribeSuccess(t *testing.T) {
db := initDB(t)
paymentDB, err := paymentsdb.NewKVPaymentsDB(db)
paymentDB, err := paymentsdb.NewKVStore(db)
require.NoError(t, err)
pControl := NewControlTower(paymentDB)
@@ -174,33 +174,33 @@ func TestControlTowerSubscribeSuccess(t *testing.T) {
}
}
// TestKVPaymentsDBSubscribeFail tests that payment updates for a
// TestKVStoreSubscribeFail tests that payment updates for a
// failed payment are properly sent to subscribers.
func TestKVPaymentsDBSubscribeFail(t *testing.T) {
func TestKVStoreSubscribeFail(t *testing.T) {
t.Parallel()
t.Run("register attempt, keep failed payments", func(t *testing.T) {
testKVPaymentsDBSubscribeFail(t, true, true)
testKVStoreSubscribeFail(t, true, true)
})
t.Run("register attempt, delete failed payments", func(t *testing.T) {
testKVPaymentsDBSubscribeFail(t, true, false)
testKVStoreSubscribeFail(t, true, false)
})
t.Run("no register attempt, keep failed payments", func(t *testing.T) {
testKVPaymentsDBSubscribeFail(t, false, true)
testKVStoreSubscribeFail(t, false, true)
})
t.Run("no register attempt, delete failed payments", func(t *testing.T) {
testKVPaymentsDBSubscribeFail(t, false, false)
testKVStoreSubscribeFail(t, false, false)
})
}
// TestKVPaymentsDBSubscribeAllSuccess tests that multiple payments are
// TestKVStoreSubscribeAllSuccess tests that multiple payments are
// properly sent to subscribers of TrackPayments.
func TestKVPaymentsDBSubscribeAllSuccess(t *testing.T) {
func TestKVStoreSubscribeAllSuccess(t *testing.T) {
t.Parallel()
db := initDB(t)
paymentDB, err := paymentsdb.NewKVPaymentsDB(
paymentDB, err := paymentsdb.NewKVStore(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
@@ -318,14 +318,14 @@ func TestKVPaymentsDBSubscribeAllSuccess(t *testing.T) {
require.Equal(t, attempt2.Route, htlc2.Route, "unexpected htlc route.")
}
// TestKVPaymentsDBSubscribeAllImmediate tests whether already inflight
// TestKVStoreSubscribeAllImmediate tests whether already inflight
// payments are reported at the start of the SubscribeAllPayments subscription.
func TestKVPaymentsDBSubscribeAllImmediate(t *testing.T) {
func TestKVStoreSubscribeAllImmediate(t *testing.T) {
t.Parallel()
db := initDB(t)
paymentDB, err := paymentsdb.NewKVPaymentsDB(
paymentDB, err := paymentsdb.NewKVStore(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
@@ -367,14 +367,14 @@ func TestKVPaymentsDBSubscribeAllImmediate(t *testing.T) {
}
}
// TestKVPaymentsDBUnsubscribeSuccess tests that when unsubscribed, there are
// TestKVStoreUnsubscribeSuccess tests that when unsubscribed, there are
// no more notifications to that specific subscription.
func TestKVPaymentsDBUnsubscribeSuccess(t *testing.T) {
func TestKVStoreUnsubscribeSuccess(t *testing.T) {
t.Parallel()
db := initDB(t)
paymentDB, err := paymentsdb.NewKVPaymentsDB(
paymentDB, err := paymentsdb.NewKVStore(
db,
paymentsdb.WithKeepFailedPaymentAttempts(true),
)
@@ -444,12 +444,12 @@ func TestKVPaymentsDBUnsubscribeSuccess(t *testing.T) {
require.Len(t, subscription2.Updates(), 0)
}
func testKVPaymentsDBSubscribeFail(t *testing.T, registerAttempt,
func testKVStoreSubscribeFail(t *testing.T, registerAttempt,
keepFailedPaymentAttempts bool) {
db := initDB(t)
paymentDB, err := paymentsdb.NewKVPaymentsDB(
paymentDB, err := paymentsdb.NewKVStore(
db,
paymentsdb.WithKeepFailedPaymentAttempts(
keepFailedPaymentAttempts,
+3 -3
View File
@@ -510,7 +510,7 @@ func (m *mockControlTowerOld) FailPayment(phash lntypes.Hash,
}
func (m *mockControlTowerOld) FetchPayment(phash lntypes.Hash) (
DBMPPayment, error) {
paymentsdb.DBMPPayment, error) {
m.Lock()
defer m.Unlock()
@@ -787,7 +787,7 @@ func (m *mockControlTower) FailPayment(phash lntypes.Hash,
}
func (m *mockControlTower) FetchPayment(phash lntypes.Hash) (
DBMPPayment, error) {
paymentsdb.DBMPPayment, error) {
args := m.Called(phash)
@@ -825,7 +825,7 @@ type mockMPPayment struct {
mock.Mock
}
var _ DBMPPayment = (*mockMPPayment)(nil)
var _ paymentsdb.DBMPPayment = (*mockMPPayment)(nil)
func (m *mockMPPayment) GetState() *paymentsdb.MPPaymentState {
args := m.Called()
+8 -4
View File
@@ -130,7 +130,7 @@ const (
// wait for results, the method will exit with `stepExit` such that the payment
// lifecycle loop will terminate.
func (p *paymentLifecycle) decideNextStep(
payment DBMPPayment) (stateStep, error) {
payment paymentsdb.DBMPPayment) (stateStep, error) {
// Check whether we could make new HTLC attempts.
allow, err := payment.AllowMoreAttempts()
@@ -326,7 +326,9 @@ lifecycle:
// terminal condition. We either return the settled preimage or the
// payment's failure reason.
//
// Optionally delete the failed attempts from the database.
// Optionally delete the failed attempts from the database. Depends on
// the database options deleting attempts is not allowed so this will
// just be a no-op.
err = p.router.cfg.Control.DeleteFailedAttempts(p.identifier)
if err != nil {
log.Errorf("Error deleting failed htlc attempts for payment "+
@@ -1110,7 +1112,9 @@ func (p *paymentLifecycle) patchLegacyPaymentHash(
// reloadInflightAttempts is called when the payment lifecycle is resumed after
// a restart. It reloads all inflight attempts from the control tower and
// collects the results of the attempts that have been sent before.
func (p *paymentLifecycle) reloadInflightAttempts() (DBMPPayment, error) {
func (p *paymentLifecycle) reloadInflightAttempts() (paymentsdb.DBMPPayment,
error) {
payment, err := p.router.cfg.Control.FetchPayment(p.identifier)
if err != nil {
return nil, err
@@ -1133,7 +1137,7 @@ func (p *paymentLifecycle) reloadInflightAttempts() (DBMPPayment, error) {
}
// reloadPayment returns the latest payment found in the db (control tower).
func (p *paymentLifecycle) reloadPayment() (DBMPPayment,
func (p *paymentLifecycle) reloadPayment() (paymentsdb.DBMPPayment,
*paymentsdb.MPPaymentState, error) {
// Read the db to get the latest state of the payment.
+3 -3
View File
@@ -7529,7 +7529,7 @@ func (r *rpcServer) ListPayments(ctx context.Context,
query.MaxPayments = math.MaxUint64
}
paymentsQuerySlice, err := r.server.kvPaymentsDB.QueryPayments(
paymentsQuerySlice, err := r.server.paymentsDB.QueryPayments(
ctx, query,
)
if err != nil {
@@ -7612,7 +7612,7 @@ func (r *rpcServer) DeletePayment(ctx context.Context,
rpcsLog.Infof("[DeletePayment] payment_identifier=%v, "+
"failed_htlcs_only=%v", hash, req.FailedHtlcsOnly)
err = r.server.kvPaymentsDB.DeletePayment(hash, req.FailedHtlcsOnly)
err = r.server.paymentsDB.DeletePayment(hash, req.FailedHtlcsOnly)
if err != nil {
return nil, err
}
@@ -7652,7 +7652,7 @@ func (r *rpcServer) DeleteAllPayments(ctx context.Context,
"failed_htlcs_only=%v", req.FailedPaymentsOnly,
req.FailedHtlcsOnly)
numDeletedPayments, err := r.server.kvPaymentsDB.DeletePayments(
numDeletedPayments, err := r.server.paymentsDB.DeletePayments(
req.FailedPaymentsOnly, req.FailedHtlcsOnly,
)
if err != nil {
+4 -6
View File
@@ -336,11 +336,9 @@ type server struct {
invoicesDB invoices.InvoiceDB
// kvPaymentsDB is the DB that contains all functions for managing
// paymentsDB is the DB that contains all functions for managing
// payments.
//
// TODO(ziggie): Replace with interface.
kvPaymentsDB *paymentsdb.KVPaymentsDB
paymentsDB paymentsdb.DB
aliasMgr *aliasmgr.Manager
@@ -683,7 +681,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
addrSource: addrSource,
miscDB: dbs.ChanStateDB,
invoicesDB: dbs.InvoiceDB,
kvPaymentsDB: dbs.KVPaymentsDB,
paymentsDB: dbs.PaymentsDB,
cc: cc,
sigPool: lnwallet.NewSigPool(cfg.Workers.Sig, cc.Signer),
writePool: writePool,
@@ -1007,7 +1005,7 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
PathFindingConfig: pathFindingConfig,
}
s.controlTower = routing.NewControlTower(dbs.KVPaymentsDB)
s.controlTower = routing.NewControlTower(dbs.PaymentsDB)
strictPruning := cfg.Bitcoin.Node == "neutrino" ||
cfg.Routing.StrictZombiePruning