Batch incoming commit_sig in PeerConnection

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

@@ -1159,7 +1159,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/fsm/Channel.scala

@@ -18,8 +18,8 @@ package fr.acinq.eclair.io
 
 import akka.actor.{ActorRef, FSM, OneForOneStrategy, PoisonPill, Props, Stash, SupervisorStrategy, Terminated}
 import akka.event.Logging.MDC
-import fr.acinq.bitcoin.scalacompat.BlockHash
 import fr.acinq.bitcoin.scalacompat.Crypto.PublicKey
+import fr.acinq.bitcoin.scalacompat.{BlockHash, ByteVector32}
 import fr.acinq.eclair.Logs.LogCategory
 import fr.acinq.eclair.crypto.Noise.KeyPair
 import fr.acinq.eclair.crypto.TransportHandler
@@ -343,10 +343,48 @@ class PeerConnection(keyPair: KeyPair, conf: PeerConnection.Conf, switchboard: A
         stay()
 
       case Event(msg: LightningMessage, d: ConnectedData) =>
-        // we acknowledge and pass all other messages to the peer
+        // We immediately acknowledge all other messages.
         d.transport ! TransportHandler.ReadAck(msg)
-        d.peer ! msg
-        stay()
+        // We immediately forward messages to the peer, unless they are part of a batch, in which case we wait to
+        // receive the whole batch before forwarding.
+        msg match {
+          case msg: CommitSig =>
+            msg.tlvStream.get[CommitSigTlv.BatchTlv].map(_.size) match {
+              case Some(batchSize) if batchSize > 25 =>
+                log.warning("received legacy batch of commit_sig exceeding our threshold ({} > 25), processing messages individually", batchSize)
+                // We don't want peers to be able to exhaust our memory by sending batches of dummy messages that we keep in RAM.
+                d.peer ! msg
+                stay()
+              case Some(batchSize) if batchSize > 1 =>
+                d.legacyCommitSigBatch_opt match {
+                  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)
+                    stay() using d.copy(legacyCommitSigBatch_opt = Some(PendingCommitSigBatch(msg.channelId, batchSize, Seq(msg))))
+                  case Some(pending) =>
+                    val received1 = pending.received :+ msg
+                    if (received1.size == batchSize) {
+                      log.debug("received last commit_sig in legacy batch for channel_id={}", msg.channelId)
+                      d.peer ! CommitSigBatch(received1)
+                      stay() using d.copy(legacyCommitSigBatch_opt = None)
+                    } else {
+                      log.debug("received commit_sig {}/{} in legacy batch for channel_id={}", received1.size, batchSize, msg.channelId)
+                      stay() using d.copy(legacyCommitSigBatch_opt = Some(pending.copy(received = received1)))
+                    }
+                  case None =>
+                    log.debug("received first commit_sig in legacy batch of size {} for channel_id={}", batchSize, msg.channelId)
+                    stay() using d.copy(legacyCommitSigBatch_opt = Some(PendingCommitSigBatch(msg.channelId, batchSize, Seq(msg))))
+                }
+              case _ =>
+                log.debug("received individual commit_sig for channel_id={}", msg.channelId)
+                d.peer ! msg
+                stay()
+            }
+          case _ =>
+            d.peer ! msg
+            stay()
+        }
 
       case Event(readAck: TransportHandler.ReadAck, d: ConnectedData) =>
         // we just forward acks to the transport (e.g. from the router)
