fix: correctness and resource-safety fixes across the HTTP stack

docs/http-body-logging-and-concurrency.md

@@ -176,39 +176,60 @@ bytes tee'd up to the failure point — useful for diagnosing a failed request.
 the delegate's replayability: `isReplayable()` delegates, and `toReplayable()` returns a new
 `LoggableRequestBody` wrapping the delegate's replayable form.
 
-### LoggableResponseBody — Drain-Once Strategy
+### LoggableResponseBody — Bounded-Capture Strategy
 
-**Problem**: Log response body content without consuming the stream that the caller needs.
+**Problem**: Log response body content without consuming the stream that the caller needs, and
+without buffering an arbitrarily large body into memory.
 
-**Solution**: Drain once. On first access, the entire delegate body is drained into an internal
-`Buffer` with a single `writeAll` (a segment transfer on Okio-backed providers) and the delegate
-is closed. All subsequent reads are served from the captured buffer:
+**Solution**: Drain a bounded prefix on first access. The wrapper reads at most `maxCaptureBytes`
+bytes from the delegate into an internal `Buffer`; what happens next depends on whether the body
+fit within that cap:
 
 ```
 First access (source / snapshot / captureException):
-  delegate.source() ──writeAll──► Buffer            (drains into pooled segments,
-  delegate closed via use {}                          then closes the source)
-                                       │
-                                       ▼
-Subsequent access:               captured.peek()
-  source() returns a fresh non-consuming view        (no copy, no consumption)
+  read up to maxCaptureBytes from delegate.source() ──► Buffer (the capture prefix)
+
+  Body fit within the cap (default, unlimited):
+    delegate closed; capture is the whole body
+        │
+        ▼
+    Subsequent access:  captured.peek()   (fresh non-consuming view, fully repeatable)
+
+  Body exceeded the cap:
+    delegate left OPEN; only the prefix is buffered
+        │
+        ▼
+    source() (once):  captured.peek()  ─then─►  live delegate tail
+                      (one-shot: replays the prefix, then continues from the wire)
 ```
 
-**Why drain-once over tee-read?**
+When the whole body fits within `maxCaptureBytes` the behavior is the classic drain-once: the
+delegate is closed and every `source()` call returns a fresh `peek()` view. When the body is
+larger than the cap, only the preview prefix is buffered, the delegate stays open, and `source()`
+returns a **one-shot** stream that replays the captured prefix and then continues from the live
+delegate tail — so the caller still receives the complete body. Because the tail is single-use, a
+second `source()` call on an over-cap body throws `IllegalStateException`.
 
-For responses, a tee-read (wrapping the source) would require the consumer to read the body to
-trigger observation — if the consumer never reads, nothing is logged. Draining ensures the body
-is always available for logging regardless of whether the consumer reads it.
+The default `maxCaptureBytes` is `Long.MAX_VALUE` (unbounded, fully repeatable). The
+instrumentation steps construct the wrapper via the internal `bounded(...)` seam capped at the
+configured `bodyPreviewMaxBytes`, so logging never buffers more than a preview while the caller
+still streams the rest.
 
-**Trade-off**: The full response is held in memory. This is acceptable for API responses
-(typically JSON, < 1 MB) but not suitable for streaming large downloads. The class documents
-this constraint and offers `snapshot(maxBytes)` for capturing only a bounded preview prefix.
+**Why bound the capture rather than always drain?**
 
-**Why `source()` returns a new non-consuming view each time:**
+A tee-read (wrapping the source) would require the consumer to read the body to trigger
+observation — if the consumer never reads, nothing is logged. Draining a bounded prefix ensures a
+preview is always available for logging regardless of whether the consumer reads, while keeping
+memory bounded and avoiding a hang on a large or unbounded streaming body.
 
-Each call returns an independent, non-consuming `BufferedSource` over the captured buffer (via
-`Buffer.peek()`). This turns a single-use body into a repeatable one — the caller can read the
-body multiple times without re-fetching or copying.
+**Trade-off**: only the preview prefix is held in memory. For a body within the cap the capture is
+the whole body and reads are fully repeatable; for a larger body the wrapper is single-consumer
+(the live tail can be read only once). `snapshot(maxBytes)` returns a bounded prefix of the
+capture, leaving the capture intact.
+
+**`contentLength()`**: returns the captured size only when the body fit within the cap (then the
+capture *is* the body); otherwise it returns the delegate's true length, since the capture is just
+a prefix.
 
 **Failure semantics**: a network error during the drain does not silently truncate. The wrapper
 retains whatever bytes were read before the failure and caches the exception:
