fix: cancellation, timeout classification, and error-handling correctness

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/steps/DefaultAsyncInstrumentationStep.kt

@@ -17,6 +17,7 @@ import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.response.LoggableResponseBody
 import org.dexpace.sdk.core.http.response.Response
 import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.HeaderValueRedactor
 import org.dexpace.sdk.core.instrumentation.LoggingEvent
 import org.dexpace.sdk.core.instrumentation.MdcSnapshot
 import org.dexpace.sdk.core.instrumentation.Span
@@ -365,29 +366,17 @@ public class DefaultAsyncInstrumentationStep
                 val typed = HttpHeaderName.fromString(nameLower)
                 when {
                     options.allowedHeaderNames.contains(typed) ->
-                        ev.field(prefix + nameLower, joinHeaderValues(typed, values))
+                        ev.field(
+                            prefix + nameLower,
+                            HeaderValueRedactor.render(typed, values, options.allowedQueryParamNames),
+                        )
                     options.isRedactedHeaderNamesLoggingEnabled ->
                         ev.field(prefix + nameLower, "REDACTED")
                     // else: silently omit
                 }
             }
         }
 
-        private fun joinHeaderValues(
-            name: HttpHeaderName?,
-            values: List<String>,
-        ): String {
-            // A URL-valued header (Location, Content-Location) can carry credentials in its query
-            // or fragment; redact it through the same UrlRedactor applied to url.full.
-            val rendered =
-                if (name in URL_VALUED_HEADERS) {
-                    values.map { UrlRedactor.redactUrlValue(it, options.allowedQueryParamNames) }
-                } else {
-                    values
-                }
-            return if (rendered.size == 1) rendered[0] else rendered.joinToString(", ")
-        }
-
         private fun safeRedact(request: Request): String =
             try {
                 UrlRedactor.redact(request.url, options.allowedQueryParamNames)
@@ -451,9 +440,5 @@ public class DefaultAsyncInstrumentationStep
             // Nanoseconds in one millisecond, expressed as Double so the division returns
             // millisecond fractions (e.g. 1.234 ms) for high-resolution latency histograms.
             private const val NANOS_PER_MILLI_DOUBLE = 1_000_000.0
-
-            // Allowed headers whose value is a URL: redacted through UrlRedactor, not logged raw.
-            private val URL_VALUED_HEADERS: Set<HttpHeaderName> =
-                setOf(HttpHeaderName.LOCATION, HttpHeaderName.CONTENT_LOCATION)
         }
     }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/steps/DefaultAsyncRetryStep.kt

@@ -19,7 +19,6 @@ import org.dexpace.sdk.core.util.Clock
 import org.dexpace.sdk.core.util.Futures
 import java.io.IOException
 import java.io.InterruptedIOException
-import java.net.SocketTimeoutException
 import java.time.Duration
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.ScheduledExecutorService
@@ -293,9 +292,7 @@ public open class DefaultAsyncRetryStep
                 // DefaultRetryStep. SocketTimeoutException extends InterruptedIOException but is a
                 // retryable read timeout, not a cancellation, so it is excluded here and left to the
                 // normal retry classification below.
-                if ((exception is InterruptedIOException && exception !is SocketTimeoutException) ||
-                    exception is InterruptedException
-                ) {
+                if (RetryPolicySupport.isNonRetryableInterrupt(exception)) {
                     Thread.currentThread().interrupt()
                     failTerminally(support.asInterruptedIo(exception))
                     return

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/steps/DefaultInstrumentationStep.kt

@@ -15,6 +15,7 @@ import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.response.LoggableResponseBody
 import org.dexpace.sdk.core.http.response.Response
 import org.dexpace.sdk.core.instrumentation.ClientLogger
+import org.dexpace.sdk.core.instrumentation.HeaderValueRedactor
 import org.dexpace.sdk.core.instrumentation.LoggingEvent
 import org.dexpace.sdk.core.instrumentation.Span
 import org.dexpace.sdk.core.instrumentation.UrlRedactor
@@ -269,30 +270,17 @@ public class DefaultInstrumentationStep
                 val typed = HttpHeaderName.fromString(nameLower)
                 when {
                     options.allowedHeaderNames.contains(typed) ->
-                        ev.field(prefix + nameLower, joinHeaderValues(typed, values))
+                        ev.field(
+                            prefix + nameLower,
+                            HeaderValueRedactor.render(typed, values, options.allowedQueryParamNames),
+                        )
                     options.isRedactedHeaderNamesLoggingEnabled ->
                         ev.field(prefix + nameLower, "REDACTED")
                     // else: silently omit
                 }
             }
         }
 
