Add prioritized transport, bounded reconnect, and configurable keepalive

Carry numeric priorities through encrypted RPC and adapter writers with bounded admission. Correct push semantics and socket lifecycle cleanup, expose dispatch gating, and add opt-in connection health.

Preserve the WebSocket setup API while allowing applications to override keepalive settings. Enable Android socket publication and document the 0.7.2-SNAPSHOT release workflow.

.gitignore

@@ -47,3 +47,7 @@ out/
 /.gigaide/gigaide.properties
 local.properties
 /kotlin-js-store/yarn.lock
+
+# JVM crash diagnostics must never be published
+*.hprof
+hs_err_pid*.log

PRIORITY.md

@@ -0,0 +1,69 @@
+# Numeric transport priority — 0.7.2-SNAPSHOT
+
+```kotlin
+import net.sergeych.kiloparsec.call
+import net.sergeych.kiloparsec.push
+
+remote.call(command, arguments, priority = 100)
+remote.push(notification, arguments, priority = 0)
+```
+
+Larger signed `Int` values go first. Equal priorities preserve FIFO. Existing
+call/push entry points remain unchanged and use zero unless invoked inside an
+explicit priority call context. The overloads are extension functions, preserving
+existing RemoteInterface implementers. A coroutine context carries the selected
+value through the encrypted RPC wrappers; it becomes serialized outer-block data.
+
+The adapter owns one outgoing writer. Waiting for a response does not occupy the
+writer: calls retain independent request IDs and response waiters. Both transport
+responses and errors copy the request's priority, including encrypted RPC errors
+returned inside an outer response. Pushes have no normal response.
+
+## Wire compatibility
+
+The legacy block tag and body are unchanged. Nonzero priority is appended as four
+big-endian two's-complement bytes. Zero omits the trailer entirely. The new reader
+defaults to zero only when the trailer is absent; a partial trailer is malformed.
+
+This relies on each Transport.Block having its own finite byte-array frame. TCP's
+existing length prefix, WebSocket messages and UDP data blocks provide that
+boundary. The old decoder consumes the legacy body and ignores trailing bytes.
+Do not use this EOF-terminated extension to concatenate unframed Block objects or
+embed them directly in a positional collection; wrap each block in its own bytes.
+The handshake remains version 1 because the original wire format is retained.
+
+Old peers interoperate but do not honor/inherit priority. New peers can decode
+both formats. BiPack @Extendable is deliberately not added: it would insert a new
+field-count prefix before existing fields and break old readers.
+
+## Adapter behavior
+
+TCP, WebSocket client/server, and UDP use a per-connection priority writer.
+The writer reads the legacy header lengths and skips payloads to locate priority;
+it does not deserialize/copy/decrypt the payload or guess from the last four bytes.
+`Transport.Device.send(block)` defaults to the existing output channel for custom
+or legacy devices. Built-in ProxyDevice-based adapters override it to enter their
+writer directly. The old output channel remains available, bridged into that writer;
+new transports should call send so pending senders are visible to priority ordering.
+Custom adapters retain their existing FIFO behavior until they implement priority.
+
+Queue admission is bounded to 128 waiting blocks and 32 MiB (plus the block
+currently being written). Overflow fails explicitly rather than allocating an
+unbounded backlog. Send awaits its adapter write callback, not a remote RPC reply.
+Queued cancellation removes the block; cancellation cannot recall a write already
+started. Closing the writer releases pending senders. There is no rate limiter or
+batch-collection delay. WebSocket writes flush after each block to avoid excessive
+look-ahead in the engine's FIFO. Socket/TLS buffers and network loss can still delay
+control behind bytes already submitted; priority is not a preemption guarantee.
+
+## Verification
+
+Focused JVM tests cover signed numeric ordering, stable ties, queued cancellation,
+bounded admission, close, absent/partial trailers, unchanged legacy bodies, and
+priority inheritance through encrypted concurrent calls and encrypted errors.
+An encrypted simulated 16 KiB/s link saturated in both directions maintains control
+round trips below 400 ms and aggregate utilization above 80% of full-duplex capacity.
+This simulation does not model kernel/TLS buffering or packet loss.
+JVM, Android and JS compilation and socket/resource lifecycle tests pass.
+Caravan also checked call/error/response fixtures with the actual packaged
+0.7.1-SNAPSHOT binary and the new 0.7.2-SNAPSHOT binary in separate JVM processes.

README.md

@@ -310,3 +310,41 @@ export restrictions are lifted. We do not support such practices here at 8-rays.
 [Sergey Chernov]: https://t.me/real_sergeych
 [^1]: On some new Xiaomi phones we found problems with websocket binary frames, probably in ktor; use text frames
 otherwise.
+
+## Numeric transport priority (0.7.2-SNAPSHOT)
+
+`call(command, args, priority = 100)` and `push(command, args, priority = 0)`
+select numeric outgoing priority. Responses inherit request priority, calls remain
+concurrent, and the framed wire extension interoperates with legacy peers.
+See [priority API, compatibility and limits](PRIORITY.md).
+
+
+## Connection health and WebSocket keepalive (0.7.2-SNAPSHOT)
+
+Encrypted application heartbeat is opt-in: `KiloClientConnection.run()` does not
+start it. Pass `ConnectionHealth` explicitly only when periodic probes are needed.
+`ReconnectionOptions` configures connection deadlines, retry delay, and an optional
+finite attempt budget; `reconnectExhausted` signals when that budget ends.
+
+The server WebSocket helper retains its existing defaults and signature. Its new
+configuration overload permits applications to disable the transport pinger:
+
+```kotlin
+setupWebsocketServer(api, path = "/rpc", serverKey = key,
+    configureWebSockets = { pingPeriodMillis = 0 }) { Session() }
+```
+
+Without heartbeat, applications must handle transport closure and bound real
+request response times. A successful socket write does not prove peer reachability.
+Proxy/NAT inactivity timeouts remain independent of these settings.
+
+## Publishing
+
+`./gradlew publishAllPublicationsToMavenRepository` publishes the configured
+version to the private Gitea Maven registry using the existing local credential
+file. Never commit that credential. Apple target publication requires a supported
+macOS builder; check task output for skipped targets on other hosts.
+
+`bin/pubdocs` generates Dokka and uploads it to the existing Kiloparsec documentation
+site. For development-only Maven Local publication without configuring the remote
+repository, pass `-PlocalPublishingOnly=true`.

build.gradle.kts

@@ -19,7 +19,7 @@ plugins {
 }
 
 group = "net.sergeych"
