Send splice commit_sigs as a batch

There has been some discussion on the splicing spec around batching the
set of `commit_sig` messages that are sent while a splice is ongoing.

In this PR, we ensure that our `commit_sig` messages are sent as a batch
(no other messages can be interleaved between `commit_sig` messages).
This will let us gradually deploy the official splicing version from the
BOLTs: we first ensure that messages are sent as a batch, and later we
will be able to rely on this on the receiver side.

Author: t-bast

modules/core/src/commonMain/kotlin/fr/acinq/lightning/channel/ChannelAction.kt

@@ -16,6 +16,7 @@ sealed class ChannelAction {
     // @formatter:off
     sealed class Message : ChannelAction() {
         data class Send(val message: LightningMessage) : Message()
+        data class SendBatch(val messages: List<CommitSig>) : Message()
         data class SendToSelf(val command: ChannelCommand) : Message()
     }
 

modules/core/src/commonMain/kotlin/fr/acinq/lightning/channel/states/Normal.kt

@@ -80,7 +80,7 @@ data class Normal(
                         val actions = buildList {
                             add(ChannelAction.Storage.StoreHtlcInfos(htlcInfos))
                             add(ChannelAction.Storage.StoreState(nextState))
-                            addAll(result.value.second.map { ChannelAction.Message.Send(it) })
+                            add(ChannelAction.Message.SendBatch(result.value.second))
                         }
                         Pair(nextState, actions)
                     }

modules/core/src/commonMain/kotlin/fr/acinq/lightning/channel/states/ShuttingDown.kt

@@ -128,7 +128,7 @@ data class ShuttingDown(
                             val actions = buildList {
                                 add(ChannelAction.Storage.StoreHtlcInfos(htlcInfos))
                                 add(ChannelAction.Storage.StoreState(nextState))
-                                addAll(result.value.second.map { ChannelAction.Message.Send(it) })
+                                add(ChannelAction.Message.SendBatch(result.value.second))
                             }
                             Pair(nextState, actions)
                         }

modules/core/src/commonMain/kotlin/fr/acinq/lightning/channel/states/Syncing.kt

@@ -199,7 +199,7 @@ data class Syncing(val state: PersistedChannelState, val channelReestablishSent:
                                         }
 
                                     // we may need to retransmit updates and/or commit_sig and/or revocation
-                                    actions.addAll(syncResult.retransmit.map { ChannelAction.Message.Send(it) })
+                                    actions.addAll(syncResult.retransmit)
 
                                     // then we clean up unsigned updates
                                     val commitments1 = discardUnsignedUpdates(state.commitments)
@@ -231,8 +231,8 @@ data class Syncing(val state: PersistedChannelState, val channelReestablishSent:
                                     val commitments1 = discardUnsignedUpdates(state.commitments)
                                     val actions = buildList {
                                         addAll(syncResult.retransmit)
-                                        add(state.localShutdown)
-                                    }.map { ChannelAction.Message.Send(it) }
+                                        add(ChannelAction.Message.Send(state.localShutdown))
+                                    }
                                     Pair(state.copy(commitments = commitments1), actions)
                                 }
                             }
