Find multi part route (#1427)

Leverage Yen's k-shortest paths and a simple split algorithm
to move MPP entirely inside the Router.

This is currently unused, the multipart payment lifecycle needs
to be updated to leverage this new algorithm.

eclair-core/src/main/resources/reference.conf

@@ -151,6 +151,11 @@ eclair {
       ratio-cltv = 0.15            // when computing the weight for a channel, consider its CLTV delta in this proportion
       ratio-channel-age = 0.35     // when computing the weight for a channel, consider its AGE in this proportion
       ratio-channel-capacity = 0.5 // when computing the weight for a channel, consider its CAPACITY in this proportion
+
+      mpp {
+        min-amount-satoshis = 15000 // minimum amount sent via partial HTLCs
+        max-parts = 6 // maximum number of HTLCs sent per payment: increasing this value will impact performance
+      }
     }
   }
 

eclair-core/src/main/scala/fr/acinq/eclair/NodeParams.scala

@@ -293,7 +293,9 @@ object NodeParams {
         searchHeuristicsEnabled = config.getBoolean("router.path-finding.heuristics-enable"),
         searchRatioCltv = config.getDouble("router.path-finding.ratio-cltv"),
         searchRatioChannelAge = config.getDouble("router.path-finding.ratio-channel-age"),
-        searchRatioChannelCapacity = config.getDouble("router.path-finding.ratio-channel-capacity")
+        searchRatioChannelCapacity = config.getDouble("router.path-finding.ratio-channel-capacity"),
+        mppMinPartAmount = Satoshi(config.getLong("router.path-finding.mpp.min-amount-satoshis")).toMilliSatoshi,
+        mppMaxParts = config.getInt("router.path-finding.mpp.max-parts")
       ),
       socksProxy_opt = socksProxy_opt,
       maxPaymentAttempts = config.getInt("max-payment-attempts"),

eclair-core/src/main/scala/fr/acinq/eclair/router/Graph.scala

@@ -375,7 +375,17 @@ object Graph {
      * @param capacity    channel capacity
      * @param balance_opt (optional) available balance that can be sent through this edge
      */
-    case class GraphEdge(desc: ChannelDesc, update: ChannelUpdate, capacity: Satoshi, balance_opt: Option[MilliSatoshi])
+    case class GraphEdge(desc: ChannelDesc, update: ChannelUpdate, capacity: Satoshi, balance_opt: Option[MilliSatoshi]) {
+
+      def maxHtlcAmount(reservedCapacity: MilliSatoshi): MilliSatoshi = Seq(
+        balance_opt.map(balance => balance - reservedCapacity),
+        update.htlcMaximumMsat,
+        Some(capacity.toMilliSatoshi - reservedCapacity)
+      ).flatten.min.max(0 msat)
+
+      def fee(amount: MilliSatoshi): MilliSatoshi = nodeFee(update.feeBaseMsat, update.feeProportionalMillionths, amount)
+
+    }
 
     /** A graph data structure that uses an adjacency list, stores the incoming edges of the neighbors */
     case class DirectedGraph(private val vertices: Map[PublicKey, List[GraphEdge]]) {

eclair-core/src/main/scala/fr/acinq/eclair/router/RouteCalculation.scala

@@ -30,6 +30,7 @@ import fr.acinq.eclair.wire.ChannelUpdate
 import fr.acinq.eclair.{ShortChannelId, _}
 
 import scala.annotation.tailrec
+import scala.collection.mutable
 import scala.concurrent.duration._
 import scala.util.{Failure, Random, Success, Try}
 
@@ -144,22 +145,24 @@ object RouteCalculation {
         ageFactor = routerConf.searchRatioChannelAge,
         capacityFactor = routerConf.searchRatioChannelCapacity
       ))
-    }
+    },
+    mpp = MultiPartParams(routerConf.mppMinPartAmount, routerConf.mppMaxParts)
   )
 
   /**
    * Find a route in the graph between localNodeId and targetNodeId, returns the route.
    * Will perform a k-shortest path selection given the @param numRoutes and randomly select one of the result.
    *
-   * @param g            graph of the whole network
-   * @param localNodeId  sender node (payer)
-   * @param targetNodeId target node (final recipient)
-   * @param amount       the amount that the target node should receive
-   * @param maxFee       the maximum fee of a resulting route
-   * @param numRoutes    the number of routes to find
-   * @param extraEdges   a set of extra edges we want to CONSIDER during the search
-   * @param ignoredEdges a set of extra edges we want to IGNORE during the search
-   * @param routeParams  a set of parameters that can restrict the route search
+   * @param g               graph of the whole network
+   * @param localNodeId     sender node (payer)
+   * @param targetNodeId    target node (final recipient)
+   * @param amount          the amount that the target node should receive
+   * @param maxFee          the maximum fee of a resulting route
+   * @param numRoutes       the number of routes to find
+   * @param extraEdges      a set of extra edges we want to CONSIDER during the search
+   * @param ignoredEdges    a set of extra edges we want to IGNORE during the search
+   * @param ignoredVertices a set of extra vertices we want to IGNORE during the search
+   * @param routeParams     a set of parameters that can restrict the route search
    * @return the computed routes to the destination @param targetNodeId
    */
   def findRoute(g: DirectedGraph,
@@ -219,4 +222,138 @@ object RouteCalculation {
     }
   }
 
