diff --git a/anthropic-java-core/src/main/kotlin/com/anthropic/core/auth/CachingAccessTokenProvider.kt b/anthropic-java-core/src/main/kotlin/com/anthropic/core/auth/CachingAccessTokenProvider.kt index f8164e2a1..a36cb4c82 100644 --- a/anthropic-java-core/src/main/kotlin/com/anthropic/core/auth/CachingAccessTokenProvider.kt +++ b/anthropic-java-core/src/main/kotlin/com/anthropic/core/auth/CachingAccessTokenProvider.kt @@ -3,6 +3,7 @@ package com.anthropic.core.auth import java.time.Duration import java.time.Instant import java.util.concurrent.CompletableFuture +import java.util.concurrent.ForkJoinPool import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -191,11 +192,10 @@ internal class CachingAccessTokenProvider( // arriving during `prior` will join us. asyncInFlight = future asyncInFlightForce = true - prior.whenComplete { _, _ -> - provider.getAsync(baseUrl, true).whenComplete { token, err -> - completeAsync(future, baseUrl, token, err, advisory = false) - } - } + prior.whenCompleteAsync( + { _, _ -> launchAsyncRefresh(baseUrl, future, force = true, advisory = false) }, + ForkJoinPool.commonPool(), + ) return future } @@ -205,9 +205,16 @@ internal class CachingAccessTokenProvider( force: Boolean, advisory: Boolean, ) { - provider.getAsync(baseUrl, force).whenComplete { token, err -> - completeAsync(future, baseUrl, token, err, advisory) - } + val refresh = + try { + provider.getAsync(baseUrl, force) + } catch (e: Exception) { + // A provider can fail before returning its future. Release the same slot + // and settle the same waiters as an exceptionally completed future. + completeAsync(future, baseUrl, null, e, advisory) + return + } + refresh.whenComplete { token, err -> completeAsync(future, baseUrl, token, err, advisory) } } private fun completeAsync( diff --git a/anthropic-java-core/src/test/kotlin/com/anthropic/auth/CachingAccessTokenProviderSynchronousFailureTest.kt b/anthropic-java-core/src/test/kotlin/com/anthropic/auth/CachingAccessTokenProviderSynchronousFailureTest.kt new file mode 100644 index 000000000..b497bcacd --- /dev/null +++ b/anthropic-java-core/src/test/kotlin/com/anthropic/auth/CachingAccessTokenProviderSynchronousFailureTest.kt @@ -0,0 +1,136 @@ +package com.anthropic.auth + +import com.anthropic.core.auth.AccessToken +import com.anthropic.core.auth.AccessTokenProvider +import com.anthropic.core.auth.CachingAccessTokenProvider +import java.time.Instant +import java.util.concurrent.CompletableFuture +import java.util.concurrent.ExecutionException +import java.util.concurrent.TimeUnit +import org.assertj.core.api.Assertions.assertThat +import org.assertj.core.api.Assertions.assertThatThrownBy +import org.junit.jupiter.api.Test + +internal class CachingAccessTokenProviderSynchronousFailureTest { + private val baseUrl = "https://api.anthropic.com" + private val failure = IllegalStateException("provider failed before returning a future") + private val fresh = AccessToken("fresh", expiresAt = null) + + private fun provider(call: (Boolean) -> CompletableFuture): AccessTokenProvider = + object : AccessTokenProvider { + override fun get(baseUrl: String, forceRefresh: Boolean): AccessToken = + throw UnsupportedOperationException("sync path not used") + + override fun getAsync(baseUrl: String, forceRefresh: Boolean) = call(forceRefresh) + } + + private fun assertFailed(future: CompletableFuture) { + assertThatThrownBy { future.get(5, TimeUnit.SECONDS) } + .isInstanceOf(ExecutionException::class.java) + .hasCause(failure) + } + + @Test + fun foregroundFailureCompletesTheFutureAndAllowsAnotherFetch() { + var calls = 0 + val cache = + CachingAccessTokenProvider( + provider { + if (++calls == 1) throw failure + CompletableFuture.completedFuture(fresh) + } + ) + assertFailed(cache.getAsync(baseUrl, false)) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(fresh) + assertThat(calls).isEqualTo(2) + } + + @Test + fun mandatoryFailureDoesNotLeaveLaterCallsWaiting() { + var calls = 0 + val cache = + CachingAccessTokenProvider( + provider { + when (++calls) { + 1 -> + CompletableFuture.completedFuture(AccessToken("expired", Instant.EPOCH)) + 2 -> throw failure + else -> CompletableFuture.completedFuture(fresh) + } + } + ) + cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS) + assertFailed(cache.getAsync(baseUrl, false)) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(fresh) + assertThat(calls).isEqualTo(3) + } + + @Test + fun advisoryFailureStillServesTheTokenAndUsesBackoff() { + var now = Instant.parse("2026-01-01T00:00:00Z") + val stale = AccessToken("stale", now.plusSeconds(60)) + var calls = 0 + val cache = + CachingAccessTokenProvider( + provider { + when (++calls) { + 1 -> CompletableFuture.completedFuture(stale) + 2 -> throw failure + else -> CompletableFuture.completedFuture(fresh) + } + }, + clock = { now }, + ) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(stale) + repeat(3) { + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(stale) + } + assertThat(calls).isEqualTo(2) + now = now.plusSeconds(6) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(stale) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(fresh) + assertThat(calls).isEqualTo(3) + } + + @Test + fun chainedForceFailureCompletesAllWaitersAndReleasesTheSlot() { + val pending = CompletableFuture() + val forces = mutableListOf() + val cache = + CachingAccessTokenProvider( + provider { force -> + forces.add(force) + when (forces.size) { + 1 -> pending + 2 -> throw failure + else -> CompletableFuture.completedFuture(fresh) + } + } + ) + val first = cache.getAsync(baseUrl, false) + val forced = cache.getAsync(baseUrl, true) + val joined = cache.getAsync(baseUrl, true) + assertThat(joined).isSameAs(forced) + pending.completeExceptionally(IllegalArgumentException("prior attempt failed")) + assertThat(first.isCompletedExceptionally).isTrue() + assertFailed(forced) + assertFailed(joined) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(fresh) + assertThat(forces).containsExactly(false, true, false) + } + + @Test + fun alreadyExceptionalFutureRetainsExistingBehavior() { + var calls = 0 + val cache = + CachingAccessTokenProvider( + provider { + if (++calls == 1) + CompletableFuture().also { it.completeExceptionally(failure) } + else CompletableFuture.completedFuture(fresh) + } + ) + assertFailed(cache.getAsync(baseUrl, false)) + assertThat(cache.getAsync(baseUrl, false).get(1, TimeUnit.SECONDS)).isSameAs(fresh) + } +}