channeldb: rename payment_control files

We rename the file to payment_kv_store to highlight that this
is the kv implementation of the payment db backend.

channeldb/payments_kv_store.go

@@ -0,0 +1,836 @@
+package channeldb
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/lightningnetwork/lnd/kvdb"
+	"github.com/lightningnetwork/lnd/lntypes"
+)
+
+const (
+	// paymentSeqBlockSize is the block size used when we batch allocate
+	// payment sequences for future payments.
+	paymentSeqBlockSize = 1000
+
+	// paymentProgressLogInterval is the interval we use limiting the
+	// logging output of payment processing.
+	paymentProgressLogInterval = 30 * time.Second
+)
+
+var (
+	// ErrAlreadyPaid signals we have already paid this payment hash.
+	ErrAlreadyPaid = errors.New("invoice is already paid")
+
+	// ErrPaymentInFlight signals that payment for this payment hash is
+	// already "in flight" on the network.
+	ErrPaymentInFlight = errors.New("payment is in transition")
+
+	// ErrPaymentExists is returned when we try to initialize an already
+	// existing payment that is not failed.
+	ErrPaymentExists = errors.New("payment already exists")
+
+	// ErrPaymentInternal is returned when performing the payment has a
+	// conflicting state, such as,
+	// - payment has StatusSucceeded but remaining amount is not zero.
+	// - payment has StatusInitiated but remaining amount is zero.
+	// - payment has StatusFailed but remaining amount is zero.
+	ErrPaymentInternal = errors.New("internal error")
+
+	// ErrPaymentNotInitiated is returned if the payment wasn't initiated.
+	ErrPaymentNotInitiated = errors.New("payment isn't initiated")
+
+	// ErrPaymentAlreadySucceeded is returned in the event we attempt to
+	// change the status of a payment already succeeded.
+	ErrPaymentAlreadySucceeded = errors.New("payment is already succeeded")
+
+	// ErrPaymentAlreadyFailed is returned in the event we attempt to alter
+	// a failed payment.
+	ErrPaymentAlreadyFailed = errors.New("payment has already failed")
+
+	// ErrUnknownPaymentStatus is returned when we do not recognize the
+	// existing state of a payment.
+	ErrUnknownPaymentStatus = errors.New("unknown payment status")
+
+	// ErrPaymentTerminal is returned if we attempt to alter a payment that
+	// already has reached a terminal condition.
+	ErrPaymentTerminal = errors.New("payment has reached terminal " +
+		"condition")
+
+	// ErrAttemptAlreadySettled is returned if we try to alter an already
+	// settled HTLC attempt.
+	ErrAttemptAlreadySettled = errors.New("attempt already settled")
+
+	// ErrAttemptAlreadyFailed is returned if we try to alter an already
+	// failed HTLC attempt.
+	ErrAttemptAlreadyFailed = errors.New("attempt already failed")
+
+	// ErrValueMismatch is returned if we try to register a non-MPP attempt
+	// with an amount that doesn't match the payment amount.
+	ErrValueMismatch = errors.New("attempted value doesn't match payment " +
+		"amount")
+
+	// ErrValueExceedsAmt is returned if we try to register an attempt that
+	// would take the total sent amount above the payment amount.
+	ErrValueExceedsAmt = errors.New("attempted value exceeds payment " +
+		"amount")
+
+	// ErrNonMPPayment is returned if we try to register an MPP attempt for
+	// a payment that already has a non-MPP attempt registered.
+	ErrNonMPPayment = errors.New("payment has non-MPP attempts")
+
+	// ErrMPPayment is returned if we try to register a non-MPP attempt for
+	// a payment that already has an MPP attempt registered.
+	ErrMPPayment = errors.New("payment has MPP attempts")
+
+	// ErrMPPRecordInBlindedPayment is returned if we try to register an
+	// attempt with an MPP record for a payment to a blinded path.
+	ErrMPPRecordInBlindedPayment = errors.New("blinded payment cannot " +
+		"contain MPP records")
+
+	// ErrBlindedPaymentTotalAmountMismatch is returned if we try to
+	// register an HTLC shard to a blinded route where the total amount
+	// doesn't match existing shards.
+	ErrBlindedPaymentTotalAmountMismatch = errors.New("blinded path " +
+		"total amount mismatch")
+
+	// ErrMPPPaymentAddrMismatch is returned if we try to register an MPP
+	// shard where the payment address doesn't match existing shards.
+	ErrMPPPaymentAddrMismatch = errors.New("payment address mismatch")
+
+	// ErrMPPTotalAmountMismatch is returned if we try to register an MPP
+	// shard where the total amount doesn't match existing shards.
+	ErrMPPTotalAmountMismatch = errors.New("mp payment total amount " +
+		"mismatch")
+
+	// ErrPaymentPendingSettled is returned when we try to add a new
+	// attempt to a payment that has at least one of its HTLCs settled.
+	ErrPaymentPendingSettled = errors.New("payment has settled htlcs")
+
+	// ErrPaymentPendingFailed is returned when we try to add a new attempt
+	// to a payment that already has a failure reason.
+	ErrPaymentPendingFailed = errors.New("payment has failure reason")
+
+	// ErrSentExceedsTotal is returned if the payment's current total sent
+	// amount exceed the total amount.
+	ErrSentExceedsTotal = errors.New("total sent exceeds total amount")
+
+	// errNoAttemptInfo is returned when no attempt info is stored yet.
+	errNoAttemptInfo = errors.New("unable to find attempt info for " +
+		"inflight payment")
+
+	// errNoSequenceNrIndex is returned when an attempt to lookup a payment
+	// index is made for a sequence number that is not indexed.
+	errNoSequenceNrIndex = errors.New("payment sequence number index " +
+		"does not exist")
+)
+
+// PaymentControl implements persistence for payments and payment attempts.
+type PaymentControl struct {
+	paymentSeqMx     sync.Mutex
+	currPaymentSeq   uint64
+	storedPaymentSeq uint64
+	db               *DB
+}
+
+// NewPaymentControl creates a new instance of the PaymentControl.
+func NewPaymentControl(db *DB) *PaymentControl {
+	return &PaymentControl{
+		db: db,
+	}
+}
+
+// InitPayment checks or records the given PaymentCreationInfo with the DB,
+// making sure it does not already exist as an in-flight payment. When this
+// method returns successfully, the payment is guaranteed to be in the InFlight
+// state.
+func (p *PaymentControl) InitPayment(paymentHash lntypes.Hash,
+	info *PaymentCreationInfo) error {
+
+	// Obtain a new sequence number for this payment. This is used
+	// to sort the payments in order of creation, and also acts as
+	// a unique identifier for each payment.
+	sequenceNum, err := p.nextPaymentSequence()
+	if err != nil {
+		return err
+	}
+
+	var b bytes.Buffer
+	if err := serializePaymentCreationInfo(&b, info); err != nil {
+		return err
+	}
+	infoBytes := b.Bytes()
+
+	var updateErr error
+	err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
+		// Reset the update error, to avoid carrying over an error
+		// from a previous execution of the batched db transaction.
+		updateErr = nil
+
+		prefetchPayment(tx, paymentHash)
+		bucket, err := createPaymentBucket(tx, paymentHash)
+		if err != nil {
+			return err
+		}
+
+		// Get the existing status of this payment, if any.
+		paymentStatus, err := fetchPaymentStatus(bucket)
+
+		switch {
+		// If no error is returned, it means we already have this
+		// payment. We'll check the status to decide whether we allow
+		// retrying the payment or return a specific error.
+		case err == nil:
+			if err := paymentStatus.initializable(); err != nil {
+				updateErr = err
+				return nil
+			}
+
+		// Otherwise, if the error is not `ErrPaymentNotInitiated`,
+		// we'll return the error.
+		case !errors.Is(err, ErrPaymentNotInitiated):
+			return err
+		}
+
+		// Before we set our new sequence number, we check whether this
+		// payment has a previously set sequence number and remove its
+		// index entry if it exists. This happens in the case where we
+		// have a previously attempted payment which was left in a state
+		// where we can retry.
+		seqBytes := bucket.Get(paymentSequenceKey)
+		if seqBytes != nil {
+			indexBucket := tx.ReadWriteBucket(paymentsIndexBucket)
+			if err := indexBucket.Delete(seqBytes); err != nil {
+				return err
+			}
+		}
+
+		// Once we have obtained a sequence number, we add an entry
+		// to our index bucket which will map the sequence number to
+		// our payment identifier.
+		err = createPaymentIndexEntry(
+			tx, sequenceNum, info.PaymentIdentifier,
+		)
+		if err != nil {
+			return err
+		}
+
+		err = bucket.Put(paymentSequenceKey, sequenceNum)
+		if err != nil {
+			return err
+		}
+
+		// Add the payment info to the bucket, which contains the
+		// static information for this payment
+		err = bucket.Put(paymentCreationInfoKey, infoBytes)
+		if err != nil {
+			return err
+		}
+
+		// We'll delete any lingering HTLCs to start with, in case we
+		// are initializing a payment that was attempted earlier, but
+		// left in a state where we could retry.
+		err = bucket.DeleteNestedBucket(paymentHtlcsBucket)
+		if err != nil && err != kvdb.ErrBucketNotFound {
+			return err
+		}
+
+		// Also delete any lingering failure info now that we are
+		// re-attempting.
+		return bucket.Delete(paymentFailInfoKey)
+	})
+	if err != nil {
+		return fmt.Errorf("unable to init payment: %w", err)
+	}
+
+	return updateErr
+}
+
+// DeleteFailedAttempts deletes all failed htlcs for a payment if configured
+// by the PaymentControl db.
+func (p *PaymentControl) DeleteFailedAttempts(hash lntypes.Hash) error {
+	if !p.db.keepFailedPaymentAttempts {
+		const failedHtlcsOnly = true
+		err := p.db.DeletePayment(hash, failedHtlcsOnly)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// paymentIndexTypeHash is a payment index type which indicates that we have
+// created an index of payment sequence number to payment hash.
+type paymentIndexType uint8
+
+// paymentIndexTypeHash is a payment index type which indicates that we have
+// created an index of payment sequence number to payment hash.
+const paymentIndexTypeHash paymentIndexType = 0
+
+// createPaymentIndexEntry creates a payment hash typed index for a payment. The
+// index produced contains a payment index type (which can be used in future to
+// signal different payment index types) and the payment identifier.
+func createPaymentIndexEntry(tx kvdb.RwTx, sequenceNumber []byte,
+	id lntypes.Hash) error {
+
+	var b bytes.Buffer
+	if err := WriteElements(&b, paymentIndexTypeHash, id[:]); err != nil {
+		return err
+	}
+
+	indexes := tx.ReadWriteBucket(paymentsIndexBucket)
+	return indexes.Put(sequenceNumber, b.Bytes())
+}
+
+// deserializePaymentIndex deserializes a payment index entry. This function
+// currently only supports deserialization of payment hash indexes, and will
+// fail for other types.
+func deserializePaymentIndex(r io.Reader) (lntypes.Hash, error) {
+	var (
+		indexType   paymentIndexType
+		paymentHash []byte
+	)
+
+	if err := ReadElements(r, &indexType, &paymentHash); err != nil {
+		return lntypes.Hash{}, err
+	}
+
+	// While we only have on payment index type, we do not need to use our
+	// index type to deserialize the index. However, we sanity check that
+	// this type is as expected, since we had to read it out anyway.
+	if indexType != paymentIndexTypeHash {
+		return lntypes.Hash{}, fmt.Errorf("unknown payment index "+
+			"type: %v", indexType)
+	}
+
+	hash, err := lntypes.MakeHash(paymentHash)
+	if err != nil {
+		return lntypes.Hash{}, err
+	}
+
+	return hash, nil
+}
+
+// RegisterAttempt atomically records the provided HTLCAttemptInfo to the
+// DB.
+func (p *PaymentControl) RegisterAttempt(paymentHash lntypes.Hash,
+	attempt *HTLCAttemptInfo) (*MPPayment, error) {
+
+	// Serialize the information before opening the db transaction.
+	var a bytes.Buffer
+	err := serializeHTLCAttemptInfo(&a, attempt)
+	if err != nil {
+		return nil, err
+	}
+	htlcInfoBytes := a.Bytes()
+
+	htlcIDBytes := make([]byte, 8)
+	binary.BigEndian.PutUint64(htlcIDBytes, attempt.AttemptID)
+
+	var payment *MPPayment
+	err = kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
+		prefetchPayment(tx, paymentHash)
+		bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
+		if err != nil {
+			return err
+		}
+
+		payment, err = fetchPayment(bucket)
+		if err != nil {
+			return err
+		}
+
+		// Check if registering a new attempt is allowed.
+		if err := payment.Registrable(); err != nil {
+			return err
+		}
+
+		// If the final hop has encrypted data, then we know this is a
+		// blinded payment. In blinded payments, MPP records are not set
+		// for split payments and the recipient is responsible for using
+		// a consistent PathID across the various encrypted data
+		// payloads that we received from them for this payment. All we
+		// need to check is that the total amount field for each HTLC
+		// in the split payment is correct.
+		isBlinded := len(attempt.Route.FinalHop().EncryptedData) != 0
+
+		// Make sure any existing shards match the new one with regards
+		// to MPP options.
+		mpp := attempt.Route.FinalHop().MPP
+
+		// MPP records should not be set for attempts to blinded paths.
+		if isBlinded && mpp != nil {
+			return ErrMPPRecordInBlindedPayment
+		}
+
+		for _, h := range payment.InFlightHTLCs() {
+			hMpp := h.Route.FinalHop().MPP
+
+			// If this is a blinded payment, then no existing HTLCs
+			// should have MPP records.
+			if isBlinded && hMpp != nil {
+				return ErrMPPRecordInBlindedPayment
+			}
+
+			// If this is a blinded payment, then we just need to
+			// check that the TotalAmtMsat field for this shard
+			// is equal to that of any other shard in the same
+			// payment.
+			if isBlinded {
+				if attempt.Route.FinalHop().TotalAmtMsat !=
+					h.Route.FinalHop().TotalAmtMsat {
+
+					//nolint:ll
+					return ErrBlindedPaymentTotalAmountMismatch
+				}
+
+				continue
+			}
+
+			switch {
+			// We tried to register a non-MPP attempt for a MPP
+			// payment.
+			case mpp == nil && hMpp != nil:
+				return ErrMPPayment
+
+			// We tried to register a MPP shard for a non-MPP
+			// payment.
+			case mpp != nil && hMpp == nil:
+				return ErrNonMPPayment
+
+			// Non-MPP payment, nothing more to validate.
+			case mpp == nil:
+				continue
+			}
+
+			// Check that MPP options match.
+			if mpp.PaymentAddr() != hMpp.PaymentAddr() {
+				return ErrMPPPaymentAddrMismatch
+			}
+
+			if mpp.TotalMsat() != hMpp.TotalMsat() {
+				return ErrMPPTotalAmountMismatch
+			}
+		}
+
+		// If this is a non-MPP attempt, it must match the total amount
+		// exactly. Note that a blinded payment is considered an MPP
+		// attempt.
+		amt := attempt.Route.ReceiverAmt()
+		if !isBlinded && mpp == nil && amt != payment.Info.Value {
+			return ErrValueMismatch
+		}
+
+		// Ensure we aren't sending more than the total payment amount.
+		sentAmt, _ := payment.SentAmt()
+		if sentAmt+amt > payment.Info.Value {
+			return fmt.Errorf("%w: attempted=%v, payment amount="+
+				"%v", ErrValueExceedsAmt, sentAmt+amt,
+				payment.Info.Value)
+		}
+
+		htlcsBucket, err := bucket.CreateBucketIfNotExists(
+			paymentHtlcsBucket,
+		)
+		if err != nil {
+			return err
+		}
+
+		err = htlcsBucket.Put(
+			htlcBucketKey(htlcAttemptInfoKey, htlcIDBytes),
+			htlcInfoBytes,
+		)
+		if err != nil {
+			return err
+		}
+
+		// Retrieve attempt info for the notification.
+		payment, err = fetchPayment(bucket)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return payment, err
+}
+
+// SettleAttempt marks the given attempt settled with the preimage. If this is
+// a multi shard payment, this might implicitly mean that the full payment
+// succeeded.
+//
+// After invoking this method, InitPayment should always return an error to
+// prevent us from making duplicate payments to the same payment hash. The
+// provided preimage is atomically saved to the DB for record keeping.
+func (p *PaymentControl) SettleAttempt(hash lntypes.Hash,
+	attemptID uint64, settleInfo *HTLCSettleInfo) (*MPPayment, error) {
+
+	var b bytes.Buffer
+	if err := serializeHTLCSettleInfo(&b, settleInfo); err != nil {
+		return nil, err
+	}
+	settleBytes := b.Bytes()
+
+	return p.updateHtlcKey(hash, attemptID, htlcSettleInfoKey, settleBytes)
+}
+
+// FailAttempt marks the given payment attempt failed.
+func (p *PaymentControl) FailAttempt(hash lntypes.Hash,
+	attemptID uint64, failInfo *HTLCFailInfo) (*MPPayment, error) {
+
+	var b bytes.Buffer
+	if err := serializeHTLCFailInfo(&b, failInfo); err != nil {
+		return nil, err
+	}
+	failBytes := b.Bytes()
+
+	return p.updateHtlcKey(hash, attemptID, htlcFailInfoKey, failBytes)
+}
+
+// updateHtlcKey updates a database key for the specified htlc.
+func (p *PaymentControl) updateHtlcKey(paymentHash lntypes.Hash,
+	attemptID uint64, key, value []byte) (*MPPayment, error) {
+
+	aid := make([]byte, 8)
+	binary.BigEndian.PutUint64(aid, attemptID)
+
+	var payment *MPPayment
+	err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
+		payment = nil
+
+		prefetchPayment(tx, paymentHash)
+		bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
+		if err != nil {
+			return err
+		}
+
+		p, err := fetchPayment(bucket)
+		if err != nil {
+			return err
+		}
+
+		// We can only update keys of in-flight payments. We allow
+		// updating keys even if the payment has reached a terminal
+		// condition, since the HTLC outcomes must still be updated.
+		if err := p.Status.updatable(); err != nil {
+			return err
+		}
+
+		htlcsBucket := bucket.NestedReadWriteBucket(paymentHtlcsBucket)
+		if htlcsBucket == nil {
+			return fmt.Errorf("htlcs bucket not found")
+		}
+
+		if htlcsBucket.Get(htlcBucketKey(htlcAttemptInfoKey, aid)) == nil {
+			return fmt.Errorf("HTLC with ID %v not registered",
+				attemptID)
+		}
+
+		// Make sure the shard is not already failed or settled.
+		if htlcsBucket.Get(htlcBucketKey(htlcFailInfoKey, aid)) != nil {
+			return ErrAttemptAlreadyFailed
+		}
+
+		if htlcsBucket.Get(htlcBucketKey(htlcSettleInfoKey, aid)) != nil {
+			return ErrAttemptAlreadySettled
+		}
+
+		// Add or update the key for this htlc.
+		err = htlcsBucket.Put(htlcBucketKey(key, aid), value)
+		if err != nil {
+			return err
+		}
+
+		// Retrieve attempt info for the notification.
+		payment, err = fetchPayment(bucket)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return payment, err
+}
+
+// Fail transitions a payment into the Failed state, and records the reason the
+// payment failed. After invoking this method, InitPayment should return nil on
+// its next call for this payment hash, allowing the switch to make a
+// subsequent payment.
+func (p *PaymentControl) Fail(paymentHash lntypes.Hash,
+	reason FailureReason) (*MPPayment, error) {
+
+	var (
+		updateErr error
+		payment   *MPPayment
+	)
+	err := kvdb.Batch(p.db.Backend, func(tx kvdb.RwTx) error {
+		// Reset the update error, to avoid carrying over an error
+		// from a previous execution of the batched db transaction.
+		updateErr = nil
+		payment = nil
+
+		prefetchPayment(tx, paymentHash)
+		bucket, err := fetchPaymentBucketUpdate(tx, paymentHash)
+		if err == ErrPaymentNotInitiated {
+			updateErr = ErrPaymentNotInitiated
+			return nil
+		} else if err != nil {
+			return err
+		}
+
+		// We mark the payment as failed as long as it is known. This
+		// lets the last attempt to fail with a terminal write its
+		// failure to the PaymentControl without synchronizing with
+		// other attempts.
+		_, err = fetchPaymentStatus(bucket)
+		if errors.Is(err, ErrPaymentNotInitiated) {
+			updateErr = ErrPaymentNotInitiated
+			return nil
+		} else if err != nil {
+			return err
+		}
+
+		// Put the failure reason in the bucket for record keeping.
+		v := []byte{byte(reason)}
+		err = bucket.Put(paymentFailInfoKey, v)
+		if err != nil {
+			return err
+		}
+
+		// Retrieve attempt info for the notification, if available.
+		payment, err = fetchPayment(bucket)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return payment, updateErr
+}
+
+// FetchPayment returns information about a payment from the database.
+func (p *PaymentControl) FetchPayment(paymentHash lntypes.Hash) (
+	*MPPayment, error) {
+
+	var payment *MPPayment
+	err := kvdb.View(p.db, func(tx kvdb.RTx) error {
+		prefetchPayment(tx, paymentHash)
+		bucket, err := fetchPaymentBucket(tx, paymentHash)
+		if err != nil {
+			return err
+		}
+
+		payment, err = fetchPayment(bucket)
+
+		return err
+	}, func() {
+		payment = nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return payment, nil
+}
+
+// prefetchPayment attempts to prefetch as much of the payment as possible to
+// reduce DB roundtrips.
+func prefetchPayment(tx kvdb.RTx, paymentHash lntypes.Hash) {
+	rb := kvdb.RootBucket(tx)
+	kvdb.Prefetch(
+		rb,
+		[]string{
+			// Prefetch all keys in the payment's bucket.
+			string(paymentsRootBucket),
+			string(paymentHash[:]),
+		},
+		[]string{
+			// Prefetch all keys in the payment's htlc bucket.
+			string(paymentsRootBucket),
+			string(paymentHash[:]),
+			string(paymentHtlcsBucket),
+		},
+	)
+}
+
+// createPaymentBucket creates or fetches the sub-bucket assigned to this
+// payment hash.
+func createPaymentBucket(tx kvdb.RwTx, paymentHash lntypes.Hash) (
+	kvdb.RwBucket, error) {
+
+	payments, err := tx.CreateTopLevelBucket(paymentsRootBucket)
+	if err != nil {
+		return nil, err
+	}
+
+	return payments.CreateBucketIfNotExists(paymentHash[:])
+}
+
+// fetchPaymentBucket fetches the sub-bucket assigned to this payment hash. If
+// the bucket does not exist, it returns ErrPaymentNotInitiated.
+func fetchPaymentBucket(tx kvdb.RTx, paymentHash lntypes.Hash) (
+	kvdb.RBucket, error) {
+
+	payments := tx.ReadBucket(paymentsRootBucket)
+	if payments == nil {
+		return nil, ErrPaymentNotInitiated
+	}
+
+	bucket := payments.NestedReadBucket(paymentHash[:])
+	if bucket == nil {
+		return nil, ErrPaymentNotInitiated
+	}
+
+	return bucket, nil
+
+}
+
+// fetchPaymentBucketUpdate is identical to fetchPaymentBucket, but it returns a
+// bucket that can be written to.
+func fetchPaymentBucketUpdate(tx kvdb.RwTx, paymentHash lntypes.Hash) (
+	kvdb.RwBucket, error) {
+
+	payments := tx.ReadWriteBucket(paymentsRootBucket)
+	if payments == nil {
+		return nil, ErrPaymentNotInitiated
+	}
+
+	bucket := payments.NestedReadWriteBucket(paymentHash[:])
+	if bucket == nil {
+		return nil, ErrPaymentNotInitiated
+	}
+
+	return bucket, nil
+}
+
+// nextPaymentSequence returns the next sequence number to store for a new
+// payment.
+func (p *PaymentControl) nextPaymentSequence() ([]byte, error) {
+	p.paymentSeqMx.Lock()
+	defer p.paymentSeqMx.Unlock()
+
+	// Set a new upper bound in the DB every 1000 payments to avoid
+	// conflicts on the sequence when using etcd.
+	if p.currPaymentSeq == p.storedPaymentSeq {
+		var currPaymentSeq, newUpperBound uint64
+		if err := kvdb.Update(p.db.Backend, func(tx kvdb.RwTx) error {
+			paymentsBucket, err := tx.CreateTopLevelBucket(
+				paymentsRootBucket,
+			)
+			if err != nil {
+				return err
+			}
+
+			currPaymentSeq = paymentsBucket.Sequence()
+			newUpperBound = currPaymentSeq + paymentSeqBlockSize
+			return paymentsBucket.SetSequence(newUpperBound)
+		}, func() {}); err != nil {
+			return nil, err
+		}
+
+		// We lazy initialize the cached currPaymentSeq here using the
+		// first nextPaymentSequence() call. This if statement will auto
+		// initialize our stored currPaymentSeq, since by default both
+		// this variable and storedPaymentSeq are zero which in turn
+		// will have us fetch the current values from the DB.
+		if p.currPaymentSeq == 0 {
+			p.currPaymentSeq = currPaymentSeq
+		}
+
+		p.storedPaymentSeq = newUpperBound
+	}
+
+	p.currPaymentSeq++
+	b := make([]byte, 8)
+	binary.BigEndian.PutUint64(b, p.currPaymentSeq)
+
+	return b, nil
+}
+
+// fetchPaymentStatus fetches the payment status of the payment. If the payment
+// isn't found, it will return error `ErrPaymentNotInitiated`.
+func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) {
+	// Creation info should be set for all payments, regardless of state.
+	// If not, it is unknown.
+	if bucket.Get(paymentCreationInfoKey) == nil {
+		return 0, ErrPaymentNotInitiated
+	}
+
+	payment, err := fetchPayment(bucket)
+	if err != nil {
+		return 0, err
+	}
+
+	return payment.Status, nil
+}
+
+// FetchInFlightPayments returns all payments with status InFlight.
+func (p *PaymentControl) FetchInFlightPayments() ([]*MPPayment, error) {
+	var (
+		inFlights      []*MPPayment
+		start          = time.Now()
+		lastLogTime    = time.Now()
+		processedCount int
+	)
+
+	err := kvdb.View(p.db, func(tx kvdb.RTx) error {
+		payments := tx.ReadBucket(paymentsRootBucket)
+		if payments == nil {
+			return nil
+		}
+
+		return payments.ForEach(func(k, _ []byte) error {
+			bucket := payments.NestedReadBucket(k)
+			if bucket == nil {
+				return fmt.Errorf("non bucket element")
+			}
+
+			p, err := fetchPayment(bucket)
+			if err != nil {
+				return err
+			}
+
+			processedCount++
+			if time.Since(lastLogTime) >=
+				paymentProgressLogInterval {
+
+				log.Debugf("Scanning inflight payments "+
+					"(in progress), processed %d, last "+
+					"processed payment: %v", processedCount,
+					p.Info)
+
+				lastLogTime = time.Now()
+			}
+
+			// Skip the payment if it's terminated.
+			if p.Terminated() {
+				return nil
+			}
+
+			inFlights = append(inFlights, p)
+			return nil
+		})
+	}, func() {
+		inFlights = nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	elapsed := time.Since(start)
+	log.Debugf("Completed scanning for inflight payments: "+
+		"total_processed=%d, found_inflight=%d, elapsed=%v",
+		processedCount, len(inFlights),
+		elapsed.Round(time.Millisecond))
+
+	return inFlights, nil
+}