diff --git a/core/src/main/kotlin/com/lightningkite/lightningserver/runtime/implementationHelpers.kt b/core/src/main/kotlin/com/lightningkite/lightningserver/runtime/implementationHelpers.kt
index 2e63372c7..069d7ad36 100644
--- a/core/src/main/kotlin/com/lightningkite/lightningserver/runtime/implementationHelpers.kt
+++ b/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 -> {
diff --git a/core/src/main/kotlin/com/lightningkite/lightningserver/serialization/registerBasicMediaTypeCoders.kt b/core/src/main/kotlin/com/lightningkite/lightningserver/serialization/registerBasicMediaTypeCoders.kt
index 3070d71a5..ca5d3f4de 100644
--- a/core/src/main/kotlin/com/lightningkite/lightningserver/serialization/registerBasicMediaTypeCoders.kt
+++ b/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) }
         )
     }
 }
diff --git a/core/src/test/kotlin/com/lightningkite/lightningserver/runtime/ImplementationHelpersHandleTest.kt b/core/src/test/kotlin/com/lightningkite/lightningserver/runtime/ImplementationHelpersHandleTest.kt
index b428e8baf..b5f82892c 100644
--- a/core/src/test/kotlin/com/lightningkite/lightningserver/runtime/ImplementationHelpersHandleTest.kt
+++ b/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)
+            }
+        }
+    }
 }
diff --git a/engine-jdk-server/src/main/kotlin/com/lightningkite/lightningserver/engine/jdk/JdkEngine.kt b/engine-jdk-server/src/main/kotlin/com/lightningkite/lightningserver/engine/jdk/JdkEngine.kt
index adaca50f1..ad89c6dea 100644
--- a/engine-jdk-server/src/main/kotlin/com/lightningkite/lightningserver/engine/jdk/JdkEngine.kt
+++ b/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) }
+        }
     }
 }
 
diff --git a/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorChannelSuspendingSink.kt b/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorChannelSuspendingSink.kt
new file mode 100644
index 000000000..32d22c4b4
--- /dev/null
+++ b/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)
+    }
+}
diff --git a/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorChannelSuspendingSource.kt b/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorChannelSuspendingSource.kt
new file mode 100644
index 000000000..63dac0b48
--- /dev/null
+++ b/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
+    }
+}
diff --git a/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorEngine.kt b/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorEngine.kt
index 5330d5955..888eab114 100644
--- a/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorEngine.kt
+++ b/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) }
                         }
                 }
             }
diff --git a/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/extensions.kt b/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/extensions.kt
index 5e443c8aa..8a1e2a65a 100644
--- a/engine-ktor/src/main/kotlin/com/lightningkite/lightningserver/engine/ktor/extensions.kt
+++ b/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,
+            )
         },
     )
 }
