Fix several on-the-fly-funding bugs (#3351)

### eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala
@@ -380,7 +380,7 @@ class Peer(val nodeParams: NodeParams,
                 // The payer is buggy and is paying the same payment_hash multiple times. We could simply claim that
                 // extra payment for ourselves, but we're nice and instead immediately fail it.
                 val proposal = OnTheFlyFunding.Proposal(htlc, cmd.upstream, cmd.onionSharedSecrets)
-                proposal.createFailureCommands(None)(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+                proposal.createFailureCommands(None).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
                 pending
             }
           case None =>
@@ -403,7 +403,7 @@ class Peer(val nodeParams: NodeParams,
                       case msg: WillFailHtlc => FailureReason.EncryptedDownstreamFailure(msg.reason, msg.attribution_opt)
                       case msg: WillFailMalformedHtlc => FailureReason.LocalFailure(createBadOnionFailure(msg.onionHash, msg.failureCode))
                     }
-                    htlc.createFailureCommands(Some(failure))(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+                    htlc.createFailureCommands(Some(failure)).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
                     val proposed1 = pending.proposed.filterNot(_.htlc.id == msg.id)
                     if (proposed1.isEmpty) {
                       Metrics.OnTheFlyFunding.withTag(Tags.OnTheFlyFundingState, Tags.OnTheFlyFundingStates.Rejected).increment()
@@ -435,7 +435,7 @@ class Peer(val nodeParams: NodeParams,
             pending.status match {
               case _: OnTheFlyFunding.Status.Proposed =>
                 log.warning("on-the-fly funding proposal timed out for payment_hash={}", timeout.paymentHash)
-                pending.createFailureCommands(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+                pending.createFailureCommands().foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
                 Metrics.OnTheFlyFunding.withTag(Tags.OnTheFlyFundingState, Tags.OnTheFlyFundingStates.Expired).increment()
                 pendingOnTheFlyFunding -= timeout.paymentHash
                 self ! Peer.OutgoingMessage(Warning(s"on-the-fly funding proposal timed out for payment_hash=${timeout.paymentHash}"), d.peerConnection)
@@ -736,20 +736,23 @@ class Peer(val nodeParams: NodeParams,
           case _: OnTheFlyFunding.Status.Proposed =>
             log.warning("proposed will_add_htlc expired for payment_hash={}", paymentHash)
             Metrics.OnTheFlyFunding.withTag(Tags.OnTheFlyFundingState, Tags.OnTheFlyFundingStates.Timeout).increment()
-            pending.createFailureCommands(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+            pending.createFailureCommands().foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
           case _: OnTheFlyFunding.Status.AddedToFeeCredit =>
             // Nothing to do, we already fulfilled the upstream HTLCs.
             log.debug("forgetting will_add_htlc added to fee credit for payment_hash={}", paymentHash)
-          case _: OnTheFlyFunding.Status.Funded =>
+          case status: OnTheFlyFunding.Status.Funded =>
             log.warning("funded will_add_htlc expired for payment_hash={}, our peer may be malicious", paymentHash)
             Metrics.OnTheFlyFunding.withTag(Tags.OnTheFlyFundingState, Tags.OnTheFlyFundingStates.Timeout).increment()
-            pending.createFailureCommands(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+            val channelCloser = context.spawnAnonymous(Behaviors.supervise(OnTheFlyFunding.ChannelCloserHtlcTimeout(nodeParams, remoteNodeId, pending, status)).onFailure(typed.SupervisorStrategy.stop))
+            channelCloser ! OnTheFlyFunding.ChannelCloserHtlcTimeout.CloseChannel(register)
             nodeParams.db.liquidity.removePendingOnTheFlyFunding(remoteNodeId, paymentHash)
         }
       }
       pendingOnTheFlyFunding = pendingOnTheFlyFunding.removedAll(expired.keys)
       d match {
-        case d: DisconnectedData if d.channels.isEmpty && pendingOnTheFlyFunding.isEmpty => stopPeer(d.peerStorage)
+        // Note that if we didn't check expired.isEmpty, we may stop the peer while a child ChannelCloserHtlcTimeout is
+        // working. We wait for the next block before stopping ourselves to make sure it has finished its work.
+        case d: DisconnectedData if d.channels.isEmpty && pendingOnTheFlyFunding.isEmpty && expired.isEmpty => stopPeer(d.peerStorage)
         case _ => stay()
       }
 
@@ -818,7 +821,7 @@ class Peer(val nodeParams: NodeParams,
         case (paymentHash, pending) => pending.status match {
           case status: OnTheFlyFunding.Status.Funded if status.channelId == e.channelId && status.txId == e.txId && status.fundingTxIndex == e.fundingTxIndex =>
             log.warning("funded will_add_htlc aborted by our peer after funding for payment_hash={} and fundingTxId={}", paymentHash, status.txId)
-            pending.createFailureCommands(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+            pending.createFailureCommands().foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
             nodeParams.db.liquidity.removePendingOnTheFlyFunding(remoteNodeId, paymentHash)
             pendingOnTheFlyFunding -= paymentHash
           case _ => ()
@@ -855,7 +858,19 @@ class Peer(val nodeParams: NodeParams,
               Metrics.OnTheFlyFundingFees.withoutTags().record(success.fees.toLong)
               nodeParams.db.liquidity.removePendingOnTheFlyFunding(remoteNodeId, success.paymentHash)
               pendingOnTheFlyFunding -= success.paymentHash
-            case None => ()
+            case None =>
+              // We have already forgotten this payment, which happens when the HTLCs reached their expiry: we then
+              // force-close the funded channel and stop tracking the proposal. But our peer may reveal the preimage
+              // anyway, either off-chain if their update_fulfill_htlc raced our expiry, or on-chain by publishing an
+              // HTLC-success transaction after our force-close. We must settle the upstream HTLCs in that case,
+              // otherwise we've paid our peer without being paid ourselves.
+              // Note that we don't emit a relay event or update our metrics here: for multi-part payments, we receive
+              // one result per downstream HTLC and only the first one finds a matching proposal, so doing that would
+              // double-count relays. Sending a settlement command for an HTLC that we've already settled is harmless.
+              log.warning("received preimage for on-the-fly payment_hash={} that we stopped tracking: fulfilling upstream HTLCs", success.paymentHash)
+              success.proposed.flatMap(_.createFulfillCommands(success.preimage)).foreach {
+                case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd)
+              }
           }
           // If this is a payment that was initially rejected, it wasn't a malicious node, but rather a temporary issue.
           nodeParams.onTheFlyFundingConfig.fromFutureHtlcFulfilled(success.paymentHash)
@@ -993,7 +1008,7 @@ class Peer(val nodeParams: NodeParams,
           case status: OnTheFlyFunding.Status.Proposed =>
             log.info("cancelling on-the-fly funding for payment_hash={}", paymentHash)
             status.timer.cancel()
-            pending.createFailureCommands(log).foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
+            pending.createFailureCommands().foreach { case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd) }
             true
           // We keep proposals that have been added to fee credit until we reach the HTLC expiry or we restart. This
           // guarantees that our peer cannot concurrently add to their fee credit a payment for which we've signed a

### eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/OnTheFlyFunding.scala
@@ -20,17 +20,17 @@ import akka.actor.Cancellable
 import akka.actor.typed.scaladsl.adapter.TypedActorRefOps
 import akka.actor.typed.scaladsl.{ActorContext, Behaviors, StashBuffer}
 import akka.actor.typed.{ActorRef, Behavior}
-import akka.event.LoggingAdapter
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
 import fr.acinq.bitcoin.scalacompat.{ByteVector32, Crypto, TxId}
 import fr.acinq.eclair.blockchain.fee.FeeratePerKw
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.crypto.Sphinx
+import fr.acinq.eclair.db.PendingCommandsDb
 import fr.acinq.eclair.payment.Monitoring.Metrics
 import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.wire.protocol.LiquidityAds.PaymentDetails
 import fr.acinq.eclair.wire.protocol._
-import fr.acinq.eclair.{Logs, MilliSatoshi, MilliSatoshiLong, NodeParams, TimestampMilli, ToMilliSatoshiConversion}
+import fr.acinq.eclair.{CltvExpiry, Logs, MilliSatoshi, MilliSatoshiLong, NodeParams, TimestampMilli, ToMilliSatoshiConversion}
 
 import scala.concurrent.duration.FiniteDuration
 
@@ -99,7 +99,7 @@ object OnTheFlyFunding {
     def maxFees(htlcMinimum: MilliSatoshi): MilliSatoshi = (htlc.amount - htlcMinimum).max(0 msat)
 
     /** Create commands to fail all upstream HTLCs. */
-    def createFailureCommands(failure_opt: Option[FailureReason])(implicit log: LoggingAdapter): Seq[(ByteVector32, CMD_FAIL_HTLC)] = upstream match {
+    def createFailureCommands(failure_opt: Option[FailureReason]): Seq[(ByteVector32, CMD_FAIL_HTLC)] = upstream match {
       case _: Upstream.Local => Nil
       case u: Upstream.Hot.Channel =>
         // Note that even in the Bolt12 case, we relay the downstream failure instead of sending back invalid_onion_blinding.
@@ -116,11 +116,10 @@ object OnTheFlyFunding {
               // the BOLTs to better handle those cases.
               Sphinx.FailurePacket.decrypt(f.packet, f.attribution_opt, onionSharedSecrets).failure match {
                 case Left(Sphinx.CannotDecryptFailurePacket(_, _)) =>
-                  log.warning("couldn't decrypt downstream on-the-fly funding failure")
+                  FailureReason.LocalFailure(TemporaryNodeFailure())
                 case Right(f) =>
-                  log.warning("downstream on-the-fly funding failure: {}", f.failureMessage.message)
+                  FailureReason.LocalFailure(TemporaryNodeFailure())
               }
-              FailureReason.LocalFailure(TemporaryNodeFailure())
             case _: FailureReason.LocalFailure => f
           }
           case None => FailureReason.LocalFailure(UnknownNextPeer())
@@ -144,15 +143,15 @@ object OnTheFlyFunding {
 
   /** A set of funding proposals for a given payment. */
   case class Pending(proposed: Seq[Proposal], status: Status) {
-    val paymentHash = proposed.head.htlc.paymentHash
-    val expiry = proposed.map(_.htlc.expiry).min
-    val amountOut = proposed.map(_.htlc.amount).sum
+    val paymentHash: ByteVector32 = proposed.head.htlc.paymentHash
+    val expiry: CltvExpiry = proposed.map(_.htlc.expiry).min
+    val amountOut: MilliSatoshi = proposed.map(_.htlc.amount).sum
 
     /** Maximum fees that can be collected from this HTLC set. */
     def maxFees(htlcMinimum: MilliSatoshi): MilliSatoshi = proposed.map(_.maxFees(htlcMinimum)).sum
 
     /** Create commands to fail all upstream HTLCs. */
-    def createFailureCommands(implicit log: LoggingAdapter): Seq[(ByteVector32, CMD_FAIL_HTLC)] = proposed.flatMap(_.createFailureCommands(None))
+    def createFailureCommands(): Seq[(ByteVector32, CMD_FAIL_HTLC)] = proposed.flatMap(_.createFailureCommands(None))
 
     /** Create commands to fulfill all upstream HTLCs. */
     def createFulfillCommands(preimage: ByteVector32): Seq[(ByteVector32, CMD_FULFILL_HTLC)] = proposed.flatMap(_.createFulfillCommands(preimage))
@@ -215,11 +214,15 @@ object OnTheFlyFunding {
       (requestFunding.fees(feerate, isChannelCreation).total.toMilliSatoshi, None)
     }
     val cancelAmountTooLow = CancelOnTheFlyFunding(channelId, paymentHashes, s"requested amount is too low to relay HTLCs: ${requestFunding.requestedAmount} < $totalPaymentAmount")
+    val cancelMultipleHash = CancelOnTheFlyFunding(channelId, paymentHashes, s"request batches ${paymentHashes.size} distinct payments: we only support funding a single payment_hash")
+    val cancelDuplicateHash = CancelOnTheFlyFunding(channelId, paymentHashes, s"request contains the same payment_hash multiple times: ${paymentHashes.mkString(",")}")
     val cancelFeesTooLow = CancelOnTheFlyFunding(channelId, paymentHashes, s"htlc amount is too low to pay liquidity fees: $availableAmountForFees < $feesOwed")
     val cancelDisabled = CancelOnTheFlyFunding(channelId, paymentHashes, "payments paid with future HTLCs are currently disabled")
     requestFunding.paymentDetails match {
       case PaymentDetails.FromChannelBalance => ValidationResult.Accept(Set.empty, None)
       case _ if requestFunding.requestedAmount.toMilliSatoshi < totalPaymentAmount => ValidationResult.Reject(cancelAmountTooLow, paymentHashes.toSet)
+      case _ if paymentHashes.toSet.size != paymentHashes.size => ValidationResult.Reject(cancelDuplicateHash, paymentHashes.toSet)
+      case _ if paymentHashes.size > 1 => ValidationResult.Reject(cancelMultipleHash, paymentHashes.toSet)
       case _: PaymentDetails.FromChannelBalanceForFutureHtlc => ValidationResult.Accept(Set.empty, useFeeCredit_opt)
       case _: PaymentDetails.FromFutureHtlc if !cfg.isFromFutureHtlcAllowed(remoteNodeId) => ValidationResult.Reject(cancelDisabled, paymentHashes.toSet)
       case _: PaymentDetails.FromFutureHtlc if availableAmountForFees < feesOwed => ValidationResult.Reject(cancelFeesTooLow, paymentHashes.toSet)
@@ -229,6 +232,20 @@ object OnTheFlyFunding {
     }
   }
 
+  private def paymentAlreadyRelayed(paymentHash: ByteVector32, commitments: Commitments): Boolean = {
+    val htlcsInCommitTxs = commitments.all.flatMap(_.localCommit.spec.htlcs.map(_.add)).toSet ++
+      commitments.all.flatMap(_.remoteCommit.spec.htlcs.map(_.add)).toSet ++
+      commitments.all.flatMap(_.nextRemoteCommit_opt).flatMap(_.spec.htlcs.map(_.add)).toSet
+    val areHtlcsBeingRelayed = commitments.changes.localChanges.all.exists {
+      case add: UpdateAddHtlc => add.paymentHash == paymentHash && add.fundingFee_opt.nonEmpty
+      case _ => false
+    }
+    val areHtlcsAlreadyRelayed = htlcsInCommitTxs.exists {
+      htlc => htlc.paymentHash == paymentHash && htlc.fundingFee_opt.nonEmpty
+    }
+    areHtlcsBeingRelayed || areHtlcsAlreadyRelayed
+  }
+
   /**
    * This actor relays HTLCs that were proposed with [[WillAddHtlc]] once funding is complete.
    * It verifies that this payment was not previously relayed, to protect against over-paying and paying multiple times.
@@ -242,7 +259,7 @@ object OnTheFlyFunding {
     private case class WrappedHtlcSettled(result: RES_ADD_SETTLED[Origin.Hot, HtlcResult]) extends Command
 
     sealed trait RelayResult
-    case class RelaySuccess(channelId: ByteVector32, paymentHash: ByteVector32, preimage: ByteVector32, fees: MilliSatoshi) extends RelayResult
+    case class RelaySuccess(channelId: ByteVector32, paymentHash: ByteVector32, preimage: ByteVector32, fees: MilliSatoshi, proposed: Seq[Proposal]) extends RelayResult
     case class RelayFailed(paymentHash: ByteVector32, failure: RelayFailure) extends RelayResult
 
     sealed trait RelayFailure
@@ -279,7 +296,7 @@ object OnTheFlyFunding {
     private def checkChannelState(): Behavior[Command] = {
       cmd.channel ! CMD_GET_CHANNEL_INFO(context.messageAdapter[RES_GET_CHANNEL_INFO](r => WrappedChannelInfo(r.state, r.data)))
       Behaviors.receiveMessagePartial {
-        case WrappedChannelInfo(_, data: DATA_NORMAL) if paymentAlreadyRelayed(paymentHash, data) =>
+        case WrappedChannelInfo(_, data: DATA_NORMAL) if paymentAlreadyRelayed(paymentHash, data.commitments) =>
           context.log.warn("payment is already being relayed, waiting for it to be settled")
           Behaviors.stopped
         case WrappedChannelInfo(_, data: DATA_NORMAL) =>
@@ -288,7 +305,7 @@ object OnTheFlyFunding {
               // We have already received the preimage for that payment, but we probably restarted before removing the
               // on-the-fly funding proposal from our DB. We must not relay the payment again, otherwise we will pay
               // the next node twice.
-              cmd.replyTo ! RelaySuccess(channelId, paymentHash, preimage, cmd.status.remainingFees)
+              cmd.replyTo ! RelaySuccess(channelId, paymentHash, preimage, cmd.status.remainingFees, cmd.proposed)
               Behaviors.stopped
             case None => relay(data)
           }
@@ -298,13 +315,6 @@ object OnTheFlyFunding {
       }
     }
 
-    private def paymentAlreadyRelayed(paymentHash: ByteVector32, data: DATA_NORMAL): Boolean = {
-      data.commitments.changes.localChanges.all.exists {
-        case add: UpdateAddHtlc => add.paymentHash == paymentHash && add.fundingFee_opt.nonEmpty
-        case _ => false
-      }
-    }
-
     private def relay(data: DATA_NORMAL): Behavior[Command] = {
       context.log.debug("relaying {} on-the-fly HTLCs that have been funded", cmd.proposed.size)
       val htlcMinimum = data.commitments.latest.remoteCommitParams.htlcMinimum
@@ -351,7 +361,7 @@ object OnTheFlyFunding {
         Behaviors.receiveMessagePartial {
           case WrappedHtlcSettled(settled) =>
             settled.result match {
-              case fulfill: HtlcResult.Fulfill => cmd.replyTo ! RelaySuccess(channelId, paymentHash, fulfill.paymentPreimage, cmd.status.remainingFees)
+              case fulfill: HtlcResult.Fulfill => cmd.replyTo ! RelaySuccess(channelId, paymentHash, fulfill.paymentPreimage, cmd.status.remainingFees, cmd.proposed)
               case fail: HtlcResult.Fail => cmd.replyTo ! RelayFailed(paymentHash, RemoteFailure(fail))
             }
             waitForSettlement(remaining - 1)
@@ -361,6 +371,84 @@ object OnTheFlyFunding {
 
   }
 
+  /**
+   * This actor closes an on-the-fly funded channel for which our peer didn't correctly pay the fees.
+   * This happens when fees must be paid with HTLCs relayed after the channel funding, which the peer ignores until
+   * they reach their expiry. We must close that channel and fail the upstream HTLCs if they hadn't been relayed yet.
+   */
+  object ChannelCloserHtlcTimeout {
+    // @formatter:off
+    sealed trait Command
+    case class CloseChannel(register: akka.actor.ActorRef) extends Command
+    private case class WrappedChannelInfo(state: ChannelState, data: ChannelData) extends Command
+    private case object ChannelNotFound extends Command
+    // @formatter:on
+
+    def apply(nodeParams: NodeParams, remoteNodeId: PublicKey, expired: OnTheFlyFunding.Pending, status: OnTheFlyFunding.Status.Funded): Behavior[Command] =
+      Behaviors.setup { context =>
+        Behaviors.withMdc(Logs.mdc(category_opt = Some(Logs.LogCategory.PAYMENT), remoteNodeId_opt = Some(remoteNodeId), channelId_opt = Some(status.channelId), paymentHash_opt = Some(expired.paymentHash))) {
+          Behaviors.receiveMessagePartial {
+            case cmd: CloseChannel => new ChannelCloserHtlcTimeout(nodeParams, expired, status, cmd.register, context).start()
+          }
+        }
+      }
+  }
+
+  class ChannelCloserHtlcTimeout private(nodeParams: NodeParams,
+                                         expired: OnTheFlyFunding.Pending,
+                                         status: OnTheFlyFunding.Status.Funded,
+                                         register: akka.actor.ActorRef,
+                                         context: ActorContext[ChannelCloserHtlcTimeout.Command]) {
+
+    import ChannelCloserHtlcTimeout._
+
+    def start(): Behavior[Command] = {
+      context.log.warn("force-closing channel with expired funded will_add_htlc")
+      // We start by closing the channel, which ensures that HTLCs cannot be relayed anymore.
+      register ! Register.Forward(
+        context.messageAdapter[Register.ForwardFailure[CMD_FORCECLOSE]](_ => ChannelNotFound),
+        status.channelId,
+        CMD_FORCECLOSE(akka.actor.ActorRef.noSender)
+      )
+      // Then we check if HTLCs had already been relayed to decide whether to fail upstream immediately or not.
+      register ! Register.Forward(
+        context.messageAdapter[Register.ForwardFailure[CMD_GET_CHANNEL_INFO]](_ => ChannelNotFound),
+        status.channelId,
+        CMD_GET_CHANNEL_INFO(context.messageAdapter[RES_GET_CHANNEL_INFO](r => WrappedChannelInfo(r.state, r.data)))
+      )
+      Behaviors.receiveMessagePartial {
+        case WrappedChannelInfo(_, data: ChannelDataWithCommitments) =>
+          if (!paymentAlreadyRelayed(expired.paymentHash, data.commitments)) {
+            context.log.info("payment hasn't been relayed downstream: failing the corresponding upstream payments")
+            // The payment hasn't been relayed to the downstream channel, it won't be relayed anymore since we've closed
+            // that channel. We thus need to fail the corresponding upstream payments to ensure that they don't close
+            // when the payments will timeout on their end. If the payment has been relayed to the downstream channel,
+            // the upstream payments will automatically be failed when the HTLC-timeout transaction confirms (or, if
+            // our peer publishes their HTLC-sucess, we will fulfill the upstream payments).
+            expired.createFailureCommands().foreach {
+              case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd)
+            }
+          }
+          Behaviors.stopped
+        case WrappedChannelInfo(state, _) =>
+          context.log.warn("expired channel in state={}: failing the corresponding upstream payments", state)
+          // If the channel doesn't have commitments, the HTLCs cannot have been relayed. It is thus safe to fail them.
+          expired.createFailureCommands().foreach {
+            case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd)
+          }
+          Behaviors.stopped
+        case ChannelNotFound =>
+          context.log.warn("cannot find channel actor (probably already closed): failing the corresponding upstream payments")
+          // If we cannot find the channel, the HTLCs cannot be claimed by our peer anymore, so it's safe to fail the
+          // corresponding upstream HTLCs.
+          expired.createFailureCommands().foreach {
+            case (channelId, cmd) => PendingCommandsDb.safeSend(register, nodeParams.db.pendingCommands, channelId, cmd)
+          }
+          Behaviors.stopped
+      }
+    }
+  }
+
   object Codecs {
 
     import fr.acinq.eclair.wire.protocol.CommonCodecs._

### eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/PostRestartHtlcCleaner.scala
@@ -160,7 +160,9 @@ class PostRestartHtlcCleaner(nodeParams: NodeParams, register: ActorRef, initial
       handleDownstreamFulfill(brokenHtlcs, o, remoteNodeId, htlc, fulfill.paymentPreimage)
 
     case RES_ADD_SETTLED(o: Origin.Cold, remoteNodeId, htlc, fail: HtlcResult.Fail) =>
-      if (htlc.fundingFee_opt.nonEmpty) {
+      // Note that an on-chain failure guarantees that our peer cannot claim that HTLC anymore. At that point, we must
+      // fail upstream, otherwise the upstream channels will force-close when their own HTLCs reach their expiry.
+      if (htlc.fundingFee_opt.nonEmpty && !fail.isInstanceOf[HtlcResult.OnChainFail]) {
         log.info("htlc #{} from channelId={} failed downstream by {} but has a pending on-the-fly funding", htlc.id, htlc.channelId, remoteNodeId)
         // We don't fail upstream: we haven't been paid our funding fee yet, so we will try relaying again.
       } else {

### eclair-core/src/main/scala/fr/acinq/eclair/transactions/Scripts.scala
@@ -247,8 +247,9 @@ object Scripts {
 
   /** Extract the payment preimage from from a fulfilled offered htlc. */
   def extractPreimageFromClaimHtlcSuccess: PartialFunction[ScriptWitness, ByteVector32] = {
-    case ScriptWitness(Seq(_, paymentPreimage, _)) if paymentPreimage.size == 32 => ByteVector32(paymentPreimage)
-    case ScriptWitness(Seq(_, paymentPreimage, _, _)) if paymentPreimage.size == 32 => ByteVector32(paymentPreimage)
+    case ScriptWitness(Seq(_, paymentPreimage, _)) if paymentPreimage.size == 32 => ByteVector32(paymentPreimage) // segwit v0
+    case ScriptWitness(Seq(_, paymentPreimage, _, _)) if paymentPreimage.size == 32 => ByteVector32(paymentPreimage) // taproot
+    case ScriptWitness(Seq(_, paymentPreimage, _, _, _)) if paymentPreimage.size == 32 => ByteVector32(paymentPreimage) // taproot with annex
   }
 
   /** Extract payment preimages from a (potentially batched) claim HTLC transaction's witnesses. */

### eclair-core/src/test/scala/fr/acinq/eclair/channel/states/h/ClosingStateSpec.scala
@@ -435,6 +435,42 @@ class ClosingStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike with
     extractPreimageFromClaimHtlcSuccess(f)
   }
 
+  test("recv WatchOutputSpentTriggered (extract preimage from Claim-HTLC-success tx with annex, taproot)", Tag(ChannelStateTestsTags.OptionSimpleTaproot)) { f =>
+    import f._
+
+    // Alice sends an htlc to Bob.
+    val (preimage, htlc) = addHtlc(50_000_000 msat, alice, bob, alice2bob, bob2alice)
+    crossSign(alice, bob, alice2bob, bob2alice)
+    // Bob has the preimage, but Alice force-closes before receiving it.
+    bob ! CMD_FULFILL_HTLC(htlc.id, preimage, None)
+    bob2alice.expectMsgType[UpdateFulfillHtlc] // ignored
+    val (lcp, _) = localClose(alice, alice2blockchain, htlcTimeoutCount = 1)
+
+    // Bob claims the htlc output from Alice's commit tx using its preimage.
+    bob ! WatchFundingSpentTriggered(lcp.commitTx)
+    bob2blockchain.expectReplaceableTxPublished[ClaimRemoteAnchorTx]
+    bob2blockchain.expectFinalTxPublished("remote-main")
+    val claimHtlcSuccessTx = bob2blockchain.expectReplaceableTxPublished[ClaimHtlcSuccessTx].sign()
+    assert(claimHtlcSuccessTx.txIn.map(_.outPoint).toSet == lcp.htlcOutputs)
+    // Bob includes an annex in its witness (the signature will be incorrect but we don't check it here).
+    val annexWitness = claimHtlcSuccessTx.txIn.head.witness.copy(stack = claimHtlcSuccessTx.txIn.head.witness.stack :+ ByteVector.fromValidHex("50deadbeef"))
+    val claimHtlcSuccessTxWithAnnex = claimHtlcSuccessTx.updateWitness(0, annexWitness)
+
+    // Alice extracts the preimage and forwards it upstream.
+    alice ! WatchOutputSpentTriggered(htlc.amountMsat.truncateToSatoshi, claimHtlcSuccessTxWithAnnex)
+    inside(alice2relayer.expectMsgType[RES_ADD_SETTLED[Origin, HtlcResult.OnChainFulfill]]) { fulfill =>
+      assert(fulfill.htlc == htlc)
+      assert(fulfill.result.paymentPreimage == preimage)
+      assert(fulfill.origin == alice.stateData.asInstanceOf[DATA_CLOSING].commitments.originChannels(htlc.id))
+    }
+
+    // The Claim-HTLC-success transaction confirms: nothing to do, preimage has already been relayed.
+    alice2blockchain.expectWatchTxConfirmed(claimHtlcSuccessTx.txid)
+    alice ! WatchTxConfirmedTriggered(alice.nodeParams.currentBlockHeight, 6, claimHtlcSuccessTx)
+    alice2blockchain.expectNoMessage(100 millis)
+    alice2relayer.expectNoMessage(100 millis)
+  }
+
   private def extractPreimageFromHtlcSuccess(f: FixtureParam): Unit = {
     import f._
 

### eclair-core/src/test/scala/fr/acinq/eclair/payment/PostRestartHtlcCleanerSpec.scala
@@ -688,9 +688,7 @@ class PostRestartHtlcCleanerSpec extends TestKitBaseClass with FixtureAnyFunSuit
     // HTLC failures are not relayed upstream, as we will retry until we reach the HTLC timeout.
     sender.send(relayer, buildForwardFail(htlc_bc(0).add, Upstream.Cold.Channel(htlc_ab(0).add, a)))
     sender.send(relayer, buildForwardFail(htlc_bc(0).add, upstreamChannel))
-    sender.send(relayer, buildForwardOnChainFail(htlc_bc(0).add, upstreamChannel))
     sender.send(relayer, buildForwardFail(htlc_bc(1).add, upstreamTrampoline))
-    sender.send(relayer, buildForwardOnChainFail(htlc_bc(1).add, upstreamTrampoline))
     register.expectNoMessage(100 millis)
 
     // HTLC fulfills are relayed upstream as soon as available.
@@ -707,6 +705,39 @@ class PostRestartHtlcCleanerSpec extends TestKitBaseClass with FixtureAnyFunSuit
     register.expectNoMessage(100 millis)
   }
 
+  test("relay on-chain htlc-fail for on-the-fly funding") { f =>
+    import f._
+
+    // Upstream HTLCs that were relayed after completing on-the-fly funding.
+    val htlc_ab = Seq(
+      buildHtlcIn(1, channelId_ab_1, paymentHash1), // channel relayed
+      buildHtlcIn(2, channelId_ab_1, paymentHash2), // trampoline relayed
+    )
+    val htlc_bc = Seq(
+      buildHtlcOut(1, channelId_bc_1, paymentHash1).modify(_.add.tlvStream).setTo(TlvStream(UpdateAddHtlcTlv.FundingFeeTlv(LiquidityAds.FundingFee(2500 msat, TxId(randomBytes32()))))),
+      buildHtlcOut(2, channelId_bc_1, paymentHash2).modify(_.add.tlvStream).setTo(TlvStream(UpdateAddHtlcTlv.FundingFeeTlv(LiquidityAds.FundingFee(1500 msat, TxId(randomBytes32()))))),
+    )
+
+    val upstreamChannel = Upstream.Cold.Channel(htlc_ab(0).add, a)
+    val upstreamTrampoline = Upstream.Cold.Trampoline(Upstream.Cold.Channel(htlc_ab(1).add, a) :: Nil)
+    val data_ab = ChannelCodecsSpec.makeChannelDataNormal(htlc_ab, Map.empty)
+    val data_bc = ChannelCodecsSpec.makeChannelDataNormal(htlc_bc, Map(1L -> Origin.Cold(upstreamChannel), 2L -> Origin.Cold(upstreamTrampoline)))
+
+    val (relayer, _) = f.createRelayer(nodeParams)
+    relayer ! PostRestartHtlcCleaner.Init(Seq(data_ab, data_bc).map(_.withChannelKeys(nodeParams)))
+
+    // An on-chain failure guarantees that our peer cannot claim that HTLC anymore, so we fail the upstream HTLCs.
+    sender.send(relayer, buildForwardOnChainFail(htlc_bc(0).add, upstreamChannel))
+    val fail1 = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
+    assert(fail1.channelId == channelId_ab_1)
+    assert(fail1.message.id == 1)
+    sender.send(relayer, buildForwardOnChainFail(htlc_bc(1).add, upstreamTrampoline))
+    val fail2 = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
+    assert(fail2.channelId == channelId_ab_1)
+    assert(fail2.message.id == 2)
+    register.expectNoMessage(100 millis)
+  }
+
   test("relayed standard->non-standard HTLC is retained") { f =>
     import f._
 

### eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/OnTheFlyFundingSpec.scala
@@ -31,6 +31,7 @@ import fr.acinq.eclair.crypto.keymanager.ChannelKeys
 import fr.acinq.eclair.io.Peer._
 import fr.acinq.eclair.io.PendingChannelsRateLimiter.AddOrRejectChannel
 import fr.acinq.eclair.io.{Peer, PeerConnection, PendingChannelsRateLimiter}
+import fr.acinq.eclair.transactions.{IncomingHtlc, OutgoingHtlc}
 import fr.acinq.eclair.wire.protocol
 import fr.acinq.eclair.wire.protocol._
 import fr.acinq.eclair.{Alias, BlockHeight, CltvExpiry, CltvExpiryDelta, FeatureSupport, Features, MilliSatoshi, MilliSatoshiLong, NodeParams, TestConstants, TestKitBaseClass, TimestampMilli, ToMilliSatoshiConversion, UInt64, randomBytes, randomBytes32, randomKey, randomLong}
@@ -164,10 +165,15 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
       assert(failed.map(_.message.id).toSet == incomingHtlcs.map(_.id).toSet)
     }
 
-    def makeChannelData(htlcMinimum: MilliSatoshi = 1 msat, localChanges: LocalChanges = LocalChanges(Nil, Nil, Nil)): DATA_NORMAL = {
+    def makeChannelData(htlcMinimum: MilliSatoshi = 1 msat, localChanges: LocalChanges = LocalChanges(Nil, Nil, Nil), crossSignedHtlcs: Seq[(UpdateAddHtlc, Origin)] = Nil): DATA_NORMAL = {
       val commitments = CommitmentsSpec.makeCommitments(500_000_000 msat, 500_000_000 msat, nodeParams.nodeId, remoteNodeId, announcement_opt = None)
-        .modify(_.active).apply(_.map(_.modify(_.remoteCommitParams.htlcMinimum).setTo(htlcMinimum)))
+        .modify(_.active).apply(_.map(_
+          .modify(_.remoteCommitParams.htlcMinimum).setTo(htlcMinimum)
+          .modify(_.localCommit.spec.htlcs).using(_ ++ crossSignedHtlcs.map { case (add, _) => OutgoingHtlc(add) })
+          .modify(_.remoteCommit.spec.htlcs).using(_ ++ crossSignedHtlcs.map { case (add, _) => IncomingHtlc(add) })
+        ))
         .modify(_.changes.localChanges).setTo(localChanges)
+        .modify(_.originChannels).setTo(crossSignedHtlcs.map { case (add, origin) => add.id -> origin }.toMap)
       DATA_NORMAL(commitments, ShortIdAliases(Alias(42), None), None, null, SpliceStatus.NoSplice, None, None, None)
     }
   }
@@ -389,7 +395,7 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
       upstreamChannel(60_000_000 msat, CltvExpiry(560), paymentHash2),
     )
     proposeFunding(40_000_000 msat, CltvExpiry(515), paymentHash2, upstream2.head)
-    signLiquidityPurchase(100_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash2 :: Nil))
+    val funded2 = signLiquidityPurchase(100_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash2 :: Nil))
     proposeExtraFunding(50_000_000 msat, CltvExpiry(525), paymentHash2, upstream2.last)
     register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
 
@@ -400,23 +406,47 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
       upstreamChannel(45_000_000 msat, CltvExpiry(560), paymentHash3),
     ))
     proposeFunding(100_000_000 msat, CltvExpiry(512), paymentHash3, upstream3)
-    signLiquidityPurchase(100_000 sat, LiquidityAds.PaymentDetails.FromChannelBalanceForFutureHtlc(paymentHash3 :: Nil))
+    val funded3 = signLiquidityPurchase(100_000 sat, LiquidityAds.PaymentDetails.FromChannelBalanceForFutureHtlc(paymentHash3 :: Nil))
 
     // A fourth funding is proposed coming from a trampoline payment.
     val paymentHash4 = randomBytes32()
     val upstream4 = Upstream.Hot.Trampoline(List(upstreamChannel(60_000_000 msat, CltvExpiry(560), paymentHash4)))
     proposeFunding(50_000_000 msat, CltvExpiry(516), paymentHash4, upstream4)
 
-    // The first three proposals reach their CLTV expiry (the extra htlc was already failed).
+    // The first three proposals reach their CLTV expiry.
     peer ! CurrentBlockHeight(BlockHeight(515))
-    val fwds = (0 until 5).map(_ => register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]])
+    // We immediately fail the upstream HTLCs for the first proposal, which hasn't been funded yet.
+    val fails = (0 until 2).map(_ => register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]])
+    assert(fails.map { f => (f.channelId, f.message.id) }.toSet == Set(
+      (upstream1.head.add.channelId, upstream1.head.add.id),
+      (upstream1.last.add.channelId, upstream1.last.add.id),
+    ))
+    fails.foreach(f => assert(f.message.reason == FailureReason.LocalFailure(UnknownNextPeer())))
+    fails.foreach(f => assert(f.message.commit))
+    // We force-close the two channels that have been funded, and then query their state to see if HTLCs have already
+    // been relayed (the extra htlc was already failed).
+    val cmds = (0 until 4).map(_ => register.expectMsgType[Register.Forward[Command]])
+    val closeCmds = cmds.collect { case cmd if cmd.message.isInstanceOf[CMD_FORCECLOSE] => cmd }
+    assert(closeCmds.map(_.channelId).toSet == Set(funded2.channelId, funded3.channelId))
+    val channelInfo = cmds.collect { case cmd if cmd.message.isInstanceOf[CMD_GET_CHANNEL_INFO] => cmd }
+    assert(channelInfo.map(_.channelId).toSet == Set(funded2.channelId, funded3.channelId))
     register.expectNoMessage(100 millis)
-    fwds.foreach(fwd => {
-      assert(fwd.message.reason == FailureReason.LocalFailure(UnknownNextPeer()))
-      assert(fwd.message.commit)
+    channelInfo.foreach(i => i.channelId match {
+      case funded2.channelId =>
+        // The HTLC has already been relayed downstream, so we don't fail the upstream HTLC and just close the channel.
+        val fundingFee = LiquidityAds.FundingFee(funded2.purchase.fees.total.toMilliSatoshi, funded2.txId)
+        val relayedHtlc = UpdateAddHtlc(funded2.channelId, 73, 40_000_000 msat, paymentHash2, CltvExpiry(515), TestConstants.emptyOnionPacket, TlvStream(UpdateAddHtlcTlv.FundingFeeTlv(fundingFee)))
+        val channelData = makeChannelData(crossSignedHtlcs = (relayedHtlc, Origin.Hot(ActorRef.noSender, upstream2.head)) :: Nil)
+        i.message.asInstanceOf[CMD_GET_CHANNEL_INFO].replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, i.channelId, ActorRef.noSender, NORMAL, channelData)
+        register.expectNoMessage(100 millis)
+      case _ =>
+        // The HTLC was *not* relayed downstream: we close the channel and fail the upstream HTLC.
+        val channelData = makeChannelData()
+        i.message.asInstanceOf[CMD_GET_CHANNEL_INFO].replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, i.channelId, ActorRef.noSender, NORMAL, channelData)
+        val fails = (0 until 2).map(_ => register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]])
+        assert(fails.map { f => (f.channelId, f.message.id) }.toSet == upstream3.received.map { u => (u.add.channelId, u.add.id) }.toSet)
+        register.expectNoMessage(100 millis)
     })
-    assert(fwds.map(_.channelId).toSet == (upstream1 ++ upstream2.slice(0, 1) ++ upstream3.received).map(_.add.channelId).toSet)
-    assert(fwds.map(_.message.id).toSet == (upstream1 ++ upstream2.slice(0, 1) ++ upstream3.received).map(_.add.id).toSet)
     awaitCond(nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId).isEmpty, interval = 100 millis)
   }
 
@@ -473,22 +503,26 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     // A first funding proposal is signed.
     val upstream1 = upstreamChannel(60_000_000 msat, CltvExpiry(560))
     proposeFunding(50_000_000 msat, CltvExpiry(520), upstream1.add.paymentHash, upstream1)
-    signLiquidityPurchase(75_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(upstream1.add.paymentHash :: Nil))
+    val funded1 = signLiquidityPurchase(75_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(upstream1.add.paymentHash :: Nil))
 
     // A second funding proposal is signed.
     val upstream2 = upstreamChannel(60_000_000 msat, CltvExpiry(560))
     proposeFunding(50_000_000 msat, CltvExpiry(525), upstream2.add.paymentHash, upstream2)
-    signLiquidityPurchase(80_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(upstream2.add.paymentHash :: Nil))
+    val funded2 = signLiquidityPurchase(80_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(upstream2.add.paymentHash :: Nil))
 
     // We don't fail signed proposals on disconnection.
     disconnect()
     register.expectNoMessage(100 millis)
 
-    // But if a funding proposal reaches its CLTV expiry, we fail it.
+    // But if a funding proposal reaches its CLTV expiry, we fail it and close the downstream channel.
     peer ! CurrentBlockHeight(BlockHeight(522))
-    val fwd1 = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
-    assert(fwd1.channelId == upstream1.add.channelId)
-    assert(fwd1.message.id == upstream1.add.id)
+    assert(register.expectMsgType[Register.Forward[CMD_FORCECLOSE]].channelId == funded1.channelId)
+    val fwd1 = register.expectMsgType[Register.Forward[CMD_GET_CHANNEL_INFO]]
+    assert(fwd1.channelId == funded1.channelId)
+    fwd1.message.replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, funded1.channelId, ActorRef.noSender, NORMAL, makeChannelData())
+    val fail1 = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
+    assert(fail1.channelId == upstream1.add.channelId)
+    assert(fail1.message.id == upstream1.add.id)
     register.expectNoMessage(100 millis)
     // We still have one pending proposal, so we don't stop.
     probe.expectNoMessage(100 millis)
@@ -500,10 +534,17 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
 
     // The last funding proposal reaches its CLTV expiry.
     peerAfterRestart ! CurrentBlockHeight(BlockHeight(525))
-    val fwd2 = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
-    assert(fwd2.channelId == upstream2.add.channelId)
-    assert(fwd2.message.id == upstream2.add.id)
+    assert(register.expectMsgType[Register.Forward[CMD_FORCECLOSE]].channelId == funded2.channelId)
+    val fwd2 = register.expectMsgType[Register.Forward[CMD_GET_CHANNEL_INFO]]
+    assert(fwd2.channelId == funded2.channelId)
+    fwd2.message.replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, funded2.channelId, ActorRef.noSender, NORMAL, makeChannelData())
+    val fail2 = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
+    assert(fail2.channelId == upstream2.add.channelId)
+    assert(fail2.message.id == upstream2.add.id)
     register.expectNoMessage(100 millis)
+    probe.expectNoMessage(100 millis)
+    // After one more block, we stop the actor.
+    peerAfterRestart ! CurrentBlockHeight(BlockHeight(526))
     probe.expectTerminated(peerAfterRestart.ref)
   }
 
@@ -1058,7 +1099,7 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     val preimage = randomBytes32()
     val paymentHash = Crypto.sha256(preimage)
     val upstream = upstreamChannel(11_000_000 msat, expiryIn, paymentHash)
-    proposeFunding(10_000_000 msat, expiryOut, paymentHash, upstream)
+    val willAdd = proposeFunding(10_000_000 msat, expiryOut, paymentHash, upstream)
     val fees = LiquidityAds.Fees(10_000 sat, 5_000 sat)
     val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(List(paymentHash)), fees = fees)
 