@@ -224,8 +245,8 @@ retains whatever bytes were read before the failure and caches the exception:
 The only logging output is a raw `ByteArray`:
 
 ```kotlin
-fun snapshot(): ByteArray            // the full captured body
-fun snapshot(maxBytes: Int): ByteArray   // a bounded prefix, leaving the capture intact
+fun snapshot(): ByteArray            // the full captured prefix (the whole body when it fit the cap)
+fun snapshot(maxBytes: Int): ByteArray   // a bounded slice of the capture, leaving it intact
 ```
 
 `snapshot()` throws `IllegalStateException` if the captured size exceeds
@@ -254,6 +275,13 @@ network resource. Direct buffer access (`tee.buffer`) is unsupported and throws
 backing buffer would reach only the tap, silently corrupting the wire body. Callers must use the
 typed `write*` methods so bytes reach both sinks.
 
+`TeeSink` accepts a `tapLimit` (default `Long.MAX_VALUE`) that bounds how many bytes are mirrored
+into the tap. Once the limit is reached the tee stops copying into the tap entirely but keeps
+forwarding the **full** payload to the primary — the wire body is never truncated. The
+instrumentation steps build the request wrapper via `LoggableRequestBody.bounded(...)` with the
+limit set to `bodyPreviewMaxBytes`, so a multi-GB `FileRequestBody` upload mirrors only a bounded
+preview into memory while streaming zero-copy to the transport.
+
 ### Buffer-Based Capture
 
 `LoggableRequestBody` allocates its tap `Buffer` lazily (only on first write or snapshot) from

sdk-core/src/main/kotlin/org/dexpace/sdk/core/client/AsyncHttpClient.kt

@@ -13,8 +13,9 @@ import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.response.Response
 import org.dexpace.sdk.core.util.Futures
 import java.io.IOException
+import java.io.InterruptedIOException
 import java.util.concurrent.CompletableFuture