diff --git a/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorRequestBodyStreamingTest.kt b/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorRequestBodyStreamingTest.kt
new file mode 100644
index 000000000..620a0e32e
--- /dev/null
+++ b/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorRequestBodyStreamingTest.kt
@@ -0,0 +1,180 @@
+package com.lightningkite.lightningserver.engine.ktor
+
+import com.lightningkite.lightningserver.definition.builder.ServerBuilder
+import com.lightningkite.lightningserver.definition.generalSettings
+import com.lightningkite.lightningserver.definition.loggingSettings
+import com.lightningkite.lightningserver.definition.secretBasis
+import com.lightningkite.lightningserver.definition.telemetrySettings
+import com.lightningkite.lightningserver.engine.local.EngineReliabilitySettings
+import com.lightningkite.lightningserver.engine.local.engineCache
+import com.lightningkite.lightningserver.engine.local.enginePubSub
+import com.lightningkite.lightningserver.engine.local.forceWebSocketPubSub
+import com.lightningkite.lightningserver.http.*
+import com.lightningkite.lightningserver.pathing.PathSpec0
+import com.lightningkite.lightningserver.plainText
+import com.lightningkite.lightningserver.serialization.registerBasicMediaTypeCoders
+import com.lightningkite.lightningserver.settings.set
+import com.lightningkite.services.data.DataSize.Companion.bytes
+import io.ktor.server.cio.CIO as ServerCIO
+import java.net.InetSocketAddress
+import java.net.ServerSocket
+import java.net.Socket
+import kotlin.concurrent.thread
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+import kotlin.test.fail
+
+/**
+ * Reproduces the failure that motivated the cooperative (non-blocking) request-body redesign.
+ *
+ * The old path read the request body with a blocking `receiveStream()` + `runBlocking` **on the event-loop thread** —
+ * the very thread responsible for feeding the body channel. A body that did not arrive all at once (slow client,
+ * segmented upload, proxy) therefore parked the feeder thread while the handler waited for bytes that could never come:
+ * a permanent deadlock. [slow_segmented_body_does_not_deadlock_event_loop] drives exactly that shape.
+ *
+ * It also pins [aborted_upload_is_not_accepted_as_complete] (finding A1): a client that dies mid-upload must surface as
+ * an error, never as a clean — but truncated — body that the handler mistakes for the whole request.
+ *
+ * These run against a real CIO server on a real socket because the bug lives in the engine's thread/event-loop
+ * handling; an in-memory `testApplication` would not exercise it.
+ */
+class KtorRequestBodyStreamingTest {
+
+    object TestServer : ServerBuilder() {
+        init { registerBasicMediaTypeCoders() }
+        val echo = path.path("echo").post bind HttpHandler<PathSpec0> { request ->
+            val bytes = request.body?.data?.bytes() ?: ByteArray(0)
+            HttpResponse.plainText("received ${bytes.size}")
+        }
+    }
+
+    private lateinit var engine: KtorEngine
+    private var port: Int = 0
+    private lateinit var serverThread: Thread
+    private val maxBody = 4L * 1024 * 1024
+
+    @AfterTest
+    fun tearDown() {
+        if (::serverThread.isInitialized) serverThread.interrupt()
+    }
+
+    private fun startServer() {
+        ServerSocket(0).use { port = (it.localSocketAddress as InetSocketAddress).port }
+        engine = KtorEngine(TestServer.build())
+        engine.settings.run {
+            generalSettings.useDefault()
+            secretBasis.useDefault()
+            loggingSettings.useDefault()
+            telemetrySettings.useDefault()
+            enginePubSub.useDefault()
+            engineCache.useDefault()
+            forceWebSocketPubSub.useDefault()
+            ktorRunConfig set KtorRuntimeSettings(
+                host = "127.0.0.1",
+                port = port,
+                reliability = EngineReliabilitySettings(maxBodySize = maxBody.bytes),
+            )
+        }
+        serverThread = thread(start = true, isDaemon = true) { engine.start(ServerCIO) }
+        val deadline = System.currentTimeMillis() + 15_000
+        while (System.currentTimeMillis() < deadline) {
+            try {
+                Socket().use { it.connect(InetSocketAddress("127.0.0.1", port), 100) }
+                return
+            } catch (_: Exception) {
+                Thread.sleep(50)
+            }
+        }
+        fail("KtorEngine never bound within 15s")
+    }
+
+    /** Reads a full HTTP/1.1 response (headers + body) from a `Connection: close` socket, to EOF. */
+    private fun Socket.readResponse(): String {
+        val out = StringBuilder()
+        getInputStream().bufferedReader().use { reader ->
+            val buf = CharArray(4096)
+            while (true) {
+                val n = reader.read(buf)
+                if (n == -1) break
+                out.append(buf, 0, n)
+            }
+        }
+        return out.toString()
+    }
+
+    @Test
+    fun slow_segmented_body_does_not_deadlock_event_loop() {
+        startServer()
+        val segment = ByteArray(1024) { 'x'.code.toByte() }
+        val segments = 64
+        val total = segment.size * segments
+
+        Socket("127.0.0.1", port).use { socket ->
+            // If the old deadlock were present, the handler would never finish reading and this read would block
+            // until the socket timeout, failing the test.
+            socket.soTimeout = 15_000
+            val out = socket.getOutputStream()
+            out.write(
+                ("POST /echo HTTP/1.1\r\n" +
+                        "Host: 127.0.0.1\r\n" +
+                        "Content-Type: application/octet-stream\r\n" +
+                        "Content-Length: $total\r\n" +
+                        "Connection: close\r\n\r\n").toByteArray(Charsets.US_ASCII)
+            )
+            out.flush()
+            // Dribble the body out in segments so it is NOT all buffered before the handler starts reading —
+            // this is the condition that deadlocked the blocking path.
+            repeat(segments) {
+                out.write(segment)
+                out.flush()
+                Thread.sleep(5)
+            }
+
+            val response = socket.readResponse()
+            assertTrue(response.startsWith("HTTP/1.1 200"), "expected 200, got:\n$response")
+            assertTrue(
+                response.trimEnd().endsWith("received $total"),
+                "handler should have read the whole segmented body ($total bytes); got:\n$response",
+            )
+        }
+    }
+
+    @Test
+    fun aborted_upload_is_not_accepted_as_complete() {
+        startServer()
+        val declared = 20_000
+        val sent = 8_000 // client dies after sending less than it promised
+
+        Socket("127.0.0.1", port).use { socket ->
+            socket.soTimeout = 15_000
+            val out = socket.getOutputStream()
+            out.write(
+                ("POST /echo HTTP/1.1\r\n" +
+                        "Host: 127.0.0.1\r\n" +
+                        "Content-Type: application/octet-stream\r\n" +
+                        "Content-Length: $declared\r\n" +
+                        "Connection: close\r\n\r\n").toByteArray(Charsets.US_ASCII)
+            )
+            out.write(ByteArray(sent) { 'y'.code.toByte() })
+            out.flush()
+            // Half-close: signal end-of-input after only `sent` of `declared` bytes, then wait for the response.
+            socket.shutdownOutput()
+
+            val response = try {
+                socket.readResponse()
+            } catch (_: Exception) {
+                "" // a dropped/reset connection is an acceptable "did not accept the truncated body" outcome
+            }
+
+            // A truncated upload must never be accepted as a successful request — neither as the full declared
+            // length nor as a "complete" body of whatever fraction happened to arrive. Any 2xx here is silent data
+            // loss; the correct outcome is an error status (400 truncated-body) or a dropped connection.
+            assertTrue(
+                !response.startsWith("HTTP/1.1 2"),
+                "a truncated upload was accepted as a successful request (silent data loss); got:\n$response",
+            )
+        }
+    }
+}
diff --git a/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorSinkResponseTest.kt b/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorSinkResponseTest.kt
index 5a0665fff..cc93add06 100644
--- a/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorSinkResponseTest.kt
+++ b/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/KtorSinkResponseTest.kt
@@ -81,6 +81,39 @@ class KtorSinkResponseTest {
         }
     }
 
