Add initial support for async payment trampoline relay

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

@@ -149,6 +149,11 @@ eclair {
       // Delay enforcement of channel fee updates
       enforcement-delay = 10 minutes
     }
+
+    async-payments {
+        enabled = false
+        timeout = 1 day // maximum time to hold an async payment while waiting for the receiver to come online
+    }
   }
 
   on-chain-fees {

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

@@ -281,6 +281,12 @@ object Features {
     val mandatory = 148
   }
 
+  // TODO: @remyers update feature bits once spec-ed (currently reserved here: https://github.com/lightning/bolts/pull/989)
+  case object AsyncPaymentPrototype extends Feature with InvoiceFeature {
+    val rfcName = "async_payment_prototype"
+    val mandatory = 52
+  }
+
   val knownFeatures: Set[Feature] = Set(
     DataLossProtect,
     InitialRoutingSync,
@@ -303,7 +309,8 @@ object Features {
     PaymentMetadata,
     ZeroConf,
     KeySend,
-    TrampolinePaymentPrototype
+    TrampolinePaymentPrototype,
+    AsyncPaymentPrototype
   )
 
   // Features may depend on other features, as specified in Bolt 9.
@@ -315,7 +322,8 @@ object Features {
     AnchorOutputsZeroFeeHtlcTx -> (StaticRemoteKey :: Nil),
     RouteBlinding -> (VariableLengthOnion :: Nil),
     TrampolinePaymentPrototype -> (PaymentSecret :: Nil),
-    KeySend -> (VariableLengthOnion :: Nil)
+    KeySend -> (VariableLengthOnion :: Nil),
+    AsyncPaymentPrototype -> (VariableLengthOnion :: TrampolinePaymentPrototype :: Nil)
   )
 
   case class FeatureException(message: String) extends IllegalArgumentException(message)

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

@@ -83,7 +83,8 @@ case class NodeParams(nodeKeyManager: NodeKeyManager,
                       blockchainWatchdogThreshold: Int,
                       blockchainWatchdogSources: Seq[String],
                       onionMessageConfig: OnionMessageConfig,
-                      purgeInvoicesInterval: Option[FiniteDuration]) {
+                      purgeInvoicesInterval: Option[FiniteDuration],
+                      asyncPaymentsTimeout: Option[FiniteDuration]) {
   val privateKey: Crypto.PrivateKey = nodeKeyManager.nodeKey.privateKey
 
   val nodeId: PublicKey = nodeKeyManager.nodeId
@@ -419,6 +420,12 @@ object NodeParams extends Logging {
       None
     }
 
+    val asyncPaymentsTimeout = if (config.getBoolean("relay.async-payments.enabled")) {
+      Some(FiniteDuration(config.getDuration("relay.async-payments.timeout").toMinutes, TimeUnit.MINUTES))
+    } else {
+      None
+    }
+
     NodeParams(
       nodeKeyManager = nodeKeyManager,
       channelKeyManager = channelKeyManager,
@@ -529,7 +536,8 @@ object NodeParams extends Logging {
         relayPolicy = onionMessageRelayPolicy,
         timeout = FiniteDuration(config.getDuration("onion-messages.reply-timeout").getSeconds, TimeUnit.SECONDS),
       ),
-      purgeInvoicesInterval = purgeInvoicesInterval
+      purgeInvoicesInterval = purgeInvoicesInterval,
+      asyncPaymentsTimeout = asyncPaymentsTimeout
     )
   }
 }

eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/NodeRelay.scala

@@ -43,6 +43,7 @@ import fr.acinq.eclair.{CltvExpiry, Features, Logs, MilliSatoshi, NodeParams, UI
 
 import java.util.UUID
 import scala.collection.immutable.Queue
+import scala.concurrent.duration.FiniteDuration
 
 /**
  * It [[NodeRelay]] aggregates incoming HTLCs (in case multi-part was used upstream) and then forwards the requested amount (using the
@@ -54,12 +55,14 @@ object NodeRelay {
   sealed trait Command
   case class Relay(nodeRelayPacket: IncomingPaymentPacket.NodeRelayPacket) extends Command
   case object Stop extends Command
+  case object AsyncPaymentTrigger extends Command
   private case class WrappedMultiPartExtraPaymentReceived(mppExtraReceived: MultiPartPaymentFSM.ExtraPaymentReceived[HtlcPart]) extends Command
   private case class WrappedMultiPartPaymentFailed(mppFailed: MultiPartPaymentFSM.MultiPartPaymentFailed) extends Command
   private case class WrappedMultiPartPaymentSucceeded(mppSucceeded: MultiPartPaymentFSM.MultiPartPaymentSucceeded) extends Command
   private case class WrappedPreimageReceived(preimageReceived: PreimageReceived) extends Command
   private case class WrappedPaymentSent(paymentSent: PaymentSent) extends Command
   private case class WrappedPaymentFailed(paymentFailed: PaymentFailed) extends Command
+  private case object AsyncPaymentTimeout extends Command
   // @formatter:on
 
   trait OutgoingPaymentFactory {
@@ -178,14 +181,15 @@ class NodeRelay private(nodeParams: NodeParams,
    * @param nextPayload relay instructions (should be identical across HTLCs in this set).
    * @param nextPacket  trampoline onion to relay to the next trampoline node.
    * @param handler     actor handling the aggregation of the incoming HTLC set.
+   * @param triggered   async payment trigger received
    */
-  private def receiving(htlcs: Queue[UpdateAddHtlc], nextPayload: IntermediatePayload.NodeRelay.Standard, nextPacket: OnionRoutingPacket, handler: ActorRef): Behavior[Command] =
+  private def receiving(htlcs: Queue[UpdateAddHtlc], nextPayload: IntermediatePayload.NodeRelay.Standard, nextPacket: OnionRoutingPacket, handler: ActorRef, triggered: Boolean = false): Behavior[Command] =
     Behaviors.receiveMessagePartial {
       case Relay(IncomingPaymentPacket.NodeRelayPacket(add, outer, _, _)) =>
         require(outer.paymentSecret == paymentSecret, "payment secret mismatch")
         context.log.debug("forwarding incoming htlc #{} from channel {} to the payment FSM", add.id, add.channelId)
         handler ! MultiPartPaymentFSM.HtlcPart(outer.totalAmount, add)
-        receiving(htlcs :+ add, nextPayload, nextPacket, handler)
+        receiving(htlcs :+ add, nextPayload, nextPacket, handler, triggered)
       case WrappedMultiPartPaymentFailed(MultiPartPaymentFSM.MultiPartPaymentFailed(_, failure, parts)) =>
         context.log.warn("could not complete incoming multi-part payment (parts={} paidAmount={} failure={})", parts.size, parts.map(_.amount).sum, failure)
         Metrics.recordPaymentRelayFailed(failure.getClass.getSimpleName, Tags.RelayType.Trampoline)
@@ -199,10 +203,43 @@ class NodeRelay private(nodeParams: NodeParams,
             context.log.warn(s"rejecting trampoline payment reason=$failure")
             rejectPayment(upstream, Some(failure))
             stopping()
+          case None =>
+            if (!triggered && nextPayload.isAsyncPayment && nodeParams.asyncPaymentsTimeout.isDefined) {
+              waitForTrigger(upstream, nextPayload, nextPacket, nodeParams.asyncPaymentsTimeout.get)
+            } else {
+              doSend(upstream, nextPayload, nextPacket)
+            }
+        }
+      case AsyncPaymentTrigger => context.log.debug("received async payment trigger while waiting to receive all incoming payment packets")
+        receiving(htlcs, nextPayload, nextPacket, handler, triggered = true)
+    }
+
+  private def waitForTrigger(upstream: Upstream.Trampoline, nextPayload: IntermediatePayload.NodeRelay.Standard, nextPacket: OnionRoutingPacket, asyncPaymentsTimeout: FiniteDuration): Behavior[Command] = {
+    context.log.info(s"waiting for async payment to trigger before relaying trampoline payment (amountIn=${upstream.amountIn} expiryIn=${upstream.expiryIn} amountOut=${nextPayload.amountToForward} expiryOut=${nextPayload.outgoingCltv}, asyncPaymentsTimeout=$asyncPaymentsTimeout)")
+    Behaviors.withTimers { timers =>
+      timers.startSingleTimer(AsyncPaymentTimeout, asyncPaymentsTimeout)
+      Behaviors.receiveMessagePartial {
+        case AsyncPaymentTimeout =>
+          context.log.warn(s"rejecting async payment that was not triggered before timeout: $asyncPaymentsTimeout")
+          rejectPayment(upstream, Some(PaymentTimeout))
+          stopping()
+        case AsyncPaymentTrigger =>
+          // check cltv timeouts again before forwarding the payment
+          validateRelay(nodeParams, upstream, nextPayload) match {
+          case Some(failure) =>
+            context.log.warn(s"rejecting async payment, reason=$failure")
+            rejectPayment(upstream, Some(failure))
+            stopping()
           case None =>
             doSend(upstream, nextPayload, nextPacket)
         }
+        case Stop =>
+          context.log.warn(s"async payment stopped while waiting for trigger")
+          rejectPayment(upstream, Some(PaymentTimeout))
+          stopping()
+      }
     }
+  }
 
   private def doSend(upstream: Upstream.Trampoline, nextPayload: IntermediatePayload.NodeRelay.Standard, nextPacket: OnionRoutingPacket): Behavior[Command] = {
     context.log.info(s"relaying trampoline payment (amountIn=${upstream.amountIn} expiryIn=${upstream.expiryIn} amountOut=${nextPayload.amountToForward} expiryOut=${nextPayload.outgoingCltv})")

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/PaymentOnion.scala

@@ -22,7 +22,7 @@ import fr.acinq.eclair.payment.Bolt11Invoice
 import fr.acinq.eclair.wire.protocol.CommonCodecs._
 import fr.acinq.eclair.wire.protocol.OnionRoutingCodecs.{ForbiddenTlv, InvalidTlvPayload, MissingRequiredTlv}
 import fr.acinq.eclair.wire.protocol.TlvCodecs._
-import fr.acinq.eclair.{CltvExpiry, Features, MilliSatoshi, MilliSatoshiLong, ShortChannelId, UInt64}
+import fr.acinq.eclair.{CltvExpiry, Features, InvoiceFeature, MilliSatoshi, MilliSatoshiLong, ShortChannelId, UInt64}
 import scodec.bits.{BitVector, ByteVector}
 
 /**
@@ -181,6 +181,9 @@ object OnionPaymentPayloadTlv {
 
   /** Pre-image included by the sender of a payment in case of a donation */
   case class KeySend(paymentPreimage: ByteVector32) extends OnionPaymentPayloadTlv
+
+  /** Invoice feature bits. Only included for intermediate trampoline nodes that should wait before forwarding this payment */
+  case class AsyncPaymentFeatures(features: ByteVector) extends OnionPaymentPayloadTlv
 }
 
 object PaymentOnion {
@@ -301,6 +304,8 @@ object PaymentOnion {
         val paymentMetadata = records.get[PaymentMetadata].map(_.data)
         val invoiceFeatures = records.get[InvoiceFeatures].map(_.features)
         val invoiceRoutingInfo = records.get[InvoiceRoutingInfo].map(_.extraHops)
+        // The following fields are only included in the async payment case.
+        val isAsyncPayment: Boolean = records.get[AsyncPaymentFeatures].isDefined
       }
 
       object Standard {
@@ -331,6 +336,17 @@ object PaymentOnion {
           ).flatten
           Standard(TlvStream(tlvs))
         }
+
+        /** Create a standard trampoline inner payload instructing the trampoline node to wait for a trigger before sending an async payment. */
+        def createNodeRelayForAsyncPayment(amount: MilliSatoshi, expiry: CltvExpiry, nextNodeId: PublicKey, invoiceFeatures: Features[InvoiceFeature]): Standard = {
+          val tlvs = Seq(
+            Some(AmountToForward(amount)),
+            Some(OutgoingCltv(expiry)),
+            Some(OutgoingNodeId(nextNodeId)),
+            Some(AsyncPaymentFeatures(invoiceFeatures.toByteVector))
+          ).flatten
+          Standard(TlvStream(tlvs))
+        }
       }
     }
   }
