feat: add async retry and bearer-auth pipeline steps

sdk-core/api/sdk-core.api

@@ -138,10 +138,14 @@ public final class org/dexpace/sdk/core/http/auth/BearerToken : org/dexpace/sdk/
 public abstract interface class org/dexpace/sdk/core/http/auth/BearerTokenProvider {
 	public fun fetch (Ljava/util/List;)Lorg/dexpace/sdk/core/http/auth/BearerToken;
 	public abstract fun fetch (Ljava/util/List;Ljava/util/Map;)Lorg/dexpace/sdk/core/http/auth/BearerToken;
+	public fun fetchAsync (Ljava/util/List;)Ljava/util/concurrent/CompletableFuture;
+	public fun fetchAsync (Ljava/util/List;Ljava/util/Map;)Ljava/util/concurrent/CompletableFuture;
 }
 
 public final class org/dexpace/sdk/core/http/auth/BearerTokenProvider$DefaultImpls {
 	public static fun fetch (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;)Lorg/dexpace/sdk/core/http/auth/BearerToken;
+	public static fun fetchAsync (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;)Ljava/util/concurrent/CompletableFuture;
+	public static fun fetchAsync (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;Ljava/util/Map;)Ljava/util/concurrent/CompletableFuture;
 }
 
 public abstract interface class org/dexpace/sdk/core/http/auth/ChallengeHandler {
@@ -747,6 +751,30 @@ public final class org/dexpace/sdk/core/http/pipeline/Stage : java/lang/Enum {
 	public static fun values ()[Lorg/dexpace/sdk/core/http/pipeline/Stage;
 }
 
+public abstract class org/dexpace/sdk/core/http/pipeline/steps/AsyncAuthStep : org/dexpace/sdk/core/http/pipeline/AsyncHttpStep {
+	public fun <init> ()V
+	protected abstract fun authorizeRequestAsync (Lorg/dexpace/sdk/core/http/request/Request;)Ljava/util/concurrent/CompletableFuture;
+	protected fun authorizeRequestOnChallengeAsync (Lorg/dexpace/sdk/core/http/request/Request;Lorg/dexpace/sdk/core/http/response/Response;)Ljava/util/concurrent/CompletableFuture;
+	public final fun getStage ()Lorg/dexpace/sdk/core/http/pipeline/Stage;
+	public final fun processAsync (Lorg/dexpace/sdk/core/http/request/Request;Lorg/dexpace/sdk/core/http/pipeline/AsyncPipelineNext;)Ljava/util/concurrent/CompletableFuture;
+}
+
+public class org/dexpace/sdk/core/http/pipeline/steps/AsyncBearerTokenAuthStep : org/dexpace/sdk/core/http/pipeline/steps/AsyncAuthStep {
+	public fun <init> (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;)V
+	public fun <init> (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;Ljava/time/Duration;)V
+	public fun <init> (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;Ljava/time/Duration;Lorg/dexpace/sdk/core/util/Clock;)V
+	public fun <init> (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;Ljava/time/Duration;Lorg/dexpace/sdk/core/util/Clock;Lorg/dexpace/sdk/core/instrumentation/ClientLogger;)V
+	public synthetic fun <init> (Lorg/dexpace/sdk/core/http/auth/BearerTokenProvider;Ljava/util/List;Ljava/time/Duration;Lorg/dexpace/sdk/core/util/Clock;Lorg/dexpace/sdk/core/instrumentation/ClientLogger;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	protected fun authorizeRequestAsync (Lorg/dexpace/sdk/core/http/request/Request;)Ljava/util/concurrent/CompletableFuture;
+	protected fun authorizeRequestOnChallengeAsync (Lorg/dexpace/sdk/core/http/request/Request;Lorg/dexpace/sdk/core/http/response/Response;)Ljava/util/concurrent/CompletableFuture;
+	protected fun bearerHeaderValue (Ljava/lang/String;)Ljava/lang/String;
+}
+
+public abstract class org/dexpace/sdk/core/http/pipeline/steps/AsyncRetryStep : org/dexpace/sdk/core/http/pipeline/AsyncHttpStep {
+	public fun <init> ()V
+	public final fun getStage ()Lorg/dexpace/sdk/core/http/pipeline/Stage;
+}
+
 public abstract class org/dexpace/sdk/core/http/pipeline/steps/AuthStep : org/dexpace/sdk/core/http/pipeline/HttpStep {
 	public fun <init> ()V
 	protected abstract fun authorizeRequest (Lorg/dexpace/sdk/core/http/request/Request;)Lorg/dexpace/sdk/core/http/request/Request;
@@ -775,6 +803,19 @@ public final class org/dexpace/sdk/core/http/pipeline/steps/DefaultAsyncInstrume
 	public fun processAsync (Lorg/dexpace/sdk/core/http/request/Request;Lorg/dexpace/sdk/core/http/pipeline/AsyncPipelineNext;)Ljava/util/concurrent/CompletableFuture;
 }
 
+public class org/dexpace/sdk/core/http/pipeline/steps/DefaultAsyncRetryStep : org/dexpace/sdk/core/http/pipeline/steps/AsyncRetryStep {
+	public static final field Companion Lorg/dexpace/sdk/core/http/pipeline/steps/DefaultAsyncRetryStep$Companion;
+	public fun <init> (Ljava/util/concurrent/ScheduledExecutorService;)V
+	public fun <init> (Ljava/util/concurrent/ScheduledExecutorService;Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryOptions;)V
+	public fun <init> (Ljava/util/concurrent/ScheduledExecutorService;Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryOptions;Lorg/dexpace/sdk/core/util/Clock;)V
+	public fun <init> (Ljava/util/concurrent/ScheduledExecutorService;Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryOptions;Lorg/dexpace/sdk/core/util/Clock;Lorg/dexpace/sdk/core/instrumentation/ClientLogger;)V
+	public synthetic fun <init> (Ljava/util/concurrent/ScheduledExecutorService;Lorg/dexpace/sdk/core/http/pipeline/steps/HttpRetryOptions;Lorg/dexpace/sdk/core/util/Clock;Lorg/dexpace/sdk/core/instrumentation/ClientLogger;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
+	public fun processAsync (Lorg/dexpace/sdk/core/http/request/Request;Lorg/dexpace/sdk/core/http/pipeline/AsyncPipelineNext;)Ljava/util/concurrent/CompletableFuture;
+}
+
+public final class org/dexpace/sdk/core/http/pipeline/steps/DefaultAsyncRetryStep$Companion {
+}
+
 public final class org/dexpace/sdk/core/http/pipeline/steps/DefaultInstrumentationStep : org/dexpace/sdk/core/http/pipeline/steps/InstrumentationStep {
 	public fun <init> ()V
 	public fun <init> (Lorg/dexpace/sdk/core/http/pipeline/steps/HttpInstrumentationOptions;)V

sdk-core/src/main/kotlin/org/dexpace/sdk/core/http/auth/BearerTokenProvider.kt

@@ -7,6 +7,8 @@
 
 package org.dexpace.sdk.core.http.auth
 
+import java.util.concurrent.CompletableFuture
+
 /**
  * Source of fresh [BearerToken]s for [org.dexpace.sdk.core.http.pipeline.steps.BearerTokenAuthStep].
  *
@@ -44,4 +46,45 @@ public fun interface BearerTokenProvider {
 
     /** Convenience for callers without extra params; forwards to [fetch] with an empty map. */
     public fun fetch(scopes: List<String>): BearerToken = fetch(scopes, emptyMap())
+
+    /**
+     * Asynchronous counterpart of [fetch], used by
+     * [org.dexpace.sdk.core.http.pipeline.steps.AsyncBearerTokenAuthStep] so a token refresh
+     * never blocks the request-dispatching thread.
+     *
+     * The default implementation invokes the blocking [fetch] **on the calling thread** and
+     * wraps the outcome into an already-completed [CompletableFuture] (completing exceptionally
+     * if [fetch] throws). That default is correct but not non-blocking: a provider that talks to
+     * a remote token endpoint should override this method to dispatch the fetch off-thread —
+     * e.g. submit to an [java.util.concurrent.Executor], or call an async OAuth client — so the
+     * returned future completes without parking the caller.
+     *
+     * Per-cloud providers (GCP / Azure / Kubernetes workload identity) and OAuth
+     * token-exchange flows belong in adapter modules, not in `sdk-core`; this seam is what they
+     * override.
+     *
+     * @param scopes OAuth scopes to request; service-specific.
+     * @param params extra parameters to pass through to the token endpoint.
+     * @return a future that completes with a fresh [BearerToken], or completes exceptionally
+     *   with whatever [fetch] threw.
+     */
+    public fun fetchAsync(
+        scopes: List<String>,
+        params: Map<String, Any>,
+    ): CompletableFuture<BearerToken> =
+        try {
+            CompletableFuture.completedFuture(fetch(scopes, params))
+        } catch (t: Throwable) {
+            // A provider's blocking fetch may throw any Throwable. Surface it through the
+            // future rather than synchronously so async callers observe a uniform error model.
+            // Error subclasses (OOM, StackOverflow) are intentionally NOT special-cased here:
+            // the default just mirrors fetch()'s outcome into the future, and an Error in a
+            // user-supplied lambda is still that lambda's failure, not a JVM-fatal one for us.
+            val failed = CompletableFuture<BearerToken>()
+            failed.completeExceptionally(t)
+            failed
+        }
+
+    /** Convenience for callers without extra params; forwards to [fetchAsync] with an empty map. */
+    public fun fetchAsync(scopes: List<String>): CompletableFuture<BearerToken> = fetchAsync(scopes, emptyMap())
 }

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

@@ -0,0 +1,150 @@
+/*
+ * 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.http.pipeline.steps
+
+import org.dexpace.sdk.core.http.common.HttpHeaderName
+import org.dexpace.sdk.core.http.pipeline.AsyncHttpStep
+import org.dexpace.sdk.core.http.pipeline.AsyncPipelineNext
+import org.dexpace.sdk.core.http.pipeline.Stage
+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.util.concurrent.CompletableFuture
+
+/**
+ * Async pillar step at [Stage.AUTH] — the [AsyncHttpStep] counterpart of [AuthStep]. Stamps
+ * credentials onto outgoing requests via an async [authorizeRequestAsync] (so a token fetch /
+ * refresh never blocks the dispatching thread) and exposes the same 401 + `WWW-Authenticate`
+ * challenge hook.
+ *
+ * The stamping and challenge semantics mirror [AuthStep] exactly:
+ *
+ *  - **HTTPS-only.** On the path that attaches a credential, [processAsync] rejects non-HTTPS
+ *    schemes before any token fetch. The guard is skipped on the marker-suppressed cross-origin
+ *    re-issue path, where no credential is attached.
+ *  - **Cross-origin redirects.** A re-issue marked by the redirect step (see
+ *    [CrossOriginRedirectMarker]) is forwarded credential-free; the marker is stripped before
+ *    the request reaches the wire.
+ *  - **Challenge retry.** On a 401 carrying `WWW-Authenticate`, [authorizeRequestOnChallengeAsync]
+ *    is consulted; a non-null replacement is driven through the chain exactly once (no further
+ *    challenge handling). The default returns a future of `null` (no retry).
+ *
+ * Unlike the synchronous [AuthStep] the credential-attaching guard checks and the downstream
+ * dispatches are composed on [CompletableFuture]s so the whole flow stays non-blocking.
+ *
+ * ## Thread-safety
+ *
+ * The stage is locked at the type level via `final override`. Concrete subclasses must be safe
+ * for concurrent invocation — see [AsyncBearerTokenAuthStep] (single-flight token refresh).
+ */
+public abstract class AsyncAuthStep : AsyncHttpStep {
+    final override val stage: Stage = Stage.AUTH
+
+    final override fun processAsync(
+        request: Request,
+        next: AsyncPipelineNext,
+    ): CompletableFuture<Response> {
+        val authorizedFuture: CompletableFuture<Request> =
+            if (CrossOriginRedirectMarker.isMarked(request)) {
+                // Cross-origin redirect re-issue: strip the marker, attach no credential.
+                CompletableFuture.completedFuture(
+                    request.newBuilder()
+                        .headers(CrossOriginRedirectMarker.strip(request.headers))
+                        .build(),
+                )
+            } else {
+                val scheme = request.url.protocol
+                if (!"https".equals(scheme, ignoreCase = true)) {
+                    Futures.failed(
+                        IllegalStateException(
+                            "${this::class.simpleName} requires HTTPS to prevent credential leak " +
+                                "(URL scheme: $scheme)",
+                        ),
+                    )
+                } else {
+                    authorizeRequestAsync(request)
+                }
+            }
+
+        return authorizedFuture.thenCompose { authorized ->
+            next.copy().processAsync(authorized).thenCompose { response ->
+                handleChallenge(authorized, response, next)
+            }
+        }
+    }
+
+    /**
+     * After the first downstream attempt, applies the 401 + `WWW-Authenticate` challenge hook.
+     * Returns the response unchanged unless [authorizeRequestOnChallengeAsync] yields a non-null
+     * replacement, in which case the original 401 is closed and the replacement is driven once.
+     */
+    private fun handleChallenge(
+        authorized: Request,
+        response: Response,
+        next: AsyncPipelineNext,
+    ): CompletableFuture<Response> {
+        if (response.status.code != SC_UNAUTHORIZED) return CompletableFuture.completedFuture(response)
+        response.headers.get(HttpHeaderName.WWW_AUTHENTICATE)
+            ?: return CompletableFuture.completedFuture(response)
+
+        val challengeFuture: CompletableFuture<Request?> =
+            try {
+                authorizeRequestOnChallengeAsync(authorized, response)
+            } catch (t: Throwable) {
+                // A sync throw from the hook (caller-bug case) must still close the 401 body.
+                response.close()
+                return Futures.failed(t)
+            }
+
+        return challengeFuture.handle { retryRequest, hookError ->
+            HookOutcome(retryRequest, hookError)
+        }.thenCompose { outcome ->
+            val hookError = outcome.error
+            if (hookError != null) {
+                response.close()
+                return@thenCompose Futures.failed<Response>(Futures.unwrap(hookError))
+            }
+            val retryRequest = outcome.request ?: return@thenCompose CompletableFuture.completedFuture(response)
+            response.close()
+            next.copy().processAsync(retryRequest)
+        }
+    }
+
+    /** Carrier so the challenge future's outcome (value or error) survives [CompletableFuture.handle]. */
+    private class HookOutcome(val request: Request?, val error: Throwable?)
+
+    /**
+     * Returns a future of [request] with the credential's auth header attached. Subclasses
+     * implement the concrete async stamping (e.g. fetch-or-refresh a bearer token off-thread,
+     * then stamp `Authorization: Bearer <token>`).
+     *
+     * Called once per request before the downstream chain is invoked.
+     */
+    protected abstract fun authorizeRequestAsync(request: Request): CompletableFuture<Request>
+
+    /**
+     * Hook invoked on a 401 response that carries a `WWW-Authenticate` header. The default
+     * returns a future of `null` — surface the 401 with no retry.
+     *
+     * Subclasses override to refresh tokens or step up auth. A non-null [Request] in the
+     * returned future triggers a single retry through the downstream chain; the original 401 is
+     * closed first.
+     *
+     * @param request the request already stamped with the credential that produced the 401.
+     * @param response the 401 response; its body is still open at this point.
+     */
+    protected open fun authorizeRequestOnChallengeAsync(
+        request: Request,
+        response: Response,
+    ): CompletableFuture<Request?> = CompletableFuture.completedFuture(null)
+
+    private companion object {
+        // HTTP 401 — the only status code AsyncAuthStep responds to.
+        private const val SC_UNAUTHORIZED = 401
+    }
+}