+    @Test
+    fun blocking_source_response_streams_full_content() = runTest {
+        // Data.Source response branch: streamed off the event loop (withContext(IO) + asSink bridge), not on it.
+        withEngine {
+            val response = client.get("/source")
+            assertEquals(HttpStatusCode.OK, response.status)
+            assertEquals("streamed-content", response.bodyAsText())
+        }
+    }
+
+    @Test
+    fun suspending_source_response_streams_full_content() = runTest {
+        // Cooperative Data.Suspending response branch: streamed via KtorChannelSuspendingSink (fully non-blocking).
+        withEngine {
+            val response = client.get("/suspendingsource")
+            assertEquals(HttpStatusCode.OK, response.status)
+            assertEquals("suspending-content", response.bodyAsText())
+        }
+    }
+
+    @Test
+    fun large_suspending_producer_is_fully_delivered() = runTest {
+        // 100k cooperative producer body: any dropped tail or backpressure-handling bug in the SuspendingSink
+        // adapter would show up as a short read.
+        withEngine {
+            val response = client.get("/suspendingproducer")
+            assertEquals(HttpStatusCode.OK, response.status)
+            val text = response.bodyAsText()
+            assertEquals(100_000, text.length)
+            assertEquals("y".repeat(100_000), text)
+        }
+    }
+
     @Test
     fun sink_response_uses_chunked_transfer_encoding() = runTest {
         // Data.Sink has no known size, so Ktor should fall back to chunked transfer.
diff --git a/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/TestServer.kt b/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/TestServer.kt
index 3808952f2..6828a1bf1 100644
--- a/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/TestServer.kt
+++ b/engine-ktor/src/test/kotlin/com/lightningkite/lightningserver/engine/ktor/TestServer.kt
@@ -7,6 +7,8 @@ import com.lightningkite.lightningserver.plainText
 import com.lightningkite.lightningserver.websockets.*
 import com.lightningkite.services.data.MediaType
 import com.lightningkite.services.data.TypedData
+import com.lightningkite.services.data.asSuspendingSource
+import com.lightningkite.services.data.writeAll
 import kotlinx.io.*
 import kotlinx.serialization.Serializable
 import java.io.ByteArrayInputStream
@@ -82,6 +84,30 @@ object TestServerBuilder : ServerBuilder() {
         )
     }
 
