fix: close the full-capture source best-effort and exactly once

On the full-capture path, drainAndCache() closed the source and only
then set the single-close guard. If that close() threw, the guard was
left unset and control fell into the catch block, which closed the same
source a second time — the double-close the wrapper's guard exists to
prevent (some sockets/streams throw on double-close). The close failure
was also stored as a drainError, so source() then re-threw on every call
and the caller could no longer read the complete body that had already
been buffered.

Close the source as best-effort on this path: swallow a close failure
(mirroring the read-failure handler) and mark the delegate closed whether
or not close() succeeds. A failure to release the handle after a complete
capture is cleanup, not a capture failure, so the buffered body stays
readable, captureException stays null, and the source is closed once.

Add tests for a fully-captured source whose close() throws: the body
remains readable and unpoisoned, and the source is closed exactly once
across the drain and a later wrapper close. Also fold the close-guard
rationale into the closeDelegateOnce KDoc to drop a duplicated comment.

Author: OmarAlJarrah

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

@@ -207,14 +207,15 @@ public class LoggableResponseBody
          * 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].
+         *
+         * The guard flips in a `finally`, so a delegate whose `close()` throws is still marked
+         * closed: the failing close propagates to the caller, but the delegate is never closed a
+         * second time. This matches the drain path, which likewise marks the delegate closed after
+         * a best-effort source close so a later [close] is a no-op.
          */
         @Throws(IOException::class)
         private fun closeDelegateOnce() {
             if (delegateClosed) return
-            // Flip the guard whether or not close() succeeds: a delegate whose handle is not safe
-            // to close twice must still see exactly one close even when that close throws. Marking
-            // it closed in a finally also matches the drain-path error handler, which marks the
-            // delegate closed after a failed source close so a later close() is a no-op.
             try {
                 delegate.close()
             } finally {
@@ -242,10 +243,11 @@ public class LoggableResponseBody
          * the source was never obtained, so the delegate remains open and a subsequent [close]
          * will still close it.
          *
-         * If EOF is reached within the cap the body is fully captured: the source is closed and
-         * [fullyCaptured] is set, preserving the fully-repeatable behavior. If the cap is hit with
-         * bytes still pending the delegate is **left open** and retained as [liveTail] so the
-         * consumer still receives the rest of the body via [source].
+         * If EOF is reached within the cap the body is fully captured: the source is closed
+         * (best-effort — a close failure does not become a [drainError], since the capture already
+         * succeeded) and [fullyCaptured] is set, preserving the fully-repeatable behavior. If the
+         * cap is hit with bytes still pending the delegate is **left open** and retained as
+         * [liveTail] so the consumer still receives the rest of the body via [source].
          *
          * If `provider.buffer()` throws (rare; would indicate a misconfigured provider), the error
          * is cached in [drainError] and an empty buffer is used as the fallback capture so
@@ -278,8 +280,19 @@ public class LoggableResponseBody
                     remaining -= n
                 }
                 if (fullyCaptured) {
-                    // Whole body captured: close the source (and via ownership the delegate).
-                    capturedSource.close()
+                    // Whole body captured: close the source (and via ownership the delegate). The
+                    // capture already succeeded, so a close failure here is best-effort cleanup, not
+                    // a drain failure — swallow it (as the read-failure handler below does) so the
+                    // complete captured body stays readable and is never reported as a [drainError].
+                    // Mark the delegate closed whether or not close() throws, so a later close() does
+                    // not close the same source a second time (some sockets / streams throw on
+                    // double-close).
+                    try {
+                        capturedSource.close()
+                    } catch (_: Throwable) {
+                        // Best-effort: the body is already fully captured; a close failure must not
+                        // poison it or leave the single-close guard unset.
+                    }
                     delegateClosed = true
                 } else {
                     // The cap was hit (or maxCaptureBytes was <= 0) with bytes still pending: retain

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

@@ -52,6 +52,34 @@ class LoggableResponseBodyTest {
         }
     }
 
+    /**
+     * A body that fits the (unbounded) cap and whose source's [close] throws every time it is
+     * called, counting invocations. Models a delegate whose handle is not safe to close twice and
+     * whose close can fail — exercises the full-capture path's best-effort, single-close behavior.
+     */
+    private fun fullyCapturedBodyWithThrowingClose(
+        text: String,
+        sourceCloseCount: AtomicInteger,
+    ): ResponseBody {
+        val backing = Io.provider.buffer().also { it.writeUtf8(text) }
+        val throwingOnClose =
+            object : BufferedSource by backing {
+                override fun close() {
+                    sourceCloseCount.incrementAndGet()
+                    throw IOException("source close failed")
+                }
+            }
+        return object : ResponseBody() {
+            override fun mediaType(): MediaType? = MediaType.parse("text/plain")
+
+            override fun contentLength(): Long = text.toByteArray(Charsets.UTF_8).size.toLong()
+
+            override fun source(): BufferedSource = throwingOnClose
+
+            override fun close() {}
+        }
+    }
+
     // ----- source() that throws before entering .use {} -----
 
     @Test
@@ -166,6 +194,44 @@ class LoggableResponseBodyTest {
         )
     }
 
+    @Test
+    fun `fully captured body whose source close throws stays readable and is not poisoned`() {
+        // Full-capture path: the body fits the cap, so the drain reads it all and then closes the
+        // source. If that close throws, the capture itself already succeeded — the complete body is
+        // in the buffer — so source() must still return it and captureException must stay null. A
+        // close failure after a complete capture is best-effort cleanup, not a drain failure.
+        val sourceCloseCount = AtomicInteger(0)
+        val wrapper = LoggableResponseBody(fullyCapturedBodyWithThrowingClose("hello", sourceCloseCount))
+
+        assertEquals(
+            "hello",
+            wrapper.source().readUtf8(),
+            "a complete capture must stay readable even when the source close fails",
+        )
+        assertNull(
+            wrapper.captureException,
+            "a successful capture must not surface a drain error for a best-effort close failure",
+        )
+    }
+
+    @Test
+    fun `fully captured body whose source close throws is closed exactly once`() {
+        // The best-effort close must still happen exactly once: the drain closes the source and,
+        // because that close threw, the guard must be marked closed so a later wrapper.close() does
+        // not close the same source a second time (some sockets / streams throw on double-close).
+        val sourceCloseCount = AtomicInteger(0)
+        val wrapper = LoggableResponseBody(fullyCapturedBodyWithThrowingClose("hello", sourceCloseCount))
+
+        wrapper.source().readUtf8()
+        wrapper.close()
+
+        assertEquals(
+            1,
+            sourceCloseCount.get(),
+            "a fully-captured source whose close() throws must be closed exactly once across drain and wrapper close",
+        )
+    }
+
     // ----- additional failure-semantics tests -----
 
     @Test