[13/15] Non-blocking streaming for request and response bodies

core/src/main/kotlin/com/lightningkite/lightningserver/runtime/implementationHelpers.kt

@@ -151,21 +151,26 @@ public suspend fun ServerRuntime.handle(request: HttpRequest<PathSpec>): HttpRes
                 // Lower compress limit. Either not worth the effort, or likely will inflate a little.
                 if (result.body.data.size?.let { it < 256 } == true) return@intercept result
 
+                // Stream-compress a body straight into the response through GZIP, with no full-body buffering. Runs
+                // inside Data.Sink.emit, which engines invoke on a blocking-capable dispatcher, so the blocking GZIP
+                // writes never touch an event loop.
+                fun gzipStream(writePlain: (kotlinx.io.Sink) -> Unit): Data.Sink = Data.Sink { outSink ->
+                    GZIPOutputStream(outSink.asOutputStream()).asSink().buffered().use { gz -> writePlain(gz) }
+                }
                 val (newData, compressed) = when (val data = result.body.data) {
-                    is Data.Sink -> {
-                        Data.Sink { outSink ->
-                            GZIPOutputStream(outSink.asOutputStream()).asSink().buffered().use { gzOut ->
-                                data.write(gzOut)
-                            }
-                        } to true
-                    }
+                    // Push producer / blocking source: drive the plaintext straight into GZIP with no buffering, so a
+                    // large streamed response (e.g. an octet-stream/CSV/JSON download) is never materialized in heap.
+                    is Data.Sink -> gzipStream { data.emit(it) } to true
+                    is Data.Source -> gzipStream { sink -> data.source.use { sink.transferFrom(it) } } to true
 
-                    is Data.Source -> {
-                        Data.Sink { outSink ->
-                            GZIPOutputStream(outSink.asOutputStream()).asSink().buffered().use { gzOut ->
-                                data.write(gzOut)
-                            }
-                        } to true
+                    // Cooperative source: there is no non-suspend way to feed it into a blocking GZIPOutputStream
+                    // without a runBlocking bridge into the response channel (deadlock-adjacent). Compress only when the
+                    // size is known and bounded (4 MiB); otherwise pass the body through uncompressed rather than risk
+                    // materializing an unbounded stream in the heap.
+                    is Data.Suspending, is Data.SuspendingProducer -> {
+                        val s = data.size
+                        if (s != null && s <= 4L * 1024 * 1024) Data.Bytes(data.bytes().gzip()) to true
+                        else return@intercept result
                     }
 
                     else -> {

core/src/main/kotlin/com/lightningkite/lightningserver/serialization/registerBasicMediaTypeCoders.kt

@@ -9,6 +9,8 @@ import com.lightningkite.lightningserver.runtime.serverRuntime
 import com.lightningkite.lightningserver.websockets.WebSocketFrame
 import com.lightningkite.services.data.*
 import com.lightningkite.services.serializers.KotlinBytesFormat
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 import kotlinx.serialization.*
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.io.decodeFromSource
@@ -33,6 +35,7 @@ public open class BinaryFormatMediaTypeCoder(
 ) : MediaTypeCoder {
 
     private var _formatCached: BinaryFormat? = null
+
     private context(runtime: ServerRuntime)
     val formatCached: BinaryFormat
         get() {
@@ -84,6 +87,7 @@ public open class StringFormatMediaTypeCoder(
 ) : MediaTypeCoder {
 
     private var _formatCached: StringFormat? = null
+
     private context(runtime: ServerRuntime)
     val formatCached: StringFormat
         get() {
@@ -139,6 +143,7 @@ public class JsonMediaTypeCoder(
     override val priority: Float get() = 1f
 
     private var _formatCached: Json? = null
+
     private context(runtime: ServerRuntime)
     val formatCached: Json
         get() {
@@ -156,7 +161,11 @@ public class JsonMediaTypeCoder(
     override context(runtime: ServerRuntime)
     suspend fun <T> invoke(content: TypedData, serializer: DeserializationStrategy<T>): T {
         return when (val body = content.data) {
-            is Data.Source -> body.source.use { formatCached.decodeFromSource(serializer, it) }
+            is Data.Source -> withContext(Dispatchers.IO) {
+                body.source.use {
+                    formatCached.decodeFromSource(serializer, it)
+                }
+            }
             else -> super.invoke(content, serializer)
         }
     }
@@ -169,9 +178,8 @@ public class JsonMediaTypeCoder(
     suspend fun <T> invoke(mediaType: MediaType, serializer: SerializationStrategy<T>, value: T): TypedData {
         return TypedData.sink(
             mediaType,
-            emit = {
-                it.use { formatCached.encodeToSink(serializer, value, it) }
-            }
+            // Do NOT close `sink` — it is caller-owned (see Data.write's contract); the consumer manages its lifecycle.
+            emit = { sink -> formatCached.encodeToSink(serializer, value, sink) }
         )
     }
 }

core/src/test/kotlin/com/lightningkite/lightningserver/runtime/ImplementationHelpersHandleTest.kt

@@ -72,6 +72,17 @@ class ImplementationHelpersHandleTest {
             )
         }
 
+        // Streaming large response backed by a blocking Data.Source at /bigsource
+        val bigSource = path.path("bigsource").get bind HttpHandler<PathSpec0> {
+            val content = "s".repeat(100_000)
+            HttpResponse(
+                body = TypedData.source(
+                    source = kotlinx.io.Buffer().also { it.writeString(content) },
+                    mediaType = MediaType.Text.Plain,
+                ),
+            )
+        }
+
         // Large plain text at /big for Range tests
         val bigGet = path.path("big").get bind HttpHandler<PathSpec0> {
             HttpResponse.plainText("z".repeat(10_000))
@@ -648,4 +659,29 @@ class ImplementationHelpersHandleTest {
             }
         }
     }
+
+    @Test
+    fun gzip_applied_on_stream_source() {
+        // The blocking Data.Source path must stream-compress (no full-body buffering) and still produce valid gzip.
+        TestServer.test(settings = {}) {
+            runBlocking {
+                val resp = serverRuntime.handle(
+                    HttpRequest(
+                        path = RawHttpEndpoint(asString = "/bigsource", method = HttpMethod.GET),
+                        queryParameters = QueryParameters.EMPTY,
+                        headers = HttpHeaders { add(HttpHeader.AcceptEncoding, "gzip") },
+                        domain = "example.com",
+                        protocol = "https",
+                        sourceIp = "local",
+                    )
+                )
+                assertEquals(HttpStatus.OK, resp.status)
+                assertEquals("gzip", resp.headers[HttpHeader.ContentEncoding]?.root)
+                val compressed = resp.body?.data?.bytes() ?: error("Expected body bytes")
+                val decompressed =
+                    GZIPInputStream(ByteArrayInputStream(compressed)).readBytes().toString(Charsets.UTF_8)
+                assertEquals("s".repeat(100_000), decompressed)
+            }
+        }
+    }
 }

engine-jdk-server/src/main/kotlin/com/lightningkite/lightningserver/engine/jdk/JdkEngine.kt

@@ -129,8 +129,10 @@ public class JdkEngine(
                 }
                 val request = exchange.requestToLightningServer(cfg.realIpHeader, this@JdkEngine, maxBody)
                 // Request timeout is enforced centrally in ServerRuntime.handle (per-handler HttpHandler.timeout).
-                val result: HttpResponse = runBlocking { this@JdkEngine.handle(request) }
-                exchange.write(result)
+                runBlocking {
+                    val result: HttpResponse = this@JdkEngine.handle(request)
+                    exchange.write(result)
+                }
             } catch (e: BodyTooLargeException) {
                 // 2.5: streamed body exceeded the cap mid-read.
                 try {
@@ -198,7 +200,7 @@ private fun HttpExchange.respondPlain(status: Int, message: String) {
  * Writes a Lightning Server HttpResponse to a JDK HttpExchange.
  * Handles all response types including empty bodies, bytes, text, sinks, and sources.
  */
-private fun HttpExchange.write(response: HttpResponse) {
+private suspend fun HttpExchange.write(response: HttpResponse) {
     // Copy headers from response
     for ((key, values) in response.headers.normalizedEntries) {
         for (value in values) {
@@ -240,6 +242,11 @@ private fun HttpExchange.write(response: HttpResponse) {
             sendResponseHeaders(status, b.size ?: 0)
             this.responseBody.asSink().buffered().use { sink -> b.source.transferTo(sink) }
         }
+
+        is Data.Suspending, is Data.SuspendingProducer -> {
+            sendResponseHeaders(status, b.size ?: 0)
+            this.responseBody.asSink().buffered().use { sink -> b.write(sink) }
+        }
     }
 }
 

engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorChannelSuspendingSink.kt

@@ -0,0 +1,43 @@
+package com.lightningkite.lightningserver.engine.ktor
+
+import com.lightningkite.services.data.StreamState
+import com.lightningkite.services.data.SuspendingSink
+import io.ktor.utils.io.ByteWriteChannel
+import io.ktor.utils.io.writeBuffer
+import kotlinx.io.Buffer
+
+/**
+ * A [SuspendingSink] over a Ktor [ByteWriteChannel] — the cooperative, non-blocking response-body channel.
+ *
+ * Writes *suspend* (yielding the event-loop thread when the channel applies backpressure) instead of blocking it, so
+ * streaming a response to a slow-reading client can never pin the event loop the way bridging through a blocking
+ * `kotlinx.io.Sink` would.
+ *
+ * Lifecycle: this does **not** close the underlying channel — `respondBytesWriter` owns and closes it when its block
+ * returns. [close] only flushes; [close] with a cause cancels the channel so the client sees a truncated response.
+ */
+internal class KtorChannelSuspendingSink(private val channel: ByteWriteChannel) : SuspendingSink {
+    override var state: StreamState = StreamState.Open
+        private set
+
+    override suspend fun write(from: Buffer, count: Long) {
+        // writeBuffer consumes `count` bytes from `from` and suspends for backpressure (never blocks the thread).
+        channel.writeBuffer(from, count)
+    }
+
+    override suspend fun flush() {
+        channel.flush()
+    }
+
+    override suspend fun close() {
+        if (state != StreamState.Open) return
+        channel.flush()
+        state = StreamState.Complete
+    }
+
+    override fun close(cause: Throwable) {
+        if (state != StreamState.Open) return
+        channel.cancel(cause)
+        state = StreamState.ClosedAbnormally(cause)
+    }
+}

engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorChannelSuspendingSource.kt

@@ -0,0 +1,69 @@
+package com.lightningkite.lightningserver.engine.ktor
+
+import com.lightningkite.lightningserver.BadRequestException
+import com.lightningkite.lightningserver.engine.local.BodyTooLargeException
+import com.lightningkite.services.data.AbstractSuspendingSource
+import io.ktor.utils.io.ByteReadChannel
+import io.ktor.utils.io.InternalAPI
+import kotlinx.io.Buffer
+
+/**
+ * A [com.lightningkite.services.data.SuspendingSource] over a Ktor [ByteReadChannel] — the cooperative, non-blocking
+ * request-body channel.
+ *
+ * Reads *suspend* (yielding the engine's event-loop thread so it can keep delivering socket bytes) instead of blocking
+ * a thread, which is what makes streamed/slow request bodies safe to consume directly on the event loop. It also
+ * enforces [maxBody], throwing [BodyTooLargeException] as soon as the cumulative body exceeds it. State/EOF bookkeeping
+ * is handled by [AbstractSuspendingSource]; this only supplies the [fill] primitive.
+ *
+ * ## Truncated uploads never look complete (finding A1)
+ *
+ * A client that dies mid-upload can end the body two ways, and both must surface as an error rather than a short-but-
+ * "clean" body the handler mistakes for the whole request:
+ * - **Reset / broken framing** (e.g. an incomplete chunked body): the channel closes *with a cause*, which
+ *   [io.ktor.utils.io.ByteReadChannel.awaitContent] rethrows.
+ * - **Content-Length underrun** (socket half-closed after fewer bytes than promised): the channel reports a *clean*
+ *   EOF with no cause, so we compare against [expectedLength] ourselves and reject the truncated body.
+ */
+internal class KtorChannelSuspendingSource(
+    private val channel: ByteReadChannel,
+    private val maxBody: Long,
+    private val expectedLength: Long?,
+) : AbstractSuspendingSource() {
+    private var readSoFar = 0L
+
+    @OptIn(InternalAPI::class)
+    override suspend fun fill(into: Buffer, count: Long): Boolean {
+        // Suspend until at least one byte is buffered. Returns false at EOF; THROWS if the client aborted with a cause.
+        if (!channel.awaitContent()) return endOfStream()
+        // Take everything currently buffered in one shot — no stalling until a fixed-size chunk fills, so trickle/
+        // segmented bodies flow through with minimal latency.
+        val moved = channel.readBuffer.readAtMostTo(into, READ_AHEAD)
+        // awaitContent() guaranteed a byte, so moved > 0 here; the guard is defensive and must honor the same
+        // truncation check rather than reporting a silent clean EOF.
+        if (moved <= 0L) return endOfStream()
+        readSoFar += moved
+        if (readSoFar > maxBody) throw BodyTooLargeException()
+        return true
+    }
+
+    /** Handle a channel EOF: a clean end unless a declared Content-Length says bytes are still owed (A1). */
+    private fun endOfStream(): Boolean {
+        if (expectedLength != null && readSoFar < expectedLength) {
+            throw BadRequestException(
+                detail = "truncated-body",
+                message = "Request body ended after $readSoFar bytes but Content-Length declared $expectedLength.",
+            )
+        }
+        return false
+    }
+
+    override fun release(cause: Throwable?) {
+        channel.cancel(cause)
+    }
+
+    private companion object {
+        // Upper bound per fill; the channel rarely has more than a segment buffered, so this just caps a single move.
+        const val READ_AHEAD = 64L * 1024
+    }
+}

engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorEngine.kt

@@ -162,9 +162,23 @@ public class KtorEngine(
                             is Data.Bytes -> call.respondBytes(body.data, type, code)
                             is Data.Text -> call.respondText(body.data, type, code)
                             is Data.Sink -> call.respondBytesWriter(contentType = type, status = code) {
-                                this.asSink().buffered().use { body.emit(it) }
+                                // emit is a blocking producer — run it on the IO pool (via Ktor's blocking asSink
+                                // bridge) so it streams to the channel without ever stalling the event-loop thread.
+                                val channel = this
+                                withContext(Dispatchers.IO) { channel.asSink().buffered().use { body.emit(it) } }
+                            }
+                            is Data.Source -> call.respondBytesWriter(contentType = type, status = code) {
+                                // Blocking streaming source: copy it to the channel on the IO pool (no full buffering).
+                                val channel = this
+                                withContext(Dispatchers.IO) {
+                                    channel.asSink().buffered().use { sink -> body.source.use { sink.transferFrom(it) } }
+                                }
+                            }
+                            is Data.Suspending, is Data.SuspendingProducer -> call.respondBytesWriter(contentType = type, status = code) {
+                                // Fully cooperative: stream the body into the ByteWriteChannel via a SuspendingSink so
+                                // response writes suspend for backpressure instead of blocking the event loop.
+                                body.writeTo(KtorChannelSuspendingSink(this))
                             }
-                            is Data.Source -> body.source.use { call.respondSource(it, type, code, body.size) }
                         }
                 }
             }

engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/extensions.kt

@@ -3,8 +3,6 @@ package com.lightningkite.lightningserver.engine.ktor
 import com.lightningkite.lightningserver.HttpMethod
 import com.lightningkite.lightningserver.http.*
 import com.lightningkite.lightningserver.http.HttpHeaders
-import com.lightningkite.lightningserver.engine.local.BodyTooLargeException
-import com.lightningkite.lightningserver.engine.local.copyLimited
 import com.lightningkite.lightningserver.logger
 import com.lightningkite.lightningserver.pathing.PathSpec
 import com.lightningkite.lightningserver.pathing.RawHttpEndpoint
@@ -49,11 +47,16 @@ internal suspend fun ApplicationCall.adapt(maxBody: Long): HttpRequest<PathSpec>
         } ?: request.origin.remoteAddress,
         body = run {
             // TODO: Add MultiPart support
-            val stream = receiveStream()
-
-            TypedData.sink(request.contentType().adapt(), request.contentLength() ?: -1) {
-                copyLimited(stream, maxBody) { b, off, len -> it.write(b, off, len) }
-            }
+            // Cooperative, non-blocking body read: back the body with Ktor's suspending ByteReadChannel so consuming
+            // it yields the event loop instead of blocking it (the original receiveStream()+runBlocking path could
+            // deadlock the event loop on slow/segmented bodies). maxBody is enforced during the suspending read.
+            val declaredLength = request.contentLength()
+            TypedData.suspending(
+                // declaredLength lets the source reject a body that ends before its declared Content-Length (A1).
+                source = KtorChannelSuspendingSource(receiveChannel(), maxBody, declaredLength),
+                mediaType = request.contentType().adapt(),
+                size = declaredLength,
+            )
         },
     )
 }