From 8bb7150dc291ad9fe1b8ee518e57a41e6db26495 Mon Sep 17 00:00:00 2001 From: ziggie Date: Tue, 11 Nov 2025 11:23:15 +0100 Subject: [PATCH] multi: thread context through RegisterAttempt method --- payments/db/interface.go | 3 +- payments/db/kv_store.go | 2 +- payments/db/kv_store_test.go | 2 +- payments/db/payment_test.go | 52 +++++++++++++++++++++++------------ payments/db/sql_store.go | 4 +-- routing/control_tower.go | 9 +++--- routing/control_tower_test.go | 28 +++++++++++++------ routing/mock_test.go | 8 +++--- routing/payment_lifecycle.go | 4 ++- 9 files changed, 70 insertions(+), 42 deletions(-) diff --git a/payments/db/interface.go b/payments/db/interface.go index 2d0335609..3af2dfb67 100644 --- a/payments/db/interface.go +++ b/payments/db/interface.go @@ -75,7 +75,8 @@ type PaymentControl interface { // - Result: 1700 sats sent, exceeding the payment amount // The payment router/controller layer is responsible for ensuring // serialized access per payment hash. - RegisterAttempt(lntypes.Hash, *HTLCAttemptInfo) (*MPPayment, error) + RegisterAttempt(context.Context, 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 diff --git a/payments/db/kv_store.go b/payments/db/kv_store.go index 81b257ce8..5511bf8bc 100644 --- a/payments/db/kv_store.go +++ b/payments/db/kv_store.go @@ -359,7 +359,7 @@ func deserializePaymentIndex(r io.Reader) (lntypes.Hash, error) { // RegisterAttempt atomically records the provided HTLCAttemptInfo to the // DB. -func (p *KVStore) RegisterAttempt(paymentHash lntypes.Hash, +func (p *KVStore) RegisterAttempt(_ context.Context, paymentHash lntypes.Hash, attempt *HTLCAttemptInfo) (*MPPayment, error) { // Serialize the information before opening the db transaction. diff --git a/payments/db/kv_store_test.go b/payments/db/kv_store_test.go index 6837134e7..0c51dccf5 100644 --- a/payments/db/kv_store_test.go +++ b/payments/db/kv_store_test.go @@ -85,7 +85,7 @@ func TestKVStoreDeleteNonInFlight(t *testing.T) { t.Fatalf("unable to send htlc message: %v", err) } _, err = paymentDB.RegisterAttempt( - info.PaymentIdentifier, attempt, + ctx, info.PaymentIdentifier, attempt, ) if err != nil { t.Fatalf("unable to send htlc message: %v", err) diff --git a/payments/db/payment_test.go b/payments/db/payment_test.go index 4b2cbcbd8..8e2c7a47c 100644 --- a/payments/db/payment_test.go +++ b/payments/db/payment_test.go @@ -151,7 +151,7 @@ func createTestPayments(t *testing.T, p DB, payments []*payment) { require.NoError(t, err, "unable to send htlc message") // Register and fail the first attempt for all payments. - _, err = p.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = p.RegisterAttempt(ctx, info.PaymentIdentifier, attempt) require.NoError(t, err, "unable to send htlc message") htlcFailure := HTLCFailUnreadable @@ -175,7 +175,7 @@ func createTestPayments(t *testing.T, p DB, payments []*payment) { require.NoError(t, err) attemptID++ - _, err = p.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = p.RegisterAttempt(ctx, info.PaymentIdentifier, attempt) require.NoError(t, err, "unable to send htlc message") switch payments[i].status { @@ -592,7 +592,7 @@ func TestMPPRecordValidation(t *testing.T) { info.Value, [32]byte{1}, ) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = paymentDB.RegisterAttempt(ctx, info.PaymentIdentifier, attempt) require.NoError(t, err, "unable to send htlc message") // Now try to register a non-MPP attempt, which should fail. @@ -604,21 +604,27 @@ func TestMPPRecordValidation(t *testing.T) { attempt2.Route.FinalHop().MPP = nil - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt2) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, attempt2, + ) require.ErrorIs(t, err, ErrMPPayment) // Try to register attempt one with a different payment address. attempt2.Route.FinalHop().MPP = record.NewMPP( info.Value, [32]byte{2}, ) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt2) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, attempt2, + ) require.ErrorIs(t, err, ErrMPPPaymentAddrMismatch) // Try registering one with a different total amount. attempt2.Route.FinalHop().MPP = record.NewMPP( info.Value/2, [32]byte{1}, ) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt2) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, attempt2, + ) require.ErrorIs(t, err, ErrMPPTotalAmountMismatch) // Create and init a new payment. This time we'll check that we cannot @@ -641,7 +647,9 @@ func TestMPPRecordValidation(t *testing.T) { require.NoError(t, err, "unable to send htlc message") attempt.Route.FinalHop().MPP = nil - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, attempt, + ) require.NoError(t, err, "unable to send htlc message") // Attempt to register an MPP attempt, which should fail. @@ -655,7 +663,9 @@ func TestMPPRecordValidation(t *testing.T) { info.Value, [32]byte{1}, ) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt2) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, attempt2, + ) require.ErrorIs(t, err, ErrNonMPPayment) } @@ -1758,7 +1768,7 @@ func TestSwitchDoubleSend(t *testing.T) { require.ErrorIs(t, err, ErrPaymentExists) // Record an attempt. - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = paymentDB.RegisterAttempt(ctx, info.PaymentIdentifier, attempt) require.NoError(t, err, "unable to send htlc message") assertDBPaymentstatus( t, paymentDB, info.PaymentIdentifier, StatusInFlight, @@ -1869,7 +1879,7 @@ func TestSwitchFail(t *testing.T) { // Record a new attempt. In this test scenario, the attempt fails. // However, this is not communicated to control tower in the current // implementation. It only registers the initiation of the attempt. - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = paymentDB.RegisterAttempt(ctx, info.PaymentIdentifier, attempt) require.NoError(t, err, "unable to register attempt") htlcReason := HTLCFailUnreadable @@ -1899,7 +1909,7 @@ func TestSwitchFail(t *testing.T) { ) require.NoError(t, err) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, attempt) + _, err = paymentDB.RegisterAttempt(ctx, info.PaymentIdentifier, attempt) require.NoError(t, err, "unable to send htlc message") assertDBPaymentstatus( t, paymentDB, info.PaymentIdentifier, StatusInFlight, @@ -2017,7 +2027,7 @@ func TestMultiShard(t *testing.T) { attempts = append(attempts, a) _, err = paymentDB.RegisterAttempt( - info.PaymentIdentifier, a, + ctx, info.PaymentIdentifier, a, ) if err != nil { t.Fatalf("unable to send htlc message: %v", err) @@ -2049,7 +2059,9 @@ func TestMultiShard(t *testing.T) { info.Value, [32]byte{1}, ) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, b) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, b, + ) require.ErrorIs(t, err, ErrValueExceedsAmt) // Fail the second attempt. @@ -2156,7 +2168,9 @@ func TestMultiShard(t *testing.T) { info.Value, [32]byte{1}, ) - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, b) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, b, + ) if test.settleFirst { require.ErrorIs( t, err, ErrPaymentPendingSettled, @@ -2255,7 +2269,9 @@ func TestMultiShard(t *testing.T) { ) // Finally assert we cannot register more attempts. - _, err = paymentDB.RegisterAttempt(info.PaymentIdentifier, b) + _, err = paymentDB.RegisterAttempt( + ctx, info.PaymentIdentifier, b, + ) require.ErrorIs(t, err, registerErr) } @@ -2658,7 +2674,7 @@ func TestQueryPayments(t *testing.T) { require.NoError(t, err) _, err = paymentDB.RegisterAttempt( - lastPaymentInfo.PaymentIdentifier, + ctx, lastPaymentInfo.PaymentIdentifier, &attempt.HTLCAttemptInfo, ) require.NoError(t, err) @@ -2842,7 +2858,7 @@ func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) { require.NoError(t, err) _, err = paymentDB.RegisterAttempt( - info.PaymentIdentifier, attempt1, + ctx, info.PaymentIdentifier, attempt1, ) require.NoError(t, err) @@ -2850,7 +2866,7 @@ func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) { require.NoError(t, err) _, err = paymentDB.RegisterAttempt( - info.PaymentIdentifier, attempt2, + ctx, info.PaymentIdentifier, attempt2, ) require.NoError(t, err) diff --git a/payments/db/sql_store.go b/payments/db/sql_store.go index cacd5cdd6..8c963d360 100644 --- a/payments/db/sql_store.go +++ b/payments/db/sql_store.go @@ -1496,11 +1496,9 @@ func (s *SQLStore) insertRouteHops(ctx context.Context, db SQLQueries, // the PaymentWriter interface and ultimately the DB interface. It represents // step 2 in the payment lifecycle control flow, called after InitPayment and // potentially multiple times for multi-path payments. -func (s *SQLStore) RegisterAttempt(paymentHash lntypes.Hash, +func (s *SQLStore) RegisterAttempt(ctx context.Context, paymentHash lntypes.Hash, attempt *HTLCAttemptInfo) (*MPPayment, error) { - ctx := context.TODO() - var mpPayment *MPPayment err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error { diff --git a/routing/control_tower.go b/routing/control_tower.go index 8df87b473..28432d732 100644 --- a/routing/control_tower.go +++ b/routing/control_tower.go @@ -31,7 +31,8 @@ type ControlTower interface { // RegisterAttempt atomically records the provided HTLCAttemptInfo. // // NOTE: Subscribers should be notified by the new state of the payment. - RegisterAttempt(lntypes.Hash, *paymentsdb.HTLCAttemptInfo) error + RegisterAttempt(context.Context, lntypes.Hash, + *paymentsdb.HTLCAttemptInfo) error // SettleAttempt marks the given attempt settled with the preimage. If // this is a multi shard payment, this might implicitly mean the the @@ -196,13 +197,13 @@ func (p *controlTower) DeleteFailedAttempts(paymentHash lntypes.Hash) error { // RegisterAttempt atomically records the provided HTLCAttemptInfo to the // DB. -func (p *controlTower) RegisterAttempt(paymentHash lntypes.Hash, - attempt *paymentsdb.HTLCAttemptInfo) error { +func (p *controlTower) RegisterAttempt(ctx context.Context, + paymentHash lntypes.Hash, attempt *paymentsdb.HTLCAttemptInfo) error { p.paymentsMtx.Lock(paymentHash) defer p.paymentsMtx.Unlock(paymentHash) - payment, err := p.db.RegisterAttempt(paymentHash, attempt) + payment, err := p.db.RegisterAttempt(ctx, paymentHash, attempt) if err != nil { return err } diff --git a/routing/control_tower_test.go b/routing/control_tower_test.go index 0993fb2a6..20bdd1756 100644 --- a/routing/control_tower_test.go +++ b/routing/control_tower_test.go @@ -92,7 +92,9 @@ func TestControlTowerSubscribeSuccess(t *testing.T) { require.NoError(t, err, "expected subscribe to succeed, but got") // Register an attempt. - err = pControl.RegisterAttempt(info.PaymentIdentifier, attempt) + err = pControl.RegisterAttempt( + t.Context(), info.PaymentIdentifier, attempt, + ) if err != nil { t.Fatal(err) } @@ -221,7 +223,9 @@ func TestKVStoreSubscribeAllSuccess(t *testing.T) { require.NoError(t, err, "expected subscribe to succeed, but got: %v") // Register an attempt. - err = pControl.RegisterAttempt(info1.PaymentIdentifier, attempt1) + err = pControl.RegisterAttempt( + t.Context(), info1.PaymentIdentifier, attempt1, + ) require.NoError(t, err) // Initiate a second payment after the subscription is already active. @@ -232,7 +236,9 @@ func TestKVStoreSubscribeAllSuccess(t *testing.T) { require.NoError(t, err) // Register an attempt on the second payment. - err = pControl.RegisterAttempt(info2.PaymentIdentifier, attempt2) + err = pControl.RegisterAttempt( + t.Context(), info2.PaymentIdentifier, attempt2, + ) require.NoError(t, err) // Mark the first payment as successful. @@ -341,7 +347,9 @@ func TestKVStoreSubscribeAllImmediate(t *testing.T) { require.NoError(t, err) // Register a payment update. - err = pControl.RegisterAttempt(info.PaymentIdentifier, attempt) + err = pControl.RegisterAttempt( + t.Context(), info.PaymentIdentifier, attempt, + ) require.NoError(t, err) subscription, err := pControl.SubscribeAllPayments() @@ -414,7 +422,9 @@ func TestKVStoreUnsubscribeSuccess(t *testing.T) { subscription1.Close() // Register a payment update. - err = pControl.RegisterAttempt(info.PaymentIdentifier, attempt) + err = pControl.RegisterAttempt( + t.Context(), info.PaymentIdentifier, attempt, + ) require.NoError(t, err) // Assert only subscription 2 receives the update. @@ -479,10 +489,10 @@ func testKVStoreSubscribeFail(t *testing.T, registerAttempt, // making any attempts at all. if registerAttempt { // Register an attempt. - err = pControl.RegisterAttempt(info.PaymentIdentifier, attempt) - if err != nil { - t.Fatal(err) - } + err = pControl.RegisterAttempt( + t.Context(), info.PaymentIdentifier, attempt, + ) + require.NoError(t, err) // Fail the payment attempt. failInfo := paymentsdb.HTLCFailInfo{ diff --git a/routing/mock_test.go b/routing/mock_test.go index 5b9d4854b..daad344fd 100644 --- a/routing/mock_test.go +++ b/routing/mock_test.go @@ -354,8 +354,8 @@ func (m *mockControlTowerOld) DeleteFailedAttempts(phash lntypes.Hash) error { return nil } -func (m *mockControlTowerOld) RegisterAttempt(phash lntypes.Hash, - a *paymentsdb.HTLCAttemptInfo) error { +func (m *mockControlTowerOld) RegisterAttempt(_ context.Context, + phash lntypes.Hash, a *paymentsdb.HTLCAttemptInfo) error { if m.registerAttempt != nil { m.registerAttempt <- registerAttemptArgs{a} @@ -746,8 +746,8 @@ func (m *mockControlTower) DeleteFailedAttempts(phash lntypes.Hash) error { return args.Error(0) } -func (m *mockControlTower) RegisterAttempt(phash lntypes.Hash, - a *paymentsdb.HTLCAttemptInfo) error { +func (m *mockControlTower) RegisterAttempt(_ context.Context, + phash lntypes.Hash, a *paymentsdb.HTLCAttemptInfo) error { args := m.Called(phash, a) return args.Error(0) diff --git a/routing/payment_lifecycle.go b/routing/payment_lifecycle.go index 4eb78c8e2..049947541 100644 --- a/routing/payment_lifecycle.go +++ b/routing/payment_lifecycle.go @@ -584,6 +584,8 @@ func (p *paymentLifecycle) collectResult( func (p *paymentLifecycle) registerAttempt(rt *route.Route, remainingAmt lnwire.MilliSatoshi) (*paymentsdb.HTLCAttempt, error) { + ctx := context.TODO() + // If this route will consume the last remaining amount to send // to the receiver, this will be our last shard (for now). isLastAttempt := rt.ReceiverAmt() == remainingAmt @@ -601,7 +603,7 @@ func (p *paymentLifecycle) registerAttempt(rt *route.Route, // Switch for its whereabouts. The route is needed to handle the result // when it eventually comes back. err = p.router.cfg.Control.RegisterAttempt( - p.identifier, &attempt.HTLCAttemptInfo, + ctx, p.identifier, &attempt.HTLCAttemptInfo, ) return attempt, err