More robust splicing edge cases (#3366)

### eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelData.scala
@@ -525,6 +525,15 @@ sealed trait SpliceStatus {
     case SpliceStatus.NoSplice | _: SpliceStatus.NegotiatingQuiescence => false
     case _ => true
   }
+  /** Our peer has sent stfu: they must not send any update until the quiescence session ends. */
+  def remoteIsQuiescing: Boolean = this match {
+    case SpliceStatus.NegotiatingQuiescence(_, status) => status match {
+      case _: QuiescenceNegotiation.NonInitiator.ReceivedStfu => true
+      case _: QuiescenceNegotiation.Initiator => false
+    }
+    case SpliceStatus.NoSplice => false
+    case _ => true
+  }
 }
 object SpliceStatus {
   case object NoSplice extends SpliceStatus
@@ -687,6 +696,7 @@ final case class DATA_NORMAL(commitments: Commitments,
   val lastAnnouncedFundingTxId_opt: Option[TxId] = lastAnnouncedCommitment_opt.map(_.fundingTxId)
   val isNegotiatingQuiescence: Boolean = spliceStatus.isNegotiatingQuiescence
   val isQuiescent: Boolean = spliceStatus.isQuiescent
+  val remoteIsQuiescing: Boolean = spliceStatus.remoteIsQuiescing
 }
 final case class DATA_SHUTDOWN(commitments: Commitments, localShutdown: Shutdown, remoteShutdown: Shutdown, closeStatus: CloseStatus) extends ChannelDataWithCommitments
 final case class DATA_NEGOTIATING(commitments: Commitments,

### eclair-core/src/main/scala/fr/acinq/eclair/channel/ChannelExceptions.scala
@@ -117,6 +117,7 @@ case class FeerateTooDifferent                     (override val channelId: Byte
 case class InvalidAnnouncementSignatures           (override val channelId: ByteVector32, annSigs: AnnouncementSignatures) extends ChannelException(channelId, s"invalid announcement signatures: $annSigs")
 case class InvalidCommitmentSignature              (override val channelId: ByteVector32, fundingTxId: TxId, commitmentNumber: Long, unsignedCommitTx: Transaction) extends ChannelException(channelId, s"invalid commitment signature: fundingTxId=$fundingTxId commitmentNumber=$commitmentNumber commitTxId=${unsignedCommitTx.txid} commitTx=$unsignedCommitTx")
 case class InvalidHtlcSignature                    (override val channelId: ByteVector32, txId: TxId) extends ChannelException(channelId, s"invalid htlc signature: txId=$txId")
+case class InvalidCommitmentNumber                 (override val channelId: ByteVector32, fundingTxId: TxId) extends ChannelException(channelId, s"invalid commitment number for pending funding attempt with fundingTxId=$fundingTxId")
 case class CannotGenerateClosingTx                 (override val channelId: ByteVector32) extends ChannelException(channelId, "failed to generate closing transaction: all outputs are trimmed")
 case class MissingCloseSignature                   (override val channelId: ByteVector32) extends ChannelException(channelId, "closing_complete is missing a signature for a closing transaction including our output")
 case class InvalidCloseSignature                   (override val channelId: ByteVector32, txId: TxId) extends ChannelException(channelId, s"invalid close signature: txId=$txId")

### eclair-core/src/main/scala/fr/acinq/eclair/channel/Commitments.scala
@@ -879,7 +879,13 @@ case class Commitments(channelParams: ChannelParams,
   val lastLocalLocked_opt: Option[Commitment] = active.filter(_.localFundingStatus.isInstanceOf[LocalFundingStatus.Locked]).sortBy(_.fundingTxIndex).lastOption
   val lastRemoteLocked_opt: Option[Commitment] = active.filter(c => c.remoteFundingStatus == RemoteFundingStatus.Locked).sortBy(_.fundingTxIndex).lastOption
 
-  def add(commitment: Commitment): Commitments = copy(active = commitment +: active)
+  def add(commitment: Commitment): Commitments = {
+    // As part of a defense-in-depth strategy, we verify that the commitment indices always match (even though the
+    // caller should always enforce that, it's good to make it explicit here).
+    require(commitment.localCommit.index == localCommitIndex, s"cannot add commitment at localCommitIndex=${commitment.localCommit.index}, we're at localCommitIndex=$localCommitIndex")
+    require(commitment.remoteCommit.index == remoteCommitIndex, s"cannot add commitment at remoteCommitIndex=${commitment.remoteCommit.index}, we're at remoteCommitIndex=$remoteCommitIndex")
+    copy(active = commitment +: active)
+  }
 
   // @formatter:off
   def localIsQuiescent: Boolean = changes.localChanges.all.isEmpty

### eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/Channel.scala
@@ -538,6 +538,26 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
       context.system.scheduler.scheduleOnce(1 second, peer, Peer.Disconnect(remoteNodeId))
       stay() sending Warning(d.channelId, error.getMessage)
 
+    case Event(msg: ForbiddenMessageWhenQuiescent, d: DATA_NORMAL) if d.remoteIsQuiescing =>
+      // Our peer has sent stfu: they must not send any update until the quiescence session ends, even though we may
+      // still be signing our own pending updates. If we applied their update, our local and remote commitments would
+      // stop being mirror images of each other, and the splice commitment that we're about to create from them would
+      // have inconsistent balances.
+      log.warning("received forbidden message {} after our peer sent stfu", msg.getClass.getSimpleName)
+      val error = ForbiddenDuringQuiescence(d.channelId, msg.getClass.getSimpleName)
+      // We forward preimages as soon as possible to the upstream channel because it allows us to pull funds.
+      msg match {
+        case fulfill: UpdateFulfillHtlc => d.commitments.receiveFulfill(fulfill) match {
+          case Right((_, origin, htlc)) => relayer ! RES_ADD_SETTLED(origin, remoteNodeId, htlc, HtlcResult.RemoteFulfill(fulfill))
+          case _ => ()
+        }
+        case _ => ()
+      }
+      // Instead of force-closing (which would cost us on-chain fees), we disconnect: this cancels the quiescence
+      // negotiation and restores the channel to a clean state.
+      context.system.scheduler.scheduleOnce(1 second, peer, Peer.Disconnect(remoteNodeId))
+      stay() sending Warning(d.channelId, error.getMessage)
+
     case Event(c: CMD_ADD_HTLC, d: DATA_NORMAL) if d.localShutdown.isDefined || d.remoteShutdown.isDefined =>
       // note: spec would allow us to keep sending new htlcs after having received their shutdown (and not sent ours)
       // but we want to converge as fast as possible and they would probably not route them anyway
@@ -675,40 +695,68 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
       }
 
     case Event(commit: CommitSigs, d: DATA_NORMAL) =>
-      (d.spliceStatus, commit) match {
-        case (s: SpliceStatus.SpliceInProgress, sig: CommitSig) =>
-          log.debug("received their commit_sig, deferring message")
-          stay() using d.copy(spliceStatus = s.copy(remoteCommitSig = Some(sig)))
-        case (SpliceStatus.SpliceAborted, _: CommitSig) =>
+      d.spliceStatus match {
+        case s: SpliceStatus.SpliceInProgress =>
+          commit match {
+            case sig: CommitSig =>
+              log.debug("received their commit_sig, deferring message")
+              stay() using d.copy(spliceStatus = s.copy(remoteCommitSig = Some(sig)))
+            case _: CommitSigBatch =>
+              log.warning("ignoring received commit_sig batch while splice is in progress")
+              // This should never happen if our peer follows the spec: we just ignore the batch, which may or may not
+              // lead to a force-close depending on how our peer behaves.
+              stay()
+          }
+        case SpliceStatus.SpliceAborted =>
           log.warning("received commit_sig after sending tx_abort, they probably sent it before receiving our tx_abort, ignoring...")
           stay()
-        case (SpliceStatus.SpliceWaitingForSigs(signingSession), sig: CommitSig) =>
-          signingSession.receiveCommitSig(d.commitments.channelParams, channelKeys, sig, nodeParams.currentBlockHeight) match {
-            case Left(f) =>
+        case SpliceStatus.SpliceWaitingForSigs(signingSession) =>
+          commit match {
+            case batch: CommitSigBatch =>
+              log.warning("we're expecting a single commit_sig for splice with txId={} but received a batch of commit_sig for txIds={}", signingSession.fundingTxId, batch.messages.flatMap(_.fundingTxId_opt).mkString(","))
               rollbackFundingAttempt(signingSession.fundingTx.tx, Nil)
-              stay() using d.copy(spliceStatus = SpliceStatus.SpliceAborted) sending TxAbort(d.channelId, f.getMessage)
-            case Right(signingSession1) => signingSession1 match {
-              case signingSession1: InteractiveTxSigningSession.WaitingForSigs =>
-                // In theory we don't have to store their commit_sig here, as they would re-send it if we disconnect, but
-                // it is more consistent with the case where we send our tx_signatures first.
-                val d1 = d.copy(spliceStatus = SpliceStatus.SpliceWaitingForSigs(signingSession1))
-                stay() using d1 storing()
-              case signingSession1: InteractiveTxSigningSession.SendingSigs =>
-                // We don't have their tx_sigs, but they have ours, and could publish the funding tx without telling us.
-                // That's why we move on immediately to the next step, and will update our unsigned funding tx when we
-                // receive their tx_sigs.
-                val minDepth_opt = d.commitments.channelParams.minDepth(nodeParams.channelConf.minDepth)
-                watchFundingConfirmed(signingSession.fundingTx.txId, minDepth_opt, delay_opt = None)
-                val commitments1 = d.commitments.add(signingSession1.commitment)
-                context.system.eventStream.publish(ChannelFundingCreated(self, d.channelId, remoteNodeId, Right(signingSession1.fundingTx.signedTx_opt.getOrElse(signingSession1.fundingTx.sharedTx.tx.buildUnsignedTx())), signingSession1.commitment.fundingTxIndex, commitments1))
-                val d1 = d.copy(commitments = commitments1, spliceStatus = SpliceStatus.NoSplice)
-                stay() using d1 storing() sending signingSession1.localSigs calling endQuiescence(d1)
-            }
+              reportLiquidityPurchaseAborted(d.channelId, signingSession)
+              handleLocalError(CommitSigCountMismatch(d.channelId, 1, batch.batchSize), d, Some(commit))
+            case _: CommitSig if !signingSession.isConsistentWith(d.commitments) =>
+              log.warning("aborting splice: commitment number mismatch")
+              rollbackFundingAttempt(signingSession.fundingTx.tx, Nil)
+              reportLiquidityPurchaseAborted(d.channelId, signingSession)
+              handleLocalError(InvalidCommitmentNumber(d.channelId, signingSession.fundingTxId), d, Some(commit))
+            case sig: CommitSig =>
+              signingSession.receiveCommitSig(d.commitments.channelParams, channelKeys, sig, nodeParams.currentBlockHeight) match {
+                case Left(f) =>
+                  rollbackFundingAttempt(signingSession.fundingTx.tx, Nil)
+                  stay() using d.copy(spliceStatus = SpliceStatus.SpliceAborted) sending TxAbort(d.channelId, f.getMessage)
+                case Right(signingSession1) => signingSession1 match {
+                  case signingSession1: InteractiveTxSigningSession.WaitingForSigs =>
+                    // In theory we don't have to store their commit_sig here, as they would re-send it if we disconnect, but
+                    // it is more consistent with the case where we send our tx_signatures first.
+                    val d1 = d.copy(spliceStatus = SpliceStatus.SpliceWaitingForSigs(signingSession1))
+                    stay() using d1 storing()
+                  case signingSession1: InteractiveTxSigningSession.SendingSigs =>
+                    // We don't have their tx_sigs, but they have ours, and could publish the funding tx without telling us.
+                    // That's why we move on immediately to the next step, and will update our unsigned funding tx when we
+                    // receive their tx_sigs.
+                    val minDepth_opt = d.commitments.channelParams.minDepth(nodeParams.channelConf.minDepth)
+                    watchFundingConfirmed(signingSession.fundingTx.txId, minDepth_opt, delay_opt = None)
+                    val commitments1 = d.commitments.add(signingSession1.commitment)
+                    context.system.eventStream.publish(ChannelFundingCreated(self, d.channelId, remoteNodeId, Right(signingSession1.fundingTx.signedTx_opt.getOrElse(signingSession1.fundingTx.sharedTx.tx.buildUnsignedTx())), signingSession1.commitment.fundingTxIndex, commitments1))
+                    val d1 = d.copy(commitments = commitments1, spliceStatus = SpliceStatus.NoSplice)
+                    stay() using d1 storing() sending signingSession1.localSigs calling endQuiescence(d1)
+                }
+              }
           }
+        case spliceStatus if spliceStatus.isQuiescent =>
+          // The channel is quiescent and we're negotiating a splice or an rbf attempt: our peer must not send
+          // commit_sig. We don't have a signing session yet, so applying it wouldn't let them steal our funds, but it
+          // would move our commitment index right before we create a commitment at the current commitment index.
+          // Instead of force-closing (which would cost us on-chain fees), we disconnect: this aborts the splice
+          // attempt, restores the channel to a clean state, and gives our peer the opportunity to fix their node.
+          log.warning("received commit_sig while quiescent with status {}, disconnecting", spliceStatus.getClass.getSimpleName)
+          context.system.scheduler.scheduleOnce(1 second, peer, Peer.Disconnect(remoteNodeId))
+          stay() sending Warning(d.channelId, ForbiddenDuringSplice(d.channelId, commit.getClass.getSimpleName).getMessage)
         case _ =>
-          // NB: in all other cases we process the commit_sigs normally. We could do a full pattern matching on all
-          // splice statuses, but it would force us to handle every corner case where our peer doesn't behave correctly
-          // whereas they will all simply lead to a force-close.
+          // NB: in all other cases we process the commit_sigs normally, since we're not in the middle of splicing.
           d.commitments.receiveCommit(commit, channelKeys) match {
             case Right((commitments1, revocation)) =>
               log.debug("received a new sig, spec:\n{}", commitments1.latest.specs2String)
@@ -727,7 +775,9 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
                   val stfu = Stfu(d.channelId, initiator = true)
                   val spliceStatus1 = SpliceStatus.NegotiatingQuiescence(cmd_opt, QuiescenceNegotiation.Initiator.SentStfu(stfu))
                   (d.copy(commitments = commitments1, spliceStatus = spliceStatus1), Seq(revocation, stfu))
-                case SpliceStatus.NegotiatingQuiescence(_, _: QuiescenceNegotiation.NonInitiator.ReceivedStfu) if commitments1.localIsQuiescent =>
+                // Note that we check that *both* sides are quiescent: our peer must not have sent updates after their
+                // stfu, otherwise the commitment we create during the splice would have inconsistent balances.
+                case SpliceStatus.NegotiatingQuiescence(_, _: QuiescenceNegotiation.NonInitiator.ReceivedStfu) if commitments1.isQuiescent =>
                   val stfu = Stfu(d.channelId, initiator = false)
                   (d.copy(commitments = commitments1, spliceStatus = SpliceStatus.NonInitiatorQuiescent), Seq(revocation, stfu))
                 case _ =>
@@ -1248,6 +1298,9 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
 
     case Event(msg: TxInitRbf, d: DATA_NORMAL) =>
       d.spliceStatus match {
+        case SpliceStatus.NonInitiatorQuiescent if !d.commitments.isQuiescent =>
+          log.info("rejecting rbf request: channel not quiescent")
+          stay() using d.copy(spliceStatus = SpliceStatus.SpliceAborted) sending TxAbort(d.channelId, InvalidSpliceNotQuiescent(d.channelId).getMessage)
         case SpliceStatus.NonInitiatorQuiescent =>
           getSpliceRbfContext(None, d) match {
             case Right(rbf) if msg.feerate < rbf.latestFundingTx.fundingParams.minNextFeerate =>
@@ -1475,6 +1528,11 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
           }
         case _ =>
           d.spliceStatus match {
+            case SpliceStatus.SpliceWaitingForSigs(signingSession) if !signingSession.isConsistentWith(d.commitments) =>
+              log.warning("aborting splice RBF: commitment number mismatch")
+              rollbackFundingAttempt(signingSession.fundingTx.tx, previousTxs = Seq.empty) // no splice rbf yet
+              reportLiquidityPurchaseAborted(d.channelId, signingSession)
+              handleLocalError(InvalidCommitmentNumber(d.channelId, signingSession.fundingTxId), d, Some(msg))
             case SpliceStatus.SpliceWaitingForSigs(signingSession) =>
               // we have not yet sent our tx_signatures
               signingSession.receiveTxSigs(channelKeys, msg, nodeParams.currentBlockHeight) match {
@@ -2660,6 +2718,14 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
           channelReestablish.nextFundingTxId_opt match {
             case Some(fundingTxId) =>
               d.status match {
+                case DualFundingStatus.RbfWaitingForSigs(signingSession) if signingSession.fundingTx.txId == fundingTxId && !signingSession.isConsistentWith(d.commitments) =>
+                  // This should never happen, but if our commitments somehow moved to the next commitment numbers while
+                  // this RBF attempt was being signed, we must not complete it: it would create a commitment at
+                  // commitment numbers that we have already revoked.
+                  log.warning("aborting rbf attempt: commitment number mismatch")
+                  rollbackRbfAttempt(signingSession, d)
+                  reportLiquidityPurchaseAborted(d.channelId, signingSession)
+                  goto(WAIT_FOR_DUAL_FUNDING_CONFIRMED) using d.copy(status = DualFundingStatus.RbfAborted) sending TxAbort(d.channelId, InvalidCommitmentNumber(d.channelId, signingSession.fundingTxId).getMessage)
                 case DualFundingStatus.RbfWaitingForSigs(signingSession) if signingSession.fundingTx.txId == fundingTxId =>
                   if (retransmitCommitSig) {
                     // They haven't received our commit_sig: we retransmit it.
@@ -3541,6 +3607,15 @@ class Channel(val nodeParams: NodeParams, val channelKeys: ChannelKeys, val wall
     val spliceStatus1 = channelReestablish.nextFundingTxId_opt match {
       case Some(fundingTxId) =>
         d.spliceStatus match {
+          case SpliceStatus.SpliceWaitingForSigs(signingSession) if signingSession.fundingTx.txId == fundingTxId && !signingSession.isConsistentWith(d.commitments) =>
+            // This should never happen, but if our commitments somehow moved to the next commitment numbers while this
+            // splice was being signed, we must not complete it: it would create a commitment at commitment numbers that
+            // we have already revoked. We abort the splice attempt instead.
+            log.warning("aborting splice attempt: commitment number mismatch")
+            rollbackFundingAttempt(signingSession.fundingTx.tx, previousTxs = Seq.empty) // no splice rbf yet
+            reportLiquidityPurchaseAborted(d.channelId, signingSession)
+            sendQueue = sendQueue :+ TxAbort(d.channelId, InvalidCommitmentNumber(d.channelId, signingSession.fundingTxId).getMessage)
+            SpliceStatus.SpliceAborted
           case SpliceStatus.SpliceWaitingForSigs(signingSession) if signingSession.fundingTx.txId == fundingTxId =>
             if (retransmitCommitSig) {
               // They haven't received our commit_sig: we retransmit it.

### eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/ChannelOpenDualFunded.scala
@@ -495,6 +495,11 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
           }
         case _: FullySignedSharedTransaction =>
           d.status match {
+            case DualFundingStatus.RbfWaitingForSigs(signingSession) if !signingSession.isConsistentWith(d.commitments) =>
+              log.warning("aborting RBF attempt: commitment number mismatch")
+              rollbackRbfAttempt(signingSession, d)
+              reportLiquidityPurchaseAborted(d.channelId, signingSession)
+              handleLocalError(InvalidCommitmentNumber(d.channelId, signingSession.fundingTxId), d, Some(txSigs))
             case DualFundingStatus.RbfWaitingForSigs(signingSession) =>
               signingSession.receiveTxSigs(channelKeys, txSigs, nodeParams.currentBlockHeight) match {
                 case Left(f) =>
@@ -680,6 +685,11 @@ trait ChannelOpenDualFunded extends DualFundingHandlers with ErrorHandlers {
         case s: DualFundingStatus.RbfInProgress =>
           log.debug("received their commit_sig, deferring message")
           stay() using d.copy(status = s.copy(remoteCommitSig = Some(commitSig)))
+        case DualFundingStatus.RbfWaitingForSigs(signingSession) if !signingSession.isConsistentWith(d.commitments) =>
+          log.warning("aborting RBF attempt: commitment number mismatch")
+          rollbackRbfAttempt(signingSession, d)
+          reportLiquidityPurchaseAborted(d.channelId, signingSession)
+          handleLocalError(InvalidCommitmentNumber(d.channelId, signingSession.fundingTxId), d, Some(commitSig))
         case DualFundingStatus.RbfWaitingForSigs(signingSession) =>
           signingSession.receiveCommitSig(d.commitments.channelParams, channelKeys, commitSig, nodeParams.currentBlockHeight) match {
             case Left(f) =>

### eclair-core/src/main/scala/fr/acinq/eclair/channel/fsm/DualFundingHandlers.scala
@@ -16,7 +16,7 @@
 
 package fr.acinq.eclair.channel.fsm
 
-import fr.acinq.bitcoin.scalacompat.{Transaction, TxIn}
+import fr.acinq.bitcoin.scalacompat.{ByteVector32, Transaction, TxIn}
 import fr.acinq.eclair.NotificationsLogger
 import fr.acinq.eclair.NotificationsLogger.NotifyNodeOperator
 import fr.acinq.eclair.blockchain.{CurrentBlockHeight, NewTransaction}
@@ -26,6 +26,7 @@ import fr.acinq.eclair.channel._
 import fr.acinq.eclair.channel.fsm.Channel.BITCOIN_FUNDING_DOUBLE_SPENT
 import fr.acinq.eclair.channel.fund.InteractiveTxBuilder._
 import fr.acinq.eclair.channel.fund.{InteractiveTxBuilder, InteractiveTxSigningSession}
+import fr.acinq.eclair.io.Peer.LiquidityPurchaseAborted
 import fr.acinq.eclair.wire.protocol.{ChannelReady, Error}
 
 import scala.concurrent.Future
@@ -144,6 +145,17 @@ trait DualFundingHandlers extends CommonFundingHandlers {
     rollbackFundingAttempt(signingSession.fundingTx.tx, d.allFundingTxs.map(_.sharedTx))
   }
 
+  /**
+   * When we abort a funding attempt for which we were selling liquidity, we must tell our peer actor: we have already
+   * told it that the purchase was signed, so it is waiting for a funding transaction that will never exist. This lets
+   * it fail the corresponding upstream HTLCs instead of holding them until they expire.
+   */
+  def reportLiquidityPurchaseAborted(channelId: ByteVector32, signingSession: InteractiveTxSigningSession.WaitingForSigs): Unit = {
+    signingSession.liquidityPurchase_opt.collect {
+      case _ if !signingSession.fundingParams.isInitiator => peer ! LiquidityPurchaseAborted(channelId, signingSession.fundingTx.txId, signingSession.fundingTxIndex)
+    }
+  }
+
   def reportRbfFailure(fundingStatus: DualFundingStatus, f: Throwable): Unit = {
     fundingStatus match {
       case DualFundingStatus.RbfRequested(cmd) => cmd.replyTo ! RES_FAILURE(cmd, f)

### eclair-core/src/main/scala/fr/acinq/eclair/channel/fund/InteractiveTxBuilder.scala
@@ -1218,6 +1218,11 @@ object InteractiveTxSigningSession {
     // If we haven't received the remote commit_sig, we will request a retransmission on reconnection.
     val retransmitRemoteCommitSig: Boolean = localCommit.isLeft
 
+    /** Verify that the pending commitment number matches the other commitments before accepting commit_sig. */
+    def isConsistentWith(commitments: Commitments): Boolean = {
+      localCommitIndex == commitments.localCommitIndex && remoteCommit.index == commitments.remoteCommitIndex
+    }
+
     // For the legacy splice protocol, we use the next_commitment_number to let our peer know whether they needed to
     // retransmit commit_sig or not. We're now using an explicit bit instead, but need to maintain backwards-compatibility.
     def nextLocalCommitmentNumber(useLegacySpliceProtocol: Boolean): Long = localCommit match {

### eclair-core/src/main/scala/fr/acinq/eclair/io/PeerConnection.scala
@@ -424,7 +424,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
                   log.warning("starting batch with incomplete previous batch ({}/{} received)", pending.received.size, pending.batchSize)
                   // This is a spec violation from our peer: this will likely lead to a force-close.
                   d.transport ! Warning(msg.channelId, "invalid start_batch message: the previous batch is not done yet")
-                  d.peer ! CommitSigBatch(pending.received)
+                  d.peer ! CommitSigs(pending.received)
                 case _ => ()
               }
               stay() using d.copy(commitSigBatch_opt = Some(PendingCommitSigBatch(msg.channelId, msg.batchSize, Nil)))
@@ -445,7 +445,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
                 }
               case _ =>
                 log.warning("received {} as part of a batch: we don't support batching that kind of messages", msg.getClass.getSimpleName)
-                if (pending.received.nonEmpty) d.peer ! CommitSigBatch(pending.received)
+                if (pending.received.nonEmpty) d.peer ! CommitSigs(pending.received)
                 d.peer ! msg
                 stay() using d.copy(commitSigBatch_opt = None)
             }
@@ -463,7 +463,7 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
                   case Some(pending) if pending.channelId != msg.channelId || pending.batchSize != batchSize =>
                     log.warning("received invalid commit_sig batch while a different batch isn't complete")
                     // This should never happen, otherwise it will likely lead to a force-close.
-                    d.peer ! CommitSigBatch(pending.received)
+                    d.peer ! CommitSigs(pending.received)
                     stay() using d.copy(legacyCommitSigBatch_opt = Some(PendingCommitSigBatch(msg.channelId, batchSize, Seq(msg))))
                   case Some(pending) =>
                     val received1 = pending.received :+ msg

### eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalQuiescentStateSpec.scala
@@ -28,6 +28,7 @@ import fr.acinq.eclair.channel.fsm.Channel
 import fr.acinq.eclair.channel.fund.InteractiveTxBuilder
 import fr.acinq.eclair.channel.states.ChannelStateTestsBase.PimpTestFSM
 import fr.acinq.eclair.channel.states.{ChannelStateTestsBase, ChannelStateTestsTags}
+import fr.acinq.eclair.crypto.keymanager.RemoteCommitmentKeys
 import fr.acinq.eclair.io.Peer
 import fr.acinq.eclair.payment.relay.Relayer.RelayForward
 import fr.acinq.eclair.reputation.Reputation
@@ -360,6 +361,57 @@ class NormalQuiescentStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteL
     bob2alice.expectMsg(Warning(channelId(alice), ForbiddenDuringSplice(channelId(alice), "UpdateAddHtlc").getMessage))
   }
 
+  test("recv (forbidden) UpdateAddHtlc message after our peer sent stfu") { f =>
+    import f._
+    initiateQuiescence(f, sendInitialStfu = false)
+    // Bob has a pending local change, so he doesn't send his stfu yet: he is still allowed to receive updates at that
+    // point, but Alice isn't allowed to send any since she already sent her stfu.
+    addHtlc(50_000_000 msat, bob, alice, bob2alice, alice2bob)
+    alice2bob.forward(bob)
+    bob2alice.expectNoMessage(100 millis)
+    val bobCommitments = bob.stateData.asInstanceOf[DATA_NORMAL].commitments
+    assert(bob.stateData.asInstanceOf[DATA_NORMAL].remoteIsQuiescing)
+    // If Bob applied that update, his local and remote commitments would stop being mirror images of each other, and
+    // the splice commitment created from them would have inconsistent balances.
+    val forbiddenMsg = UpdateAddHtlc(channelId = channelId(bob), id = 5656, amountMsat = 50000000 msat, cltvExpiry = CltvExpiryDelta(144).toCltvExpiry(currentBlockHeight), paymentHash = randomBytes32(), onionRoutingPacket = TestConstants.emptyOnionPacket, pathKey_opt = None, accountable = false, fundingFee_opt = None)
+    alice2bob.forward(bob, forbiddenMsg)
+    bob2alice.expectMsg(Warning(channelId(bob), ForbiddenDuringQuiescence(channelId(bob), "UpdateAddHtlc").getMessage))
+    assert(bob.stateData.asInstanceOf[DATA_NORMAL].commitments.changes.remoteChanges == bobCommitments.changes.remoteChanges)
+  }
+
+  test("recv (forbidden) commit_sig while quiescent") { f =>
+    import f._
+    initiateQuiescence(f, sendInitialStfu = true)
+    // Alice is waiting for Bob's splice_ack while Bob is waiting for Alice's splice_init: neither of them must apply a
+    // commit_sig, which would move their commitment index while the splice commitment is about to be created at the
+    // current commitment index.
+    val aliceCommitments = alice.stateData.asInstanceOf[DATA_NORMAL].commitments
+    val bobCommitments = bob.stateData.asInstanceOf[DATA_NORMAL].commitments
+    assert(alice.stateData.asInstanceOf[DATA_NORMAL].spliceStatus.isInstanceOf[SpliceStatus.SpliceRequested])
+    assert(bob.stateData.asInstanceOf[DATA_NORMAL].spliceStatus == SpliceStatus.NonInitiatorQuiescent)
+
+    val commitSigAlice = {
+      val channelKeys = alice.underlyingActor.channelKeys
+      val nextPerCommitmentPoint = aliceCommitments.remoteNextCommitInfo.toOption.get
+      val remoteKeys = RemoteCommitmentKeys(aliceCommitments.channelParams, channelKeys, nextPerCommitmentPoint)
+      aliceCommitments.active.head.sendCommit(aliceCommitments.channelParams, channelKeys, remoteKeys, aliceCommitments.changes, nextPerCommitmentPoint, batchSize = 1, nextRemoteNonce_opt = None).toOption.get._2
+    }
+    val commitSigBob = {
+      val channelKeys = bob.underlyingActor.channelKeys
+      val nextPerCommitmentPoint = bobCommitments.remoteNextCommitInfo.toOption.get
+      val remoteKeys = RemoteCommitmentKeys(bobCommitments.channelParams, channelKeys, nextPerCommitmentPoint)
+      bobCommitments.active.head.sendCommit(bobCommitments.channelParams, channelKeys, remoteKeys, bobCommitments.changes, nextPerCommitmentPoint, batchSize = 1, nextRemoteNonce_opt = None).toOption.get._2
+    }
+
+    // Both parties respond with a warning (and disconnect) instead of revoking their current commitment.
+    alice2bob.forward(bob, commitSigAlice)
+    bob2alice.expectMsg(Warning(channelId(bob), ForbiddenDuringSplice(channelId(bob), "CommitSig").getMessage))
+    assert(bob.stateData.asInstanceOf[DATA_NORMAL].commitments.localCommitIndex == bobCommitments.localCommitIndex)
+    bob2alice.forward(alice, commitSigBob)
+    alice2bob.expectMsg(Warning(channelId(alice), ForbiddenDuringSplice(channelId(alice), "CommitSig").getMessage))
+    assert(alice.stateData.asInstanceOf[DATA_NORMAL].commitments.localCommitIndex == aliceCommitments.localCommitIndex)
+  }
+
   test("recv stfu from splice initiator that is not quiescent") { f =>
     import f._
     addHtlc(50_000_000 msat, alice, bob, alice2bob, bob2alice)

### eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala
@@ -31,9 +31,11 @@ import fr.acinq.eclair.channel.LocalFundingStatus.DualFundedUnconfirmedFundingTx
 import fr.acinq.eclair.channel._
 import fr.acinq.eclair.channel.fsm.Channel
 import fr.acinq.eclair.channel.fund.InteractiveTxBuilder.FullySignedSharedTransaction
+import fr.acinq.eclair.channel.fund.InteractiveTxSigningSession
 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.crypto.keymanager.RemoteCommitmentKeys
 import fr.acinq.eclair.db.RevokedHtlcInfoCleaner.ForgetHtlcInfos
 import fr.acinq.eclair.io.Peer.{LiquidityPurchaseAborted, LiquidityPurchaseSigned}
 import fr.acinq.eclair.payment.relay.Relayer
@@ -1967,6 +1969,204 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     (channelReestablishAlice, channelReestablishBob)
   }
 
+  /** Simulate a splice signing session that isn't consistent with our commitments anymore. */
+  private def desynchronizeSpliceSigningSession(alice: TestFSMRef[ChannelState, ChannelData, Channel]): InteractiveTxSigningSession.WaitingForSigs = {
+    val d = alice.stateData.asInstanceOf[DATA_NORMAL]
+    val signingSession = d.spliceStatus.asInstanceOf[SpliceStatus.SpliceWaitingForSigs].signingSession
+    assert(signingSession.isConsistentWith(d.commitments))
+    // We simulate a state where our commitments moved to the next commitment numbers while this splice was being
+    // signed: the splice commitment would then use commitment numbers that we have already revoked.
+    val desynchronized = signingSession.copy(remoteCommit = signingSession.remoteCommit.copy(index = signingSession.remoteCommit.index + 1))
+    assert(!desynchronized.isConsistentWith(d.commitments))
+    alice.setState(NORMAL, d.copy(spliceStatus = SpliceStatus.SpliceWaitingForSigs(desynchronized)))
+    desynchronized
+  }
+
+  test("recv CommitSigBatch after receiving the splice commit_sig") { f =>
+    import f._
+
+    initiateSpliceWithoutSigs(f, spliceIn_opt = Some(SpliceIn(500_000 sat)))
+    alice2bob.expectMsgType[CommitSig]
+    alice2bob.forward(bob)
+    // Alice receives Bob's commit_sig for the splice and is waiting for his tx_signatures.
+    val spliceSigBob = bob2alice.expectMsgType[CommitSig]
+    bob2alice.forward(alice)
+    bob2alice.expectMsgType[TxSignatures] // we don't forward it yet
+
+    val aliceCommitments = alice.stateData.asInstanceOf[DATA_NORMAL].commitments
+    assert(aliceCommitments.active.size == 1)
+    val parentCommitment = aliceCommitments.active.head
+    val commitIndex = parentCommitment.localCommit.index
+    inside(alice.stateData.asInstanceOf[DATA_NORMAL].spliceStatus) {
+      case SpliceStatus.SpliceWaitingForSigs(signingSession) =>
+        // We have already applied their commit_sig for the splice, which is created at the parent's commitment index.
+        assert(signingSession.localCommit.isRight)
+        assert(signingSession.localCommitIndex == commitIndex)
+    }
+
+    // Bob then sends a batch that includes the *next* commit_sig for the parent commitment. If Alice applied it, she
+    // would reveal the per-commitment secret for the index at which the splice commitment is being created, and Bob
+    // could claim the whole channel once the splice completes.
+    val bobKeys = bob.underlyingActor.channelKeys
+    val bobCommitments = bob.stateData.asInstanceOf[DATA_NORMAL].commitments
+    val aliceNextPerCommitmentPoint = alice.underlyingActor.channelKeys.commitmentPoint(commitIndex + 1)
+    val remoteKeys = RemoteCommitmentKeys(bobCommitments.channelParams, bobKeys, aliceNextPerCommitmentPoint)
+    val bobParentCommitment = bobCommitments.active.find(_.fundingTxId == parentCommitment.fundingTxId).get
+    val Right((_, parentSigBob)) = bobParentCommitment.sendCommit(bobCommitments.channelParams, bobKeys, remoteKeys, bobCommitments.changes, aliceNextPerCommitmentPoint, batchSize = 2, nextRemoteNonce_opt = None)
+    alice ! CommitSigBatch(Seq(parentSigBob, spliceSigBob))
+    // Alice force-closes instead of revoking that commitment index.
+    assert(alice2bob.expectMsgType[Error].toAscii == CommitSigCountMismatch(channelId(alice), 1, 2).getMessage)
+    alice2blockchain.expectFinalTxPublished(parentCommitment.localCommit.txId)
+  }
+
+  test("recv CommitSigBatch containing a single commit_sig while signing a splice") { f =>
+    import f._
+
+    initiateSpliceWithoutSigs(f, spliceIn_opt = Some(SpliceIn(500_000 sat)))
+    alice2bob.expectMsgType[CommitSig]
+    alice2bob.forward(bob)
+    val spliceSigBob = bob2alice.expectMsgType[CommitSig]
+    val spliceTxSigsBob = bob2alice.expectMsgType[TxSignatures]
+
+    // Our peer connection flushes incomplete batches of commit_sig: when such a batch contains a single message, it
+    // must be delivered as an individual commit_sig, otherwise we would needlessly force-close the channel.
+    assert(CommitSigs(Seq(spliceSigBob)) == spliceSigBob)
+    alice ! CommitSigs(Seq(spliceSigBob))
+    alice ! spliceTxSigsBob
+    alice2bob.expectMsgType[TxSignatures]
+    awaitCond(alice.stateData.asInstanceOf[DATA_NORMAL].spliceStatus == SpliceStatus.NoSplice)
+    assert(alice.stateData.asInstanceOf[DATA_NORMAL].commitments.active.size == 2)
+  }
+
+  test("recv TxSignatures with an inconsistent splice signing session") { f =>
+    import f._
+
+    initiateSpliceWithoutSigs(f, spliceIn_opt = Some(SpliceIn(500_000 sat)))
+    alice2bob.expectMsgType[CommitSig]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[CommitSig]
+    bob2alice.forward(alice)
+    val spliceTxSigsBob = bob2alice.expectMsgType[TxSignatures]
+
+    val aliceCommitments = alice.stateData.asInstanceOf[DATA_NORMAL].commitments
+    val signingSession = desynchronizeSpliceSigningSession(alice)
+    // We must not add that commitment to our commitments, even though we have their tx_signatures.
+    alice ! spliceTxSigsBob
+    val expected = InvalidCommitmentNumber(channelId(alice), signingSession.fundingTxId)
+    assert(alice2bob.expectMsgType[Error].toAscii == expected.getMessage)
+    alice2blockchain.expectFinalTxPublished(aliceCommitments.latest.localCommit.txId)
+  }
+
+  test("reconnect with an inconsistent splice signing session") { f =>
+    import f._
+
+    initiateSpliceWithoutSigs(f, spliceIn_opt = Some(SpliceIn(500_000 sat)))
+    alice2bob.expectMsgType[CommitSig]
+    alice2bob.forward(bob)
+    bob2alice.expectMsgType[CommitSig]
+    bob2alice.forward(alice)
+    bob2alice.expectMsgType[TxSignatures]
+
+    val aliceCommitments = alice.stateData.asInstanceOf[DATA_NORMAL].commitments
+    val signingSession = desynchronizeSpliceSigningSession(alice)
+    disconnect(f)
+    reconnect(f)
+
+    // We abort the splice attempt instead of resuming it at commitment numbers that we may have revoked.
+    val expected = InvalidCommitmentNumber(channelId(alice), signingSession.fundingTxId)
+    // Note that we also retransmit channel_ready, so we look for our tx_abort in the messages we send.
+    val txAbort = alice2bob.fishForSpecificMessage() { case msg: TxAbort => msg }
+    assert(txAbort.toAscii == expected.getMessage)
+    awaitCond(alice.stateData.asInstanceOf[DATA_NORMAL].spliceStatus == SpliceStatus.SpliceAborted)
+    assert(alice.stateData.asInstanceOf[DATA_NORMAL].commitments.active.size == 1)
+  }
+
+  test("reconnect with an inconsistent splice signing session (liquidity purchase)") { f =>
+    import f._
+
+    // Alice buys liquidity from Bob while splicing in.
+    val sender = TestProbe()
+    val fundingRequest = LiquidityAds.RequestFunding(400_000 sat, TestConstants.defaultLiquidityRates.fundingRates.head, LiquidityAds.PaymentDetails.FromChannelBalance)
+    alice ! CMD_SPLICE(sender.ref, Some(SpliceIn(500_000 sat)), None, Some(fundingRequest), None)
+    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 that liquidity purchase: his peer actor is now waiting for the funding transaction.
+    val purchase = bobPeer.fishForSpecificMessage() { case l: LiquidityPurchaseSigned => l }
+    assert(purchase.fundingTxIndex == 1)
+    alice2bob.expectMsgType[CommitSig] // we don't forward it
+    bob2alice.expectMsgType[CommitSig] // we don't forward it
+    awaitCond(bob.stateData.asInstanceOf[DATA_NORMAL].spliceStatus.isInstanceOf[SpliceStatus.SpliceWaitingForSigs])
+
+    // Bob's signing session isn't consistent with his commitments anymore: he aborts the splice on reconnection, and
+    // must tell his peer actor so that it fails the corresponding upstream HTLCs instead of holding them.
+    val signingSession = desynchronizeSpliceSigningSession(bob)
+    disconnect(f)
+    reconnect(f)
+
+    val txAbort = bob2alice.fishForSpecificMessage() { case msg: TxAbort => msg }
+    assert(txAbort.toAscii == InvalidCommitmentNumber(channelId(bob), signingSession.fundingTxId).getMessage)
+    val aborted = bobPeer.fishForSpecificMessage() { case l: LiquidityPurchaseAborted => l }
+    assert(aborted.txId == purchase.txId)
+    assert(aborted.fundingTxIndex == purchase.fundingTxIndex)
+  }
+
+  test("recv CommitSigBatch including the parent commitment while signing a splice") { f =>
+    import f._
+
+    initiateSpliceWithoutSigs(f, spliceIn_opt = Some(SpliceIn(500_000 sat)))
+    alice2bob.expectMsgType[CommitSig]
+    alice2bob.forward(bob)
+    val spliceSigBob = bob2alice.expectMsgType[CommitSig]
+    bob2alice.expectMsgType[TxSignatures]
+
+    val aliceCommitments = alice.stateData.asInstanceOf[DATA_NORMAL].commitments
+    assert(aliceCommitments.active.size == 1)
+    val parentCommitment = aliceCommitments.active.head
+    val commitIndex = parentCommitment.localCommit.index
+    val signingSession = alice.stateData.asInstanceOf[DATA_NORMAL].spliceStatus.asInstanceOf[SpliceStatus.SpliceWaitingForSigs].signingSession
+    // The splice commitment is being signed at the same commitment index as its parent.
+    assert(signingSession.localCommitIndex == commitIndex)
+
+    // Instead of sending his commit_sig for the splice individually, Bob includes the *next* commit_sig for the parent
+    // commitment as well, which would desynchronize the parent and splice commitments.
+    val bobKeys = bob.underlyingActor.channelKeys
+    val bobCommitments = bob.stateData.asInstanceOf[DATA_NORMAL].commitments
+    val aliceNextPerCommitmentPoint = alice.underlyingActor.channelKeys.commitmentPoint(commitIndex + 1)
+    assert(bobCommitments.remoteNextCommitInfo == Right(aliceNextPerCommitmentPoint))
+    val remoteKeys = RemoteCommitmentKeys(bobCommitments.channelParams, bobKeys, aliceNextPerCommitmentPoint)
+    val bobParentCommitment = bobCommitments.active.find(_.fundingTxId == parentCommitment.fundingTxId).get
+    val Right((_, parentSigBob)) = bobParentCommitment.sendCommit(bobCommitments.channelParams, bobKeys, remoteKeys, bobCommitments.changes, aliceNextPerCommitmentPoint, batchSize = 2, nextRemoteNonce_opt = None)
+    assert(parentSigBob.fundingTxId_opt.contains(parentCommitment.fundingTxId))
+    // Bob sends that commit_sig as a batch, which must not be applied to the parent commitment while a splice is being
+    // signed: otherwise Alice would revoke the commitment index that the splice commitment is being created at.
+    alice ! CommitSigBatch(Seq(parentSigBob, spliceSigBob))
+    // Alice immediately force-closes.
+    assert(alice2bob.expectMsgType[Error].toAscii == CommitSigCountMismatch(channelId(alice), 1, 2).getMessage)
+    alice2blockchain.expectFinalTxPublished(aliceCommitments.latest.localCommit.txId)
+  }
+
   test("disconnect (tx_complete not received)") { f =>
     import f._
     // Disconnection with one side sending commit_sig

### eclair-core/src/test/scala/fr/acinq/eclair/io/PeerConnectionSpec.scala
@@ -442,7 +442,7 @@ class PeerConnectionSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike wi
     peer.expectNoMessage(100 millis)
     transport.send(peerConnection, commitSigs4(1))
     transport.expectMsg(TransportHandler.ReadAck(commitSigs4(1)))
-    peer.expectMsg(CommitSigBatch(commitSigs4.take(1)))
+    peer.expectMsg(commitSigs4.head)
     transport.send(peerConnection, commitSigs4.last)
     transport.expectMsg(TransportHandler.ReadAck(commitSigs4.last))
     peer.expectMsg(CommitSigBatch(commitSigs4.tail))
@@ -527,7 +527,7 @@ class PeerConnectionSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike wi
     peer.expectNoMessage(100 millis)
     transport.send(peerConnection, commitSigs4(1))
     transport.expectMsg(TransportHandler.ReadAck(commitSigs4(1)))
-    peer.expectMsg(CommitSigBatch(commitSigs4.take(1)))
+    peer.expectMsg(commitSigs4.head)
     peer.expectMsg(commitSigs4(1))
     peer.expectNoMessage(100 millis)
     transport.send(peerConnection, commitSigs4(2))