routing: remove the abstraction shardHandler
This commit removes the unclear abstraction `shardHandler` that's used in our payment lifecycle. As we'll see in the following commits, `shardHandler` is an unnecessary layer and everything can be cleanly managed inside `paymentLifecycle`.
This commit is contained in:
2 files changed
+46
-59
No files matched your search
@@ -30,6 +30,19 @@ type paymentLifecycle struct {
|
|||||||
shardTracker shards.ShardTracker
|
shardTracker shards.ShardTracker
|
||||||
timeoutChan <-chan time.Time
|
timeoutChan <-chan time.Time
|
||||||
currentHeight int32
|
currentHeight int32
|
||||||
|
|
||||||
|
// shardErrors is a channel where errors collected by calling
|
||||||
|
// collectResultAsync will be delivered. These results are meant to be
|
||||||
|
// inspected by calling waitForShard or checkShards, and the channel
|
||||||
|
// doesn't need to be initiated if the caller is using the sync
|
||||||
|
// collectResult directly.
|
||||||
|
// TODO(yy): delete.
|
||||||
|
shardErrors chan error
|
||||||
|
|
||||||
|
// quit is closed to signal the sub goroutines of the payment lifecycle
|
||||||
|
// to stop.
|
||||||
|
quit chan struct{}
|
||||||
|
wg sync.WaitGroup
|
||||||
}
|
}
|
||||||
|
|
||||||
// newPaymentLifecycle initiates a new payment lifecycle and returns it.
|
// newPaymentLifecycle initiates a new payment lifecycle and returns it.
|
||||||
@@ -45,6 +58,8 @@ func newPaymentLifecycle(r *ChannelRouter, feeLimit lnwire.MilliSatoshi,
|
|||||||
paySession: paySession,
|
paySession: paySession,
|
||||||
shardTracker: shardTracker,
|
shardTracker: shardTracker,
|
||||||
currentHeight: currentHeight,
|
currentHeight: currentHeight,
|
||||||
|
shardErrors: make(chan error),
|
||||||
|
quit: make(chan struct{}),
|
||||||
}
|
}
|
||||||
|
|
||||||
// If a timeout is specified, create a timeout channel. If no timeout is
|
// If a timeout is specified, create a timeout channel. If no timeout is
|
||||||
@@ -78,19 +93,10 @@ func (p *paymentLifecycle) calcFeeBudget(
|
|||||||
|
|
||||||
// resumePayment resumes the paymentLifecycle from the current state.
|
// resumePayment resumes the paymentLifecycle from the current state.
|
||||||
func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
||||||
shardHandler := &shardHandler{
|
|
||||||
router: p.router,
|
|
||||||
identifier: p.identifier,
|
|
||||||
shardTracker: p.shardTracker,
|
|
||||||
shardErrors: make(chan error),
|
|
||||||
quit: make(chan struct{}),
|
|
||||||
paySession: p.paySession,
|
|
||||||
}
|
|
||||||
|
|
||||||
// When the payment lifecycle loop exits, we make sure to signal any
|
// When the payment lifecycle loop exits, we make sure to signal any
|
||||||
// sub goroutine of the shardHandler to exit, then wait for them to
|
// sub goroutine of the HTLC attempt to exit, then wait for them to
|
||||||
// return.
|
// return.
|
||||||
defer shardHandler.stop()
|
defer p.stop()
|
||||||
|
|
||||||
// If we had any existing attempts outstanding, we'll start by spinning
|
// If we had any existing attempts outstanding, we'll start by spinning
|
||||||
// up goroutines that'll collect their results and deliver them to the
|
// up goroutines that'll collect their results and deliver them to the
|
||||||
@@ -106,7 +112,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
log.Infof("Resuming payment shard %v for payment %v",
|
log.Infof("Resuming payment shard %v for payment %v",
|
||||||
a.AttemptID, p.identifier)
|
a.AttemptID, p.identifier)
|
||||||
|
|
||||||
shardHandler.collectResultAsync(&a)
|
p.collectResultAsync(&a)
|
||||||
}
|
}
|
||||||
|
|
||||||
// exitWithErr is a helper closure that logs and returns an error.
|
// exitWithErr is a helper closure that logs and returns an error.
|
||||||
@@ -122,7 +128,7 @@ lifecycle:
|
|||||||
for {
|
for {
|
||||||
// Start by quickly checking if there are any outcomes already
|
// Start by quickly checking if there are any outcomes already
|
||||||
// available to handle before we reevaluate our state.
|
// available to handle before we reevaluate our state.
|
||||||
if err := shardHandler.checkShards(); err != nil {
|
if err := p.checkShards(); err != nil {
|
||||||
return exitWithErr(err)
|
return exitWithErr(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -187,7 +193,7 @@ lifecycle:
|
|||||||
// We still have outstanding shards, so wait for a new
|
// We still have outstanding shards, so wait for a new
|
||||||
// outcome to be available before re-evaluating our
|
// outcome to be available before re-evaluating our
|
||||||
// state.
|
// state.
|
||||||
if err := shardHandler.waitForShard(); err != nil {
|
if err := p.waitForShard(); err != nil {
|
||||||
return exitWithErr(err)
|
return exitWithErr(err)
|
||||||
}
|
}
|
||||||
continue lifecycle
|
continue lifecycle
|
||||||
@@ -259,7 +265,7 @@ lifecycle:
|
|||||||
|
|
||||||
// We still have active shards, we'll wait for an
|
// We still have active shards, we'll wait for an
|
||||||
// outcome to be available before retrying.
|
// outcome to be available before retrying.
|
||||||
if err := shardHandler.waitForShard(); err != nil {
|
if err := p.waitForShard(); err != nil {
|
||||||
return exitWithErr(err)
|
return exitWithErr(err)
|
||||||
}
|
}
|
||||||
continue lifecycle
|
continue lifecycle
|
||||||
@@ -272,7 +278,7 @@ lifecycle:
|
|||||||
lastShard := rt.ReceiverAmt() == ps.RemainingAmt
|
lastShard := rt.ReceiverAmt() == ps.RemainingAmt
|
||||||
|
|
||||||
// We found a route to try, launch a new shard.
|
// We found a route to try, launch a new shard.
|
||||||
attempt, outcome, err := shardHandler.launchShard(rt, lastShard)
|
attempt, outcome, err := p.launchShard(rt, lastShard)
|
||||||
switch {
|
switch {
|
||||||
// We may get a terminal error if we've processed a shard with
|
// We may get a terminal error if we've processed a shard with
|
||||||
// a terminal state (settled or permanent failure), while we
|
// a terminal state (settled or permanent failure), while we
|
||||||
@@ -298,7 +304,7 @@ lifecycle:
|
|||||||
// We must inspect the error to know whether it was
|
// We must inspect the error to know whether it was
|
||||||
// critical or not, to decide whether we should
|
// critical or not, to decide whether we should
|
||||||
// continue trying.
|
// continue trying.
|
||||||
err := shardHandler.handleSwitchErr(
|
err := p.handleSwitchErr(
|
||||||
attempt, outcome.err,
|
attempt, outcome.err,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -312,39 +318,18 @@ lifecycle:
|
|||||||
|
|
||||||
// Now that the shard was successfully sent, launch a go
|
// Now that the shard was successfully sent, launch a go
|
||||||
// routine that will handle its result when its back.
|
// routine that will handle its result when its back.
|
||||||
shardHandler.collectResultAsync(attempt)
|
p.collectResultAsync(attempt)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// shardHandler holds what is necessary to send and collect the result of
|
|
||||||
// shards.
|
|
||||||
type shardHandler struct {
|
|
||||||
identifier lntypes.Hash
|
|
||||||
router *ChannelRouter
|
|
||||||
shardTracker shards.ShardTracker
|
|
||||||
paySession PaymentSession
|
|
||||||
|
|
||||||
// shardErrors is a channel where errors collected by calling
|
|
||||||
// collectResultAsync will be delivered. These results are meant to be
|
|
||||||
// inspected by calling waitForShard or checkShards, and the channel
|
|
||||||
// doesn't need to be initiated if the caller is using the sync
|
|
||||||
// collectResult directly.
|
|
||||||
shardErrors chan error
|
|
||||||
|
|
||||||
// quit is closed to signal the sub goroutines of the payment lifecycle
|
|
||||||
// to stop.
|
|
||||||
quit chan struct{}
|
|
||||||
wg sync.WaitGroup
|
|
||||||
}
|
|
||||||
|
|
||||||
// stop signals any active shard goroutine to exit and waits for them to exit.
|
// stop signals any active shard goroutine to exit and waits for them to exit.
|
||||||
func (p *shardHandler) stop() {
|
func (p *paymentLifecycle) stop() {
|
||||||
close(p.quit)
|
close(p.quit)
|
||||||
p.wg.Wait()
|
p.wg.Wait()
|
||||||
}
|
}
|
||||||
|
|
||||||
// waitForShard blocks until any of the outstanding shards return.
|
// waitForShard blocks until any of the outstanding shards return.
|
||||||
func (p *shardHandler) waitForShard() error {
|
func (p *paymentLifecycle) waitForShard() error {
|
||||||
select {
|
select {
|
||||||
case err := <-p.shardErrors:
|
case err := <-p.shardErrors:
|
||||||
return err
|
return err
|
||||||
@@ -359,7 +344,7 @@ func (p *shardHandler) waitForShard() error {
|
|||||||
|
|
||||||
// checkShards is a non-blocking method that check if any shards has finished
|
// checkShards is a non-blocking method that check if any shards has finished
|
||||||
// their execution.
|
// their execution.
|
||||||
func (p *shardHandler) checkShards() error {
|
func (p *paymentLifecycle) checkShards() error {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case err := <-p.shardErrors:
|
case err := <-p.shardErrors:
|
||||||
@@ -400,7 +385,7 @@ type launchOutcome struct {
|
|||||||
// whether the attempt was successfully sent. If the launchOutcome wraps a
|
// whether the attempt was successfully sent. If the launchOutcome wraps a
|
||||||
// non-nil error, it means that the attempt was not sent onto the network, so
|
// non-nil error, it means that the attempt was not sent onto the network, so
|
||||||
// no result will be available in the future for it.
|
// no result will be available in the future for it.
|
||||||
func (p *shardHandler) launchShard(rt *route.Route,
|
func (p *paymentLifecycle) launchShard(rt *route.Route,
|
||||||
lastShard bool) (*channeldb.HTLCAttempt, *launchOutcome, error) {
|
lastShard bool) (*channeldb.HTLCAttempt, *launchOutcome, error) {
|
||||||
|
|
||||||
// Using the route received from the payment session, create a new
|
// Using the route received from the payment session, create a new
|
||||||
@@ -455,7 +440,7 @@ type shardResult struct {
|
|||||||
// collectResultAsync launches a goroutine that will wait for the result of the
|
// collectResultAsync launches a goroutine that will wait for the result of the
|
||||||
// given HTLC attempt to be available then handle its result. It will fail the
|
// given HTLC attempt to be available then handle its result. It will fail the
|
||||||
// payment with the control tower if a terminal error is encountered.
|
// payment with the control tower if a terminal error is encountered.
|
||||||
func (p *shardHandler) collectResultAsync(attempt *channeldb.HTLCAttempt) {
|
func (p *paymentLifecycle) collectResultAsync(attempt *channeldb.HTLCAttempt) {
|
||||||
// errToSend is the error to be sent to sh.shardErrors.
|
// errToSend is the error to be sent to sh.shardErrors.
|
||||||
var errToSend error
|
var errToSend error
|
||||||
|
|
||||||
@@ -510,7 +495,7 @@ func (p *shardHandler) collectResultAsync(attempt *channeldb.HTLCAttempt) {
|
|||||||
// collectResult waits for the result for the given attempt to be available
|
// collectResult waits for the result for the given attempt to be available
|
||||||
// from the Switch, then records the attempt outcome with the control tower. A
|
// from the Switch, then records the attempt outcome with the control tower. A
|
||||||
// shardResult is returned, indicating the final outcome of this HTLC attempt.
|
// shardResult is returned, indicating the final outcome of this HTLC attempt.
|
||||||
func (p *shardHandler) collectResult(attempt *channeldb.HTLCAttempt) (
|
func (p *paymentLifecycle) collectResult(attempt *channeldb.HTLCAttempt) (
|
||||||
*shardResult, error) {
|
*shardResult, error) {
|
||||||
|
|
||||||
// We'll retrieve the hash specific to this shard from the
|
// We'll retrieve the hash specific to this shard from the
|
||||||
@@ -632,7 +617,7 @@ func (p *shardHandler) collectResult(attempt *channeldb.HTLCAttempt) (
|
|||||||
}
|
}
|
||||||
|
|
||||||
// createNewPaymentAttempt creates a new payment attempt from the given route.
|
// createNewPaymentAttempt creates a new payment attempt from the given route.
|
||||||
func (p *shardHandler) createNewPaymentAttempt(rt *route.Route,
|
func (p *paymentLifecycle) createNewPaymentAttempt(rt *route.Route,
|
||||||
lastShard bool) (*channeldb.HTLCAttempt, error) {
|
lastShard bool) (*channeldb.HTLCAttempt, error) {
|
||||||
|
|
||||||
// Generate a new key to be used for this attempt.
|
// Generate a new key to be used for this attempt.
|
||||||
@@ -683,7 +668,7 @@ func (p *shardHandler) createNewPaymentAttempt(rt *route.Route,
|
|||||||
// sendAttempt attempts to send the current attempt to the switch to complete
|
// sendAttempt attempts to send the current attempt to the switch to complete
|
||||||
// the payment. If this attempt fails, then we'll continue on to the next
|
// the payment. If this attempt fails, then we'll continue on to the next
|
||||||
// available route.
|
// available route.
|
||||||
func (p *shardHandler) sendAttempt(
|
func (p *paymentLifecycle) sendAttempt(
|
||||||
attempt *channeldb.HTLCAttempt) error {
|
attempt *channeldb.HTLCAttempt) error {
|
||||||
|
|
||||||
log.Tracef("Attempting to send payment %v (pid=%v), "+
|
log.Tracef("Attempting to send payment %v (pid=%v), "+
|
||||||
@@ -744,7 +729,7 @@ func (p *shardHandler) sendAttempt(
|
|||||||
// the error type, the error is either the final outcome of the payment or we
|
// the error type, the error is either the final outcome of the payment or we
|
||||||
// need to continue with an alternative route. A final outcome is indicated by
|
// need to continue with an alternative route. A final outcome is indicated by
|
||||||
// a non-nil reason value.
|
// a non-nil reason value.
|
||||||
func (p *shardHandler) handleSwitchErr(attempt *channeldb.HTLCAttempt,
|
func (p *paymentLifecycle) handleSwitchErr(attempt *channeldb.HTLCAttempt,
|
||||||
sendErr error) error {
|
sendErr error) error {
|
||||||
|
|
||||||
internalErrorReason := channeldb.FailureReasonError
|
internalErrorReason := channeldb.FailureReasonError
|
||||||
@@ -841,7 +826,7 @@ func (p *shardHandler) handleSwitchErr(attempt *channeldb.HTLCAttempt,
|
|||||||
|
|
||||||
// handleFailureMessage tries to apply a channel update present in the failure
|
// handleFailureMessage tries to apply a channel update present in the failure
|
||||||
// message if any.
|
// message if any.
|
||||||
func (p *shardHandler) handleFailureMessage(rt *route.Route,
|
func (p *paymentLifecycle) handleFailureMessage(rt *route.Route,
|
||||||
errorSourceIdx int, failure lnwire.FailureMessage) error {
|
errorSourceIdx int, failure lnwire.FailureMessage) error {
|
||||||
|
|
||||||
if failure == nil {
|
if failure == nil {
|
||||||
@@ -913,7 +898,7 @@ func (p *shardHandler) handleFailureMessage(rt *route.Route,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// failAttempt calls control tower to fail the current payment attempt.
|
// failAttempt calls control tower to fail the current payment attempt.
|
||||||
func (p *shardHandler) failAttempt(attemptID uint64,
|
func (p *paymentLifecycle) failAttempt(attemptID uint64,
|
||||||
sendError error) (*channeldb.HTLCAttempt, error) {
|
sendError error) (*channeldb.HTLCAttempt, error) {
|
||||||
|
|
||||||
log.Warnf("Attempt %v for payment %v failed: %v", attemptID,
|
log.Warnf("Attempt %v for payment %v failed: %v", attemptID,
|
||||||
|
|||||||
+11
-9
@@ -2496,15 +2496,17 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
|||||||
// shard we'll now launch.
|
// shard we'll now launch.
|
||||||
shardTracker := shards.NewSimpleShardTracker(htlcHash, nil)
|
shardTracker := shards.NewSimpleShardTracker(htlcHash, nil)
|
||||||
|
|
||||||
// Launch a shard along the given route.
|
// Create a payment lifecycle using the given route with,
|
||||||
sh := &shardHandler{
|
// - zero fee limit as we are not requesting routes.
|
||||||
router: r,
|
// - nil payment session (since we already have a route).
|
||||||
identifier: paymentIdentifier,
|
// - no payment timeout.
|
||||||
shardTracker: shardTracker,
|
// - no current block height.
|
||||||
}
|
p := newPaymentLifecycle(
|
||||||
|
r, 0, paymentIdentifier, nil, shardTracker, 0, 0,
|
||||||
|
)
|
||||||
|
|
||||||
var shardError error
|
var shardError error
|
||||||
attempt, outcome, err := sh.launchShard(rt, false)
|
attempt, outcome, err := p.launchShard(rt, false)
|
||||||
|
|
||||||
// With SendToRoute, it can happen that the route exceeds protocol
|
// With SendToRoute, it can happen that the route exceeds protocol
|
||||||
// constraints. Mark the payment as failed with an internal error.
|
// constraints. Mark the payment as failed with an internal error.
|
||||||
@@ -2536,7 +2538,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
|||||||
|
|
||||||
// Shard successfully launched, wait for the result to be available.
|
// Shard successfully launched, wait for the result to be available.
|
||||||
default:
|
default:
|
||||||
result, err := sh.collectResult(attempt)
|
result, err := p.collectResult(attempt)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -2556,7 +2558,7 @@ func (r *ChannelRouter) sendToRoute(htlcHash lntypes.Hash, rt *route.Route,
|
|||||||
// the error to check if it maps into a terminal error code, if not use
|
// the error to check if it maps into a terminal error code, if not use
|
||||||
// a generic NO_ROUTE error.
|
// a generic NO_ROUTE error.
|
||||||
var failureReason *channeldb.FailureReason
|
var failureReason *channeldb.FailureReason
|
||||||
err = sh.handleSwitchErr(attempt, shardError)
|
err = p.handleSwitchErr(attempt, shardError)
|
||||||
|
|
||||||
switch {
|
switch {
|
||||||
// If a non-terminal error is returned and `skipTempErr` is false, then
|
// If a non-terminal error is returned and `skipTempErr` is false, then
|
||||||
|
|||||||
Reference in new issue
Block a user