From 4b0549e7b4672abcdf871427a9bcded003de9041 Mon Sep 17 00:00:00 2001 From: OmarAlJarrah Date: Mon, 15 Jun 2026 21:55:07 +0300 Subject: [PATCH 1/2] fix: correct transport request-body handling for body-less and unbufferable bodies The SDK's Request model treats a request body as optional for every method, but the two transports mishandled two edge cases of that contract. OkHttp transport: a body-less POST/PUT/PATCH crashed. RequestAdapter passed a null body straight to okhttp3.Request.Builder.method, which throws IllegalArgumentException for the methods OkHttp treats as requiring a body. A body-less POST is a valid, publicly-constructible SDK request, so this turned a normal request into an unchecked exception that bypassed the pipeline and diverged from the JDK transport (which already substitutes an empty body). The adapter now substitutes a shared zero-length body for POST/PUT/PATCH when no body is present; other methods keep their null body. An explicit three-token check is used rather than OkHttp's internal HttpMethod.requiresRequestBody: that symbol is internal to OkHttp's Kotlin module, and the SDK's Method enum cannot produce the WebDAV tokens (PROPPATCH/REPORT) that distinguish the two sets. JDK transport: a non-replayable streaming body whose writeTo failed mid-write surfaced the wrong exception. When toReplayable() threw, streamingPublisher fell back to bufferToByteArray, which called writeTo a second time. The SDK's consume-once bodies flip their consumed guard before writing, so the second writeTo tripped that guard and threw IllegalStateException, masking the original IOException and escaping the @Throws(IOException) contract on both execute paths. The partial bytes from the first attempt were already gone. The catch now rethrows the original IOException wrapped in an UncheckedIOException with a clear message instead of re-driving the consumed body. Adds MockWebServer coverage for body-less POST/PUT/PATCH on the OkHttp transport and a non-replayable mid-write-failure body on the JDK transport. --- .../jdkhttp/internal/BodyPublishers.kt | 28 +++++--- .../transport/jdkhttp/JdkHttpTransportTest.kt | 69 +++++++++++++++++++ .../okhttp/internal/RequestAdapter.kt | 43 ++++++++++-- .../transport/okhttp/OkHttpTransportTest.kt | 55 +++++++++++++++ 4 files changed, 180 insertions(+), 15 deletions(-) diff --git a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt index 29871b05..bfab1169 100644 --- a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt +++ b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt @@ -15,6 +15,7 @@ import java.io.InputStream import java.io.InterruptedIOException import java.io.PipedInputStream import java.io.PipedOutputStream +import java.io.UncheckedIOException import java.net.http.HttpRequest import java.util.concurrent.ExecutorService import java.util.concurrent.Executors @@ -142,9 +143,12 @@ internal object BodyPublishers { * therefore re-reads the body, so the body must be replayable for the resent request to * carry the correct bytes. If [body] is not already replayable it is buffered once into an * in-memory copy via [SdkRequestBody.toReplayable]. A body that cannot be made replayable - * (its `toReplayable` throws) falls back to a single eager byte-array publisher — correct - * for the first send and harmless on a resend — rather than a one-shot pipe that would - * corrupt the body on resubscribe. + * (its `toReplayable` throws mid-write) has already been partially consumed and cannot be + * recovered — the bytes drained by the failed buffering attempt are gone and `writeTo` + * cannot be driven a second time on a consume-once body. This method therefore fails loudly + * with an [UncheckedIOException] wrapping the original [IOException] rather than masking it + * or shipping a truncated body; callers that need resilience here must supply a replayable + * body. * * For each subscription the supplier: * 1. creates a fresh [PipedOutputStream] / [PipedInputStream] pair; @@ -169,16 +173,20 @@ internal object BodyPublishers { try { body.toReplayable() } catch (e: IOException) { - // The body could not be buffered into a replayable copy. A one-shot pipe - // would corrupt the body on the JDK's resubscribe, so fall back to a single - // eager byte-array publisher: the body has already been partially consumed - // by the failed buffering attempt, but emitting whatever was captured is - // strictly better than handing back a stream that is wrong on resend. + // toReplayable drained the body once and failed mid-write; a consume-once + // body has already flipped its guard, so a second writeTo would trip that + // guard and surface an IllegalStateException that masks this IOException. The + // partially captured bytes are local to toReplayable and gone. Fail loudly + // rather than re-driving the body or shipping a truncated copy. log.atVerbose() .event("transport.jdkhttp.body.replayable.failed") .field("error.message", e.message ?: "") - .log("could not buffer streaming body as replayable; falling back to eager publisher") - return HttpRequest.BodyPublishers.ofByteArray(bufferToByteArray(body)) + .log("could not buffer streaming body as replayable; failing the request") + throw UncheckedIOException( + "streaming request body could not be buffered for the JDK transport and " + + "has been partially consumed; supply a replayable body", + e, + ) } } return HttpRequest.BodyPublishers.ofInputStream { newSubscriptionStream(replayable) } diff --git a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt index 42e2e4c6..cbc28b42 100644 --- a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt +++ b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt @@ -24,6 +24,7 @@ import org.dexpace.sdk.io.OkioIoProvider import org.dexpace.sdk.transport.jdkhttp.internal.BodyPublishers import java.io.ByteArrayOutputStream import java.io.IOException +import java.io.UncheckedIOException import java.net.InetSocketAddress import java.net.ServerSocket import java.net.URL @@ -341,6 +342,45 @@ class JdkHttpTransportTest { assertEquals(md5(bytes), md5(received), "streaming body must round-trip intact") } + @Test + fun `streamingBodyThatCannotBeBufferedFailsLoudly`() { + // A non-replayable streaming body (> 64 KiB, so it takes the piped path) whose writeTo + // aborts with an IOException after a partial write cannot be turned into a replayable + // copy: toReplayable's internal writeTo throws, and the body — like the SDK's + // consume-once bodies — has already flipped its consumed guard. The adapter must NOT + // attempt a second writeTo (which would trip the guard and surface an + // IllegalStateException, masking the real cause and escaping the @Throws(IOException) + // contract). It must instead fail loudly with a clear UncheckedIOException that wraps + // the original IOException. + val body = PartialThenFailingBody(CommonMediaTypes.APPLICATION_OCTET_STREAM) + val request = + Request.builder() + .method(Method.POST) + .url(server.url("/unbufferable").toUrl()) + .body(body) + .build() + val ex = + assertFails { + transport.execute(request).close() + } + assertTrue( + ex is UncheckedIOException, + "expected UncheckedIOException, got ${ex::class}: ${ex.message}", + ) + assertTrue( + ex.message?.contains("supply a replayable body") == true, + "message must explain the body could not be buffered, was: ${ex.message}", + ) + val cause = ex.cause + // UncheckedIOException is a RuntimeException, so `is IOException` already excludes a + // re-wrapped IllegalStateException/UncheckedIOException — the masking failure mode. + assertTrue( + cause is IOException, + "the original IOException must be preserved as the cause, was: ${cause?.let { it::class }}", + ) + assertEquals("simulated mid-write failure", cause.message) + } + @Test fun `abandonedStreamingPublisherDoesNotStrandWriter`() { // A subscription that is acquired but never drained (connect failure, cancellation @@ -727,6 +767,35 @@ class JdkHttpTransportTest { } } + /** + * Non-replayable body that writes a partial chunk and then aborts with an [IOException]. Its + * declared length is large enough to route through the streaming (piped) publisher path, and + * a consume-once guard mirrors the SDK's stream-backed bodies: a second [writeTo] throws + * [IllegalStateException]. Used to verify the adapter does not re-drive an already-consumed + * body when buffering it into a replayable copy fails mid-write. + */ + private class PartialThenFailingBody( + private val mediaType: MediaType, + ) : RequestBody() { + private val consumed = AtomicBoolean(false) + + override fun mediaType(): MediaType = mediaType + + // > 64 KiB so adaptBody routes to the streaming publisher rather than the eager path. + override fun contentLength(): Long = 128L * 1024L + + override fun isReplayable(): Boolean = false + + override fun writeTo(sink: BufferedSink) { + check(consumed.compareAndSet(false, true)) { + "PartialThenFailingBody.writeTo was already called — the body is single-use" + } + sink.write(ByteArray(4096)) + sink.flush() + throw IOException("simulated mid-write failure") + } + } + /** Minimal [org.dexpace.sdk.core.http.auth.ChallengeHandler] stub for the M7 acceptance test. */ private object NoopChallengeHandler : org.dexpace.sdk.core.http.auth.ChallengeHandler { override fun handleChallenges( diff --git a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt index e9bf5131..88af552b 100644 --- a/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt +++ b/sdk-transport-okhttp/src/main/kotlin/org/dexpace/sdk/transport/okhttp/internal/RequestAdapter.kt @@ -9,6 +9,7 @@ package org.dexpace.sdk.transport.okhttp.internal import okhttp3.Request import okhttp3.RequestBody +import okhttp3.RequestBody.Companion.toRequestBody import org.dexpace.sdk.core.http.request.FileRequestBody import org.dexpace.sdk.core.instrumentation.ClientLogger import org.dexpace.sdk.core.http.request.Request as SdkRequest @@ -25,9 +26,13 @@ import org.dexpace.sdk.core.http.request.RequestBody as SdkRequestBody * computes them from the body / connection. * 3. Method + body: a [FileRequestBody] is wrapped in [FileRequestBodyAdapter] so the * bytes stream zero-copy through okio's [okio.FileHandle]; any other SDK body is wrapped - * in the generic [SdkRequestBodyAdapter]. Methods that require a body - * (POST/PUT/PATCH/DELETE) pass the adapter; methods without one (GET/HEAD/OPTIONS/TRACE) - * pass `null` per OkHttp's contract. + * in the generic [SdkRequestBodyAdapter]. When the SDK request carries no body, the + * adapter normally passes `null`, but OkHttp's `Request.Builder.method` rejects a null + * body for POST/PUT/PATCH (the methods OkHttp treats as requiring one). The SDK model + * makes a body optional for every method, so a body-less POST is a valid request; to keep + * it from crashing with `IllegalArgumentException`, the adapter substitutes a zero-length + * body for those methods (see [requiresRequestBody]). GET/HEAD/OPTIONS/TRACE/DELETE keep + * their `null` body. * * The adapter is stateless — the [logger] is the only field it carries so the DEBUG log * naming dropped headers attributes to the transport. @@ -52,11 +57,31 @@ internal class RequestAdapter( builder.addHeader(rawName, value) } } - val okhttpBody = request.body?.let { toOkHttpBody(it) } - builder.method(request.method.method, okhttpBody) + val methodToken = request.method.method + val okhttpBody = + when { + request.body != null -> toOkHttpBody(request.body!!) + // OkHttp rejects a null body for the methods it treats as requiring one. The + // SDK allows a body-less request for any method, so substitute an empty body + // here rather than letting Request.Builder.method throw. + requiresRequestBody(methodToken) -> EMPTY_BODY + else -> null + } + builder.method(methodToken, okhttpBody) return builder.build() } + /** + * Whether OkHttp's `Request.Builder.method` rejects a null body for [methodToken]. OkHttp's + * own set is `{POST, PUT, PATCH, PROPPATCH, REPORT}`; the SDK's [org.dexpace.sdk.core.http.request.Method] + * enum cannot produce the two WebDAV tokens, so the reachable set is exactly POST/PUT/PATCH. + * An explicit check is used instead of OkHttp's `okhttp3.internal.http.HttpMethod` because + * that is an `internal` Kotlin symbol — depending on it couples this transport to an + * unstable cross-module API for no real gain over a three-token comparison. + */ + private fun requiresRequestBody(methodToken: String): Boolean = + methodToken == "POST" || methodToken == "PUT" || methodToken == "PATCH" + /** * Selects the OkHttp [RequestBody] adapter for an SDK body. A [FileRequestBody] streams * zero-copy via [FileRequestBodyAdapter] (okio `FileHandle` → OkHttp's `BufferedSink`, @@ -68,4 +93,12 @@ internal class RequestAdapter( is FileRequestBody -> FileRequestBodyAdapter(body) else -> SdkRequestBodyAdapter(body) } + + private companion object { + /** + * Shared zero-length body substituted for a body-less POST/PUT/PATCH. Immutable and + * stateless, so a single instance is safe to reuse across requests and threads. + */ + private val EMPTY_BODY: RequestBody = ByteArray(0).toRequestBody(null, 0, 0) + } } diff --git a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt index c466801f..aad6bf12 100644 --- a/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt +++ b/sdk-transport-okhttp/src/test/kotlin/org/dexpace/sdk/transport/okhttp/OkHttpTransportTest.kt @@ -102,6 +102,61 @@ class OkHttpTransportTest { assertEquals("/echo", recorded.url.encodedPath) } + // -------- body-less methods that OkHttp requires a body for -------- + + @Test + fun executeBodylessPostSendsEmptyBody() { + // The SDK treats a body as optional for every method, so a body-less POST is a valid, + // publicly-constructible request. OkHttp's Request.Builder.method, however, rejects a + // null body for POST/PUT/PATCH. The adapter must substitute a zero-length body so the + // request dispatches with an empty payload instead of throwing IllegalArgumentException. + server.enqueue(MockResponse.Builder().code(200).build()) + val request = + Request.builder() + .method(Method.POST) + .url(server.url("/bodyless-post").toUrl()) + .build() + transport.execute(request).use { response -> + assertEquals(200, response.status.code) + } + val recorded = server.takeRequest() + assertEquals("", recorded.body?.utf8() ?: "") + assertEquals("0", recorded.headers["Content-Length"], "empty body must report zero length") + assertEquals("/bodyless-post", recorded.url.encodedPath) + } + + @Test + fun executeBodylessPutSendsEmptyBody() { + server.enqueue(MockResponse.Builder().code(200).build()) + val request = + Request.builder() + .method(Method.PUT) + .url(server.url("/bodyless-put").toUrl()) + .build() + transport.execute(request).use { response -> + assertEquals(200, response.status.code) + } + val recorded = server.takeRequest() + assertEquals("", recorded.body?.utf8() ?: "") + assertEquals("0", recorded.headers["Content-Length"], "empty body must report zero length") + } + + @Test + fun executeBodylessPatchSendsEmptyBody() { + server.enqueue(MockResponse.Builder().code(201).build()) + val request = + Request.builder() + .method(Method.PATCH) + .url(server.url("/bodyless-patch").toUrl()) + .build() + transport.execute(request).use { response -> + assertEquals(201, response.status.code) + } + val recorded = server.takeRequest() + assertEquals("", recorded.body?.utf8() ?: "") + assertEquals("0", recorded.headers["Content-Length"], "empty body must report zero length") + } + // -------- async golden paths -------- @Test From 18409aebbcf77bdfc119205a9f392d9cce7e12b3 Mon Sep 17 00:00:00 2001 From: OmarAlJarrah Date: Tue, 16 Jun 2026 18:34:40 +0300 Subject: [PATCH 2/2] fix: surface unbufferable JDK streaming body as a checked IOException MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When toReplayable() fails mid-write while buffering a non-replayable streaming body, the JDK transport now rethrows a checked IOException wrapping the original cause, instead of an UncheckedIOException. The body is already partially consumed and unrecoverable at that point, so the request must fail — but it is an I/O failure and belongs in execute()'s @Throws(IOException) contract. Throwing it unchecked bypassed that contract and diverged from the eager (<=64 KiB) path, which already propagates a mid-write failure as an IOException. The retry pipeline gates on isReplayable(), so a non-replayable body is never re-driven regardless of exception type. --- .../jdkhttp/internal/BodyPublishers.kt | 20 +++++++++++-------- .../transport/jdkhttp/JdkHttpTransportTest.kt | 16 +++++++-------- 2 files changed, 20 insertions(+), 16 deletions(-) diff --git a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt index bfab1169..38b8f151 100644 --- a/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt +++ b/sdk-transport-jdkhttp/src/main/kotlin/org/dexpace/sdk/transport/jdkhttp/internal/BodyPublishers.kt @@ -15,7 +15,6 @@ import java.io.InputStream import java.io.InterruptedIOException import java.io.PipedInputStream import java.io.PipedOutputStream -import java.io.UncheckedIOException import java.net.http.HttpRequest import java.util.concurrent.ExecutorService import java.util.concurrent.Executors @@ -145,10 +144,13 @@ internal object BodyPublishers { * in-memory copy via [SdkRequestBody.toReplayable]. A body that cannot be made replayable * (its `toReplayable` throws mid-write) has already been partially consumed and cannot be * recovered — the bytes drained by the failed buffering attempt are gone and `writeTo` - * cannot be driven a second time on a consume-once body. This method therefore fails loudly - * with an [UncheckedIOException] wrapping the original [IOException] rather than masking it - * or shipping a truncated body; callers that need resilience here must supply a replayable - * body. + * cannot be driven a second time on a consume-once body. This method therefore fails with a + * checked [IOException] that wraps the original cause rather than masking it (a second + * `writeTo` would trip a consume-once guard and surface an `IllegalStateException`) or + * shipping a truncated body. Surfacing it as a checked [IOException] keeps the transport's + * `@Throws(IOException)` contract intact and matches the eager path, which already propagates + * a mid-write failure as an [IOException]; callers that need resilience here must supply a + * replayable body. * * For each subscription the supplier: * 1. creates a fresh [PipedOutputStream] / [PipedInputStream] pair; @@ -176,13 +178,15 @@ internal object BodyPublishers { // toReplayable drained the body once and failed mid-write; a consume-once // body has already flipped its guard, so a second writeTo would trip that // guard and surface an IllegalStateException that masks this IOException. The - // partially captured bytes are local to toReplayable and gone. Fail loudly - // rather than re-driving the body or shipping a truncated copy. + // partially captured bytes are local to toReplayable and gone. Rethrow as a + // checked IOException wrapping the cause — honouring the transport's + // @Throws(IOException) contract and matching the eager path — rather than + // re-driving the body or shipping a truncated copy. log.atVerbose() .event("transport.jdkhttp.body.replayable.failed") .field("error.message", e.message ?: "") .log("could not buffer streaming body as replayable; failing the request") - throw UncheckedIOException( + throw IOException( "streaming request body could not be buffered for the JDK transport and " + "has been partially consumed; supply a replayable body", e, diff --git a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt index cbc28b42..696c6196 100644 --- a/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt +++ b/sdk-transport-jdkhttp/src/test/kotlin/org/dexpace/sdk/transport/jdkhttp/JdkHttpTransportTest.kt @@ -24,7 +24,6 @@ import org.dexpace.sdk.io.OkioIoProvider import org.dexpace.sdk.transport.jdkhttp.internal.BodyPublishers import java.io.ByteArrayOutputStream import java.io.IOException -import java.io.UncheckedIOException import java.net.InetSocketAddress import java.net.ServerSocket import java.net.URL @@ -349,9 +348,9 @@ class JdkHttpTransportTest { // copy: toReplayable's internal writeTo throws, and the body — like the SDK's // consume-once bodies — has already flipped its consumed guard. The adapter must NOT // attempt a second writeTo (which would trip the guard and surface an - // IllegalStateException, masking the real cause and escaping the @Throws(IOException) - // contract). It must instead fail loudly with a clear UncheckedIOException that wraps - // the original IOException. + // IllegalStateException, masking the real cause). It must instead fail with a checked + // IOException — honouring execute's @Throws(IOException) contract, and matching the eager + // path — that carries a clear message and preserves the original IOException as its cause. val body = PartialThenFailingBody(CommonMediaTypes.APPLICATION_OCTET_STREAM) val request = Request.builder() @@ -363,17 +362,18 @@ class JdkHttpTransportTest { assertFails { transport.execute(request).close() } + // IOException is checked; UncheckedIOException and IllegalStateException are + // RuntimeExceptions, so asserting `is IOException` rules out both a contract-bypassing + // unchecked throw and the IllegalStateException masking failure mode. assertTrue( - ex is UncheckedIOException, - "expected UncheckedIOException, got ${ex::class}: ${ex.message}", + ex is IOException, + "expected a checked IOException, got ${ex::class}: ${ex.message}", ) assertTrue( ex.message?.contains("supply a replayable body") == true, "message must explain the body could not be buffered, was: ${ex.message}", ) val cause = ex.cause - // UncheckedIOException is a RuntimeException, so `is IOException` already excludes a - // re-wrapped IllegalStateException/UncheckedIOException — the masking failure mode. assertTrue( cause is IOException, "the original IOException must be preserved as the cause, was: ${cause?.let { it::class }}",