Stricter batching of commit_sig messages on the wire (#3083)

We introduce a `CommitSigBatch` class to group `commit_sig` messages
when splice transactions are pending. We use this class to ensure that
all the `commit_sig` messages in the batch are sent together to our
peer, without any other messages in-between.

We move the incoming `commit_sig` batching logic outside of the channel
and into the `PeerConnection` instead. This slightly simplifies the
channel FSM and its tests, since the `PeerConnection` actor is simpler.

We unfortunately cannot easily do this in the `TransportHandler` because
of our buffered read of the encrypted messages, which may split batches
and make it more complex to correctly group messages.

Author: t-bast

eclair-core/src/main/scala/fr/acinq/eclair/channel/Commitments.scala

@@ -1012,7 +1012,7 @@ case class Commitments(params: ChannelParams,
     }
   }
 
-  def sendCommit(channelKeys: ChannelKeys)(implicit log: LoggingAdapter): Either[ChannelException, (Commitments, Seq[CommitSig])] = {
+  def sendCommit(channelKeys: ChannelKeys)(implicit log: LoggingAdapter): Either[ChannelException, (Commitments, CommitSigs)] = {
     remoteNextCommitInfo match {
       case Right(_) if !changes.localHasChanges => Left(CannotSignWithoutChanges(channelId))
       case Right(remoteNextPerCommitmentPoint) =>
@@ -1026,21 +1026,24 @@ case class Commitments(params: ChannelParams,
           active = active1,
           remoteNextCommitInfo = Left(WaitForRev(localCommitIndex))
         )
-        Right(commitments1, sigs)
+        Right(commitments1, CommitSigs(sigs))
       case Left(_) => Left(CannotSignBeforeRevocation(channelId))
     }
   }
 
-  def receiveCommit(commits: Seq[CommitSig], channelKeys: ChannelKeys)(implicit log: LoggingAdapter): Either[ChannelException, (Commitments, RevokeAndAck)] = {
+  def receiveCommit(commitSigs: CommitSigs, channelKeys: ChannelKeys)(implicit log: LoggingAdapter): Either[ChannelException, (Commitments, RevokeAndAck)] = {
     // We may receive more commit_sig than the number of active commitments, because there can be a race where we send
     // splice_locked while our peer is sending us a batch of commit_sig. When that happens, we simply need to discard
     // the commit_sig that belong to commitments we deactivated.
-    if (commits.size < active.size) {
-      return Left(CommitSigCountMismatch(channelId, active.size, commits.size))
+    val sigs = commitSigs match {
+      case batch: CommitSigBatch if batch.batchSize < active.size => return Left(CommitSigCountMismatch(channelId, active.size, batch.batchSize))
+      case batch: CommitSigBatch => batch.messages
+      case _: CommitSig if active.size > 1 => return Left(CommitSigCountMismatch(channelId, active.size, 1))
+      case commitSig: CommitSig => Seq(commitSig)
     }
     val commitKeys = LocalCommitmentKeys(params, channelKeys, localCommitIndex + 1)
     // Signatures are sent in order (most recent first), calling `zip` will drop trailing sigs that are for deactivated/pruned commitments.
-    val active1 = active.zip(commits).map { case (commitment, commit) =>
+    val active1 = active.zip(sigs).map { case (commitment, commit) =>
       commitment.receiveCommit(params, channelKeys, commitKeys, changes, commit) match {
         case Left(f) => return Left(f)
         case Right(commitment1) => commitment1
@@ -1171,7 +1174,7 @@ case class Commitments(params: ChannelParams,
       case ChannelSpendSignature.IndividualSignature(latestRemoteSig) => latestRemoteSig == commitSig.signature
       case ChannelSpendSignature.PartialSignatureWithNonce(_, _) => ???
     }
-    params.channelFeatures.hasFeature(Features.DualFunding) && commitSig.batchSize == 1 && isLatestSig
+    params.channelFeatures.hasFeature(Features.DualFunding) && isLatestSig
   }
 
   def localFundingSigs(fundingTxId: TxId): Option[TxSignatures] = {

eclair-core/src/main/scala/fr/acinq/eclair/channel/Helpers.scala

@@ -479,14 +479,14 @@ object Helpers {
             // we just sent a new commit_sig but they didn't receive it
             // we resend the same updates and the same sig, and preserve the same ordering
             val signedUpdates = commitments.changes.localChanges.signed
-            val commitSigs = commitments.active.flatMap(_.nextRemoteCommit_opt).map(_.sig)
+            val commitSigs = CommitSigs(commitments.active.flatMap(_.nextRemoteCommit_opt).map(_.sig))
             retransmitRevocation_opt match {
               case None =>
-                SyncResult.Success(retransmit = signedUpdates ++ commitSigs)
+                SyncResult.Success(retransmit = signedUpdates :+ commitSigs)
               case Some(revocation) if commitments.localCommitIndex > waitingForRevocation.sentAfterLocalCommitIndex =>
-                SyncResult.Success(retransmit = signedUpdates ++ commitSigs ++ Seq(revocation))
+                SyncResult.Success(retransmit = signedUpdates :+ commitSigs :+ revocation)
               case Some(revocation) =>
-                SyncResult.Success(retransmit = Seq(revocation) ++ signedUpdates ++ commitSigs)
+                SyncResult.Success(retransmit = revocation +: signedUpdates :+ commitSigs)
             }
           case Left(_) if remoteChannelReestablish.nextLocalCommitmentNumber == (commitments.nextRemoteCommitIndex + 1) =>
             // we just sent a new commit_sig, they have received it but we haven't received their revocation