Merge pull request #5214 from carlaKC/4788-terminalshard
routing: handle failure to launch shard after permanent failure
This commit is contained in:
5 files changed
+641
-422
No files matched your search
@@ -290,18 +290,19 @@ func (p *PaymentControl) RegisterAttempt(paymentHash lntypes.Hash,
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
// Ensure the payment is in-flight.
|
|
||||||
if err := ensureInFlight(p); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// We cannot register a new attempt if the payment already has
|
// We cannot register a new attempt if the payment already has
|
||||||
// reached a terminal condition:
|
// reached a terminal condition. We check this before
|
||||||
|
// ensureInFlight because it is a more general check.
|
||||||
settle, fail := p.TerminalInfo()
|
settle, fail := p.TerminalInfo()
|
||||||
if settle != nil || fail != nil {
|
if settle != nil || fail != nil {
|
||||||
return ErrPaymentTerminal
|
return ErrPaymentTerminal
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Ensure the payment is in-flight.
|
||||||
|
if err := ensureInFlight(p); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
// Make sure any existing shards match the new one with regards
|
// Make sure any existing shards match the new one with regards
|
||||||
// to MPP options.
|
// to MPP options.
|
||||||
mpp := attempt.Route.FinalHop().MPP
|
mpp := attempt.Route.FinalHop().MPP
|
||||||
|
|||||||
@@ -1013,19 +1013,15 @@ func TestPaymentControlMultiShard(t *testing.T) {
|
|||||||
// up in the Succeeded state. If both failed the payment should
|
// up in the Succeeded state. If both failed the payment should
|
||||||
// also be Failed at this poinnt.
|
// also be Failed at this poinnt.
|
||||||
finalStatus := StatusFailed
|
finalStatus := StatusFailed
|
||||||
expRegErr := ErrPaymentAlreadyFailed
|
|
||||||
if test.settleFirst || test.settleLast {
|
if test.settleFirst || test.settleLast {
|
||||||
finalStatus = StatusSucceeded
|
finalStatus = StatusSucceeded
|
||||||
expRegErr = ErrPaymentAlreadySucceeded
|
|
||||||
}
|
}
|
||||||
|
|
||||||
assertPaymentStatus(t, pControl, info.PaymentHash, finalStatus)
|
assertPaymentStatus(t, pControl, info.PaymentHash, finalStatus)
|
||||||
|
|
||||||
// Finally assert we cannot register more attempts.
|
// Finally assert we cannot register more attempts.
|
||||||
_, err = pControl.RegisterAttempt(info.PaymentHash, &b)
|
_, err = pControl.RegisterAttempt(info.PaymentHash, &b)
|
||||||
if err != expRegErr {
|
require.Equal(t, ErrPaymentTerminal, err)
|
||||||
t.Fatalf("expected error %v, got: %v", expRegErr, err)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, test := range tests {
|
for _, test := range tests {
|
||||||
|
|||||||
+71
-33
@@ -83,7 +83,8 @@ func (m *mockPaymentAttemptDispatcher) setPaymentResult(
|
|||||||
}
|
}
|
||||||
|
|
||||||
type mockPaymentSessionSource struct {
|
type mockPaymentSessionSource struct {
|
||||||
routes []*route.Route
|
routes []*route.Route
|
||||||
|
routeRelease chan struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ PaymentSessionSource = (*mockPaymentSessionSource)(nil)
|
var _ PaymentSessionSource = (*mockPaymentSessionSource)(nil)
|
||||||
@@ -91,7 +92,10 @@ var _ PaymentSessionSource = (*mockPaymentSessionSource)(nil)
|
|||||||
func (m *mockPaymentSessionSource) NewPaymentSession(
|
func (m *mockPaymentSessionSource) NewPaymentSession(
|
||||||
_ *LightningPayment) (PaymentSession, error) {
|
_ *LightningPayment) (PaymentSession, error) {
|
||||||
|
|
||||||
return &mockPaymentSession{m.routes}, nil
|
return &mockPaymentSession{
|
||||||
|
routes: m.routes,
|
||||||
|
release: m.routeRelease,
|
||||||
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *mockPaymentSessionSource) NewPaymentSessionForRoute(
|
func (m *mockPaymentSessionSource) NewPaymentSessionForRoute(
|
||||||
@@ -137,6 +141,11 @@ func (m *mockMissionControl) GetProbability(fromNode, toNode route.Vertex,
|
|||||||
|
|
||||||
type mockPaymentSession struct {
|
type mockPaymentSession struct {
|
||||||
routes []*route.Route
|
routes []*route.Route
|
||||||
|
|
||||||
|
// release is a channel that optionally blocks requesting a route
|
||||||
|
// from our mock payment channel. If this value is nil, we will just
|
||||||
|
// release the route automatically.
|
||||||
|
release chan struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ PaymentSession = (*mockPaymentSession)(nil)
|
var _ PaymentSession = (*mockPaymentSession)(nil)
|
||||||
@@ -144,6 +153,10 @@ var _ PaymentSession = (*mockPaymentSession)(nil)
|
|||||||
func (m *mockPaymentSession) RequestRoute(_, _ lnwire.MilliSatoshi,
|
func (m *mockPaymentSession) RequestRoute(_, _ lnwire.MilliSatoshi,
|
||||||
_, height uint32) (*route.Route, error) {
|
_, height uint32) (*route.Route, error) {
|
||||||
|
|
||||||
|
if m.release != nil {
|
||||||
|
m.release <- struct{}{}
|
||||||
|
}
|
||||||
|
|
||||||
if len(m.routes) == 0 {
|
if len(m.routes) == 0 {
|
||||||
return nil, errNoPathFound
|
return nil, errNoPathFound
|
||||||
}
|
}
|
||||||
@@ -155,10 +168,9 @@ func (m *mockPaymentSession) RequestRoute(_, _ lnwire.MilliSatoshi,
|
|||||||
}
|
}
|
||||||
|
|
||||||
type mockPayer struct {
|
type mockPayer struct {
|
||||||
sendResult chan error
|
sendResult chan error
|
||||||
paymentResultErr chan error
|
paymentResult chan *htlcswitch.PaymentResult
|
||||||
paymentResult chan *htlcswitch.PaymentResult
|
quit chan struct{}
|
||||||
quit chan struct{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ PaymentAttemptDispatcher = (*mockPayer)(nil)
|
var _ PaymentAttemptDispatcher = (*mockPayer)(nil)
|
||||||
@@ -180,12 +192,16 @@ func (m *mockPayer) GetPaymentResult(paymentID uint64, _ lntypes.Hash,
|
|||||||
_ htlcswitch.ErrorDecrypter) (<-chan *htlcswitch.PaymentResult, error) {
|
_ htlcswitch.ErrorDecrypter) (<-chan *htlcswitch.PaymentResult, error) {
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case res := <-m.paymentResult:
|
case res, ok := <-m.paymentResult:
|
||||||
resChan := make(chan *htlcswitch.PaymentResult, 1)
|
resChan := make(chan *htlcswitch.PaymentResult, 1)
|
||||||
resChan <- res
|
if !ok {
|
||||||
|
close(resChan)
|
||||||
|
} else {
|
||||||
|
resChan <- res
|
||||||
|
}
|
||||||
|
|
||||||
return resChan, nil
|
return resChan, nil
|
||||||
case err := <-m.paymentResultErr:
|
|
||||||
return nil, err
|
|
||||||
case <-m.quit:
|
case <-m.quit:
|
||||||
return nil, fmt.Errorf("test quitting")
|
return nil, fmt.Errorf("test quitting")
|
||||||
}
|
}
|
||||||
@@ -248,13 +264,13 @@ func makeMockControlTower() *mockControlTower {
|
|||||||
func (m *mockControlTower) InitPayment(phash lntypes.Hash,
|
func (m *mockControlTower) InitPayment(phash lntypes.Hash,
|
||||||
c *channeldb.PaymentCreationInfo) error {
|
c *channeldb.PaymentCreationInfo) error {
|
||||||
|
|
||||||
m.Lock()
|
|
||||||
defer m.Unlock()
|
|
||||||
|
|
||||||
if m.init != nil {
|
if m.init != nil {
|
||||||
m.init <- initArgs{c}
|
m.init <- initArgs{c}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
m.Lock()
|
||||||
|
defer m.Unlock()
|
||||||
|
|
||||||
// Don't allow re-init a successful payment.
|
// Don't allow re-init a successful payment.
|
||||||
if _, ok := m.successful[phash]; ok {
|
if _, ok := m.successful[phash]; ok {
|
||||||
return channeldb.ErrAlreadyPaid
|
return channeldb.ErrAlreadyPaid
|
||||||
@@ -279,27 +295,49 @@ func (m *mockControlTower) InitPayment(phash lntypes.Hash,
|
|||||||
func (m *mockControlTower) RegisterAttempt(phash lntypes.Hash,
|
func (m *mockControlTower) RegisterAttempt(phash lntypes.Hash,
|
||||||
a *channeldb.HTLCAttemptInfo) error {
|
a *channeldb.HTLCAttemptInfo) error {
|
||||||
|
|
||||||
m.Lock()
|
|
||||||
defer m.Unlock()
|
|
||||||
|
|
||||||
if m.registerAttempt != nil {
|
if m.registerAttempt != nil {
|
||||||
m.registerAttempt <- registerAttemptArgs{a}
|
m.registerAttempt <- registerAttemptArgs{a}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Cannot register attempts for successful or failed payments.
|
m.Lock()
|
||||||
if _, ok := m.successful[phash]; ok {
|
defer m.Unlock()
|
||||||
return channeldb.ErrPaymentAlreadySucceeded
|
|
||||||
}
|
|
||||||
|
|
||||||
if _, ok := m.failed[phash]; ok {
|
|
||||||
return channeldb.ErrPaymentAlreadyFailed
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// Lookup payment.
|
||||||
p, ok := m.payments[phash]
|
p, ok := m.payments[phash]
|
||||||
if !ok {
|
if !ok {
|
||||||
return channeldb.ErrPaymentNotInitiated
|
return channeldb.ErrPaymentNotInitiated
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var inFlight bool
|
||||||
|
for _, a := range p.attempts {
|
||||||
|
if a.Settle != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
if a.Failure != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
inFlight = true
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cannot register attempts for successful or failed payments.
|
||||||
|
_, settled := m.successful[phash]
|
||||||
|
_, failed := m.failed[phash]
|
||||||
|
|
||||||
|
if settled || failed {
|
||||||
|
return channeldb.ErrPaymentTerminal
|
||||||
|
}
|
||||||
|
|
||||||
|
if settled && !inFlight {
|
||||||
|
return channeldb.ErrPaymentAlreadySucceeded
|
||||||
|
}
|
||||||
|
|
||||||
|
if failed && !inFlight {
|
||||||
|
return channeldb.ErrPaymentAlreadyFailed
|
||||||
|
}
|
||||||
|
|
||||||
|
// Add attempt to payment.
|
||||||
p.attempts = append(p.attempts, channeldb.HTLCAttempt{
|
p.attempts = append(p.attempts, channeldb.HTLCAttempt{
|
||||||
HTLCAttemptInfo: *a,
|
HTLCAttemptInfo: *a,
|
||||||
})
|
})
|
||||||
@@ -312,13 +350,13 @@ func (m *mockControlTower) SettleAttempt(phash lntypes.Hash,
|
|||||||
pid uint64, settleInfo *channeldb.HTLCSettleInfo) (
|
pid uint64, settleInfo *channeldb.HTLCSettleInfo) (
|
||||||
*channeldb.HTLCAttempt, error) {
|
*channeldb.HTLCAttempt, error) {
|
||||||
|
|
||||||
m.Lock()
|
|
||||||
defer m.Unlock()
|
|
||||||
|
|
||||||
if m.settleAttempt != nil {
|
if m.settleAttempt != nil {
|
||||||
m.settleAttempt <- settleAttemptArgs{settleInfo.Preimage}
|
m.settleAttempt <- settleAttemptArgs{settleInfo.Preimage}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
m.Lock()
|
||||||
|
defer m.Unlock()
|
||||||
|
|
||||||
// Only allow setting attempts if the payment is known.
|
// Only allow setting attempts if the payment is known.
|
||||||
p, ok := m.payments[phash]
|
p, ok := m.payments[phash]
|
||||||
if !ok {
|
if !ok {
|
||||||
@@ -353,13 +391,13 @@ func (m *mockControlTower) SettleAttempt(phash lntypes.Hash,
|
|||||||
func (m *mockControlTower) FailAttempt(phash lntypes.Hash, pid uint64,
|
func (m *mockControlTower) FailAttempt(phash lntypes.Hash, pid uint64,
|
||||||
failInfo *channeldb.HTLCFailInfo) (*channeldb.HTLCAttempt, error) {
|
failInfo *channeldb.HTLCFailInfo) (*channeldb.HTLCAttempt, error) {
|
||||||
|
|
||||||
m.Lock()
|
|
||||||
defer m.Unlock()
|
|
||||||
|
|
||||||
if m.failAttempt != nil {
|
if m.failAttempt != nil {
|
||||||
m.failAttempt <- failAttemptArgs{failInfo}
|
m.failAttempt <- failAttemptArgs{failInfo}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
m.Lock()
|
||||||
|
defer m.Unlock()
|
||||||
|
|
||||||
// Only allow failing attempts if the payment is known.
|
// Only allow failing attempts if the payment is known.
|
||||||
p, ok := m.payments[phash]
|
p, ok := m.payments[phash]
|
||||||
if !ok {
|
if !ok {
|
||||||
@@ -437,13 +475,13 @@ func (m *mockControlTower) FetchPayment(phash lntypes.Hash) (
|
|||||||
func (m *mockControlTower) FetchInFlightPayments() (
|
func (m *mockControlTower) FetchInFlightPayments() (
|
||||||
[]*channeldb.InFlightPayment, error) {
|
[]*channeldb.InFlightPayment, error) {
|
||||||
|
|
||||||
m.Lock()
|
|
||||||
defer m.Unlock()
|
|
||||||
|
|
||||||
if m.fetchInFlight != nil {
|
if m.fetchInFlight != nil {
|
||||||
m.fetchInFlight <- struct{}{}
|
m.fetchInFlight <- struct{}{}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
m.Lock()
|
||||||
|
defer m.Unlock()
|
||||||
|
|
||||||
// In flight are all payments not successful or failed.
|
// In flight are all payments not successful or failed.
|
||||||
var fl []*channeldb.InFlightPayment
|
var fl []*channeldb.InFlightPayment
|
||||||
for hash, p := range m.payments {
|
for hash, p := range m.payments {
|
||||||
|
|||||||
@@ -115,6 +115,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
|
|
||||||
// We'll continue until either our payment succeeds, or we encounter a
|
// We'll continue until either our payment succeeds, or we encounter a
|
||||||
// critical error during path finding.
|
// critical error during path finding.
|
||||||
|
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.
|
||||||
@@ -171,7 +172,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
if err := shardHandler.waitForShard(); err != nil {
|
if err := shardHandler.waitForShard(); err != nil {
|
||||||
return [32]byte{}, nil, err
|
return [32]byte{}, nil, err
|
||||||
}
|
}
|
||||||
continue
|
continue lifecycle
|
||||||
}
|
}
|
||||||
|
|
||||||
// Before we attempt any new shard, we'll check to see if
|
// Before we attempt any new shard, we'll check to see if
|
||||||
@@ -195,7 +196,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
return [32]byte{}, nil, saveErr
|
return [32]byte{}, nil, saveErr
|
||||||
}
|
}
|
||||||
|
|
||||||
continue
|
continue lifecycle
|
||||||
|
|
||||||
case <-p.router.quit:
|
case <-p.router.quit:
|
||||||
return [32]byte{}, nil, ErrRouterShuttingDown
|
return [32]byte{}, nil, ErrRouterShuttingDown
|
||||||
@@ -234,7 +235,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
return [32]byte{}, nil, saveErr
|
return [32]byte{}, nil, saveErr
|
||||||
}
|
}
|
||||||
|
|
||||||
continue
|
continue lifecycle
|
||||||
}
|
}
|
||||||
|
|
||||||
// We still have active shards, we'll wait for an
|
// We still have active shards, we'll wait for an
|
||||||
@@ -242,12 +243,23 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
if err := shardHandler.waitForShard(); err != nil {
|
if err := shardHandler.waitForShard(); err != nil {
|
||||||
return [32]byte{}, nil, err
|
return [32]byte{}, nil, err
|
||||||
}
|
}
|
||||||
continue
|
continue lifecycle
|
||||||
}
|
}
|
||||||
|
|
||||||
// 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)
|
attempt, outcome, err := shardHandler.launchShard(rt)
|
||||||
if err != nil {
|
switch {
|
||||||
|
// We may get a terminal error if we've processed a shard with
|
||||||
|
// a terminal state (settled or permanent failure), while we
|
||||||
|
// were pathfinding. We know we're in a terminal state here,
|
||||||
|
// so we can continue and wait for our last shards to return.
|
||||||
|
case err == channeldb.ErrPaymentTerminal:
|
||||||
|
log.Infof("Payment: %v in terminal state, abandoning "+
|
||||||
|
"shard", p.paymentHash)
|
||||||
|
|
||||||
|
continue lifecycle
|
||||||
|
|
||||||
|
case err != nil:
|
||||||
return [32]byte{}, nil, err
|
return [32]byte{}, nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -270,7 +282,7 @@ func (p *paymentLifecycle) resumePayment() ([32]byte, *route.Route, error) {
|
|||||||
|
|
||||||
// Error was handled successfully, continue to make a
|
// Error was handled successfully, continue to make a
|
||||||
// new attempt.
|
// new attempt.
|
||||||
continue
|
continue lifecycle
|
||||||
}
|
}
|
||||||
|
|
||||||
// Now that the shard was successfully sent, launch a go
|
// Now that the shard was successfully sent, launch a go
|
||||||
|
|||||||
+544
-372
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user