@@ -1073,7 +1114,8 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     awaitCond(!nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
 
     // When we retry relaying the HTLC, our peer fulfills it: we re-enable from_future_htlc.
-    peer ! OnTheFlyFunding.PaymentRelayer.RelaySuccess(purchase.channelId, paymentHash, preimage, fees.total.toMilliSatoshi)
+    val proposal = OnTheFlyFunding.Proposal(willAdd, upstream, Nil)
+    peer ! OnTheFlyFunding.PaymentRelayer.RelaySuccess(purchase.channelId, paymentHash, preimage, fees.total.toMilliSatoshi, proposal :: Nil)
     awaitCond(nodeParams.onTheFlyFundingConfig.isFromFutureHtlcAllowed(remoteNodeId))
   }
 
@@ -1131,9 +1173,13 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     peer ! ChannelReadyForPayments(channel.ref, remoteNodeId, purchase.channelId, purchase.txId, fundingTxIndex = 0)
     channel.expectNoMessage(100 millis)
     peer ! CurrentBlockHeight(BlockHeight(TestConstants.defaultBlockHeight))
-    val fwd = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
-    assert(fwd.channelId == upstream.add.channelId)
-    assert(fwd.message.id == upstream.add.id)
+    assert(register.expectMsgType[Register.Forward[CMD_FORCECLOSE]].channelId == purchase.channelId)
+    val fwd = register.expectMsgType[Register.Forward[CMD_GET_CHANNEL_INFO]]
+    assert(fwd.channelId == purchase.channelId)
+    fwd.message.replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, purchase.channelId, ActorRef.noSender, NORMAL, makeChannelData())
+    val fail = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
+    assert(fail.channelId == upstream.add.channelId)
+    assert(fail.message.id == upstream.add.id)
     awaitCond(nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId).isEmpty, interval = 100 millis)
   }
 