-version = "0.7.1-SNAPSHOT"
+version = "0.7.2-SNAPSHOT"
 
 repositories {
     google()
@@ -48,7 +48,9 @@ kotlin {
 
     mingwX64()
 
-    androidTarget()
+    androidTarget {
+        publishLibraryVariants("release")
+    }
     @OptIn(ExperimentalWasmDsl::class)
     wasmJs {
         browser()
@@ -84,6 +86,7 @@ kotlin {
                 implementation("io.ktor:ktor-network:$ktor_version")
             }
         }
+        androidMain.dependsOn(ktorSocketMain)
         val commonTest by getting {
             dependencies {
                 implementation(kotlin("test"))
@@ -155,6 +158,7 @@ publishing {
         File("${System.getProperty("user.home")}/.gitea_token").readText()
     }
     repositories {
+        if (!providers.gradleProperty("localPublishingOnly").map(String::toBoolean).getOrElse(false)) {
             maven {
                 credentials(HttpHeaderCredentials::class) {
                     name = "Authorization"
@@ -167,6 +171,7 @@ publishing {
             }
         }
     }
+}
 
 
 tasks.dokkaHtml.configure {

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

@@ -0,0 +1,18 @@
+package net.sergeych.kiloparsec
+
+import kotlinx.coroutines.withContext
+import kotlin.coroutines.AbstractCoroutineContextElement
+import kotlin.coroutines.CoroutineContext
+
+/** Local propagation through the encrypted RPC layers; serialized on the outer transport block. */
+internal class CallPriority(val value: Int) : AbstractCoroutineContextElement(Key) {
+    companion object Key : CoroutineContext.Key<CallPriority>
+}
+
+/** Larger values are sent first. Existing two-argument calls remain priority zero. */
+suspend fun <A, R> RemoteInterface.call(cmd: Command<A, R>, args: A, priority: Int): R =
+    withContext(CallPriority(priority)) { call(cmd, args) }
+suspend fun <R> RemoteInterface.call(cmd: Command<Unit, R>, priority: Int): R = call(cmd, Unit, priority)
+suspend fun <A> RemoteInterface.push(cmd: Command<A, Unit>, args: A, priority: Int) =
+    withContext(CallPriority(priority)) { push(cmd, args) }
+suspend fun RemoteInterface.push(cmd: Command<Unit, Unit>, priority: Int) = push(cmd, Unit, priority)

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

@@ -0,0 +1,27 @@
+package net.sergeych.kiloparsec
+
+import kotlinx.serialization.builtins.serializer
+
+/** Opt-in end-to-end encrypted heartbeat, independent of the carrier's keepalive. */
+data class ConnectionHealth(
+    val intervalMillis: Long = 15_000,
+    val timeoutMillis: Long = 5_000,
+) {
+    init { require(intervalMillis > 0 && timeoutMillis > 0) }
+}
+
+/** Null attempt limit preserves legacy unlimited reconnect behavior. */
+data class ReconnectionOptions(
+    val health: ConnectionHealth? = null,
+    val maxAttempts: Int? = null,
+    val connectTimeoutMillis: Long = 15_000,
+    val retryDelayMillis: Long = 700,
+) {
+    init {
+        require(maxAttempts == null || maxAttempts > 0)
+        require(connectTimeoutMillis > 0 && retryDelayMillis >= 0)
+    }
+}
+
+// Reserved inner command: encrypted with the current session and handled before application dispatch.
+internal val connectionPing = Command("@kiloparsec/ping", Long.serializer(), Long.serializer())

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

@@ -13,6 +13,7 @@ package net.sergeych.kiloparsec
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.isActive
@@ -24,14 +25,13 @@ import net.sergeych.crypto2.VerifyingPublicKey
 import net.sergeych.mp_logger.LogTag
 import net.sergeych.mp_logger.Loggable
 import net.sergeych.mp_logger.debug
-import net.sergeych.mp_logger.exception
 import net.sergeych.mp_tools.globalLaunch
 
 /**
  * The auto-connecting client that reconnects to the kiloparsec server,
  * [KiloServer],
  * and maintain connection state flow. Client factory launches a disconnected
- * set of coroutines to support automatic reconnection, so you _must_ [close]
+ * set of coroutines to support automatic reconnection (unlimited by default; see [options]), so you _must_ [close]
  * it manually when it is unnecessary, otherwise it will continue to reconnect.
  *
  * ## Usage
@@ -59,9 +59,18 @@ import net.sergeych.mp_tools.globalLaunch
 class KiloClient<S>(
     val localInterface: KiloInterface<S>,
     secretKey: SigningKey? = null,
+    val options: ReconnectionOptions = ReconnectionOptions(),
     connectionDataFactory: ConnectionDataFactory<S>,
 ) : RemoteInterface,
     Loggable by LogTag("CLIF") {
+    constructor(localInterface: KiloInterface<S>, secretKey: SigningKey? = null, connectionDataFactory: ConnectionDataFactory<S>)
+        : this(localInterface, secretKey, ReconnectionOptions(), connectionDataFactory)
+
+    private val stopped = MutableStateFlow(false)
+    private val exhausted = MutableStateFlow(false)
+    /** True when the finite attempt budget has ended. Create a new client for another carrier. */
+    val reconnectExhausted = exhausted.asStateFlow()
+
     val _state = MutableStateFlow(false)
 
     /**
@@ -81,49 +90,66 @@ class KiloClient<S>(
 
     private var deferredClient = CompletableDeferred<KiloClientConnection<S>>()
 
+    private val defMutex = Mutex()
+
     private val job =
         globalLaunch {
             debug { "starting connector" }
-            while (isActive) {
+            var attempts = 0
             try {
-                    debug { "getting connection" }
-                    val kc = connectionDataFactory()
-                    debug { "get device and session" }
-                    val client = KiloClientConnection(localInterface, kc, secretKey)
-                    deferredClient.complete(client)
-                    debug { "starting client run"}
-                    val r = runCatching {  client.run {
-                        _state.value = it
-                    } }
-                    debug { "----------- client run finished: $r" }
-                    resetDeferredClient()
-                    debug { "client run finished" }
-                } catch (_: RemoteInterface.ClosedException) {
-                    debug { "remote closed" }
-                    delay(1000)
-                } catch (_: CancellationException) {
-                    debug { "cancelled" }
-                    break
-                } catch (t: Throwable) {
-                    exception { "unexpected exception" to t }
-                    delay(1000)
+                while (isActive && (options.maxAttempts == null || attempts < options.maxAttempts)) {
+                    attempts++
+                    var connectedAt: kotlin.time.TimeMark? = null
+                    try {
+                        val client = kotlinx.coroutines.withTimeout(options.connectTimeoutMillis) {
+                            KiloClientConnection(localInterface, connectionDataFactory(), secretKey)
                         }
+                        deferredClient.complete(client)
+                        kotlinx.coroutines.coroutineScope {
+                            val setupDeadline = launch {
+                                delay(options.connectTimeoutMillis)
+                                throw RemoteInterface.ClosedException()
+                            }
+                            val onState: (Boolean) -> Unit = {
+                                _state.value = it
+                                if (it) {
+                                    connectedAt = kotlin.time.TimeSource.Monotonic.markNow()
+                                    setupDeadline.cancel()
+                                }
+                            }
+                            if (options.health == null) client.run(onState)
+                            else client.run(options.health, onState)
+                            setupDeadline.cancel()
+                        }
+                    } catch (t: kotlinx.coroutines.TimeoutCancellationException) {
+                        debug { "connection setup timed out" }
+                    } catch (t: CancellationException) { throw t }
+                    catch (t: Throwable) { debug { "connection ended: $t" } }
+                    finally {
                         _state.value = false
                         resetDeferredClient()
-                // reconnection timeout
-                delay(700)
+                    }
+                    // Short authenticated-but-dead sessions must still exhaust the budget.
+                    val stableMillis = options.health?.let { 3 * (it.intervalMillis + it.timeoutMillis) } ?: 30_000L
+                    if ((connectedAt?.elapsedNow()?.inWholeMilliseconds ?: 0) > stableMillis) attempts = 0
+                    if (options.maxAttempts == null || attempts < options.maxAttempts) delay(options.retryDelayMillis)
+                }
+                exhausted.value = true
+            } finally {
+                _state.value = false
+                deferredClient.completeExceptionally(RemoteInterface.ClosedException())
             }
         }
 
     fun close() {
+        stopped.value = true
         job.cancel()
         debug { "client is closed" }
     }
 
-    private val defMutex = Mutex()
     private suspend fun resetDeferredClient() {
         defMutex.withLock {
-            if (!deferredClient.isActive) {
+            if (!deferredClient.isActive && !exhausted.value && !stopped.value) {
                 deferredClient = CompletableDeferred()
             }
 
@@ -147,6 +173,9 @@ class KiloClient<S>(
         }
     }
 
+    /** Perform an encrypted liveness round trip on the active connection. */
+    suspend fun ping() = deferredClient.await().ping()
+
     /**
      * Current session token. This is a per-connection unique random value same on the client and server part so
      * it could be used as a nonce to pair MITM and like attacks, be sure that the server is actually
@@ -174,6 +203,8 @@ class KiloClient<S>(
             }
             private var connectionBuilder: (suspend () -> Transport.Device)? = null
 
+            var reconnectionOptions: ReconnectionOptions = ReconnectionOptions()
+
             var secretIdKey: SigningKey? = null
 
             /**
@@ -228,7 +259,7 @@ class KiloClient<S>(
                 for (ep in errorProviders) i.addErrorProvider(ep)
                 interfaceBuilder?.let { i.it() }
                 val connector = connectionBuilder ?: throw IllegalArgumentException("connect handler was not set")
-                return KiloClient(i, secretIdKey) {
+                return KiloClient(i, secretIdKey, reconnectionOptions) {
                     KiloConnectionData(connector(), sessionBuilder())
                 }
             }

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

@@ -43,7 +43,12 @@ class KiloClientConnection<S>(
      * will stop the transport, or the calling scope will be canceled.
      * Cancelling the scope where server is running is a preferred way to stop the client.
      */
-    suspend fun run(onConnectedStateChanged: ((Boolean) -> Unit)? = null) {
+    suspend fun run(onConnectedStateChanged: ((Boolean) -> Unit)? = null) = runConfigured(null, onConnectedStateChanged)
+
+    suspend fun run(health: ConnectionHealth, onConnectedStateChanged: ((Boolean) -> Unit)? = null) =
+        runConfigured(health, onConnectedStateChanged)
+
+    private suspend fun runConfigured(health: ConnectionHealth?, onConnectedStateChanged: ((Boolean) -> Unit)?) {
         coroutineScope {
             var job: Job? = null
             var connectedScope: KiloScope<S>? = null
@@ -94,20 +99,42 @@ class KiloClientConnection<S>(
                 connectedScope = params.scope
                 clientInterface.onConnectHandlers.invokeAll(connectedScope)
                 onConnectedStateChanged?.invoke(true)
-                job.join()
+                val heartbeat = health?.let { options ->
+                    launch {
+                        while (isActive) {
+                            delay(options.intervalMillis)
+                            if (withTimeoutOrNull(options.timeoutMillis) {
+                                withContext(CallPriority(100)) { ping() }; true
+                            } != true) throw RemoteInterface.ClosedException()
+                        }
+                    }
+                }
+                try { job.join() }
+                finally { heartbeat?.cancelAndJoin() }
 
             } catch (x: CancellationException) {
                 info { "client is cancelled" }
             } catch (x: RemoteInterface.ClosedException) {
                 debug { "connection closed/refused by remote" }
             } finally {
+                if (!deferredParams.isCompleted) deferredParams.completeExceptionally(RemoteInterface.ClosedException())
+                if (!kiloRemoteInterface.isCompleted) kiloRemoteInterface.completeExceptionally(RemoteInterface.ClosedException())
                 connectedScope?.let { clientInterface.onDisconnectHandlers.invokeAll(it) }
                 onConnectedStateChanged?.invoke(false)
-                job?.cancel()
+                withContext(NonCancellable) {
+                    job?.cancelAndJoin()
                     device.apply { runCatching { close() } }
                 }
             }
         }
+    }
+
+    /** Encrypted round trip. Older peers' encrypted UnknownCommand also proves liveness. */
+    suspend fun ping() {
+        val nonce = kotlin.random.Random.nextLong()
+        try { check(call(connectionPing, nonce) == nonce) { "Invalid heartbeat response" } }
+        catch (_: RemoteInterface.UnknownCommand) { /* Compatible with peers predating native ping. */ }
+    }
 
     suspend fun token() = deferredParams.await().token
     override suspend fun <A, R> call(cmd: Command<A, R>, args: A): R =

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

@@ -41,7 +41,8 @@ internal class KiloL0Interface<T>(
             val result: Transport.Block = try {
                 Transport.Block.Response(
                     0u,
-                    clientInterface.execute(params.scope, call.name, call.serializedArgs)
+                    if (call.name == connectionPing.name) connectionPing.exec(call.serializedArgs) { it }
+                    else clientInterface.execute(params.scope, call.name, call.serializedArgs)
                 )
             } catch(t: BreakConnectionException) {
                 throw t

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

@@ -11,12 +11,17 @@
 package net.sergeych.kiloparsec
 
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.serialization.builtins.serializer
 import net.sergeych.mp_logger.LogTag
 import net.sergeych.mp_logger.Loggable
 import net.sergeych.utools.unpack
 
 private var L1IdCounter = 0
 
+// Same encrypted L0Call wire command, sent as an existing id=0 notification.
+// The result serializer is unused: neither endpoint sends/decodes a reply to a push.
+private val l0Notification = Command(L0Call.name, L0Call.argsSerializer, Unit.serializer())
+
 /**
  * Represents a remote interface for interacting with a Kiloparsec service.
  *
@@ -48,7 +53,6 @@ class KiloRemoteInterface<S>(
 
     override suspend fun <A> push(cmd: Command<A, Unit>, args: A) {
         val params = deferredParams.await()
-        params.transport.call(L0Call, params.encrypt(cmd.packCall(args)))
+        params.transport.push(l0Notification, params.encrypt(cmd.packCall(args)))
     }
 }
-

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

@@ -72,7 +72,8 @@ open class LocalInterface<S> : Loggable by LogTag("LocalInterface${idCounter.inc
         }
     }
 
-    suspend fun execute(
+    /** Override to gate dispatch before command lookup and argument deserialization. */
+    open suspend fun execute(
         scope: S,
         name: String,
         packedArgs: UByteArray,

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

@@ -68,10 +68,18 @@ class Transport<S>(
         suspend fun close()
 
         suspend fun flush() {}
+
+        /** Adapter admission; legacy devices retain their existing output-channel behavior. */
+        suspend fun send(block: UByteArray) { output.send(block) }
     }
 
     @Serializable(TransportBlockSerializer::class)
     sealed class Block {
+        // Kept outside the legacy generated body serializer. Only the outer custom
+        // serializer appends it, so old field positions and constructors stay intact.
+        @kotlinx.serialization.Transient
+        var priority: Int = 0
+
         @Serializable
         data class Call(
             @Unsigned
@@ -121,37 +129,27 @@ class Transport<S>(
     private suspend fun sendCallBlock(name: String, packedArgs: UByteArray): UByteArray {
         if (isClosed) throw RemoteInterface.ClosedException()
 
-        val b: Block
+        val b: Block.Call
         val deferred = CompletableDeferred<UByteArray>()
 
         // We need to shield calls and lastID with mutex, but nothing more:
         access.withLock {
             if (isClosed) throw RemoteInterface.ClosedException()
             // the order is important: first id in use MUST BE >= 1, not zero:
-            b = Block.Call(++lastId, name, packedArgs)
+            b = Block.Call(++lastId, name, packedArgs).also { it.priority = currentCoroutineContext()[CallPriority]?.value ?: 0 }
             calls[b.id] = deferred
         }
 
-        // now we have mutex freed so we can call:
-        val r = runCatching {
-            do {
-                val cr = device.output.trySend(pack(b))
-                if( cr.isClosed ) throw ClosedSendChannelException("can't send block: channel is closed")
-                delay(100)
-            } while(!cr.isSuccess)
-        }
-        if (!r.isSuccess) {
-            r.exceptionOrNull()?.let {
-                exception { "failed to send output block" to it }
-            } ?: run {
-                error { "It should not happen: empty exception on block send failure" }
-                throw RuntimeException("unexpected failure in sending block")
-            }
-            deferred.completeExceptionally(RemoteInterface.ClosedException())
-        }
-
-        // it returns packed result or throws a proper error:
+        // Suspend on actual channel backpressure; never poll or delay a successful send.
+        try {
+            device.send(pack<Block>(b))
             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) } }
+        }
     }
 
     /**
@@ -164,9 +162,10 @@ class Transport<S>(
         if (isClosed) throw RemoteInterface.ClosedException()
 
         // All push blocks have the same id == 0:
-        val b = Block.Call(0u, name, packedArgs)
-        val r = runCatching { device.output.send(pack(b).also { debug { ">>$\n${it.toDump()}" } }) }
+        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()}" } }) }
         when(val e = r.exceptionOrNull()) {
+            null -> return
             is RemoteInterface.ClosedException, is CancellationException, is RemoteInterface.RemoteException
                 -> throw e
             else -> throw RemoteInterface.ClosedException()
@@ -224,7 +223,7 @@ class Transport<S>(
                                             Block.Response(
                                                 b.id,
                                                 localInterface.execute(commandContext, b.name, b.packedArgs)
-                                            )
+                                            ).also { it.priority = b.priority }
                                         )
                                 } catch (x: LocalInterface.BreakConnectionException) {
                                     // handler forced close
@@ -233,9 +232,9 @@ class Transport<S>(
                                     if (x.flushSendQueue) device.flush()
                                     device.close()
                                 } catch (x: RemoteInterface.RemoteException) {
-                                    send(Block.Error(b.id, x.code, x.text, x.extra))
+                                    send(Block.Error(b.id, x.code, x.text, x.extra).also { it.priority = b.priority })
                                 } catch (t: Throwable) {
-                                    send(Block.Error(b.id, "UnknownError", t.message))
+                                    send(Block.Error(b.id, "UnknownError", t.message).also { it.priority = b.priority })
                                 }
                             }
                         }
@@ -281,7 +280,7 @@ class Transport<S>(
 
     private suspend fun send(block: Block) {
         try {
-            device.output.send(pack(block))
+            device.send(pack(block))
         } catch (_: ClosedSendChannelException) {
             throw RemoteInterface.ClosedException()
         }
@@ -309,14 +308,26 @@ object TransportBlockSerializer : KSerializer<Transport.Block> {
                 encoder.encodeSerializableValue(serializer<Transport.Block.Response>(), value)
             }
         }
+        // Fixed-size trailer is omitted for zero, preserving exact legacy bytes.
+        if (value.priority != 0) {
+            for (shift in 24 downTo 0 step 8) encoder.encodeByte((value.priority ushr shift).toByte())
+        }
     }
 
 
-    override fun deserialize(decoder: Decoder): Transport.Block =
-        when (val id = decoder.decodeByte().toInt()) {
+    override fun deserialize(decoder: Decoder): Transport.Block {
+        val block = when (val id = decoder.decodeByte().toInt()) {
             0 -> decoder.decodeSerializableValue(serializer<Transport.Block.Call>())
             1 -> decoder.decodeSerializableValue(serializer<Transport.Block.Error>())
             2 -> decoder.decodeSerializableValue(serializer<Transport.Block.Response>())
             else -> throw RemoteInterface.InvalidDataException("wrong block type: $id")
         }
+        val first = try { decoder.decodeByte().toInt() and 255 }
+        catch (_: net.sergeych.bintools.DataSource.EndOfData) { return block }
+        // A present but truncated trailer is malformed, not a legacy priority-zero block.
+        var priority = first
+        repeat(3) { priority = (priority shl 8) or (decoder.decodeByte().toInt() and 255) }
+        block.priority = priority
+        return block
+    }
 }

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

@@ -0,0 +1,113 @@
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import net.sergeych.bintools.DataSource
+import net.sergeych.kiloparsec.RemoteInterface
+
+/** Reads only headers and skips length-delimited payloads; never copies or decrypts payload data. */
+internal fun blockPriority(bytes: UByteArray): Int {
+    val source = object : DataSource {
+        var position = 0
+        override fun readByte(): Byte {
+            if (position >= bytes.size) throw DataSource.EndOfData()
+            return bytes[position++].toByte()
+        }
+        fun skipBlob() {
+            val size = readSmartUInt().toLong()
+            require(size <= bytes.size - position)
+            position += size.toInt()
+        }
+    }
+    return try {
+        val tag = source.readUByte().toInt()
+        source.readSmartUInt() // call ID or response ID
+        when (tag) {
+            0 -> { source.skipBlob(); source.skipBlob() }
+            1 -> {
+                source.skipBlob()
+                if (source.readByte() != 0.toByte()) source.skipBlob()
+                if (source.readByte() != 0.toByte()) source.skipBlob()
+            }
+            2 -> source.skipBlob()
+            else -> return 0 // Opaque legacy device users need not send RPC blocks.
+        }
+        if (bytes.size - source.position != 4) return 0
+        var result = 0
+        repeat(4) { result = (result shl 8) or source.readUByte().toInt() }
+        result
+    } catch (_: Exception) { 0 }
+}
+
+/**
+ * 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;
+ * 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 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 var closed = false
+    private var writing: Pending? = null
+    private val job = scope.launch {
+        var current: Pending? = null
+        try {
+            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 }
+                }
+                val entry = current
+                if (entry == null) { wake.receive(); continue }
+                if (entry.done.isActive) write(entry.bytes)
+                entry.done.complete(Unit)
+                mutex.withLock { writing = null }
+                current = null
+            }
+        } catch (failure: Throwable) {
+            current?.done?.completeExceptionally(failure)
+            throw failure
+        } finally {
+            withContext(NonCancellable) {
+                mutex.withLock {
+                    closed = true
+                    pending.forEach { it.done.completeExceptionally(RemoteInterface.ClosedException()) }
+                    pending.clear()
+                }
+            }
+        }
+    }
+
+    suspend fun send(bytes: UByteArray) {
+        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 })
+                throw RemoteInterface.Exception("Outgoing block queue is full")
+            pending.add(entry)
+        }
+        wake.trySend(Unit)
+        try { entry.done.await() }
+        finally {
+            entry.done.cancel()
+            withContext(NonCancellable) { mutex.withLock { pending.remove(entry) } }
+        }
+    }
+    suspend fun flush() {
+        val queued = mutex.withLock { (pending + listOfNotNull(writing)).map { it.done } }
+        queued.forEach { it.await() }
+    }
+    fun close() { job.cancel() }
+}

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

@@ -25,16 +25,26 @@ open class ProxyDevice(
     private val doFlush: (suspend ()->Unit)? = null,
 ): Transport.Device {
 
+    internal var prioritizedSend: (suspend (UByteArray) -> Unit)? = null
+    internal var priorityWriter: PriorityBlockWriter? = null
+    override suspend fun send(block: UByteArray) {
+        prioritizedSend?.let { it(block); return }
+        val writer = priorityWriter
+        if (writer == null) output.send(block) else writer.send(block)
+    }
+
     override val input: ReceiveChannel<UByteArray> = inputChannel
     override val output: SendChannel<UByteArray> = outputChannel
 
     override suspend fun close() {
+        priorityWriter?.close()
         kotlin.runCatching {  doClose?.invoke() }
         runCatching { inputChannel.close() }
         runCatching { outputChannel.close() }
     }
 
     override suspend fun flush() {
+        priorityWriter?.flush()
         doFlush?.invoke()
         var cnt = 10
         while(!outputChannel.isEmpty) {

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

@@ -78,10 +78,10 @@ fun websocketTransportDevice(
 
     val input = Channel<UByteArray>()
     val output = Channel<UByteArray>()
+    val writerReady = CompletableDeferred<PriorityBlockWriter>()
     val closeHandle = CompletableDeferred<Boolean>()
-    val readyHandle = CompletableDeferred<Unit>()
 
-    globalLaunch {
+    val connectionJob = globalLaunch {
         val log = LogTag("KC:${counter.incrementAndGet()}")
         try {
             actualClient.webSocket({
@@ -95,17 +95,15 @@ fun websocketTransportDevice(
                 log.info { "connected to the server" }
 //                println("SENDING!!!")
 //                send("Helluva")
-                readyHandle.complete(Unit)
+                val writer = PriorityBlockWriter(this) { block ->
+                    if (useTextFrames) send(Frame.Text(block.asByteArray().encodeToBase64Compact()))
+                    else send(block.asByteArray())
+                    flush() // Do not turn the WebSocket outgoing FIFO into an unbounded look-ahead.
+                }
+                writerReady.complete(writer)
                 launch {
                     try {
-                        for (block in output) {
-                            if (useTextFrames)
-                                send(
-                                    Frame.Text(block.asByteArray().encodeToBase64Compact())
-                                )
-                            else
-                                send(block.toByteArray())
-                        }
+                        for (block in output) writer.send(block)
                         log.info { "input is closed, closing the websocket" }
                         if (closeHandle.isActive) closeHandle.complete(true)
                     } catch (_: ClosedSendChannelException) {
@@ -115,6 +113,7 @@ fun websocketTransportDevice(
                         log.info { "unexpected exception in websock sender: ${t.stackTraceToString()}" }
                         closeHandle.completeExceptionally(t)
                     }
+                    writer.close()
                     if (closeHandle.isActive) closeHandle.complete(false)
                 }
                 launch {
@@ -151,6 +150,7 @@ fun websocketTransportDevice(
             runCatching { output.close() }
             runCatching { input.close() }
         } finally {
+            writerReady.completeExceptionally(RemoteInterface.ClosedException())
             if (ownsClient) actualClient.close()
         }
         log.info { "closing connection" }
@@ -162,7 +162,13 @@ fun websocketTransportDevice(
         runCatching { output.close() }
         runCatching { input.close() }
         closeHandle.complete(true)
+        connectionJob.cancel()
 //            job.cancel()
     })
+    device.prioritizedSend = { block ->
+        val writer = writerReady.await()
+        device.priorityWriter = writer
+        writer.send(block)
+    }
     return device
 }

src/commonTest/kotlin/PriorityRpcTest.kt

@@ -0,0 +1,49 @@
+@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.unpack
+import kotlin.test.*
+
+class PriorityRpcTest {
+    @Test fun encryptedConcurrentRepliesAndErrorsInheritPriority() = runTest {
+        initCrypto()
+        val echo by command<Int, Int>()
+        val wait by command<Int, Int>()
+        val fail by command<Int, Int>()
+        val release = CompletableDeferred<Unit>()
+        val entered = CompletableDeferred<Unit>()
+        val ready = CompletableDeferred<Unit>()
+        val requests = mutableListOf<Transport.Block>()
+        val replies = mutableListOf<Transport.Block>()
+        val (a, b) = createTestDevice()
+        fun recording(device: Transport.Device, blocks: MutableList<Transport.Block>) = object : Transport.Device by device {
+            override suspend fun send(block: UByteArray) {
+                blocks += unpack<Transport.Block>(block)
+                device.send(block)
+            }
+        }
+        val server = KiloServerConnection(KiloInterface<Unit>().apply {
+            on(echo) { it }
+            on(wait) { entered.complete(Unit); release.await(); it }
+            on(fail) { throw RemoteInterface.RemoteException("test-error") }
+        }, recording(b, replies), Unit)
+        val client = KiloClientConnection(KiloInterface<Unit>(), recording(a, requests), Unit)
+        val jobs = listOf(launch { server.run() }, launch { client.run { if (it) ready.complete(Unit) } })
+        try {
+            ready.await()
+            val slow = async { client.call(wait, 1, priority = 7) }
+            entered.await()
+            assertEquals(2, client.call(echo, 2, priority = 123))
+            assertFalse(slow.isCompleted, "A pending reply must not prevent another concurrent call")
+            assertFailsWith<RemoteInterface.RemoteException> { client.call(fail, 3, priority = 81) }
+            release.complete(Unit); assertEquals(1, slow.await())
+            for (priority in listOf(7, 123, 81)) {
+                val request = requests.filterIsInstance<Transport.Block.Call>().single { it.priority == priority }
+                val response = replies.filterIsInstance<Transport.Block.Response>().single { it.forId == request.id }
+                assertEquals(priority, response.priority)
+            }
+        } finally { release.complete(Unit); jobs.forEach { it.cancelAndJoin() } }
+    }
+}

src/commonTest/kotlin/TransportAdmissionTest.kt

@@ -0,0 +1,65 @@
+@file:OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class, ExperimentalUnsignedTypes::class)
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import kotlin.test.*
+import kotlin.time.Duration.Companion.seconds
+
+class TransportAdmissionTest {
+    @Test fun readyCallsDoNotAddPollingDelay() = runTest(timeout = 10.seconds) {
+        val ping by command<Int, Int>()
+        val (a, b) = createTestDevice()
+        val local = LocalInterface<Unit>().apply { on(ping) { it } }
+        val client = Transport(a, local, Unit)
+        val server = Transport(b, local, Unit)
+        val jobs = listOf(launch { client.run() }, launch { server.run() })
+        try {
+            val start = currentTime
+            repeat(20) { assertEquals(it, client.call(ping, it)) }
+            assertEquals(0L, currentTime - start, "Ready transports must not impose an artificial per-call delay")
+        } finally { jobs.forEach { it.cancelAndJoin() } }
+    }
+
+    @Test fun successfulPlainPushReturnsNormally() = runTest(timeout = 10.seconds) {
+        val notify by command<Int, Unit>()
+        val received = CompletableDeferred<Int>()
+        val (a, b) = createTestDevice()
+        val client = Transport(a, LocalInterface<Unit>(), Unit)
+        val server = Transport(b, LocalInterface<Unit>().apply { on(notify) { received.complete(it); Unit } }, Unit)
+        val jobs = listOf(launch { client.run() }, launch { server.run() })
+        try {
+            client.push(notify, 7)
+            assertEquals(7, received.await())
+        } finally { jobs.forEach { it.cancelAndJoin() } }
+    }
+
+    @Test fun encryptedPushDoesNotWaitForRemoteHandler() = runTest(timeout = 10.seconds) {
+        initCrypto()
+        val notify by command<Int, Unit>()
+        val ping by command<Int, Int>()
+        val entered = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        val ready = CompletableDeferred<Unit>()
+        val (a, b) = createTestDevice()
+        val server = KiloServerConnection(KiloInterface<Unit>().apply {
+            on(notify) { entered.complete(Unit); release.await() }
+            on(ping) { it }
+        }, b, Unit)
+        val client = KiloClientConnection(KiloInterface<Unit>(), a, Unit)
+        val jobs = listOf(launch { server.run() }, launch { client.run { if (it) ready.complete(Unit) } })
+        try {
+            ready.await()
+            val notification = async { client.push(notify, 7) }
+            entered.await()
+            runCurrent()
+            assertTrue(notification.isCompleted, "Push must return before the remote handler finishes")
+            notification.await()
+            assertEquals(9, client.call(ping, 9), "A pending notification must not block ordinary RPC")
+        } finally {
+            release.complete(Unit)
+            jobs.forEach { it.cancelAndJoin() }
+        }
+    }
+}

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

@@ -0,0 +1,59 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.test.*
+import net.sergeych.crypto2.initCrypto
+import net.sergeych.kiloparsec.*
+import kotlin.math.ceil
+import kotlin.test.*
+
+class PriorityLinkTest {
+    @Test fun saturatedEncryptedLinkKeepsConcurrentControlResponsiveWithoutRateCap() = runTest {
+        initCrypto()
+        val bulk by command<UByteArray, Unit>()
+        val ping by command<Int, Int>()
+        val received = Channel<UByteArray>(Channel.UNLIMITED)
+        val aInput = Channel<UByteArray>(Channel.UNLIMITED)
+        val bInput = Channel<UByteArray>(Channel.UNLIMITED)
+        var sentBytes = 0L
+        val rate = 16_384
+        fun endpoint(input: Channel<UByteArray>, destination: Channel<UByteArray>): Transport.Device {
+            val writer = PriorityBlockWriter(backgroundScope) { block ->
+                delay(ceil(block.size * 1000.0 / rate).toLong())
+                sentBytes += block.size
+                destination.send(block)
+            }
+            return ProxyDevice(input, Channel(), doClose = { writer.close(); input.close() }).also { it.priorityWriter = writer }
+        }
+        val a = endpoint(aInput, bInput)
+        val b = endpoint(bInput, aInput)
+        val ready = CompletableDeferred<Unit>()
+        val api = KiloInterface<Unit>().apply {
+            on(bulk) { received.send(it) }
+            on(ping) { it }
+        }
+        val server = KiloServerConnection(api, b, Unit)
+        val client = KiloClientConnection(api, a, Unit)
+        val jobs = listOf(backgroundScope.launch { server.run() }, backgroundScope.launch { client.run { if (it) ready.complete(Unit) } })
+        try {
+            ready.await()
+            val start = currentTime
+            val initialBytes = sentBytes
+            val producers = listOf<RemoteInterface>(client, server).flatMap { sender ->
+                List(8) { backgroundScope.launch { while (isActive) sender.push(bulk, UByteArray(2000), priority = 0) } }
+            }
+            received.receive()
+            repeat(10) {
+                val began = currentTime
+                assertEquals(it, client.call(ping, it, priority = 100))
+                assertTrue(currentTime - began < 400, "control latency=${currentTime - began}")
+                delay(80)
+            }
+            val elapsed = currentTime - start
+            assertTrue((sentBytes - initialBytes) * 1000 / elapsed > rate * 1.6, "Writer must use available capacity")
+            producers.forEach { it.cancelAndJoin() }
+        } finally { jobs.forEach { it.cancelAndJoin() }; a.close(); b.close() }
+    }
+}

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

@@ -0,0 +1,85 @@
+@file:OptIn(ExperimentalUnsignedTypes::class, kotlinx.coroutines.ExperimentalCoroutinesApi::class)
+package net.sergeych.kiloparsec.adapter
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.test.*
+import kotlinx.serialization.serializer
+import net.sergeych.bintools.DataSource
+import net.sergeych.bipack.BipackDecoder
+import net.sergeych.bipack.BipackEncoder
+import net.sergeych.kiloparsec.*
+import net.sergeych.utools.pack
+import net.sergeych.utools.unpack
+import kotlin.test.*
+
+class PriorityTest {
+    private fun block(priority: Int, id: Int = 1): UByteArray = pack<Transport.Block>(
+        Transport.Block.Call(id.toUInt(), "example", ubyteArrayOf(1u, 2u)).also { it.priority = priority })
+
+    @Test fun legacyBodyAndTrailerCompatibility() {
+        for (original in listOf(
+            Transport.Block.Call(123456u, "test", UByteArray(3500) { it.toUByte() }),
+            Transport.Block.Response(123456u, UByteArray(256)),
+            Transport.Block.Error(123456u, "code", "message", UByteArray(180)),
+            Transport.Block.Error(0u, "code", null, null),
+        )) {
+            val legacy = when (original) {
+                is Transport.Block.Call -> ubyteArrayOf(0u) + BipackEncoder.encode(original).asUByteArray()
+                is Transport.Block.Error -> ubyteArrayOf(1u) + BipackEncoder.encode(original).asUByteArray()
+                is Transport.Block.Response -> ubyteArrayOf(2u) + BipackEncoder.encode(original).asUByteArray()
+            }
+            assertContentEquals(legacy, pack<Transport.Block>(original))
+            assertEquals(0, unpack<Transport.Block>(legacy).priority)
+            for (priority in listOf(Int.MIN_VALUE, -1, 0, 1, 100, Int.MAX_VALUE)) {
+                original.priority = priority
+                val packed = pack<Transport.Block>(original)
+                assertEquals(priority, unpack<Transport.Block>(packed).priority)
+                assertEquals(priority, blockPriority(packed))
+                assertContentEquals(legacy, packed.copyOf(legacy.size))
+                // Legacy decoding invokes the original body serializer and ignores trailing bytes.
+                val oldRead = when (original) {
+                    is Transport.Block.Call -> BipackDecoder.decode<Transport.Block.Call>(packed.drop(1).toUByteArray())
+                    is Transport.Block.Response -> BipackDecoder.decode<Transport.Block.Response>(packed.drop(1).toUByteArray())
+                    is Transport.Block.Error -> BipackDecoder.decode<Transport.Block.Error>(packed.drop(1).toUByteArray())
+                }
+                assertEquals(0, oldRead.priority)
+            }
+            assertFailsWith<DataSource.EndOfData> { unpack<Transport.Block>(legacy + ubyteArrayOf(1u)) }
+        }
+    }
+
+    @Test fun numericOrderingStableTiesAndCancellation() = runTest {
+        val release = CompletableDeferred<Unit>()
+        val started = CompletableDeferred<Unit>()
+        val sent = mutableListOf<Pair<Int, UInt>>()
+        val writer = PriorityBlockWriter(backgroundScope) { bytes ->
+            val b = unpack<Transport.Block>(bytes) as Transport.Block.Call
+            if (b.id == 1u) { started.complete(Unit); release.await() }
+            sent += b.priority to b.id
+        }
+        val first = async { writer.send(block(0, 1)) }
+        started.await()
+        val queued = listOf(0 to 2, 100 to 3, 2 to 4, 100 to 5, -10 to 6).map { (p, id) ->
+            async { writer.send(block(p, id)) }
+        }
+        val cancelled = async { writer.send(block(Int.MAX_VALUE, 7)) }
+        runCurrent(); cancelled.cancelAndJoin()
+        release.complete(Unit)
+        first.await(); queued.awaitAll()
+        assertEquals(listOf(0 to 1u, 100 to 3u, 100 to 5u, 2 to 4u, 0 to 2u, -10 to 6u), sent)
+        writer.close()
+    }
+
+    @Test fun boundedAdmissionAndCloseReleaseWaiters() = runTest {
+        val started = CompletableDeferred<Unit>()
+        val writer = PriorityBlockWriter(backgroundScope, maxQueuedItems = 1) { started.complete(Unit); awaitCancellation() }
+        val sending = async { runCatching { writer.send(block(0)) } }
+        started.await()
+        val queued = async { runCatching { writer.send(block(2)) } }
+        runCurrent()
+        assertFailsWith<RemoteInterface.Exception> { writer.send(block(100)) }
+        writer.close()
+        assertIs<CancellationException>(sending.await().exceptionOrNull())
+        assertIs<RemoteInterface.ClosedException>(queued.await().exceptionOrNull())
+    }
+}

src/jvmMain/kotlin/net/sergeych/kiloparsec/adapter/WebsocketServer.kt

@@ -54,12 +54,22 @@ fun <S> Application.setupWebsocketServer(
     path: String = "/kp",
     serverKey: SigningKey? = null,
     createSession: () -> S,
+) = setupWebsocketServer(localInterface, path, serverKey, {}, createSession)
+
+/** Configure transport keepalive without changing the legacy defaults for other applications. */
+fun <S> Application.setupWebsocketServer(
+    localInterface: KiloInterface<S>,
+    path: String = "/kp",
+    serverKey: SigningKey? = null,
+    configureWebSockets: WebSockets.WebSocketOptions.() -> Unit,
+    createSession: () -> S,
 ) {
     install(WebSockets) {
         pingPeriod = 60.seconds //Duration.ofSeconds(15)
         timeout = 45.seconds
         maxFrameSize = Long.MAX_VALUE
         masking = false
+        configureWebSockets()
     }
     val counter = AtomicCounter()
     routing {
@@ -67,28 +77,20 @@ fun <S> Application.setupWebsocketServer(
             val log = LogTag("KWS:${counter.incrementAndGet()}")
             log.debug { "opening the connection" }
             val input = Channel<UByteArray>(256)
-            val output = Channel<UByteArray>(256)
+            val output = Channel<UByteArray>()
             var useBinary: Boolean? = null
+            val writer = PriorityBlockWriter(this) { block ->
+                if (useBinary == false) send(block.asByteArray().encodeToBase64Compact())
+                else send(block.asByteArray())
+                flush()
+            }
             launch {
-                log.debug { "starting output pump" }
-                while (isActive) {
-                    try {
-                        val block = output.receive()
-                        if (useBinary == false)
-                            send(block.asByteArray().encodeToBase64Compact())
-                        else
-                            send(block.toByteArray())
-
-                    } catch (_: ClosedReceiveChannelException) {
-                        log.debug { "closing output pump as output channel is closed" }
-                        break
-                    }
-                }
-                log.debug { "closing output pump" }
+                try { for (block in output) writer.send(block) }
+                finally { writer.close() }
             }
             val server = KiloServerConnection(
                 localInterface,
-                ProxyDevice(input, output),
+                ProxyDevice(input, output).also { it.priorityWriter = writer },
                 createSession(),
                 serverKey
             )

src/jvmTest/kotlin/net/sergeych/kiloparsec/ConnectionHealthTest.kt

@@ -0,0 +1,95 @@
+package net.sergeych.kiloparsec
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.first
+import net.sergeych.crypto2.initCrypto
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.*
+
+class ConnectionHealthTest {
+    @Test(timeout = 15_000)
+    fun silentHandshakesExhaustBudgetAndReleaseWaitingCalls() = runBlocking {
+        initCrypto()
+        val attempts = AtomicInteger()
+        val closes = AtomicInteger()
+        val client = KiloClient(KiloInterface<Unit>(), options = ReconnectionOptions(
+            health = ConnectionHealth(50, 50), maxAttempts = 3,
+            connectTimeoutMillis = 100, retryDelayMillis = 10,
+        )) {
+            attempts.incrementAndGet()
+            KiloConnectionData(object : Transport.Device {
+                val closed = java.util.concurrent.atomic.AtomicBoolean()
+                override val input = Channel<UByteArray>(16)
+                override val output = Channel<UByteArray>(16)
+                override suspend fun close() { if (closed.compareAndSet(false, true)) closes.incrementAndGet(); input.cancel(); output.cancel() }
+            }, Unit)
+        }
+        try {
+            val waiter = async { runCatching { client.token() } }
+            withTimeout(10_000) { client.reconnectExhausted.first { it } }
+            assertEquals(3, attempts.get())
+            assertEquals(3, closes.get())
+            assertFalse(client.connectedStateFlow.value)
+            assertTrue(withTimeout(1_000) { waiter.await() }.isFailure)
+            val missing by command<Unit, Unit>()
+            repeat(2) {
+                withTimeout(1_000) {
+                    assertFailsWith<RemoteInterface.ClosedException> { client.call(missing, Unit) }
+                }
+            }
+            delay(200)
+            assertEquals(3, attempts.get(), "An exhausted client must not reconnect forever")
+        } finally { client.close() }
+    }
+    @Test(timeout = 15_000)
+    fun establishedSilentLossTriggersHeartbeatAndBoundedRecovery() = runBlocking {
+        initCrypto()
+        val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        val blocked = java.util.concurrent.atomic.AtomicBoolean()
+        val attempts = AtomicInteger()
+        val api = object : KiloInterface<Unit>() {
+            override suspend fun execute(scope: KiloScope<Unit>, name: String, packedArgs: UByteArray): UByteArray {
+                error("Heartbeat must bypass application command dispatch")
+            }
+        }
+        val client = KiloClient(KiloInterface<Unit>(), options = ReconnectionOptions(
+            health = ConnectionHealth(100, 100), maxAttempts = 3,
+            connectTimeoutMillis = 300, retryDelayMillis = 10,
+        )) {
+            attempts.incrementAndGet()
+            val requests = Channel<UByteArray>(16)
+            val forwarded = Channel<UByteArray>(16)
+            val responses = Channel<UByteArray>(16)
+            val pump = scope.launch { for (block in requests) if (!blocked.get()) forwarded.send(block) }
+            suspend fun retire() { pump.cancel(); requests.cancel(); forwarded.cancel(); responses.cancel() }
+            val serverDevice = object : Transport.Device {
+                override val input = forwarded
+                override val output = responses
+                override suspend fun close() { retire() }
+            }
+            scope.launch { runCatching { KiloServerConnection(api, serverDevice, Unit).run() } }
+            KiloConnectionData(object : Transport.Device {
+                override val input = responses
+                override val output = requests
+                override suspend fun close() { retire() }
+            }, Unit)
+        }
+        try {
+            withTimeout(5_000) { client.connectedStateFlow.first { it } }
+            // Several successful native encrypted heartbeats; no application RPC is exported.
+            client.ping()
+            delay(800)
+            assertTrue(client.connectedStateFlow.value)
+            assertEquals(1, attempts.get())
+            blocked.set(true)
+            withTimeout(5_000) { client.reconnectExhausted.first { it } }
+            // The healthy session earns a new budget: three failed recovery connections.
+            assertEquals(4, attempts.get())
+            assertFalse(client.connectedStateFlow.value)
+            delay(400)
+            assertEquals(4, attempts.get())
+        } finally { client.close(); scope.coroutineContext[Job]!!.cancelAndJoin() }
+    }
+
+}

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

@@ -0,0 +1,95 @@
+package net.sergeych.kiloparsec.adapter
+
+import java.net.DatagramPacket
+import java.net.DatagramSocket
+import java.net.InetAddress
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.Channel
+import kotlin.test.*
+
+class SocketLifecycleTest {
+    @Test fun tcpRoundTripAndRebind() = runBlocking {
+        withTimeout(10_000) {
+            val server = TcpServer(0, "127.0.0.1")
+            val port = server.boundPort()
+            val accepted = Channel<InetTransportDevice>(1)
+            val collector = launch { runCatching { server.transportFlow.collect { accepted.send(it) } } }
+            val client = connectTcpDevice(NetworkAddress("127.0.0.1", port))
+            val peer = accepted.receive()
+            try {
+                val payload = ubyteArrayOf(2u, 3u, 4u)
+                client.output.send(payload)
+                assertContentEquals(payload, peer.input.receive())
+                peer.output.send(payload)
+                assertContentEquals(payload, client.input.receive())
+            } finally { client.close(); server.close(); collector.cancelAndJoin() }
+            val replacement = TcpServer(port, "127.0.0.1")
+            try { assertEquals(port, replacement.boundPort()) } finally { replacement.close() }
+        }
+    }
+
+    @Test fun udpMalformedPacketDoesNotStopListenerAndRebind() = runBlocking {
+        withTimeout(10_000) {
+            val server = UdpServer(0, "127.0.0.1")
+            val port = server.boundPort()
+            val accepted = Channel<InetTransportDevice>(1)
+            val collector = launch { server.transportFlow.collect { accepted.send(it) } }
+            DatagramSocket().use { socket ->
+                socket.send(DatagramPacket(byteArrayOf(1), 1, InetAddress.getLoopbackAddress(), port))
+            }
+            val client = connectUdpDevice(NetworkAddress("127.0.0.1", port))
+            try {
+                val payload = ubyteArrayOf(2u, 3u, 4u)
+                client.output.send(payload)
+                val peer = accepted.receive()
+                assertContentEquals(payload, peer.input.receive())
+                peer.output.send(payload)
+                assertContentEquals(payload, client.input.receive())
+            } finally { client.close(); server.close(); collector.cancelAndJoin() }
+            val replacement = UdpServer(port, "127.0.0.1")
+            try { assertEquals(port, replacement.boundPort()) } finally { replacement.close() }
+        }
+    }
+
+    @Test fun deviceCloseWorksInCancelledContext() = runBlocking {
+        withTimeout(10_000) {
+            val tcp = TcpServer(0, "127.0.0.1")
+            val udp = UdpServer(0, "127.0.0.1")
+            val peers = Channel<InetTransportDevice>(2)
+            val tcpCollector = launch { runCatching { tcp.transportFlow.collect { peers.send(it) } } }
+            val udpCollector = launch { udp.transportFlow.collect { peers.send(it) } }
+            val tcpClient = connectTcpDevice(NetworkAddress("127.0.0.1", tcp.boundPort()))
+            val udpClient = connectUdpDevice(NetworkAddress("127.0.0.1", udp.boundPort()))
+            udpClient.output.send(ubyteArrayOf(2u, 3u))
+            val firstPeer = peers.receive()
+            val secondPeer = peers.receive()
+            try {
+                val closer = launch(start = CoroutineStart.UNDISPATCHED) {
+                    currentCoroutineContext().cancel()
+                    tcpClient.close()
+                    udpClient.close()
+                }
+                closer.join()
+                // EOF reaches both peers even when caller cancellation precedes close.
+                for (peer in listOf(firstPeer, secondPeer)) {
+                    while (peer.input.receiveCatching().isSuccess) { }
+                }
+            } finally {
+                tcpClient.close(); udpClient.close(); tcp.close(); udp.close()
+                tcpCollector.cancelAndJoin(); udpCollector.cancelAndJoin()
+            }
+        }
+    }
+
+    @Test fun bindFailuresAreReported() = runBlocking {
+        val tcp = TcpServer(0, "127.0.0.1")
+        val udp = UdpServer(0, "127.0.0.1")
+        try {
+            val duplicateTcp = TcpServer(tcp.boundPort(), "127.0.0.1")
+            try { assertFails { duplicateTcp.boundPort() } } finally { duplicateTcp.close() }
+            val duplicateUdp = UdpServer(udp.boundPort(), "127.0.0.1")
+            try { assertFails { duplicateUdp.boundPort() } } finally { duplicateUdp.close() }
+        } finally { tcp.close(); udp.close() }
+        Unit
+    }
+}

src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpProvider.kt

@@ -40,18 +40,11 @@ internal val PING_INACTIVITY_TIME = 30.seconds
  * Listen for incoming TCP/IP connections on all local interfaces and the specified [port]
  * anc create flow of [InetTransportDevice] suitable for [KiloClient].
  */
-fun acceptTcpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> {
-    val selectorManager = SelectorManager(Dispatchers.IO)
-    return flow {
-        val serverSocket = aSocket(selectorManager).tcp().bind(localInterface, port)
-        while (true) {
-            serverSocket.accept().let { sock ->
-                emit(inetTransportDevice(sock, "srv"))
+fun acceptTcpDevice(port: Int, localInterface: String = "0.0.0.0"): Flow<InetTransportDevice> = flow {
+    val server = TcpServer(port, localInterface)
+    try { server.transportFlow.collect { emit(it) } }
+    finally { withContext(NonCancellable) { server.close() } }
 }
-        }
-    }
-}
-
 
 /**
  * Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
@@ -64,10 +57,49 @@ suspend fun connectTcpDevice(hostPortAddress: String): InetTransportDevice =
  * Connect to the TCP/IP server (see [KiloServer]) at the specified address and provide th compatible
  * [InetTransportDevice] to use with [KiloClient].
  */
-suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice {
+suspend fun connectTcpDevice(address: NetworkAddress): InetTransportDevice = withContext(Dispatchers.IO) {
     val selectorManager = SelectorManager(Dispatchers.IO)
+    try {
         val socket = aSocket(selectorManager).tcp().connect(address.host, address.port)
-    return inetTransportDevice(socket)
+        return@withContext inetTransportDevice(socket, onClose = { selectorManager.close() })
+    } catch (failure: Throwable) {
+        selectorManager.close()
+        throw failure
+    }
+}
+
+/** Establish the existing TCP carrier through a caller-selected HTTP CONNECT relay. */
+suspend fun connectTcpDeviceThroughHttpProxy(
+    address: NetworkAddress,
+    proxy: NetworkAddress,
+    authorization: String,
+): InetTransportDevice = withContext(Dispatchers.IO) {
+    require(!authorization.contains('\r') && !authorization.contains('\n'))
+    require(!address.host.any { it.isWhitespace() || it == '\r' || it == '\n' })
+    val selector = SelectorManager(Dispatchers.IO)
+    var socket: Socket? = null
+    try {
+        socket = aSocket(selector).tcp().connect(proxy.host, proxy.port)
+        val input = socket.openReadChannel()
+        val output = socket.openWriteChannel()
+        val target = "${address.host}:${address.port}"
+        output.writeStringUtf8("CONNECT $target HTTP/1.1\r\nHost: $target\r\nProxy-Authorization: $authorization\r\n\r\n")
+        output.flush()
+        val status = input.readUTF8Line(4096) ?: error("Proxy closed before response")
+        require(status.startsWith("HTTP/1.1 200 ") || status.startsWith("HTTP/1.0 200 ")) { "CONNECT refused" }
+        var headerBytes = status.length
+        while (true) {
+            val line = input.readUTF8Line(4096) ?: error("Incomplete proxy response")
+            headerBytes += line.length
+            require(headerBytes <= 16384) { "Proxy headers too large" }
+            if (line.isEmpty()) break
+        }
+        return@withContext inetTransportDevice(socket, onClose = { selector.close() }, providedInput = input, providedOutput = output)
+    } catch (failure: Throwable) {
+        socket?.close()
+        selector.close()
+        throw failure
+    }
 }
 
 /**
@@ -78,147 +110,84 @@ fun String.toNetworkAddress(): NetworkAddress {
     return NetworkAddress(host, port.toInt())
 }
 
-private fun inetTransportDevice(
+internal fun inetTransportDevice(
     sock: Socket,
     suffix: String = "cli",
+    providedInput: ByteReadChannel? = null,
+    providedOutput: ByteWriteChannel? = null,
+    onClose: suspend () -> Unit = {},
 ): InetTransportDevice {
     val networkAddress = (sock.remoteAddress as InetSocketAddress).let { NetworkAddress(it.hostname, it.port) }
     val inputBlocks = Channel<UByteArray>(4096)
-    val outputBlocks = Channel<UByteArray>(4096)
-
-    val log = LogTag("TCPT${logCounter.incrementAndGet()}:$suffix:$networkAddress")
-    val job = AtomicAsyncValue<Job?>(null)
-
-    val sockOutput = sock.openWriteChannel()
-    val sockInput = runCatching { sock.openReadChannel() }.getOrElse {
-        log.warning { "failed to open read channel $it" }
-        throw IllegalStateException("failed to open read channel")
-    }
-
-    suspend fun stop() {
-        job.mutate {
-            if (it != null) {
-                log.debug { "stopping" }
-                runCatching { inputBlocks.close() }
-                runCatching { outputBlocks.close() }
-                // The problem: on mac platofrms closing the socket does not close its input
-                // and output channels!
-                runCatching { sockInput.cancel() }
-                runCatching { sockOutput.flushAndClose() }
-                if (!sock.isClosed)
-                    runCatching {
-                        log.debug { "closing socket by stop" }
+    // A push must backpressure on the writer, not build thousands of blocks ahead
+    // of latency-sensitive calls sharing this carrier.
+    val outputBlocks = Channel<UByteArray>(Channel.RENDEZVOUS)
+    val sockOutput = providedOutput ?: sock.openWriteChannel()
+    val sockInput = providedInput ?: sock.openReadChannel()
+    val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    val closeAccess = Mutex()
+    var closed = false
+    suspend fun stop() = withContext(NonCancellable) {
+        closeAccess.withLock {
+            if (!closed) {
+                closed = true
+                inputBlocks.close()
+                outputBlocks.close()
+                sockInput.cancel()
                 sock.close()
-                    }
-                else
-                    log.debug { "socket is already closed when stop is called" }
-                it.cancel()
-                log.debug { "implementation job cancel called" }
-            } else
-                log.debug { "already stopped" }
-            null
+                scope.cancel()
+                onClose()
+            }
         }
     }
-
     var lastActiveAt = Clock.System.now()
-    globalLaunch {
-        job.reset(globalLaunch {
-            launch {
-
-                log.debug { "opening read channel" }
-
-                while (isActive && sock.isActive) {
+    val writeAccess = Mutex()
+    scope.launch {
         try {
-                        val size = AsyncVarint.decodeUnsigned(sockInput).toInt()
-                        if (size > MAX_TCP_BLOCK_SIZE) // 16M is a max command block
-                            throw ProtocolException("Illegal block size: $size should be < $MAX_TCP_BLOCK_SIZE")
-                        val data = ByteArray(size)
-                        if (size == 0) {
-                            log.debug { "ping received" }
+            while (isActive) {
+                val size = AsyncVarint.decodeUnsigned(sockInput)
+                if (size > MAX_TCP_BLOCK_SIZE.toULong())
+                    throw ProtocolException("Illegal block size: $size")
                 lastActiveAt = Clock.System.now()
-                        } else {
-                            sockInput.readFully(data, 0, size)
+                if (size != 0uL) {
+                    val data = ByteArray(size.toInt())
+                    sockInput.readFully(data, 0, data.size)
                     inputBlocks.send(data.toUByteArray())
                 }
-                    } catch (_: ClosedReceiveChannelException) {
-                        log.error { "closed receive channel " }
-                        stop()
-                        break
-                    } catch (_: CancellationException) {
-                        log.error { "cancellation exception " }
-                        break
-                    } catch (e: Exception) {
-                        log.exception { "unexpected exception in TCP socket read" to e }
-                        stop()
-                        break
             }
+        } catch (_: Exception) {
+            // EOF, cancellation, or invalid framing ends only this connection.
+        } finally { stop() }
     }
-            }
-        })
-        launch {
-            val outAccess = Mutex()
-            var lastSentAt = Clock.System.now()
-            launch {
-                while (isActive && sock.isActive) {
-                    delay(500)
-                    val activityTime = if (lastSentAt > lastActiveAt) lastSentAt else lastActiveAt
-                    if (Clock.System.now() - activityTime > PING_INACTIVITY_TIME) {
-                        log.debug { "pinging for inactivity" }
-                        val repeat = outAccess.withLock {
+    val priorityWriter = PriorityBlockWriter(scope) { block ->
         try {
+            writeAccess.withLock {
+                AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
+                sockOutput.writeFully(block.asByteArray(), 0, block.size)
+                sockOutput.flush()
+                lastActiveAt = Clock.System.now()
+            }
+        } catch (failure: Throwable) { stop(); throw failure }
+    }
+    // Preserve the public legacy output channel; current transports use send directly.
+    scope.launch {
+        try { for (block in outputBlocks) priorityWriter.send(block) }
+        finally { priorityWriter.close(); stop() }
+    }
+    scope.launch {
+        try {
+            while (isActive) {
+                delay(500)
+                writeAccess.withLock {
+                    if (Clock.System.now() - lastActiveAt > PING_INACTIVITY_TIME) {
                         sockOutput.writeByte(0)
                         sockOutput.flush()
-                                lastSentAt = Clock.System.now()
-                                true
-                            } catch (e: ClosedReceiveChannelException) {
-                                e.printStackTrace()
-                                stop()
-                                false
-                            } catch (_: CancellationException) {
-                                false
-                            } catch (e: Throwable) {
-                                e.printStackTrace()
-                                stop()
-                                false
-                            }
-                        }
-                        if (!repeat) break
+                        lastActiveAt = Clock.System.now()
                     }
                 }
             }
-
-            while (isActive && sock.isActive) {
-                try {
-                    val block = outputBlocks.receive()
-                    outAccess.withLock {
-                        AsyncVarint.encodeUnsigned(block.size.toULong(), sockOutput)
-                        sockOutput.writeFully(block.toByteArray(), 0, block.size)
-                        sockOutput.flush()
-                        lastSentAt = Clock.System.now()
+        } catch (_: Exception) {
+        } finally { stop() }
     }
-                } catch (_: CancellationException) {
-                    log.debug { "cancellation exception on output" }
-                    stop()
-                    break
-                } catch (_: LocalInterface.BreakConnectionException) {
-                    log.debug { "requested connection break" }
-                    stop()
-                    break
-                } catch (_: ClosedReceiveChannelException) {
-                    log.debug { "receive block channel closed, closing the socket" }
-                    stop()
-                    break
-                } catch (e: Exception) {
-                    log.exception { "unexpected exception. closing." to e }
-                    stop()
-                    break
-                }
-            }
-        }
-    }
-    val device = InetTransportDevice(inputBlocks, outputBlocks, networkAddress, {
-        stop()
-    })
-    log.debug { "Transport ready" }
-    return device
+    return InetTransportDevice(inputBlocks, outputBlocks, networkAddress, { stop() }).also { it.priorityWriter = priorityWriter }
 }

src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/TcpServer.kt

@@ -0,0 +1,54 @@
+package net.sergeych.kiloparsec.adapter
+
+import io.ktor.network.selector.SelectorManager
+import io.ktor.network.sockets.*
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+
+/** Managed listener using the same framing as [connectTcpDevice]. Collect once. */
+class TcpServer(val port: Int, localInterface: String = "0.0.0.0") {
+    private val selector = SelectorManager(Dispatchers.IO)
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    private val socket = scope.async { aSocket(selector).tcp().bind(localInterface, port) }
+    private val access = Mutex()
+    private val connections = mutableSetOf<InetTransportDevice>()
+    private var closed = false
+
+    /** Wait for binding, propagating bind errors; supports port zero. */
+    suspend fun boundPort(): Int = (socket.await().localAddress as InetSocketAddress).port
+
+    val transportFlow = flow {
+        try {
+            val listener = socket.await()
+            while (currentCoroutineContext().isActive) {
+                val accepted = listener.accept()
+                val device = access.withLock {
+                    if (closed) { accepted.close(); null }
+                    else {
+                        lateinit var device: InetTransportDevice
+                        device = inetTransportDevice(accepted, "srv") {
+                            access.withLock { connections.remove(device) }
+                        }
+                        connections.add(device)
+                        device
+                    }
+                }
+                if (device != null) emit(device)
+            }
+        } finally { withContext(NonCancellable) { close() } }
+    }
+
+    suspend fun close() = withContext(NonCancellable) {
+        val pending = access.withLock {
+            if (closed) return@withContext
+            closed = true
+            connections.toList().also { connections.clear() }
+        }
+        runCatching { socket.await().close() }
+        pending.forEach { runCatching { it.close() } }
+        scope.cancel()
+        selector.close()
+    }
+}

src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpProvider.kt

@@ -122,17 +122,20 @@ suspend fun connectUdpDevice(
     maxInactivityTimeout: Duration = 2.minutes,
 ): InetTransportDevice {
     val selectorManager = SelectorManager(Dispatchers.IO)
-    val remoteAddress = InetSocketAddress(addr.host, addr.port)
+    val remoteAddress = withContext(Dispatchers.IO) { InetSocketAddress(addr.host, addr.port) }
 
     val done = CompletableDeferred<Unit>()
 
-    val socket = aSocket(selectorManager).udp().connect(remoteAddress)
+    val socket = try { aSocket(selectorManager).udp().connect(remoteAddress) }
+    catch (failure: Throwable) { selectorManager.close(); throw failure }
     val transport = UdpSocketTransport(object : UdpConnector {
         override suspend fun sendBlock(block: UdpBlock, toAddress: SocketAddress) {
             socket.send(block.toDatagram(remoteAddress))
         }
 
         override suspend fun disconnectClient(address: SocketAddress) {
+            socket.close()
+            selectorManager.close()
             done.complete(Unit)
         }
 
@@ -151,7 +154,8 @@ suspend fun connectUdpDevice(
                 }
             }
         }
-        done.await()
+        try { done.await() }
+        finally { withContext(NonCancellable) { transport.close() }; coroutineContext.cancelChildren() }
     }
 
     return transport.transportDevice

src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpServer.kt

@@ -12,7 +12,7 @@ package net.sergeych.kiloparsec.adapter
 
 import io.ktor.network.selector.*
 import io.ktor.network.sockets.*
-import io.ktor.utils.io.*
+import kotlinx.coroutines.*
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.IO
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
@@ -60,12 +60,16 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
 
     private val sessions = mutableMapOf<SocketAddress, UdpSocketTransport>()
     private val access = Mutex()
+    private var closed = false
 
     private val selectorManager = SelectorManager(Dispatchers.IO)
-    private val serverSocket = globalDefer {
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
+    private val serverSocket = scope.async {
         aSocket(selectorManager).udp().bind(InetSocketAddress(localInterface, port))
     }
 
+    suspend fun boundPort(): Int = (serverSocket.await().localAddress as InetSocketAddress).port
+
     override suspend fun disconnectClient(address: SocketAddress) {
         access.withLock { sessions.remove(address) }
     }
@@ -76,13 +80,16 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
      */
     val transportFlow by lazy {
         flow {
+            try {
             while (true) {
                 try {
                     val datagram = serverSocket.await().receive()
-                    val block = UdpBlock.decode(datagram)
+                    val block = try { UdpBlock.decode(datagram) }
+                    catch (_: Exception) { continue } // Malformed traffic must not stop the listener.
                     val remoteAddress = datagram.address
 
                     access.withLock {
+                        if (closed) return@withLock
                         if (block == UdpBlock.Cancel) {
                             // if the cancel comes to already closed transport, do nothing
                             sessions.remove(remoteAddress)?.processIncoming(block)
@@ -106,6 +113,7 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
                     break
                 }
             }
+            } finally { withContext(NonCancellable) { close() } }
         }
     }
 
@@ -124,17 +132,14 @@ class UdpServer(val port: Int, localInterface: String = "0.0.0.0", maxInactivity
      *
      * Call suspends until socket and all sessions are closed. Later calls do nothing.
      */
-    suspend fun close() {
-        access.withLock {
-            if (!isClosed()) {
+    suspend fun close() = withContext(NonCancellable) {
         runCatching { serverSocket.await().close() }
-            }
-        }
-        while (sessions.isNotEmpty()) {
-            runCatching {
-                access.withLock { sessions.values.firstOrNull() }
-                    ?.close()
-            }
-        }
+        val pending = access.withLock {
+            closed = true
+            sessions.values.toList().also { sessions.clear() }
+        }
+        pending.forEach { runCatching { it.close() } }
+        scope.cancel()
+        selectorManager.close()
     }
 }

src/ktorSocketMain/kotlin/net/sergeych/kiloparsec/adapter/UdpSocketTransport.kt

@@ -16,6 +16,9 @@ import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import kotlinx.coroutines.channels.ClosedSendChannelException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.withContext
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
@@ -54,10 +57,13 @@ internal class UdpSocketTransport(
     private val pingMinTimeout = pingTimeout * 2 / 3
 
     val inputDataBlocks = Channel<UByteArray>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
-    val outputDataBlocks = Channel<UByteArray>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
+    val outputDataBlocks = Channel<UByteArray>()
+    private val writerReady = CompletableDeferred<PriorityBlockWriter>()
 
     val inputUdpBlocks = Channel<UdpBlock>(256, onBufferOverflow = BufferOverflow.DROP_OLDEST)
 
+    private val closedFlag = SyncValue(false)
+
     private val job = globalLaunch {
         coroutineScope {
             launch { convertOutput() }
@@ -75,20 +81,22 @@ internal class UdpSocketTransport(
 
 
     val transportDevice: InetTransportDevice by lazy {
-        InetTransportDevice(inputDataBlocks, outputDataBlocks, address, { close() }, {})
+        InetTransportDevice(inputDataBlocks, outputDataBlocks, address, { close() }, {}).also { device ->
+            device.prioritizedSend = { writerReady.await().send(it) }
+        }
     }
 
-    private val closedFlag = SyncValue(false)
+    val isClosed: Boolean get() = closedFlag.value
 
-    val isClosed: Boolean = closedFlag.value
-
-    suspend fun close() {
+    suspend fun close(): Unit = withContext(NonCancellable) {
         closedFlag.mutate {
             if (!it) {
                 runCatching { server.sendBlock(UdpBlock.Cancel, socketAddress) }
                 server.disconnectClient(socketAddress)
                 runCatching { inputDataBlocks.close() }
                 runCatching { outputDataBlocks.close() }
+                runCatching { inputUdpBlocks.close() }
+                writerReady.completeExceptionally(net.sergeych.kiloparsec.RemoteInterface.ClosedException())
                 job.cancel()
             }
             true
@@ -167,10 +175,13 @@ internal class UdpSocketTransport(
         }
     }
 
-    private suspend fun convertOutput() {
+    private suspend fun convertOutput(): Unit = coroutineScope {
+        val writer = PriorityBlockWriter(this) { server.sendBlock(UdpBlock.Data(it), socketAddress) }
+        writerReady.complete(writer)
+        try {
             while (!isClosed) {
                 try {
-                server.sendBlock(UdpBlock.Data(outputDataBlocks.receive()), socketAddress)
+                    writer.send(outputDataBlocks.receive())
                 } catch (_: CancellationException) {
                     // this is ok
                     break
@@ -184,6 +195,7 @@ internal class UdpSocketTransport(
                     break
                 }
             }
+        } finally { writer.close() }
         debug { "exiting convertOutput" }
     }
 }