Explicitly match on-the-fly HTLCs after a restart (#3357)

### eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/OnTheFlyFunding.scala
@@ -98,6 +98,13 @@ object OnTheFlyFunding {
     /** Maximum fees that can be collected from this HTLC. */
     def maxFees(htlcMinimum: MilliSatoshi): MilliSatoshi = (htlc.amount - htlcMinimum).max(0 msat)
 
+    /** Incoming HTLCs that are paying for this proposal, identified by their channel and HTLC id. */
+    def upstreamHtlcs: Set[(ByteVector32, Long)] = upstream match {
+      case _: Upstream.Local => Set.empty
+      case u: Upstream.Hot.Channel => Set((u.add.channelId, u.add.id))
+      case u: Upstream.Hot.Trampoline => u.received.map(r => (r.add.channelId, r.add.id)).toSet
+    }
+
     /** Create commands to fail all upstream HTLCs. */
     def createFailureCommands(failure_opt: Option[FailureReason]): Seq[(ByteVector32, CMD_FAIL_HTLC)] = upstream match {
       case _: Upstream.Local => Nil
@@ -150,6 +157,9 @@ object OnTheFlyFunding {
     /** Maximum fees that can be collected from this HTLC set. */
     def maxFees(htlcMinimum: MilliSatoshi): MilliSatoshi = proposed.map(_.maxFees(htlcMinimum)).sum
 
+    /** Incoming HTLCs that are paying for this payment, identified by their channel and HTLC id. */
+    def upstreamHtlcs: Set[(ByteVector32, Long)] = proposed.flatMap(_.upstreamHtlcs).toSet
+
     /** Create commands to fail all upstream HTLCs. */
     def createFailureCommands(): Seq[(ByteVector32, CMD_FAIL_HTLC)] = proposed.flatMap(_.createFailureCommands(None))
 

### eclair-core/src/main/scala/fr/acinq/eclair/payment/relay/PostRestartHtlcCleaner.scala
@@ -69,7 +69,11 @@ class PostRestartHtlcCleaner(nodeParams: NodeParams, register: ActorRef, initial
       // result upstream to preserve channels.
       val brokenHtlcs: BrokenHtlcs = {
         val channels = listLocalChannels(init.channels)
-        val onTheFlyPayments = nodeParams.db.liquidity.listPendingOnTheFlyPayments().values.flatten.toSet
+        // Note that we identify those HTLCs by their channel and HTLC id, and not by their payment_hash: otherwise a
+        // malicious peer could pin unrelated HTLCs by reusing the payment_hash of a pending on-the-fly payment, which
+        // could trigger a force-close on the corresponding channel because we wouldn't correctly fail back the HTLC
+        // before its timeout.
+        val onTheFlyHtlcs: Set[(ByteVector32, Long)] = nodeParams.db.liquidity.listPendingOnTheFlyFunding().values.flatMap(_.values.flatMap(_.upstreamHtlcs)).toSet
         val nonStandardIncomingHtlcs: Seq[IncomingHtlc] = nodeParams.pluginParams.collect { case p: CustomCommitmentsPlugin => p.getIncomingHtlcs(nodeParams, log) }.flatten
         val htlcsIn: Seq[IncomingHtlc] = getIncomingHtlcs(channels.map(_.channelData), nodeParams.db.payments, nodeParams.privateKey, nodeParams.features) ++ nonStandardIncomingHtlcs
         val nonStandardRelayedOutHtlcs: Map[Origin.Cold, Set[(ByteVector32, Long)]] = nodeParams.pluginParams.collect { case p: CustomCommitmentsPlugin => p.getHtlcsRelayedOut(htlcsIn, nodeParams, log) }.flatten.toMap
@@ -87,7 +91,7 @@ class PostRestartHtlcCleaner(nodeParams: NodeParams, register: ActorRef, initial
         log.info(s"htlcsIn=${htlcsIn.length} notRelayed=${notRelayed.length} relayedOut=${relayedOut.values.flatten.size}")
         log.info("notRelayed={}", notRelayed.map(htlc => (htlc.add.channelId, htlc.add.id)))
         log.info("relayedOut={}", relayedOut)
-        BrokenHtlcs(notRelayed, relayedOut, Set.empty, onTheFlyPayments)
+        BrokenHtlcs(notRelayed, relayedOut, Set.empty, onTheFlyHtlcs)
       }
 
       Metrics.PendingNotRelayed.update(brokenHtlcs.notRelayed.size)
@@ -122,7 +126,7 @@ class PostRestartHtlcCleaner(nodeParams: NodeParams, register: ActorRef, initial
                 } else {
                   log.info(s"got preimage but upstream channel is closed for htlc=$htlc")
                 }
-              case None if brokenHtlcs.pendingPayments.contains(htlc.paymentHash) =>
+              case None if brokenHtlcs.pendingOnTheFly.contains((htlc.channelId, htlc.id)) =>
                 // We don't fail on-the-fly HTLCs that have been funded: we haven't been paid our fee yet, so we will
                 // retry relaying them unless we reach the HTLC timeout.
                 log.info("htlc #{} from channelId={} wasn't relayed, but has a pending on-the-fly relay (paymentHash={})", htlc.id, htlc.channelId, htlc.paymentHash)
@@ -343,9 +347,10 @@ object PostRestartHtlcCleaner {
    * @param notRelayed      incoming HTLCs that were committed upstream but not relayed downstream.
    * @param relayedOut      outgoing HTLC sets that may have been incompletely sent and need to be watched.
    * @param settledUpstream upstream payments that have already been settled (failed or fulfilled) by this actor.
-   * @param pendingPayments payments that are pending and will be relayed: we mustn't fail them upstream.
+   * @param pendingOnTheFly incoming HTLCs that are paying for a pending on-the-fly funding proposal that we will retry
+   *                        relaying: we mustn't fail them upstream.
    */
-  case class BrokenHtlcs(notRelayed: Seq[IncomingHtlc], relayedOut: Map[Origin.Cold, Set[(ByteVector32, Long)]], settledUpstream: Set[Origin.Cold], pendingPayments: Set[ByteVector32])
+  case class BrokenHtlcs(notRelayed: Seq[IncomingHtlc], relayedOut: Map[Origin.Cold, Set[(ByteVector32, Long)]], settledUpstream: Set[Origin.Cold], pendingOnTheFly: Set[(ByteVector32, Long)])
 
   /** Returns true if the given HTLC matches the given origin. */
   private def matchesOrigin(htlcIn: UpdateAddHtlc, origin: Origin.Cold): Boolean = origin.upstream match {

### eclair-core/src/test/scala/fr/acinq/eclair/payment/PostRestartHtlcCleanerSpec.scala
@@ -190,6 +190,43 @@ class PostRestartHtlcCleanerSpec extends TestKitBaseClass with FixtureAnyFunSuit
     channel.expectNoMessage(100 millis)
   }
 
+  test("fail upstream HTLCs that reuse an on-the-fly funding payment_hash") { f =>
+    import f._
+
+    // Our peer has a pending on-the-fly funding proposal that was funded, and then sends us more HTLCs that reuse the
+    // same payment_hash: those additional HTLCs must be failed, otherwise our peer can force-close our upstream channels.
+    val paymentHash = randomBytes32()
+
+    val htlc_ab_1 = Seq(
+      buildHtlcIn(0, channelId_ab_1, paymentHash),
+      buildHtlcIn(1, channelId_ab_1, paymentHash),
+    )
+    val htlc_ab_2 = Seq(
+      buildHtlcIn(2, channelId_ab_2, paymentHash),
+    )
+
+    val channels = Seq(
+      ChannelCodecsSpec.makeChannelDataNormal(htlc_ab_1, Map.empty),
+      ChannelCodecsSpec.makeChannelDataNormal(htlc_ab_2, Map.empty)
+    )
+
+    // Only the first HTLC is paying for the on-the-fly funding proposal.
+    val upstream = Upstream.Hot.Channel(htlc_ab_1.head.add, TimestampMilli.now(), a, 0.1)
+    val pending = OnTheFlyFunding.Pending(Seq(OnTheFlyFunding.Proposal(createWillAdd(100_000 msat, paymentHash, CltvExpiry(500)), upstream, Nil)), createStatus())
+    nodeParams.db.liquidity.addPendingOnTheFlyFunding(randomKey().publicKey, pending)
+
+    val channel = TestProbe()
+    val (relayer, _) = f.createRelayer(nodeParams)
+    relayer ! PostRestartHtlcCleaner.Init(channels.map(_.withChannelKeys(nodeParams)))
+    // The HTLC that is paying for the on-the-fly funding is kept, but the other ones are failed.
+    system.eventStream.publish(ChannelStateChanged(channel.ref, channels.head.channelId, system.deadLetters, a, OFFLINE, NORMAL, Some(channels.head.commitments)))
+    channel.expectMsg(CMD_FAIL_HTLC(1, FailureReason.LocalFailure(TemporaryNodeFailure()), None, commit = true))
+    channel.expectNoMessage(100 millis)
+    system.eventStream.publish(ChannelStateChanged(channel.ref, channels(1).channelId, system.deadLetters, a, OFFLINE, NORMAL, Some(channels(1).commitments)))
+    channel.expectMsg(CMD_FAIL_HTLC(2, FailureReason.LocalFailure(TemporaryNodeFailure()), None, commit = true))
+    channel.expectNoMessage(100 millis)
+  }
+
   test("clean up upstream HTLCs for which we're the final recipient") { f =>
     import f._