Skip to content

MPP routing improvements #1219

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
t-bast merged 6 commits into from
Nov 21, 2019
Merged

MPP routing improvements #1219

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9307953
MPP split/retry improvements
t-bast Nov 20, 2019
c0a57d5
Fix flaky MPP test in IntegrationSpec:
t-bast Nov 20, 2019
a9b269e
MPP: rename WAIT_FOR_CHANNEL_BALANCES
t-bast Nov 20, 2019
f91f93a
Add a (true) WAIT_FOR_CHANNEL_BALANCES state
t-bast Nov 20, 2019
a999ca8
Apply suggested refactoring
t-bast Nov 20, 2019
08f8ca5
Update comment
t-bast Nov 20, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 91 additions & 24 deletions eclair-core/src/main/scala/fr/acinq/eclair/payment/send/MultiPartPaymentLifecycle.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,13 @@ import fr.acinq.eclair.channel.Commitments
import fr.acinq.eclair.crypto.Sphinx
import fr.acinq.eclair.payment.PaymentRequest.ExtraHop
import fr.acinq.eclair.payment.PaymentSent.PartialPayment
import fr.acinq.eclair.payment.relay.Relayer.{GetOutgoingChannels, OutgoingChannel, OutgoingChannels}
import fr.acinq.eclair.payment._
import fr.acinq.eclair.payment.relay.Relayer.{GetOutgoingChannels, OutgoingChannel, OutgoingChannels}
import fr.acinq.eclair.payment.send.PaymentInitiator.SendPaymentConfig
import fr.acinq.eclair.payment.send.PaymentLifecycle.SendPayment
import fr.acinq.eclair.router._
import fr.acinq.eclair.wire.{Onion, OnionRoutingPacket, OnionTlv, PaymentTimeout, UpdateAddHtlc}
import fr.acinq.eclair.{CltvExpiry, FSMDiagnosticActorLogging, Logs, LongToBtcAmount, MilliSatoshi, NodeParams, ToMilliSatoshiConversion}
import fr.acinq.eclair.wire._
import fr.acinq.eclair.{CltvExpiry, FSMDiagnosticActorLogging, Logs, LongToBtcAmount, MilliSatoshi, NodeParams, ShortChannelId, ToMilliSatoshiConversion}
import scodec.bits.ByteVector

import scala.annotation.tailrec
Expand Down Expand Up @@ -72,29 +72,54 @@ class MultiPartPaymentLifecycle(nodeParams: NodeParams, cfg: SendPaymentConfig,
router ! TickComputeNetworkStats
}
relayer ! GetOutgoingChannels()
goto(WAIT_FOR_CHANNEL_BALANCES) using PaymentProgress(d.sender, d.request, s.stats, d.request.totalAmount, d.request.maxAttempts, Map.empty, Nil)
goto(WAIT_FOR_CHANNEL_BALANCES) using WaitingForChannelBalances(d.sender, d.request, s.stats)
}

when(WAIT_FOR_CHANNEL_BALANCES) {
case Event(OutgoingChannels(channels), d: PaymentProgress) =>
log.debug("trying to send {} with local channels: {}", d.toSend, channels.map(_.toUsableBalance).mkString(","))
val randomize = d.failures.nonEmpty // we randomize channel selection when we retry
val (remaining, payments) = splitPayment(nodeParams, d.toSend, channels, d.networkStats, d.request, randomize)
case Event(OutgoingChannels(channels), d: WaitingForChannelBalances) =>
log.debug("trying to send {} with local channels: {}", d.request.totalAmount, channels.map(_.toUsableBalance).mkString(","))
val (remaining, payments) = splitPayment(nodeParams, d.request.totalAmount, channels, d.networkStats, d.request, randomize = false)
if (remaining > 0.msat) {
log.warning(s"cannot send ${d.toSend} with our current balance")
goto(PAYMENT_ABORTED) using PaymentAborted(d.sender, d.request, d.failures :+ LocalFailure(new RuntimeException("balance is too low")), d.pending.keySet)
log.warning(s"cannot send ${d.request.totalAmount} with our current balance")
goto(PAYMENT_ABORTED) using PaymentAborted(d.sender, d.request, LocalFailure(BalanceTooLow) :: Nil, Set.empty)
} else {
val pending = setFees(d.request.routeParams, payments, payments.size + d.pending.size)
val pending = setFees(d.request.routeParams, payments, payments.size)
pending.foreach { case (childId, payment) => spawnChildPaymentFsm(childId) ! payment }
goto(PAYMENT_IN_PROGRESS) using d.copy(toSend = 0 msat, remainingAttempts = d.remainingAttempts - 1, pending = d.pending ++ pending)
goto(PAYMENT_IN_PROGRESS) using PaymentProgress(d.sender, d.request, d.networkStats, channels.length, 0 msat, d.request.maxAttempts - 1, pending, Set.empty, Nil)
}
}

