mirror/lnd
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

multi: thread context through payment lifecyle functions

Browse Source
This commit is contained in:
ziggie committed 2026-02-25 08:02:25 +01:00
1 parent dfe0c43c7d
commit 8038dc0054
6 files changed
+91 -54
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

lnrpc/routerrpc/router_server.go
+2 -2
View File
@@ -1088,11 +1088,11 @@ func (s *Server) SendToRouteV2(ctx context.Context,
// db.
if req.SkipTempErr {
attempt, err = s.cfg.Router.SendToRouteSkipTempErr(
hash, route, firstHopRecords,
ctx, hash, route, firstHopRecords,
)
} else {
attempt, err = s.cfg.Router.SendToRoute(
hash, route, firstHopRecords,
ctx, hash, route, firstHopRecords,
)
}
if attempt != nil {
routing/payment_lifecycle.go
+18 -12
View File
@@ -128,7 +128,7 @@ const (
// results is sent back. then process its result here. When there's no need to
// wait for results, the method will exit with `stepExit` such that the payment
// lifecycle loop will terminate.
func (p *paymentLifecycle) decideNextStep(
func (p *paymentLifecycle) decideNextStep(ctx context.Context,
payment paymentsdb.DBMPPayment) (stateStep, error) {
// Check whether we could make new HTLC attempts.
@@ -168,7 +168,7 @@ func (p *paymentLifecycle) decideNextStep(
// stepSkip and move to the next lifecycle iteration, which will
// refresh the payment and wait for the next attempt result, if
// any.
_, err := p.handleAttemptResult(r.attempt, r.result)
_, err := p.handleAttemptResult(ctx, r.attempt, r.result)
// We would only get a DB-related error here, which will cause
// us to abort the payment flow.
@@ -192,6 +192,13 @@ func (p *paymentLifecycle) resumePayment(ctx context.Context) ([32]byte,
// We need to make sure we can still do db operations after the context
// is cancelled.
//
// TODO(ziggie): This is a workaround to avoid a greater refactor of the
// payment lifecycle. We can currently not rely on the parent context
// because this method is also collecting the results of inflight HTLCs
// after the context is cancelled. So we need to make sure we only use
// the current context to stop creating new attempts but use this
// cleanupCtx to do all the db operations.
cleanupCtx := context.WithoutCancel(ctx)
// When the payment lifecycle loop exits, we make sure to signal any
@@ -264,7 +271,7 @@ lifecycle:
//
// Now decide the next step of the current lifecycle.
step, err := p.decideNextStep(payment)
step, err := p.decideNextStep(cleanupCtx, payment)
if err != nil {
return exitWithErr(err)
}
@@ -307,7 +314,9 @@ lifecycle:
log.Tracef("Found route: %s", lnutils.SpewLogClosure(rt.Hops))
// We found a route to try, create a new HTLC attempt to try.
attempt, err := p.registerAttempt(rt, ps.RemainingAmt)
attempt, err := p.registerAttempt(
cleanupCtx, rt, ps.RemainingAmt,
)
if err != nil {
return exitWithErr(err)
}
@@ -596,11 +605,9 @@ func (p *paymentLifecycle) collectResult(
// registerAttempt is responsible for creating and saving an HTLC attempt in db
// by using the route info provided. The `remainingAmt` is used to decide
// whether this is the last attempt.
func (p *paymentLifecycle) registerAttempt(rt *route.Route,
func (p *paymentLifecycle) registerAttempt(ctx context.Context, rt *route.Route,
remainingAmt lnwire.MilliSatoshi) (*paymentsdb.HTLCAttempt, error) {
ctx := context.TODO()
// If this route will consume the last remaining amount to send
// to the receiver, this will be our last shard (for now).
isLastAttempt := rt.ReceiverAmt() == remainingAmt
@@ -1184,11 +1191,10 @@ func (p *paymentLifecycle) reloadPayment() (paymentsdb.DBMPPayment,
// handleAttemptResult processes the result of an HTLC attempt returned from
// the htlcswitch.
func (p *paymentLifecycle) handleAttemptResult(attempt *paymentsdb.HTLCAttempt,
func (p *paymentLifecycle) handleAttemptResult(ctx context.Context,
attempt *paymentsdb.HTLCAttempt,
result *htlcswitch.PaymentResult) (*attemptResult, error) {
ctx := context.TODO()
// If the result has an error, we need to further process it by failing
// the attempt and maybe fail the payment.
if result.Error != nil {
@@ -1235,7 +1241,7 @@ func (p *paymentLifecycle) handleAttemptResult(attempt *paymentsdb.HTLCAttempt,
// available from the Switch, then records the attempt outcome with the control
// tower. An attemptResult is returned, indicating the final outcome of this
// HTLC attempt.
func (p *paymentLifecycle) collectAndHandleResult(
func (p *paymentLifecycle) collectAndHandleResult(ctx context.Context,
attempt *paymentsdb.HTLCAttempt) (*attemptResult, error) {
result, err := p.collectResult(attempt)
@@ -1243,5 +1249,5 @@ func (p *paymentLifecycle) collectAndHandleResult(
return nil, err
}
return p.handleAttemptResult(attempt, result)
return p.handleAttemptResult(ctx, attempt, result)
}
routing/payment_lifecycle_test.go
+18 -14
View File
@@ -599,7 +599,7 @@ func TestDecideNextStep(t *testing.T) {
// Once the setup is finished, run the test cases.
t.Run(tc.name, func(t *testing.T) {
step, err := p.decideNextStep(payment)
step, err := p.decideNextStep(t.Context(), payment)
require.Equal(t, tc.expectedStep, step)
require.ErrorIs(t, tc.expectedErr, err)
})
@@ -628,7 +628,7 @@ func TestDecideNextStepOnRouterQuit(t *testing.T) {
close(p.router.quit)
// Call the method under test.
step, err := p.decideNextStep(payment)
step, err := p.decideNextStep(t.Context(), payment)
// We expect stepExit and an error to be returned.
require.Equal(t, stepExit, step)
@@ -657,7 +657,7 @@ func TestDecideNextStepOnLifecycleQuit(t *testing.T) {
close(p.quit)
// Call the method under test.
step, err := p.decideNextStep(payment)
step, err := p.decideNextStep(t.Context(), payment)
// We expect stepExit and an error to be returned.
require.Equal(t, stepExit, step)
@@ -716,7 +716,7 @@ func TestDecideNextStepHandleAttemptResultSucceed(t *testing.T) {
mock.Anything).Return(attempt, nil).Once()
// Call the method under test.
step, err := p.decideNextStep(payment)
step, err := p.decideNextStep(t.Context(), payment)
// We expect stepSkip and no error to be returned.
require.Equal(t, stepSkip, step)
@@ -774,7 +774,7 @@ func TestDecideNextStepHandleAttemptResultFail(t *testing.T) {
mock.Anything).Return(attempt, errDummy).Once()
// Call the method under test.
step, err := p.decideNextStep(payment)
step, err := p.decideNextStep(t.Context(), payment)
// We expect stepExit and the above error to be returned.
require.Equal(t, stepExit, step)
@@ -1467,7 +1467,7 @@ func TestCollectResultExitOnErr(t *testing.T) {
m.clock.On("Now").Return(time.Now())
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.ErrorIs(t, err, errDummy, "expected dummy error")
require.Nil(t, result, "expected nil attempt")
}
@@ -1513,7 +1513,7 @@ func TestCollectResultExitOnResultErr(t *testing.T) {
m.clock.On("Now").Return(time.Now())
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.ErrorIs(t, err, errDummy, "expected dummy error")
require.Nil(t, result, "expected nil attempt")
}
@@ -1539,7 +1539,7 @@ func TestCollectResultExitOnSwitchQuit(t *testing.T) {
})
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.ErrorIs(t, err, htlcswitch.ErrSwitchExiting,
"expected switch exit")
require.Nil(t, result, "expected nil attempt")
@@ -1566,7 +1566,7 @@ func TestCollectResultExitOnRouterQuit(t *testing.T) {
})
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.ErrorIs(t, err, ErrRouterShuttingDown, "expected router exit")
require.Nil(t, result, "expected nil attempt")
}
@@ -1592,7 +1592,7 @@ func TestCollectResultExitOnLifecycleQuit(t *testing.T) {
})
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.ErrorIs(t, err, ErrPaymentLifecycleExiting,
"expected lifecycle exit")
require.Nil(t, result, "expected nil attempt")
@@ -1636,7 +1636,7 @@ func TestCollectResultExitOnSettleErr(t *testing.T) {
m.clock.On("Now").Return(time.Now())
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.ErrorIs(t, err, errDummy, "expected settle error")
require.Nil(t, result, "expected nil attempt")
}
@@ -1678,7 +1678,7 @@ func TestCollectResultSuccess(t *testing.T) {
m.clock.On("Now").Return(time.Now())
// Now call the method under test.
result, err := p.collectAndHandleResult(attempt)
result, err := p.collectAndHandleResult(t.Context(), attempt)
require.NoError(t, err, "expected no error")
require.Equal(t, preimage, result.attempt.Settle.Preimage,
"preimage mismatch")
@@ -1762,7 +1762,9 @@ func TestHandleAttemptResultWithError(t *testing.T) {
// Call the method under test and expect the dummy error to be
// returned.
attemptResult, err := p.handleAttemptResult(attempt, result)
attemptResult, err := p.handleAttemptResult(
t.Context(), attempt, result,
)
require.ErrorIs(t, err, errDummy, "expected fail error")
require.Nil(t, attemptResult, "expected nil attempt result")
}
@@ -1800,7 +1802,9 @@ func TestHandleAttemptResultSuccess(t *testing.T) {
// Call the method under test and expect the dummy error to be
// returned.
attemptResult, err := p.handleAttemptResult(attempt, result)
attemptResult, err := p.handleAttemptResult(
t.Context(), attempt, result,
)
require.NoError(t, err, "expected no error")
require.Equal(t, attempt, attemptResult.attempt)
}
routing/router.go
+11 -5
View File
@@ -1038,7 +1038,8 @@ func (r *ChannelRouter) PreparePayment(payment *LightningPayment) (
// SendToRoute sends a payment using the provided route and fails the payment
// when an error is returned from the attempt.
func (r *ChannelRouter) SendToRoute(htlcHash lntypes.Hash, rt *route.Route,
func (r *ChannelRouter) SendToRoute(_ context.Context, htlcHash lntypes.Hash,
rt *route.Route,
firstHopCustomRecords lnwire.CustomRecords) (*paymentsdb.HTLCAttempt,
error) {
@@ -1047,8 +1048,8 @@ func (r *ChannelRouter) SendToRoute(htlcHash lntypes.Hash, rt *route.Route,
// SendToRouteSkipTempErr sends a payment using the provided route and fails
// the payment ONLY when a terminal error is returned from the attempt.
func (r *ChannelRouter) SendToRouteSkipTempErr(htlcHash lntypes.Hash,
rt *route.Route,
func (r *ChannelRouter) SendToRouteSkipTempErr(_ context.Context,
htlcHash lntypes.Hash, rt *route.Route,
firstHopCustomRecords lnwire.CustomRecords) (*paymentsdb.HTLCAttempt,
error) {
@@ -1066,6 +1067,11 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
firstHopCustomRecords lnwire.CustomRecords) (*paymentsdb.HTLCAttempt,
error) {
// TODO(ziggie): We cannot easily thread the context from the caller
// of this method because the payment lifecycle depends on the context
// to update the db. The Sending and Receiving of results is currently
// not cleanly separated which is the reason that we cannot easily
// cancel the context and therefore cancel the ongoing payment.
ctx := context.TODO()
// Helper function to fail a payment. It makes sure the payment is only
@@ -1179,7 +1185,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
// NOTE: we use zero `remainingAmt` here to simulate the same effect of
// setting the lastShard to be false, which is used by previous
// implementation.
attempt, err := p.registerAttempt(rt, 0)
attempt, err := p.registerAttempt(ctx, rt, 0)
if err != nil {
return nil, err
}
@@ -1216,7 +1222,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
// The attempt was successfully sent, wait for the result to be
// available.
result, err = p.collectAndHandleResult(attempt)
result, err = p.collectAndHandleResult(ctx, attempt)
if err != nil {
return nil, err
}
routing/router_test.go
+19 -9
View File
@@ -522,7 +522,7 @@ func TestChannelUpdateValidation(t *testing.T) {
// Send off the payment request to the router. The specified route
// should be attempted and the channel update should be received by
// graph and ignored because it is missing a valid signature.
_, err = ctx.router.SendToRoute(payment, rt, nil)
_, err = ctx.router.SendToRoute(t.Context(), payment, rt, nil)
require.Error(t, err, "expected route to fail with channel update")
_, e1, e2, err = ctx.graph.FetchChannelEdgesByID(
@@ -542,7 +542,7 @@ func TestChannelUpdateValidation(t *testing.T) {
ctx.graphBuilder.setNextReject(false)
// Retry the payment using the same route as before.
_, err = ctx.router.SendToRoute(payment, rt, nil)
_, err = ctx.router.SendToRoute(t.Context(), payment, rt, nil)
require.Error(t, err, "expected route to fail with channel update")
// This time a valid signature was supplied and the policy change should
@@ -1427,7 +1427,9 @@ func TestSendToRouteStructuredError(t *testing.T) {
// update should be received by router and ignored
// because it is missing a valid
// signature.
_, err = ctx.router.SendToRoute(payment, rt, nil)
_, err = ctx.router.SendToRoute(
t.Context(), payment, rt, nil,
)
fErr, ok := err.(*htlcswitch.ForwardingError)
require.True(
@@ -1506,7 +1508,7 @@ func TestSendToRouteMaxHops(t *testing.T) {
// Send off the payment request to the router. We expect an error back
// indicating that the route is too long.
var payHash lntypes.Hash
_, err = ctx.router.SendToRoute(payHash, rt, nil)
_, err = ctx.router.SendToRoute(t.Context(), payHash, rt, nil)
if err != route.ErrMaxRouteHopsExceeded {
t.Fatalf("expected ErrMaxRouteHopsExceeded, but got %v", err)
}
@@ -2221,7 +2223,9 @@ func TestSendToRouteSkipTempErrSuccess(t *testing.T) {
).Return(nil)
// Expect a successful send to route.
attempt, err := router.SendToRouteSkipTempErr(payHash, rt, nil)
attempt, err := router.SendToRouteSkipTempErr(
t.Context(), payHash, rt, nil,
)
require.NoError(t, err)
require.Equal(t, testAttempt, attempt)
@@ -2276,7 +2280,9 @@ func TestSendToRouteSkipTempErrNonMPP(t *testing.T) {
}}
// Expect an error to be returned.
attempt, err := router.SendToRouteSkipTempErr(payHash, rt, nil)
attempt, err := router.SendToRouteSkipTempErr(
t.Context(), payHash, rt, nil,
)
require.ErrorIs(t, ErrSkipTempErr, err)
require.Nil(t, attempt)
@@ -2356,7 +2362,9 @@ func TestSendToRouteSkipTempErrTempFailure(t *testing.T) {
).Return(nil, nil)
// Expect a failed send to route.
attempt, err := router.SendToRouteSkipTempErr(payHash, rt, nil)
attempt, err := router.SendToRouteSkipTempErr(
t.Context(), payHash, rt, nil,
)
require.Equal(t, tempErr, err)
require.Equal(t, testAttempt, attempt)
@@ -2440,7 +2448,9 @@ func TestSendToRouteSkipTempErrPermanentFailure(t *testing.T) {
).Return(&failureReason, nil)
// Expect a failed send to route.
attempt, err := router.SendToRouteSkipTempErr(payHash, rt, nil)
attempt, err := router.SendToRouteSkipTempErr(
t.Context(), payHash, rt, nil,
)
require.Equal(t, permErr, err)
require.Equal(t, testAttempt, attempt)
@@ -2529,7 +2539,7 @@ func TestSendToRouteTempFailure(t *testing.T) {
).Return(nil, nil)
// Expect a failed send to route.
attempt, err := router.SendToRoute(payHash, rt, nil)
attempt, err := router.SendToRoute(t.Context(), payHash, rt, nil)
require.Equal(t, tempErr, err)
require.Equal(t, testAttempt, attempt)
rpcserver.go
+23 -12
View File
@@ -5581,8 +5581,9 @@ func (r *rpcServer) SubscribeChannelEvents(req *lnrpc.ChannelEventSubscription,
// execute sendPayment. We use this struct as a sort of bridge to enable code
// re-use between SendPayment and SendToRoute.
type paymentStream struct {
recv func() (*rpcPaymentRequest, error)
send func(*lnrpc.SendResponse) error
getCtx func() context.Context
recv func() (*rpcPaymentRequest, error)
send func(*lnrpc.SendResponse) error
}
// rpcPaymentRequest wraps lnrpc.SendRequest so that routes from
@@ -5596,10 +5597,13 @@ type rpcPaymentRequest struct {
// through the Lightning Network. A single RPC invocation creates a persistent
// bi-directional stream allowing clients to rapidly send payments through the
// Lightning Network with a single persistent connection.
func (r *rpcServer) SendPayment(stream lnrpc.Lightning_SendPaymentServer) error {
func (r *rpcServer) SendPayment(
stream lnrpc.Lightning_SendPaymentServer) error {
var lock sync.Mutex
return r.sendPayment(&paymentStream{
getCtx: stream.Context,
recv: func() (*rpcPaymentRequest, error) {
req, err := stream.Recv()
if err != nil {
@@ -5624,10 +5628,13 @@ func (r *rpcServer) SendPayment(stream lnrpc.Lightning_SendPaymentServer) error
// invocation creates a persistent bi-directional stream allowing clients to
// rapidly send payments through the Lightning Network with a single persistent
// connection.
func (r *rpcServer) SendToRoute(stream lnrpc.Lightning_SendToRouteServer) error {
func (r *rpcServer) SendToRoute(
stream lnrpc.Lightning_SendToRouteServer) error {
var lock sync.Mutex
return r.sendPayment(&paymentStream{
getCtx: stream.Context,
recv: func() (*rpcPaymentRequest, error) {
req, err := stream.Recv()
if err != nil {
@@ -5697,7 +5704,11 @@ type rpcPaymentIntent struct {
// dispatch a client from the information presented by an RPC client. There are
// three ways a client can specify their payment details: a payment request,
// via manual details, or via a complete route.
func (r *rpcServer) extractPaymentIntent(rpcPayReq *rpcPaymentRequest) (rpcPaymentIntent, error) {
//
//nolint:funlen
func (r *rpcServer) extractPaymentIntent(
rpcPayReq *rpcPaymentRequest) (rpcPaymentIntent, error) {
payIntent := rpcPaymentIntent{}
// If a route was specified, then we can use that directly.
@@ -5969,7 +5980,7 @@ type paymentIntentResponse struct {
// pre-built route. The first error this method returns denotes if we were
// unable to save the payment. The second error returned denotes if the payment
// didn't succeed.
func (r *rpcServer) dispatchPaymentIntent(
func (r *rpcServer) dispatchPaymentIntent(ctx context.Context,
payIntent *rpcPaymentIntent) (*paymentIntentResponse, error) {
// Construct a payment request to send to the channel router. If the
@@ -6016,7 +6027,7 @@ func (r *rpcServer) dispatchPaymentIntent(
} else {
var attempt *paymentsdb.HTLCAttempt
attempt, routerErr = r.server.chanRouter.SendToRoute(
payIntent.rHash, payIntent.route, nil,
ctx, payIntent.rHash, payIntent.route, nil,
)
if routerErr == nil {
@@ -6189,7 +6200,7 @@ sendLoop:
}()
resp, saveErr := r.dispatchPaymentIntent(
payIntent,
stream.getCtx(), payIntent,
)
switch {
@@ -6267,7 +6278,7 @@ sendLoop:
func (r *rpcServer) SendPaymentSync(ctx context.Context,
nextPayment *lnrpc.SendRequest) (*lnrpc.SendResponse, error) {
return r.sendPaymentSync(&rpcPaymentRequest{
return r.sendPaymentSync(ctx, &rpcPaymentRequest{
SendRequest: nextPayment,
})
}
@@ -6288,12 +6299,12 @@ func (r *rpcServer) SendToRouteSync(ctx context.Context,
return nil, err
}
return r.sendPaymentSync(paymentRequest)
return r.sendPaymentSync(ctx, paymentRequest)
}
// sendPaymentSync is the synchronous variant of sendPayment. It will block and
// wait until the payment has been fully completed.
func (r *rpcServer) sendPaymentSync(
func (r *rpcServer) sendPaymentSync(ctx context.Context,
nextPayment *rpcPaymentRequest) (*lnrpc.SendResponse, error) {
// We don't allow payments to be sent while the daemon itself is still
@@ -6312,7 +6323,7 @@ func (r *rpcServer) sendPaymentSync(
// With the payment validated, we'll now attempt to dispatch the
// payment.
resp, saveErr := r.dispatchPaymentIntent(&payIntent)
resp, saveErr := r.dispatchPaymentIntent(ctx, &payIntent)
switch {
case saveErr != nil:
return nil, saveErr