From 25d05be0384373da2c0053555814e90835ba62ed Mon Sep 17 00:00:00 2001 From: ziggie Date: Mon, 20 Oct 2025 20:16:47 +0200 Subject: [PATCH] multi: thread context through FetchPayment --- payments/db/interface.go | 3 ++- payments/db/kv_store.go | 4 ++-- payments/db/kv_store_test.go | 10 ++++++---- payments/db/payment_test.go | 16 +++++++++++----- payments/db/sql_store.go | 4 ++-- payments/db/test_kvdb.go | 4 +++- routing/control_tower.go | 17 ++++++++++++----- routing/mock_test.go | 9 +++++---- routing/payment_lifecycle.go | 8 ++++++-- routing/router.go | 4 +++- routing/router_test.go | 4 +++- 11 files changed, 55 insertions(+), 28 deletions(-) diff --git a/payments/db/interface.go b/payments/db/interface.go index c6f1bf353..5368d53d3 100644 --- a/payments/db/interface.go +++ b/payments/db/interface.go @@ -21,7 +21,8 @@ type PaymentReader interface { // FetchPayment fetches the payment corresponding to the given payment // hash. - FetchPayment(paymentHash lntypes.Hash) (*MPPayment, error) + FetchPayment(ctx context.Context, + paymentHash lntypes.Hash) (*MPPayment, error) // FetchInFlightPayments returns all payments with status InFlight. FetchInFlightPayments() ([]*MPPayment, error) diff --git a/payments/db/kv_store.go b/payments/db/kv_store.go index d3d347afe..86b37edcf 100644 --- a/payments/db/kv_store.go +++ b/payments/db/kv_store.go @@ -585,8 +585,8 @@ func (p *KVStore) Fail(paymentHash lntypes.Hash, } // FetchPayment returns information about a payment from the database. -func (p *KVStore) FetchPayment(paymentHash lntypes.Hash) ( - *MPPayment, error) { +func (p *KVStore) FetchPayment(_ context.Context, + paymentHash lntypes.Hash) (*MPPayment, error) { var payment *MPPayment err := kvdb.View(p.db, func(tx kvdb.RTx) error { diff --git a/payments/db/kv_store_test.go b/payments/db/kv_store_test.go index cf5a9a9a0..910c1812d 100644 --- a/payments/db/kv_store_test.go +++ b/payments/db/kv_store_test.go @@ -409,6 +409,8 @@ func deletePayment(t *testing.T, db kvdb.Backend, paymentHash lntypes.Hash, func TestFetchPaymentWithSequenceNumber(t *testing.T) { paymentDB := NewKVTestDB(t) + ctx := t.Context() + // Generate a test payment which does not have duplicates. noDuplicates, _, err := genInfo(t) require.NoError(t, err) @@ -421,7 +423,7 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) { // Fetch the payment so we can get its sequence nr. noDuplicatesPayment, err := paymentDB.FetchPayment( - noDuplicates.PaymentIdentifier, + ctx, noDuplicates.PaymentIdentifier, ) require.NoError(t, err) @@ -437,7 +439,7 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) { // Fetch the payment so we can get its sequence nr. hasDuplicatesPayment, err := paymentDB.FetchPayment( - hasDuplicates.PaymentIdentifier, + ctx, hasDuplicates.PaymentIdentifier, ) require.NoError(t, err) @@ -749,7 +751,7 @@ func TestKVStoreQueryPaymentsDuplicates(t *testing.T) { // Immediately delete the payment with index 2. if i == 1 { pmt, err := paymentDB.FetchPayment( - info.PaymentIdentifier, + ctx, info.PaymentIdentifier, ) require.NoError(t, err) @@ -766,7 +768,7 @@ func TestKVStoreQueryPaymentsDuplicates(t *testing.T) { // duplicate payments will always be succeeded. if i == (nonDuplicatePayments - 1) { pmt, err := paymentDB.FetchPayment( - info.PaymentIdentifier, + ctx, info.PaymentIdentifier, ) require.NoError(t, err) diff --git a/payments/db/payment_test.go b/payments/db/payment_test.go index 4a9a69b0c..8954f8a5a 100644 --- a/payments/db/payment_test.go +++ b/payments/db/payment_test.go @@ -235,7 +235,9 @@ func assertPaymentInfo(t *testing.T, p DB, hash lntypes.Hash, t.Helper() - payment, err := p.FetchPayment(hash) + ctx := t.Context() + + payment, err := p.FetchPayment(ctx, hash) if err != nil { t.Fatal(err) } @@ -303,7 +305,9 @@ func assertDBPaymentstatus(t *testing.T, p DB, hash lntypes.Hash, t.Helper() - payment, err := p.FetchPayment(hash) + ctx := t.Context() + + payment, err := p.FetchPayment(ctx, hash) if errors.Is(err, ErrPaymentNotInitiated) { return } @@ -1796,6 +1800,8 @@ func TestSwitchDoubleSend(t *testing.T) { func TestSwitchFail(t *testing.T) { t.Parallel() + ctx := t.Context() + paymentDB, harness := NewTestDB(t) preimg, err := genPreimage(t) @@ -1834,7 +1840,7 @@ func TestSwitchFail(t *testing.T) { // Lookup the payment so we can get its old sequence number before it is // overwritten. - payment, err := paymentDB.FetchPayment(info.PaymentIdentifier) + payment, err := paymentDB.FetchPayment(ctx, info.PaymentIdentifier) require.NoError(t, err) // Sends the htlc again, which should succeed since the prior payment @@ -2609,7 +2615,7 @@ func TestQueryPayments(t *testing.T) { // Now delete the payment at index 1 (the second // payment). pmt, err := paymentDB.FetchPayment( - paymentInfos[1].PaymentIdentifier, + ctx, paymentInfos[1].PaymentIdentifier, ) require.NoError(t, err) @@ -2621,7 +2627,7 @@ func TestQueryPayments(t *testing.T) { // Verify the payment is deleted. _, err = paymentDB.FetchPayment( - paymentInfos[1].PaymentIdentifier, + ctx, paymentInfos[1].PaymentIdentifier, ) require.ErrorIs( t, err, ErrPaymentNotInitiated, diff --git a/payments/db/sql_store.go b/payments/db/sql_store.go index ef0d96c93..94233649e 100644 --- a/payments/db/sql_store.go +++ b/payments/db/sql_store.go @@ -922,8 +922,8 @@ func fetchPaymentByHash(ctx context.Context, db SQLQueries, // Returns ErrPaymentNotInitiated if no payment with the given hash exists. // // This is part of the DB interface. -func (s *SQLStore) FetchPayment(paymentHash lntypes.Hash) (*MPPayment, error) { - ctx := context.TODO() +func (s *SQLStore) FetchPayment(ctx context.Context, + paymentHash lntypes.Hash) (*MPPayment, error) { var mpPayment *MPPayment diff --git a/payments/db/test_kvdb.go b/payments/db/test_kvdb.go index ed1710b14..c2de0b43f 100644 --- a/payments/db/test_kvdb.go +++ b/payments/db/test_kvdb.go @@ -57,9 +57,11 @@ func (h *kvTestHarness) AssertPaymentIndex(t *testing.T, t.Helper() + ctx := t.Context() + // Lookup the payment so that we have its sequence number and check // that it has correctly been indexed in the payment indexes bucket. - pmt, err := h.db.FetchPayment(expectedHash) + pmt, err := h.db.FetchPayment(ctx, expectedHash) require.NoError(t, err) hash, err := h.fetchPaymentIndexEntry(t, pmt.SequenceNum) diff --git a/routing/control_tower.go b/routing/control_tower.go index 2b9e7dd9d..310289487 100644 --- a/routing/control_tower.go +++ b/routing/control_tower.go @@ -1,6 +1,7 @@ package routing import ( + "context" "sync" "github.com/lightningnetwork/lnd/lntypes" @@ -52,7 +53,8 @@ type ControlTower interface { // FetchPayment fetches the payment corresponding to the given payment // hash. - FetchPayment(paymentHash lntypes.Hash) (paymentsdb.DBMPPayment, error) + FetchPayment(ctx context.Context, + paymentHash lntypes.Hash) (paymentsdb.DBMPPayment, error) // FailPayment transitions a payment into the Failed state, and records // the ultimate reason the payment failed. Note that this should only @@ -164,6 +166,8 @@ func NewControlTower(db paymentsdb.DB) ControlTower { func (p *controlTower) InitPayment(paymentHash lntypes.Hash, info *paymentsdb.PaymentCreationInfo) error { + ctx := context.TODO() + err := p.db.InitPayment(paymentHash, info) if err != nil { return err @@ -174,7 +178,7 @@ func (p *controlTower) InitPayment(paymentHash lntypes.Hash, p.paymentsMtx.Lock(paymentHash) defer p.paymentsMtx.Unlock(paymentHash) - payment, err := p.db.FetchPayment(paymentHash) + payment, err := p.db.FetchPayment(ctx, paymentHash) if err != nil { return err } @@ -250,10 +254,11 @@ func (p *controlTower) FailAttempt(paymentHash lntypes.Hash, } // FetchPayment fetches the payment corresponding to the given payment hash. -func (p *controlTower) FetchPayment(paymentHash lntypes.Hash) ( +func (p *controlTower) FetchPayment(ctx context.Context, + paymentHash lntypes.Hash) ( paymentsdb.DBMPPayment, error) { - return p.db.FetchPayment(paymentHash) + return p.db.FetchPayment(ctx, paymentHash) } // FailPayment transitions a payment into the Failed state, and records the @@ -293,12 +298,14 @@ func (p *controlTower) FetchInFlightPayments() ([]*paymentsdb.MPPayment, func (p *controlTower) SubscribePayment(paymentHash lntypes.Hash) ( ControlTowerSubscriber, error) { + ctx := context.TODO() + // Take lock before querying the db to prevent missing or duplicating an // update. p.paymentsMtx.Lock(paymentHash) defer p.paymentsMtx.Unlock(paymentHash) - payment, err := p.db.FetchPayment(paymentHash) + payment, err := p.db.FetchPayment(ctx, paymentHash) if err != nil { return nil, err } diff --git a/routing/mock_test.go b/routing/mock_test.go index 19a76ee90..556601ecd 100644 --- a/routing/mock_test.go +++ b/routing/mock_test.go @@ -1,6 +1,7 @@ package routing import ( + "context" "errors" "fmt" "sync" @@ -509,8 +510,8 @@ func (m *mockControlTowerOld) FailPayment(phash lntypes.Hash, return nil } -func (m *mockControlTowerOld) FetchPayment(phash lntypes.Hash) ( - paymentsdb.DBMPPayment, error) { +func (m *mockControlTowerOld) FetchPayment(_ context.Context, + phash lntypes.Hash) (paymentsdb.DBMPPayment, error) { m.Lock() defer m.Unlock() @@ -786,8 +787,8 @@ func (m *mockControlTower) FailPayment(phash lntypes.Hash, return args.Error(0) } -func (m *mockControlTower) FetchPayment(phash lntypes.Hash) ( - paymentsdb.DBMPPayment, error) { +func (m *mockControlTower) FetchPayment(_ context.Context, + phash lntypes.Hash) (paymentsdb.DBMPPayment, error) { args := m.Called(phash) diff --git a/routing/payment_lifecycle.go b/routing/payment_lifecycle.go index 8353cba15..4eb78c8e2 100644 --- a/routing/payment_lifecycle.go +++ b/routing/payment_lifecycle.go @@ -1114,7 +1114,9 @@ func (p *paymentLifecycle) patchLegacyPaymentHash( func (p *paymentLifecycle) reloadInflightAttempts() (paymentsdb.DBMPPayment, error) { - payment, err := p.router.cfg.Control.FetchPayment(p.identifier) + ctx := context.TODO() + + payment, err := p.router.cfg.Control.FetchPayment(ctx, p.identifier) if err != nil { return nil, err } @@ -1139,8 +1141,10 @@ func (p *paymentLifecycle) reloadInflightAttempts() (paymentsdb.DBMPPayment, func (p *paymentLifecycle) reloadPayment() (paymentsdb.DBMPPayment, *paymentsdb.MPPaymentState, error) { + ctx := context.TODO() + // Read the db to get the latest state of the payment. - payment, err := p.router.cfg.Control.FetchPayment(p.identifier) + payment, err := p.router.cfg.Control.FetchPayment(ctx, p.identifier) if err != nil { return nil, nil, err } diff --git a/routing/router.go b/routing/router.go index 19df5b921..8aa5acf5a 100644 --- a/routing/router.go +++ b/routing/router.go @@ -1064,13 +1064,15 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route, firstHopCustomRecords lnwire.CustomRecords) (*paymentsdb.HTLCAttempt, error) { + ctx := context.TODO() + // Helper function to fail a payment. It makes sure the payment is only // failed once so that the failure reason is not overwritten. failPayment := func(paymentIdentifier lntypes.Hash, reason paymentsdb.FailureReason) error { payment, fetchErr := r.cfg.Control.FetchPayment( - paymentIdentifier, + ctx, paymentIdentifier, ) if fetchErr != nil { return fetchErr diff --git a/routing/router_test.go b/routing/router_test.go index 115c02c2c..733943272 100644 --- a/routing/router_test.go +++ b/routing/router_test.go @@ -1097,7 +1097,9 @@ func TestSendPaymentErrorPathPruning(t *testing.T) { require.Equal(t, paymentsdb.FailureReasonNoRoute, err) // Inspect the two attempts that were made before the payment failed. - p, err := ctx.router.cfg.Control.FetchPayment(*payment.paymentHash) + p, err := ctx.router.cfg.Control.FetchPayment( + t.Context(), *payment.paymentHash, + ) require.NoError(t, err) htlcs := p.GetHTLCs()