+    // GET /suspendingsource — exercises the cooperative Data.Suspending response branch in KtorEngine
+    val suspendingSource = path.path("suspendingsource").get bind HttpHandler<PathSpec0> {
+        val content = "suspending-content"
+        HttpResponse(
+            body = TypedData.suspending(
+                source = Buffer().also { it.writeString(content) }.asSuspendingSource(),
+                mediaType = MediaType.Text.Plain,
+                size = content.length.toLong(),
+            ),
+            status = HttpStatus.OK,
+        )
+    }
+
+    // GET /suspendingproducer — large cooperative producer body; verifies full delivery + backpressure via SuspendingSink
+    val suspendingProducer = path.path("suspendingproducer").get bind HttpHandler<PathSpec0> {
+        val content = "y".repeat(100_000)
+        HttpResponse(
+            body = TypedData.suspendingProducer(MediaType.Text.Plain) { sink ->
+                sink.writeAll(Buffer().also { it.writeString(content) })
+            },
+            status = HttpStatus.OK,
+        )
+    }
+
     // GET /empty triggers 204 branch with preset CT + CL headers
     val empty = path.path("empty").get bind HttpHandler<PathSpec0> {
         HttpResponse(
diff --git a/engine-netty/src/main/kotlin/com/lightningkite/lightningserver/engine/netty/NettyEngine.kt b/engine-netty/src/main/kotlin/com/lightningkite/lightningserver/engine/netty/NettyEngine.kt
index 892229462..ce7a3d9aa 100644
--- a/engine-netty/src/main/kotlin/com/lightningkite/lightningserver/engine/netty/NettyEngine.kt
+++ b/engine-netty/src/main/kotlin/com/lightningkite/lightningserver/engine/netty/NettyEngine.kt
@@ -675,7 +675,7 @@ public class NettyEngine(
             )
         }
 
-        private fun HttpResponse.toNettyResponse(version: HttpVersion): FullHttpResponse {
+        private suspend fun HttpResponse.toNettyResponse(version: HttpVersion): FullHttpResponse {
             val contentBuf = this.body?.data?.bytes()
                 ?.let { Unpooled.wrappedBuffer(it) }
                 ?: Unpooled.EMPTY_BUFFER
diff --git a/files/src/main/kotlin/com/lightningkite/lightningserver/files/ranging.kt b/files/src/main/kotlin/com/lightningkite/lightningserver/files/ranging.kt
index d0ffab01a..ad90edac4 100644
--- a/files/src/main/kotlin/com/lightningkite/lightningserver/files/ranging.kt
+++ b/files/src/main/kotlin/com/lightningkite/lightningserver/files/ranging.kt
@@ -4,9 +4,49 @@ import com.lightningkite.lightningserver.BadRequestException
 import com.lightningkite.lightningserver.http.HttpHeader
 import com.lightningkite.lightningserver.http.HttpHeaders
 import com.lightningkite.services.data.*
+import kotlinx.io.Buffer
 import kotlinx.io.Sink
 import kotlinx.io.writeString
 
+/** Sentinel for [RangeSlicingSink.forward] meaning "forward everything after the skip, to end of stream". */
+private const val RANGE_UNTIL_END: Long = -1L
+
+/**
+ * A [SuspendingSink] that slices a passing byte stream: it discards the first [skip] bytes, forwards the next
+ * [forward] bytes to [downstream] (or all remaining bytes if [forward] is [RANGE_UNTIL_END]), and discards anything
+ * past that. Lets a range be served straight from a streaming body without buffering the whole thing in memory.
+ *
+ * Does not close [downstream] — the enclosing producer owns its lifecycle.
+ */
+private class RangeSlicingSink(
+    private val downstream: SuspendingSink,
+    skip: Long,
+    private val forward: Long,
+) : SuspendingSink {
+    private var toSkip = skip
+    private var forwarded = 0L
+
+    override val state: StreamState get() = downstream.state
+
+    override suspend fun write(from: Buffer, count: Long) {
+        var remaining = count
+        if (toSkip > 0L) {
+            val dropped = minOf(toSkip, remaining)
+            from.skip(dropped); toSkip -= dropped; remaining -= dropped
+        }
+        if (remaining <= 0L) return
+        val allowed = if (forward == RANGE_UNTIL_END) remaining else minOf(remaining, forward - forwarded)
+        if (allowed > 0L) {
+            downstream.write(from, allowed); forwarded += allowed; remaining -= allowed
+        }
+        if (remaining > 0L) from.skip(remaining) // past the window — discard, but still consume `count` from `from`
+    }
+
+    override suspend fun flush(): Unit = downstream.flush()
+    override suspend fun close() {} // downstream is caller-owned
+    override fun close(cause: Throwable) {}
+}
+
 /**
  * Represents a single range value requested by a `Range` header.
  *
@@ -27,7 +67,9 @@ public sealed interface HttpRange {
      * A range of the form `<range-start>-<range-end>`. Range starts at [rangeStart] and ends at [rangeEnd], end inclusive.
      * */
     public data class Bounded(val rangeStart: Long, val rangeEnd: Long) : HttpRange {
-        val size: Long get() = rangeEnd - rangeStart
+        // RFC 9110: both ends are inclusive, so `bytes=0-0` is one byte and `bytes=100-109` is ten. This is the byte
+        // count the body must carry to match the `Content-Range: bytes start-end/total` header the endpoint sends.
+        val size: Long get() = rangeEnd - rangeStart + 1
 
         override fun rangeStart(resourceSize: Long): Long = rangeStart
         override fun rangeEnd(resourceSize: Long): Long = rangeEnd
@@ -124,29 +166,38 @@ public fun ByteArray.sliceArray(range: HttpRange): ByteArray = sliceArray(
     range.rangeStart(size.toLong()).toInt()..range.rangeEnd(size.toLong()).toInt()
 )
 
-public fun TypedData.getRange(range: HttpRange, dataSize: Long): TypedData =
+public suspend fun TypedData.getRange(range: HttpRange, dataSize: Long): TypedData =
     TypedData(
         mediaType = mediaType,
         data = when (data) {
             is Data.Bytes, is Data.Text -> Data.Bytes(data.bytes().sliceArray(range))
-            is Data.Sink, is Data.Source -> Data.Sink { sink ->
-                data.source().use { source ->
-                    source.skip(range.rangeStart(dataSize))
-                    when (range) {
-                        is HttpRange.Bounded -> sink.write(source, range.size)
-                        is HttpRange.Last, is HttpRange.UntilEnd -> source.transferTo(sink)
-                    }
+            is Data.Sink, is Data.Source, is Data.Suspending, is Data.SuspendingProducer -> {
+                // Stream the requested window through instead of buffering the whole body into the heap: pipe the
+                // source into a slicing sink that drops everything before the range and forwards only the window.
+                // (This still *reads* through the whole upstream — post-hoc slicing can't seek — but heap stays flat,
+                // which is what matters for large media served from a streaming source.)
+                val bytesToForward: Long = when (range) {
+                    is HttpRange.Bounded -> range.size
+                    is HttpRange.Last, is HttpRange.UntilEnd -> RANGE_UNTIL_END
+                }
+                val start = range.rangeStart(dataSize)
+                Data.SuspendingProducer(size = bytesToForward.takeIf { it != RANGE_UNTIL_END }) { out ->
+                    data.writeTo(RangeSlicingSink(out, skip = start, forward = bytesToForward))
                 }
             }
         }
     )
 
-public fun TypedData.getRanges(
+public suspend fun TypedData.getRanges(
     ranges: List<HttpRange>,
     dataSize: Long,
     rangeBoundary: String = "CONTENT_BOUNDARY",
-): TypedData =
-    TypedData(
+): TypedData {
+    // Multi-range (multipart/byteranges) responses buffer the whole body once: the boundary framing interleaves
+    // headers between arbitrary, possibly-overlapping windows, so a single streaming pass isn't enough. Multi-range
+    // requests are rare; single-range seeking (the large-media case) streams via getRange without buffering.
+    val raw = data.bytes()
+    return TypedData(
         mediaType = MediaType.MultiPart.ByteRanges.copy(parameters = mapOf("boundary" to rangeBoundary)),
         data = Data.Sink { sink ->
             fun Sink.writeRangeHeaders(range: HttpRange) {
@@ -157,7 +208,7 @@ public fun TypedData.getRanges(
 
             if (data is Data.Bytes || data is Data.Text || ranges.mergeOverlaps(dataSize).size != ranges.size) {
                 // read all bytes if available or ranges overlap
-                val bytes = data.bytes()
+                val bytes = raw
 
                 for (range in ranges) {
                     sink.writeRangeHeaders(range)
@@ -166,7 +217,7 @@ public fun TypedData.getRanges(
                     sink.writeString(LINE_FEED)
                 }
                 sink.writeString(rangeBoundary)
-            } else data.source().use { source ->  // use source if possible
+            } else kotlinx.io.Buffer().apply { write(raw) }.let { source ->  // use source if possible
                 var pos = 0L
                 for (range in ranges) {
                     sink.writeRangeHeaders(range)
@@ -190,5 +241,6 @@ public fun TypedData.getRanges(
             }
         }
     )
+}
 
 private const val LINE_FEED = "\r\n"
\ No newline at end of file
diff --git a/files/src/test/kotlin/com/lightningkite/lightningserver/files/TypedDataGetRangeTest.kt b/files/src/test/kotlin/com/lightningkite/lightningserver/files/TypedDataGetRangeTest.kt
index 37aa81683..5ef1bb5b5 100644
--- a/files/src/test/kotlin/com/lightningkite/lightningserver/files/TypedDataGetRangeTest.kt
+++ b/files/src/test/kotlin/com/lightningkite/lightningserver/files/TypedDataGetRangeTest.kt
@@ -4,6 +4,7 @@ import com.lightningkite.services.data.Data
 import com.lightningkite.services.data.MediaType
 import com.lightningkite.services.data.TypedData
 import kotlinx.io.Buffer
+import kotlinx.coroutines.runBlocking
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertSame
@@ -33,26 +34,26 @@ class TypedDataGetRangeTest {
     // -------- Bytes --------
 
     @Test
-    fun bounded_on_bytes_slices_inclusive() {
+    fun bounded_on_bytes_slices_inclusive(): Unit = runBlocking {
         val out = typedBytes().getRange(HttpRange.Bounded(0, 9), size).data.bytes()
         assertEquals(10, out.size)
         assertEquals(payload.sliceArray(0..9).toList(), out.toList())
     }
 
     @Test
-    fun until_end_on_bytes_returns_tail() {
+    fun until_end_on_bytes_returns_tail(): Unit = runBlocking {
         val out = typedBytes().getRange(HttpRange.UntilEnd(250), size).data.bytes()
         assertEquals(payload.sliceArray(250..255).toList(), out.toList())
     }
 
     @Test
-    fun last_on_bytes_returns_suffix() {
+    fun last_on_bytes_returns_suffix(): Unit = runBlocking {
         val out = typedBytes().getRange(HttpRange.Last(8), size).data.bytes()
         assertEquals(payload.sliceArray(248..255).toList(), out.toList())
     }
 
     @Test
-    fun bounded_on_bytes_preserves_media_type() {
+    fun bounded_on_bytes_preserves_media_type(): Unit = runBlocking {
         val original = typedBytes()
         val ranged = original.getRange(HttpRange.Bounded(0, 1), size)
         assertSame(original.mediaType, ranged.mediaType)
@@ -61,7 +62,7 @@ class TypedDataGetRangeTest {
     // -------- Text --------
 
     @Test
-    fun bounded_on_text_slices_underlying_bytes() {
+    fun bounded_on_text_slices_underlying_bytes(): Unit = runBlocking {
         val text = "abcdefghij" // 10 bytes ASCII
         val out = typedText(text).getRange(HttpRange.Bounded(2, 5), text.length.toLong()).data.bytes()
         assertEquals("cdef", out.decodeToString())
@@ -70,76 +71,89 @@ class TypedDataGetRangeTest {
     // -------- Source --------
 
     @Test
-    fun bounded_on_source_slices_inclusive() {
+    fun bounded_on_source_slices_inclusive(): Unit = runBlocking {
+        // Both ends inclusive (RFC 9110): bytes=10-19 is 10 bytes, and must match the Bytes path exactly.
         val out = typedSource().getRange(HttpRange.Bounded(10, 19), size).data.bytes()
-        // Bounded.size == rangeEnd - rangeStart, so 19-10 = 9 bytes emitted.
-        // This matches the current contract of getRange for streaming data.
-        assertEquals(9, out.size)
-        assertEquals(payload.sliceArray(10..18).toList(), out.toList())
+        assertEquals(10, out.size)
+        assertEquals(payload.sliceArray(10..19).toList(), out.toList())
     }
 
     @Test
-    fun until_end_on_source_returns_tail() {
+    fun single_byte_bounded_on_source_is_not_empty(): Unit = runBlocking {
+        // A `bytes=42-42` probe (common from video players) must return exactly that one byte, not an empty body.
+        val out = typedSource().getRange(HttpRange.Bounded(42, 42), size).data.bytes()
+        assertEquals(1, out.size)
+        assertEquals(payload[42], out[0])
+    }
+
+    @Test
+    fun until_end_on_source_returns_tail(): Unit = runBlocking {
         val out = typedSource().getRange(HttpRange.UntilEnd(200), size).data.bytes()
         assertEquals(payload.sliceArray(200..255).toList(), out.toList())
     }
 
     @Test
-    fun last_on_source_returns_suffix() {
+    fun last_on_source_returns_suffix(): Unit = runBlocking {
         val out = typedSource().getRange(HttpRange.Last(16), size).data.bytes()
         assertEquals(payload.sliceArray(240..255).toList(), out.toList())
     }
 
     @Test
-    fun source_range_produces_sink_data() {
+    fun source_range_stays_streaming(): Unit = runBlocking {
+        // A range over a streaming source must stay streaming (not be materialized to Bytes) so large media served
+        // from a Source is sliced without buffering the whole payload in the heap.
         val ranged = typedSource().getRange(HttpRange.Bounded(0, 1), size)
-        assert(ranged.data is Data.Sink) { "Expected ranged source to become Data.Sink, got ${ranged.data::class}" }
+        assert(ranged.data is Data.SuspendingProducer) {
+            "Expected ranged source to stream via Data.SuspendingProducer, got ${ranged.data::class}"
+        }
     }
 
     // -------- Sink --------
 
     @Test
-    fun bounded_on_sink_slices_inclusive() {
+    fun bounded_on_sink_slices_inclusive(): Unit = runBlocking {
         val out = typedSink().getRange(HttpRange.Bounded(100, 109), size).data.bytes()
-        assertEquals(9, out.size)
-        assertEquals(payload.sliceArray(100..108).toList(), out.toList())
+        assertEquals(10, out.size)
+        assertEquals(payload.sliceArray(100..109).toList(), out.toList())
     }
 
     @Test
-    fun until_end_on_sink_returns_tail() {
+    fun until_end_on_sink_returns_tail(): Unit = runBlocking {
         val out = typedSink().getRange(HttpRange.UntilEnd(128), size).data.bytes()
         assertEquals(payload.sliceArray(128..255).toList(), out.toList())
     }
 
     @Test
-    fun last_on_sink_returns_suffix() {
+    fun last_on_sink_returns_suffix(): Unit = runBlocking {
         val out = typedSink().getRange(HttpRange.Last(32), size).data.bytes()
         assertEquals(payload.sliceArray(224..255).toList(), out.toList())
     }
 
     @Test
-    fun sink_range_produces_sink_data() {
+    fun sink_range_stays_streaming(): Unit = runBlocking {
         val ranged = typedSink().getRange(HttpRange.UntilEnd(0), size)
-        assert(ranged.data is Data.Sink) { "Expected ranged sink to remain Data.Sink, got ${ranged.data::class}" }
+        assert(ranged.data is Data.SuspendingProducer) {
+            "Expected ranged sink to stream via Data.SuspendingProducer, got ${ranged.data::class}"
+        }
     }
 
     // -------- Edge cases --------
 
     @Test
-    fun single_byte_bounded_on_bytes() {
+    fun single_byte_bounded_on_bytes(): Unit = runBlocking {
         val out = typedBytes().getRange(HttpRange.Bounded(42, 42), size).data.bytes()
         assertEquals(1, out.size)
         assertEquals(payload[42], out[0])
     }
 
     @Test
-    fun until_end_from_zero_returns_full_payload_for_source() {
+    fun until_end_from_zero_returns_full_payload_for_source(): Unit = runBlocking {
         val out = typedSource().getRange(HttpRange.UntilEnd(0), size).data.bytes()
         assertEquals(payload.toList(), out.toList())
     }
 
     @Test
-    fun last_full_size_returns_full_payload_for_sink() {
+    fun last_full_size_returns_full_payload_for_sink(): Unit = runBlocking {
         val out = typedSink().getRange(HttpRange.Last(size), size).data.bytes()
         assertEquals(payload.toList(), out.toList())
     }
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 6beda83e8..4e21de32b 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -30,7 +30,7 @@ orgCrac = "0.1.3"
 proguard = "7.9.1"
 scrimage = "4.5.1"
 serializationLibs = "1.11.0"
-serviceAbstractions = "1.2.0-5-295db54f"
+serviceAbstractions = "1.2.0-9-dad48bf4"
 shadow = "8.1.1"
 vanniktechMavenPublish = "0.36.0"
 webauthn4jCore = "0.31.3.RELEASE"