multi: thread context through RegisterAttempt method

This commit is contained in:
ziggie committed 2026-02-25 08:02:24 +01:00
1 parent 60026494b1
commit 8bb7150dc2
9 files changed
+70 -42

No files matched your search

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