when(PAYMENT_IN_PROGRESS) {
case Event(pf: PaymentFailed, d: PaymentProgress) => handleChildFailure(pf, d) match {
case Some(paymentAborted) =>
goto(PAYMENT_ABORTED) using paymentAborted
case None =>
// Get updated local channels (will take into account the child payments that are in-flight).
relayer ! GetOutgoingChannels()
val failedPayment = d.pending(pf.id)
stay using d.copy(toSend = d.toSend + failedPayment.finalPayload.amount, pending = d.pending - pf.id, failures = d.failures ++ pf.failures)
val shouldBlacklist = shouldBlacklistChannel(pf)
if (shouldBlacklist) {
log.debug(s"ignoring channel ${getFirstHopShortChannelId(failedPayment)} to ${failedPayment.routePrefix.head.nextNodeId}")
}
val ignoreChannels = if (shouldBlacklist) d.ignoreChannels + getFirstHopShortChannelId(failedPayment) else d.ignoreChannels
val remainingAttempts = if (shouldBlacklist && Random.nextDouble() * math.log(d.channelsCount) > 2.0) {
// When we have a lot of channels, many of them may end up being a bad route prefix for the destination we're
// trying to reach. This is a cheap error that is detected quickly (RouteNotFound), so we don't want to count
// it in our payment attempts to avoid failing too fast.
// However we don't want to test all of our channels either which would be expensive, so we only probabilistically
// count the failure in our payment attempts.
// With the log-scale used, here are the probabilities and the corresponding number of retries:
// * 10 channels -> refund 13% of failures -> with 5 initial retries we will actually try 5/(1-0.13) = ~6 times
// * 20 channels -> refund 32% of failures -> with 5 initial retries we will actually try 5/(1-0.32) = ~7 times
// * 50 channels -> refund 50% of failures -> with 5 initial retries we will actually try 5/(1-0.50) = ~10 times
// * 100 channels -> refund 56% of failures -> with 5 initial retries we will actually try 5/(1-0.56) = ~11 times
// * 1000 channels -> refund 70% of failures -> with 5 initial retries we will actually try 5/(1-0.70) = ~17 times
// NB: this hack won't be necessary once multi-part is directly handled by the router.
d.remainingAttempts + 1
} else {
d.remainingAttempts
}
goto(RETRY_WITH_UPDATED_BALANCES) using d.copy(toSend = d.toSend + failedPayment.finalPayload.amount, pending = d.pending - pf.id, failures = d.failures ++ pf.failures, ignoreChannels = ignoreChannels, remainingAttempts = remainingAttempts)
}

case Event(ps: PaymentSent, d: PaymentProgress) =>
Expand All @@ -103,15 +128,27 @@ class MultiPartPaymentLifecycle(nodeParams: NodeParams, cfg: SendPaymentConfig,
goto(PAYMENT_SUCCEEDED) using PaymentSucceeded(d.sender, d.request, ps.paymentPreimage, ps.parts, d.pending.keySet - ps.id)
}

when(PAYMENT_IN_PROGRESS) {
when(RETRY_WITH_UPDATED_BALANCES) {
case Event(OutgoingChannels(channels), d: PaymentProgress) =>
log.debug("trying to send {} with local channels: {}", d.toSend, channels.map(_.toUsableBalance).mkString(","))
val filteredChannels = channels.filter(c => !d.ignoreChannels.contains(c.channelUpdate.shortChannelId))
val (remaining, payments) = splitPayment(nodeParams, d.toSend, filteredChannels, d.networkStats, d.request, randomize = true) // we randomize channel selection when we retry
if (remaining > 0.msat) {
log.warning(s"cannot send ${d.toSend} with our current balance")
goto(PAYMENT_ABORTED) using PaymentAborted(d.sender, d.request, d.failures :+ LocalFailure(BalanceTooLow), d.pending.keySet)
} else {
val pending = setFees(d.request.routeParams, payments, payments.size + d.pending.size)
pending.foreach { case (childId, payment) => spawnChildPaymentFsm(childId) ! payment }
goto(PAYMENT_IN_PROGRESS) using d.copy(toSend = 0 msat, remainingAttempts = d.remainingAttempts - 1, pending = d.pending ++ pending, channelsCount = channels.length)
}

case Event(pf: PaymentFailed, d: PaymentProgress) => handleChildFailure(pf, d) match {
case Some(paymentAborted) =>
goto(PAYMENT_ABORTED) using paymentAborted
case None =>
// Get updated local channels (will take into account the child payments that are in-flight).
relayer ! GetOutgoingChannels()
val failedPayment = d.pending(pf.id)
goto(WAIT_FOR_CHANNEL_BALANCES) using d.copy(toSend = d.toSend + failedPayment.finalPayload.amount, pending = d.pending - pf.id, failures = d.failures ++ pf.failures)
val ignoreChannels = if (shouldBlacklistChannel(pf)) d.ignoreChannels + getFirstHopShortChannelId(failedPayment) else d.ignoreChannels
stay using d.copy(toSend = d.toSend + failedPayment.finalPayload.amount, pending = d.pending - pf.id, failures = d.failures ++ pf.failures, ignoreChannels = ignoreChannels)
}

case Event(ps: PaymentSent, d: PaymentProgress) =>
Expand Down Expand Up @@ -203,7 +240,7 @@ class MultiPartPaymentLifecycle(nodeParams: NodeParams, cfg: SendPaymentConfig,
if (paymentTimedOut) {
Some(PaymentAborted(d.sender, d.request, d.failures ++ pf.failures, d.pending.keySet - pf.id))
} else if (d.remainingAttempts == 0) {
val failure = LocalFailure(new RuntimeException("payment attempts exhausted without success"))
val failure = LocalFailure(RetryExhausted)
Some(PaymentAborted(d.sender, d.request, d.failures ++ pf.failures :+ failure, d.pending.keySet - pf.id))
} else {
None
Expand Down Expand Up @@ -234,12 +271,18 @@ object MultiPartPaymentLifecycle {
require(totalAmount > 0.msat, s"total amount must be > 0")
}

// @formatter:off
object BalanceTooLow extends RuntimeException("outbound capacity is too low")
object RetryExhausted extends RuntimeException("payment attempts exhausted without success")
// @formatter:on

// @formatter:off
sealed trait State
case object WAIT_FOR_PAYMENT_REQUEST extends State
case object WAIT_FOR_NETWORK_STATS extends State
case object WAIT_FOR_CHANNEL_BALANCES extends State
case object WAIT_FOR_CHANNEL_BALANCES extends State
case object PAYMENT_IN_PROGRESS extends State
case object RETRY_WITH_UPDATED_BALANCES extends State
case object PAYMENT_ABORTED extends State
case object PAYMENT_SUCCEEDED extends State

Expand All @@ -251,23 +294,33 @@ object MultiPartPaymentLifecycle {
/**
* During initialization, we collect network statistics to help us decide how to best split a big payment.
*
* @param sender the sender of the payment request.
* @param request payment request containing the total amount to send.
*/
case class WaitingForNetworkStats(sender: ActorRef, request: SendMultiPartPayment) extends Data
/**
* During initialization, we request our local channels balances.
*
* @param sender the sender of the payment request.
* @param request payment request containing the total amount to send.
* @param networkStats network statistics help us decide how to best split a big payment.
*/
case class WaitingForNetworkStats(sender: ActorRef, request: SendMultiPartPayment) extends Data
case class WaitingForChannelBalances(sender: ActorRef, request: SendMultiPartPayment, networkStats: Option[NetworkStats]) extends Data
/**
* While the payment is in progress, we listen to child payment failures. When we receive such failures, we request
* our up-to-date local channels balances and retry the failed child payments with a potentially different route.
*
* @param sender the sender of the payment request.
* @param request payment request containing the total amount to send.
* @param networkStats network statistics help us decide how to best split a big payment.
* @param channelsCount number of local channels.
* @param toSend remaining amount that should be split and sent.
* @param remainingAttempts remaining attempts (after child payments fail).
* @param pending pending child payments (payment sent, we are waiting for a fulfill or a failure).
* @param ignoreChannels channels that should be ignored (previously returned a permanent error).
* @param failures previous child payment failures.
*/
case class PaymentProgress(sender: ActorRef, request: SendMultiPartPayment, networkStats: Option[NetworkStats], toSend: MilliSatoshi, remainingAttempts: Int, pending: Map[UUID, SendPayment], failures: Seq[PaymentFailure]) extends Data
case class PaymentProgress(sender: ActorRef, request: SendMultiPartPayment, networkStats: Option[NetworkStats], channelsCount: Int, toSend: MilliSatoshi, remainingAttempts: Int, pending: Map[UUID, SendPayment], ignoreChannels: Set[ShortChannelId], failures: Seq[PaymentFailure]) extends Data
/**
* When we exhaust our retry attempts without success, we abort the payment.
* Once we're in that state, we wait for all the pending child payments to settle.
Expand All @@ -292,6 +345,17 @@ object MultiPartPaymentLifecycle {
case class PaymentSucceeded(sender: ActorRef, request: SendMultiPartPayment, preimage: ByteVector32, parts: Seq[PartialPayment], pending: Set[UUID]) extends Data
// @formatter:on

/** If the payment failed immediately with a RouteNotFound, the channel we selected should be ignored in retries. */
private def shouldBlacklistChannel(pf: PaymentFailed): Boolean = pf.failures match {
case LocalFailure(RouteNotFound) :: Nil => true
case _ => false
}

def getFirstHopShortChannelId(payment: SendPayment): ShortChannelId = {
require(payment.routePrefix.nonEmpty, "multi-part payment must have a route prefix")
payment.routePrefix.head.lastUpdate.shortChannelId
}

/**
* If fee limits are provided, we need to divide them between all child payments. Otherwise we could end up paying
* N * maxFee (where N is the number of child payments).
Expand Down Expand Up @@ -415,14 +479,17 @@ object MultiPartPaymentLifecycle {
// If we have direct channels to the target, we use them without splitting the payment inside each channel.
val channelsToTarget = localChannels.filter(p => p.nextNodeId == request.targetNodeId).sortBy(_.commitments.availableBalanceForSend)
val directPayments = split(toSend, Seq.empty, channelsToTarget, (remaining: MilliSatoshi, channel: OutgoingChannel) => {
createChildPayment(nodeParams, request, remaining.min(channel.commitments.availableBalanceForSend), channel) :: Nil
// When using direct channels to the destination, it doesn't make sense to use retries so we set maxAttempts to 1.
createChildPayment(nodeParams, request.copy(maxAttempts = 1), remaining.min(channel.commitments.availableBalanceForSend), channel) :: Nil
})

// Otherwise we need to split the amount based on network statistics and pessimistic fees estimates.
// We filter out unannounced channels: they are very likely leading to a non-routing node.
// Note that this will be handled more gracefully once this logic is migrated inside the router.
val channels = if (randomize) {
Random.shuffle(localChannels.filter(p => p.nextNodeId != request.targetNodeId))
Random.shuffle(localChannels.filter(p => p.commitments.announceChannel && p.nextNodeId != request.targetNodeId))
} else {
localChannels.filter(p => p.nextNodeId != request.targetNodeId).sortBy(_.commitments.availableBalanceForSend)
localChannels.filter(p => p.commitments.announceChannel && p.nextNodeId != request.targetNodeId).sortBy(_.commitments.availableBalanceForSend)
}
val remotePayments = split(toSend - directPayments.map(_.finalPayload.amount).sum, Seq.empty, channels, (remaining: MilliSatoshi, channel: OutgoingChannel) => {
// We re-generate a split threshold for each channel to randomize the amounts.
Expand Down
35 changes: 16 additions & 19 deletions eclair-core/src/main/scala/fr/acinq/eclair/payment/send/PaymentInitiator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,27 +42,24 @@ class PaymentInitiator(nodeParams: NodeParams, router: ActorRef, relayer: ActorR
override def receive: Receive = {
case r: SendPaymentRequest =>
val paymentId = UUID.randomUUID()
sender ! paymentId
val paymentCfg = SendPaymentConfig(paymentId, paymentId, r.externalId, r.paymentHash, r.targetNodeId, r.paymentRequest, storeInDb = true, publishEvent = true)
val finalExpiry = r.finalExpiry(nodeParams.currentBlockHeight)
if (r.paymentRequest.exists(!_.features.supported)) {
sender ! paymentId
sender ! PaymentFailed(paymentId, r.paymentHash, LocalFailure(new IllegalArgumentException(s"can't send payment: unknown invoice features (${r.paymentRequest.get.features})")) :: Nil)
} else {
r.paymentRequest match {
case Some(invoice) if invoice.features.allowMultiPart =>
r.predefinedRoute match {
case Nil => spawnMultiPartPaymentFsm(paymentCfg) forward SendMultiPartPayment(r.paymentHash, invoice.paymentSecret.get, r.targetNodeId, r.amount, finalExpiry, r.maxAttempts, r.assistedRoutes, r.routeParams)
case hops => spawnPaymentFsm(paymentCfg) forward SendPaymentToRoute(r.paymentHash, hops, Onion.createMultiPartPayload(r.amount, invoice.amount.getOrElse(r.amount), finalExpiry, invoice.paymentSecret.get))
}
case _ =>
val payFsm = spawnPaymentFsm(paymentCfg)
// NB: we only generate legacy payment onions for now for maximum compatibility.
r.predefinedRoute match {
case Nil => payFsm forward SendPayment(r.paymentHash, r.targetNodeId, FinalLegacyPayload(r.amount, finalExpiry), r.maxAttempts, r.assistedRoutes, r.routeParams)
case hops => payFsm forward SendPaymentToRoute(r.paymentHash, hops, FinalLegacyPayload(r.amount, finalExpiry))
}
}
sender ! paymentId
r.paymentRequest match {
case Some(invoice) if !invoice.features.supported =>
sender ! PaymentFailed(paymentId, r.paymentHash, LocalFailure(new IllegalArgumentException(s"can't send payment: unknown invoice features (${r.paymentRequest.get.features})")) :: Nil)
case Some(invoice) if invoice.features.allowMultiPart =>
r.predefinedRoute match {
case Nil => spawnMultiPartPaymentFsm(paymentCfg) forward SendMultiPartPayment(r.paymentHash, invoice.paymentSecret.get, r.targetNodeId, r.amount, finalExpiry, r.maxAttempts, r.assistedRoutes, r.routeParams)
case hops => spawnPaymentFsm(paymentCfg) forward SendPaymentToRoute(r.paymentHash, hops, Onion.createMultiPartPayload(r.amount, invoice.amount.getOrElse(r.amount), finalExpiry, invoice.paymentSecret.get))
}
case _ =>
val payFsm = spawnPaymentFsm(paymentCfg)
// NB: we only generate legacy payment onions for now for maximum compatibility.
r.predefinedRoute match {
case Nil => payFsm forward SendPayment(r.paymentHash, r.targetNodeId, FinalLegacyPayload(r.amount, finalExpiry), r.maxAttempts, r.assistedRoutes, r.routeParams)
case hops => payFsm forward SendPaymentToRoute(r.paymentHash, hops, FinalLegacyPayload(r.amount, finalExpiry))
}
}

case r: SendTrampolinePaymentRequest =>
Expand Down
4 changes: 3 additions & 1 deletion eclair-core/src/main/scala/fr/acinq/eclair/payment/send/PaymentLifecycle.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,7 +186,9 @@ class PaymentLifecycle(nodeParams: NodeParams, cfg: SendPaymentConfig, router: A
stay

case Event(Status.Failure(t), WaitingForComplete(s, c, _, failures, _, ignoreNodes, ignoreChannels, hops)) =>
if (failures.size + 1 >= c.maxAttempts) {
// If the first hop was selected by the sender (in routePrefix) and it failed, it doesn't make sense to retry (we
// will end up retrying over that same faulty channel).
if (failures.size + 1 >= c.maxAttempts || c.routePrefix.nonEmpty) {
onFailure(s, PaymentFailed(id, c.paymentHash, failures :+ LocalFailure(t)))
stop(FSM.Normal)
} else {
Expand Down
Loading