+  /**
+   * Find a multi-part route in the graph between localNodeId and targetNodeId.
+   *
+   * @param g               graph of the whole network
+   * @param localNodeId     sender node (payer)
+   * @param targetNodeId    target node (final recipient)
+   * @param amount          the amount that the target node should receive
+   * @param maxFee          the maximum fee of a resulting route
+   * @param extraEdges      a set of extra edges we want to CONSIDER during the search
+   * @param ignoredEdges    a set of extra edges we want to IGNORE during the search
+   * @param ignoredVertices a set of extra vertices we want to IGNORE during the search
+   * @param pendingHtlcs    a list of htlcs that have already been sent for that multi-part payment (used to avoid finding conflicting HTLCs)
+   * @param routeParams     a set of parameters that can restrict the route search
+   * @return a set of disjoint routes to the destination @param targetNodeId with the payment amount split between them
+   */
+  def findMultiPartRoute(g: DirectedGraph,
+                         localNodeId: PublicKey,
+                         targetNodeId: PublicKey,
+                         amount: MilliSatoshi,
+                         maxFee: MilliSatoshi,
+                         extraEdges: Set[GraphEdge] = Set.empty,
+                         ignoredEdges: Set[ChannelDesc] = Set.empty,
+                         ignoredVertices: Set[PublicKey] = Set.empty,
+                         pendingHtlcs: Seq[Route] = Nil,
+                         routeParams: RouteParams,
+                         currentBlockHeight: Long): Try[Seq[Route]] = Try {
+    val result = findMultiPartRouteInternal(g, localNodeId, targetNodeId, amount, maxFee, extraEdges, ignoredEdges, ignoredVertices, pendingHtlcs, routeParams, currentBlockHeight) match {
+      case Right(routes) => Right(routes)
+      case Left(RouteNotFound) if routeParams.randomize =>
+        // If we couldn't find a randomized solution, fallback to a deterministic one.
+        findMultiPartRouteInternal(g, localNodeId, targetNodeId, amount, maxFee, extraEdges, ignoredEdges, ignoredVertices, pendingHtlcs, routeParams.copy(randomize = false), currentBlockHeight)
+      case Left(ex) => Left(ex)
+    }
+    result match {
+      case Right(routes) => routes
+      case Left(ex) => return Failure(ex)
+    }
+  }
+
+  private def findMultiPartRouteInternal(g: DirectedGraph,
+                                         localNodeId: PublicKey,
+                                         targetNodeId: PublicKey,
+                                         amount: MilliSatoshi,
+                                         maxFee: MilliSatoshi,
+                                         extraEdges: Set[GraphEdge] = Set.empty,
+                                         ignoredEdges: Set[ChannelDesc] = Set.empty,
+                                         ignoredVertices: Set[PublicKey] = Set.empty,
+                                         pendingHtlcs: Seq[Route] = Nil,
+                                         routeParams: RouteParams,
+                                         currentBlockHeight: Long): Either[RouterException, Seq[Route]] = {
+    // We use Yen's k-shortest paths to find many paths for chunks of the total amount.
+    val numRoutes = {
+      val directChannelsCount = g.getEdgesBetween(localNodeId, targetNodeId).length
+      routeParams.mpp.maxParts.max(directChannelsCount) // if we have direct channels to the target, we can use them all
+    }
+    val routeAmount = routeParams.mpp.minPartAmount.min(amount)
+    findRouteInternal(g, localNodeId, targetNodeId, routeAmount, maxFee, numRoutes, extraEdges, ignoredEdges, ignoredVertices, routeParams, currentBlockHeight) match {
+      case Right(routes) =>
+        // We use these shortest paths to find a set of non-conflicting HTLCs that send the total amount.
+        split(amount, mutable.Queue(routes: _*), initializeUsedCapacity(pendingHtlcs), routeParams) match {
+          case Right(routes) if validateMultiPartRoute(amount, maxFee, routes) => Right(routes)
+          case _ => Left(RouteNotFound)
+        }
+      case Left(ex) => Left(ex)
+    }
+  }
+
+  @tailrec
+  private def split(amount: MilliSatoshi, paths: mutable.Queue[Graph.WeightedPath], usedCapacity: mutable.Map[ShortChannelId, MilliSatoshi], routeParams: RouteParams, selectedRoutes: Seq[Route] = Nil): Either[RouterException, Seq[Route]] = {
+    if (amount == 0.msat) {
+      Right(selectedRoutes)
+    } else if (paths.isEmpty) {
+      Left(RouteNotFound)
+    } else {
+      val current = paths.dequeue()
+      val candidate = computeRouteMaxAmount(current.path, usedCapacity)
+      if (candidate.amount < routeParams.mpp.minPartAmount.min(amount)) {
+        // this route doesn't have enough capacity left: we remove it and continue.
+        split(amount, paths, usedCapacity, routeParams, selectedRoutes)
+      } else {
+        val route = if (routeParams.randomize) {
+          // randomly choose the amount to be between 20% and 100% of the available capacity.
+          val randomizedAmount = candidate.amount * ((20d + Random.nextInt(81)) / 100)
+          if (randomizedAmount < routeParams.mpp.minPartAmount) {
+            candidate.copy(amount = routeParams.mpp.minPartAmount.min(amount))
+          } else {
+            candidate.copy(amount = randomizedAmount.min(amount))
+          }
+        } else {
+          candidate.copy(amount = candidate.amount.min(amount))
+        }
+        updateUsedCapacity(route, usedCapacity)
+        // NB: we re-enqueue the current path, it may still have capacity for a second HTLC.
+        split(amount - route.amount, paths.enqueue(current), usedCapacity, routeParams, route +: selectedRoutes)
+      }
+    }
+  }
+
+  /** Compute the maximum amount that we can send through the given route. */
+  private def computeRouteMaxAmount(route: Seq[GraphEdge], usedCapacity: mutable.Map[ShortChannelId, MilliSatoshi]): Route = {
+    val firstHopMaxAmount = route.head.maxHtlcAmount(usedCapacity.getOrElse(route.head.update.shortChannelId, 0 msat))
+    val amount = route.drop(1).foldLeft(firstHopMaxAmount) { case (amount, edge) =>
+      // We compute fees going forward instead of backwards. That means we will slightly overestimate the fees of some
+      // edges, but we will always stay inside the capacity bounds we computed.
+      val amountMinusFees = amount - edge.fee(amount)
+      val edgeMaxAmount = edge.maxHtlcAmount(usedCapacity.getOrElse(edge.update.shortChannelId, 0 msat))
+      amountMinusFees.min(edgeMaxAmount)
+    }
+    Route(amount.max(0 msat), route.map(graphEdgeToHop))
+  }
+
+  /** Initialize known used capacity based on pending HTLCs. */
+  private def initializeUsedCapacity(pendingHtlcs: Seq[Route]): mutable.Map[ShortChannelId, MilliSatoshi] = {
+    val usedCapacity = mutable.Map.empty[ShortChannelId, MilliSatoshi]
+    // We always skip the first hop: since they are local channels, we already take into account those sent HTLCs in the
+    // channel balance (which overrides the channel capacity in route calculation).
+    pendingHtlcs.filter(_.hops.length > 1).foreach(route => updateUsedCapacity(route.copy(hops = route.hops.tail), usedCapacity))
+    usedCapacity
+  }
+
+  /** Update used capacity by taking into account an HTLC sent to the given route. */
+  private def updateUsedCapacity(route: Route, usedCapacity: mutable.Map[ShortChannelId, MilliSatoshi]): Unit = {
+    route.hops.reverse.foldLeft(route.amount) { case (amount, hop) =>
+      usedCapacity.updateWith(hop.lastUpdate.shortChannelId)(previous => Some(amount + previous.getOrElse(0 msat)))
+      amount + hop.fee(amount)
+    }
+  }
+
+  private def validateMultiPartRoute(amount: MilliSatoshi, maxFee: MilliSatoshi, routes: Seq[Route]): Boolean = {
+    val amountOk = routes.map(_.amount).sum == amount
+    val feeOk = routes.map(_.fee).sum <= maxFee
+    amountOk && feeOk
+  }
+
 }