@@ -566,8 +604,19 @@ object PeerConnection {
   case class AuthenticatingData(pendingAuth: PendingAuth, transport: ActorRef, isPersistent: Boolean) extends Data with HasTransport
   case class BeforeInitData(remoteNodeId: PublicKey, pendingAuth: PendingAuth, transport: ActorRef, isPersistent: Boolean) extends Data with HasTransport
   case class InitializingData(chainHash: BlockHash, pendingAuth: PendingAuth, remoteNodeId: PublicKey, transport: ActorRef, peer: ActorRef, localInit: protocol.Init, doSync: Boolean, isPersistent: Boolean) extends Data with HasTransport
-  case class ConnectedData(chainHash: BlockHash, remoteNodeId: PublicKey, transport: ActorRef, peer: ActorRef, localInit: protocol.Init, remoteInit: protocol.Init, rebroadcastDelay: FiniteDuration, gossipTimestampFilter: Option[GossipTimestampFilter] = None, behavior: Behavior = Behavior(), expectedPong_opt: Option[ExpectedPong] = None, isPersistent: Boolean) extends Data with HasTransport
-
+  case class ConnectedData(chainHash: BlockHash,
+                           remoteNodeId: PublicKey,
+                           transport: ActorRef,
+                           peer: ActorRef,
+                           localInit: protocol.Init, remoteInit: protocol.Init,
+                           rebroadcastDelay: FiniteDuration,
+                           gossipTimestampFilter: Option[GossipTimestampFilter] = None,
+                           behavior: Behavior = Behavior(),
+                           expectedPong_opt: Option[ExpectedPong] = None,
+                           legacyCommitSigBatch_opt: Option[PendingCommitSigBatch] = None,
+                           isPersistent: Boolean) extends Data with HasTransport
+
+  case class PendingCommitSigBatch(channelId: ByteVector32, batchSize: Int, received: Seq[CommitSig])
   case class ExpectedPong(ping: Ping, timestamp: TimestampMilli = TimestampMilli.now())
   case class PingTimeout(ping: Ping)
 

eclair-core/src/main/scala/fr/acinq/eclair/io/PeerConnection.scala

@@ -440,9 +440,7 @@ object CommitSigs {
 case class CommitSig(channelId: ByteVector32,
                      signature: ByteVector64,
                      htlcSignatures: List[ByteVector64],
-                     tlvStream: TlvStream[CommitSigTlv] = TlvStream.empty) extends CommitSigs {
-  val batchSize: Int = tlvStream.get[CommitSigTlv.BatchTlv].map(_.size).getOrElse(1)
-}
+                     tlvStream: TlvStream[CommitSigTlv] = TlvStream.empty) extends CommitSigs
 
 case class CommitSigBatch(messages: Seq[CommitSig]) extends CommitSigs {
   require(messages.map(_.channelId).toSet.size == 1, "commit_sig messages in a batch must be for the same channel")

eclair-core/src/main/scala/fr/acinq/eclair/wire/protocol/LightningMessageTypes.scala

@@ -463,23 +463,17 @@ trait ChannelStateTestsBase extends Assertions with Eventually {
     val rHasChanges = r.stateData.asInstanceOf[ChannelDataWithCommitments].commitments.changes.localHasChanges
     s ! CMD_SIGN(Some(sender.ref))
     sender.expectMsgType[RES_SUCCESS[CMD_SIGN]]
-    s2r.expectMsgType[CommitSigs] match {
-      case sig: CommitSig => s2r.forward(r, sig)
-      case batch: CommitSigBatch => batch.messages.foreach(sig => s2r.forward(r, sig))
-    }
+    s2r.expectMsgType[CommitSigs]
+    s2r.forward(r)
     r2s.expectMsgType[RevokeAndAck]
     r2s.forward(s)
-    r2s.expectMsgType[CommitSigs] match {
-      case sig: CommitSig => r2s.forward(s, sig)
-      case batch: CommitSigBatch => batch.messages.foreach(sig => r2s.forward(s, sig))
-    }
+    r2s.expectMsgType[CommitSigs]
+    r2s.forward(s)
     s2r.expectMsgType[RevokeAndAck]
     s2r.forward(r)
     if (rHasChanges) {
-      s2r.expectMsgType[CommitSigs] match {
-        case sig: CommitSig => s2r.forward(r, sig)
-        case batch: CommitSigBatch => batch.messages.foreach(sig => s2r.forward(r, sig))
-      }
+      s2r.expectMsgType[CommitSigs]
+      s2r.forward(r)
       r2s.expectMsgType[RevokeAndAck]
       r2s.forward(s)
       eventually {

eclair-core/src/test/scala/fr/acinq/eclair/channel/states/ChannelStateTestsHelperMethods.scala

@@ -1519,12 +1519,12 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     alice ! CMD_SIGN()
     val sigsA = alice2bob.expectMsgType[CommitSigBatch]
     assert(sigsA.batchSize == 2)
-    sigsA.messages.foreach(sig => alice2bob.forward(bob, sig))
+    alice2bob.forward(bob, sigsA)
     bob2alice.expectMsgType[RevokeAndAck]
     bob2alice.forward(alice)
     val sigsB = bob2alice.expectMsgType[CommitSigBatch]
     assert(sigsB.batchSize == 2)
-    sigsB.messages.foreach(sig => bob2alice.forward(alice, sig))
+    bob2alice.forward(alice, sigsB)
     alice2bob.expectMsgType[RevokeAndAck]
     alice2bob.forward(bob)
     awaitCond(alice.stateData.asInstanceOf[DATA_NORMAL].commitments.active.forall(_.localCommit.spec.htlcs.size == 1))
@@ -1548,12 +1548,12 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     alice2bob.forward(bob)
     val sigsA = alice2bob.expectMsgType[CommitSigBatch]
     assert(sigsA.batchSize == 2)
-    sigsA.messages.foreach(sig => alice2bob.forward(bob, sig))
+    alice2bob.forward(bob, sigsA)
     bob2alice.expectMsgType[RevokeAndAck]
     bob2alice.forward(alice)
     val sigsB = bob2alice.expectMsgType[CommitSigBatch]
     assert(sigsB.batchSize == 2)
-    sigsB.messages.foreach(sig => bob2alice.forward(alice, sig))
+    bob2alice.forward(alice, sigsB)
     alice2bob.expectMsgType[RevokeAndAck]
     alice2bob.forward(bob)
     awaitCond(alice.stateData.asInstanceOf[DATA_NORMAL].commitments.active.forall(_.localCommit.spec.htlcs.size == 1))
@@ -1672,16 +1672,14 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
     alice ! CMD_SIGN()
     val commitSigsAlice = alice2bob.expectMsgType[CommitSigBatch]
     assert(commitSigsAlice.batchSize == 3)
-    alice2bob.forward(bob, commitSigsAlice.messages(0))
     bob ! WatchPublishedTriggered(spliceTx2)
     val spliceLockedBob = bob2alice.expectMsgType[SpliceLocked]
     assert(spliceLockedBob.fundingTxId == spliceTx2.txid)
     bob2alice.forward(alice, spliceLockedBob)
-    alice2bob.forward(bob, commitSigsAlice.messages(1))
-    alice2bob.forward(bob, commitSigsAlice.messages(2))
+    alice2bob.forward(bob, commitSigsAlice)
     bob2alice.expectMsgType[RevokeAndAck]
     bob2alice.forward(alice)
-    assert(bob2alice.expectMsgType[CommitSig].batchSize == 1)
+    bob2alice.expectMsgType[CommitSig]
     bob2alice.forward(alice)
     alice2bob.expectMsgType[RevokeAndAck]
     alice2bob.forward(bob)
@@ -3335,13 +3333,13 @@ class NormalSplicesStateSpec extends TestKitBaseClass with FixtureAnyFunSuiteLik
       bob ! CMD_SIGN()
       inside(bob2alice.expectMsgType[CommitSigBatch]) { batch =>
         assert(batch.batchSize == 3)
-        batch.messages.foreach(sig => bob2alice.forward(alice, sig))
+        bob2alice.forward(alice, batch)
       }
       alice2bob.expectMsgType[RevokeAndAck]
       alice2bob.forward(bob)
       inside(alice2bob.expectMsgType[CommitSigBatch]) { batch =>
         assert(batch.batchSize == 3)
-        batch.messages.foreach(sig => alice2bob.forward(bob, sig))
+        alice2bob.forward(bob, batch)
       }
       bob2alice.expectMsgType[RevokeAndAck]
       bob2alice.forward(alice)

eclair-core/src/test/scala/fr/acinq/eclair/channel/states/e/NormalSplicesStateSpec.scala

@@ -23,6 +23,7 @@ import fr.acinq.bitcoin.scalacompat.{Block, ByteVector32}
 import fr.acinq.eclair.FeatureSupport.{Mandatory, Optional}
 import fr.acinq.eclair.Features._
 import fr.acinq.eclair.TestConstants._
+import fr.acinq.eclair.TestUtils.randomTxId
 import fr.acinq.eclair.crypto.TransportHandler
 import fr.acinq.eclair.io.Peer.ConnectionDown
 import fr.acinq.eclair.message.OnionMessages.{Recipient, buildMessage}
@@ -348,6 +349,85 @@ class PeerConnectionSpec extends TestKitBaseClass with FixtureAnyFunSuiteLike wi
     transport.expectNoMessage(100 millis)
   }
 
+  test("receive legacy batch of commit_sig messages") { f =>
+    import f._
+    connect(nodeParams, remoteNodeId, switchboard, router, connection, transport, peerConnection, peer)
+
+    // We receive a batch of commit_sig messages from a first channel.
+    val channelId1 = randomBytes32()
+    val commitSigs1 = Seq(
+      CommitSig(channelId1, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+      CommitSig(channelId1, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+    )
+    transport.send(peerConnection, commitSigs1.head)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs1.head))
+    peer.expectNoMessage(100 millis)
+    transport.send(peerConnection, commitSigs1.last)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs1.last))
+    peer.expectMsg(CommitSigBatch(commitSigs1))
+
+    // We receive a batch of commit_sig messages from a second channel.
+    val channelId2 = randomBytes32()
+    val commitSigs2 = Seq(
+      CommitSig(channelId2, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(3))),
+      CommitSig(channelId2, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(3))),
+      CommitSig(channelId2, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(3))),
+    )
+    commitSigs2.dropRight(1).foreach(commitSig => {
+      transport.send(peerConnection, commitSig)
+      transport.expectMsg(TransportHandler.ReadAck(commitSig))
+    })
+    peer.expectNoMessage(100 millis)
+    transport.send(peerConnection, commitSigs2.last)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs2.last))
+    peer.expectMsg(CommitSigBatch(commitSigs2))
+
+    // We receive another batch of commit_sig messages from the first channel, with unrelated messages in the batch.
+    val commitSigs3 = Seq(
+      CommitSig(channelId1, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+      CommitSig(channelId1, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+    )
+    transport.send(peerConnection, commitSigs3.head)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs3.head))
+    val spliceLocked1 = SpliceLocked(channelId1, randomTxId())
+    transport.send(peerConnection, spliceLocked1)
+    transport.expectMsg(TransportHandler.ReadAck(spliceLocked1))
+    peer.expectMsg(spliceLocked1)
+    val spliceLocked2 = SpliceLocked(channelId2, randomTxId())
+    transport.send(peerConnection, spliceLocked2)
+    transport.expectMsg(TransportHandler.ReadAck(spliceLocked2))
+    peer.expectMsg(spliceLocked2)
+    peer.expectNoMessage(100 millis)
+    transport.send(peerConnection, commitSigs3.last)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs3.last))
+    peer.expectMsg(CommitSigBatch(commitSigs3))
+
+    // We start receiving a batch of commit_sig messages from the first channel, interleaved with a batch from the second
+    // channel, which is not supported.
+    val commitSigs4 = Seq(
+      CommitSig(channelId1, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+      CommitSig(channelId2, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+      CommitSig(channelId2, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(2))),
+    )
+    transport.send(peerConnection, commitSigs4.head)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs4.head))
+    peer.expectNoMessage(100 millis)
+    transport.send(peerConnection, commitSigs4(1))
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs4(1)))
+    peer.expectMsg(CommitSigBatch(commitSigs4.take(1)))
+    transport.send(peerConnection, commitSigs4.last)
+    transport.expectMsg(TransportHandler.ReadAck(commitSigs4.last))
+    peer.expectMsg(CommitSigBatch(commitSigs4.tail))
+
+    // We receive a batch that exceeds our threshold: we process them individually.
+    val invalidCommitSigs = (0 until 30).map(_ => CommitSig(channelId2, randomBytes64(), Nil, TlvStream(CommitSigTlv.BatchTlv(30))))
+    invalidCommitSigs.foreach(commitSig => {
+      transport.send(peerConnection, commitSig)
+      transport.expectMsg(TransportHandler.ReadAck(commitSig))
+      peer.expectMsg(commitSig)
+    })
+  }
+
   test("react to peer's bad behavior") { f =>
     import f._
     val probe = TestProbe()