diff --git a/lnrpc/routerrpc/router_server.go b/lnrpc/routerrpc/router_server.go index a4031b154..7f2514aee 100644 --- a/lnrpc/routerrpc/router_server.go +++ b/lnrpc/routerrpc/router_server.go @@ -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 { diff --git a/routing/payment_lifecycle.go b/routing/payment_lifecycle.go index c8a59a323..43d56331d 100644 --- a/routing/payment_lifecycle.go +++ b/routing/payment_lifecycle.go @@ -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) } diff --git a/routing/payment_lifecycle_test.go b/routing/payment_lifecycle_test.go index a03218bfe..61ae83a31 100644 --- a/routing/payment_lifecycle_test.go +++ b/routing/payment_lifecycle_test.go @@ -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) } diff --git a/routing/router.go b/routing/router.go index fe8d06797..c67fe42e3 100644 --- a/routing/router.go +++ b/routing/router.go @@ -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 } diff --git a/routing/router_test.go b/routing/router_test.go index 733943272..9bc7bdbe1 100644 --- a/routing/router_test.go +++ b/routing/router_test.go @@ -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) diff --git a/rpcserver.go b/rpcserver.go index 2f88b2db8..3909fb1a7 100644 --- a/rpcserver.go +++ b/rpcserver.go @@ -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