multi: thread context through FetchPayment
This commit is contained in:
11 files changed
+55
-28
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
+3
-1
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user