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,13 @@ eclair {
       // Delay enforcement of channel fee updates
       enforcement-delay = 10 minutes
     }
+
+    async-payments {
+      // Maximum number of blocks to hold an async payment while waiting to receive a trigger from the receiver
+      hold-timeout-blocks = 1008
+      // Number of blocks before the incoming HTLC expires that an async payment must be triggered by the receiver
+      cancel-safety-before-timeout-blocks = 144
+    }
   }
 
   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 InitFeature 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

@@ -30,7 +30,7 @@ import fr.acinq.eclair.db._
 import fr.acinq.eclair.io.MessageRelay.{NoRelay, RelayAll, RelayChannelsOnly, RelayPolicy}
 import fr.acinq.eclair.io.PeerConnection
 import fr.acinq.eclair.message.OnionMessages.OnionMessageConfig
-import fr.acinq.eclair.payment.relay.Relayer.{RelayFees, RelayParams}
+import fr.acinq.eclair.payment.relay.Relayer.{AsyncPaymentsParams, RelayFees, RelayParams}
 import fr.acinq.eclair.router.Announcements.AddressException
 import fr.acinq.eclair.router.Graph.{HeuristicsConstants, WeightRatios}
 import fr.acinq.eclair.router.PathFindingExperimentConf
@@ -419,6 +419,12 @@ object NodeParams extends Logging {
       None
     }
 
+    val asyncPaymentCancelSafetyBeforeTimeoutBlocks = CltvExpiryDelta(config.getInt("relay.async-payments.cancel-safety-before-timeout-blocks"))
+    require(asyncPaymentCancelSafetyBeforeTimeoutBlocks >= expiryDelta, "relay.async-payments.cancel-safety-before-timeout-blocks must not be less than channel.expiry-delta-blocks; this may lead to undesired channel closure")
+
+    val asyncPaymentHoldTimeoutBlocks = config.getInt("relay.async-payments.hold-timeout-blocks")
+    require(asyncPaymentHoldTimeoutBlocks >= (asyncPaymentCancelSafetyBeforeTimeoutBlocks + expiryDelta).toInt, "relay.async-payments.hold-timeout-blocks must not be less than relay.async-payments.cancel-safety-before-timeout-blocks + channel.expiry-delta-blocks; otherwise it will have no effect")
+
     NodeParams(
       nodeKeyManager = nodeKeyManager,
       channelKeyManager = channelKeyManager,
@@ -488,7 +494,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),
+        asyncPaymentsParams = AsyncPaymentsParams(asyncPaymentHoldTimeoutBlocks, asyncPaymentCancelSafetyBeforeTimeoutBlocks)
       ),
       db = database,
       autoReconnect = config.getBoolean("auto-reconnect"),

eclair-core/src/main/scala/fr/acinq/eclair/payment/PaymentEvents.scala

