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..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 @@ -142,9 +142,15 @@ 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 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; @@ -169,16 +175,22 @@ 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. 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; falling back to eager publisher") - return HttpRequest.BodyPublishers.ofByteArray(bufferToByteArray(body)) + .log("could not buffer streaming body as replayable; failing the request") + throw IOException( + "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..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 @@ -341,6 +341,46 @@ 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). 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() + .method(Method.POST) + .url(server.url("/unbufferable").toUrl()) + .body(body) + .build() + val ex = + 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 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 + 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