routing: serialize payment lifecycles by hash

Author: yyforyongyu

lnrpc/routerrpc/router_server.go

@@ -375,7 +375,8 @@ func (s *Server) SendPaymentV2(req *SendPaymentRequest,
 	payHash := payment.Identifier()
 
 	// Init the payment in db.
-	paySession, shardTracker, err := s.cfg.Router.PreparePayment(payment)
+	paySession, shardTracker, releaseLifecycle, err := s.cfg.Router.
+		PreparePayment(payment)
 	if err != nil {
 		log.Errorf("SendPayment async error for payment %x: %v",
 			payment.Identifier(), err)
@@ -397,6 +398,8 @@ func (s *Server) SendPaymentV2(req *SendPaymentRequest,
 	// miss events.
 	sub, err := s.subscribePayment(payHash)
 	if err != nil {
+		releaseLifecycle()
+
 		return err
 	}
 
@@ -415,7 +418,9 @@ func (s *Server) SendPaymentV2(req *SendPaymentRequest,
 	}
 
 	// Send the payment asynchronously.
-	s.cfg.Router.SendPaymentAsync(ctx, payment, paySession, shardTracker)
+	s.cfg.Router.SendPaymentAsync(
+		ctx, payment, paySession, shardTracker, releaseLifecycle,
+	)
 
 	// Track the payment and return.
 	return s.trackPayment(sub, payHash, stream, req.NoInflightUpdates)

routing/router.go

@@ -22,6 +22,7 @@ import (
 	"github.com/lightningnetwork/lnd/lnutils"
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
+	"github.com/lightningnetwork/lnd/multimutex"
 	paymentsdb "github.com/lightningnetwork/lnd/payments/db"
 	"github.com/lightningnetwork/lnd/record"
 	"github.com/lightningnetwork/lnd/routing/route"
@@ -334,6 +335,13 @@ type ChannelRouter struct {
 
 	quit chan struct{}
 	wg   sync.WaitGroup
+
+	// paymentLifecycleMtx ensures only one payment lifecycle for a given
+	// payment hash can be active at a time. Failed payments are retryable,
+	// so the lock must be held until resumePayment fully exits rather than
+	// only until the terminal status update is published.
+	paymentLifecycleMtx     *multimutex.Mutex[lntypes.Hash]
+	paymentLifecycleMtxOnce sync.Once
 }
 
 // New creates a new instance of the ChannelRouter with the specified
@@ -343,8 +351,9 @@ type ChannelRouter struct {
 // to fully sync to the latest state of the UTXO set.
 func New(cfg Config) (*ChannelRouter, error) {
 	return &ChannelRouter{
-		cfg:  &cfg,
-		quit: make(chan struct{}),
+		cfg:                 &cfg,
+		quit:                make(chan struct{}),
+		paymentLifecycleMtx: multimutex.NewMutex[lntypes.Hash](),
 	}, nil
 }
 
@@ -903,7 +912,9 @@ func (l *LightningPayment) Identifier() [32]byte {
 func (r *ChannelRouter) SendPayment(ctx context.Context,
 	payment *LightningPayment) ([32]byte, *route.Route, error) {
 
-	paySession, shardTracker, err := r.PreparePayment(payment)
+	paySession, shardTracker, releaseLifecycle, err := r.PreparePayment(
+		payment,
+	)
 	if err != nil {
 		return [32]byte{}, nil, err
 	}
@@ -914,14 +925,15 @@ func (r *ChannelRouter) SendPayment(ctx context.Context,
 	return r.sendPayment(
 		ctx, payment.FeeLimit, payment.Identifier(),
 		payment.PayAttemptTimeout, paySession, shardTracker,
-		payment.FirstHopCustomRecords,
+		payment.FirstHopCustomRecords, releaseLifecycle,
 	)
 }
 
 // SendPaymentAsync is the non-blocking version of SendPayment. The payment
 // result needs to be retrieved via the control tower.
 func (r *ChannelRouter) SendPaymentAsync(ctx context.Context,
-	payment *LightningPayment, ps PaymentSession, st shards.ShardTracker) {
+	payment *LightningPayment, ps PaymentSession, st shards.ShardTracker,
+	releaseLifecycle func()) {
 
 	// Since this is the first time this payment is being made, we pass nil
 	// for the existing attempt.
@@ -935,7 +947,7 @@ func (r *ChannelRouter) SendPaymentAsync(ctx context.Context,
 		_, _, err := r.sendPayment(
 			ctx, payment.FeeLimit, payment.Identifier(),
 			payment.PayAttemptTimeout, ps, st,
-			payment.FirstHopCustomRecords,
+			payment.FirstHopCustomRecords, releaseLifecycle,
 		)
 		if err != nil {
 			log.Errorf("Payment %x failed: %v",
@@ -966,24 +978,43 @@ func spewPayment(payment *LightningPayment) lnutils.LogClosure {
 	})
 }
 
+// lockPaymentLifecycle locks the payment lifecycle mutex for the given payment
+// hash and returns a release function.
+func (r *ChannelRouter) lockPaymentLifecycle(paymentHash lntypes.Hash) func() {
+	r.paymentLifecycleMtxOnce.Do(func() {
+		if r.paymentLifecycleMtx == nil {
+			r.paymentLifecycleMtx = multimutex.
+				NewMutex[lntypes.Hash]()
+		}
+	})
+
+	r.paymentLifecycleMtx.Lock(paymentHash)
+
+	return func() {
+		r.paymentLifecycleMtx.Unlock(paymentHash)
+	}
+}
+
 // PreparePayment creates the payment session and registers the payment with the
-// control tower.
+// control tower. The returned release function must be called after the payment
+// lifecycle has fully exited.
 func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
-	PaymentSession, shards.ShardTracker, error) {
+	PaymentSession, shards.ShardTracker, func(), error) {
 
 	ctx := context.TODO()
 
 	// Assemble any custom data we want to send to the first hop only.
 	var firstHopData fn.Option[tlv.Blob]
 	if len(payment.FirstHopCustomRecords) > 0 {
 		if err := payment.FirstHopCustomRecords.Validate(); err != nil {
-			return nil, nil, fmt.Errorf("invalid first hop custom "+
-				"records: %w", err)
+			return nil, nil, nil, fmt.Errorf(
+				"invalid first hop custom records: %w", err,
+			)
 		}
 
 		firstHopBlob, err := payment.FirstHopCustomRecords.Serialize()
 		if err != nil {
-			return nil, nil, fmt.Errorf("unable to serialize "+
+			return nil, nil, nil, fmt.Errorf("unable to serialize "+
 				"first hop custom records: %w", err)
 		}
 
@@ -997,7 +1028,7 @@ func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
 		payment, firstHopData, r.cfg.TrafficShaper,
 	)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
 	// Record this payment hash with the ControlTower, ensuring it is not
@@ -1032,12 +1063,16 @@ func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
 		)
 	}
 
+	releaseLifecycle := r.lockPaymentLifecycle(payment.Identifier())
+
 	err = r.cfg.Control.InitPayment(ctx, payment.Identifier(), info)
 	if err != nil {
-		return nil, nil, err
+		releaseLifecycle()
+
+		return nil, nil, nil, err
 	}
 
-	return paySession, shardTracker, nil
+	return paySession, shardTracker, releaseLifecycle, nil
 }
 
 // SendToRoute sends a payment using the provided route and fails the payment
@@ -1264,8 +1299,12 @@ func (r *ChannelRouter) sendPayment(ctx context.Context,
 	feeLimit lnwire.MilliSatoshi, identifier lntypes.Hash,
 	paymentAttemptTimeout time.Duration, paySession PaymentSession,
 	shardTracker shards.ShardTracker,
-	firstHopCustomRecords lnwire.CustomRecords) ([32]byte, *route.Route,
-	error) {
+	firstHopCustomRecords lnwire.CustomRecords,
+	releaseLifecycle func()) ([32]byte, *route.Route, error) {
+
+	if releaseLifecycle != nil {
+		defer releaseLifecycle()
+	}
 
 	// If the user provides a timeout, we will additionally wrap the context
 	// in a deadline.
@@ -1511,9 +1550,11 @@ func (r *ChannelRouter) resumePayments() error {
 		// attempt has finished anyway. We also set a zero fee limit,
 		// as no more routes should be tried.
 		noTimeout := time.Duration(0)
+		releaseLifecycle := r.lockPaymentLifecycle(payHash)
 		_, _, err := r.sendPayment(
 			context.Background(), 0, payHash, noTimeout, paySession,
 			shardTracker, payment.Info.FirstHopCustomRecords,
+			releaseLifecycle,
 		)
 		if err != nil {
 			log.Errorf("Resuming payment %v failed: %v", payHash,

routing/router_test.go

@@ -1357,6 +1357,93 @@ func TestUnknownErrorSource(t *testing.T) {
 	}
 }
 
+// TestPreparePaymentWaitsForActiveLifecycle asserts that a retry with the same
+// payment hash cannot create a new control tower payment until the existing
+// payment lifecycle has fully exited.
+func TestPreparePaymentWaitsForActiveLifecycle(t *testing.T) {
+	t.Parallel()
+
+	payment := createDummyLightningPayment(
+		t, route.Vertex{}, lnwire.MilliSatoshi(1000),
+	)
+	payHash := lntypes.Hash(payment.Identifier())
+
+	controlTower := &mockControlTower{}
+	sessionSource := &mockPaymentSessionSource{}
+	r := &ChannelRouter{
+		cfg: &Config{
+			Clock:         clock.NewDefaultClock(),
+			Control:       controlTower,
+			SessionSource: sessionSource,
+			TrafficShaper: fn.None[htlcswitch.AuxTrafficShaper](),
+		},
+		quit: make(chan struct{}),
+	}
+
+	sessionCreated := make(chan struct{})
+	sessionSource.On(
+		"NewPaymentSession", payment, mock.Anything, mock.Anything,
+	).Run(func(_ mock.Arguments) {
+		close(sessionCreated)
+	}).Return(&mockPaymentSession{}, nil).Once()
+
+	initCalled := make(chan struct{})
+	controlTower.On(
+		"InitPayment", payHash,
+		mock.MatchedBy(func(info *paymentsdb.PaymentCreationInfo) bool {
+			return info.PaymentIdentifier == payHash
+		}),
+	).Run(func(_ mock.Arguments) {
+		close(initCalled)
+	}).Return(nil).Once()
+
+	releaseActiveLifecycle := r.lockPaymentLifecycle(payHash)
+
+	result := make(chan error, 1)
+	go func() {
+		_, _, releaseLifecycle, err := r.PreparePayment(payment)
+		if err == nil {
+			releaseLifecycle()
+		}
+
+		result <- err
+	}()
+
+	select {
+	case <-sessionCreated:
+
+	case <-time.After(testTimeout):
+		require.Fail(t, "payment session not created")
+	}
+
+	select {
+	case <-initCalled:
+		require.Fail(t, "payment initialized before lifecycle exit")
+
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	releaseActiveLifecycle()
+
+	select {
+	case err := <-result:
+		require.NoError(t, err)
+
+	case <-time.After(testTimeout):
+		require.Fail(t, "prepare payment did not complete")
+	}
+
+	select {
+	case <-initCalled:
+
+	default:
+		require.Fail(t, "payment was not initialized")
+	}
+
+	controlTower.AssertExpectations(t)
+	sessionSource.AssertExpectations(t)
+}
+
 // TestSendToRouteStructuredError asserts that SendToRoute returns a structured
 // error.
 func TestSendToRouteStructuredError(t *testing.T) {