-        private fun joinHeaderValues(
-            name: HttpHeaderName?,
-            values: List<String>,
-        ): String {
-            // A URL-valued header (Location, Content-Location) can carry credentials in its query
-            // or fragment (OAuth code, pre-signed signature, implicit-flow token). Redact its value
-            // through the same UrlRedactor applied to url.full instead of logging it verbatim.
-            val rendered =
-                if (name in URL_VALUED_HEADERS) {
-                    values.map { UrlRedactor.redactUrlValue(it, options.allowedQueryParamNames) }
-                } else {
-                    values
-                }
-            return if (rendered.size == 1) rendered[0] else rendered.joinToString(", ")
-        }
-
         private fun safeRedact(request: Request): String =
             try {
                 UrlRedactor.redact(request.url, options.allowedQueryParamNames)
@@ -356,10 +344,5 @@ public class DefaultInstrumentationStep
             // Nanoseconds in one millisecond, expressed as Double so the division returns
             // millisecond fractions (e.g. 1.234 ms) for high-resolution latency histograms.
             private const val NANOS_PER_MILLI_DOUBLE = 1_000_000.0
-
-            // Allowed headers whose value is a URL: their query/fragment can carry credentials, so
-            // they are redacted through UrlRedactor rather than logged verbatim.
-            private val URL_VALUED_HEADERS: Set<HttpHeaderName> =
-                setOf(HttpHeaderName.LOCATION, HttpHeaderName.CONTENT_LOCATION)
         }
     }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/steps/DefaultRetryStep.kt

@@ -18,7 +18,6 @@ import org.dexpace.sdk.core.pipeline.step.retry.RetrySettings
 import org.dexpace.sdk.core.util.Clock
 import java.io.IOException
 import java.io.InterruptedIOException
-import java.net.SocketTimeoutException
 import java.time.Duration
 
 /**
@@ -229,9 +228,7 @@ public open class DefaultRetryStep
                 // the @Throws(IOException) contract holds, with prior failures attached.
                 // SocketTimeoutException extends InterruptedIOException but is a read timeout, not a
                 // cancellation, so it is excluded here and left to the normal retry classification.
-                if ((exception is InterruptedIOException && exception !is SocketTimeoutException) ||
-                    exception is InterruptedException
-                ) {
+                if (RetryPolicySupport.isNonRetryableInterrupt(exception)) {
                     Thread.currentThread().interrupt()
                     val ioe = support.asInterruptedIo(exception)
                     suppressed?.forEach(ioe::addSuppressed)

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/pipeline/steps/RetryPolicySupport.kt

@@ -7,13 +7,14 @@
 
 package org.dexpace.sdk.core.http.pipeline.steps
 
-import org.dexpace.sdk.core.http.request.Method
+import org.dexpace.sdk.core.http.request.IDEMPOTENT_METHODS
 import org.dexpace.sdk.core.http.request.Request
 import org.dexpace.sdk.core.http.request.RequestOptions
 import org.dexpace.sdk.core.instrumentation.ClientLogger
 import org.dexpace.sdk.core.pipeline.step.retry.BackoffCalculator
 import org.dexpace.sdk.core.pipeline.step.retry.RetrySettings
 import java.io.InterruptedIOException
+import java.net.SocketTimeoutException
 import java.time.Duration
 
 /**
@@ -50,18 +51,14 @@ internal class RetryPolicySupport(
             .build()
 
     /**
-     * Resolves the retry budget for a single call from its [RequestOptions]. A per-call
-     * [RequestOptions.maxRetries] override wins only when it is non-negative; a `null` (no
-     * override) or a negative override falls back to the configured, already-clamped
-     * [options]`.maxRetries`.
-     *
-     * The negative-override fallback is deliberate: it mirrors [clampOptions]' handling of a
-     * negative *configured* `maxRetries`, so a negative per-call value means "use the configured
-     * default", NOT "0 retries" (which is what `maxRetries = 0` requests). Callers should read this
-     * once per call — the options are constant across retry re-drives.
+     * Resolves the retry budget for a single call from its [RequestOptions]. A non-`null` per-call
+     * [RequestOptions.maxRetries] override wins — the builder rejects a negative value, so a
+     * present override is always `>= 0` here — while a `null` override falls back to the configured,
+     * already-clamped [options]`.maxRetries`. `0` is a real value (one attempt, no retries), not
+     * "unset". Callers should read this once per call — the options are constant across retry
+     * re-drives.
      */
-    fun effectiveMaxRetries(callOptions: RequestOptions): Int =
-        callOptions.maxRetries?.takeIf { it >= 0 } ?: options.maxRetries
+    fun effectiveMaxRetries(callOptions: RequestOptions): Int = callOptions.maxRetries ?: options.maxRetries
 
     /**
      * Returns `true` when [request] may be re-sent: a body-less request only when its method is
@@ -127,10 +124,16 @@ internal class RetryPolicySupport(
         return opts.withMaxRetries(DefaultRetryStep.DEFAULT_MAX_RETRIES)
     }
 
-    private companion object {
-        // Methods safe to re-send regardless of body replayability (idempotent per RFC 9110).
-        // Mirrors RetrySettings.DEFAULT_RETRYABLE_METHODS.
-        private val IDEMPOTENT_METHODS: Set<Method> =
-            setOf(Method.GET, Method.HEAD, Method.OPTIONS, Method.PUT, Method.DELETE)
+    internal companion object {
+        /**
+         * Classifies [exception] as a non-retryable interrupt / cancellation, per the SDK-wide
+         * convention: an [InterruptedIOException] other than a [SocketTimeoutException] (a read
+         * timeout, which stays retryable), or a bare [InterruptedException]. Shared by
+         * [DefaultRetryStep] and [DefaultAsyncRetryStep] so the blocking and async stacks classify
+         * cancellation identically.
+         */
+        internal fun isNonRetryableInterrupt(exception: Throwable): Boolean =
+            (exception is InterruptedIOException && exception !is SocketTimeoutException) ||
+                exception is InterruptedException
     }
 }

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/request/Method.kt

