htlcswitch+router: define PaymentResult, GetPaymentResult

This lets us distinguish an critical error from a actual payment result
(success or failure). This is important since we know that we can only
attempt another payment when a final result from the previous payment
attempt is received.
This commit is contained in:
Johan T. Halseth 2019-05-16 15:27:29 +02:00
parent be129eb7c7
commit ec087a9f73
No known key found for this signature in database
GPG Key ID: 15BAADA29DA20D26
7 changed files with 251 additions and 68 deletions

@ -54,6 +54,8 @@ func newConcurrentTester(t *testing.T) *concurrentTester {
} }
func (c *concurrentTester) Fatalf(format string, args ...interface{}) { func (c *concurrentTester) Fatalf(format string, args ...interface{}) {
c.T.Helper()
c.mtx.Lock() c.mtx.Lock()
defer c.mtx.Unlock() defer c.mtx.Unlock()
@ -1108,13 +1110,30 @@ func TestChannelLinkMultiHopUnknownPaymentHash(t *testing.T) {
} }
// Send payment and expose err channel. // Send payment and expose err channel.
_, err = n.aliceServer.htlcSwitch.SendHTLC( err = n.aliceServer.htlcSwitch.SendHTLC(
n.firstBobChannelLink.ShortChanID(), pid, htlc, n.firstBobChannelLink.ShortChanID(), pid, htlc,
newMockDeobfuscator(), newMockDeobfuscator(),
) )
if !strings.Contains(err.Error(), lnwire.CodeUnknownPaymentHash.String()) { if err != nil {
t.Fatalf("expected %v got %v", err, t.Fatalf("unable to get send payment: %v", err)
lnwire.CodeUnknownPaymentHash) }
resultChan, err := n.aliceServer.htlcSwitch.GetPaymentResult(pid)
if err != nil {
t.Fatalf("unable to get payment result: %v", err)
}
var result *PaymentResult
select {
case result = <-resultChan:
case <-time.After(5 * time.Second):
t.Fatalf("no result arrive")
}
fErr := result.Error
if !strings.Contains(fErr.Error(), lnwire.CodeUnknownPaymentHash.String()) {
t.Fatalf("expected %v got %v", lnwire.CodeUnknownPaymentHash, fErr)
} }
// Wait for Alice to receive the revocation. // Wait for Alice to receive the revocation.
@ -3867,7 +3886,7 @@ func TestChannelLinkAcceptDuplicatePayment(t *testing.T) {
// With the invoice now added to Carol's registry, we'll send the // With the invoice now added to Carol's registry, we'll send the
// payment. It should succeed w/o any issues as it has been crafted // payment. It should succeed w/o any issues as it has been crafted
// properly. // properly.
_, err = n.aliceServer.htlcSwitch.SendHTLC( err = n.aliceServer.htlcSwitch.SendHTLC(
n.firstBobChannelLink.ShortChanID(), pid, htlc, n.firstBobChannelLink.ShortChanID(), pid, htlc,
newMockDeobfuscator(), newMockDeobfuscator(),
) )
@ -3875,9 +3894,23 @@ func TestChannelLinkAcceptDuplicatePayment(t *testing.T) {
t.Fatalf("unable to send payment to carol: %v", err) t.Fatalf("unable to send payment to carol: %v", err)
} }
resultChan, err := n.aliceServer.htlcSwitch.GetPaymentResult(pid)
if err != nil {
t.Fatalf("unable to get payment result: %v", err)
}
select {
case result := <-resultChan:
if result.Error != nil {
t.Fatalf("payment failed: %v", result.Error)
}
case <-time.After(5 * time.Second):
t.Fatalf("payment result did not arrive")
}
// Now, if we attempt to send the payment *again* it should be rejected // Now, if we attempt to send the payment *again* it should be rejected
// as it's a duplicate request. // as it's a duplicate request.
_, err = n.aliceServer.htlcSwitch.SendHTLC( err = n.aliceServer.htlcSwitch.SendHTLC(
n.firstBobChannelLink.ShortChanID(), pid, htlc, n.firstBobChannelLink.ShortChanID(), pid, htlc,
newMockDeobfuscator(), newMockDeobfuscator(),
) )

@ -0,0 +1,25 @@
package htlcswitch
import "errors"
var (
// ErrPaymentIDNotFound is an error returned if the given paymentID is
// not found.
ErrPaymentIDNotFound = errors.New("paymentID not found")
// ErrPaymentIDAlreadyExists is returned if we try to write a pending
// payment whose paymentID already exists.
ErrPaymentIDAlreadyExists = errors.New("paymentID already exists")
)
// PaymentResult wraps a result received from the network after a payment
// attempt was made.
type PaymentResult struct {
// Preimage is set by the switch in case a sent HTLC was settled.
Preimage [32]byte
// Error is non-nil in case a HTLC send failed, and the HTLC is now
// irrevocably cancelled. If the payment failed during forwarding, this
// error will be a *ForwardingError.
Error error
}

@ -71,8 +71,7 @@ type pendingPayment struct {
paymentHash lntypes.Hash paymentHash lntypes.Hash
amount lnwire.MilliSatoshi amount lnwire.MilliSatoshi
preimage chan [sha256.Size]byte resultChan chan *PaymentResult
err chan error
// deobfuscator is a serializable entity which is used if we received // deobfuscator is a serializable entity which is used if we received
// an error, it deobfuscates the onion failure blob, and extracts the // an error, it deobfuscates the onion failure blob, and extracts the
@ -346,32 +345,52 @@ func (s *Switch) ProcessContractResolution(msg contractcourt.ResolutionMsg) erro
return nil return nil
} }
// GetPaymentResult returns the the result of the payment attempt with the
// given paymentID. The method returns a channel where the payment result will
// be sent when available, or an error is encountered. If the paymentID is
// unknown, ErrPaymentIDNotFound will be returned.
func (s *Switch) GetPaymentResult(paymentID uint64) (<-chan *PaymentResult, error) {
s.pendingMutex.Lock()
payment, ok := s.pendingPayments[paymentID]
s.pendingMutex.Unlock()
if !ok {
return nil, ErrPaymentIDNotFound
}
return payment.resultChan, nil
}
// SendHTLC is used by other subsystems which aren't belong to htlc switch // SendHTLC is used by other subsystems which aren't belong to htlc switch
// package in order to send the htlc update. The paymentID used MUST be unique // package in order to send the htlc update. The paymentID used MUST be unique
// for this HTLC, and MUST be used only once, otherwise the switch might reject // for this HTLC, and MUST be used only once, otherwise the switch might reject
// it. // it.
func (s *Switch) SendHTLC(firstHop lnwire.ShortChannelID, paymentID uint64, func (s *Switch) SendHTLC(firstHop lnwire.ShortChannelID, paymentID uint64,
htlc *lnwire.UpdateAddHTLC, htlc *lnwire.UpdateAddHTLC, deobfuscator ErrorDecrypter) error {
deobfuscator ErrorDecrypter) ([sha256.Size]byte, error) {
// Before sending, double check that we don't already have 1) an // Before sending, double check that we don't already have 1) an
// in-flight payment to this payment hash, or 2) a complete payment for // in-flight payment to this payment hash, or 2) a complete payment for
// the same hash. // the same hash.
if err := s.control.ClearForTakeoff(htlc); err != nil { if err := s.control.ClearForTakeoff(htlc); err != nil {
return zeroPreimage, err return err
} }
// Create payment and add to the map of payment in order later to be // Create payment and add to the map of payment in order later to be
// able to retrieve it and return response to the user. // able to retrieve it and return response to the user.
payment := &pendingPayment{ payment := &pendingPayment{
err: make(chan error, 1), resultChan: make(chan *PaymentResult, 1),
preimage: make(chan [sha256.Size]byte, 1),
paymentHash: htlc.PaymentHash, paymentHash: htlc.PaymentHash,
amount: htlc.Amount, amount: htlc.Amount,
deobfuscator: deobfuscator, deobfuscator: deobfuscator,
} }
s.pendingMutex.Lock() s.pendingMutex.Lock()
if _, ok := s.pendingPayments[paymentID]; ok {
s.pendingMutex.Unlock()
return ErrPaymentIDAlreadyExists
}
s.pendingPayments[paymentID] = payment s.pendingPayments[paymentID] = payment
s.pendingMutex.Unlock() s.pendingMutex.Unlock()
@ -388,32 +407,13 @@ func (s *Switch) SendHTLC(firstHop lnwire.ShortChannelID, paymentID uint64,
if err := s.forward(packet); err != nil { if err := s.forward(packet); err != nil {
s.removePendingPayment(paymentID) s.removePendingPayment(paymentID)
if err := s.control.Fail(htlc.PaymentHash); err != nil { if err := s.control.Fail(htlc.PaymentHash); err != nil {
return zeroPreimage, err return err
} }
return zeroPreimage, err return err
} }
// Returns channels so that other subsystem might wait/skip the return nil
// waiting of handling of payment.
var preimage [sha256.Size]byte
var err error
select {
case e := <-payment.err:
err = e
case <-s.quit:
return zeroPreimage, ErrSwitchExiting
}
select {
case p := <-payment.preimage:
preimage = p
case <-s.quit:
return zeroPreimage, ErrSwitchExiting
}
return preimage, err
} }
// UpdateForwardingPolicies sends a message to the switch to update the // UpdateForwardingPolicies sends a message to the switch to update the
@ -880,10 +880,7 @@ func (s *Switch) handleLocalResponse(pkt *htlcPacket) {
// has been restarted since sending the payment. // has been restarted since sending the payment.
payment := s.findPayment(pkt.incomingHTLCID) payment := s.findPayment(pkt.incomingHTLCID)
var ( var result *PaymentResult
preimage [32]byte
paymentErr error
)
switch htlc := pkt.htlc.(type) { switch htlc := pkt.htlc.(type) {
@ -900,7 +897,9 @@ func (s *Switch) handleLocalResponse(pkt *htlcPacket) {
return return
} }
preimage = htlc.PaymentPreimage result = &PaymentResult{
Preimage: htlc.PaymentPreimage,
}
// We've received a fail update which means we can finalize the user // We've received a fail update which means we can finalize the user
// payment and return fail response. // payment and return fail response.
@ -918,10 +917,13 @@ func (s *Switch) handleLocalResponse(pkt *htlcPacket) {
// The error reason will be unencypted in case this a local // The error reason will be unencypted in case this a local
// failure or a converted error. // failure or a converted error.
unencrypted := pkt.localFailure || pkt.convertedError unencrypted := pkt.localFailure || pkt.convertedError
paymentErr = s.parseFailedPayment( paymentErr := s.parseFailedPayment(
payment, pkt.incomingHTLCID, payment.paymentHash, payment, pkt.incomingHTLCID, payment.paymentHash,
unencrypted, pkt.isResolution, htlc, unencrypted, pkt.isResolution, htlc,
) )
result = &PaymentResult{
Error: paymentErr,
}
default: default:
log.Warnf("Received unknown response type: %T", pkt.htlc) log.Warnf("Received unknown response type: %T", pkt.htlc)
@ -931,9 +933,7 @@ func (s *Switch) handleLocalResponse(pkt *htlcPacket) {
// Deliver the payment error and preimage to the application, if it is // Deliver the payment error and preimage to the application, if it is
// waiting for a response. // waiting for a response.
if payment != nil { if payment != nil {
payment.err <- paymentErr payment.resultChan <- result
payment.preimage <- preimage
s.removePendingPayment(pkt.incomingHTLCID)
} }
} }

@ -1417,7 +1417,7 @@ func testSkipLinkLocalForward(t *testing.T, eligible bool,
// We'll attempt to send out a new HTLC that has Alice as the first // We'll attempt to send out a new HTLC that has Alice as the first
// outgoing link. This should fail as Alice isn't yet able to forward // outgoing link. This should fail as Alice isn't yet able to forward
// any active HTLC's. // any active HTLC's.
_, err = s.SendHTLC(aliceChannelLink.ShortChanID(), 0, addMsg, nil) err = s.SendHTLC(aliceChannelLink.ShortChanID(), 0, addMsg, nil)
if err == nil { if err == nil {
t.Fatalf("local forward should fail due to inactive link") t.Fatalf("local forward should fail due to inactive link")
} }
@ -1738,14 +1738,39 @@ func TestSwitchSendPayment(t *testing.T) {
PaymentHash: rhash, PaymentHash: rhash,
Amount: 1, Amount: 1,
} }
paymentID := uint64(123)
// First check that the switch will correctly respond that this payment
// ID is unknown.
_, err = s.GetPaymentResult(paymentID)
if err != ErrPaymentIDNotFound {
t.Fatalf("expected ErrPaymentIDNotFound, got %v", err)
}
// Handle the request and checks that bob channel link received it. // Handle the request and checks that bob channel link received it.
errChan := make(chan error) errChan := make(chan error)
go func() { go func() {
_, err := s.SendHTLC( err := s.SendHTLC(
aliceChannelLink.ShortChanID(), 0, update, aliceChannelLink.ShortChanID(), paymentID, update,
newMockDeobfuscator()) newMockDeobfuscator())
if err != nil {
errChan <- err errChan <- err
return
}
resultChan, err := s.GetPaymentResult(paymentID)
if err != nil {
errChan <- err
return
}
result := <-resultChan
if result.Error != nil {
errChan <- result.Error
return
}
errChan <- nil
}() }()
select { select {

@ -794,10 +794,23 @@ func preparePayment(sendingPeer, receivingPeer lnpeer.Peer,
// Send payment and expose err channel. // Send payment and expose err channel.
return invoice, func() error { return invoice, func() error {
_, err := sender.htlcSwitch.SendHTLC( err := sender.htlcSwitch.SendHTLC(
firstHop, pid, htlc, newMockDeobfuscator(), firstHop, pid, htlc, newMockDeobfuscator(),
) )
if err != nil {
return err return err
}
resultChan, err := sender.htlcSwitch.GetPaymentResult(pid)
if err != nil {
return err
}
result := <-resultChan
if result.Error != nil {
return result.Error
}
return nil
}, nil }, nil
} }
@ -1261,10 +1274,26 @@ func (n *twoHopNetwork) makeHoldPayment(sendingPeer, receivingPeer lnpeer.Peer,
// Send payment and expose err channel. // Send payment and expose err channel.
go func() { go func() {
_, err := sender.htlcSwitch.SendHTLC( err := sender.htlcSwitch.SendHTLC(
firstHop, pid, htlc, newMockDeobfuscator(), firstHop, pid, htlc, newMockDeobfuscator(),
) )
if err != nil {
paymentErr <- err paymentErr <- err
return
}
resultChan, err := sender.htlcSwitch.GetPaymentResult(pid)
if err != nil {
paymentErr <- err
return
}
result := <-resultChan
if result.Error != nil {
paymentErr <- result.Error
return
}
paymentErr <- nil
}() }()
return paymentErr return paymentErr

@ -1,28 +1,61 @@
package routing package routing
import ( import (
"crypto/sha256"
"github.com/lightningnetwork/lnd/htlcswitch" "github.com/lightningnetwork/lnd/htlcswitch"
"github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/lnwire"
) )
type mockPaymentAttemptDispatcher struct { type mockPaymentAttemptDispatcher struct {
onPayment func(firstHop lnwire.ShortChannelID) ([32]byte, error) onPayment func(firstHop lnwire.ShortChannelID) ([32]byte, error)
results map[uint64]*htlcswitch.PaymentResult
} }
var _ PaymentAttemptDispatcher = (*mockPaymentAttemptDispatcher)(nil) var _ PaymentAttemptDispatcher = (*mockPaymentAttemptDispatcher)(nil)
func (m *mockPaymentAttemptDispatcher) SendHTLC(firstHop lnwire.ShortChannelID, func (m *mockPaymentAttemptDispatcher) SendHTLC(firstHop lnwire.ShortChannelID,
_ uint64, pid uint64,
_ *lnwire.UpdateAddHTLC, _ *lnwire.UpdateAddHTLC,
_ htlcswitch.ErrorDecrypter) ([sha256.Size]byte, error) { _ htlcswitch.ErrorDecrypter) error {
if m.onPayment != nil { if m.onPayment == nil {
return m.onPayment(firstHop) return nil
} }
return [sha256.Size]byte{}, nil if m.results == nil {
m.results = make(map[uint64]*htlcswitch.PaymentResult)
}
var result *htlcswitch.PaymentResult
preimage, err := m.onPayment(firstHop)
if err != nil {
fwdErr, ok := err.(*htlcswitch.ForwardingError)
if !ok {
return err
}
result = &htlcswitch.PaymentResult{
Error: fwdErr,
}
} else {
result = &htlcswitch.PaymentResult{Preimage: preimage}
}
m.results[pid] = result
return nil
}
func (m *mockPaymentAttemptDispatcher) GetPaymentResult(paymentID uint64) (
<-chan *htlcswitch.PaymentResult, error) {
c := make(chan *htlcswitch.PaymentResult, 1)
res, ok := m.results[paymentID]
if !ok {
return nil, htlcswitch.ErrPaymentIDNotFound
}
c <- res
return c, nil
} }
func (m *mockPaymentAttemptDispatcher) setPaymentResult( func (m *mockPaymentAttemptDispatcher) setPaymentResult(

@ -2,7 +2,6 @@ package routing
import ( import (
"bytes" "bytes"
"crypto/sha256"
"fmt" "fmt"
"runtime" "runtime"
"sync" "sync"
@ -136,7 +135,15 @@ type PaymentAttemptDispatcher interface {
SendHTLC(firstHop lnwire.ShortChannelID, SendHTLC(firstHop lnwire.ShortChannelID,
paymentID uint64, paymentID uint64,
htlcAdd *lnwire.UpdateAddHTLC, htlcAdd *lnwire.UpdateAddHTLC,
deobfuscator htlcswitch.ErrorDecrypter) ([sha256.Size]byte, error) deobfuscator htlcswitch.ErrorDecrypter) error
// GetPaymentResult returns the the result of the payment attempt with
// the given paymentID. The method returns a channel where the payment
// result will be sent when available, or an error is encountered. If
// the paymentID is unknown, htlcswitch.ErrPaymentIDNotFound will be
// returned.
GetPaymentResult(paymentID uint64) (
<-chan *htlcswitch.PaymentResult, error)
} }
// FeeSchema is the set fee configuration for a Lightning Node on the network. // FeeSchema is the set fee configuration for a Lightning Node on the network.
@ -1711,19 +1718,50 @@ func (r *ChannelRouter) sendPaymentAttempt(paySession *paymentSession,
return [32]byte{}, true, err return [32]byte{}, true, err
} }
preimage, err := r.cfg.Payer.SendHTLC( err = r.cfg.Payer.SendHTLC(
firstHop, paymentID, htlcAdd, errorDecryptor, firstHop, paymentID, htlcAdd, errorDecryptor,
) )
if err == nil { if err != nil {
return preimage, true, nil log.Errorf("Failed sending attempt %d for payment %x to "+
} "switch: %v", paymentID, paymentHash, err)
log.Errorf("Attempt to send payment %x failed: %v", // We must inspect the error to know whether it was critical or
paymentHash, err) // not, to decide whether we should continue trying.
finalOutcome := r.processSendError(
finalOutcome := r.processSendError(paySession, route, err) paySession, route, err,
)
return [32]byte{}, finalOutcome, err return [32]byte{}, finalOutcome, err
}
// Now ask the switch to return the result of the payment when
// available.
resultChan, err := r.cfg.Payer.GetPaymentResult(paymentID)
if err != nil {
log.Errorf("Failed getting result for paymentID %d "+
"from switch: %v", paymentID, err)
return [32]byte{}, true, err
}
var result *htlcswitch.PaymentResult
select {
case result = <-resultChan:
case <-r.quit:
return [32]byte{}, true, ErrRouterShuttingDown
}
if result.Error != nil {
log.Errorf("Attempt to send payment %x failed: %v",
paymentHash, result.Error)
finalOutcome := r.processSendError(
paySession, route, result.Error,
)
return [32]byte{}, finalOutcome, result.Error
}
return result.Preimage, true, nil
} }
// processSendError analyzes the error for the payment attempt received from the // processSendError analyzes the error for the payment attempt received from the