-import java.util.concurrent.CompletionException
+import java.util.concurrent.ExecutionException
 import java.util.concurrent.Executor
 
 /**
@@ -56,9 +57,10 @@ import java.util.concurrent.Executor
  * - `HttpClient.asAsync(executor)` wraps a blocking transport in an async facade by submitting
  *   each `execute(...)` call to the given [Executor]. Pair with a virtual-thread executor on
  *   JDK 21+ for cheap concurrency.
- * - `AsyncHttpClient.asBlocking()` returns a sync [HttpClient] that calls
- *   `executeAsync(...).join()` and unwraps the [CompletionException] so callers see the
- *   original [Exception] / [Response] semantics.
+ * - `AsyncHttpClient.asBlocking()` returns a sync [HttpClient] that blocks on
+ *   `executeAsync(...).get()` and unwraps the [ExecutionException] so callers see the
+ *   original [Exception] / [Response] semantics. The blocking wait honours
+ *   `Thread.interrupt()`.
  */
 public fun interface AsyncHttpClient : AutoCloseable {
     /**
@@ -102,23 +104,37 @@ public fun HttpClient.asAsync(executor: Executor): AsyncHttpClient =
     }
 
 /**
- * Wraps an [AsyncHttpClient] as a blocking [HttpClient] by calling `executeAsync(...).join()`
- * and unwrapping the [CompletionException] so callers see the original exception. The current
+ * Wraps an [AsyncHttpClient] as a blocking [HttpClient] by blocking on `executeAsync(...).get()`
+ * and unwrapping the [ExecutionException] so callers see the original exception. The current
  * thread blocks until the future completes — pair with virtual threads (JDK 21+) to keep
  * carrier threads available.
  *
+ * The blocking wait honours `Thread.interrupt()`: interrupting the calling thread restores the
+ * interrupt flag, cancels the in-flight future, and throws an [InterruptedIOException] (an
+ * [IOException] subtype, so the `@Throws(IOException::class)` contract holds).
+ *
  * The returned [Response] must be closed by the caller, per the [HttpClient.execute] contract.
  */
 @Throws(IOException::class)
 public fun AsyncHttpClient.asBlocking(): HttpClient =
     HttpClient { request ->
+        val future = executeAsync(request)
         try {
-            executeAsync(request).join()
-        } catch (ce: CompletionException) {
-            // `join()` wraps every exceptional completion in CompletionException; unwrap so
+            future.get()
+        } catch (ie: InterruptedException) {
+            // `get()` parks interruptibly (unlike `join()`). Restore the interrupt flag, abort
+            // the in-flight exchange, and surface an InterruptedIOException so callers' I/O
+            // error handling terminates cleanly. See RetryStep.awaitDelay for the same pattern.
+            Thread.currentThread().interrupt()
+            future.cancel(true)
+            val ioe = InterruptedIOException("Interrupted while waiting for response")
+            ioe.initCause(ie)
+            throw ioe
+        } catch (ee: ExecutionException) {
+            // `get()` wraps every exceptional completion in ExecutionException; unwrap so
             // callers' `catch (IOException)` blocks see the original failure rather than the
             // JDK wrapper. `CancellationException` from a cancelled future is unaffected — it
-            // is a RuntimeException, not a CompletionException, so it propagates as-is.
-            throw Futures.unwrap(ce)
+            // is a RuntimeException, not an ExecutionException, so it propagates as-is.
+            throw Futures.unwrap(ee)
         }
     }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/common/Headers.kt

@@ -160,6 +160,7 @@ public data class Headers private constructor(
             values: List<String>,
         ): Builder =
             apply {
+                validateValues(name, values)
                 headersMap.computeIfAbsent(sanitizeName(name)) { mutableListOf() }.addAll(values)
             }
 
@@ -179,6 +180,7 @@ public data class Headers private constructor(
             values: List<String>,
         ): Builder =
             apply {
+                validateValues(name.caseInsensitiveName, values)
                 headersMap.computeIfAbsent(name.caseInsensitiveName) { mutableListOf() }.addAll(values)
             }
 
@@ -216,6 +218,7 @@ public data class Headers private constructor(
             values: List<String>,
         ): Builder =
             apply {
+                validateValues(name, values)
                 headersMap[sanitizeName(name)] = values.toMutableList()
             }
 
@@ -243,6 +246,7 @@ public data class Headers private constructor(
             values: List<String>,
         ): Builder =
             apply {
+                validateValues(name.caseInsensitiveName, values)
                 headersMap[name.caseInsensitiveName] = values.toMutableList()
             }
 
@@ -305,5 +309,29 @@ public data class Headers private constructor(
          * so locale-sensitive folding (Turkish `i`, etc.) would be incorrect here.
          */
         private fun sanitizeName(value: String): String = value.lowercase(Locale.US).trim()
+
+        /**
+         * Rejects header values that would enable request/header splitting before they reach a
+         * transport. A bare carriage return (`\r`) or line feed (`\n`) in a value lets an
+         * attacker inject a new header or even a second request once the value is serialised;
+         * OkHttp throws unchecked on such values and the JDK transport silently drops them, so we
+         * validate here at the transport-agnostic model layer to fail fast and uniformly.
+         *
+         * Policy: reject **only** CR/LF, not OkHttp's stricter printable-ASCII-only rule. CR/LF
+         * are the splitting vector and are illegal in every transport; tightening further would
+         * reject legitimate UTF-8 values that some transports (and the JDK) accept, so the
+         * conservative CR/LF check is the right model-layer contract.
+         */
+        private fun validateValues(
+            name: String,
+            values: List<String>,
+        ) {
+            values.forEach { value ->
+                require(value.indexOf('\r') < 0 && value.indexOf('\n') < 0) {
+                    "Header value for '$name' must not contain a carriage return or line feed " +
+                        "(\\r / \\n); such characters enable request/header splitting."
+                }
+            }
+        }
     }
 }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/context/ContextStore.kt

@@ -22,6 +22,25 @@ import java.util.concurrent.ConcurrentHashMap
  *
  * Entries are removed by [CallContext.close] when callers honour the close contract.
  *
+ * ## Lifecycle and leaks
+ *
+ * Eviction is **primarily close-only**: the normal path is that [CallContext.close] removes
+ * the entry. Only the **terminal** promotion context (the [ExchangeContext] that ends the
+ * chain) should be closed — closing an earlier link uses identity-conditional [remove] so it
+ * cannot evict the live successor that replaced it. A caller that fails to close on an
+ * exception path drops an entry that **pins the full Request + Response graph** — possibly an
+ * unread response body holding a transport connection open.
+ *
+ * To stop a missed close from leaking unboundedly, the backing map is **bounded** to
+ * [MAX_TRACKED_CONTEXTS] entries: after each insert ([set] / [put]) the map is drained back
+ * under the cap, mirroring the proven bound in
+ * [DigestChallengeHandler][org.dexpace.sdk.core.http.auth.DigestChallengeHandler]'s nonce
+ * counters. The bound is a backstop, not a substitute for closing — it caps memory but a
+ * leaked entry still holds its graph alive until evicted, and eviction is by arbitrary victim
+ * (see [drainToCap]), so a still-live call could lose its registry entry under heavy leak
+ * pressure. The store is the only strong reference on the live path, so a [java.lang.ref.WeakReference]
+ * cannot be used here: it would let an in-flight context be collected mid-call.
+ *
  * ## Thread-safety
  *
  * Thread-safe. The backing map is a [ConcurrentHashMap] so concurrent calls with distinct
