multi: thread context through InitPayment
This commit is contained in:
9 files changed
+46
-39
No files matched your search
@@ -61,7 +61,7 @@ type PaymentControl interface {
|
||||
// exists in the database before creating a new payment. However, it
|
||||
// should allow the user making a subsequent payment if the payment is
|
||||
// in a Failed state.
|
||||
InitPayment(lntypes.Hash, *PaymentCreationInfo) error
|
||||
InitPayment(context.Context, lntypes.Hash, *PaymentCreationInfo) error
|
||||
|
||||
// RegisterAttempt atomically records the provided HTLCAttemptInfo.
|
||||
//
|
||||
|
||||
@@ -186,7 +186,7 @@ func initKVStore(db kvdb.Backend) error {
|
||||
// making sure it does not already exist as an in-flight payment. When this
|
||||
// method returns successfully, the payment is guaranteed to be in the InFlight
|
||||
// state.
|
||||
func (p *KVStore) InitPayment(paymentHash lntypes.Hash,
|
||||
func (p *KVStore) InitPayment(_ context.Context, paymentHash lntypes.Hash,
|
||||
info *PaymentCreationInfo) error {
|
||||
|
||||
// Obtain a new sequence number for this payment. This is used
|
||||
|
||||
@@ -80,7 +80,7 @@ func TestKVStoreDeleteNonInFlight(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
// Sends base htlc message which initiate StatusInFlight.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
if err != nil {
|
||||
t.Fatalf("unable to send htlc message: %v", err)
|
||||
}
|
||||
@@ -417,7 +417,7 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) {
|
||||
|
||||
// Create a new payment entry in the database.
|
||||
err = paymentDB.InitPayment(
|
||||
noDuplicates.PaymentIdentifier, noDuplicates,
|
||||
ctx, noDuplicates.PaymentIdentifier, noDuplicates,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -433,7 +433,7 @@ func TestFetchPaymentWithSequenceNumber(t *testing.T) {
|
||||
|
||||
// Create a new payment entry in the database.
|
||||
err = paymentDB.InitPayment(
|
||||
hasDuplicates.PaymentIdentifier, hasDuplicates,
|
||||
ctx, hasDuplicates.PaymentIdentifier, hasDuplicates,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -744,7 +744,7 @@ func TestKVStoreQueryPaymentsDuplicates(t *testing.T) {
|
||||
|
||||
// Create a new payment entry in the database.
|
||||
err = paymentDB.InitPayment(
|
||||
info.PaymentIdentifier, info,
|
||||
ctx, info.PaymentIdentifier, info,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
|
||||
+21
-13
@@ -125,6 +125,8 @@ type payment struct {
|
||||
func createTestPayments(t *testing.T, p DB, payments []*payment) {
|
||||
t.Helper()
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
attemptID := uint64(0)
|
||||
|
||||
for i := 0; i < len(payments); i++ {
|
||||
@@ -145,7 +147,7 @@ func createTestPayments(t *testing.T, p DB, payments []*payment) {
|
||||
attemptID++
|
||||
|
||||
// Init the payment.
|
||||
err = p.InitPayment(info.PaymentIdentifier, info)
|
||||
err = p.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err, "unable to send htlc message")
|
||||
|
||||
// Register and fail the first attempt for all payments.
|
||||
@@ -559,6 +561,8 @@ func testDeleteFailedAttempts(t *testing.T, keepFailedPaymentAttempts bool) {
|
||||
func TestMPPRecordValidation(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
paymentDB, _ := NewTestDB(t)
|
||||
|
||||
preimg, err := genPreimage(t)
|
||||
@@ -575,7 +579,7 @@ func TestMPPRecordValidation(t *testing.T) {
|
||||
require.NoError(t, err, "unable to generate htlc message")
|
||||
|
||||
// Init the payment.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err, "unable to send htlc message")
|
||||
|
||||
// Create three unique attempts we'll use for the test, and
|
||||
@@ -633,7 +637,7 @@ func TestMPPRecordValidation(t *testing.T) {
|
||||
|
||||
require.NoError(t, err, "unable to generate htlc message")
|
||||
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err, "unable to send htlc message")
|
||||
|
||||
attempt.Route.FinalHop().MPP = nil
|
||||
@@ -1722,6 +1726,8 @@ func TestDeletePayments(t *testing.T) {
|
||||
func TestSwitchDoubleSend(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
paymentDB, harness := NewTestDB(t)
|
||||
|
||||
preimg, err := genPreimage(t)
|
||||
@@ -1734,7 +1740,7 @@ func TestSwitchDoubleSend(t *testing.T) {
|
||||
|
||||
// Sends base htlc message which initiate base status and move it to
|
||||
// StatusInFlight and verifies that it was changed.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err, "unable to send htlc message")
|
||||
|
||||
harness.AssertPaymentIndex(t, info.PaymentIdentifier)
|
||||
@@ -1748,7 +1754,7 @@ func TestSwitchDoubleSend(t *testing.T) {
|
||||
// Try to initiate double sending of htlc message with the same
|
||||
// payment hash, should result in error indicating that payment has
|
||||
// already been sent.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.ErrorIs(t, err, ErrPaymentExists)
|
||||
|
||||
// Record an attempt.
|
||||
@@ -1766,7 +1772,7 @@ func TestSwitchDoubleSend(t *testing.T) {
|
||||
)
|
||||
|
||||
// Sends base htlc message which initiate StatusInFlight.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
if !errors.Is(err, ErrPaymentInFlight) {
|
||||
t.Fatalf("payment control wrong behaviour: " +
|
||||
"double sending must trigger ErrPaymentInFlight error")
|
||||
@@ -1789,7 +1795,7 @@ func TestSwitchDoubleSend(t *testing.T) {
|
||||
t, paymentDB, info.PaymentIdentifier, info, nil, htlc,
|
||||
)
|
||||
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
if !errors.Is(err, ErrAlreadyPaid) {
|
||||
t.Fatalf("unable to send htlc message: %v", err)
|
||||
}
|
||||
@@ -1813,7 +1819,7 @@ func TestSwitchFail(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
|
||||
// Sends base htlc message which initiate StatusInFlight.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err, "unable to send htlc message")
|
||||
|
||||
harness.AssertPaymentIndex(t, info.PaymentIdentifier)
|
||||
@@ -1845,7 +1851,7 @@ func TestSwitchFail(t *testing.T) {
|
||||
|
||||
// Sends the htlc again, which should succeed since the prior payment
|
||||
// failed.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err, "unable to send htlc message")
|
||||
|
||||
// Check that our index has been updated, and the old index has been
|
||||
@@ -1940,7 +1946,7 @@ func TestSwitchFail(t *testing.T) {
|
||||
|
||||
// Attempt a final payment, which should now fail since the prior
|
||||
// payment succeed.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
if !errors.Is(err, ErrAlreadyPaid) {
|
||||
t.Fatalf("unable to send htlc message: %v", err)
|
||||
}
|
||||
@@ -1951,6 +1957,8 @@ func TestSwitchFail(t *testing.T) {
|
||||
func TestMultiShard(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := t.Context()
|
||||
|
||||
// We will register three HTLC attempts, and always fail the second
|
||||
// one. We'll generate all combinations of settling/failing the first
|
||||
// and third HTLC, and assert that the payment status end up as we
|
||||
@@ -1977,7 +1985,7 @@ func TestMultiShard(t *testing.T) {
|
||||
info := genPaymentCreationInfo(t, rhash)
|
||||
|
||||
// Init the payment, moving it to the StatusInFlight state.
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err)
|
||||
|
||||
harness.AssertPaymentIndex(t, info.PaymentIdentifier)
|
||||
@@ -2607,7 +2615,7 @@ func TestQueryPayments(t *testing.T) {
|
||||
|
||||
// Create a new payment entry in the database.
|
||||
err = paymentDB.InitPayment(
|
||||
info.PaymentIdentifier, info,
|
||||
ctx, info.PaymentIdentifier, info,
|
||||
)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
@@ -2826,7 +2834,7 @@ func TestFetchInFlightPaymentsMultipleAttempts(t *testing.T) {
|
||||
|
||||
// Init payment with double the amount to allow two attempts.
|
||||
info.Value *= 2
|
||||
err = paymentDB.InitPayment(info.PaymentIdentifier, info)
|
||||
err = paymentDB.InitPayment(ctx, info.PaymentIdentifier, info)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Register two attempts for the same payment.
|
||||
|
||||
@@ -1265,11 +1265,9 @@ func (s *SQLStore) DeletePayment(ctx context.Context, paymentHash lntypes.Hash,
|
||||
// This method is part of the PaymentControl interface, which is embedded in
|
||||
// the PaymentWriter interface and ultimately the DB interface, representing
|
||||
// the first step in the payment lifecycle control flow.
|
||||
func (s *SQLStore) InitPayment(paymentHash lntypes.Hash,
|
||||
func (s *SQLStore) InitPayment(ctx context.Context, paymentHash lntypes.Hash,
|
||||
paymentCreationInfo *PaymentCreationInfo) error {
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
// Create the payment in the database.
|
||||
err := s.db.ExecTx(ctx, sqldb.WriteTxOpt(), func(db SQLQueries) error {
|
||||
existingPayment, err := db.FetchPayment(ctx, paymentHash[:])
|
||||
|
||||
@@ -20,7 +20,8 @@ type ControlTower interface {
|
||||
// also notifies subscribers of the payment creation.
|
||||
//
|
||||
// NOTE: Subscribers should be notified by the new state of the payment.
|
||||
InitPayment(lntypes.Hash, *paymentsdb.PaymentCreationInfo) error
|
||||
InitPayment(context.Context, lntypes.Hash,
|
||||
*paymentsdb.PaymentCreationInfo) error
|
||||
|
||||
// DeleteFailedAttempts removes all failed HTLCs from the db. It should
|
||||
// be called for a given payment whenever all inflight htlcs are
|
||||
@@ -164,12 +165,10 @@ func NewControlTower(db paymentsdb.DB) ControlTower {
|
||||
// making sure it does not already exist as an in-flight payment. Then this
|
||||
// method returns successfully, the payment is guaranteed to be in the
|
||||
// Initiated state.
|
||||
func (p *controlTower) InitPayment(paymentHash lntypes.Hash,
|
||||
info *paymentsdb.PaymentCreationInfo) error {
|
||||
func (p *controlTower) InitPayment(ctx context.Context,
|
||||
paymentHash lntypes.Hash, info *paymentsdb.PaymentCreationInfo) error {
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
err := p.db.InitPayment(paymentHash, info)
|
||||
err := p.db.InitPayment(ctx, paymentHash, info)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -81,7 +81,7 @@ func TestControlTowerSubscribeSuccess(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = pControl.InitPayment(info.PaymentIdentifier, info)
|
||||
err = pControl.InitPayment(t.Context(), info.PaymentIdentifier, info)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -212,7 +212,7 @@ func TestKVStoreSubscribeAllSuccess(t *testing.T) {
|
||||
info1, attempt1, preimg1, err := genInfo()
|
||||
require.NoError(t, err)
|
||||
|
||||
err = pControl.InitPayment(info1.PaymentIdentifier, info1)
|
||||
err = pControl.InitPayment(t.Context(), info1.PaymentIdentifier, info1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Subscription should succeed and immediately report the Initiated
|
||||
@@ -228,7 +228,7 @@ func TestKVStoreSubscribeAllSuccess(t *testing.T) {
|
||||
info2, attempt2, preimg2, err := genInfo()
|
||||
require.NoError(t, err)
|
||||
|
||||
err = pControl.InitPayment(info2.PaymentIdentifier, info2)
|
||||
err = pControl.InitPayment(t.Context(), info2.PaymentIdentifier, info2)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Register an attempt on the second payment.
|
||||
@@ -337,7 +337,7 @@ func TestKVStoreSubscribeAllImmediate(t *testing.T) {
|
||||
info, attempt, _, err := genInfo()
|
||||
require.NoError(t, err)
|
||||
|
||||
err = pControl.InitPayment(info.PaymentIdentifier, info)
|
||||
err = pControl.InitPayment(t.Context(), info.PaymentIdentifier, info)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Register a payment update.
|
||||
@@ -392,7 +392,7 @@ func TestKVStoreUnsubscribeSuccess(t *testing.T) {
|
||||
info, attempt, _, err := genInfo()
|
||||
require.NoError(t, err)
|
||||
|
||||
err = pControl.InitPayment(info.PaymentIdentifier, info)
|
||||
err = pControl.InitPayment(t.Context(), info.PaymentIdentifier, info)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Assert all subscriptions receive the update.
|
||||
@@ -465,7 +465,7 @@ func testKVStoreSubscribeFail(t *testing.T, registerAttempt,
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = pControl.InitPayment(info.PaymentIdentifier, info)
|
||||
err = pControl.InitPayment(t.Context(), info.PaymentIdentifier, info)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
@@ -297,8 +297,8 @@ func makeMockControlTower() *mockControlTowerOld {
|
||||
}
|
||||
}
|
||||
|
||||
func (m *mockControlTowerOld) InitPayment(phash lntypes.Hash,
|
||||
c *paymentsdb.PaymentCreationInfo) error {
|
||||
func (m *mockControlTowerOld) InitPayment(_ context.Context,
|
||||
phash lntypes.Hash, c *paymentsdb.PaymentCreationInfo) error {
|
||||
|
||||
if m.init != nil {
|
||||
m.init <- initArgs{c}
|
||||
@@ -734,7 +734,7 @@ type mockControlTower struct {
|
||||
|
||||
var _ ControlTower = (*mockControlTower)(nil)
|
||||
|
||||
func (m *mockControlTower) InitPayment(phash lntypes.Hash,
|
||||
func (m *mockControlTower) InitPayment(_ context.Context, phash lntypes.Hash,
|
||||
c *paymentsdb.PaymentCreationInfo) error {
|
||||
|
||||
args := m.Called(phash, c)
|
||||
|
||||
+4
-2
@@ -967,6 +967,8 @@ func spewPayment(payment *LightningPayment) lnutils.LogClosure {
|
||||
func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
|
||||
PaymentSession, shards.ShardTracker, error) {
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
// Assemble any custom data we want to send to the first hop only.
|
||||
var firstHopData fn.Option[tlv.Blob]
|
||||
if len(payment.FirstHopCustomRecords) > 0 {
|
||||
@@ -1026,7 +1028,7 @@ func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
|
||||
)
|
||||
}
|
||||
|
||||
err = r.cfg.Control.InitPayment(payment.Identifier(), info)
|
||||
err = r.cfg.Control.InitPayment(ctx, payment.Identifier(), info)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
@@ -1131,7 +1133,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
||||
FirstHopCustomRecords: firstHopCustomRecords,
|
||||
}
|
||||
|
||||
err := r.cfg.Control.InitPayment(paymentIdentifier, info)
|
||||
err := r.cfg.Control.InitPayment(ctx, paymentIdentifier, info)
|
||||
switch {
|
||||
// If this is an MPP attempt and the hash is already registered with
|
||||
// the database, we can go on to launch the shard.
|
||||
|
||||
Reference in new issue
Block a user