fix: harden LoggableResponseBody capture against zero-read spin and double-close

Two defects in the bounded response-body capture path:

- The bounded drain loop subtracted the read count from its remaining budget and
  handled only EOF (-1) and positive reads. A delegate Source that returns 0 for a
  positive byteCount — a Source.read contract violation — made the subtraction a
  no-op, so the loop spun forever and hung whatever thread triggered the capture.
  The loop now fails fast with an IOException on a 0-read, matching TeeSink.writeAll;
  the existing failure path caches it and source() re-throws it deterministically
  while snapshot() still returns the partial bytes. A 0-read is not treated as EOF,
  which would silently truncate the body.

- On the over-cap path the wrapper retains the delegate's source as the live tail and
  hands it out inside a one-shot prefix+tail stream. That stream's close() closed the
  tail directly while the wrapper's own close() independently closed the delegate, so
  for a delegate whose source() returns the same instance and whose close() closes it,
  the underlying source was closed twice. Both paths now funnel through a single
  closeDelegateOnce() guard under the existing lock, so whichever closes first wins and
  the other is a no-op. The peek-based prefix view is still closed directly. This was
  masked by the self-guarded close in the shipped Okio adapter but was latent for a
  bring-your-own IoProvider.

Also tightens the over-cap source() to assert the live-tail invariant with checkNotNull
rather than silently falling back to a peek view, since the tail is always present once
control reaches that point.

Author: OmarAlJarrah

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/response/LoggableResponseBody.kt

@@ -143,7 +143,10 @@ public class LoggableResponseBody
                 "LoggableResponseBody capture exceeded maxCaptureBytes; source() over the live tail " +
                     "is single-use and was already consumed."
             }
-            val tail = liveTail ?: return buf.peek()
+            // On the over-cap path the live tail is always set by the time control reaches here:
+            // drainError throws above and fullyCaptured returns above, so the only remaining regime
+            // is "cap hit with bytes pending", which always assigns liveTail.
+            val tail = checkNotNull(liveTail) { "over-cap source() reached with no live tail to replay" }
             return provider.bufferedSource(PrefixThenTailSource(buf.peek(), tail))
         }
 
@@ -190,15 +193,29 @@ public class LoggableResponseBody
                 // On an over-cap capture the delegate is still open and must be closed here; the
                 // delegate owns the live-tail source, so closing the delegate releases it too (we
                 // must NOT close the live tail separately, or the source would be closed twice).
-                if (!delegateClosed) {
-                    delegate.close()
-                    delegateClosed = true
-                }
+                // Both this path and the over-cap one-shot source funnel through the same guard so
+                // whichever fires first wins and the second is a no-op.
+                closeDelegateOnce()
                 // The captured buffer intentionally survives close — it holds in-memory bytes
                 // and no network resources, and is needed for post-mortem snapshot logging.
             }
         }
 
