Add initial support for async payment trampoline relay

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

@@ -70,6 +70,7 @@ eclair {
     option_zeroconf = disabled
     keysend = disabled
     trampoline_payment_prototype = disabled
+    async_payment_prototype = disabled
   }
   // The following section lets you customize features for specific nodes.
   // The overrides will be applied on top of the default features settings.
@@ -149,6 +150,9 @@ eclair {
       // Delay enforcement of channel fee updates
       enforcement-delay = 10 minutes
     }
+
+    // Maximum time to hold an async payment while waiting for the receiver to come online
+    timeout = 1 day
   }
 
   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 = 152
+  }
+
   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 -> (TrampolinePaymentPrototype :: Nil)
   )
 
   case class FeatureException(message: String) extends IllegalArgumentException(message)

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

@@ -488,7 +488,8 @@ object NodeParams extends Logging {
         publicChannelFees = getRelayFees(config.getConfig("relay.fees.public-channels")),
         privateChannelFees = getRelayFees(config.getConfig("relay.fees.private-channels")),
         minTrampolineFees = getRelayFees(config.getConfig("relay.fees.min-trampoline")),
-        enforcementDelay = FiniteDuration(config.getDuration("relay.fees.enforcement-delay").getSeconds, TimeUnit.SECONDS)
+        enforcementDelay = FiniteDuration(config.getDuration("relay.fees.enforcement-delay").getSeconds, TimeUnit.SECONDS),
+        timeout = FiniteDuration(config.getDuration("relay.timeout").toMinutes, TimeUnit.MINUTES)
       ),
       db = database,
       autoReconnect = config.getBoolean("auto-reconnect"),

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

@@ -54,12 +54,14 @@ object NodeRelay {
   sealed trait Command
   case class Relay(nodeRelayPacket: IncomingPaymentPacket.NodeRelayPacket) extends Command
   case object Stop extends Command
+  case object RelayAsyncPayment 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 {
@@ -199,10 +201,43 @@ class NodeRelay private(nodeParams: NodeParams,
             context.log.warn(s"rejecting trampoline payment reason=$failure")
             rejectPayment(upstream, Some(failure))
             stopping()
+          case None =>
+            if (nextPayload.isAsyncPayment && nodeParams.features.hasFeature(Features.AsyncPaymentPrototype)) {
+              waitForTrigger(upstream, nextPayload, nextPacket)
+            } else {
+              doSend(upstream, nextPayload, nextPacket)
+            }
+        }
+      case RelayAsyncPayment => context.log.debug("received async payment trigger while waiting to receive all incoming payment packets")
+        receiving(htlcs, nextPayload, nextPacket, handler)
+    }
+
+  private def waitForTrigger(upstream: Upstream.Trampoline, nextPayload: IntermediatePayload.NodeRelay.Standard, nextPacket: OnionRoutingPacket): 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}, timeout=${nodeParams.relayParams.timeout})")
+    Behaviors.withTimers { timers =>
+      timers.startSingleTimer(AsyncPaymentTimeout, nodeParams.relayParams.timeout)
+      Behaviors.receiveMessagePartial {
+        case AsyncPaymentTimeout =>
+          context.log.warn(s"rejecting async payment that was not triggered before relay timeout: ${nodeParams.relayParams.timeout}")
+          rejectPayment(upstream, Some(PaymentTimeout))
+          stopping()
+        case RelayAsyncPayment =>
+          // 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/payment/relay/Relayer.scala

@@ -129,7 +129,8 @@ object Relayer extends Logging {
   case class RelayParams(publicChannelFees: RelayFees,
                          privateChannelFees: RelayFees,
                          minTrampolineFees: RelayFees,
-                         enforcementDelay: FiniteDuration) {
+                         enforcementDelay: FiniteDuration,
+                         timeout: FiniteDuration) {
     def defaultFees(announceChannel: Boolean): RelayFees = {
       if (announceChannel) {
         publicChannelFees

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

@@ -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
+
+  /** Only included for intermediate trampoline nodes that should wait before forwarding this payment */
+  case class AsyncPayment() 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[AsyncPayment].isDefined
       }
 
       object Standard {
@@ -331,6 +336,11 @@ 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): Standard = {
+          Standard(TlvStream(AmountToForward(amount), OutgoingCltv(expiry), OutgoingNodeId(nextNodeId), AsyncPayment()))
+        }
       }
     }
   }
@@ -475,6 +485,8 @@ object PaymentOnionCodecs {
 
   private val keySend: Codec[KeySend] = variableSizeBytesLong(varintoverflow, bytes32).as[KeySend]
 
+  private val asyncPayment: Codec[AsyncPayment] = provide(AsyncPayment())
+
   private val onionTlvCodec = discriminated[OnionPaymentPayloadTlv].by(varint)
     .typecase(UInt64(2), amountToForward)
     .typecase(UInt64(4), outgoingCltv)
@@ -489,6 +501,7 @@ object PaymentOnionCodecs {
     .typecase(UInt64(66098), outgoingNodeId)
     .typecase(UInt64(66099), invoiceRoutingInfo)
     .typecase(UInt64(66100), trampolineOnion)
+    .typecase(UInt64(181324718L), asyncPayment)
     .typecase(UInt64(5482373484L), keySend)
 
   val perHopPayloadCodec: Codec[TlvStream[OnionPaymentPayloadTlv]] = TlvCodecs.lengthPrefixedTlvStream[OnionPaymentPayloadTlv](onionTlvCodec).complete

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

@@ -142,7 +142,8 @@ object TestConstants {
         minTrampolineFees = RelayFees(
           feeBase = 548000 msat,
           feeProportionalMillionths = 30),
-        enforcementDelay = 10 minutes),
+        enforcementDelay = 10 minutes,
+        timeout = 1 day),
       db = TestDatabases.inMemoryDb(),
       autoReconnect = false,
       initialRandomReconnectDelay = 5 seconds,
@@ -202,7 +203,7 @@ object TestConstants {
         relayPolicy = RelayAll,
         timeout = 1 minute
       ),
-      purgeInvoicesInterval = None,
+      purgeInvoicesInterval = None
     )
 
     def channelParams: LocalParams = Peer.makeChannelParams(
@@ -284,7 +285,8 @@ object TestConstants {
         minTrampolineFees = RelayFees(
           feeBase = 548000 msat,
           feeProportionalMillionths = 30),
-        enforcementDelay = 10 minutes),
+        enforcementDelay = 10 minutes,
+        timeout = 1 day),
       db = TestDatabases.inMemoryDb(),
       autoReconnect = false,
       initialRandomReconnectDelay = 5 seconds,