fix: correct transport request-body handling for body-less and unbufferable streaming bodies

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) }

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(

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)
+    }
 }

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