@@ -49,6 +68,7 @@ public object ContextStore {
     ) {
         val previous = contexts.putIfAbsent(callKey, context)
         require(previous == null) { "Call context key already registered: $callKey" }
+        drainToCap()
     }
 
     /**
@@ -61,6 +81,7 @@ public object ContextStore {
         context: CallContext,
     ) {
         contexts[callKey] = context
+        drainToCap()
     }
 
     /**
@@ -100,4 +121,29 @@ public object ContextStore {
         }
         return removed
     }
+
+    /**
+     * Drains the backing map back under [MAX_TRACKED_CONTEXTS] after an insert. Mirrors
+     * [DigestChallengeHandler][org.dexpace.sdk.core.http.auth.DigestChallengeHandler]'s
+     * post-insert drain loop: rather than a single check-then-evict-then-insert (a non-atomic
+     * race that a burst of concurrent inserts can overshoot and then sit at the cap forever),
+     * each insert drains until the map is back under the cap, so the map converges to the
+     * bound. `ConcurrentHashMap` has no insertion order, so any single key is an acceptable
+     * victim — the first the iterator yields is dropped.
+     */
+    private fun drainToCap() {
+        while (contexts.size > MAX_TRACKED_CONTEXTS) {
+            val iterator = contexts.keys.iterator()
+            if (!iterator.hasNext()) break
+            contexts.remove(iterator.next())
+        }
+    }
+
+    // Upper bound on call contexts tracked at once. The store is normally drained by
+    // close(), so this is a backstop against a missed close on an exception path leaking an
+    // entry that pins a full Request + Response graph. 4096 is large enough to never bite a
+    // realistic in-flight concurrency level (each entry is one live or recently-finished
+    // call), yet small enough to cap worst-case retained memory at a few thousand graphs.
+    // Matches the bounded-map style of DigestChallengeHandler.MAX_TRACKED_NONCES.
+    private const val MAX_TRACKED_CONTEXTS = 4096
 }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/paging/PagedIterable.kt

@@ -100,7 +100,8 @@ public class PagedIterable<T>
          * **Page ownership.** Each yielded [PagedResponse] is owned by the consumer; this method
          * does not close them. Consumers MUST `close()` each page after use (or wrap iteration
          * in `use {}` / try-with-resources) to release the underlying response body. The
-         * flattening [iterator] path closes pages automatically — only direct callers of
+         * flattening [iterator] path closes pages automatically — each page is closed eagerly,
+         * as soon as its (already fully-materialized) items are taken — so only direct callers of
          * [byPage] need to manage page lifecycle.
          *
          * [maxPages] defaults to `Long.MAX_VALUE`. **Production callers should set a finite
@@ -132,8 +133,11 @@ public class PagedIterable<T>
          * Flattens all pages into an iterator of items. Each call starts a fresh iteration
          * (re-invokes [firstPage]).
          *
-         * The iterator owns the pages it pulls through [byPage] and closes each one once its
-         * items are exhausted — callers don't need to manage page lifecycle when iterating by
+         * The iterator owns the pages it pulls through [byPage] and closes each one eagerly,
+         * as soon as its items iterator is taken. Because [PagedResponse.value] is a
+         * fully-materialized list, the items survive the close, so a partial consume
+         * (`first()` / `take(n)` / `stream().findFirst()`) never strands an open response body
+         * or pooled connection. Callers don't need to manage page lifecycle when iterating by
          * item.
          *
          * **Error propagation.** If `pages.next()` throws, the exception propagates immediately
@@ -146,7 +150,6 @@ public class PagedIterable<T>
             object : AbstractIterator<T>() {
                 private val pages = byPage().iterator()
                 private var currentItems: Iterator<T>? = null
-                private var currentPage: PagedResponse<T>? = null
 
                 override fun computeNext() {
                     // Advance through pages until we find one with items left, or pages are exhausted.
@@ -156,8 +159,6 @@ public class PagedIterable<T>
                             setNext(items.next())
                             return
                         }
-                        currentPage?.close()
-                        currentPage = null
                         currentItems = null
                         if (!pages.hasNext()) {
                             done()
@@ -167,8 +168,12 @@ public class PagedIterable<T>
                         // AbstractIterator transitions to FAILED state, so further hasNext() calls
                         // raise IllegalArgumentException — this is the correct fail-fast behavior.
                         val next = pages.next()
-                        currentPage = next
+                        // PagedResponse.value is a fully-materialized list, so we can take the items
+                        // iterator and close the page immediately. Closing eagerly means a partial
+                        // consume (first()/take(n)) never abandons an open response body or pooled
+                        // connection waiting for a later computeNext() that may never run.
                         currentItems = next.value.iterator()
+                        next.close()
                     }
                 }
             }