@@ -1157,6 +1203,101 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     register.expectNoMessage(100 millis)
   }
 
+  test("don't relay payments that have already been relayed") { f =>
+    import f._
+
+    // We create a channel, that can later be spliced.
+    connect(peer)
+    val channelId = openChannel(250_000 sat)
+
+    // We relay an on-the-fly payment.
+    val upstream = upstreamChannel(50_000_000 msat, expiryIn, paymentHash)
+    proposeFunding(50_000_000 msat, expiryOut, paymentHash, upstream)
+    val fees = LiquidityAds.Fees(1000 sat, 1000 sat)
+    val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromChannelBalanceForFutureHtlc(paymentHash :: Nil), channelId, fees, fundingTxIndex = 1)
+    peer ! ChannelReadyForPayments(channel.ref, remoteNodeId, channelId, purchase.txId, fundingTxIndex = 1)
+    channel.expectMsgType[CMD_GET_CHANNEL_INFO].replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, channelId, channel.ref, NORMAL, makeChannelData())
+    val cmd = channel.expectMsgType[CMD_ADD_HTLC]
+    cmd.replyTo ! RES_SUCCESS(cmd, channelId)
+    val htlc = UpdateAddHtlc(channelId, randomHtlcId(), cmd.amount, paymentHash, cmd.cltvExpiry, cmd.onion, cmd.nextPathKey_opt, cmd.reputationScore.accountable, cmd.fundingFee_opt)
+    channel.expectNoMessage(100 millis)
+
+    // On restart, the HTLC has been fully cross-signed: it isn't part of our local changes anymore, it only appears in
+    // our commitments. We must still detect it, otherwise we would relay the same payment twice.
+    val peerAfterRestart = TestFSMRef(new Peer(nodeParams, remoteNodeId, new DummyOnChainWallet(), FakeChannelFactory(remoteNodeId, channel), TestProbe().ref, register.ref, TestProbe().ref, TestProbe().ref))
+    peerAfterRestart ! Peer.Init(Set.empty, nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId))
+    connect(peerAfterRestart)
+    peerAfterRestart ! ChannelReadyForPayments(channel.ref, remoteNodeId, channelId, purchase.txId, fundingTxIndex = 1)
+    val channelData = makeChannelData(crossSignedHtlcs = Seq(htlc -> Origin.Cold(Upstream.Cold(upstream))))
+    channel.expectMsgType[CMD_GET_CHANNEL_INFO].replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, channelId, channel.ref, NORMAL, channelData)
+    channel.expectNoMessage(100 millis)
+    register.expectNoMessage(100 millis)
+  }
+
+  test("fulfill upstream when preimage is received after the htlc expiry") { f =>
+    import f._
+
+    connect(peer)
+
+    // We relay an on-the-fly payment.
+    val upstream = upstreamChannel(50_000_000 msat, expiryIn, paymentHash)
+    proposeFunding(50_000_000 msat, expiryOut, paymentHash, upstream)
+    val fees = LiquidityAds.Fees(1000 sat, 1000 sat)
+    val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash :: Nil), fees = fees)
+    peer ! ChannelReadyForPayments(channel.ref, remoteNodeId, purchase.channelId, purchase.txId, fundingTxIndex = 0)
+    channel.expectMsgType[CMD_GET_CHANNEL_INFO].replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, purchase.channelId, channel.ref, NORMAL, makeChannelData())
+    val cmd = channel.expectMsgType[CMD_ADD_HTLC]
+    cmd.replyTo ! RES_SUCCESS(cmd, purchase.channelId)
+    val htlc = UpdateAddHtlc(purchase.channelId, randomHtlcId(), cmd.amount, paymentHash, cmd.cltvExpiry, cmd.onion, cmd.nextPathKey_opt, cmd.reputationScore.accountable, cmd.fundingFee_opt)
+    channel.expectNoMessage(100 millis)
+
+    // Our peer doesn't settle that HTLC: when it reaches its expiry, we force-close the channel and stop tracking the
+    // payment. The HTLC has been cross-signed, so we must not fail the upstream HTLCs.
+    peer ! CurrentBlockHeight(expiryOut.blockHeight)
+    assert(register.expectMsgType[Register.Forward[CMD_FORCECLOSE]].channelId == purchase.channelId)
+    val channelInfo = register.expectMsgType[Register.Forward[CMD_GET_CHANNEL_INFO]]
+    assert(channelInfo.channelId == purchase.channelId)
+    channelInfo.message.replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, purchase.channelId, ActorRef.noSender, NORMAL, makeChannelData(crossSignedHtlcs = Seq(htlc -> cmd.origin)))
+    register.expectNoMessage(100 millis)
+    awaitCond(nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId).isEmpty, interval = 100 millis)
+
+    // Our peer claims that HTLC on-chain by revealing the preimage: even though we've stopped tracking that payment,
+    // we must fulfill the upstream HTLCs, otherwise we've paid our peer without being paid ourselves.
+    cmd.replyTo ! RES_ADD_SETTLED(cmd.origin, remoteNodeId, htlc, HtlcResult.OnChainFulfill(preimage))
+    verifyFulfilledUpstream(upstream, preimage)
+    register.expectNoMessage(100 millis)
+  }
+
+  test("fulfill upstream when fulfill races the htlc expiry") { f =>
+    import f._
+
+    connect(peer)
+
+    // We relay an on-the-fly payment.
+    val upstream = upstreamChannel(50_000_000 msat, expiryIn, paymentHash)
+    proposeFunding(50_000_000 msat, expiryOut, paymentHash, upstream)
+    val fees = LiquidityAds.Fees(1000 sat, 1000 sat)
+    val purchase = signLiquidityPurchase(200_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(paymentHash :: Nil), fees = fees)
+    peer ! ChannelReadyForPayments(channel.ref, remoteNodeId, purchase.channelId, purchase.txId, fundingTxIndex = 0)
+    channel.expectMsgType[CMD_GET_CHANNEL_INFO].replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, purchase.channelId, channel.ref, NORMAL, makeChannelData())
+    val cmd = channel.expectMsgType[CMD_ADD_HTLC]
+    cmd.replyTo ! RES_SUCCESS(cmd, purchase.channelId)
+    val htlc = UpdateAddHtlc(purchase.channelId, randomHtlcId(), cmd.amount, paymentHash, cmd.cltvExpiry, cmd.onion, cmd.nextPathKey_opt, cmd.reputationScore.accountable, cmd.fundingFee_opt)
+    channel.expectNoMessage(100 millis)
+
+    // We reach the HTLC expiry and stop tracking the payment.
+    peer ! CurrentBlockHeight(expiryOut.blockHeight)
+    assert(register.expectMsgType[Register.Forward[CMD_FORCECLOSE]].channelId == purchase.channelId)
+    val channelInfo = register.expectMsgType[Register.Forward[CMD_GET_CHANNEL_INFO]]
+    channelInfo.message.replyTo ! RES_GET_CHANNEL_INFO(remoteNodeId, purchase.channelId, ActorRef.noSender, NORMAL, makeChannelData(crossSignedHtlcs = Seq(htlc -> cmd.origin)))
+    awaitCond(nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId).isEmpty, interval = 100 millis)
+
+    // Our peer had actually fulfilled that HTLC off-chain, racing our expiry: we must fulfill the upstream HTLCs.
+    cmd.replyTo ! RES_ADD_SETTLED(cmd.origin, remoteNodeId, htlc, HtlcResult.RemoteFulfill(UpdateFulfillHtlc(purchase.channelId, htlc.id, preimage)))
+    verifyFulfilledUpstream(upstream, preimage)
+    register.expectNoMessage(100 millis)
+  }
+
   test("stop when disconnecting without pending proposals") { f =>
     import f._