Allow non-initiator RBF for dual funding (#3021)

diff --git a/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelExceptions.scala b/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelExceptions.scala
index 8169685..ed61f8f 100644
--- a/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelExceptions.scala
+++ b/eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelExceptions.scala
@@ -95,8 +95,8 @@ case class InvalidSpliceTxAbortNotAcked            (override val channelId: Byte
 case class InvalidSpliceNotQuiescent               (override val channelId: ByteVector32) extends ChannelException(channelId, "invalid splice attempt: the channel is not quiescent")
 case class InvalidSpliceWithUnconfirmedTx          (override val channelId: ByteVector32, fundingTx: TxId) extends ChannelException(channelId, s"invalid splice attempt: the current funding transaction is still unconfirmed (txId=$fundingTx), you should use tx_init_rbf instead")
 case class InvalidRbfTxConfirmed                   (override val channelId: ByteVector32) extends ChannelException(channelId, "no need to rbf, transaction is already confirmed")
-case class InvalidRbfNonInitiator                  (override val channelId: ByteVector32) extends ChannelException(channelId, "cannot initiate rbf: we're not the initiator of this interactive-tx attempt")
 case class InvalidRbfZeroConf                      (override val channelId: ByteVector32) extends ChannelException(channelId, "cannot initiate rbf: we're using zero-conf for this interactive-tx attempt")
+case class InvalidRbfOverridesLiquidityPurchase    (override val channelId: ByteVector32, purchasedAmount: Satoshi) extends ChannelException(channelId, s"cannot initiate rbf attempt: our peer wanted to purchase $purchasedAmount of liquidity that we would override, they must initiate rbf")
 case class InvalidRbfMissingLiquidityPurchase      (override val channelId: ByteVector32, expectedAmount: Satoshi) extends ChannelException(channelId, s"cannot accept rbf attempt: the previous attempt contained a liquidity purchase of $expectedAmount but this one doesn't contain any liquidity purchase")
 case class InvalidRbfAttempt                       (override val channelId: ByteVector32) extends ChannelException(channelId, "invalid rbf attempt")
 case class NoMoreHtlcsClosingInProgress            (override val channelId: ByteVector32) extends ChannelException(channelId, "cannot send new htlcs, closing in progress")
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 3cc1fce..0877492 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
@@ -481,17 +481,21 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
       }
 
     case Event(cmd: CMD_BUMP_FUNDING_FEE, d: DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED) =>
-      if (!d.latestFundingTx.fundingParams.isInitiator) {
-        cmd.replyTo ! RES_FAILURE(cmd, InvalidRbfNonInitiator(d.channelId))
-        stay()
-      } else if (d.commitments.channelParams.zeroConf) {
-        cmd.replyTo ! RES_FAILURE(cmd, InvalidRbfZeroConf(d.channelId))
-        stay()
-      } else if (cmd.requestFunding_opt.isEmpty && d.latestFundingTx.liquidityPurchase_opt.nonEmpty) {
-        cmd.replyTo ! RES_FAILURE(cmd, InvalidRbfMissingLiquidityPurchase(d.channelId, d.latestFundingTx.liquidityPurchase_opt.get.amount))
-        stay()
-      } else {
-        d.status match {
+      d.latestFundingTx.liquidityPurchase_opt match {
+        case Some(purchase) if !d.latestFundingTx.fundingParams.isInitiator =>
+          // If we're not the channel initiator and they are purchasing liquidity, they must initiate RBF, otherwise
+          // the liquidity purchase will be lost (since only the initiator can purchase liquidity).
+          cmd.replyTo ! RES_FAILURE(cmd, InvalidRbfOverridesLiquidityPurchase(d.channelId, purchase.amount))
+          stay()
+        case Some(purchase) if cmd.requestFunding_opt.isEmpty =>
+          // If we were purchasing liquidity, we must keep purchasing liquidity across RBF attempts, otherwise our
+          // peer will simply reject the RBF attempt.
+          cmd.replyTo ! RES_FAILURE(cmd, InvalidRbfMissingLiquidityPurchase(d.channelId, purchase.amount))
+          stay()
+        case _ if d.commitments.channelParams.zeroConf =>
+          cmd.replyTo ! RES_FAILURE(cmd, InvalidRbfZeroConf(d.channelId))
+          stay()
+        case _ => d.status match {
           case DualFundingStatus.WaitingForConfirmations =>
             val minNextFeerate = d.latestFundingTx.fundingParams.minNextFeerate
             if (cmd.targetFeerate < minNextFeerate) {
@@ -509,11 +513,7 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
       }
 
     case Event(msg: TxInitRbf, d: DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED) =>
-      if (d.latestFundingTx.fundingParams.isInitiator) {
-        // Only the initiator is allowed to initiate RBF.
-        log.info("rejecting tx_init_rbf, we're the initiator, not them!")
-        stay() sending Error(d.channelId, InvalidRbfNonInitiator(d.channelId).getMessage)
-      } else if (d.commitments.channelParams.zeroConf) {
+      if (d.commitments.channelParams.zeroConf) {
         log.info("rejecting tx_init_rbf, we're using zero-conf")
         stay() using d.copy(status = DualFundingStatus.RbfAborted) sending TxAbort(d.channelId, InvalidRbfZeroConf(d.channelId).getMessage)
       } else {
@@ -551,6 +551,7 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
                   val fundingContribution = willFund_opt.map(_.purchase.amount).getOrElse(d.latestFundingTx.fundingParams.localContribution)
                   log.info("accepting rbf with remote.in.amount={} local.in.amount={}", msg.fundingContribution, fundingContribution)
                   val fundingParams = d.latestFundingTx.fundingParams.copy(
+                    isInitiator = false,
                     localContribution = fundingContribution,
                     remoteContribution = msg.fundingContribution,
                     lockTime = msg.lockTime,
@@ -594,6 +595,7 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
           stay() using d.copy(status = DualFundingStatus.RbfAborted) sending TxAbort(d.channelId, error.getMessage)
         case DualFundingStatus.RbfRequested(cmd) =>
           val fundingParams = d.latestFundingTx.fundingParams.copy(
+            isInitiator = true,
             // we don't change our funding contribution
             remoteContribution = msg.fundingContribution,
             lockTime = cmd.lockTime,
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/ChannelStateTestsHelperMethods.scala b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/ChannelStateTestsHelperMethods.scala
index a06b49b..1e83d05 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/ChannelStateTestsHelperMethods.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/ChannelStateTestsHelperMethods.scala
@@ -93,6 +93,8 @@ object ChannelStateTestsTags {
   val RejectRbfAttempts = "reject_rbf_attempts"
   /** If set, the non-initiator will require a 1-block delay between RBF attempts. */
   val DelayRbfAttempts = "delay_rbf_attempts"
+  /** If set, the non-initiator will not enforce any restriction between RBF attempts. */
+  val UnlimitedRbfAttempts = "unlimited_rbf_attempts"
   /** If set, channels will adapt their max HTLC amount to the available balance. */
   val AdaptMaxHtlcAmount = "adapt_max_htlc_amount"
   /** If set, closing will use option_simple_close. */
@@ -225,6 +227,8 @@ trait ChannelStateTestsBase extends Assertions with Eventually {
       .modify(_.channelConf.dustLimit).setToIf(tags.contains(ChannelStateTestsTags.HighDustLimitDifferenceBobAlice))(1000 sat)
       .modify(_.channelConf.maxRemoteDustLimit).setToIf(tags.contains(ChannelStateTestsTags.HighDustLimitDifferenceAliceBob))(10000 sat)
       .modify(_.channelConf.maxRemoteDustLimit).setToIf(tags.contains(ChannelStateTestsTags.HighDustLimitDifferenceBobAlice))(10000 sat)
+      .modify(_.channelConf.remoteRbfLimits.maxAttempts).setToIf(tags.contains(ChannelStateTestsTags.UnlimitedRbfAttempts))(100)
+      .modify(_.channelConf.remoteRbfLimits.attemptDeltaBlocks).setToIf(tags.contains(ChannelStateTestsTags.UnlimitedRbfAttempts))(0)
       .modify(_.onChainFeeConf.defaultFeerateTolerance.ratioLow).setToIf(tags.contains(ChannelStateTestsTags.HighFeerateMismatchTolerance))(0.000001)
       .modify(_.onChainFeeConf.defaultFeerateTolerance.ratioHigh).setToIf(tags.contains(ChannelStateTestsTags.HighFeerateMismatchTolerance))(1000000)
       .modify(_.onChainFeeConf.spendAnchorWithoutHtlcs).setToIf(tags.contains(ChannelStateTestsTags.DontSpendAnchorWithoutHtlcs))(false)
@@ -235,7 +239,9 @@ trait ChannelStateTestsBase extends Assertions with Eventually {
       .modify(_.channelConf.maxRemoteDustLimit).setToIf(tags.contains(ChannelStateTestsTags.HighDustLimitDifferenceAliceBob))(10000 sat)
       .modify(_.channelConf.maxRemoteDustLimit).setToIf(tags.contains(ChannelStateTestsTags.HighDustLimitDifferenceBobAlice))(10000 sat)
       .modify(_.channelConf.remoteRbfLimits.maxAttempts).setToIf(tags.contains(ChannelStateTestsTags.RejectRbfAttempts))(0)
+      .modify(_.channelConf.remoteRbfLimits.maxAttempts).setToIf(tags.contains(ChannelStateTestsTags.UnlimitedRbfAttempts))(100)
       .modify(_.channelConf.remoteRbfLimits.attemptDeltaBlocks).setToIf(tags.contains(ChannelStateTestsTags.DelayRbfAttempts))(1)
+      .modify(_.channelConf.remoteRbfLimits.attemptDeltaBlocks).setToIf(tags.contains(ChannelStateTestsTags.UnlimitedRbfAttempts))(0)
       .modify(_.onChainFeeConf.defaultFeerateTolerance.ratioLow).setToIf(tags.contains(ChannelStateTestsTags.HighFeerateMismatchTolerance))(0.000001)
       .modify(_.onChainFeeConf.defaultFeerateTolerance.ratioHigh).setToIf(tags.contains(ChannelStateTestsTags.HighFeerateMismatchTolerance))(1000000)
       .modify(_.onChainFeeConf.spendAnchorWithoutHtlcs).setToIf(tags.contains(ChannelStateTestsTags.DontSpendAnchorWithoutHtlcs))(false)
diff --git a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/c/WaitForDualFundingConfirmedStateSpec.scala b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/c/WaitForDualFundingConfirmedStateSpec.scala
index 6e110fe..c7aa3ba 100644
--- a/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/c/WaitForDualFundingConfirmedStateSpec.scala
+++ b/eclair-core/src/test/scala/fr/acinq/eclair/channel/states/c/WaitForDualFundingConfirmedStateSpec.scala
@@ -338,49 +338,61 @@ class WaitForDualFundingConfirmedStateSpec extends TestKitBaseClass with Fixture
   }
 
   def testBumpFundingFees(f: FixtureParam, feerate_opt: Option[FeeratePerKw] = None, requestFunding_opt: Option[LiquidityAds.RequestFunding] = None): FullySignedSharedTransaction = {
+    testBumpFundingFees(f, f.alice, f.bob, f.alice2bob, f.bob2alice, feerate_opt, requestFunding_opt)
+  }
+
+  def testBumpFundingFees(f: FixtureParam, s: TestFSMRef[ChannelState, ChannelData, Channel], r: TestFSMRef[ChannelState, ChannelData, Channel], s2r: TestProbe, r2s: TestProbe, feerate_opt: Option[FeeratePerKw], requestFunding_opt: Option[LiquidityAds.RequestFunding]): FullySignedSharedTransaction = {
     import f._
 
     val probe = TestProbe()
-    val currentFundingTx = alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].latestFundingTx.sharedTx.asInstanceOf[FullySignedSharedTransaction]
-    val previousFundingTxs = alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs
-    alice ! CMD_BUMP_FUNDING_FEE(probe.ref, feerate_opt.getOrElse(currentFundingTx.feerate * 1.1), fundingFeeBudget = 100_000.sat, 0, requestFunding_opt)
-    assert(alice2bob.expectMsgType[TxInitRbf].fundingContribution == TestConstants.fundingSatoshis)
-    alice2bob.forward(bob)
-    val txAckRbf = bob2alice.expectMsgType[TxAckRbf]
-    assert(txAckRbf.fundingContribution == requestFunding_opt.map(_.requestedAmount).getOrElse(TestConstants.nonInitiatorFundingSatoshis))
+    val currentFundingParams = s.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].latestFundingTx.fundingParams
+    val currentFundingTx = s.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].latestFundingTx.sharedTx.asInstanceOf[FullySignedSharedTransaction]
+    val previousFundingTxs = s.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs
+    s ! CMD_BUMP_FUNDING_FEE(probe.ref, feerate_opt.getOrElse(currentFundingTx.feerate * 1.1), fundingFeeBudget = 100_000.sat, 0, requestFunding_opt)
+    assert(s2r.expectMsgType[TxInitRbf].fundingContribution == currentFundingParams.localContribution)
+    s2r.forward(r)
+    val txAckRbf = r2s.expectMsgType[TxAckRbf]
+    assert(txAckRbf.fundingContribution == requestFunding_opt.map(_.requestedAmount).getOrElse(currentFundingParams.remoteContribution))
     requestFunding_opt.foreach(_ => assert(txAckRbf.willFund_opt.nonEmpty))
-    bob2alice.forward(alice)
+    r2s.forward(s)
 
     // Alice and Bob build a new version of the funding transaction, with one new input every time.
     val inputCount = previousFundingTxs.length + 2
     (1 to inputCount).foreach(_ => {
-      alice2bob.expectMsgType[TxAddInput]
-      alice2bob.forward(bob)
-      bob2alice.expectMsgType[TxAddInput]
-      bob2alice.forward(alice)
+      s2r.expectMsgType[TxAddInput]
+      s2r.forward(r)
+      r2s.expectMsgType[TxAddInput]
+      r2s.forward(s)
     })
-    alice2bob.expectMsgType[TxAddOutput]
-    alice2bob.forward(bob)
-    bob2alice.expectMsgType[TxAddOutput]
-    bob2alice.forward(alice)
-    alice2bob.expectMsgType[TxAddOutput]
-    alice2bob.forward(bob)
-    bob2alice.expectMsgType[TxComplete]
-    bob2alice.forward(alice)
-    alice2bob.expectMsgType[TxComplete]
-    alice2bob.forward(bob)
-    bob2alice.expectMsgType[CommitSig]
-    bob2alice.forward(alice)
-    alice2bob.expectMsgType[CommitSig]
-    alice2bob.forward(bob)
-    bob2alice.expectMsgType[TxSignatures]
-    bob2alice.forward(alice)
-    alice2bob.expectMsgType[TxSignatures]
-    alice2bob.forward(bob)
+    s2r.expectMsgType[TxAddOutput]
+    s2r.forward(r)
+    r2s.expectMsgType[TxAddOutput]
+    r2s.forward(s)
+    s2r.expectMsgType[TxAddOutput]
+    s2r.forward(r)
+    r2s.expectMsgType[TxComplete]
+    r2s.forward(s)
+    s2r.expectMsgType[TxComplete]
+    s2r.forward(r)
+    r2s.expectMsgType[CommitSig]
+    r2s.forward(s)
+    s2r.expectMsgType[CommitSig]
+    s2r.forward(r)
+    if (currentFundingParams.localContribution < currentFundingParams.remoteContribution) {
+      s2r.expectMsgType[TxSignatures]
+      s2r.forward(r)
+      r2s.expectMsgType[TxSignatures]
+      r2s.forward(s)
+    } else {
+      r2s.expectMsgType[TxSignatures]
+      r2s.forward(s)
+      s2r.expectMsgType[TxSignatures]
+      s2r.forward(r)
+    }
 
     probe.expectMsgType[RES_BUMP_FUNDING_FEE]
 
-    val nextFundingTx = alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].latestFundingTx.sharedTx.asInstanceOf[FullySignedSharedTransaction]
+    val nextFundingTx = s.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].latestFundingTx.sharedTx.asInstanceOf[FullySignedSharedTransaction]
     assert(aliceListener.expectMsgType[TransactionPublished].tx.txid == nextFundingTx.signedTx.txid)
     assert(alice2blockchain.expectMsgType[WatchFundingConfirmed].txId == nextFundingTx.signedTx.txid)
     assert(bobListener.expectMsgType[TransactionPublished].tx.txid == nextFundingTx.signedTx.txid)
@@ -389,12 +401,12 @@ class WaitForDualFundingConfirmedStateSpec extends TestKitBaseClass with Fixture
     assert(currentFundingTx.feerate < nextFundingTx.feerate)
     // The new transaction double-spends previous inputs.
     currentFundingTx.signedTx.txIn.map(_.outPoint).foreach(o => assert(nextFundingTx.signedTx.txIn.exists(_.outPoint == o)))
-    assert(alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.length == previousFundingTxs.length + 1)
-    assert(alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.head.sharedTx == currentFundingTx)
+    assert(s.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.length == previousFundingTxs.length + 1)
+    assert(s.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.head.sharedTx == currentFundingTx)
     nextFundingTx
   }
 
-  test("recv CMD_BUMP_FUNDING_FEE", Tag(ChannelStateTestsTags.DualFunding)) { f =>
+  test("recv CMD_BUMP_FUNDING_FEE", Tag(ChannelStateTestsTags.DualFunding), Tag(ChannelStateTestsTags.UnlimitedRbfAttempts)) { f =>
     import f._
 
     // Bob contributed to the funding transaction.
@@ -420,9 +432,21 @@ class WaitForDualFundingConfirmedStateSpec extends TestKitBaseClass with Fixture
     assert(FeeratePerKw(15_000 sat) <= fundingTx3.feerate && fundingTx3.feerate < FeeratePerKw(15_700 sat))
     assert(alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.length == 2)
 
-    // The initial funding transaction confirms
+    // Bob RBFs the funding transaction: Alice keeps contributing the same amount.
+    val feerate4 = FeeratePerKw(20_000 sat)
+    testBumpFundingFees(f, bob, alice, bob2alice, alice2bob, Some(feerate4), requestFunding_opt = None)
+    val balanceBob4 = bob.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].commitments.latest.localCommit.spec.toLocal
+    assert(balanceBob4 == TestConstants.nonInitiatorFundingSatoshis.toMilliSatoshi)
+    val balanceAlice4 = alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].commitments.latest.localCommit.spec.toLocal
+    assert(balanceAlice4 == TestConstants.fundingSatoshis.toMilliSatoshi)
+    val fundingTx4 = alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].commitments.latest.localFundingStatus.asInstanceOf[DualFundedUnconfirmedFundingTx].sharedTx.asInstanceOf[FullySignedSharedTransaction]
+    assert(FeeratePerKw(20_000 sat) <= fundingTx4.feerate && fundingTx4.feerate < FeeratePerKw(20_500 sat))
+    assert(bob.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.length == 3)
+    assert(alice.stateData.asInstanceOf[DATA_WAIT_FOR_DUAL_FUNDING_CONFIRMED].previousFundingTxs.length == 3)
+
+    // The initial funding transaction confirms: we rollback unused inputs.
     alice ! WatchFundingConfirmedTriggered(BlockHeight(42000), 42, fundingTx1.signedTx)
-    testUnusedInputsUnlocked(wallet, Seq(fundingTx2, fundingTx3))
+    testUnusedInputsUnlocked(wallet, Seq(fundingTx2, fundingTx3, fundingTx4))
   }
 
   test("recv CMD_BUMP_FUNDING_FEE (liquidity ads)", Tag(ChannelStateTestsTags.DualFunding), Tag(liquidityPurchase)) { f =>
@@ -478,6 +502,10 @@ class WaitForDualFundingConfirmedStateSpec extends TestKitBaseClass with Fixture
     bob2alice.forward(alice)
     alice2bob.expectMsgType[TxAbort]
     alice2bob.forward(bob)
+
+    // Bob tries to RBF: this is disabled because it would override Alice's liquidity purchase.
+    bob ! CMD_BUMP_FUNDING_FEE(sender.ref, FeeratePerKw(20_000 sat), 100_000 sat, 0, requestFunding_opt = None)
+    assert(sender.expectMsgType[RES_FAILURE[_, ChannelException]].t.isInstanceOf[InvalidRbfOverridesLiquidityPurchase])
   }
 
   test("recv CMD_BUMP_FUNDING_FEE (aborted)", Tag(ChannelStateTestsTags.DualFunding)) { f =>