Merge pull request #8174 from yyforyongyu/fix-inflight-payments

routing: fix stuck inflight payments

Author: yyforyongyu

channeldb/forwarding_package.go

@@ -211,6 +211,11 @@ func (f *PkgFilter) Decode(r io.Reader) error {
 	return err
 }
 
+// String returns a human-readable string.
+func (f *PkgFilter) String() string {
+	return fmt.Sprintf("count=%v, filter=%v", f.count, f.filter)
+}
+
 // FwdPkg records all adds, settles, and fails that were locked in as a result
 // of the remote peer sending us a revocation. Each package is identified by
 // the short chanid and remote commitment height corresponding to the revocation

channeldb/payment_control.go

@@ -424,7 +424,9 @@ func (p *PaymentControl) RegisterAttempt(paymentHash lntypes.Hash,
 		// Ensure we aren't sending more than the total payment amount.
 		sentAmt, _ := payment.SentAmt()
 		if sentAmt+amt > payment.Info.Value {
-			return ErrValueExceedsAmt
+			return fmt.Errorf("%w: attempted=%v, payment amount="+
+				"%v", ErrValueExceedsAmt, sentAmt+amt,
+				payment.Info.Value)
 		}
 
 		htlcsBucket, err := bucket.CreateBucketIfNotExists(

channeldb/payment_control_test.go

@@ -734,10 +734,7 @@ func TestPaymentControlMultiShard(t *testing.T) {
 		b := *attempt
 		b.AttemptID = 3
 		_, err = pControl.RegisterAttempt(info.PaymentIdentifier, &b)
-		if err != ErrValueExceedsAmt {
-			t.Fatalf("expected ErrValueExceedsAmt, got: %v",
-				err)
-		}
+		require.ErrorIs(t, err, ErrValueExceedsAmt)
 
 		// Fail the second attempt.
 		a := attempts[1]

docs/release-notes/release-notes-0.18.3.md

@@ -48,6 +48,15 @@
 bumping an anchor channel closing was not possible when no HTLCs were on the
 commitment when the channel was force closed.
 
+* [Fixed](https://github.com/lightningnetwork/lnd/pull/8174) old payments that
+  are stuck inflight. Though the root cause is unknown, it's possible the
+  network result for a given HTLC attempt was not saved, which is now fixed.
+  Check
+  [here](https://github.com/lightningnetwork/lnd/pull/8174#issue-1992055103)
+  for the details about the analysis, and
+  [here](https://github.com/lightningnetwork/lnd/issues/8146) for a summary of
+  the issue.
+
 # New Features
 ## Functional Enhancements
 ## RPC Additions

htlcswitch/link.go

@@ -3297,7 +3297,7 @@ func (l *channelLink) processRemoteSettleFails(fwdPkg *channeldb.FwdPkg,
 		return
 	}
 
-	l.log.Debugf("settle-fail-filter %v", fwdPkg.SettleFailFilter)
+	l.log.Debugf("settle-fail-filter: %v", fwdPkg.SettleFailFilter)
 
 	var switchPackets []*htlcPacket
 	for i, pd := range settleFails {

htlcswitch/payment_result.go

@@ -90,41 +90,41 @@ type networkResultStore struct {
 	results    map[uint64][]chan *networkResult
 	resultsMtx sync.Mutex
 
-	// paymentIDMtx is a multimutex used to make sure the database and
-	// result subscribers map is consistent for each payment ID in case of
+	// attemptIDMtx is a multimutex used to make sure the database and
+	// result subscribers map is consistent for each attempt ID in case of
 	// concurrent callers.
-	paymentIDMtx *multimutex.Mutex[uint64]
+	attemptIDMtx *multimutex.Mutex[uint64]
 }
 
 func newNetworkResultStore(db kvdb.Backend) *networkResultStore {
 	return &networkResultStore{
 		backend:      db,
 		results:      make(map[uint64][]chan *networkResult),
-		paymentIDMtx: multimutex.NewMutex[uint64](),
+		attemptIDMtx: multimutex.NewMutex[uint64](),
 	}
 }
 
-// storeResult stores the networkResult for the given paymentID, and
-// notifies any subscribers.
-func (store *networkResultStore) storeResult(paymentID uint64,
+// storeResult stores the networkResult for the given attemptID, and notifies
+// any subscribers.
+func (store *networkResultStore) storeResult(attemptID uint64,
 	result *networkResult) error {
 
-	// We get a mutex for this payment ID. This is needed to ensure
+	// We get a mutex for this attempt ID. This is needed to ensure
 	// consistency between the database state and the subscribers in case
 	// of concurrent calls.
-	store.paymentIDMtx.Lock(paymentID)
-	defer store.paymentIDMtx.Unlock(paymentID)
+	store.attemptIDMtx.Lock(attemptID)
+	defer store.attemptIDMtx.Unlock(attemptID)
 
-	log.Debugf("Storing result for paymentID=%v", paymentID)
+	log.Debugf("Storing result for attemptID=%v", attemptID)
 
 	// Serialize the payment result.
 	var b bytes.Buffer
 	if err := serializeNetworkResult(&b, result); err != nil {
 		return err
 	}
 
-	var paymentIDBytes [8]byte
-	binary.BigEndian.PutUint64(paymentIDBytes[:], paymentID)
+	var attemptIDBytes [8]byte
+	binary.BigEndian.PutUint64(attemptIDBytes[:], attemptID)
 
 	err := kvdb.Batch(store.backend, func(tx kvdb.RwTx) error {
 		networkResults, err := tx.CreateTopLevelBucket(
@@ -134,7 +134,7 @@ func (store *networkResultStore) storeResult(paymentID uint64,
 			return err
 		}
 
-		return networkResults.Put(paymentIDBytes[:], b.Bytes())
+		return networkResults.Put(attemptIDBytes[:], b.Bytes())
 	})
 	if err != nil {
 		return err
@@ -143,28 +143,27 @@ func (store *networkResultStore) storeResult(paymentID uint64,
 	// Now that the result is stored in the database, we can notify any
 	// active subscribers.
 	store.resultsMtx.Lock()
-	for _, res := range store.results[paymentID] {
+	for _, res := range store.results[attemptID] {
 		res <- result
 	}
-	delete(store.results, paymentID)
+	delete(store.results, attemptID)
 	store.resultsMtx.Unlock()
 
 	return nil
 }
 
-// subscribeResult is used to get the payment result for the given
-// payment ID. It returns a channel on which the result will be delivered when
-// ready.
-func (store *networkResultStore) subscribeResult(paymentID uint64) (
+// subscribeResult is used to get the HTLC attempt result for the given attempt
+// ID.  It returns a channel on which the result will be delivered when ready.
+func (store *networkResultStore) subscribeResult(attemptID uint64) (
 	<-chan *networkResult, error) {
 
 	// We get a mutex for this payment ID. This is needed to ensure
 	// consistency between the database state and the subscribers in case
 	// of concurrent calls.
-	store.paymentIDMtx.Lock(paymentID)
-	defer store.paymentIDMtx.Unlock(paymentID)
+	store.attemptIDMtx.Lock(attemptID)
+	defer store.attemptIDMtx.Unlock(attemptID)
 
-	log.Debugf("Subscribing to result for paymentID=%v", paymentID)
+	log.Debugf("Subscribing to result for attemptID=%v", attemptID)
 
 	var (
 		result     *networkResult
@@ -173,7 +172,7 @@ func (store *networkResultStore) subscribeResult(paymentID uint64) (
 
 	err := kvdb.View(store.backend, func(tx kvdb.RTx) error {
 		var err error
-		result, err = fetchResult(tx, paymentID)
+		result, err = fetchResult(tx, attemptID)
 		switch {
 
 		// Result not yet available, we will notify once a result is
@@ -205,8 +204,8 @@ func (store *networkResultStore) subscribeResult(paymentID uint64) (
 	// Otherwise we store the result channel for when the result is
 	// available.
 	store.resultsMtx.Lock()
-	store.results[paymentID] = append(
-		store.results[paymentID], resultChan,
+	store.results[attemptID] = append(
+		store.results[attemptID], resultChan,
 	)
 	store.resultsMtx.Unlock()
 
@@ -234,16 +233,16 @@ func (store *networkResultStore) getResult(pid uint64) (
 }
 
 func fetchResult(tx kvdb.RTx, pid uint64) (*networkResult, error) {
-	var paymentIDBytes [8]byte
-	binary.BigEndian.PutUint64(paymentIDBytes[:], pid)
+	var attemptIDBytes [8]byte
+	binary.BigEndian.PutUint64(attemptIDBytes[:], pid)
 
 	networkResults := tx.ReadBucket(networkResultStoreBucketKey)
 	if networkResults == nil {
 		return nil, ErrPaymentIDNotFound
 	}
 
 	// Check whether a result is already available.
-	resultBytes := networkResults.Get(paymentIDBytes[:])
+	resultBytes := networkResults.Get(attemptIDBytes[:])
 	if resultBytes == nil {
 		return nil, ErrPaymentIDNotFound
 	}