@@ -120,6 +120,8 @@ case class PaymentMetadataReceived(paymentHash: ByteVector32, paymentMetadata: B
 
 case class PaymentSettlingOnChain(id: UUID, amount: MilliSatoshi, paymentHash: ByteVector32, timestamp: TimestampMilli = TimestampMilli.now()) extends PaymentEvent
 
+case class WaitingToRelayPayment(remoteNodeId: PublicKey, paymentHash: ByteVector32, timestamp: TimestampMilli = TimestampMilli.now()) extends PaymentEvent
+
 sealed trait PaymentFailure {
   // @formatter:off
   def amount: MilliSatoshi

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

@@ -23,6 +23,7 @@ import akka.actor.typed.scaladsl.adapter.{TypedActorContextOps, TypedActorRefOps
 import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
 import com.softwaremill.quicklens.ModifyPimp
 import fr.acinq.bitcoin.scalacompat.ByteVector32
+import fr.acinq.eclair.blockchain.CurrentBlockHeight
 import fr.acinq.eclair.channel.{CMD_FAIL_HTLC, CMD_FULFILL_HTLC}
 import fr.acinq.eclair.db.PendingCommandsDb
 import fr.acinq.eclair.payment.IncomingPaymentPacket.NodeRelayPacket
@@ -39,7 +40,7 @@ import fr.acinq.eclair.router.Router.RouteParams
 import fr.acinq.eclair.router.{BalanceTooLow, RouteNotFound}
 import fr.acinq.eclair.wire.protocol.PaymentOnion.{FinalPayload, IntermediatePayload}
 import fr.acinq.eclair.wire.protocol._
-import fr.acinq.eclair.{CltvExpiry, Features, Logs, MilliSatoshi, NodeParams, UInt64, nodeFee, randomBytes32}
+import fr.acinq.eclair.{BlockHeight, CltvExpiry, Features, Logs, MilliSatoshi, NodeParams, UInt64, nodeFee, randomBytes32}
 
 import java.util.UUID
 import scala.collection.immutable.Queue
@@ -54,12 +55,15 @@ object NodeRelay {
   sealed trait Command
   case class Relay(nodeRelayPacket: IncomingPaymentPacket.NodeRelayPacket) extends Command
   case object Stop extends Command
+  case object RelayAsyncPayment extends Command
+  case object CancelAsyncPayment 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 class WrappedCurrentBlockHeight(currentBlockHeight: BlockHeight) extends Command
   // @formatter:on
 
   trait OutgoingPaymentFactory {
@@ -200,10 +204,45 @@ class NodeRelay private(nodeParams: NodeParams,
             rejectPayment(upstream, Some(failure))
             stopping()
           case None =>
-            doSend(upstream, nextPayload, nextPacket)
+            if (nextPayload.isAsyncPayment && nodeParams.features.hasFeature(Features.AsyncPaymentPrototype)) {
+              waitForTrigger(upstream, nextPayload, nextPacket)
+            } else {
+              doSend(upstream, nextPayload, nextPacket)
+            }
         }
     }
 
+  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}, asyncPaymentsParams=${nodeParams.relayParams.asyncPaymentsParams})")
+    // a trigger must be received before waiting more than `holdTimeoutBlocks`
+    val timeoutBlock: BlockHeight = nodeParams.currentBlockHeight + nodeParams.relayParams.asyncPaymentsParams.holdTimeoutBlocks
+    // a trigger must be received `cancelSafetyBeforeTimeoutBlocks` before the incoming payment cltv expiry
+    val safetyBlock: BlockHeight = (upstream.expiryIn - nodeParams.relayParams.asyncPaymentsParams.cancelSafetyBeforeTimeout).blockHeight
+    val messageAdapter = context.messageAdapter[CurrentBlockHeight](cbc => WrappedCurrentBlockHeight(cbc.blockHeight))
+    context.system.eventStream ! EventStream.Subscribe[CurrentBlockHeight](messageAdapter)
+
+    // TODO: send the WaitingToRelayPayment message to an actor that watches for the payment receiver to come back online before sending the RelayAsyncPayment message
+    context.system.eventStream ! EventStream.Publish(WaitingToRelayPayment(nextPayload.outgoingNodeId, paymentHash))
+    Behaviors.receiveMessagePartial {
+      case WrappedCurrentBlockHeight(blockHeight) if blockHeight >= safetyBlock =>
+        context.log.warn(s"rejecting async payment at block $blockHeight; was not triggered ${nodeParams.relayParams.asyncPaymentsParams.cancelSafetyBeforeTimeout} safety blocks before upstream cltv expiry at ${upstream.expiryIn}")
+        rejectPayment(upstream, Some(TemporaryNodeFailure)) // TODO: replace failure type when async payment spec is finalized
+        stopping()
+      case WrappedCurrentBlockHeight(blockHeight) if blockHeight >= timeoutBlock =>
+        context.log.warn(s"rejecting async payment at block $blockHeight; was not triggered after waiting ${nodeParams.relayParams.asyncPaymentsParams.holdTimeoutBlocks} blocks")
+        rejectPayment(upstream, Some(TemporaryNodeFailure)) // TODO: replace failure type when async payment spec is finalized
+        stopping()
+      case WrappedCurrentBlockHeight(blockHeight) =>
+        Behaviors.same
+      case CancelAsyncPayment =>
+        context.log.warn(s"payment sender canceled a waiting async payment")
+        rejectPayment(upstream, Some(TemporaryNodeFailure)) // TODO: replace failure type when async payment spec is finalized
+        stopping()
+      case RelayAsyncPayment =>
+        doSend(upstream, nextPayload, nextPacket)
+    }
+  }
+
   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})")
     relay(upstream, nextPayload, nextPacket)

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

@@ -29,7 +29,7 @@ import fr.acinq.eclair.channel._
 import fr.acinq.eclair.db.PendingCommandsDb
 import fr.acinq.eclair.payment._
 import fr.acinq.eclair.wire.protocol._
-import fr.acinq.eclair.{Logs, MilliSatoshi, NodeParams}
+import fr.acinq.eclair.{CltvExpiryDelta, Logs, MilliSatoshi, NodeParams}
 import grizzled.slf4j.Logging
 
 import scala.concurrent.Promise
@@ -126,10 +126,13 @@ object Relayer extends Logging {
     require(feeProportionalMillionths >= 0.0, "feeProportionalMillionths must be nonnegative")
   }
 
+  case class AsyncPaymentsParams(holdTimeoutBlocks: Int, cancelSafetyBeforeTimeout: CltvExpiryDelta)
+
   case class RelayParams(publicChannelFees: RelayFees,
                          privateChannelFees: RelayFees,
                          minTrampolineFees: RelayFees,
-                         enforcementDelay: FiniteDuration) {
+                         enforcementDelay: FiniteDuration,
+                         asyncPaymentsParams: AsyncPaymentsParams) {
     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] = variableSizeBytesLong(varintoverflow, provide(AsyncPayment())).as[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

@@ -26,7 +26,7 @@ import fr.acinq.eclair.crypto.keymanager.{LocalChannelKeyManager, LocalNodeKeyMa
 import fr.acinq.eclair.io.MessageRelay.RelayAll
 import fr.acinq.eclair.io.{Peer, PeerConnection}
 import fr.acinq.eclair.message.OnionMessages.OnionMessageConfig
-import fr.acinq.eclair.payment.relay.Relayer.{RelayFees, RelayParams}
+import fr.acinq.eclair.payment.relay.Relayer.{AsyncPaymentsParams, RelayFees, RelayParams}
 import fr.acinq.eclair.router.Graph.WeightRatios
 import fr.acinq.eclair.router.PathFindingExperimentConf
 import fr.acinq.eclair.router.Router.{MultiPartParams, PathFindingConf, RouterConf, SearchBoundaries}
@@ -142,7 +142,8 @@ object TestConstants {
         minTrampolineFees = RelayFees(
           feeBase = 548000 msat,
           feeProportionalMillionths = 30),
-        enforcementDelay = 10 minutes),
+        enforcementDelay = 10 minutes,
+       asyncPaymentsParams = AsyncPaymentsParams(1008, CltvExpiryDelta(144))),
       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,
+        asyncPaymentsParams = AsyncPaymentsParams(1008, CltvExpiryDelta(144))),
       db = TestDatabases.inMemoryDb(),
       autoReconnect = false,
       initialRandomReconnectDelay = 5 seconds,