Merge pull request #8734 from hieblmi/cancel-estimateroutefee
routing: cancelable payment loop
This commit is contained in:
9 files changed
+317
-146
No files matched your search
@@ -139,6 +139,17 @@ var (
|
||||
Usage: "(blinded paths) the total cltv delay for the " +
|
||||
"blinded portion of the route",
|
||||
}
|
||||
|
||||
cancelableFlag = cli.BoolFlag{
|
||||
Name: "cancelable",
|
||||
Usage: "if set to true, the payment loop can be interrupted " +
|
||||
"by manually canceling the payment context, even " +
|
||||
"before the payment timeout is reached. Note that " +
|
||||
"the payment may still succeed after cancellation, " +
|
||||
"as in-flight attempts can still settle afterwards. " +
|
||||
"Canceling will only prevent further attempts from " +
|
||||
"being sent",
|
||||
}
|
||||
)
|
||||
|
||||
// paymentFlags returns common flags for sendpayment and payinvoice.
|
||||
@@ -166,6 +177,7 @@ func paymentFlags() []cli.Flag {
|
||||
"after the timeout has elapsed",
|
||||
Value: paymentTimeout,
|
||||
},
|
||||
cancelableFlag,
|
||||
cltvLimitFlag,
|
||||
lastHopFlag,
|
||||
cli.Int64SliceFlag{
|
||||
@@ -329,6 +341,7 @@ func sendPayment(ctx *cli.Context) error {
|
||||
Amt: ctx.Int64("amt"),
|
||||
DestCustomRecords: make(map[uint64][]byte),
|
||||
Amp: ctx.Bool(ampFlag.Name),
|
||||
Cancelable: ctx.Bool(cancelableFlag.Name),
|
||||
}
|
||||
|
||||
// We'll attempt to parse a payment address as well, given that
|
||||
@@ -387,6 +400,7 @@ func sendPayment(ctx *cli.Context) error {
|
||||
Amt: amount,
|
||||
DestCustomRecords: make(map[uint64][]byte),
|
||||
Amp: ctx.Bool(ampFlag.Name),
|
||||
Cancelable: ctx.Bool(cancelableFlag.Name),
|
||||
}
|
||||
|
||||
var rHash []byte
|
||||
@@ -888,6 +902,7 @@ func payInvoice(ctx *cli.Context) error {
|
||||
Amt: ctx.Int64("amt"),
|
||||
DestCustomRecords: make(map[uint64][]byte),
|
||||
Amp: ctx.Bool(ampFlag.Name),
|
||||
Cancelable: ctx.Bool(cancelableFlag.Name),
|
||||
}
|
||||
|
||||
return sendPaymentRequest(ctx, req)
|
||||
|
||||
@@ -22,15 +22,25 @@
|
||||
* `closedchannels` now [successfully reports](https://github.com/lightningnetwork/lnd/pull/8800)
|
||||
settled balances even if the delivery address is set to an address that
|
||||
LND does not control.
|
||||
|
||||
* [SendPaymentV2](https://github.com/lightningnetwork/lnd/pull/8734) now cancels
|
||||
the background payment loop if the user cancels the stream context.
|
||||
|
||||
# New Features
|
||||
## Functional Enhancements
|
||||
## RPC Additions
|
||||
|
||||
* The [SendPaymentRequest](https://github.com/lightningnetwork/lnd/pull/8734)
|
||||
message receives a new flag `cancelable` which indicates if the payment loop
|
||||
is cancelable. The cancellation can either occur manually by cancelling the
|
||||
send payment stream context, or automatically at the end of the timeout period
|
||||
if the user provided `timeout_seconds`.
|
||||
|
||||
## lncli Additions
|
||||
|
||||
* [Added](https://github.com/lightningnetwork/lnd/pull/8491) the `cltv_expiry`
|
||||
argument to `addinvoice` and `addholdinvoice`, allowing users to set the
|
||||
`min_final_cltv_expiry_delta`
|
||||
`min_final_cltv_expiry_delta`.
|
||||
|
||||
* The [`lncli wallet estimatefeerate`](https://github.com/lightningnetwork/lnd/pull/8730)
|
||||
command returns the fee rate estimate for on-chain transactions in sat/kw and
|
||||
@@ -72,3 +82,4 @@
|
||||
|
||||
* Andras Banki-Horvath
|
||||
* Bufo
|
||||
* Slyghtning
|
||||
@@ -492,6 +492,12 @@ type SendPaymentRequest struct {
|
||||
// The time preference for this payment. Set to -1 to optimize for fees
|
||||
// only, to 1 to optimize for reliability only or a value inbetween for a mix.
|
||||
TimePref float64 `protobuf:"fixed64,23,opt,name=time_pref,json=timePref,proto3" json:"time_pref,omitempty"`
|
||||
// If set, the payment loop can be interrupted by manually canceling the
|
||||
// payment context, even before the payment timeout is reached. Note that the
|
||||
// payment may still succeed after cancellation, as in-flight attempts can
|
||||
// still settle afterwards. Canceling will only prevent further attempts from
|
||||
// being sent.
|
||||
Cancelable bool `protobuf:"varint,24,opt,name=cancelable,proto3" json:"cancelable,omitempty"`
|
||||
}
|
||||
|
||||
func (x *SendPaymentRequest) Reset() {
|
||||
@@ -688,6 +694,13 @@ func (x *SendPaymentRequest) GetTimePref() float64 {
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *SendPaymentRequest) GetCancelable() bool {
|
||||
if x != nil {
|
||||
return x.Cancelable
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
type TrackPaymentRequest struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
@@ -3339,7 +3352,7 @@ var file_routerrpc_router_proto_rawDesc = []byte{
|
||||
0x0a, 0x16, 0x72, 0x6f, 0x75, 0x74, 0x65, 0x72, 0x72, 0x70, 0x63, 0x2f, 0x72, 0x6f, 0x75, 0x74,
|
||||
0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x09, 0x72, 0x6f, 0x75, 0x74, 0x65, 0x72,
|
||||
0x72, 0x70, 0x63, 0x1a, 0x0f, 0x6c, 0x69, 0x67, 0x68, 0x74, 0x6e, 0x69, 0x6e, 0x67, 0x2e, 0x70,
|
||||
0x72, 0x6f, 0x74, 0x6f, 0x22, 0xf4, 0x07, 0x0a, 0x12, 0x53, 0x65, 0x6e, 0x64, 0x50, 0x61, 0x79,
|
||||
0x72, 0x6f, 0x74, 0x6f, 0x22, 0x94, 0x08, 0x0a, 0x12, 0x53, 0x65, 0x6e, 0x64, 0x50, 0x61, 0x79,
|
||||
0x6d, 0x65, 0x6e, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x64,
|
||||
0x65, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x64, 0x65, 0x73, 0x74, 0x12,
|
||||
0x10, 0x0a, 0x03, 0x61, 0x6d, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x03, 0x61, 0x6d,
|
||||
@@ -3398,7 +3411,9 @@ var file_routerrpc_router_proto_rawDesc = []byte{
|
||||
0x7a, 0x65, 0x4d, 0x73, 0x61, 0x74, 0x12, 0x10, 0x0a, 0x03, 0x61, 0x6d, 0x70, 0x18, 0x16, 0x20,
|
||||
0x01, 0x28, 0x08, 0x52, 0x03, 0x61, 0x6d, 0x70, 0x12, 0x1b, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65,
|
||||
0x5f, 0x70, 0x72, 0x65, 0x66, 0x18, 0x17, 0x20, 0x01, 0x28, 0x01, 0x52, 0x08, 0x74, 0x69, 0x6d,
|
||||
0x65, 0x50, 0x72, 0x65, 0x66, 0x1a, 0x44, 0x0a, 0x16, 0x44, 0x65, 0x73, 0x74, 0x43, 0x75, 0x73,
|
||||
0x65, 0x50, 0x72, 0x65, 0x66, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x61, 0x6e, 0x63, 0x65, 0x6c, 0x61,
|
||||
0x62, 0x6c, 0x65, 0x18, 0x18, 0x20, 0x01, 0x28, 0x08, 0x52, 0x0a, 0x63, 0x61, 0x6e, 0x63, 0x65,
|
||||
0x6c, 0x61, 0x62, 0x6c, 0x65, 0x1a, 0x44, 0x0a, 0x16, 0x44, 0x65, 0x73, 0x74, 0x43, 0x75, 0x73,
|
||||
0x74, 0x6f, 0x6d, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12,
|
||||
0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x03, 0x6b, 0x65,
|
||||
0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c,
|
||||
|
||||
@@ -330,6 +330,15 @@ message SendPaymentRequest {
|
||||
only, to 1 to optimize for reliability only or a value inbetween for a mix.
|
||||
*/
|
||||
double time_pref = 23;
|
||||
|
||||
/*
|
||||
If set, the payment loop can be interrupted by manually canceling the
|
||||
payment context, even before the payment timeout is reached. Note that the
|
||||
payment may still succeed after cancellation, as in-flight attempts can
|
||||
still settle afterwards. Canceling will only prevent further attempts from
|
||||
being sent.
|
||||
*/
|
||||
bool cancelable = 24;
|
||||
}
|
||||
|
||||
message TrackPaymentRequest {
|
||||
|
||||
@@ -1804,6 +1804,10 @@
|
||||
"type": "number",
|
||||
"format": "double",
|
||||
"description": "The time preference for this payment. Set to -1 to optimize for fees\nonly, to 1 to optimize for reliability only or a value inbetween for a mix."
|
||||
},
|
||||
"cancelable": {
|
||||
"type": "boolean",
|
||||
"description": "If set, the payment loop can be interrupted by manually canceling the\npayment context, even before the payment timeout is reached. Note that the\npayment may still succeed after cancellation, as in-flight attempts can\nstill settle afterwards. Canceling will only prevent further attempts from\nbeing sent."
|
||||
}
|
||||
}
|
||||
},
|
||||
|
||||
@@ -149,14 +149,14 @@ var (
|
||||
DefaultRouterMacFilename = "router.macaroon"
|
||||
)
|
||||
|
||||
// ServerShell a is shell struct holding a reference to the actual sub-server.
|
||||
// ServerShell is a shell struct holding a reference to the actual sub-server.
|
||||
// It is used to register the gRPC sub-server with the root server before we
|
||||
// have the necessary dependencies to populate the actual sub-server.
|
||||
type ServerShell struct {
|
||||
RouterServer
|
||||
}
|
||||
|
||||
// Server is a stand alone sub RPC server which exposes functionality that
|
||||
// Server is a stand-alone sub RPC server which exposes functionality that
|
||||
// allows clients to route arbitrary payment through the Lightning Network.
|
||||
type Server struct {
|
||||
started int32 // To be used atomically.
|
||||
@@ -181,7 +181,7 @@ var _ RouterServer = (*Server)(nil)
|
||||
// that contains all external dependencies. If the target macaroon exists, and
|
||||
// we're unable to create it, then an error will be returned. We also return
|
||||
// the set of permissions that we require as a server. At the time of writing
|
||||
// of this documentation, this is the same macaroon as as the admin macaroon.
|
||||
// of this documentation, this is the same macaroon as the admin macaroon.
|
||||
func New(cfg *Config) (*Server, lnrpc.MacaroonPerms, error) {
|
||||
// If the path of the router macaroon wasn't generated, then we'll
|
||||
// assume that it's found at the default network directory.
|
||||
@@ -360,13 +360,25 @@ func (s *Server) SendPaymentV2(req *SendPaymentRequest,
|
||||
return err
|
||||
}
|
||||
|
||||
// The payment context is influenced by two user-provided parameters,
|
||||
// the cancelable flag and the payment attempt timeout.
|
||||
// If the payment is cancelable, we will use the stream context as the
|
||||
// payment context. That way, if the user ends the stream, the payment
|
||||
// loop will be canceled.
|
||||
// The second context parameter is the timeout. If the user provides a
|
||||
// timeout, we will additionally wrap the context in a deadline. If the
|
||||
// user provided 'cancelable' and ends the stream before the timeout is
|
||||
// reached the payment will be canceled.
|
||||
ctx := context.Background()
|
||||
if req.Cancelable {
|
||||
ctx = stream.Context()
|
||||
}
|
||||
|
||||
// Send the payment asynchronously.
|
||||
s.cfg.Router.SendPaymentAsync(payment, paySession, shardTracker)
|
||||
s.cfg.Router.SendPaymentAsync(ctx, payment, paySession, shardTracker)
|
||||
|
||||
// Track the payment and return.
|
||||
return s.trackPayment(
|
||||
sub, payHash, stream, req.NoInflightUpdates,
|
||||
)
|
||||
return s.trackPayment(sub, payHash, stream, req.NoInflightUpdates)
|
||||
}
|
||||
|
||||
// EstimateRouteFee allows callers to obtain an expected value w.r.t how much it
|
||||
@@ -986,9 +998,8 @@ func (s *Server) SetMissionControlConfig(ctx context.Context,
|
||||
AprioriHopProbability: float64(
|
||||
req.Config.HopProbability,
|
||||
),
|
||||
AprioriWeight: float64(req.Config.Weight),
|
||||
CapacityFraction: float64(
|
||||
routing.DefaultCapacityFraction),
|
||||
AprioriWeight: float64(req.Config.Weight),
|
||||
CapacityFraction: routing.DefaultCapacityFraction, //nolint:lll
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1032,8 +1043,8 @@ func (s *Server) SetMissionControlConfig(ctx context.Context,
|
||||
|
||||
// QueryMissionControl exposes the internal mission control state to callers. It
|
||||
// is a development feature.
|
||||
func (s *Server) QueryMissionControl(ctx context.Context,
|
||||
req *QueryMissionControlRequest) (*QueryMissionControlResponse, error) {
|
||||
func (s *Server) QueryMissionControl(_ context.Context,
|
||||
_ *QueryMissionControlRequest) (*QueryMissionControlResponse, error) {
|
||||
|
||||
snapshot := s.cfg.RouterBackend.MissionControl.GetHistorySnapshot()
|
||||
|
||||
@@ -1080,7 +1091,7 @@ func toRPCPairData(data *routing.TimedPairResult) *PairData {
|
||||
|
||||
// XImportMissionControl imports the state provided to our internal mission
|
||||
// control. Only entries that are fresher than our existing state will be used.
|
||||
func (s *Server) XImportMissionControl(ctx context.Context,
|
||||
func (s *Server) XImportMissionControl(_ context.Context,
|
||||
req *XImportMissionControlRequest) (*XImportMissionControlResponse,
|
||||
error) {
|
||||
|
||||
@@ -1273,8 +1284,9 @@ func (s *Server) subscribePayment(identifier lntypes.Hash) (
|
||||
sub, err := router.Tower.SubscribePayment(identifier)
|
||||
|
||||
switch {
|
||||
case err == channeldb.ErrPaymentNotInitiated:
|
||||
case errors.Is(err, channeldb.ErrPaymentNotInitiated):
|
||||
return nil, status.Error(codes.NotFound, err.Error())
|
||||
|
||||
case err != nil:
|
||||
return nil, err
|
||||
}
|
||||
@@ -1385,7 +1397,7 @@ func (s *Server) trackPaymentStream(context context.Context,
|
||||
}
|
||||
|
||||
// BuildRoute builds a route from a list of hop addresses.
|
||||
func (s *Server) BuildRoute(ctx context.Context,
|
||||
func (s *Server) BuildRoute(_ context.Context,
|
||||
req *BuildRouteRequest) (*BuildRouteResponse, error) {
|
||||
|
||||
// Unmarshall hop list.
|
||||
@@ -1446,7 +1458,7 @@ func (s *Server) BuildRoute(ctx context.Context,
|
||||
|
||||
// SubscribeHtlcEvents creates a uni-directional stream from the server to
|
||||
// the client which delivers a stream of htlc events.
|
||||
func (s *Server) SubscribeHtlcEvents(req *SubscribeHtlcEventsRequest,
|
||||
func (s *Server) SubscribeHtlcEvents(_ *SubscribeHtlcEventsRequest,
|
||||
stream Router_SubscribeHtlcEventsServer) error {
|
||||
|
||||
htlcClient, err := s.cfg.RouterBackend.SubscribeHtlcEvents()
|
||||
@@ -1495,7 +1507,7 @@ func (s *Server) SubscribeHtlcEvents(req *SubscribeHtlcEventsRequest,
|
||||
|
||||
// HtlcInterceptor is a bidirectional stream for streaming interception
|
||||
// requests to the caller.
|
||||
// Upon connection it does the following:
|
||||
// Upon connection, it does the following:
|
||||
// 1. Check if there is already a live stream, if yes it rejects the request.
|
||||
// 2. Registered a ForwardInterceptor
|
||||
// 3. Delivers to the caller every √√ and detect his answer.
|
||||
@@ -1525,7 +1537,7 @@ func extractOutPoint(req *UpdateChanStatusRequest) (*wire.OutPoint, error) {
|
||||
}
|
||||
|
||||
// UpdateChanStatus allows channel state to be set manually.
|
||||
func (s *Server) UpdateChanStatus(ctx context.Context,
|
||||
func (s *Server) UpdateChanStatus(_ context.Context,
|
||||
req *UpdateChanStatusRequest) (*UpdateChanStatusResponse, error) {
|
||||
|
||||
outPoint, err := extractOutPoint(req)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package routing
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
@@ -29,7 +30,6 @@ type paymentLifecycle struct {
|
||||
identifier lntypes.Hash
|
||||
paySession PaymentSession
|
||||
shardTracker shards.ShardTracker
|
||||
timeoutChan <-chan time.Time
|
||||
currentHeight int32
|
||||
|
||||
// quit is closed to signal the sub goroutines of the payment lifecycle
|
||||
@@ -52,7 +52,7 @@ type paymentLifecycle struct {
|
||||
// newPaymentLifecycle initiates a new payment lifecycle and returns it.
|
||||
func newPaymentLifecycle(r *ChannelRouter, feeLimit lnwire.MilliSatoshi,
|
||||
identifier lntypes.Hash, paySession PaymentSession,
|
||||
shardTracker shards.ShardTracker, timeout time.Duration,
|
||||
shardTracker shards.ShardTracker,
|
||||
currentHeight int32) *paymentLifecycle {
|
||||
|
||||
p := &paymentLifecycle{
|
||||
@@ -69,13 +69,6 @@ func newPaymentLifecycle(r *ChannelRouter, feeLimit lnwire.MilliSatoshi,
|
||||
// Mount the result collector.
|
||||
p.resultCollector = p.collectResultAsync
|
||||
|
||||
// If a timeout is specified, create a timeout channel. If no timeout is
|
||||
// specified, the channel is left nil and will never abort the payment
|
||||
// loop.
|
||||
if timeout != 0 {
|
||||
p.timeoutChan = time.After(timeout)
|
||||
}
|
||||
|
||||
return p
|
||||
}
|
||||
|
||||
@@ -167,7 +160,9 @@ func (p *paymentLifecycle) decideNextStep(
|
||||
}
|
||||
|
||||
// resumePayment resumes the paymentLifecycle from the current state.
|
||||
func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
||||
func (p *paymentLifecycle) resumePayment(ctx context.Context) ([32]byte,
|
||||
*route.Route, error) {
|
||||
|
||||
// When the payment lifecycle loop exits, we make sure to signal any
|
||||
// sub goroutine of the HTLC attempt to exit, then wait for them to
|
||||
// return.
|
||||
@@ -221,18 +216,17 @@ lifecycle:
|
||||
|
||||
// We now proceed our lifecycle with the following tasks in
|
||||
// order,
|
||||
// 1. check timeout.
|
||||
// 1. check context.
|
||||
// 2. request route.
|
||||
// 3. create HTLC attempt.
|
||||
// 4. send HTLC attempt.
|
||||
// 5. collect HTLC attempt result.
|
||||
//
|
||||
// Before we attempt any new shard, we'll check to see if
|
||||
// either we've gone past the payment attempt timeout, or the
|
||||
// router is exiting. In either case, we'll stop this payment
|
||||
// attempt short. If a timeout is not applicable, timeoutChan
|
||||
// will be nil.
|
||||
if err := p.checkTimeout(); err != nil {
|
||||
// Before we attempt any new shard, we'll check to see if we've
|
||||
// gone past the payment attempt timeout, or if the context was
|
||||
// cancelled, or the router is exiting. In any of these cases,
|
||||
// we'll stop this payment attempt short.
|
||||
if err := p.checkContext(ctx); err != nil {
|
||||
return exitWithErr(err)
|
||||
}
|
||||
|
||||
@@ -318,19 +312,30 @@ lifecycle:
|
||||
return [32]byte{}, nil, *failure
|
||||
}
|
||||
|
||||
// checkTimeout checks whether the payment has reached its timeout.
|
||||
func (p *paymentLifecycle) checkTimeout() error {
|
||||
// checkContext checks whether the payment context has been canceled.
|
||||
// Cancellation occurs manually or if the context times out.
|
||||
func (p *paymentLifecycle) checkContext(ctx context.Context) error {
|
||||
select {
|
||||
case <-p.timeoutChan:
|
||||
log.Warnf("payment attempt not completed before timeout")
|
||||
case <-ctx.Done():
|
||||
// If the context was canceled, we'll mark the payment as
|
||||
// failed. There are two cases to distinguish here: Either a
|
||||
// user-provided timeout was reached, or the context was
|
||||
// canceled, either to a manual cancellation or due to an
|
||||
// unknown error.
|
||||
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
|
||||
log.Warnf("Payment attempt not completed before "+
|
||||
"timeout, id=%s", p.identifier.String())
|
||||
} else {
|
||||
log.Warnf("Payment attempt context canceled, id=%s",
|
||||
p.identifier.String())
|
||||
}
|
||||
|
||||
// By marking the payment failed, depending on whether it has
|
||||
// inflight HTLCs or not, its status will now either be
|
||||
// `StatusInflight` or `StatusFailed`. In either case, no more
|
||||
// HTLCs will be attempted.
|
||||
err := p.router.cfg.Control.FailPayment(
|
||||
p.identifier, channeldb.FailureReasonTimeout,
|
||||
)
|
||||
reason := channeldb.FailureReasonTimeout
|
||||
err := p.router.cfg.Control.FailPayment(p.identifier, reason)
|
||||
if err != nil {
|
||||
return fmt.Errorf("FailPayment got %w", err)
|
||||
}
|
||||
@@ -368,7 +373,7 @@ func (p *paymentLifecycle) requestRoute(
|
||||
log.Warnf("Failed to find route for payment %v: %v", p.identifier, err)
|
||||
|
||||
// If the error belongs to `noRouteError` set, it means a non-critical
|
||||
// error has happened during path finding and we will mark the payment
|
||||
// error has happened during path finding, and we will mark the payment
|
||||
// failed with this reason. Otherwise, we'll return the critical error
|
||||
// found to abort the lifecycle.
|
||||
var routeErr noRouteError
|
||||
@@ -377,9 +382,9 @@ func (p *paymentLifecycle) requestRoute(
|
||||
}
|
||||
|
||||
// It's the `paymentSession`'s responsibility to find a route for us
|
||||
// with best effort. When it cannot find a path, we need to treat it as
|
||||
// a terminal condition and fail the payment no matter it has inflight
|
||||
// HTLCs or not.
|
||||
// with the best effort. When it cannot find a path, we need to treat it
|
||||
// as a terminal condition and fail the payment no matter it has
|
||||
// inflight HTLCs or not.
|
||||
failureCode := routeErr.FailureReason()
|
||||
log.Warnf("Marking payment %v permanently failed with no route: %v",
|
||||
p.identifier, failureCode)
|
||||
@@ -415,7 +420,7 @@ type attemptResult struct {
|
||||
|
||||
// collectResultAsync launches a goroutine that will wait for the result of the
|
||||
// given HTLC attempt to be available then handle its result. Once received, it
|
||||
// will send a nil error to channel `resultCollected` to indicate there's an
|
||||
// will send a nil error to channel `resultCollected` to indicate there's a
|
||||
// result.
|
||||
func (p *paymentLifecycle) collectResultAsync(attempt *channeldb.HTLCAttempt) {
|
||||
log.Debugf("Collecting result for attempt %v in payment %v",
|
||||
@@ -484,7 +489,7 @@ func (p *paymentLifecycle) collectResult(attempt *channeldb.HTLCAttempt) (
|
||||
return p.failAttempt(attempt.AttemptID, err)
|
||||
}
|
||||
|
||||
// Using the created circuit, initialize the error decrypter so we can
|
||||
// Using the created circuit, initialize the error decrypter, so we can
|
||||
// parse+decode any failures incurred by this payment within the
|
||||
// switch.
|
||||
errorDecryptor := &htlcswitch.SphinxErrorDecrypter{
|
||||
@@ -786,7 +791,7 @@ func (p *paymentLifecycle) handleSwitchErr(attempt *channeldb.HTLCAttempt,
|
||||
return p.failAttempt(attemptID, sendErr)
|
||||
}
|
||||
|
||||
if sendErr == htlcswitch.ErrUnreadableFailureMessage {
|
||||
if errors.Is(sendErr, htlcswitch.ErrUnreadableFailureMessage) {
|
||||
log.Warn("Unreadable failure when sending htlc: id=%v, hash=%v",
|
||||
attempt.AttemptID, attempt.Hash)
|
||||
|
||||
@@ -801,7 +806,8 @@ func (p *paymentLifecycle) handleSwitchErr(attempt *channeldb.HTLCAttempt,
|
||||
// down the route. If the error is not related to the propagation of
|
||||
// our payment, we can stop trying because an internal error has
|
||||
// occurred.
|
||||
rtErr, ok := sendErr.(htlcswitch.ClearTextError)
|
||||
var rtErr htlcswitch.ClearTextError
|
||||
ok := errors.As(sendErr, &rtErr)
|
||||
if !ok {
|
||||
return p.failPaymentAndAttempt(
|
||||
attemptID, &internalErrorReason, sendErr,
|
||||
@@ -815,7 +821,8 @@ func (p *paymentLifecycle) handleSwitchErr(attempt *channeldb.HTLCAttempt,
|
||||
// ForwardingError, it did not originate at our node, so we set
|
||||
// failureSourceIdx to the index of the node where the failure occurred.
|
||||
failureSourceIdx := 0
|
||||
source, ok := rtErr.(*htlcswitch.ForwardingError)
|
||||
var source *htlcswitch.ForwardingError
|
||||
ok = errors.As(rtErr, &source)
|
||||
if ok {
|
||||
failureSourceIdx = source.FailureSourceIdx
|
||||
}
|
||||
@@ -863,7 +870,7 @@ func (p *paymentLifecycle) handleFailureMessage(rt *route.Route,
|
||||
|
||||
// Parse pubkey to allow validation of the channel update. This should
|
||||
// always succeed, otherwise there is something wrong in our
|
||||
// implementation. Therefore return an error.
|
||||
// implementation. Therefore, return an error.
|
||||
errVertex := rt.Hops[errorSourceIdx-1].PubKeyBytes
|
||||
errSource, err := btcec.ParsePubKey(errVertex[:])
|
||||
if err != nil {
|
||||
@@ -951,17 +958,18 @@ func marshallError(sendError error, time time.Time) *channeldb.HTLCFailInfo {
|
||||
FailTime: time,
|
||||
}
|
||||
|
||||
switch sendError {
|
||||
case htlcswitch.ErrPaymentIDNotFound:
|
||||
switch {
|
||||
case errors.Is(sendError, htlcswitch.ErrPaymentIDNotFound):
|
||||
response.Reason = channeldb.HTLCFailInternal
|
||||
return response
|
||||
|
||||
case htlcswitch.ErrUnreadableFailureMessage:
|
||||
case errors.Is(sendError, htlcswitch.ErrUnreadableFailureMessage):
|
||||
response.Reason = channeldb.HTLCFailUnreadable
|
||||
return response
|
||||
}
|
||||
|
||||
rtErr, ok := sendError.(htlcswitch.ClearTextError)
|
||||
var rtErr htlcswitch.ClearTextError
|
||||
ok := errors.As(sendError, &rtErr)
|
||||
if !ok {
|
||||
response.Reason = channeldb.HTLCFailInternal
|
||||
return response
|
||||
@@ -981,7 +989,8 @@ func marshallError(sendError error, time time.Time) *channeldb.HTLCFailInfo {
|
||||
// failure occurred. If the error is not a ForwardingError, the failure
|
||||
// occurred at our node, so we leave the index as 0 to indicate that
|
||||
// we failed locally.
|
||||
fErr, ok := rtErr.(*htlcswitch.ForwardingError)
|
||||
var fErr *htlcswitch.ForwardingError
|
||||
ok = errors.As(rtErr, &fErr)
|
||||
if ok {
|
||||
response.FailureSourceIndex = uint32(fErr.FailureSourceIdx)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package routing
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -88,7 +89,7 @@ func newTestPaymentLifecycle(t *testing.T) (*paymentLifecycle, *mockers) {
|
||||
// Create a test payment lifecycle with no fee limit and no timeout.
|
||||
p := newPaymentLifecycle(
|
||||
rt, noFeeLimit, paymentHash, mockPaymentSession,
|
||||
mockShardTracker, 0, 0,
|
||||
mockShardTracker, 0,
|
||||
)
|
||||
|
||||
// Create a mock payment which is returned from mockControlTower.
|
||||
@@ -151,9 +152,9 @@ type resumePaymentResult struct {
|
||||
err error
|
||||
}
|
||||
|
||||
// sendPaymentAndAssertFailed calls `resumePayment` and asserts that an error
|
||||
// is returned.
|
||||
func sendPaymentAndAssertFailed(t *testing.T,
|
||||
// sendPaymentAndAssertError calls `resumePayment` and asserts that an error is
|
||||
// returned.
|
||||
func sendPaymentAndAssertError(t *testing.T, ctx context.Context,
|
||||
p *paymentLifecycle, errExpected error) {
|
||||
|
||||
resultChan := make(chan *resumePaymentResult, 1)
|
||||
@@ -161,7 +162,7 @@ func sendPaymentAndAssertFailed(t *testing.T,
|
||||
// We now make a call to `resumePayment` and expect it to return the
|
||||
// error.
|
||||
go func() {
|
||||
preimage, _, err := p.resumePayment()
|
||||
preimage, _, err := p.resumePayment(ctx)
|
||||
resultChan <- &resumePaymentResult{
|
||||
preimage: preimage,
|
||||
err: err,
|
||||
@@ -189,7 +190,7 @@ func sendPaymentAndAssertSucceeded(t *testing.T,
|
||||
// We now make a call to `resumePayment` and expect it to return the
|
||||
// preimage.
|
||||
go func() {
|
||||
preimage, _, err := p.resumePayment()
|
||||
preimage, _, err := p.resumePayment(context.Background())
|
||||
resultChan <- &resumePaymentResult{
|
||||
preimage: preimage,
|
||||
err: err,
|
||||
@@ -278,6 +279,10 @@ func makeAttemptInfo(t *testing.T, amt int) channeldb.HTLCAttemptInfo {
|
||||
func TestCheckTimeoutTimedOut(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
deadline := time.Now().Add(time.Nanosecond)
|
||||
ctx, cancel := context.WithDeadline(context.Background(), deadline)
|
||||
defer cancel()
|
||||
|
||||
p := createTestPaymentLifecycle()
|
||||
|
||||
// Mock the control tower's `FailPayment` method.
|
||||
@@ -288,14 +293,11 @@ func TestCheckTimeoutTimedOut(t *testing.T) {
|
||||
// Mount the mocked control tower.
|
||||
p.router.cfg.Control = ct
|
||||
|
||||
// Make the timeout happens instantly.
|
||||
p.timeoutChan = time.After(1 * time.Nanosecond)
|
||||
|
||||
// Sleep one millisecond to make sure it timed out.
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
|
||||
// Call the function and expect no error.
|
||||
err := p.checkTimeout()
|
||||
err := p.checkContext(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Assert that `FailPayment` is called as expected.
|
||||
@@ -313,13 +315,15 @@ func TestCheckTimeoutTimedOut(t *testing.T) {
|
||||
p.router.cfg.Control = ct
|
||||
|
||||
// Make the timeout happens instantly.
|
||||
p.timeoutChan = time.After(1 * time.Nanosecond)
|
||||
deadline = time.Now().Add(time.Nanosecond)
|
||||
ctx, cancel = context.WithDeadline(context.Background(), deadline)
|
||||
defer cancel()
|
||||
|
||||
// Sleep one millisecond to make sure it timed out.
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
|
||||
// Call the function and expect an error.
|
||||
err = p.checkTimeout()
|
||||
err = p.checkContext(ctx)
|
||||
require.ErrorIs(t, err, errDummy)
|
||||
|
||||
// Assert that `FailPayment` is called as expected.
|
||||
@@ -331,10 +335,13 @@ func TestCheckTimeoutTimedOut(t *testing.T) {
|
||||
func TestCheckTimeoutOnRouterQuit(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
p := createTestPaymentLifecycle()
|
||||
|
||||
close(p.router.quit)
|
||||
err := p.checkTimeout()
|
||||
err := p.checkContext(ctx)
|
||||
require.ErrorIs(t, err, ErrRouterShuttingDown)
|
||||
}
|
||||
|
||||
@@ -627,7 +634,7 @@ func TestResumePaymentFailOnFetchPayment(t *testing.T) {
|
||||
m.control.On("FetchPayment", p.identifier).Return(nil, errDummy)
|
||||
|
||||
// Send the payment and assert it failed.
|
||||
sendPaymentAndAssertFailed(t, p, errDummy)
|
||||
sendPaymentAndAssertError(t, context.Background(), p, errDummy)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
@@ -656,14 +663,15 @@ func TestResumePaymentFailOnTimeout(t *testing.T) {
|
||||
}
|
||||
m.payment.On("GetState").Return(ps).Once()
|
||||
|
||||
// NOTE: GetStatus is only used to populate the logs which is
|
||||
// not critical so we loosen the checks on how many times it's
|
||||
// been called.
|
||||
// NOTE: GetStatus is only used to populate the logs which is not
|
||||
// critical, so we loosen the checks on how many times it's been called.
|
||||
m.payment.On("GetStatus").Return(channeldb.StatusInFlight)
|
||||
|
||||
// 3. make the timeout happens instantly and sleep one millisecond to
|
||||
// make sure it timed out.
|
||||
p.timeoutChan = time.After(1 * time.Nanosecond)
|
||||
deadline := time.Now().Add(time.Nanosecond)
|
||||
ctx, cancel := context.WithDeadline(context.Background(), deadline)
|
||||
defer cancel()
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
|
||||
// 4. the payment should be failed with reason timeout.
|
||||
@@ -683,7 +691,7 @@ func TestResumePaymentFailOnTimeout(t *testing.T) {
|
||||
m.payment.On("TerminalInfo").Return(nil, &reason)
|
||||
|
||||
// Send the payment and assert it failed with the timeout reason.
|
||||
sendPaymentAndAssertFailed(t, p, reason)
|
||||
sendPaymentAndAssertError(t, ctx, p, reason)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
@@ -721,7 +729,65 @@ func TestResumePaymentFailOnTimeoutErr(t *testing.T) {
|
||||
close(p.router.quit)
|
||||
|
||||
// Send the payment and assert it failed when router is shutting down.
|
||||
sendPaymentAndAssertFailed(t, p, ErrRouterShuttingDown)
|
||||
sendPaymentAndAssertError(
|
||||
t, context.Background(), p, ErrRouterShuttingDown,
|
||||
)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
}
|
||||
|
||||
// TestResumePaymentFailContextCancel checks that the lifecycle fails when the
|
||||
// context is canceled.
|
||||
//
|
||||
// NOTE: No parallel test because it overwrites global variables.
|
||||
//
|
||||
//nolint:paralleltest
|
||||
func TestResumePaymentFailContextCancel(t *testing.T) {
|
||||
// Create a test paymentLifecycle with the initial two calls mocked.
|
||||
p, m := setupTestPaymentLifecycle(t)
|
||||
|
||||
// Create the cancelable payment context.
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
paymentAmt := lnwire.MilliSatoshi(10000)
|
||||
|
||||
// We now enter the payment lifecycle loop.
|
||||
//
|
||||
// 1. calls `FetchPayment` and return the payment.
|
||||
m.control.On("FetchPayment", p.identifier).Return(m.payment, nil).Once()
|
||||
|
||||
// 2. calls `GetState` and return the state.
|
||||
ps := &channeldb.MPPaymentState{
|
||||
RemainingAmt: paymentAmt,
|
||||
}
|
||||
m.payment.On("GetState").Return(ps).Once()
|
||||
|
||||
// NOTE: GetStatus is only used to populate the logs which is not
|
||||
// critical, so we loosen the checks on how many times it's been called.
|
||||
m.payment.On("GetStatus").Return(channeldb.StatusInFlight)
|
||||
|
||||
// 3. Cancel the context and skip the FailPayment error to trigger the
|
||||
// context cancellation of the payment.
|
||||
cancel()
|
||||
|
||||
m.control.On(
|
||||
"FailPayment", p.identifier, channeldb.FailureReasonTimeout,
|
||||
).Return(nil).Once()
|
||||
|
||||
// 5. decideNextStep now returns stepExit.
|
||||
m.payment.On("AllowMoreAttempts").Return(false, nil).Once().
|
||||
On("NeedWaitAttempts").Return(false, nil).Once()
|
||||
|
||||
// 6. Control tower deletes failed attempts.
|
||||
m.control.On("DeleteFailedAttempts", p.identifier).Return(nil).Once()
|
||||
|
||||
// 7. We will observe FailureReasonError if the context was cancelled.
|
||||
reason := channeldb.FailureReasonError
|
||||
m.payment.On("TerminalInfo").Return(nil, &reason)
|
||||
|
||||
// Send the payment and assert it failed with the timeout reason.
|
||||
sendPaymentAndAssertError(t, ctx, p, reason)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
@@ -759,7 +825,7 @@ func TestResumePaymentFailOnStepErr(t *testing.T) {
|
||||
m.payment.On("AllowMoreAttempts").Return(false, errDummy).Once()
|
||||
|
||||
// Send the payment and assert it failed.
|
||||
sendPaymentAndAssertFailed(t, p, errDummy)
|
||||
sendPaymentAndAssertError(t, context.Background(), p, errDummy)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
@@ -803,7 +869,7 @@ func TestResumePaymentFailOnRequestRouteErr(t *testing.T) {
|
||||
).Return(nil, errDummy).Once()
|
||||
|
||||
// Send the payment and assert it failed.
|
||||
sendPaymentAndAssertFailed(t, p, errDummy)
|
||||
sendPaymentAndAssertError(t, context.Background(), p, errDummy)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
@@ -863,7 +929,7 @@ func TestResumePaymentFailOnRegisterAttemptErr(t *testing.T) {
|
||||
).Return(nil, errDummy).Once()
|
||||
|
||||
// Send the payment and assert it failed.
|
||||
sendPaymentAndAssertFailed(t, p, errDummy)
|
||||
sendPaymentAndAssertError(t, context.Background(), p, errDummy)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
@@ -955,7 +1021,7 @@ func TestResumePaymentFailOnSendAttemptErr(t *testing.T) {
|
||||
).Return(nil, errDummy).Once()
|
||||
|
||||
// Send the payment and assert it failed.
|
||||
sendPaymentAndAssertFailed(t, p, errDummy)
|
||||
sendPaymentAndAssertError(t, context.Background(), p, errDummy)
|
||||
|
||||
// Expected collectResultAsync to not be called.
|
||||
require.Zero(t, m.collectResultsCount)
|
||||
|
||||
+92
-62
@@ -2,6 +2,7 @@ package routing
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"runtime"
|
||||
@@ -45,11 +46,11 @@ const (
|
||||
// DefaultPayAttemptTimeout is the default payment attempt timeout. The
|
||||
// payment attempt timeout defines the duration after which we stop
|
||||
// trying more routes for a payment.
|
||||
DefaultPayAttemptTimeout = time.Duration(time.Second * 60)
|
||||
DefaultPayAttemptTimeout = time.Second * 60
|
||||
|
||||
// DefaultChannelPruneExpiry is the default duration used to determine
|
||||
// if a channel should be pruned or not.
|
||||
DefaultChannelPruneExpiry = time.Duration(time.Hour * 24 * 14)
|
||||
DefaultChannelPruneExpiry = time.Hour * 24 * 14
|
||||
|
||||
// DefaultFirstTimePruneDelay is the time we'll wait after startup
|
||||
// before attempting to prune the graph for zombie channels. We don't
|
||||
@@ -376,9 +377,9 @@ type Config struct {
|
||||
FirstTimePruneDelay time.Duration
|
||||
|
||||
// QueryBandwidth is a method that allows the router to query the lower
|
||||
// link layer to determine the up to date available bandwidth at a
|
||||
// link layer to determine the up-to-date available bandwidth at a
|
||||
// prospective link to be traversed. If the link isn't available, then
|
||||
// a value of zero should be returned. Otherwise, the current up to
|
||||
// a value of zero should be returned. Otherwise, the current up-to-
|
||||
// date knowledge of the available bandwidth of the link should be
|
||||
// returned.
|
||||
GetLink getLinkQuery
|
||||
@@ -389,7 +390,7 @@ type Config struct {
|
||||
// the switch can properly handle the HTLC.
|
||||
NextPaymentID func() (uint64, error)
|
||||
|
||||
// AssumeChannelValid toggles whether or not the router will check for
|
||||
// AssumeChannelValid toggles whether the router will check for
|
||||
// spentness of channel outpoints. For neutrino, this saves long rescans
|
||||
// from blocking initial usage of the daemon.
|
||||
AssumeChannelValid bool
|
||||
@@ -423,7 +424,7 @@ type EdgeLocator struct {
|
||||
Direction uint8
|
||||
}
|
||||
|
||||
// String returns a human readable version of the edgeLocator values.
|
||||
// String returns a human-readable version of the edgeLocator values.
|
||||
func (e *EdgeLocator) String() string {
|
||||
return fmt.Sprintf("%v:%v", e.ChannelID, e.Direction)
|
||||
}
|
||||
@@ -551,9 +552,10 @@ func (r *ChannelRouter) Start() error {
|
||||
// then we don't treat this as an explicit error.
|
||||
if _, _, err := r.cfg.Graph.PruneTip(); err != nil {
|
||||
switch {
|
||||
case err == channeldb.ErrGraphNeverPruned:
|
||||
case errors.Is(err, channeldb.ErrGraphNeverPruned):
|
||||
fallthrough
|
||||
case err == channeldb.ErrGraphNotFound:
|
||||
|
||||
case errors.Is(err, channeldb.ErrGraphNotFound):
|
||||
// If the graph has never been pruned, then we'll set
|
||||
// the prune height to the current best height of the
|
||||
// chain backend.
|
||||
@@ -563,6 +565,7 @@ func (r *ChannelRouter) Start() error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
default:
|
||||
return err
|
||||
}
|
||||
@@ -603,7 +606,10 @@ func (r *ChannelRouter) Start() error {
|
||||
// we may miss on-chain events as the filter hasn't properly
|
||||
// been applied.
|
||||
channelView, err := r.cfg.Graph.ChannelView()
|
||||
if err != nil && err != channeldb.ErrGraphNoEdgesFound {
|
||||
if err != nil && !errors.Is(
|
||||
err, channeldb.ErrGraphNoEdgesFound,
|
||||
) {
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -638,7 +644,10 @@ func (r *ChannelRouter) Start() error {
|
||||
// from the graph in order to ensure we maintain a tight graph
|
||||
// of "useful" nodes.
|
||||
err = r.cfg.Graph.PruneGraphNodes()
|
||||
if err != nil && err != channeldb.ErrGraphNodesNotFound {
|
||||
if err != nil && !errors.Is(
|
||||
err, channeldb.ErrGraphNodesNotFound,
|
||||
) {
|
||||
|
||||
return err
|
||||
}
|
||||
}
|
||||
@@ -707,13 +716,15 @@ func (r *ChannelRouter) Start() error {
|
||||
// result for the in-flight attempt is received.
|
||||
paySession := r.cfg.SessionSource.NewPaymentSessionEmpty()
|
||||
|
||||
// We pass in a zero timeout value, to indicate we
|
||||
// We pass in a non-timeout context, to indicate we
|
||||
// don't need it to timeout. It will stop immediately
|
||||
// after the existing attempt has finished anyway. We
|
||||
// also set a zero fee limit, as no more routes should
|
||||
// be tried.
|
||||
noTimeout := time.Duration(0)
|
||||
_, _, err := r.sendPayment(
|
||||
0, payment.Info.PaymentIdentifier, 0,
|
||||
context.Background(), 0,
|
||||
payment.Info.PaymentIdentifier, noTimeout,
|
||||
paySession, shardTracker,
|
||||
)
|
||||
if err != nil {
|
||||
@@ -776,8 +787,8 @@ func (r *ChannelRouter) syncGraphWithChain() error {
|
||||
switch {
|
||||
// If the graph has never been pruned, or hasn't fully been
|
||||
// created yet, then we don't treat this as an explicit error.
|
||||
case err == channeldb.ErrGraphNeverPruned:
|
||||
case err == channeldb.ErrGraphNotFound:
|
||||
case errors.Is(err, channeldb.ErrGraphNeverPruned):
|
||||
case errors.Is(err, channeldb.ErrGraphNotFound):
|
||||
default:
|
||||
return err
|
||||
}
|
||||
@@ -826,10 +837,12 @@ func (r *ChannelRouter) syncGraphWithChain() error {
|
||||
// can exit as this entails we are back to the point
|
||||
// where it hasn't seen any block or created channels,
|
||||
// alas there's nothing left to prune.
|
||||
case err == channeldb.ErrGraphNeverPruned:
|
||||
case errors.Is(err, channeldb.ErrGraphNeverPruned):
|
||||
return nil
|
||||
case err == channeldb.ErrGraphNotFound:
|
||||
|
||||
case errors.Is(err, channeldb.ErrGraphNotFound):
|
||||
return nil
|
||||
|
||||
default:
|
||||
return err
|
||||
}
|
||||
@@ -1047,7 +1060,10 @@ func (r *ChannelRouter) pruneZombieChans() error {
|
||||
}
|
||||
|
||||
for _, u := range oldEdges {
|
||||
filterPruneChans(u.Info, u.Policy1, u.Policy2)
|
||||
err = filterPruneChans(u.Info, u.Policy1, u.Policy2)
|
||||
if err != nil {
|
||||
log.Warnf("Filter pruning channels: %w\n", err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Infof("Pruning %v zombie channels", len(chansToPrune))
|
||||
@@ -1072,7 +1088,7 @@ func (r *ChannelRouter) pruneZombieChans() error {
|
||||
// With the channels pruned, we'll also attempt to prune any nodes that
|
||||
// were a part of them.
|
||||
err = r.cfg.Graph.PruneGraphNodes()
|
||||
if err != nil && err != channeldb.ErrGraphNodesNotFound {
|
||||
if err != nil && !errors.Is(err, channeldb.ErrGraphNodesNotFound) {
|
||||
return fmt.Errorf("unable to prune graph nodes: %w", err)
|
||||
}
|
||||
|
||||
@@ -1125,7 +1141,7 @@ func (r *ChannelRouter) handleNetworkUpdate(vb *ValidationBarrier,
|
||||
if err != nil {
|
||||
// We now decide to log an error or not. If allowDependents is
|
||||
// false, it means there is an error and the error is neither
|
||||
// ErrIgnored or ErrOutdated. In this case, we'll log an error.
|
||||
// ErrIgnored nor ErrOutdated. In this case, we'll log an error.
|
||||
// Otherwise, we'll add debug log only.
|
||||
if allowDependents {
|
||||
log.Debugf("process network updates got: %v", err)
|
||||
@@ -1225,7 +1241,7 @@ func (r *ChannelRouter) networkHandler() {
|
||||
|
||||
// Since this block is stale, we update our best height
|
||||
// to the previous block.
|
||||
blockHeight := uint32(chainUpdate.Height)
|
||||
blockHeight := chainUpdate.Height
|
||||
atomic.StoreUint32(&r.bestHeight, blockHeight-1)
|
||||
|
||||
// Update the channel graph to reflect that this block
|
||||
@@ -1485,9 +1501,9 @@ func (r *ChannelRouter) assertNodeAnnFreshness(node route.Vertex,
|
||||
}
|
||||
|
||||
// addZombieEdge adds a channel that failed complete validation into the zombie
|
||||
// index so we can avoid having to re-validate it in the future.
|
||||
// index, so we can avoid having to re-validate it in the future.
|
||||
func (r *ChannelRouter) addZombieEdge(chanID uint64) error {
|
||||
// If the edge fails validation we'll mark the edge itself as a zombie
|
||||
// If the edge fails validation we'll mark the edge itself as a zombie,
|
||||
// so we don't continue to request it. We use the "zero key" for both
|
||||
// node pubkeys so this edge can't be resurrected.
|
||||
var zeroKey [33]byte
|
||||
@@ -1598,7 +1614,10 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
_, _, exists, isZombie, err := r.cfg.Graph.HasChannelEdge(
|
||||
msg.ChannelID,
|
||||
)
|
||||
if err != nil && err != channeldb.ErrGraphNoEdgesFound {
|
||||
if err != nil && !errors.Is(
|
||||
err, channeldb.ErrGraphNoEdgesFound,
|
||||
) {
|
||||
|
||||
return errors.Errorf("unable to check for edge "+
|
||||
"existence: %v", err)
|
||||
}
|
||||
@@ -1655,7 +1674,7 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
case strings.Contains(err.Error(), "out of range"):
|
||||
// If the funding transaction isn't found at
|
||||
// all, then we'll mark the edge itself as a
|
||||
// zombie so we don't continue to request it.
|
||||
// zombie, so we don't continue to request it.
|
||||
// We use the "zero key" for both node pubkeys
|
||||
// so this edge can't be resurrected.
|
||||
zErr := r.addZombieEdge(msg.ChannelID)
|
||||
@@ -1681,8 +1700,8 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
return err
|
||||
}
|
||||
|
||||
// Next we'll validate that this channel is actually well
|
||||
// formed. If this check fails, then this channel either
|
||||
// Next we'll validate that this channel is actually
|
||||
// well-formed. If this check fails, then this channel either
|
||||
// doesn't exist, or isn't the one that was meant to be created
|
||||
// according to the passed channel proofs.
|
||||
fundingPoint, err := chanvalidate.Validate(&chanvalidate.Context{
|
||||
@@ -1693,7 +1712,7 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
FundingTx: fundingTx,
|
||||
})
|
||||
if err != nil {
|
||||
// Mark the edge as a zombie so we won't try to
|
||||
// Mark the edge as a zombie, so we won't try to
|
||||
// re-validate it on start up.
|
||||
if err := r.addZombieEdge(msg.ChannelID); err != nil {
|
||||
return err
|
||||
@@ -1705,7 +1724,7 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
|
||||
// Now that we have the funding outpoint of the channel, ensure
|
||||
// that it hasn't yet been spent. If so, then this channel has
|
||||
// been closed so we'll ignore it.
|
||||
// been closed, so we'll ignore it.
|
||||
chanUtxo, err := r.cfg.Chain.GetUtxo(
|
||||
fundingPoint, fundingPkScript, channelID.BlockHeight,
|
||||
r.quit,
|
||||
@@ -1740,7 +1759,7 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
|
||||
// As a new edge has been added to the channel graph, we'll
|
||||
// update the current UTXO filter within our active
|
||||
// FilteredChainView so we are notified if/when this channel is
|
||||
// FilteredChainView, so we are notified if/when this channel is
|
||||
// closed.
|
||||
filterUpdate := []channeldb.EdgePoint{
|
||||
{
|
||||
@@ -1768,7 +1787,10 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
|
||||
edge1Timestamp, edge2Timestamp, exists, isZombie, err :=
|
||||
r.cfg.Graph.HasChannelEdge(msg.ChannelID)
|
||||
if err != nil && err != channeldb.ErrGraphNoEdgesFound {
|
||||
if err != nil && !errors.Is(
|
||||
err, channeldb.ErrGraphNoEdgesFound,
|
||||
) {
|
||||
|
||||
return errors.Errorf("unable to check for edge "+
|
||||
"existence: %v", err)
|
||||
|
||||
@@ -1795,8 +1817,8 @@ func (r *ChannelRouter) processUpdate(msg interface{},
|
||||
}
|
||||
|
||||
// As edges are directional edge node has a unique policy for
|
||||
// the direction of the edge they control. Therefore we first
|
||||
// check if we already have the most up to date information for
|
||||
// the direction of the edge they control. Therefore, we first
|
||||
// check if we already have the most up-to-date information for
|
||||
// that edge. If this message has a timestamp not strictly
|
||||
// newer than what we already know of we can exit early.
|
||||
switch {
|
||||
@@ -2099,8 +2121,8 @@ func (r *ChannelRouter) FindRoute(req *RouteRequest) (*route.Route, float64,
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
// We'll fetch the current block height so we can properly calculate the
|
||||
// required HTLC time locks within the route.
|
||||
// We'll fetch the current block height, so we can properly calculate
|
||||
// the required HTLC time locks within the route.
|
||||
_, currentHeight, err := r.cfg.Chain.GetBestBlock()
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
@@ -2172,7 +2194,7 @@ func generateSphinxPacket(rt *route.Route, paymentHash []byte,
|
||||
// Now that we know we have an actual route, we'll map the route into a
|
||||
// sphinx payment path which includes per-hop payloads for each hop
|
||||
// that give each node within the route the necessary information
|
||||
// (fees, CLTV value, etc) to properly forward the payment.
|
||||
// (fees, CLTV value, etc.) to properly forward the payment.
|
||||
sphinxPath, err := rt.ToSphinxPath()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
@@ -2285,11 +2307,11 @@ type LightningPayment struct {
|
||||
LastHop *route.Vertex
|
||||
|
||||
// DestFeatures specifies the set of features we assume the final node
|
||||
// has for pathfinding. Typically these will be taken directly from an
|
||||
// has for pathfinding. Typically, these will be taken directly from an
|
||||
// invoice, but they can also be manually supplied or assumed by the
|
||||
// sender. If a nil feature vector is provided, the router will try to
|
||||
// fallback to the graph in order to load a feature vector for a node in
|
||||
// the public graph.
|
||||
// fall back to the graph in order to load a feature vector for a node
|
||||
// in the public graph.
|
||||
DestFeatures *lnwire.FeatureVector
|
||||
|
||||
// PaymentAddr is the payment address specified by the receiver. This
|
||||
@@ -2387,18 +2409,16 @@ func (r *ChannelRouter) SendPayment(payment *LightningPayment) ([32]byte,
|
||||
log.Tracef("Dispatching SendPayment for lightning payment: %v",
|
||||
spewPayment(payment))
|
||||
|
||||
// Since this is the first time this payment is being made, we pass nil
|
||||
// for the existing attempt.
|
||||
return r.sendPayment(
|
||||
payment.FeeLimit, payment.Identifier(),
|
||||
context.Background(), payment.FeeLimit, payment.Identifier(),
|
||||
payment.PayAttemptTimeout, paySession, shardTracker,
|
||||
)
|
||||
}
|
||||
|
||||
// SendPaymentAsync is the non-blocking version of SendPayment. The payment
|
||||
// result needs to be retrieved via the control tower.
|
||||
func (r *ChannelRouter) SendPaymentAsync(payment *LightningPayment,
|
||||
ps PaymentSession, st shards.ShardTracker) {
|
||||
func (r *ChannelRouter) SendPaymentAsync(ctx context.Context,
|
||||
payment *LightningPayment, ps PaymentSession, st shards.ShardTracker) {
|
||||
|
||||
// Since this is the first time this payment is being made, we pass nil
|
||||
// for the existing attempt.
|
||||
@@ -2410,7 +2430,7 @@ func (r *ChannelRouter) SendPaymentAsync(payment *LightningPayment,
|
||||
spewPayment(payment))
|
||||
|
||||
_, _, err := r.sendPayment(
|
||||
payment.FeeLimit, payment.Identifier(),
|
||||
ctx, payment.FeeLimit, payment.Identifier(),
|
||||
payment.PayAttemptTimeout, ps, st,
|
||||
)
|
||||
if err != nil {
|
||||
@@ -2520,8 +2540,8 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
||||
// Calculate amount paid to receiver.
|
||||
amt := rt.ReceiverAmt()
|
||||
|
||||
// If this is meant as a MP payment shard, we set the amount
|
||||
// for the creating info to the total amount of the payment.
|
||||
// If this is meant as an MP payment shard, we set the amount for the
|
||||
// creating info to the total amount of the payment.
|
||||
finalHop := rt.Hops[len(rt.Hops)-1]
|
||||
mpp := finalHop.MPP
|
||||
if mpp != nil {
|
||||
@@ -2575,7 +2595,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
||||
|
||||
// Since the HTLC hashes and preimages are specified manually over the
|
||||
// RPC for SendToRoute requests, we don't have to worry about creating
|
||||
// a ShardTracker that can generate hashes for AMP payments. Instead we
|
||||
// a ShardTracker that can generate hashes for AMP payments. Instead, we
|
||||
// create a simple tracker that can just return the hash for the single
|
||||
// shard we'll now launch.
|
||||
shardTracker := shards.NewSimpleShardTracker(htlcHash, nil)
|
||||
@@ -2585,9 +2605,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
||||
// - nil payment session (since we already have a route).
|
||||
// - no payment timeout.
|
||||
// - no current block height.
|
||||
p := newPaymentLifecycle(
|
||||
r, 0, paymentIdentifier, nil, shardTracker, 0, 0,
|
||||
)
|
||||
p := newPaymentLifecycle(r, 0, paymentIdentifier, nil, shardTracker, 0)
|
||||
|
||||
// We found a route to try, create a new HTLC attempt to try.
|
||||
//
|
||||
@@ -2608,7 +2626,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// We now lookup the payment to see if it's already failed.
|
||||
// We now look up the payment to see if it's already failed.
|
||||
payment, err := p.router.cfg.Control.FetchPayment(p.identifier)
|
||||
if err != nil {
|
||||
return result.attempt, err
|
||||
@@ -2677,15 +2695,27 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
||||
// returned.
|
||||
//
|
||||
// This method relies on the ControlTower's internal payment state machine to
|
||||
// carry out its execution. After restarts it is safe, and assumed, that the
|
||||
// carry out its execution. After restarts, it is safe, and assumed, that the
|
||||
// router will call this method for every payment still in-flight according to
|
||||
// the ControlTower.
|
||||
func (r *ChannelRouter) sendPayment(feeLimit lnwire.MilliSatoshi,
|
||||
identifier lntypes.Hash, timeout time.Duration,
|
||||
paySession PaymentSession,
|
||||
func (r *ChannelRouter) sendPayment(ctx context.Context,
|
||||
feeLimit lnwire.MilliSatoshi, identifier lntypes.Hash,
|
||||
paymentAttemptTimeout time.Duration, paySession PaymentSession,
|
||||
shardTracker shards.ShardTracker) ([32]byte, *route.Route, error) {
|
||||
|
||||
// We'll also fetch the current block height so we can properly
|
||||
// If the user provides a timeout, we will additionally wrap the context
|
||||
// in a deadline.
|
||||
cancel := func() {}
|
||||
if paymentAttemptTimeout > 0 {
|
||||
ctx, cancel = context.WithTimeout(ctx, paymentAttemptTimeout)
|
||||
}
|
||||
|
||||
// Since resumePayment is a blocking call, we'll cancel this
|
||||
// context if the payment completes before the optional
|
||||
// deadline.
|
||||
defer cancel()
|
||||
|
||||
// We'll also fetch the current block height, so we can properly
|
||||
// calculate the required HTLC time locks within the route.
|
||||
_, currentHeight, err := r.cfg.Chain.GetBestBlock()
|
||||
if err != nil {
|
||||
@@ -2695,11 +2725,11 @@ func (r *ChannelRouter) sendPayment(feeLimit lnwire.MilliSatoshi,
|
||||
// Now set up a paymentLifecycle struct with these params, such that we
|
||||
// can resume the payment from the current state.
|
||||
p := newPaymentLifecycle(
|
||||
r, feeLimit, identifier, paySession,
|
||||
shardTracker, timeout, currentHeight,
|
||||
r, feeLimit, identifier, paySession, shardTracker,
|
||||
currentHeight,
|
||||
)
|
||||
|
||||
return p.resumePayment()
|
||||
return p.resumePayment(ctx)
|
||||
}
|
||||
|
||||
// extractChannelUpdate examines the error and extracts the channel update.
|
||||
@@ -3022,8 +3052,8 @@ func (r *ChannelRouter) IsStaleEdgePolicy(chanID lnwire.ShortChannelID,
|
||||
}
|
||||
|
||||
// As edges are directional edge node has a unique policy for the
|
||||
// direction of the edge they control. Therefore we first check if we
|
||||
// already have the most up to date information for that edge. If so,
|
||||
// direction of the edge they control. Therefore, we first check if we
|
||||
// already have the most up-to-date information for that edge. If so,
|
||||
// then we can exit early.
|
||||
switch {
|
||||
// A flag set of 0 indicates this is an announcement for the "first"
|
||||
@@ -3054,7 +3084,7 @@ type ErrNoChannel struct {
|
||||
fromNode route.Vertex
|
||||
}
|
||||
|
||||
// Error returns a human readable string describing the error.
|
||||
// Error returns a human-readable string describing the error.
|
||||
func (e ErrNoChannel) Error() string {
|
||||
return fmt.Sprintf("no matching outgoing channel available for "+
|
||||
"node %v (%v)", e.position, e.fromNode)
|
||||
|
||||
Reference in new issue
Block a user