+        /**
+         * Closes the delegate (and, by ownership, its source) at most once. Both [close] and the
+         * over-cap one-shot source close the same underlying source on the over-cap path; routing
+         * both through this single guard prevents a double-close on a delegate whose [source]
+         * returns the same instance and whose [close] closes it (some sockets / streams throw on
+         * double-close). Callers must hold [lock].
+         */
+        @Throws(IOException::class)
+        private fun closeDelegateOnce() {
+            if (!delegateClosed) {
+                delegate.close()
+                delegateClosed = true
+            }
+        }
+
         private fun ensureCaptured(): Buffer {
             captured?.let { return it }
             return lock.withLock {
@@ -244,6 +261,14 @@ public class LoggableResponseBody
                         fullyCaptured = true
                         break
                     }
+                    // A 0-read for a positive byteCount violates the Source.read contract; treating
+                    // it as a no-op would spin this loop forever. Fail fast (the catch below caches
+                    // it and source() re-throws it). Do NOT treat 0 as EOF — that would truncate.
+                    if (n == 0L) {
+                        throw IOException(
+                            "Source returned 0 for byteCount=$chunk which violates the Source.read contract",
+                        )
+                    }
                     remaining -= n
                 }
                 if (fullyCaptured) {
@@ -285,9 +310,12 @@ public class LoggableResponseBody
 
         /**
          * A one-shot [Source] that yields the captured [prefix] bytes first, then continues from
-         * the live [tail]. Used only on the over-cap path; closing it closes the tail.
+         * the live [tail]. Used only on the over-cap path. Closing it closes the independent peek
+         * [prefix] view directly, but routes the live-tail close through [closeDelegateOnce] — the
+         * same single-close guard the wrapper's own [close] uses — so the underlying delegate
+         * source is never closed twice when both this source and the wrapper are closed.
          */
-        private class PrefixThenTailSource(
+        private inner class PrefixThenTailSource(
             private val prefix: BufferedSource,
             private val tail: BufferedSource,
         ) : Source {
@@ -303,9 +331,12 @@ public class LoggableResponseBody
             @Throws(IOException::class)
             override fun close() {
                 try {
+                    // The peek view is independent of the delegate source and cheap to close.
                     prefix.close()
                 } finally {
-                    tail.close()
+                    // The tail IS the delegate's source; close it through the shared guard so the
+                    // wrapper's close() and this close() cannot both close the underlying source.
+                    lock.withLock { closeDelegateOnce() }
                 }
             }
         }

sdk-core/src/test/kotlin/org/dexpace/sdk/core/http/response/LoggableResponseBodyTest.kt

@@ -12,7 +12,9 @@ import org.dexpace.sdk.core.io.Buffer
 import org.dexpace.sdk.core.io.BufferedSource
 import org.dexpace.sdk.core.io.Io
 import org.dexpace.sdk.io.OkioIoProvider
+import org.junit.jupiter.api.assertTimeoutPreemptively
 import java.io.IOException
+import java.time.Duration
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.Executors
 import java.util.concurrent.atomic.AtomicInteger
@@ -374,4 +376,120 @@ class LoggableResponseBodyTest {
             assertEquals(payload, results[i], "Thread $i got unexpected data")
         }
     }
+
+    // ----- bounded drain must not spin on a zero-read source -----
+
+    @Test
+    fun `drain on a zero-read source fails fast instead of spinning forever`() {
+        // A delegate Source that returns 0 for a positive byteCount violates the Source.read
+        // contract. The bounded drain subtracts the read count from `remaining`, so a 0-read
+        // is a no-op that loops forever. The drain must instead fail fast with an IOException,
+        // and source() must re-throw it (rather than hang the caller / worker / completion thread).
+        val seed = Io.provider.buffer().also { it.writeUtf8("ignored") }
+        val zeroReadSource =
+            object : BufferedSource by seed.peek() {
+                override fun read(
+                    sink: Buffer,
+                    byteCount: Long,
+                ): Long = 0L
+
+                override fun close() {}
+            }
+        val delegate =
+            object : ResponseBody() {
+                override fun mediaType(): MediaType? = null
+
+                override fun contentLength(): Long = -1
+
+                override fun source(): BufferedSource = zeroReadSource
+
+                override fun close() {}
+            }
+        val wrapper = LoggableResponseBody(delegate)
+
+        assertTimeoutPreemptively(Duration.ofSeconds(5)) {
+            // The drain must terminate; it must not spin on the zero-read.
+            val ex = assertFailsWith<IOException> { wrapper.source() }
+            assertNotNull(ex)
+        }
+
+        // The drain failure is cached as an IOException and surfaced via captureException.
+        val captured = wrapper.captureException
+        assertNotNull(captured, "a zero-read drain must cache a failure, not truncate silently")
+        assertTrue(
+            captured is IOException,
+            "the cached drain failure must be an IOException, got: ${captured::class}",
+        )
+        // The partial snapshot is still available (documented failure semantics).
+        assertEquals(0, wrapper.snapshot().size, "no bytes were read before the contract violation")
+    }
+
+    // ----- over-cap live tail must be closed exactly once -----
+
+    /**
+     * A delegate that returns the SAME [BufferedSource] instance on every [source] call and whose
+     * [close] actually closes that instance, counting underlying closes. Models a BYO transport
+     * whose `source()` returns one source and whose `close()` closes it — the case where routing
+     * the live-tail close separately from the delegate close double-closes the socket.
+     */
+    private class SharedSourceBody(
+        private val text: String,
+        sourceCloseCount: AtomicInteger,
+    ) : ResponseBody() {
+        private val backing: BufferedSource = Io.provider.buffer().also { it.writeUtf8(text) }
+        private val shared: BufferedSource =
+            object : BufferedSource by backing {
+                override fun close() {
+                    sourceCloseCount.incrementAndGet()
+                }
+            }
+
+        override fun mediaType(): MediaType? = MediaType.parse("text/plain")
+
+        override fun contentLength(): Long = text.toByteArray(Charsets.UTF_8).size.toLong()
+
+        override fun source(): BufferedSource = shared
+
+        override fun close() {
+            shared.close()
+        }
+    }
+
+    @Test
+    fun `over-cap source then close closes the underlying source exactly once`() {
+        // Over-cap: the wrapper retains delegate.source() as the live tail and hands it out inside
+        // a one-shot stream. Closing that stream and then closing the wrapper must not close the
+        // same underlying source twice (some sockets / streams throw on double-close).
+        val sourceCloseCount = AtomicInteger(0)
+        val payload = "abcdefghijklmnopqrstuvwxyz" // 26 bytes > cap
+        val wrapper = LoggableResponseBody.bounded(SharedSourceBody(payload, sourceCloseCount), Io.provider, 5L)
+
+        wrapper.source().use { it.readUtf8() }
+        wrapper.close()
+
+        assertEquals(
+            1,
+            sourceCloseCount.get(),
+            "the underlying source must be closed exactly once across the one-shot stream and wrapper close",
+        )
+    }
+
+    @Test
+    fun `over-cap close then source close closes the underlying source exactly once`() {
+        // The reverse order: wrapper.close() first, then closing the already-handed-out one-shot
+        // stream. Whichever path closes the delegate first wins; the second must be a no-op.
+        val sourceCloseCount = AtomicInteger(0)
+        val payload = "abcdefghijklmnopqrstuvwxyz" // 26 bytes > cap
+        val wrapper = LoggableResponseBody.bounded(SharedSourceBody(payload, sourceCloseCount), Io.provider, 5L)
+
+        val tail = wrapper.source()
+        wrapper.close()
+        tail.close()
+
+        assertEquals(
+            1,
+            sourceCloseCount.get(),
+            "the underlying source must be closed exactly once regardless of close ordering",
+        )
+    }
 }