multi: add new payments package

We add a new payments package and create a db interface for the
paymentDB there. This is already implemented for the KV db side
and will be implemented for the SQL side in future PRS.

Author: ziggie1984

config_builder.go

@@ -48,6 +48,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet/rpcwallet"
 	"github.com/lightningnetwork/lnd/macaroons"
 	"github.com/lightningnetwork/lnd/msgmux"
+	pymtpkg "github.com/lightningnetwork/lnd/payments"
 	"github.com/lightningnetwork/lnd/rpcperms"
 	"github.com/lightningnetwork/lnd/signal"
 	"github.com/lightningnetwork/lnd/sqldb"
@@ -924,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
+	// PaymentDB is the database that stores all payment related
 	// information.
-	KVPaymentsDB *channeldb.KVPaymentsDB
+	PaymentDB pymtpkg.PaymentDB
 
 	// MacaroonDB is the database that stores macaroon root keys.
 	MacaroonDB kvdb.Backend
@@ -1194,7 +1195,7 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 	kvPaymentsDB := channeldb.NewKVPaymentsDB(
 		dbs.ChanStateDB,
 	)
-	dbs.KVPaymentsDB = kvPaymentsDB
+	dbs.PaymentDB = kvPaymentsDB
 
 	// Wrap the watchtower client DB and make sure we clean up.
 	if cfg.WtClient.Active {

payments/interface.go

@@ -0,0 +1,71 @@
+package payments
+
+import (
+	"context"
+
+	"github.com/lightningnetwork/lnd/channeldb"
+	"github.com/lightningnetwork/lnd/lntypes"
+)
+
+// PaymentDB is the interface that represents the underlying payments database.
+type PaymentDB interface {
+	// QueryPayments queries the payments database and should support
+	// pagination.
+	QueryPayments(ctx context.Context,
+		query channeldb.PaymentsQuery) (channeldb.PaymentsResponse,
+		error)
+
+	// DeletePayment deletes a payment from the DB given its payment hash.
+	DeletePayment(paymentHash lntypes.Hash, failedHtlcsOnly bool) error
+
+	// DeletePayments deletes all payments from the DB given the specified
+	// flags.
+	//
+	// TODO(ziggie): use DeletePayment here as well to make the interface
+	// more leaner.
+	DeletePayments(failedOnly, failedHtlcsOnly bool) (int, error)
+
+	// This method checks that no succeeded payment exist for this payment
+	// hash.
+	InitPayment(lntypes.Hash, *channeldb.PaymentCreationInfo) 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 settled state.
+	DeleteFailedAttempts(lntypes.Hash) error
+
+	// RegisterAttempt atomically records the provided HTLCAttemptInfo.
+	RegisterAttempt(lntypes.Hash,
+		*channeldb.HTLCAttemptInfo) (*channeldb.MPPayment, error)
+
+	// SettleAttempt marks the given attempt settled with the preimage. If
+	// this is a multi shard payment, this might implicitly mean the 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, *channeldb.HTLCSettleInfo) (
+		*channeldb.MPPayment, error)
+
+	// FailAttempt marks the given payment attempt failed.
+	FailAttempt(lntypes.Hash, uint64, *channeldb.HTLCFailInfo) (
+		*channeldb.MPPayment, error)
+
+	// FetchPayment fetches the payment corresponding to the given payment
+	// hash.
+	FetchPayment(paymentHash lntypes.Hash) (*channeldb.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, channeldb.FailureReason) (*channeldb.MPPayment,
+		error)
+
+	// FetchInFlightPayments returns all payments with status InFlight.
+	FetchInFlightPayments() ([]*channeldb.MPPayment, error)
+}

routing/control_tower.go

@@ -6,6 +6,7 @@ import (
 	"github.com/lightningnetwork/lnd/channeldb"
 	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/multimutex"
+	pymtpkg "github.com/lightningnetwork/lnd/payments"
 	"github.com/lightningnetwork/lnd/queue"
 )
 
@@ -151,7 +152,7 @@ func (s *controlTowerSubscriberImpl) Updates() <-chan interface{} {
 // controlTower is persistent implementation of ControlTower to restrict
 // double payment sending.
 type controlTower struct {
-	db *channeldb.KVPaymentsDB
+	db pymtpkg.PaymentDB
 
 	// 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 +169,7 @@ type controlTower struct {
 }
 
 // NewControlTower creates a new instance of the controlTower.
-func NewControlTower(db *channeldb.KVPaymentsDB) ControlTower {
+func NewControlTower(db pymtpkg.PaymentDB) ControlTower {
 	return &controlTower{
 		db: db,
 		subscribersAllPayments: make(

rpcserver.go

@@ -7403,9 +7403,7 @@ func (r *rpcServer) ListPayments(ctx context.Context,
 		query.MaxPayments = math.MaxUint64
 	}
 
-	paymentsQuerySlice, err := r.server.kvPaymentsDB.QueryPayments(
-		ctx, query,
-	)
+	paymentsQuerySlice, err := r.server.paymentsDB.QueryPayments(ctx, query)
 	if err != nil {
 		return nil, err
 	}
@@ -7447,7 +7445,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
 	}
@@ -7487,7 +7485,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 {

server.go

@@ -64,6 +64,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/lightningnetwork/lnd/nat"
 	"github.com/lightningnetwork/lnd/netann"
+	pymtpkg "github.com/lightningnetwork/lnd/payments"
 	"github.com/lightningnetwork/lnd/peer"
 	"github.com/lightningnetwork/lnd/peernotifier"
 	"github.com/lightningnetwork/lnd/pool"
@@ -317,11 +318,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 *channeldb.KVPaymentsDB
+	paymentsDB pymtpkg.PaymentDB
 
 	aliasMgr *aliasmgr.Manager
 
@@ -664,7 +663,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		addrSource:     addrSource,
 		miscDB:         dbs.ChanStateDB,
 		invoicesDB:     dbs.InvoiceDB,
-		kvPaymentsDB:   dbs.KVPaymentsDB,
+		paymentsDB:     dbs.PaymentDB,
 		cc:             cc,
 		sigPool:        lnwallet.NewSigPool(cfg.Workers.Sig, cc.Signer),
 		writePool:      writePool,
@@ -1088,7 +1087,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		PathFindingConfig:   pathFindingConfig,
 	}
 
-	s.controlTower = routing.NewControlTower(dbs.KVPaymentsDB)
+	s.controlTower = routing.NewControlTower(dbs.PaymentDB)
 
 	strictPruning := cfg.Bitcoin.Node == "neutrino" ||
 		cfg.Routing.StrictZombiePruning