diff --git a/payments/db/interface.go b/payments/db/interface.go index 5368d53d3..616906c73 100644 --- a/payments/db/interface.go +++ b/payments/db/interface.go @@ -25,7 +25,7 @@ type PaymentReader interface { paymentHash lntypes.Hash) (*MPPayment, error) // FetchInFlightPayments returns all payments with status InFlight. - FetchInFlightPayments() ([]*MPPayment, error) + FetchInFlightPayments(ctx context.Context) ([]*MPPayment, error) } // PaymentWriter represents the interface to write operations to the payments diff --git a/payments/db/kv_store.go b/payments/db/kv_store.go index 86b37edcf..1b48cac67 100644 --- a/payments/db/kv_store.go +++ b/payments/db/kv_store.go @@ -741,7 +741,9 @@ func fetchPaymentStatus(bucket kvdb.RBucket) (PaymentStatus, error) { } // FetchInFlightPayments returns all payments with status InFlight. -func (p *KVStore) FetchInFlightPayments() ([]*MPPayment, error) { +func (p *KVStore) FetchInFlightPayments(_ context.Context) ([]*MPPayment, + error) { + var ( inFlights []*MPPayment start = time.Now() diff --git a/payments/db/payment_test.go b/payments/db/payment_test.go index 8954f8a5a..cc1c6e919 100644 --- a/payments/db/payment_test.go +++ b/payments/db/payment_test.go @@ -2740,6 +2740,8 @@ func TestQueryPayments(t *testing.T) { func TestFetchInFlightPayments(t *testing.T) { t.Parallel() + ctx := t.Context() + paymentDB, _ := NewTestDB(t) // Register payments with different statuses: @@ -2765,7 +2767,7 @@ func TestFetchInFlightPayments(t *testing.T) { assertDBPayments(t, paymentDB, payments) // Fetch in-flight payments. - inFlightPayments, err := paymentDB.FetchInFlightPayments() + inFlightPayments, err := paymentDB.FetchInFlightPayments(ctx) require.NoError(t, err) // We should only get the two in-flight payments. @@ -2795,7 +2797,7 @@ func TestFetchInFlightPayments(t *testing.T) { require.NoError(t, err) // Fetch in-flight payments again. - inFlightPayments, err = paymentDB.FetchInFlightPayments() + inFlightPayments, err = paymentDB.FetchInFlightPayments(ctx) require.NoError(t, err) // We should now only get one in-flight payment. @@ -2812,6 +2814,8 @@ func TestFetchInFlightPayments(t *testing.T) { func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) { t.Parallel() + ctx := t.Context() + paymentDB, _ := NewTestDB(t) preimg, err := genPreimage(t) @@ -2843,7 +2847,7 @@ func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) { require.NoError(t, err) // Both attempts are in-flight. Fetch in-flight payments. - inFlightPayments, err := paymentDB.FetchInFlightPayments() + inFlightPayments, err := paymentDB.FetchInFlightPayments(ctx) require.NoError(t, err) // We should only get one payment even though it has 2 in-flight diff --git a/payments/db/sql_store.go b/payments/db/sql_store.go index 94233649e..7fad7cc7e 100644 --- a/payments/db/sql_store.go +++ b/payments/db/sql_store.go @@ -972,11 +972,9 @@ func (s *SQLStore) FetchPayment(ctx context.Context, // While inflight payments are typically a small subset, this would improve // memory efficiency for nodes with unusually high numbers of concurrent // payments and would better leverage the existing pagination infrastructure. -func (s *SQLStore) FetchInFlightPayments() ([]*MPPayment, +func (s *SQLStore) FetchInFlightPayments(ctx context.Context) ([]*MPPayment, error) { - ctx := context.TODO() - var mpPayments []*MPPayment err := s.db.ExecTx(ctx, sqldb.ReadTxOpt(), func(db SQLQueries) error { diff --git a/routing/control_tower.go b/routing/control_tower.go index 310289487..718dca3ff 100644 --- a/routing/control_tower.go +++ b/routing/control_tower.go @@ -67,7 +67,8 @@ type ControlTower interface { FailPayment(lntypes.Hash, paymentsdb.FailureReason) error // FetchInFlightPayments returns all payments with status InFlight. - FetchInFlightPayments() ([]*paymentsdb.MPPayment, error) + FetchInFlightPayments(ctx context.Context) ([]*paymentsdb.MPPayment, + error) // SubscribePayment subscribes to updates for the payment with the given // hash. A first update with the current state of the payment is always @@ -286,10 +287,10 @@ func (p *controlTower) FailPayment(paymentHash lntypes.Hash, } // FetchInFlightPayments returns all payments with status InFlight. -func (p *controlTower) FetchInFlightPayments() ([]*paymentsdb.MPPayment, - error) { +func (p *controlTower) FetchInFlightPayments( + ctx context.Context) ([]*paymentsdb.MPPayment, error) { - return p.db.FetchInFlightPayments() + return p.db.FetchInFlightPayments(ctx) } // SubscribePayment subscribes to updates for the payment with the given hash. A @@ -342,6 +343,8 @@ func (p *controlTower) SubscribePayment(paymentHash lntypes.Hash) ( func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) { subscriber := newControlTowerSubscriber() + ctx := context.TODO() + // Add the subscriber to the list before fetching in-flight payments, so // no events are missed. If a payment attempt update occurs after // appending and before fetching in-flight payments, an out-of-order @@ -353,7 +356,7 @@ func (p *controlTower) SubscribeAllPayments() (ControlTowerSubscriber, error) { p.subscribersMtx.Unlock() log.Debugf("Scanning for inflight payments") - inflightPayments, err := p.db.FetchInFlightPayments() + inflightPayments, err := p.db.FetchInFlightPayments(ctx) if err != nil { return nil, err } diff --git a/routing/mock_test.go b/routing/mock_test.go index 556601ecd..b30627165 100644 --- a/routing/mock_test.go +++ b/routing/mock_test.go @@ -546,7 +546,7 @@ func (m *mockControlTowerOld) fetchPayment(phash lntypes.Hash) ( return mp, nil } -func (m *mockControlTowerOld) FetchInFlightPayments() ( +func (m *mockControlTowerOld) FetchInFlightPayments(_ context.Context) ( []*paymentsdb.MPPayment, error) { if m.fetchInFlight != nil { @@ -801,7 +801,7 @@ func (m *mockControlTower) FetchPayment(_ context.Context, return payment, args.Error(1) } -func (m *mockControlTower) FetchInFlightPayments() ( +func (m *mockControlTower) FetchInFlightPayments(_ context.Context) ( []*paymentsdb.MPPayment, error) { args := m.Called() diff --git a/routing/router.go b/routing/router.go index 8aa5acf5a..2aa5745e9 100644 --- a/routing/router.go +++ b/routing/router.go @@ -1417,9 +1417,11 @@ func (r *ChannelRouter) BuildRoute(amt fn.Option[lnwire.MilliSatoshi], // resumePayments fetches inflight payments and resumes their payment // lifecycles. func (r *ChannelRouter) resumePayments() error { + ctx := context.TODO() + // Get all payments that are inflight. log.Debugf("Scanning for inflight payments") - payments, err := r.cfg.Control.FetchInFlightPayments() + payments, err := r.cfg.Control.FetchInFlightPayments(ctx) if err != nil { return err }