Allow aborting liquidity purchases after signing (#3206)

diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala b/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala
index 928ab70..4c82bac 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala
@@ -46,7 +46,7 @@ import fr.acinq.eclair.crypto.keymanager.ChannelKeys
 import fr.acinq.eclair.db.DbEventHandler.ChannelEvent.EventType
 import fr.acinq.eclair.db.PendingCommandsDb
 import fr.acinq.eclair.io.Peer
-import fr.acinq.eclair.io.Peer.LiquidityPurchaseSigned
+import fr.acinq.eclair.io.Peer.{LiquidityPurchaseAborted, LiquidityPurchaseSigned}
 import fr.acinq.eclair.payment.relay.Relayer
 import fr.acinq.eclair.payment.{Bolt11Invoice, PaymentSettlingOnChain}
 import fr.acinq.eclair.reputation.Reputation
@@ -1363,6 +1363,9 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
         case SpliceStatus.SpliceWaitingForSigs(signingSession) =>
           log.info("our peer aborted the splice attempt: ascii='{}' bin={}", msg.toAscii, msg.data)
           rollbackFundingAttempt(signingSession.fundingTx.tx, previousTxs = Seq.empty) // no splice rbf yet
+          signingSession.liquidityPurchase_opt.collect {
+            case purchase if !signingSession.fundingParams.isInitiator => peer ! LiquidityPurchaseAborted(d.channelId, signingSession.fundingTx.txId, signingSession.fundingTxIndex)
+          }
           stay() using d.copy(spliceStatus = SpliceStatus.NoSplice) sending TxAbort(d.channelId, SpliceAttemptAborted(d.channelId).getMessage) calling endQuiescence(d)
         case SpliceStatus.SpliceRequested(cmd, _) =>
           log.info("our peer rejected our splice attempt: ascii='{}' bin={}", msg.toAscii, msg.data)
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenDualFunded.scala b/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenDualFunded.scala
index 586b14a..b051e08 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenDualFunded.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenDualFunded.scala
@@ -25,7 +25,7 @@ import fr.acinq.eclair.channel.fund.InteractiveTxBuilder.{FullySignedSharedTrans
 import fr.acinq.eclair.channel.fund.{InteractiveTxBuilder, InteractiveTxSigningSession}
 import fr.acinq.eclair.channel.publish.TxPublisher.SetChannelId
 import fr.acinq.eclair.crypto.ShaChain
-import fr.acinq.eclair.io.Peer.{LiquidityPurchaseSigned, OpenChannelResponse}
+import fr.acinq.eclair.io.Peer.{LiquidityPurchaseAborted, LiquidityPurchaseSigned, OpenChannelResponse}
 import fr.acinq.eclair.transactions.Transactions
 import fr.acinq.eclair.wire.protocol._
 import fr.acinq.eclair.{ToMilliSatoshiConversion, randomBytes32}
@@ -412,6 +412,9 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
         case msg: TxAbort =>
           log.info("our peer aborted the dual funding flow: ascii='{}' bin={}", msg.toAscii, msg.data)
           rollbackFundingAttempt(d.signingSession.fundingTx.tx, Nil)
+          d.signingSession.liquidityPurchase_opt.collect {
+            case purchase if !d.signingSession.fundingParams.isInitiator => peer ! LiquidityPurchaseAborted(d.channelId, d.signingSession.fundingTx.txId, d.signingSession.fundingTxIndex)
+          }
           goto(CLOSED) using IgnoreClosedData(d) sending TxAbort(d.channelId, DualFundingAborted(d.channelId).getMessage)
         case msg: InteractiveTxConstructionMessage =>
           log.info("received unexpected interactive-tx message: {}", msg.getClass.getSimpleName)
diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala b/eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala
index 555a053..c782a1f 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/io/Peer.scala
@@ -740,9 +740,8 @@ class Peer(val nodeParams: NodeParams,
       }
       // We signed a liquidity purchase from our peer. At that point we're not 100% sure yet it will succeed: if
       // we disconnect before our peer sends their signature, the funding attempt may be cancelled when reconnecting.
-      // If that happens, the on-the-fly proposal will stay in our state until we reach the CLTV expiry, at which
-      // point we will forget it and fail the upstream HTLCs. This is also what would happen if we successfully
-      // funded the channel, but it closed before we could relay the HTLCs.
+      // If that happens, we will receive a LiquidityPurchaseAborted event, which is handled below, at which point we
+      // will forget the purchase and fail the upstream HTLCs.
       val (paymentHashes, feesOwed) = e.purchase.paymentDetails match {
         case LiquidityAds.PaymentDetails.FromChannelBalance => (Nil, 0 msat)
         case p: LiquidityAds.PaymentDetails.FromChannelBalanceForFutureHtlc => (p.paymentHashes, 0 msat)
@@ -786,6 +785,19 @@ class Peer(val nodeParams: NodeParams,
         })
       stay()
 
+    case Event(e: LiquidityPurchaseAborted, _: ConnectedData) =>
+      pendingOnTheFlyFunding.foreach {
+        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) }
+            nodeParams.db.liquidity.removePendingOnTheFlyFunding(remoteNodeId, paymentHash)
+            pendingOnTheFlyFunding -= paymentHash
+          case _ => ()
+        }
+      }
+      stay()
+
     case Event(e: OnTheFlyFunding.PaymentRelayer.RelayResult, _) =>
       e match {
         case success: OnTheFlyFunding.PaymentRelayer.RelaySuccess =>
@@ -1133,6 +1145,7 @@ object Peer {
 
   /** We signed a funding transaction where our peer purchased some liquidity. */
   case class LiquidityPurchaseSigned(channelId: ByteVector32, txId: TxId, fundingTxIndex: Long, htlcMinimum: MilliSatoshi, purchase: LiquidityAds.Purchase)
+  case class LiquidityPurchaseAborted(channelId: ByteVector32, txId: TxId, fundingTxIndex: Long)
 
   case class OnTheFlyFundingTimeout(paymentHash: ByteVector32)
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/b/WaitForDualFundingSignedStateSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/b/WaitForDualFundingSignedStateSpec.scala
index 04c6a37..05b0909 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/b/WaitForDualFundingSignedStateSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/b/WaitForDualFundingSignedStateSpec.scala
@@ -29,7 +29,7 @@ import fr.acinq.eclair.channel.fund.InteractiveTxBuilder.{FullySignedSharedTrans
 import fr.acinq.eclair.channel.publish.TxPublisher
 import fr.acinq.eclair.channel.states.{ChannelStateTestsBase, ChannelStateTestsTags}
 import fr.acinq.eclair.crypto.NonceGenerator
-import fr.acinq.eclair.io.Peer.{LiquidityPurchaseSigned, OpenChannelResponse}
+import fr.acinq.eclair.io.Peer.{LiquidityPurchaseAborted, LiquidityPurchaseSigned, OpenChannelResponse}
 import fr.acinq.eclair.transactions.Transactions._
 import fr.acinq.eclair.wire.protocol._
 import fr.acinq.eclair.{Features, MilliSatoshiLong, TestConstants, TestKitBaseClass, ToMilliSatoshiConversion, randomBytes32, randomKey}
@@ -233,6 +233,35 @@ class WaitForDualFundingSignedStateSpec extends TestKitBaseClass with FixtureAny
     assert(aliceData.latestFundingTx.sharedTx.asInstanceOf[FullySignedSharedTransaction].signedTx.txid == fundingTxId)
   }
 
+  test("complete interactive-tx protocol then aborts (with liquidity ads)", Tag(ChannelStateTestsTags.DualFunding), Tag(ChannelStateTestsTags.LiquidityAds)) { f =>
+    import f._
+
+    // We don't forward signatures to simulate an abort.
+    bob2alice.expectMsgType[CommitSig]
+    alice2bob.expectMsgType[CommitSig]
+
+    // Bob signed a liquidity purchase.
+    val purchase = bobPeer.fishForMessage() {
+      case l: LiquidityPurchaseSigned =>
+        assert(l.purchase.paymentDetails == LiquidityAds.PaymentDetails.FromChannelBalance)
+        assert(l.fundingTxIndex == 0)
+        true
+      case _ => false
+    }
+
+    // Alice aborts the channel open.
+    alice2bob.forward(bob, TxAbort(alice.stateData.channelId, "changed my mind"))
+    bobPeer.fishForMessage() {
+      case l: LiquidityPurchaseAborted =>
+        assert(l.txId == purchase.asInstanceOf[LiquidityPurchaseSigned].txId)
+        assert(l.fundingTxIndex == 0)
+        true
+      case _ => false
+    }
+    bobListener.expectMsgType[ChannelAborted]
+    awaitCond(bob.stateName == CLOSED)
+  }
+
   test("recv invalid CommitSig", Tag(ChannelStateTestsTags.DualFunding)) { f =>
     import f._
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala
index 9f997cb..19f9ff9 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala
@@ -35,7 +35,7 @@ import fr.acinq.eclair.channel.publish.TxPublisher.SetChannelId
 import fr.acinq.eclair.channel.states.ChannelStateTestsBase.{FakeTxPublisherFactory, PimpTestFSM}
 import fr.acinq.eclair.channel.states.{ChannelStateTestsBase, ChannelStateTestsTags}
 import fr.acinq.eclair.db.RevokedHtlcInfoCleaner.ForgetHtlcInfos
-import fr.acinq.eclair.io.Peer.LiquidityPurchaseSigned
+import fr.acinq.eclair.io.Peer.{LiquidityPurchaseAborted, LiquidityPurchaseSigned}
 import fr.acinq.eclair.payment.relay.Relayer
 import fr.acinq.eclair.reputation.Reputation
 import fr.acinq.eclair.testutils.PimpTestProbe.convert
@@ -466,6 +466,59 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     }
   }
 
+  test("recv CMD_SPLICE (splice-in, liquidity ads, aborted)") { f =>
+    import f._
+
+    val sender = TestProbe()
+    val fundingRequest = LiquidityAds.RequestFunding(400_000 sat, TestConstants.defaultLiquidityRates.fundingRates.head, LiquidityAds.PaymentDetails.FromChannelBalance)
+    val cmd = CMD_SPLICE(sender.ref, Some(SpliceIn(500_000 sat)), None, Some(fundingRequest), None)
+    alice ! cmd
+
+    exchangeStfu(alice, bob, alice2bob, bob2alice)
+    assert(alice2bob.expectMsgType[SpliceInit].requestFunding_opt.nonEmpty)
+    alice2bob.forward(bob)
+    assert(bob2alice.expectMsgType[SpliceAck].willFund_opt.nonEmpty)
+    bob2alice.forward(alice)
+    alice2bob.expectMsgType[TxAddInput]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[TxAddInput]
+    bob2alice.forward(alice)
+    alice2bob.expectMsgType[TxAddInput]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[TxAddOutput]
+    bob2alice.forward(alice)
+    alice2bob.expectMsgType[TxAddOutput]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[TxComplete]
+    bob2alice.forward(alice)
+    alice2bob.expectMsgType[TxAddOutput]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[TxComplete]
+    bob2alice.forward(alice)
+    alice2bob.expectMsgType[TxComplete]
+    alice2bob.forward(bob)
+
+    // Bob signed a liquidity purchase.
+    val purchase = bobPeer.fishForMessage() {
+      case l: LiquidityPurchaseSigned =>
+        assert(l.purchase.paymentDetails == LiquidityAds.PaymentDetails.FromChannelBalance)
+        assert(l.fundingTxIndex == 1)
+        true
+      case _ => false
+    }
+
+    // Alice aborts the splice instead of signing it: Bob will notify the peer actor to allow failing the corresponding
+    // upstream HTLCs instead of holding them until they expire.
+    alice2bob.forward(bob, TxAbort(alice.stateData.channelId, "thanks but no thanks"))
+    bobPeer.fishForMessage() {
+      case l: LiquidityPurchaseAborted =>
+        assert(l.txId == purchase.asInstanceOf[LiquidityPurchaseSigned].txId)
+        assert(l.fundingTxIndex == 1)
+        true
+      case _ => false
+    }
+  }
+
   test("recv CMD_SPLICE (splice-in, liquidity ads, invalid will_fund signature)") { f =>
     import f._
 
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/OnTheFlyFundingSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/OnTheFlyFundingSpec.scala
index da6f822..bdbde36 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/OnTheFlyFundingSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/payment/relay/OnTheFlyFundingSpec.scala
@@ -22,6 +22,7 @@ import akka.testkit.{TestFSMRef, TestProbe}
 import com.softwaremill.quicklens.ModifyPimp
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
 import fr.acinq.bitcoin.scalacompat.{Block, ByteVector32, Crypto, Satoshi, SatoshiLong, TxId}
+import fr.acinq.eclair.TestUtils.randomTxId
 import fr.acinq.eclair.blockchain.{CurrentBlockHeight, DummyOnChainWallet}
 import fr.acinq.eclair.channel.Upstream.Hot
 import fr.acinq.eclair.channel._
@@ -436,6 +437,34 @@ class OnTheFlyFundingSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike {
     peerConnection.expectNoMessage(100 millis)
   }
 
+  test("signed on-the-fly funding aborted") { f =>
+    import f._
+
+    connect(peer)
+
+    // A funding proposal is signed.
+    val upstream1 = upstreamChannel(60_000_000 msat, CltvExpiry(560))
+    proposeFunding(50_000_000 msat, CltvExpiry(520), upstream1.add.paymentHash, upstream1)
+    val purchase = signLiquidityPurchase(75_000 sat, LiquidityAds.PaymentDetails.FromFutureHtlc(upstream1.add.paymentHash :: Nil))
+
+    // An unrelated liquidity purchase is aborted.
+    peer ! LiquidityPurchaseAborted(purchase.channelId, randomTxId(), purchase.fundingTxIndex)
+    register.expectNoMessage(100 millis)
+    awaitCond(nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId).contains(upstream1.add.paymentHash))
+
+    // Our peer then decides to abort instead of exchanging splice signatures.
+    peer ! LiquidityPurchaseAborted(purchase.channelId, purchase.txId, purchase.fundingTxIndex)
+
+    // We fail the corresponding upstream HTLC.
+    val fwd = register.expectMsgType[Register.Forward[CMD_FAIL_HTLC]]
+    assert(fwd.channelId == upstream1.add.channelId)
+    assert(fwd.message.id == upstream1.add.id)
+    register.expectNoMessage(100 millis)
+
+    // And forget the liquidity purchase.
+    awaitCond(nodeParams.db.liquidity.listPendingOnTheFlyFunding(remoteNodeId).isEmpty)
+  }
+
   test("signed on-the-fly funding HTLC timeout after disconnection") { f =>
     import f._