@@ -47,3 +47,12 @@ public enum class Method(public val permitsRequestBody: Boolean) {
     @Suppress("MemberNameEqualsClassName")
     public val method: String get() = name
 }
+
+/**
+ * HTTP methods that are idempotent per RFC 9110 §9.2.2 and therefore safe to replay without a
+ * replayable request body. Single canonical source both retry defaults derive from:
+ * `RetrySettings.DEFAULT_RETRYABLE_METHODS` (the configurable retry allow-list) and the inherent
+ * replay-safety gate in `RetryPolicySupport`. `linkedSetOf` fixes a stable iteration order.
+ */
+internal val IDEMPOTENT_METHODS: Set<Method> =
+    linkedSetOf(Method.GET, Method.HEAD, Method.OPTIONS, Method.PUT, Method.DELETE)

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/request/RequestOptions.kt

@@ -45,7 +45,7 @@ import org.dexpace.sdk.core.generics.Builder as SdkBuilder
  *   single-attempt bound.
  * @property tags Opaque per-call tags, keyed by string. Read-only; never `null`, may be empty.
  * @property maxRetries Per-call retry-count override, or `null` to keep the retry step's configured
- *   budget. `0` disables retries for this call.
+ *   budget. `0` disables retries for this call; a negative value is rejected by the builder.
  */
 public class RequestOptions private constructor(
     public val timeout: Duration?,
@@ -123,10 +123,21 @@ public class RequestOptions private constructor(
 
         /**
          * Sets the per-call retry-count override, or clears it when [maxRetries] is `null` (fall
-         * back to the retry step's configured budget). `0` disables retries for the call.
+         * back to the retry step's configured budget).
+         *
+         * A non-`null` [maxRetries] must be non-negative: a negative value is rejected here rather
+         * than reaching the retry step, where it would be silently reinterpreted as "use the
+         * configured default" — the opposite of the caller's likely intent to suppress retries.
+         * Pass `0` to disable retries for the call, or `null` to use the transport / pipeline
+         * default.
+         *
+         * @throws IllegalArgumentException if [maxRetries] is negative.
          */
         public fun maxRetries(maxRetries: Int?): Builder =
             apply {
+                require(maxRetries == null || maxRetries >= 0) {
+                    "maxRetries must be non-negative (or null to use the transport default), was $maxRetries"
+                }
                 this.maxRetries = maxRetries
             }
 

sdk-core/src/main/kotlin/org/dexpace/sdk/core/instrumentation/HeaderValueRedactor.kt

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2026 dexpace and Omar Aljarrah
+ *
+ * Licensed under the MIT License. See LICENSE in the project root.
+ * SPDX-License-Identifier: MIT
+ */
+
+package org.dexpace.sdk.core.instrumentation
+
+import org.dexpace.sdk.core.http.common.HttpHeaderName
+
+/**
+ * Renders a header's values for structured logging, redacting URL-valued headers.
+ *
+ * Shared by the sync and async instrumentation steps so the redaction policy lives in one place
+ * — a change here (or a new URL-valued header) applies to both without drift, which is what keeps
+ * the credential-leak this closes from re-opening.
+ */
+internal object HeaderValueRedactor {
+    /**
+     * Allowed headers whose value is a URL: their query/fragment can carry credentials (an OAuth
+     * code, a pre-signed signature, an implicit-flow token), so they are redacted through
+     * [UrlRedactor] rather than logged verbatim. Kept to the plain-URL-valued headers only —
+     * headers like `Link`/`Refresh` have non-plain-URL syntax [UrlRedactor.redactUrlValue] is not
+     * built for.
+     */
+    internal val URL_VALUED_HEADERS: Set<HttpHeaderName> =
+        setOf(HttpHeaderName.LOCATION, HttpHeaderName.CONTENT_LOCATION)
+
+    /**
+     * Joins [values] into a single loggable string. When [name] is a URL-valued header, each value
+     * is redacted through [UrlRedactor.redactUrlValue] (using [allowedQueryParams] to gate which
+     * query parameters survive); otherwise values pass through verbatim.
+     */
+    internal fun render(
+        name: HttpHeaderName?,
+        values: List<String>,
+        allowedQueryParams: Set<String>,
+    ): String {
+        val rendered =
+            if (name in URL_VALUED_HEADERS) {
+                values.map { UrlRedactor.redactUrlValue(it, allowedQueryParams) }
+            } else {
+                values
+            }
+        return if (rendered.size == 1) rendered[0] else rendered.joinToString(", ")
+    }
+}