Release 0.8.0 with replay-safe prioritized transport

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloClientConnection.kt

@@ -87,7 +87,8 @@ class KiloClientConnection<S>(
 
                 val identity = pack(ClientIdentity(secretIdKey?.verifyingKey,
                     secretIdKey?.sign(clientAuthentication(params.token))))
-                val finished = params.decryptMessage(transport.call(L0ClientId, params.encrypt(identity)))
+                val finished = unpack<ProtectedMessage>(transport.callWithId(L0ClientId,
+                    readResponse = params::decrypt) { params.encode(identity) })
                 if (finished.kind != ProtectedKind.SERVER_FINISHED || finished.correlation != 0uL ||
                     finished.priority != 0 || !finished.payload.contentEquals(identity))
                     throw RemoteInterface.SecurityException("invalid server finished")
@@ -145,4 +146,13 @@ class KiloClientConnection<S>(
 }
 
 internal fun <S>Collection<KiloHandler<S>>.invokeAll(scope: KiloScope<S>) =
-    forEach { runCatching { scope.it() } }
+    forEach { handler ->
+        try {
+            scope.handler()
+        } catch (e: LocalInterface.BreakConnectionException) {
+            // Includes SecurityException and application-defined rejection subclasses.
+            throw e
+        } catch (_: Throwable) {
+            // Preserve best-effort notification for other handler failures.
+        }
+    }

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloL0Interface.kt

@@ -20,12 +20,13 @@ internal class KiloL0Interface<T>(
     // Plaintext transport errors must never use application exception constructors.
     override fun getErrorBuilder(code: String): ((String, UByteArray?) -> Throwable)? = null
 
-    override suspend fun executeTransportBlock(scope: Unit, block: Transport.Block.Call): UByteArray {
+    override suspend fun executeTransportBlock(scope: Unit, block: Transport.Block.Call): PreparedPayload {
         if (block.name != L0Call.name) return super.executeTransportBlock(scope, block)
         if (!allowHandshakeCallbacks && !deferredParams.isCompleted)
             throw RemoteInterface.SecurityException("application call before authentication")
         val params = deferredParams.await()
         val frame: ProtectedMessage = try { params.decryptMessage(unpack(block.packedArgs)) }
+            catch (e: DiscardPacket) { throw e }
             catch (e: Throwable) { throw RemoteInterface.SecurityException("invalid authenticated call") }
         val expected = if (block.id == 0u) ProtectedKind.PUSH else ProtectedKind.CALL
         if (frame.kind != expected || frame.correlation != block.id.toULong() || frame.priority != block.priority)
@@ -40,8 +41,12 @@ internal class KiloL0Interface<T>(
           catch (e: CancellationException) { throw e }
           catch (e: Throwable) { clientInterface.encodeError(0u, e) }
         // Push has no reply; its authenticated kind cannot be changed by rewriting id=0.
-        if (frame.kind == ProtectedKind.PUSH) return pack(Unit)
+        if (frame.kind == ProtectedKind.PUSH) return PreparedPayload { OutgoingData(pack(Unit)) }
         val kind = if (result is Transport.Block.Error) ProtectedKind.ERROR else ProtectedKind.RESPONSE
-        return pack(params.encryptMessage(ProtectedMessage(kind, frame.correlation, frame.priority, pack(result))))
+        val payload = pack(result)
+        return PreparedPayload {
+            val encoded = params.encodeMessage(ProtectedMessage(kind, frame.correlation, frame.priority, payload))
+            OutgoingData(pack(encoded.bytes), encoded.nonce)
+        }
     }
 }

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloParams.kt

@@ -45,8 +45,7 @@ data class KiloParams<S>(
     )
 
     private var nonce = 0UL
-    private var highestReceived: ULong? = null
-    private val received = mutableSetOf<ULong>()
+    private var received = ReplayWindow()
 
     val scope: KiloScope<S> by lazy {
         object : KiloScope<S> {
@@ -68,7 +67,9 @@ data class KiloParams<S>(
     /**
      * Encrypt using send keys and proper nonce
      */
-    fun encrypt(message: UByteArray, fillFactor: Float = 0f): UByteArray {
+    fun encrypt(message: UByteArray, fillFactor: Float = 0f): UByteArray = encode(message, fillFactor).bytes
+
+    internal fun encode(message: UByteArray, fillFactor: Float = 0f): OutgoingData {
         val fill = if (fillFactor > 0f)
             0..(message.size * fillFactor).roundToInt()
         else
@@ -79,24 +80,20 @@ data class KiloParams<S>(
             nonce++
         }
 
-        return pack(
+        return OutgoingData(pack(
             Package(n, sessionKey.encryptWithNonce(message, numericNonce.withULong(n), fill))
-        )
+        ), n)
     }
 
     /** Authenticate and consume one incoming nonce, allowing bounded out-of-order delivery. */
-    fun decrypt(encryptedMessage: UByteArray): UByteArray = protectDecryption {
-        val p: Package = BipackDecoder.decode(encryptedMessage.toDataSource())
-        protectedOp.invoke {
-            val high = highestReceived
-            if (p.nonce in received || (high != null && high >= 4096uL && p.nonce <= high - 4096uL))
-                throw RemoteInterface.SecurityException("replayed or stale encrypted packet")
-            val clear = sessionKey.decryptWithNonce(p.encryptedMessage, numericNonce.withULong(p.nonce))
-            received.add(p.nonce)
-            if (high == null || p.nonce > high) {
-                highestReceived = p.nonce
-                if (p.nonce >= 4096uL) received.removeAll { it <= p.nonce - 4096uL }
+    fun decrypt(encryptedMessage: UByteArray): UByteArray {
+        val p = protectDecryption { BipackDecoder.decode<Package>(encryptedMessage.toDataSource()) }
+        return protectedOp.invoke {
+            if (received.containsOrStale(p.nonce)) throw DiscardPacket()
+            val clear = protectDecryption {
+                sessionKey.decryptWithNonce(p.encryptedMessage, numericNonce.withULong(p.nonce))
             }
+            received.record(p.nonce)
             clear
         }
     }
@@ -105,11 +102,12 @@ data class KiloParams<S>(
     internal fun authenticated(identity: VerifyingPublicKey): KiloParams<S> = protectedOp.invoke {
         copy(remoteIdentity = identity).also {
             it.nonce = nonce
-            it.highestReceived = highestReceived
-            it.received.addAll(received)
+            it.received = received.copy()
         }
     }
 
+    internal fun encodeMessage(message: ProtectedMessage): OutgoingData = encode(pack(message))
+
     internal fun encryptMessage(message: ProtectedMessage): UByteArray = encrypt(pack(message))
     internal fun decryptMessage(message: UByteArray): ProtectedMessage =
         net.sergeych.utools.unpack(decrypt(message))

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloRemoteInterface.kt

@@ -4,12 +4,8 @@ package net.sergeych.kiloparsec
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.currentCoroutineContext
-import kotlinx.serialization.builtins.serializer
-import net.sergeych.utools.pack
 import net.sergeych.utools.unpack
 
-private val l0Notification = Command(L0Call.name, L0Call.argsSerializer, Unit.serializer())
-
 /** Full-duplex remote interface with authenticated request/result correlation. */
 class KiloRemoteInterface<S>(
     private val deferredParams: CompletableDeferred<KiloParams<S>>,
@@ -22,21 +18,20 @@ class KiloRemoteInterface<S>(
         var requestId = 0u
         val call = cmd.packCall(args)
         val encoded = try {
-            transport.callWithId(L0Call) { id ->
+            transport.callWithId(L0Call, readResponse = params::decrypt) { id ->
                 requestId = id
-                params.encryptMessage(ProtectedMessage(ProtectedKind.CALL, id.toULong(), priority, call))
+                params.encodeMessage(ProtectedMessage(ProtectedKind.CALL, id.toULong(), priority, call))
             }
         } catch (e: CancellationException) { throw e }
           catch (e: RemoteInterface.ClosedException) { throw e }
-          catch (e: Throwable) {
-            // An outer plaintext error is not an authenticated business result.
+          catch (e: RemoteInterface.SecurityException) {
             transport.abortSecurity()
-            throw RemoteInterface.SecurityException("unauthenticated outer error")
+            throw e
         }
         val frame: ProtectedMessage
         val block: Transport.Block
         try {
-            frame = params.decryptMessage(encoded)
+            frame = unpack(encoded)
             if (frame.correlation != requestId.toULong() || frame.priority != priority)
                 throw RemoteInterface.SecurityException("authenticated response routing mismatch")
             block = unpack(frame.payload)
@@ -60,7 +55,9 @@ class KiloRemoteInterface<S>(
     override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
         val params = deferredParams.await()
         val priority = currentCoroutineContext()[CallPriority]?.value ?: 0
-        params.transport.push(l0Notification, params.encryptMessage(
-            ProtectedMessage(ProtectedKind.PUSH, 0uL, priority, cmd.packCall(args))))
+        val payload = cmd.packCall(args)
+        (params.transport as Transport<*>).pushPrepared(L0Call.name, priority) {
+            params.encodeMessage(ProtectedMessage(ProtectedKind.PUSH, 0uL, priority, payload))
+        }
     }
 }

src/commonMain/kotlin/net/sergeych/kiloparsec/KiloServerConnection.kt

@@ -71,10 +71,10 @@ class KiloServerConnection<S>(
                     serverSigningKey?.seal(serverAuthentication(params!!.token)))
             } }
 
-            on(L0ClientId) { encoded -> handshakeLock.withLock {
-                if (established) throw RemoteInterface.SecurityException("repeated client identity")
+            onPrepared(L0ClientId) { encoded -> handshakeLock.withLock {
                 var p = params ?: throw RemoteInterface.SecurityException("wrong handshake sequence")
                 val identity = p.decrypt(encoded)
+                if (established) throw RemoteInterface.SecurityException("repeated client identity")
                 val ci = unpack<ClientIdentity>(identity)
                 if ((ci.clientIdKey == null) != (ci.signature == null))
                     throw RemoteInterface.SecurityException("incomplete client identity")
@@ -89,7 +89,10 @@ class KiloServerConnection<S>(
                 kiloRemoteInterface.complete(KiloRemoteInterface(deferredParams, clientInterface))
                 connectedScope = p.scope
                 clientInterface.onConnectHandlers.invokeAll(connectedScope)
-                p.encryptMessage(ProtectedMessage(ProtectedKind.SERVER_FINISHED, 0uL, 0, identity))
+                PreparedPayload {
+                    val result = p.encodeMessage(ProtectedMessage(ProtectedKind.SERVER_FINISHED, 0uL, 0, identity))
+                    OutgoingData(net.sergeych.utools.pack(result.bytes), result.nonce)
+                }
             } }
 
         }

src/commonMain/kotlin/net/sergeych/kiloparsec/LocalInterface.kt

@@ -16,6 +16,7 @@ import net.sergeych.mp_logger.exception
 import net.sergeych.mp_logger.info
 import net.sergeych.tools.AtomicCounter
 import net.sergeych.utools.firstNonNull
+import net.sergeych.utools.unpack
 import kotlin.reflect.KClass
 
 private typealias RawCommandHandler<C> = suspend (C, UByteArray) -> UByteArray
@@ -82,9 +83,19 @@ open class LocalInterface<S> : Loggable by LogTag("LocalInterface${idCounter.inc
             .invoke(scope, packedArgs)
 
 
-    /** Transport hook keeps outer routing metadata available to the authenticated layer. */
-    internal open suspend fun executeTransportBlock(scope: S, block: Transport.Block.Call): UByteArray =
-        execute(scope, block.name, block.packedArgs)
+    private val preparedHandlers = mutableMapOf<String, suspend (S, UByteArray) -> PreparedPayload>()
+
+    /** Secure handshake replies can defer sealing until atomic outgoing admission. */
+    internal fun <A> onPrepared(command: Command<A, UByteArray>, handler: suspend S.(A) -> PreparedPayload) {
+        preparedHandlers[command.name] = { scope, args -> scope.handler(unpack(command.argsSerializer, args)) }
+    }
+
+    /** Dispatch now; encode/seal the result later under the outgoing admission mutex. */
+    internal open suspend fun executeTransportBlock(scope: S, block: Transport.Block.Call): PreparedPayload {
+        preparedHandlers[block.name]?.let { return it(scope, block.packedArgs) }
+        val result = execute(scope, block.name, block.packedArgs)
+        return PreparedPayload { OutgoingData(result) }
+    }
 
     private val errorByClass = mutableMapOf<KClass<*>, String>()
     private val errorBuilder = mutableMapOf<String, (String, UByteArray?) -> Throwable>()

src/commonMain/kotlin/net/sergeych/kiloparsec/ReplayWindow.kt

@@ -0,0 +1,47 @@
+package net.sergeych.kiloparsec
+
+internal const val REPLAY_WINDOW_SIZE = 4096
+internal const val MAX_NONCE_OVERTAKING = REPLAY_WINDOW_SIZE * 2 / 3
+
+/** A packet disposition, not a session failure. Never dispatched or replied to. */
+internal class DiscardPacket : RuntimeException("Duplicate or stale encrypted packet")
+
+/** Fixed circular bitmap. Callers serialize check/authenticate/record as one operation. */
+internal class ReplayWindow {
+    private val bits = LongArray(REPLAY_WINDOW_SIZE / 64)
+    private var highest: ULong? = null
+    fun copy(): ReplayWindow = ReplayWindow().also {
+        bits.copyInto(it.bits)
+        it.highest = highest
+    }
+    private fun index(n: ULong) = (n % REPLAY_WINDOW_SIZE.toULong()).toInt()
+    fun containsOrStale(n: ULong): Boolean {
+        val h = highest ?: return false
+        if (n > h) return false
+        if (h - n >= REPLAY_WINDOW_SIZE.toULong()) return true
+        val i = index(n)
+        return bits[i / 64] and (1L shl (i % 64)) != 0L
+    }
+    fun record(n: ULong) {
+        val h = highest
+        if (h == null || n > h) {
+            if (h == null || n - h >= REPLAY_WINDOW_SIZE.toULong()) bits.fill(0)
+            else {
+                // Clear only newly entered positions, not every remembered nonce.
+                var next = h
+                repeat((n - h).toInt()) {
+                    val i = index(++next)
+                    bits[i / 64] = bits[i / 64] and (1L shl (i % 64)).inv()
+                }
+            }
+            highest = n
+        }
+        val i = index(n)
+        bits[i / 64] = bits[i / 64] or (1L shl (i % 64))
+    }
+}
+
+/** Local-only metadata; does not change wire encoding. */
+internal data class OutgoingData(val bytes: UByteArray, val nonce: ULong? = null)
+internal fun interface PreparedPayload { fun encode(): OutgoingData }
+internal class UnauthenticatedOuterError : RemoteInterface.SecurityException("Unauthenticated outer error")

src/commonMain/kotlin/net/sergeych/kiloparsec/Transport.kt

@@ -26,7 +26,7 @@ import kotlinx.serialization.encoding.Decoder
 import kotlinx.serialization.encoding.Encoder
 import kotlinx.serialization.serializer
 import net.sergeych.bipack.Unsigned
-import net.sergeych.crypto2.toDump
+import net.sergeych.kiloparsec.adapter.PriorityBlockWriter
 import net.sergeych.mp_logger.*
 import net.sergeych.utools.pack
 import net.sergeych.utools.unpack
@@ -115,6 +115,9 @@ class Transport<S>(
         }
     }
 
+    private val outgoing = Mutex()
+    private val writerReady = CompletableDeferred<PriorityBlockWriter>()
+    private val responseReaders = mutableMapOf<UInt, (UByteArray) -> UByteArray>()
     private val access = Mutex()
     private var lastId = 0u
     private val calls = mutableMapOf<UInt, CompletableDeferred<UByteArray>>()
@@ -127,13 +130,18 @@ class Transport<S>(
      * @throws RemoteInterface.ClosedException
      */
     private suspend fun sendCallBlock(name: String, packedArgs: UByteArray): UByteArray =
-        sendBoundCallBlock(name) { packedArgs }
+        sendBoundCallBlock(name) { OutgoingData(packedArgs) }
 
     /** Allocate the outer ID before sealing its authenticated inner counterpart. */
-    private suspend fun sendBoundCallBlock(name: String, packedArgs: (UInt) -> UByteArray): UByteArray {
+    private suspend fun sendBoundCallBlock(
+        name: String,
+        readResponse: ((UByteArray) -> UByteArray)? = null,
+        packedArgs: (UInt) -> OutgoingData,
+    ): UByteArray {
         if (isClosed) throw RemoteInterface.ClosedException()
 
-        val b: Block.Call
+        val id: UInt
+        val priority = currentCoroutineContext()[CallPriority]?.value ?: 0
         val deferred = CompletableDeferred<UByteArray>()
 
         // We need to shield calls and lastID with mutex, but nothing more:
@@ -141,20 +149,23 @@ class Transport<S>(
             if (isClosed) throw RemoteInterface.ClosedException()
             // the order is important: first id in use MUST BE >= 1, not zero:
             if (lastId == UInt.MAX_VALUE) throw RemoteInterface.ClosedException("RPC ID space exhausted")
-            val id = ++lastId
-            b = Block.Call(id, name, packedArgs(id)).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 }
-            calls[b.id] = deferred
+            id = ++lastId
+            calls[id] = deferred
+            if (readResponse != null) responseReaders[id] = readResponse
         }
 
         // Suspend on actual channel backpressure; never poll or delay a successful send.
         try {
-            device.send(pack<Block>(b))
+            sendPrepared {
+                val payload = packedArgs(id)
+                OutgoingData(pack<Block>(Block.Call(id, name, payload.bytes).also { it.priority = priority }), payload.nonce)
+            }
             return deferred.await()
         } catch (_: ClosedSendChannelException) {
             throw RemoteInterface.ClosedException()
         } finally {
             // Cancellation must not retain a response waiter for the connection lifetime.
-            withContext(NonCancellable) { access.withLock { calls.remove(b.id) } }
+            withContext(NonCancellable) { access.withLock { calls.remove(id); responseReaders.remove(id) } }
         }
     }
 
@@ -169,7 +180,7 @@ class Transport<S>(
 
         // All push blocks have the same id == 0:
         val b = Block.Call(0u, name, packedArgs).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 }
-        val r = runCatching { device.send(pack<Block>(b).also { debug { ">>$\n${it.toDump()}" } }) }
+        val r = runCatching { send(b) }
         when(val e = r.exceptionOrNull()) {
             null -> return
             is RemoteInterface.ClosedException, is CancellationException, is RemoteInterface.RemoteException
@@ -180,9 +191,34 @@ class Transport<S>(
 
     internal suspend fun callWithId(
         cmd: Command<UByteArray, UByteArray>,
-        build: (UInt) -> UByteArray,
-    ): UByteArray = unpack(cmd.resultSerializer,
-        sendBoundCallBlock(cmd.name) { id -> pack(cmd.argsSerializer, build(id)) })
+        readResponse: (UByteArray) -> UByteArray = { it },
+        build: (UInt) -> OutgoingData,
+    ): UByteArray = sendBoundCallBlock(cmd.name, { raw ->
+        try { readResponse(unpack(cmd.resultSerializer, raw)) }
+        catch (e: DiscardPacket) { throw e }
+        catch (e: Throwable) { throw RemoteInterface.SecurityException("Invalid authenticated response") }
+    }) { id ->
+        val payload = build(id)
+        OutgoingData(pack(cmd.argsSerializer, payload.bytes), payload.nonce)
+    }
+
+    internal suspend fun pushPrepared(name: String, priority: Int, prepare: () -> OutgoingData) {
+        sendPrepared {
+            val payload = prepare()
+            OutgoingData(pack<Block>(Block.Call(0u, name, pack(payload.bytes)).also { it.priority = priority }), payload.nonce)
+        }
+    }
+
+    /** Atomic preparation/admission, with carrier completion awaited outside the mutex. */
+    private suspend fun sendPrepared(prepare: () -> OutgoingData) {
+        val writer = writerReady.await()
+        val awaitSent = outgoing.withLock {
+            if (isClosed) throw RemoteInterface.ClosedException()
+            val frame = prepare()
+            writer.enqueue(frame.bytes, frame.nonce)
+        }
+        awaitSent()
+    }
 
     internal suspend fun abortSecurity() {
         isClosed = true
@@ -210,6 +246,10 @@ class Transport<S>(
      */
     suspend fun run() {
         coroutineScope {
+            // One ordered submission to the carrier at a time: adapter queues cannot
+            // undo the nonce-aware scheduling performed here.
+            val writer = PriorityBlockWriter(this, continueOnWriteFailure = true) { device.send(it) }
+            writerReady.complete(writer)
             debug { "awaiting incoming blocks" }
             // todo: rewrite it to close the job with no exceptions at all, always
             while (isActive && !isClosed) {
@@ -218,35 +258,56 @@ class Transport<S>(
                         val b = unpack<Block>(packed)
                         when (b) {
                             is Block.Error -> access.withLock {
-                                val error = localInterface.decodeError(b)
+                                val error = if (responseReaders.containsKey(b.forId)) UnauthenticatedOuterError() else localInterface.decodeError(b)
                                 warning { "decoded error: ${error::class.simpleName}: $error" }
-                                calls.remove(b.forId)?.completeExceptionally(localInterface.decodeError(b))
+                                responseReaders.remove(b.forId)
+                                calls.remove(b.forId)?.completeExceptionally(error)
                                     ?: warning { "error handler not found for ${b.forId}" }
                                 info { "error processed" }
                             }
 
                             is Block.Response -> access.withLock {
-                                calls.remove(b.forId)?.complete(b.packedResult)
-                                    ?: warning { "wait handle not found for ${b.forId}" }
+                                val pending = calls[b.forId]
+                                if (pending != null) {
+                                    try {
+                                        val result = responseReaders[b.forId]?.invoke(b.packedResult) ?: b.packedResult
+                                        calls.remove(b.forId)
+                                        responseReaders.remove(b.forId)
+                                        pending.complete(result)
+                                    } catch (_: DiscardPacket) {
+                                        // Keep the waiter: a valid response may still arrive.
+                                    } catch (failure: Throwable) {
+                                        calls.remove(b.forId)
+                                        responseReaders.remove(b.forId)
+                                        pending.completeExceptionally(failure)
+                                    }
+                                }
                             }
 
                             is Block.Call -> launch {
                                 try {
-                                    if (b.id == 0u)
-                                        // Command does not waits return
-                                        localInterface.executeTransportBlock(commandContext, b)
-                                    else
-                                        send(
-                                            Block.Response(
-                                                b.id,
-                                                localInterface.executeTransportBlock(commandContext, b)
-                                            ).also { it.priority = b.priority }
-                                        )
+                                    val result = localInterface.executeTransportBlock(commandContext, b)
+                                    if (b.id != 0u) {
+                                        try {
+                                            sendPrepared {
+                                                val encoded = result.encode()
+                                                OutgoingData(pack<Block>(Block.Response(b.id, encoded.bytes).also { it.priority = b.priority }), encoded.nonce)
+                                            }
+                                        } catch (failure: CancellationException) { throw failure }
+                                          catch (failure: LocalInterface.BreakConnectionException) { throw failure }
+                                          catch (failure: Throwable) {
+                                            // Local reply admission/I/O failure is not a plaintext
+                                            // remote business error. Leave peer timeout policy intact.
+                                            warning { "Reply send failed: ${failure.message}" }
+                                        }
+                                    }
+                                } catch (_: DiscardPacket) {
+                                    // Duplicate/stale input is silently discarded, including pushes.
                                 } catch (x: LocalInterface.BreakConnectionException) {
                                     // handler forced close
                                     warning { "handler requested closing of the connection (${x.flushSendQueue}" }
                                     isClosed = true
-                                    if (x.flushSendQueue) device.flush()
+                                    if (x.flushSendQueue) { writer.flush(); device.flush() }
                                     device.close()
                                 } catch (x: CancellationException) {
                                     throw x
@@ -290,6 +351,8 @@ class Transport<S>(
                 for (c in calls.values)
                     c.completeExceptionally(RemoteInterface.ClosedException())
                 calls.clear()
+                responseReaders.clear()
+                writer.close()
                 debug { "calls clear has been called" }
                 // Closing a device also releases dispatch children waiting on unfinished handshakes.
                 currentCoroutineContext().cancelChildren()
@@ -301,7 +364,7 @@ class Transport<S>(
 
     private suspend fun send(block: Block) {
         try {
-            device.send(pack(block))
+            sendPrepared { OutgoingData(pack(block)) }
         } catch (_: ClosedSendChannelException) {
             throw RemoteInterface.ClosedException()
         }

src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/OutQueue.kt

@@ -0,0 +1,25 @@
+package net.sergeych.kiloparsec.adapter
+
+import net.sergeych.kiloparsec.MAX_NONCE_OVERTAKING
+
+/** Externally synchronized. Nonces belong to one sending-key stream. */
+internal class OutQueue<T>(private val allowance: ULong = MAX_NONCE_OVERTAKING.toULong()) {
+    private data class Entry<T>(val value: T, val nonce: ULong?, val priority: Int)
+    private val entries = mutableListOf<Entry<T>>()
+    val values get() = entries.map { it.value }
+    fun push(value: T, nonce: ULong?, priority: Int) { entries.add(Entry(value, nonce, priority)) }
+    fun pop(): T? {
+        val oldest = entries.mapNotNull { it.nonce }.minOrNull()
+        var best: Entry<T>? = null
+        for (entry in entries) {
+            if (oldest != null && entry.nonce != null && entry.nonce - oldest > allowance) continue
+            val previous = best
+            if (previous == null || entry.priority > previous.priority ||
+                (entry.priority == previous.priority && entry.nonce != null && previous.nonce != null && entry.nonce < previous.nonce)) best = entry
+        }
+        return best?.also { entries.remove(it) }?.value
+    }
+    fun remove(value: T) { entries.removeAll { it.value === value } }
+    fun removeAll(predicate: (T) -> Boolean) { entries.removeAll { predicate(it.value) } }
+    fun clear() { entries.clear() }
+}

src/commonMain/kotlin/net/sergeych/kiloparsec/adapter/PriorityBlockWriter.kt

@@ -43,19 +43,20 @@ internal fun blockPriority(bytes: UByteArray): Int {
 
 /**
  * One adapter writer. Send completes after its write callback, not after an RPC reply.
- * Queue ordering is numeric priority then FIFO. Cancellation removes queued work;
+ * Queue ordering is nonce-bounded numeric priority; unnumbered ties remain FIFO. Cancellation removes queued work;
  * bytes already being written cannot be recalled. No rate limit or idle batching delay.
  */
 internal class PriorityBlockWriter(
     scope: CoroutineScope,
     private val maxQueuedItems: Int = 128,
     private val maxQueuedBytes: Int = 32 * 1024 * 1024,
+    private val continueOnWriteFailure: Boolean = false,
     private val write: suspend (UByteArray) -> Unit,
 ) {
     private class Pending(val bytes: UByteArray, val priority: Int, val done: CompletableDeferred<Unit>)
     private val mutex = Mutex()
     private val wake = Channel<Unit>(Channel.CONFLATED)
-    private val pending = mutableListOf<Pending>()
+    private val pending = OutQueue<Pending>()
     private var closed = false
     private var writing: Pending? = null
     private val job = scope.launch {
@@ -64,13 +65,17 @@ internal class PriorityBlockWriter(
             while (isActive) {
                 current = mutex.withLock {
                     pending.removeAll { !it.done.isActive }
-                    // maxByOrNull retains the first equal value: stable FIFO ties.
-                    pending.maxByOrNull { it.priority }?.also { pending.remove(it) }.also { writing = it }
+                    pending.pop().also { writing = it }
                 }
                 val entry = current
                 if (entry == null) { wake.receive(); continue }
-                if (entry.done.isActive) write(entry.bytes)
-                entry.done.complete(Unit)
+                try {
+                    if (entry.done.isActive) write(entry.bytes)
+                    entry.done.complete(Unit)
+                } catch (failure: Throwable) {
+                    entry.done.completeExceptionally(failure)
+                    if (!continueOnWriteFailure || failure is CancellationException) throw failure
+                }
                 mutex.withLock { writing = null }
                 current = null
             }
@@ -81,32 +86,36 @@ internal class PriorityBlockWriter(
             withContext(NonCancellable) {
                 mutex.withLock {
                     closed = true
-                    pending.forEach { it.done.completeExceptionally(RemoteInterface.ClosedException()) }
+                    pending.values.forEach { it.done.completeExceptionally(RemoteInterface.ClosedException()) }
                     pending.clear()
                 }
             }
         }
     }
 
-    suspend fun send(bytes: UByteArray) {
+    /** Admission never waits for the carrier. Caller may hold its preparation mutex here. */
+    suspend fun enqueue(bytes: UByteArray, nonce: ULong? = null): suspend () -> Unit {
         currentCoroutineContext().ensureActive()
         val entry = Pending(bytes, blockPriority(bytes), CompletableDeferred())
         mutex.withLock {
             if (closed || !job.isActive) throw RemoteInterface.ClosedException()
             pending.removeAll { !it.done.isActive }
-            if (pending.size >= maxQueuedItems || bytes.size > maxQueuedBytes - pending.sumOf { it.bytes.size })
+            if (pending.values.size >= maxQueuedItems || bytes.size > maxQueuedBytes - pending.values.sumOf { it.bytes.size })
                 throw RemoteInterface.Exception("Outgoing block queue is full")
-            pending.add(entry)
+            pending.push(entry, nonce, entry.priority)
         }
         wake.trySend(Unit)
-        try { entry.done.await() }
-        finally {
-            entry.done.cancel()
-            withContext(NonCancellable) { mutex.withLock { pending.remove(entry) } }
+        return {
+            try { entry.done.await() }
+            finally {
+                entry.done.cancel()
+                withContext(NonCancellable) { mutex.withLock { pending.remove(entry) } }
+            }
         }
     }
+    suspend fun send(bytes: UByteArray) { enqueue(bytes).invoke() }
     suspend fun flush() {
-        val queued = mutex.withLock { (pending + listOfNotNull(writing)).map { it.done } }
+        val queued = mutex.withLock { (pending.values + listOfNotNull(writing)).map { it.done } }
         queued.forEach { it.await() }
     }
     fun close() { job.cancel() }

src/commonTest/kotlin/AuthenticatedEnvelopeTest.kt

@@ -127,10 +127,14 @@ class AuthenticatedEnvelopeTest {
         val clientJob = launch { runCatching { client.run { if (it) ready.complete(Unit) } } }
         try {
             ready.await()
-            withTimeout(2000) { runCatching { client.call(mutate, 1) } }
-            withTimeout(2000) { serverJob.join() }
+            assertEquals(1, withTimeout(2000) { client.call(mutate, 1) })
+            runCurrent()
             assertTrue(attacked)
-            assertTrue(executions <= 1, "The replay must never execute the application operation again")
+            assertEquals(1, executions, "The replay must never execute the application operation again")
+            assertTrue(serverJob.isActive)
+            assertEquals(2, client.call(mutate, 2))
+            runCurrent()
+            assertEquals(2, executions)
         } finally { listOf(serverJob, clientJob).forEach { it.cancelAndJoin() } }
     }
 

src/commonTest/kotlin/ConnectionInitializationTest.kt

@@ -0,0 +1,88 @@
+@file:OptIn(ExperimentalUnsignedTypes::class)
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.runTest
+import net.sergeych.crypto2.SigningSecretKey
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import kotlin.test.*
+import kotlin.time.Duration.Companion.seconds
+
+class ConnectionInitializationTest {
+    private class PolicyRejected : RemoteInterface.SecurityException("Peer rejected by policy")
+    private class InitializationAborted : LocalInterface.BreakConnectionException("Initialization aborted")
+
+    @Test fun serverSecurityRejectionClosesConnection() = rejectsInitialization(true, PolicyRejected())
+    @Test fun clientSecurityRejectionClosesConnection() = rejectsInitialization(false, PolicyRejected())
+    @Test fun serverBreakSubclassClosesConnection() = rejectsInitialization(true, InitializationAborted())
+    @Test fun clientBreakSubclassClosesConnection() = rejectsInitialization(false, InitializationAborted())
+
+    private fun rejectsInitialization(rejectOnServer: Boolean, rejection: Throwable) = runTest(timeout = 10.seconds) {
+        initCrypto()
+        val (a, b) = createTestDevice()
+        val serverKey = SigningSecretKey.new()
+        val clientKey = SigningSecretKey.new()
+        val clientInterface = KiloInterface<Unit>()
+        val serverInterface = KiloInterface<Unit>()
+        var policyCalled = false
+        var identityMatched = false
+        var laterHandlerCalled = false
+        var clientConnected = false
+        (if (rejectOnServer) serverInterface else clientInterface).apply {
+            onConnected {
+                identityMatched = remoteIdentity == if (rejectOnServer) clientKey.verifyingKey else serverKey.verifyingKey
+                policyCalled = true
+                throw rejection
+            }
+            onConnected { laterHandlerCalled = true }
+        }
+        val server = KiloServerConnection(serverInterface, b, Unit, serverKey)
+        val client = KiloClientConnection(clientInterface, a, Unit, clientKey)
+        val serverRun = async { runCatching { server.run() } }
+        val clientRun = async { runCatching { client.run { if (it) clientConnected = true } } }
+        try {
+            withTimeout(2000) { serverRun.await(); clientRun.await() }
+            assertTrue(policyCalled, "Initialization policy must run")
+            assertTrue(identityMatched, "Policy must receive the verified remote identity")
+            assertFalse(laterHandlerCalled, "Rejection must stop subsequent initialization handlers")
+            assertFalse(clientConnected, "Rejected initialization must not announce a connected client")
+            if (!rejectOnServer) assertSame(rejection, clientRun.await().exceptionOrNull())
+            val echo by command<String, String>()
+            assertIs<RemoteInterface.ClosedException>(runCatching { client.call(echo, "after rejection") }.exceptionOrNull())
+        } finally {
+            serverRun.cancelAndJoin()
+            clientRun.cancelAndJoin()
+        }
+    }
+
+    @Test fun ordinaryHandlerFailuresStillAllowInitialization() = runTest(timeout = 10.seconds) {
+        initCrypto()
+        val (a, b) = createTestDevice()
+        val echo by command<String, String>()
+        var serverInitialized = false
+        var clientInitialized = false
+        val serverInterface = KiloInterface<Unit>().apply {
+            onConnected { throw IllegalStateException("Nonfatal handler failure") }
+            onConnected { serverInitialized = true }
+            on(echo) { it }
+        }
+        val clientInterface = KiloInterface<Unit>().apply {
+            onConnected { throw IllegalStateException("Nonfatal handler failure") }
+            onConnected { clientInitialized = true }
+        }
+        val ready = CompletableDeferred<Unit>()
+        val server = KiloServerConnection(serverInterface, b, Unit)
+        val client = KiloClientConnection(clientInterface, a, Unit)
+        val serverJob = launch { server.run() }
+        val clientJob = launch { client.run { if (it) ready.complete(Unit) } }
+        try {
+            ready.await()
+            assertTrue(serverInitialized)
+            assertTrue(clientInitialized)
+            assertEquals("ok", client.call(echo, "ok"))
+        } finally {
+            serverJob.cancelAndJoin()
+            clientJob.cancelAndJoin()
+        }
+    }
+}

src/commonTest/kotlin/OutgoingSecurityTest.kt

@@ -0,0 +1,114 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.pack
+import net.sergeych.utools.unpack
+import kotlin.test.*
+import kotlin.time.Duration.Companion.seconds
+
+class OutgoingSecurityTest {
+    @Test fun replayedReplyLeavesWaiterForValidReply() = runTest(timeout = 10.seconds) {
+        initCrypto()
+        val echo by command<String, String>()
+        val (a, b) = createTestDevice()
+        var armed = false
+        var saved: UByteArray? = null
+        val adversary = object : Transport.Device by b {
+            override suspend fun send(block: UByteArray) {
+                val bytes = block
+                val decoded = unpack<Transport.Block>(bytes)
+                if (armed && decoded is Transport.Block.Response) {
+                    saved?.let { b.send(pack<Transport.Block>(decoded.copy(packedResult = it))) }
+                    saved = decoded.packedResult
+                }
+                b.send(bytes)
+            }
+        }
+        val ready = CompletableDeferred<Unit>()
+        val server = KiloServerConnection(KiloInterface<Unit>().apply { on(echo) { it } }, adversary, Unit)
+        val client = KiloClientConnection(KiloInterface<Unit>(), a, Unit)
+        val sj = launch { server.run() }
+        val cj = launch { client.run { if (it) ready.complete(Unit) } }
+        try {
+            ready.await(); armed = true
+            assertEquals("first", client.call(echo, "first"))
+            assertEquals("second", client.call(echo, "second"))
+            assertTrue(sj.isActive); assertTrue(cj.isActive)
+        } finally { sj.cancelAndJoin(); cj.cancelAndJoin() }
+    }
+
+    @Test fun localReplySendFailureDoesNotForgePlaintextErrorOrClose() = runTest(timeout = 10.seconds) {
+        initCrypto()
+        val echo by command<Int, Int>()
+        val (a, b) = createTestDevice()
+        var fail = false
+        var plaintextErrors = 0
+        val device = object : Transport.Device by b {
+            override suspend fun send(block: UByteArray) {
+                val bytes = block
+                val decoded = unpack<Transport.Block>(bytes)
+                if (decoded is Transport.Block.Error) plaintextErrors++
+                if (fail && decoded is Transport.Block.Response) throw RemoteInterface.Exception("Local write failure")
+                b.send(bytes)
+            }
+        }
+        val ready = CompletableDeferred<Unit>()
+        val server = KiloServerConnection(KiloInterface<Unit>().apply { on(echo) { it } }, device, Unit)
+        val client = KiloClientConnection(KiloInterface<Unit>(), a, Unit)
+        val sj = launch { server.run() }
+        val cj = launch { client.run { if (it) ready.complete(Unit) } }
+        try {
+            ready.await(); fail = true
+            assertNull(withTimeoutOrNull(100) { client.call(echo, 1) })
+            assertEquals(0, plaintextErrors)
+            fail = false
+            assertEquals(2, client.call(echo, 2))
+        } finally { sj.cancelAndJoin(); cj.cancelAndJoin() }
+    }
+
+    @Test fun allSecurePathsShareOrderedNonceAdmission() = runTest(timeout = 20.seconds) {
+        initCrypto()
+        val echo by command<Int, Int>()
+        val notify by command<Int, Unit>()
+        val (a, b) = createTestDevice()
+        fun observing(device: Transport.Device, nonces: MutableList<ULong>, skipHandshakeReply: Boolean = false) = object : Transport.Device by device {
+            override suspend fun send(block: UByteArray) {
+                val bytes = block
+                when (val decoded = unpack<Transport.Block>(bytes)) {
+                    is Transport.Block.Call -> if (decoded.name != "L0Request")
+                        nonces += unpack<KiloParams.Package>(unpack<UByteArray>(decoded.packedArgs)).nonce
+                    is Transport.Block.Response -> if (!skipHandshakeReply || decoded.forId != 1u)
+                        nonces += unpack<KiloParams.Package>(unpack<UByteArray>(decoded.packedResult)).nonce
+                    else -> Unit
+                }
+                device.send(bytes)
+            }
+        }
+        val an = mutableListOf<ULong>(); val bn = mutableListOf<ULong>()
+        val api = KiloInterface<Unit>().apply { on(echo) { it }; on(notify) {} }
+        val ready = CompletableDeferred<Unit>()
+        val server = KiloServerConnection(api, observing(b, bn, true), Unit)
+        val client = KiloClientConnection(api, observing(a, an), Unit)
+        val sj = launch { server.run() }
+        val cj = launch { client.run { if (it) ready.complete(Unit) } }
+        try {
+            ready.await()
+            // Real JVM parallel preparation, requests in both directions, replies and pushes.
+            withContext(Dispatchers.Default) {
+                (0 until 32).map { n -> async {
+                    val remote: RemoteInterface = if (n % 2 == 0) client else server
+                    remote.push(notify, n)
+                    assertEquals(n, remote.call(echo, n))
+                } }.awaitAll()
+            }
+            for (nonces in listOf(an, bn)) {
+                assertEquals(0uL, nonces.first())
+                assertTrue(nonces.zipWithNext().all { (x, y) -> y > x }, "Equal-priority admission must follow nonce allocation: $nonces")
+                assertTrue(nonces.size >= 49)
+            }
+        } finally { sj.cancelAndJoin(); cj.cancelAndJoin() }
+    }
+}

src/commonTest/kotlin/ReplayDeliveryTest.kt

@@ -0,0 +1,97 @@
+@file:OptIn(ExperimentalUnsignedTypes::class)
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.*
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.pack
+import net.sergeych.utools.unpack
+import kotlin.test.*
+import kotlin.time.Duration.Companion.seconds
+
+class ReplayDeliveryTest {
+ @Test fun replayWindowBoundaries() = runTest {
+  initCrypto()
+  val a = SafeKeyExchange(); val b = SafeKeyExchange()
+  val (da, db) = createTestDevice()
+  val ta = Transport(da, LocalInterface<Unit>(), Unit)
+  val tb = Transport(db, LocalInterface<Unit>(), Unit)
+  val sender = KiloParams(false, ta, a.clientSessionKey(b.publicKey), Unit, null, ta)
+  val receiver = KiloParams(true, tb, b.serverSessionKey(a.publicKey), Unit, null, tb)
+  val packets = (0..4096).map { sender.encrypt(pack(it)) }
+  assertEquals(4096, unpack<Int>(receiver.decrypt(packets[4096])))
+  assertEquals(2, unpack<Int>(receiver.decrypt(packets[2])))
+  assertEquals(1, unpack<Int>(receiver.decrypt(packets[1])))
+  assertFails { receiver.decrypt(packets[0]) }
+  assertFails { receiver.decrypt(packets[1]) }
+  da.close(); db.close()
+ }
+ @Test fun localSendFailurePreservesHealthyConnection() = runTest(timeout = 10.seconds) {
+  initCrypto()
+  val echo by command<String,String>()
+  val (a,b) = createTestDevice()
+  val ready = CompletableDeferred<Unit>()
+  var fail = true
+  val localFailure = object: Transport.Device by a {
+   override suspend fun send(block: UByteArray) {
+    val decoded = unpack<Transport.Block>(block)
+    if(fail && decoded is Transport.Block.Call && decoded.name == "L0Call")
+      throw RemoteInterface.Exception("Outgoing block queue is full")
+    a.send(block)
+   }
+  }
+  val server=KiloServerConnection(KiloInterface<Unit>().apply { on(echo) { it } }, b, Unit)
+  val client=KiloClientConnection(KiloInterface<Unit>(),localFailure,Unit)
+  val sj=launch { runCatching { server.run() } }
+  val cj=launch { runCatching { client.run { if(it) ready.complete(Unit) } } }
+  try {
+   ready.await()
+   val result=runCatching { client.call(echo,"x") }
+   assertIs<RemoteInterface.Exception>(result.exceptionOrNull())
+   assertFalse(result.exceptionOrNull() is RemoteInterface.SecurityException)
+   assertTrue(cj.isActive); assertTrue(sj.isActive)
+   fail = false
+   assertEquals("recovered", client.call(echo,"recovered"))
+  } finally { sj.cancelAndJoin(); cj.cancelAndJoin() }
+ }
+
+ @Test fun adjacentReorderingSurvives() = reorderedPushes(1, false)
+ @Test fun deepReorderingDiscardsWithoutClosing() = reorderedPushes(4096, true)
+ private fun reorderedPushes(overtakers: Int, shouldDiscard: Boolean) = runTest(timeout = 20.seconds) {
+  initCrypto()
+  val notify by command<Int,Unit>()
+  val (a,b)=createTestDevice()
+  val ready=CompletableDeferred<Unit>()
+  val newerReceived=CompletableDeferred<Unit>()
+  val heldReceived=CompletableDeferred<Unit>()
+  var held: UByteArray?=null
+  val reorder=object: Transport.Device by a {
+   override suspend fun send(block: UByteArray) {
+    val decoded=unpack<Transport.Block>(block)
+    if(decoded is Transport.Block.Call && decoded.name=="L0Call" && held==null) held=block
+    else a.send(block)
+   }
+  }
+  val server=KiloServerConnection(KiloInterface<Unit>().apply { on(notify) {
+   if(it==overtakers) newerReceived.complete(Unit)
+   if(it==0) heldReceived.complete(Unit)
+  } }, b, Unit)
+  val client=KiloClientConnection(KiloInterface<Unit>(),reorder,Unit)
+  val sj=launch { runCatching { server.run() } }
+  val cj=launch { runCatching { client.run { if(it) ready.complete(Unit) } } }
+  try {
+   ready.await()
+   for(i in 0..overtakers) client.push(notify,i)
+   newerReceived.await()
+   a.send(held!!)
+   if(shouldDiscard) {
+    runCurrent()
+    assertTrue(sj.isActive); assertTrue(cj.isActive)
+    assertFalse(heldReceived.isCompleted)
+    client.ping()
+   } else {
+    withTimeout(2000) { heldReceived.await() }
+    assertTrue(sj.isActive); assertTrue(cj.isActive)
+   }
+  } finally { sj.cancelAndJoin(); cj.cancelAndJoin() }
+ }
+}

src/commonTest/kotlin/ReplayWindowTest.kt

@@ -0,0 +1,28 @@
+@file:OptIn(ExperimentalUnsignedTypes::class)
+package net.sergeych.kiloparsec
+
+import kotlin.random.Random
+import kotlin.test.*
+
+class ReplayWindowTest {
+    @Test fun boundariesWraparoundAndLargeJumpsMatchReference() {
+        val window = ReplayWindow()
+        val seen = mutableSetOf<ULong>()
+        var highest: ULong? = null
+        val random = Random(41)
+        val inputs = mutableListOf(4096uL, 1uL, 2uL, 0uL, 1uL, 8191uL, 4096uL, 8192uL)
+        repeat(15000) { inputs += random.nextInt(0, 30000).toULong() }
+        inputs += listOf(ULong.MAX_VALUE - 1u, ULong.MAX_VALUE, ULong.MAX_VALUE - 4095u, ULong.MAX_VALUE - 4096u)
+        for (n in inputs) {
+            val h = highest
+            val reject = n in seen || (h != null && n <= h && h - n >= 4096u)
+            assertEquals(reject, window.containsOrStale(n), "nonce=$n highest=$h")
+            if (!reject) {
+                window.record(n)
+                seen += n
+                if (h == null || n > h) highest = n
+                assertTrue(window.containsOrStale(n))
+            }
+        }
+    }
+}

src/commonTest/kotlin/net/sergeych/kiloparsec/adapter/OutQueueTest.kt

@@ -0,0 +1,123 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+package net.sergeych.kiloparsec.adapter
+
+import createTestDevice
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.pack
+import net.sergeych.utools.unpack
+import kotlin.test.*
+
+class OutQueueTest {
+    @Test fun priorityWithinWindowAndOldestProgress() {
+        val queue = OutQueue<ULong>()
+        queue.push(100u, 100u, -100)
+        for (n in 101uL..5000uL) queue.push(n, n, 100)
+        val sent = mutableListOf<ULong>()
+        while (true) sent += queue.pop() ?: break
+        assertEquals((101uL..2830uL).toList(), sent.take(2730))
+        assertEquals(100uL, sent[2730])
+        assertEquals(4901, sent.toSet().size)
+    }
+    @Test fun gapsAndMaximumNonceDoNotOverflow() {
+        val queue = OutQueue<ULong>()
+        queue.push(ULong.MAX_VALUE, ULong.MAX_VALUE, 100)
+        queue.push(ULong.MAX_VALUE - 2731u, ULong.MAX_VALUE - 2731u, -100)
+        assertEquals(ULong.MAX_VALUE - 2731u, queue.pop())
+        assertEquals(ULong.MAX_VALUE, queue.pop())
+        assertNull(queue.pop())
+    }
+    @Test fun writerHandlesCancellationCapacityAndContinuesAfterLocalFailure() = runTest {
+        val gate = CompletableDeferred<Unit>()
+        val entered = CompletableDeferred<Unit>()
+        val observed = mutableListOf<Int>()
+        val writer = PriorityBlockWriter(backgroundScope, maxQueuedItems = 2, continueOnWriteFailure = true) {
+            val n = it[0].toInt()
+            if (n == 0) { entered.complete(Unit); gate.await() }
+            if (n == 3) throw IllegalStateException("Local carrier rejection")
+            observed += n
+        }
+        val first = async { writer.enqueue(ubyteArrayOf(0u), 0u).invoke() }
+        entered.await()
+        val cancelled = async { writer.enqueue(ubyteArrayOf(1u), 1u).invoke() }
+        runCurrent()
+        val second = async { writer.enqueue(ubyteArrayOf(2u), 2u).invoke() }
+        runCurrent()
+        assertFailsWith<RemoteInterface.Exception> { writer.enqueue(ubyteArrayOf(9u), 9u) }
+        cancelled.cancelAndJoin()
+        gate.complete(Unit)
+        first.await(); second.await()
+        assertFailsWith<IllegalStateException> { writer.enqueue(ubyteArrayOf(3u), 3u).invoke() }
+        writer.enqueue(ubyteArrayOf(4u), 4u).invoke()
+        assertEquals(listOf(0, 2, 4), observed)
+        writer.close(); runCurrent()
+        assertFailsWith<RemoteInterface.ClosedException> { writer.enqueue(ubyteArrayOf(5u), 5u) }
+    }
+    @Test fun writerUsesNonceBoundBeforePriority() = runTest {
+        val gate = CompletableDeferred<Unit>()
+        val entered = CompletableDeferred<Unit>()
+        val order = mutableListOf<Int>()
+        val writer = PriorityBlockWriter(backgroundScope) { bytes ->
+            val n = (unpack<Transport.Block>(bytes) as Transport.Block.Call).id.toInt()
+            if (n == 0) { entered.complete(Unit); gate.await() }
+            order += n
+        }
+        fun frame(n: Int, priority: Int) = pack<Transport.Block>(Transport.Block.Call(n.toUInt(), "x", ubyteArrayOf()).also { it.priority = priority })
+        val first = async { writer.enqueue(frame(0, 0), 0u).invoke() }
+        entered.await()
+        val old = writer.enqueue(frame(1, 0), 1u)
+        val newer = writer.enqueue(frame(2, 100), 2732u)
+        gate.complete(Unit); first.await(); old(); newer()
+        assertEquals(3, order.size)
+        assertEquals(listOf(0, 1, 2), order)
+        writer.close()
+    }
+
+    @Test fun encryptedBacklogCrossingWindowStillDeliversEveryMessage() = runTest {
+        net.sergeych.crypto2.initCrypto()
+        val a = net.sergeych.crypto2.SafeKeyExchange()
+        val b = net.sergeych.crypto2.SafeKeyExchange()
+        val (da, db) = createTestDevice()
+        val ta = Transport(da, LocalInterface<Unit>(), Unit)
+        val tb = Transport(db, LocalInterface<Unit>(), Unit)
+        val sender = KiloParams(false, ta, a.clientSessionKey(b.publicKey), Unit, null, ta)
+        val receiver = KiloParams(true, tb, b.serverSessionKey(a.publicKey), Unit, null, tb)
+        val gate = CompletableDeferred<Unit>()
+        val entered = CompletableDeferred<Unit>()
+        val order = mutableListOf<Int>()
+        // Larger test capacity reaches the boundary in one batch; production retains 128.
+        val writer = PriorityBlockWriter(backgroundScope, maxQueuedItems = 5000) { bytes ->
+            val frame = unpack<Transport.Block>(bytes) as Transport.Block.Call
+            if (frame.id == 0u) { entered.complete(Unit); gate.await() }
+            order += unpack<Int>(receiver.decrypt(unpack(frame.packedArgs)))
+        }
+        suspend fun admit(n: Int, priority: Int): suspend () -> Unit {
+            val encoded = sender.encode(pack(n))
+            val frame = Transport.Block.Call(n.toUInt(), "L0Call", pack(encoded.bytes)).also { it.priority = priority }
+            return writer.enqueue(pack<Transport.Block>(frame), encoded.nonce)
+        }
+        try {
+            val first = admit(0, 0)
+            entered.await()
+            val completions = mutableListOf(admit(1, -100))
+            for (n in 2..4097) completions += admit(n, 100)
+            gate.complete(Unit); first()
+            completions.forEach { it() }
+            assertEquals(4098, order.size)
+            assertEquals((0..4097).toSet(), order.toSet())
+            assertEquals(2731, order.indexOf(1))
+        } finally { writer.close(); da.close(); db.close() }
+    }
+
+    @Test fun closingWriterReleasesCurrentAndQueuedSenders() = runTest {
+        val entered = CompletableDeferred<Unit>()
+        val writer = PriorityBlockWriter(backgroundScope) { entered.complete(Unit); awaitCancellation() }
+        val first = async { runCatching { writer.send(ubyteArrayOf(0u)) } }
+        entered.await()
+        val second = async { runCatching { writer.enqueue(ubyteArrayOf(1u), 1u).invoke() } }
+        runCurrent(); writer.close()
+        assertTrue(first.await().isFailure)
+        assertTrue(second.await().isFailure)
+    }
+}

src/jvmTest/kotlin/net/sergeych/kiloparsec/adapter/CarrierReleaseTest.kt

@@ -0,0 +1,91 @@
+@file:OptIn(ExperimentalUnsignedTypes::class)
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import net.sergeych.crypto2.SigningSecretKey
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.unpack
+import kotlin.test.*
+
+/** Real loopback carriers; impairment occurs before writing actual datagrams. */
+class CarrierReleaseTest {
+    @Test fun udpDuplicatesReorderingAndConcurrentDuplex() = carrier(udp = true)
+    @Test fun tcpConcurrentDuplexAndPriority() = carrier(udp = false)
+
+    private fun carrier(udp: Boolean) = runBlocking {
+        withTimeout(20_000) {
+            initCrypto()
+            val tcp = if (!udp) TcpServer(0, "127.0.0.1") else null
+            val datagrams = if (udp) UdpServer(0, "127.0.0.1") else null
+            val received = Channel<Int>(Channel.UNLIMITED)
+            val echo by command<Int, Int>()
+            val notify by command<Int, Unit>()
+            val api = KiloInterface<Unit>().apply {
+                on(echo) { it }
+                on(notify) { received.send(it) }
+            }
+            val serverReady = CompletableDeferred<KiloServerConnection<Unit>>()
+            val serverKey = SigningSecretKey.new()
+            val collector = launch {
+                (tcp?.transportFlow ?: datagrams!!.transportFlow).collect { device ->
+                    val server = KiloServerConnection(api, device, Unit, serverKey)
+                    serverReady.complete(server)
+                    launch { server.run() }
+                }
+            }
+            val port = tcp?.boundPort() ?: datagrams!!.boundPort()
+            val raw = if (udp) connectUdpDevice(NetworkAddress("127.0.0.1", port))
+                      else connectTcpDevice(NetworkAddress("127.0.0.1", port))
+            var held: UByteArray? = null
+            var impaired = false
+            val device = object : Transport.Device by raw {
+                override suspend fun send(block: UByteArray) {
+                    val decoded = unpack<Transport.Block>(block)
+                    if (udp && !impaired && decoded is Transport.Block.Call && decoded.name == "L0Call" && decoded.id == 0u) {
+                        val first = held
+                        if (first == null) held = block
+                        else {
+                            raw.send(block)
+                            raw.send(first)
+                            raw.send(first) // exact duplicate over the real UDP socket
+                            impaired = true
+                        }
+                    } else raw.send(block)
+                }
+            }
+            val ready = CompletableDeferred<Unit>()
+            val clientApi = KiloInterface<Unit>().apply {
+                onConnected { if (remoteIdentity != serverKey.verifyingKey) throw RemoteInterface.SecurityException() }
+                on(echo) { it }
+            }
+            val client = KiloClientConnection(clientApi, device, Unit)
+            val clientJob = launch { client.run { if (it) ready.complete(Unit) } }
+            try {
+                ready.await()
+                val server = serverReady.await()
+                client.push(notify, 1)
+                client.push(notify, 2)
+                val notices = setOf(received.receive(), received.receive())
+                assertEquals(setOf(1, 2), notices)
+                coroutineScope {
+                    (0 until 32).map { n -> async {
+                        val peer: RemoteInterface = if (n % 2 == 0) client else server
+                        assertEquals(n, peer.call(echo, n, priority = if (n % 3 == 0) 100 else 0))
+                    } }.awaitAll()
+                }
+                client.ping()
+                assertNull(withTimeoutOrNull(100) { received.receive() }, "Duplicate must not execute again")
+                assertTrue(clientJob.isActive)
+                if (udp) assertTrue(impaired)
+            } finally {
+                clientJob.cancelAndJoin()
+                raw.close()
+                collector.cancelAndJoin()
+                tcp?.close(); datagrams?.close()
+                received.close()
+            }
+        }
+    }
+}