eclair-core/src/main/scala/fr/acinq/eclair/router/Router.scala

@@ -274,7 +274,9 @@ object Router {
                         searchHeuristicsEnabled: Boolean,
                         searchRatioCltv: Double,
                         searchRatioChannelAge: Double,
-                        searchRatioChannelCapacity: Double)
+                        searchRatioChannelCapacity: Double,
+                        mppMinPartAmount: MilliSatoshi,
+                        mppMaxParts: Int)
 
   // @formatter:off
   case class ChannelDesc(shortChannelId: ShortChannelId, a: PublicKey, b: PublicKey)
@@ -363,7 +365,9 @@ object Router {
     override def fee(amount: MilliSatoshi): MilliSatoshi = fee
   }
 
-  case class RouteParams(randomize: Boolean, maxFeeBase: MilliSatoshi, maxFeePct: Double, routeMaxLength: Int, routeMaxCltv: CltvExpiryDelta, ratios: Option[WeightRatios]) {
+  case class MultiPartParams(minPartAmount: MilliSatoshi, maxParts: Int)
+
+  case class RouteParams(randomize: Boolean, maxFeeBase: MilliSatoshi, maxFeePct: Double, routeMaxLength: Int, routeMaxCltv: CltvExpiryDelta, ratios: Option[WeightRatios], mpp: MultiPartParams) {
     def getMaxFee(amount: MilliSatoshi): MilliSatoshi = {
       // The payment fee must satisfy either the flat fee or the percentage fee, not necessarily both.
       maxFeeBase.max(amount * maxFeePct)
@@ -384,6 +388,10 @@ object Router {
   case class Route(amount: MilliSatoshi, hops: Seq[ChannelHop], allowEmpty: Boolean = false) {
     require(allowEmpty || hops.nonEmpty, "route cannot be empty")
     val length = hops.length
+    lazy val fee: MilliSatoshi = {
+      val amountToSend = hops.drop(1).reverse.foldLeft(amount) { case (amount1, hop) => amount1 + hop.fee(amount1) }
+      amountToSend - amount
+    }
 
     /** This method retrieves the channel update that we used when we built the route. */
     def getChannelUpdateForNode(nodeId: PublicKey): Option[ChannelUpdate] = hops.find(_.nodeId == nodeId).map(_.lastUpdate)

eclair-core/src/test/scala/fr/acinq/eclair/TestConstants.scala

@@ -133,7 +133,9 @@ object TestConstants {
         searchHeuristicsEnabled = false,
         searchRatioCltv = 0.0,
         searchRatioChannelAge = 0.0,
-        searchRatioChannelCapacity = 0.0
+        searchRatioChannelCapacity = 0.0,
+        mppMinPartAmount = 15000000 msat,
+        mppMaxParts = 10
       ),
       socksProxy_opt = None,
       maxPaymentAttempts = 5,
@@ -217,7 +219,9 @@ object TestConstants {
         searchHeuristicsEnabled = false,
         searchRatioCltv = 0.0,
         searchRatioChannelAge = 0.0,
-        searchRatioChannelCapacity = 0.0
+        searchRatioChannelCapacity = 0.0,
+        mppMinPartAmount = 15000000 msat,
+        mppMaxParts = 10
       ),
       socksProxy_opt = None,
       maxPaymentAttempts = 5,

eclair-core/src/test/scala/fr/acinq/eclair/integration/IntegrationSpec.scala

@@ -37,8 +37,8 @@ import fr.acinq.eclair.channel._
 import fr.acinq.eclair.crypto.Sphinx.DecryptedFailurePacket
 import fr.acinq.eclair.crypto.TransportHandler
 import fr.acinq.eclair.db._
-import fr.acinq.eclair.io.{Peer, PeerConnection}
 import fr.acinq.eclair.io.Peer.{Disconnect, PeerRoutingMessage}
+import fr.acinq.eclair.io.{Peer, PeerConnection}
 import fr.acinq.eclair.payment.PaymentRequest.ExtraHop
 import fr.acinq.eclair.payment._
 import fr.acinq.eclair.payment.receive.MultiPartHandler.ReceivePayment
@@ -49,7 +49,7 @@ import fr.acinq.eclair.payment.send.PaymentInitiator.{SendPaymentRequest, SendTr
 import fr.acinq.eclair.payment.send.PaymentLifecycle.{State => _}
 import fr.acinq.eclair.router.Graph.WeightRatios
 import fr.acinq.eclair.router.RouteCalculation.ROUTE_MAX_LENGTH
-import fr.acinq.eclair.router.Router.{GossipDecision, PublicChannel, RouteParams, NORMAL => _, State => _}
+import fr.acinq.eclair.router.Router.{GossipDecision, MultiPartParams, PublicChannel, RouteParams, NORMAL => _, State => _}
 import fr.acinq.eclair.router.{Announcements, AnnouncementsBatchValidationSpec, Router}
 import fr.acinq.eclair.transactions.Transactions
 import fr.acinq.eclair.transactions.Transactions.{HtlcSuccessTx, HtlcTimeoutTx}
@@ -86,7 +86,8 @@ class IntegrationSpec extends TestKitBaseClass with BitcoindService with AnyFunS
       cltvDeltaFactor = 0.1,
       ageFactor = 0,
       capacityFactor = 0
-    ))
+    )),
+    mpp = MultiPartParams(15000000 msat, 6)
   ))
 
   val commonConfig = ConfigFactory.parseMap(Map(

eclair-core/src/test/scala/fr/acinq/eclair/payment/MultiPartPaymentLifecycleSpec.scala

@@ -283,7 +283,7 @@ class MultiPartPaymentLifecycleSpec extends TestKitBaseClass with FixtureAnyFunS
 
   test("split fees between child payments") { f =>
     import f._
-    val routeParams = RouteParams(randomize = false, 100 msat, 0.05, 20, CltvExpiryDelta(144), None)
+    val routeParams = RouteParams(randomize = false, 100 msat, 0.05, 20, CltvExpiryDelta(144), None, MultiPartParams(10000 msat, 5))
     val payment = SendMultiPartPayment(randomBytes32, e, 3000 * 1000 msat, expiry, 3, routeParams = Some(routeParams))
     initPayment(f, payment, emptyStats.copy(capacity = Stats.generate(Seq(1000), d => Satoshi(d.toLong))), localChannels())
     waitUntilAmountSent(f, 3000 * 1000 msat)
@@ -494,7 +494,7 @@ class MultiPartPaymentLifecycleSpec extends TestKitBaseClass with FixtureAnyFunS
       // We have a total of 6500 satoshis across all channels. We try to send lower amounts to take fees into account.
       val toSend = ((1 + Random.nextInt(3500)) * 1000).msat
       val networkStats = emptyStats.copy(capacity = Stats.generate(Seq(400 + Random.nextInt(1600)), d => Satoshi(d.toLong)))
-      val routeParams = RouteParams(randomize = true, Random.nextInt(1000).msat, Random.nextInt(10).toDouble / 100, 20, CltvExpiryDelta(144), None)
+      val routeParams = RouteParams(randomize = true, Random.nextInt(1000).msat, Random.nextInt(10).toDouble / 100, 20, CltvExpiryDelta(144), None, MultiPartParams(10000 msat, 5))
       val request = SendMultiPartPayment(randomBytes32, e, toSend, CltvExpiry(561), 1, Nil, Some(routeParams))
       val fuzzParams = s"(sending $toSend with network capacity ${networkStats.capacity.percentile75.toMilliSatoshi}, fee base ${routeParams.maxFeeBase} and fee percentage ${routeParams.maxFeePct})"
       val (remaining, payments) = splitPayment(f.nodeParams, toSend, testChannels.channels, Some(networkStats), request, randomize = true)

eclair-core/src/test/scala/fr/acinq/eclair/payment/PaymentInitiatorSpec.scala

@@ -32,7 +32,7 @@ import fr.acinq.eclair.payment.send.MultiPartPaymentLifecycle.SendMultiPartPayme
 import fr.acinq.eclair.payment.send.PaymentInitiator._
 import fr.acinq.eclair.payment.send.PaymentLifecycle.{SendPayment, SendPaymentToRoute}
 import fr.acinq.eclair.payment.send.{PaymentError, PaymentInitiator}
-import fr.acinq.eclair.router.Router.{NodeHop, RouteParams}
+import fr.acinq.eclair.router.Router.{MultiPartParams, NodeHop, RouteParams}
 import fr.acinq.eclair.wire.Onion.{FinalLegacyPayload, FinalTlvPayload}
 import fr.acinq.eclair.wire.OnionTlv.{AmountToForward, OutgoingCltv}
 import fr.acinq.eclair.wire.{Onion, OnionCodecs, OnionTlv, TrampolineFeeInsufficient, _}
@@ -122,7 +122,7 @@ class PaymentInitiatorSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike
   test("forward legacy payment") { f =>
     import f._
     val hints = Seq(Seq(ExtraHop(b, channelUpdate_bc.shortChannelId, feeBase = 10 msat, feeProportionalMillionths = 1, cltvExpiryDelta = CltvExpiryDelta(12))))
-    val routeParams = RouteParams(randomize = true, 15 msat, 1.5, 5, CltvExpiryDelta(561), None)
+    val routeParams = RouteParams(randomize = true, 15 msat, 1.5, 5, CltvExpiryDelta(561), None, MultiPartParams(10000 msat, 5))
     sender.send(initiator, SendPaymentRequest(finalAmount, paymentHash, c, 1, CltvExpiryDelta(42), assistedRoutes = hints, routeParams = Some(routeParams)))
     val id1 = sender.expectMsgType[UUID]
     payFsm.expectMsg(SendPaymentConfig(id1, id1, None, paymentHash, finalAmount, c, Upstream.Local(id1), None, storeInDb = true, publishEvent = true, Nil))