@@ -379,7 +379,7 @@ data class Syncing(val state: PersistedChannelState, val channelReestablishSent:
 
         // @formatter:off
         sealed class SyncResult {
-            data class Success(val retransmit: List<LightningMessage>) : SyncResult()
+            data class Success(val retransmit: List<ChannelAction.Message>) : SyncResult()
             sealed class Failure : SyncResult() {
                 data class LocalLateProven(val ourLocalCommitmentNumber: Long, val theirRemoteCommitmentNumber: Long) : Failure()
                 data class LocalLateUnproven(val ourRemoteCommitmentNumber: Long, val theirLocalCommitmentNumber: Long) : Failure()
@@ -400,31 +400,32 @@ data class Syncing(val state: PersistedChannelState, val channelReestablishSent:
             // step 2 depends on step 1 because we need to preserve ordering between commit_sig and revocation
 
             // step 2: we check the remote commitment
-            fun checkRemoteCommit(remoteChannelReestablish: ChannelReestablish, retransmitRevocation: RevokeAndAck?): SyncResult {
+            fun checkRemoteCommit(remoteChannelReestablish: ChannelReestablish, revocation: RevokeAndAck?): SyncResult {
+                val retransmitRevocation = revocation?.let { ChannelAction.Message.Send(it) }
                 return when (val rnci = commitments.remoteNextCommitInfo) {
                     is Either.Left -> {
                         when {
                             remoteChannelReestablish.nextLocalCommitmentNumber == commitments.nextRemoteCommitIndex -> {
                                 // 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.map { it.nextRemoteCommit }.filterIsInstance<NextRemoteCommit>().map { it.sig }
+                                val signedUpdates = commitments.changes.localChanges.signed.map { ChannelAction.Message.Send(it) }
+                                val commitSigs = ChannelAction.Message.SendBatch(commitments.active.mapNotNull { it.nextRemoteCommit }.map { it.sig })
                                 val retransmit = when (retransmitRevocation) {
                                     null -> buildList {
                                         addAll(signedUpdates)
-                                        addAll(commitSigs)
+                                        add(commitSigs)
                                     }
                                     else -> if (commitments.localCommitIndex > rnci.value.sentAfterLocalCommitIndex) {
                                         buildList {
                                             addAll(signedUpdates)
-                                            addAll(commitSigs)
+                                            add(commitSigs)
                                             add(retransmitRevocation)
                                         }
                                     } else {
                                         buildList {
                                             add(retransmitRevocation)
                                             addAll(signedUpdates)
-                                            addAll(commitSigs)
+                                            add(commitSigs)
                                         }
                                     }
                                 }
@@ -477,7 +478,7 @@ data class Syncing(val state: PersistedChannelState, val channelReestablishSent:
             // step 1: we check our local commitment
             return if (commitments.localCommitIndex == remoteChannelReestablish.nextRemoteRevocationNumber) {
                 // our local commitment is in sync, let's check the remote commitment
-                checkRemoteCommit(remoteChannelReestablish, retransmitRevocation = null)
+                checkRemoteCommit(remoteChannelReestablish, revocation = null)
             } else if (commitments.localCommitIndex == remoteChannelReestablish.nextRemoteRevocationNumber + 1) {
                 // they just sent a new commit_sig, we have received it but they didn't receive our revocation
                 val localPerCommitmentSecret = channelKeys.commitmentSecret(commitments.localCommitIndex - 1)
@@ -487,7 +488,7 @@ data class Syncing(val state: PersistedChannelState, val channelReestablishSent:
                     perCommitmentSecret = localPerCommitmentSecret,
                     nextPerCommitmentPoint = localNextPerCommitmentPoint
                 )
-                checkRemoteCommit(remoteChannelReestablish, retransmitRevocation = revocation)
+                checkRemoteCommit(remoteChannelReestablish, revocation)
             } else if (commitments.localCommitIndex > remoteChannelReestablish.nextRemoteRevocationNumber + 1) {
                 SyncResult.Failure.RemoteLate
             } else {

modules/core/src/commonMain/kotlin/fr/acinq/lightning/io/Peer.kt

@@ -859,6 +859,10 @@ class Peer(
                         }
                         peerConnection?.send(action.message) // ignore if disconnected
                     }
+                    is ChannelAction.Message.SendBatch -> {
+                        updatePeerStorage(nodeParams, channels + (channelId to state), peerConnection, theirInit?.features, logger)
+                        action.messages.forEach { peerConnection?.send(it) }
+                    }
                     // sometimes channel actions include "self" command (such as ChannelCommand.Commitment.Sign)
                     is ChannelAction.Message.SendToSelf -> input.send(WrappedChannelCommand(actualChannelId, action.command))
                     is ChannelAction.Blockchain.SendWatch -> watcher.watch(action.watch)

modules/core/src/commonTest/kotlin/fr/acinq/lightning/channel/TestsHelper.kt

@@ -29,7 +29,14 @@ import kotlinx.coroutines.runBlocking
 import kotlin.test.*
 
 // LN Message
-internal inline fun <reified T : LightningMessage> List<ChannelAction>.findOutgoingMessages(): List<T> = filterIsInstance<ChannelAction.Message.Send>().map { it.message }.filterIsInstance<T>()
+internal inline fun <reified T : LightningMessage> List<ChannelAction>.findOutgoingMessages(): List<T> = filterIsInstance<ChannelAction.Message>().flatMap {
+    when (it) {
+        is ChannelAction.Message.Send -> listOf(it.message)
+        is ChannelAction.Message.SendBatch -> it.messages
+        is ChannelAction.Message.SendToSelf -> listOf()
+    }
+}.filterIsInstance<T>()
+
 internal inline fun <reified T : LightningMessage> List<ChannelAction>.findOutgoingMessageOpt(): T? = findOutgoingMessages<T>().firstOrNull()
 internal inline fun <reified T : LightningMessage> List<ChannelAction>.findOutgoingMessage(): T = findOutgoingMessageOpt<T>() ?: fail("cannot find outgoing message ${T::class}")
 internal inline fun <reified T : LightningMessage> List<ChannelAction>.hasOutgoingMessage() = assertNotNull(findOutgoingMessageOpt<T>(), "cannot find outgoing message ${T::class}")
@@ -545,7 +552,7 @@ object TestsHelper {
         val rHasChanges = nodeB.state.commitments.changes.localHasChanges()
 
         val (sender0, sActions0) = nodeA.process(ChannelCommand.Commitment.Sign)
-        val commitSigs0 = sActions0.findOutgoingMessages<CommitSig>()
+        val commitSigs0 = sActions0.find<ChannelAction.Message.SendBatch>().messages
         assertEquals(commitmentsCount, commitSigs0.size)
         commitSigs0.forEach { assertEquals(commitmentsCount, it.batchSize) }
 
@@ -556,7 +563,7 @@ object TestsHelper {
         val (sender1, _) = sender0.process(ChannelCommand.MessageReceived(revokeAndAck0))
         assertIs<LNChannel<T>>(sender1)
         val (receiver1, rActions1) = receiver0.process(commandSign0)
-        val commitSigs1 = rActions1.findOutgoingMessages<CommitSig>()
+        val commitSigs1 = rActions1.find<ChannelAction.Message.SendBatch>().messages
         assertEquals(commitmentsCount, commitSigs1.size)
         commitSigs1.forEach { assertEquals(commitmentsCount, it.batchSize) }
 
@@ -568,7 +575,7 @@ object TestsHelper {
         if (rHasChanges) {
             val commandSign1 = sActions2.findCommand<ChannelCommand.Commitment.Sign>()
             val (sender3, sActions3) = sender2.process(commandSign1)
-            val commitSigs2 = sActions3.findOutgoingMessages<CommitSig>()
+            val commitSigs2 = sActions3.find<ChannelAction.Message.SendBatch>().messages
             assertEquals(commitmentsCount, commitSigs2.size)
 
             val (receiver3, rActions3) = receiveCommitSigs(receiver2, commitSigs2)