@@ -475,6 +491,8 @@ object PaymentOnionCodecs {
 
   private val keySend: Codec[KeySend] = variableSizeBytesLong(varintoverflow, bytes32).as[KeySend]
 
+  private val asyncPayment: Codec[AsyncPaymentFeatures] = variableSizeBytesLong(varintoverflow, bytes).as[AsyncPaymentFeatures]
+
   private val onionTlvCodec = discriminated[OnionPaymentPayloadTlv].by(varint)
     .typecase(UInt64(2), amountToForward)
     .typecase(UInt64(4), outgoingCltv)
@@ -490,6 +508,7 @@ object PaymentOnionCodecs {
     .typecase(UInt64(66099), invoiceRoutingInfo)
     .typecase(UInt64(66100), trampolineOnion)
     .typecase(UInt64(5482373484L), keySend)
+    .typecase(UInt64(181324718L), asyncPayment)
 
   val perHopPayloadCodec: Codec[TlvStream[OnionPaymentPayloadTlv]] = TlvCodecs.lengthPrefixedTlvStream[OnionPaymentPayloadTlv](onionTlvCodec).complete
 

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

@@ -203,6 +203,7 @@ object TestConstants {
         timeout = 1 minute
       ),
       purgeInvoicesInterval = None,
+      asyncPaymentsTimeout = None
     )
 
     def channelParams: LocalParams = Peer.makeChannelParams(
@@ -344,7 +345,8 @@ object TestConstants {
         relayPolicy = RelayAll,
         timeout = 1 minute
       ),
-      purgeInvoicesInterval = None
+      purgeInvoicesInterval = None,
+      asyncPaymentsTimeout = None
     )
 
     def channelParams: LocalParams = Peer.makeChannelParams(