diff --git a/.changes/next-release/feature-AWSSDKforJavav2-55b76bf.json b/.changes/next-release/feature-AWSSDKforJavav2-55b76bf.json index afbf9065e31c..ba415d9885f7 100644 --- a/.changes/next-release/feature-AWSSDKforJavav2-55b76bf.json +++ b/.changes/next-release/feature-AWSSDKforJavav2-55b76bf.json @@ -2,5 +2,5 @@ "type": "feature", "category": "AWS SDK for Java v2", "contributor": "", - "description": "Standardize cached credential stale/prefetch windows, improve credential refresh resilliancy (static-stability) and add automatic credential invalidation/refresh." + "description": "Standardize cached credential stale/prefetch windows, improve credential refresh resiliency (static-stability) and add automatic credential invalidation/refresh. Requests rejected with `ExpiredToken` or `InvalidToken` are now retried, and each retry attempt resolves credentials again." } diff --git a/core/aws-core/src/main/java/software/amazon/awssdk/awscore/internal/AwsErrorCode.java b/core/aws-core/src/main/java/software/amazon/awssdk/awscore/internal/AwsErrorCode.java index 3dabecf6289a..c93c7a68a3f6 100644 --- a/core/aws-core/src/main/java/software/amazon/awssdk/awscore/internal/AwsErrorCode.java +++ b/core/aws-core/src/main/java/software/amazon/awssdk/awscore/internal/AwsErrorCode.java @@ -61,11 +61,13 @@ public final class AwsErrorCode { possibleClockSkewErrorCodes.add("AuthFailure"); POSSIBLE_CLOCK_SKEW_ERROR_CODES = unmodifiableSet(possibleClockSkewErrorCodes); - Set retryableErrorCodes = new HashSet<>(1); + Set retryableErrorCodes = new HashSet<>(6); retryableErrorCodes.add("PriorRequestNotComplete"); retryableErrorCodes.add("RequestTimeout"); retryableErrorCodes.add("RequestTimeoutException"); retryableErrorCodes.add("InternalError"); + retryableErrorCodes.add("ExpiredToken"); + retryableErrorCodes.add("InvalidToken"); RETRYABLE_ERROR_CODES = unmodifiableSet(retryableErrorCodes); Set authErrorCodes = new HashSet<>(2); diff --git a/core/aws-core/src/main/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategy.java b/core/aws-core/src/main/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategy.java index ee694dd370f2..c07219375d7c 100644 --- a/core/aws-core/src/main/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategy.java +++ b/core/aws-core/src/main/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategy.java @@ -217,7 +217,7 @@ public static AdaptiveRetryStrategy adaptiveRetryStrategy(Boolean newRetries2026 private static boolean retryOnAwsRetryableErrors(Throwable ex) { if (ex instanceof AwsServiceException) { AwsServiceException exception = (AwsServiceException) ex; - return AwsErrorCode.RETRYABLE_ERROR_CODES.contains(exception.awsErrorDetails().errorCode()); + return AwsErrorCode.isRetryableErrorCode(exception.awsErrorDetails().errorCode()); } return false; } diff --git a/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryPolicyTest.java b/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryPolicyTest.java index b318381ca41f..1766aca3da1b 100644 --- a/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryPolicyTest.java +++ b/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryPolicyTest.java @@ -40,6 +40,18 @@ public void retriesOnRetryableErrorCodes() { assertTrue(shouldRetry(applyErrorCode("PriorRequestNotComplete"))); } + @Test + public void retriesOnAuthenticationErrorCodes() { + assertTrue(shouldRetry(applyErrorCode("ExpiredToken"))); + assertTrue(shouldRetry(applyErrorCode("InvalidToken"))); + } + + @Test + public void doesNotRetryOnAuthorizationErrorCodes() { + assertFalse(shouldRetry(applyErrorCode("AccessDenied"))); + assertFalse(shouldRetry(applyErrorCode("AccessDeniedException"))); + } + @Test public void retriesOnThrottlingExceptions() { assertTrue(shouldRetry(applyErrorCode("ThrottlingException"))); diff --git a/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategyTest.java b/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategyTest.java index 12e4f86020e5..d295814e159c 100644 --- a/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategyTest.java +++ b/core/aws-core/src/test/java/software/amazon/awssdk/awscore/retry/AwsRetryStrategyTest.java @@ -24,9 +24,11 @@ import org.junit.jupiter.params.provider.CsvSource; import software.amazon.awssdk.awscore.exception.AwsErrorDetails; import software.amazon.awssdk.awscore.exception.AwsServiceException; +import software.amazon.awssdk.core.retry.RetryMode; import software.amazon.awssdk.retries.StandardRetryStrategy; import software.amazon.awssdk.retries.api.AcquireInitialTokenRequest; import software.amazon.awssdk.retries.api.RefreshRetryTokenRequest; +import software.amazon.awssdk.retries.api.RetryStrategy; import software.amazon.awssdk.retries.api.RetryToken; import software.amazon.awssdk.retries.api.TokenAcquisitionFailedException; import software.amazon.awssdk.retries.internal.DefaultRetryToken; @@ -89,6 +91,45 @@ void standardRetryStrategy_retry21_throttlingBehaviorCorrect(String errorCode) { } } + @ParameterizedTest + @CsvSource({"STANDARD, false, ExpiredToken", + "STANDARD, false, InvalidToken", + "STANDARD, true, ExpiredToken", + "STANDARD, true, InvalidToken", + "ADAPTIVE_V2, false, ExpiredToken", + "ADAPTIVE_V2, true, InvalidToken", + "LEGACY, false, ExpiredToken", + "LEGACY, false, InvalidToken"}) + void forRetryMode_authenticationErrorCode_isRetried(RetryMode mode, boolean newRetries2026Enabled, String errorCode) { + RetryStrategy strategy = AwsRetryStrategy.forRetryMode(mode, newRetries2026Enabled); + + RetryToken token = strategy.acquireInitialToken(AcquireInitialTokenRequest.create("test")).token(); + RefreshRetryTokenRequest refresh = RefreshRetryTokenRequest.builder() + .failure(createTestException(errorCode)) + .token(token) + .build(); + + assertThat(strategy.refreshRetryToken(refresh).delay()).isGreaterThanOrEqualTo(Duration.ZERO); + } + + @ParameterizedTest + @CsvSource({"STANDARD, AccessDenied", + "STANDARD, AccessDeniedException", + "STANDARD, ExpiredTokenException", + "LEGACY, AccessDenied"}) + void forRetryMode_errorCodeThatIsNotAnAuthenticationError_isNotRetried(RetryMode mode, String errorCode) { + RetryStrategy strategy = AwsRetryStrategy.forRetryMode(mode, false); + + RetryToken token = strategy.acquireInitialToken(AcquireInitialTokenRequest.create("test")).token(); + RefreshRetryTokenRequest refresh = RefreshRetryTokenRequest.builder() + .failure(createTestException(errorCode)) + .token(token) + .build(); + + assertThatThrownBy(() -> strategy.refreshRetryToken(refresh)) + .isInstanceOf(TokenAcquisitionFailedException.class); + } + private static AwsServiceException createTestException(String errorCode) { AwsErrorDetails details = AwsErrorDetails.builder() .errorCode(errorCode) diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/SelectedAuthScheme.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/SelectedAuthScheme.java index 26f1645692d0..9246c188313d 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/SelectedAuthScheme.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/SelectedAuthScheme.java @@ -84,6 +84,10 @@ public AuthSchemeOption authSchemeOption() { return authSchemeOption; } + /** + * The identity provider that resolved the {@link #identity()}, or null if it is unknown. When present, it is used to + * invalidate the identity after an authentication error, and to resolve the identity again before each retry attempt. + */ public IdentityProvider identityProvider() { return identityProvider; } @@ -121,7 +125,11 @@ public interface Builder { Builder authSchemeOption(AuthSchemeOption authSchemeOption); /** - * The identity provider that resolved the identity. Used for invalidation on auth errors. + * The identity provider that resolved the identity. Used for invalidation on auth errors, and to resolve the identity + * again, using the identity properties of the {@link #authSchemeOption(AuthSchemeOption)}, before each retry attempt. + * + *

When an identity provider is set, the {@link #identity(CompletableFuture)} must have been resolved by it. A + * retry attempt replaces the identity with one newly resolved from this provider. */ Builder identityProvider(IdentityProvider identityProvider); diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/http/auth/AuthSchemeResolver.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/http/auth/AuthSchemeResolver.java index befba9595790..f21cc74587cc 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/http/auth/AuthSchemeResolver.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/http/auth/AuthSchemeResolver.java @@ -15,7 +15,6 @@ package software.amazon.awssdk.core.http.auth; -import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -29,22 +28,17 @@ import software.amazon.awssdk.core.exception.SdkException; import software.amazon.awssdk.core.interceptor.ExecutionAttributes; import software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute; -import software.amazon.awssdk.core.internal.util.MetricUtils; -import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.core.internal.http.pipeline.stages.utils.IdentityResolutionHelper; import software.amazon.awssdk.core.spi.identity.AuthSchemeOptionsResolver; import software.amazon.awssdk.core.spi.identity.RequestIdentityProviderResolver; import software.amazon.awssdk.http.auth.spi.scheme.AuthScheme; import software.amazon.awssdk.http.auth.spi.scheme.AuthSchemeOption; import software.amazon.awssdk.http.auth.spi.signer.HttpSigner; import software.amazon.awssdk.http.auth.spi.signer.SignerProperty; -import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; import software.amazon.awssdk.identity.spi.Identity; import software.amazon.awssdk.identity.spi.IdentityProvider; import software.amazon.awssdk.identity.spi.IdentityProviders; -import software.amazon.awssdk.identity.spi.ResolveIdentityRequest; -import software.amazon.awssdk.identity.spi.TokenIdentity; import software.amazon.awssdk.metrics.MetricCollector; -import software.amazon.awssdk.metrics.SdkMetric; import software.amazon.awssdk.utils.Logger; /** @@ -280,11 +274,8 @@ private static SelectedAuthScheme trySelectAuthScheme( return null; } - ResolveIdentityRequest.Builder identityRequestBuilder = ResolveIdentityRequest.builder(); - authOption.forEachIdentityProperty(identityRequestBuilder::putProperty); - - CompletableFuture identity = resolveIdentity( - identityProvider, identityRequestBuilder.build(), metricCollector); + CompletableFuture identity = IdentityResolutionHelper.resolveIdentity( + identityProvider, IdentityResolutionHelper.resolveIdentityRequest(authOption), metricCollector); return SelectedAuthScheme.builder() .identity(identity) @@ -293,27 +284,4 @@ private static SelectedAuthScheme trySelectAuthScheme( .identityProvider(identityProvider) .build(); } - - private static CompletableFuture resolveIdentity( - IdentityProvider identityProvider, - ResolveIdentityRequest request, - MetricCollector metricCollector) { - - SdkMetric metric = getIdentityMetric(identityProvider); - if (metric == null || metricCollector == null) { - return identityProvider.resolveIdentity(request); - } - return MetricUtils.reportDuration(() -> identityProvider.resolveIdentity(request), metricCollector, metric); - } - - private static SdkMetric getIdentityMetric(IdentityProvider identityProvider) { - Class identityType = identityProvider.identityType(); - if (identityType == AwsCredentialsIdentity.class) { - return CoreMetric.CREDENTIALS_FETCH_DURATION; - } - if (identityType == TokenIdentity.class) { - return CoreMetric.TOKEN_FETCH_DURATION; - } - return null; - } } diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java index 6d96569ffdb3..7595ea09d04f 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java @@ -21,11 +21,14 @@ import java.time.Duration; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import software.amazon.awssdk.annotations.SdkInternalApi; import software.amazon.awssdk.core.Response; import software.amazon.awssdk.core.async.AsyncRequestBody; +import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption; import software.amazon.awssdk.core.client.config.SdkClientOption; import software.amazon.awssdk.core.exception.SdkException; import software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute; @@ -51,6 +54,7 @@ public final class AsyncRetryableStage implements RequestPipeline> responseHandler; private final RequestPipeline>> requestPipeline; private final ScheduledExecutorService scheduledExecutor; + private final Executor futureCompletionExecutor; private final HttpClientDependencies dependencies; public AsyncRetryableStage(TransformingAsyncResponseHandler> responseHandler, @@ -59,6 +63,8 @@ public AsyncRetryableStage(TransformingAsyncResponseHandler> r this.responseHandler = responseHandler; this.dependencies = dependencies; this.scheduledExecutor = dependencies.clientConfiguration().option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE); + this.futureCompletionExecutor = + dependencies.clientConfiguration().option(SdkAdvancedAsyncClientOption.FUTURE_COMPLETION_EXECUTOR); this.requestPipeline = requestPipeline; } @@ -116,6 +122,7 @@ private void attemptExecute(CompletableFuture> future) { CompletableFuture> responseFuture; try { retryableStageHelper.startingAttempt(); + retryableStageHelper.resolveIdentityForAttempt(); retryableStageHelper.logSendingRequest(); responseFuture = requestPipeline.execute(retryableStageHelper.requestToSend(), context); @@ -183,13 +190,34 @@ public void maybeAttemptExecute(CompletableFuture> future) { Duration successDelay = backoffDelay.left().get(); retryableStageHelper.logBackingOff(successDelay); long totalDelayMillis = successDelay.toMillis(); - scheduleOrFail(() -> attemptExecute(future), totalDelayMillis, MILLISECONDS, future); + scheduleOrFail(() -> attemptRetryExecute(future), totalDelayMillis, MILLISECONDS, future); } catch (Throwable t) { future.completeExceptionally(t); } }); } + /** + * Runs a retry attempt on the future completion executor, off the scheduled executor that runs the client's timers. + * A retry attempt resolves the identity again, which can block while credentials are refreshed. Falls back to the + * current thread if there is no future completion executor or it rejects the attempt. + */ + private void attemptRetryExecute(CompletableFuture> future) { + if (futureCompletionExecutor == null) { + attemptExecute(future); + return; + } + try { + futureCompletionExecutor.execute(() -> attemptExecute(future)); + } catch (RejectedExecutionException e) { + LOG.debug(() -> String.format("Could not run the retry attempt on the provided FUTURE_COMPLETION_EXECUTOR. " + + "The attempt will run on thread %s. This may be an indication that the " + + "executor is being overwhelmed by too many requests.", + Thread.currentThread().getName()), e); + attemptExecute(future); + } + } + private void maybeRetryExecute(CompletableFuture> future, Exception exception) { retryableStageHelper.setLastException(exception); try { diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/RetryableStage.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/RetryableStage.java index 3407429f7455..b47cba067245 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/RetryableStage.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/RetryableStage.java @@ -59,6 +59,7 @@ public Response execute(SdkHttpFullRequest request, RequestExecutionCon while (true) { try { retryableStageHelper.startingAttempt(); + retryableStageHelper.resolveIdentityForAttempt(); Response response = executeRequest(retryableStageHelper, context); retryableStageHelper.recordAttemptSucceeded(); return response; diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/AuthErrorInvalidationHelper.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/AuthErrorInvalidationHelper.java index 10320bbf5c7e..a41014c45bfe 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/AuthErrorInvalidationHelper.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/AuthErrorInvalidationHelper.java @@ -34,9 +34,11 @@ * {@link SelectedAuthScheme} from the execution context and calls * {@link IdentityProvider#invalidate} so that the provider refreshes before it vends credentials again. * - *

Identity is resolved once per API call, by a stage that sits outside the retry loop, and every attempt of that - * call reuses it. Invalidation therefore does not affect the attempt currently being retried; it takes effect on the - * next API call that resolves credentials. + *

Each retry attempt resolves the identity again (see {@link IdentityResolutionHelper}), so invalidation takes effect on + * the retry of the rejected request, as well as on later API calls. Invalidation is best-effort and the retry path does not + * wait for it. A retry observes it when it completes before the retry attempt resolves the identity, which is always the case + * for the SDK's identity providers because they invalidate inline. If an invalidation is still in progress, the retry + * attempt may reuse the rejected identity. * *

Both the synchronous and asynchronous request paths must call * {@link #invalidateIfAuthError(Throwable, RequestExecutionContext)} for the behavior to apply to both client types. @@ -62,8 +64,8 @@ private AuthErrorInvalidationHelper() { * resolved by the time a response has been received, so the invalidation completes inline. * *

The returned future never completes exceptionally. Invalidation is best-effort, and any failure is logged at - * debug level instead of being propagated. Callers are not required to await it: identity is resolved outside the - * retry loop, so a pending invalidation cannot affect the attempt currently being retried. + * debug level instead of being propagated. Callers are not required to await it, and the retry path does not, so that a + * slow or never-completing invalidation cannot delay or block the request. * * @param exception The exception from the failed request attempt * @param context The request execution context containing auth scheme info diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/IdentityResolutionHelper.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/IdentityResolutionHelper.java new file mode 100644 index 000000000000..adb6a0e1c706 --- /dev/null +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/IdentityResolutionHelper.java @@ -0,0 +1,116 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.core.internal.http.pipeline.stages.utils; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import software.amazon.awssdk.annotations.SdkInternalApi; +import software.amazon.awssdk.core.SelectedAuthScheme; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.SdkExecutionAttribute; +import software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute; +import software.amazon.awssdk.core.internal.util.MetricUtils; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.auth.spi.scheme.AuthSchemeOption; +import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; +import software.amazon.awssdk.identity.spi.Identity; +import software.amazon.awssdk.identity.spi.IdentityProvider; +import software.amazon.awssdk.identity.spi.ResolveIdentityRequest; +import software.amazon.awssdk.identity.spi.TokenIdentity; +import software.amazon.awssdk.metrics.MetricCollector; +import software.amazon.awssdk.metrics.SdkMetric; +import software.amazon.awssdk.utils.CompletableFutureUtils; + +/** + * Resolves the identity used to sign a request, both when the auth scheme is first selected and again before each retry + * attempt. + */ +@SdkInternalApi +public final class IdentityResolutionHelper { + + private IdentityResolutionHelper() { + } + + /** + * Resolves the identity again for a retry attempt and replaces the + * {@link SdkInternalExecutionAttribute#SELECTED_AUTH_SCHEME} with a copy holding the newly resolved identity. + * The identity is resolved from the {@link SelectedAuthScheme#identityProvider()} using the identity properties of the + * selected {@link AuthSchemeOption}. + * + * @param executionAttributes The execution attributes of the request being retried. + */ + public static void reResolveIdentityForRetry(ExecutionAttributes executionAttributes) { + SelectedAuthScheme selectedAuthScheme = + executionAttributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME); + if (selectedAuthScheme == null || selectedAuthScheme.identityProvider() == null) { + return; + } + + MetricCollector metricCollector = executionAttributes.getAttribute(SdkExecutionAttribute.API_CALL_METRIC_COLLECTOR); + executionAttributes.putAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME, + withReResolvedIdentity(selectedAuthScheme, metricCollector)); + } + + private static SelectedAuthScheme withReResolvedIdentity(SelectedAuthScheme selectedAuthScheme, + MetricCollector metricCollector) { + CompletableFuture identity; + try { + identity = resolveIdentity(selectedAuthScheme.identityProvider(), + resolveIdentityRequest(selectedAuthScheme.authSchemeOption()), + metricCollector); + } catch (RuntimeException e) { + identity = CompletableFutureUtils.failedFuture(e); + } + return selectedAuthScheme.toBuilder() + .identity(identity) + .build(); + } + + /** + * Returns the request used to resolve the identity for the given auth scheme option, carrying the option's identity + * properties. + */ + public static ResolveIdentityRequest resolveIdentityRequest(AuthSchemeOption authSchemeOption) { + ResolveIdentityRequest.Builder identityRequestBuilder = ResolveIdentityRequest.builder(); + authSchemeOption.forEachIdentityProperty(identityRequestBuilder::putProperty); + return identityRequestBuilder.build(); + } + + /** + * Resolves an identity from the given provider, reporting the fetch duration to the metric collector when the identity + * type has a corresponding fetch-duration metric and a metric collector is provided. + */ + public static CompletableFuture resolveIdentity(IdentityProvider identityProvider, + ResolveIdentityRequest request, + MetricCollector metricCollector) { + SdkMetric metric = identityFetchDurationMetric(identityProvider); + if (metric == null || metricCollector == null) { + return identityProvider.resolveIdentity(request); + } + return MetricUtils.reportDuration(() -> identityProvider.resolveIdentity(request), metricCollector, metric); + } + + private static SdkMetric identityFetchDurationMetric(IdentityProvider identityProvider) { + Class identityType = identityProvider.identityType(); + if (identityType == AwsCredentialsIdentity.class) { + return CoreMetric.CREDENTIALS_FETCH_DURATION; + } + if (identityType == TokenIdentity.class) { + return CoreMetric.TOKEN_FETCH_DURATION; + } + return null; + } +} diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java index ca52e10b6025..25b6c005a86e 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelper.java @@ -102,6 +102,17 @@ public void startingAttempt() { context.executionAttributes().putAttribute(EXECUTION_ATTEMPT, attemptNumber); } + /** + * Invoke after {@link #startingAttempt()} and before executing the attempt. Identity is only re-resolved on retry attempts. + * The first attempt uses the identity resolved when the auth scheme was selected. + */ + public void resolveIdentityForAttempt() { + if (isInitialAttempt()) { + return; + } + IdentityResolutionHelper.reResolveIdentityForRetry(context.executionAttributes()); + } + /** * Invoke when starting the first attempt. This method will acquire the initial token and store it as an execution attribute. * This method returns a delay that the caller have to wait before attempting the first request. If this method returns @@ -159,6 +170,7 @@ public void recordAttemptSucceeded() { */ public Either tryRefreshToken(Duration suggestedDelay) { // Invalidate cached credentials if this failure is an auth error, before the retry strategy evaluates. + // Not awaited: invalidation is best-effort and must not delay or block the retry path. AuthErrorInvalidationHelper.invalidateIfAuthError(this.lastException, context); RetryToken retryToken; @@ -198,7 +210,7 @@ private RefreshRetryTokenResponse doBlockingRefreshRetryToken(Duration suggested public CompletableFuture> tryRefreshTokenAsync(Duration suggestedDelay) { // Invalidate cached credentials if this failure is an auth error, before the retry strategy evaluates. - // Not awaited: invalidation is best-effort and must not delay the retry path. + // Not awaited: invalidation is best-effort and must not delay or block the retry path. AuthErrorInvalidationHelper.invalidateIfAuthError(this.lastException, context); CompletableFuture> cf = new CompletableFuture<>(); diff --git a/core/sdk-core/src/main/java/software/amazon/awssdk/core/metrics/CoreMetric.java b/core/sdk-core/src/main/java/software/amazon/awssdk/core/metrics/CoreMetric.java index 60939d869dd5..87a7227e9c85 100644 --- a/core/sdk-core/src/main/java/software/amazon/awssdk/core/metrics/CoreMetric.java +++ b/core/sdk-core/src/main/java/software/amazon/awssdk/core/metrics/CoreMetric.java @@ -66,7 +66,7 @@ public final class CoreMetric { * {@link java.util.concurrent.CompletableFuture} completes, which for a streaming operation is when the response * transformer completes. * - *

{@code API_CALL_DURATION ~= CREDENTIALS_FETCH_DURATION + MARSHALLING_DURATION + ENDPOINT_RESOLVE_DURATION + + *

{@code API_CALL_DURATION ~= SUM_ALL(CREDENTIALS_FETCH_DURATION) + MARSHALLING_DURATION + ENDPOINT_RESOLVE_DURATION + * SUM_ALL(BACKOFF_DELAY_DURATION) + SUM_ALL(SIGNING_DURATION) + SUM_ALL(SERVICE_CALL_DURATION) + * SUM_ALL(UNMARSHALLING_DURATION)} * @@ -82,12 +82,16 @@ public final class CoreMetric { /** * The duration of time taken to fetch signing credentials for the API call. + * + *

Credentials are fetched again for each retry attempt, so an API call that retries reports more than one value. */ public static final SdkMetric CREDENTIALS_FETCH_DURATION = metric("CredentialsFetchDuration", Duration.class, MetricLevel.INFO); /** - * The duration of time taken to fetch signing credentials for the API call. + * The duration of time taken to fetch the signing token for the API call. + * + *

The token is fetched again for each retry attempt, so an API call that retries reports more than one value. */ public static final SdkMetric TOKEN_FETCH_DURATION = metric("TokenFetchDuration", Duration.class, MetricLevel.INFO); diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java index 659623108d13..7f57ca259fcf 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java @@ -28,7 +28,11 @@ import java.io.IOException; import java.net.URI; import java.time.Duration; +import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Phaser; import java.util.concurrent.RejectedExecutionException; @@ -47,6 +51,7 @@ import software.amazon.awssdk.core.Response; import software.amazon.awssdk.core.SdkRequest; import software.amazon.awssdk.core.SdkResponse; +import software.amazon.awssdk.core.client.config.SdkAdvancedAsyncClientOption; import software.amazon.awssdk.core.client.config.SdkClientConfiguration; import software.amazon.awssdk.core.client.config.SdkClientOption; import software.amazon.awssdk.core.exception.SdkException; @@ -518,12 +523,104 @@ void execute_isFirstAttempt_acquireInitialTokenCompleteDifferentThread_schedules verify(mockExec).schedule(any(Runnable.class), eq(0L), eq(TimeUnit.MILLISECONDS)); } + @Test + void execute_isRetryAttempt_runsAttemptOnFutureCompletionExecutor() throws Exception { + ExecutorService completionExecutor = Executors.newSingleThreadExecutor(r -> new Thread(r, "test-completion")); + try { + List attemptThreads = stubFailingAttemptsRecordingThreads(); + stubOneRetry(); + + AsyncRetryableStage retryableStage = + createRetryableStage(mockRetryStrategy, executorService, completionExecutor); + + assertThatThrownBy(retryableStage.execute(testRequest(), createRequestExecutionContext(true))::join); + + assertThat(attemptThreads).hasSize(2); + assertThat(attemptThreads.get(0)).isEqualTo(Thread.currentThread().getName()); + assertThat(attemptThreads.get(1)).isEqualTo("test-completion"); + } finally { + completionExecutor.shutdownNow(); + } + } + + @Test + void execute_isRetryAttempt_futureCompletionExecutorRejects_runsAttemptOnSchedulerThread() throws Exception { + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> new Thread(r, "test-scheduler")); + Executor rejectingExecutor = r -> { + throw new RejectedExecutionException("full"); + }; + try { + List attemptThreads = stubFailingAttemptsRecordingThreads(); + stubOneRetry(); + + AsyncRetryableStage retryableStage = + createRetryableStage(mockRetryStrategy, scheduler, rejectingExecutor); + + assertThatThrownBy(retryableStage.execute(testRequest(), createRequestExecutionContext(true))::join) + .hasRootCauseInstanceOf(SdkException.class); + + assertThat(attemptThreads).hasSize(2); + assertThat(attemptThreads.get(1)).isEqualTo("test-scheduler"); + } finally { + scheduler.shutdownNow(); + } + } + + /** + * Makes every attempt fail with a retryable 502, and returns the names of the threads that ran each attempt. + */ + private List stubFailingAttemptsRecordingThreads() throws Exception { + Response response = Response.builder() + .httpResponse(SdkHttpFullResponse.builder().statusCode(502).build()) + .isSuccess(false) + .exception(SdkException.builder().build()) + .build(); + List attemptThreads = new CopyOnWriteArrayList<>(); + when(mockDelegatePipeline.execute(any(), any())).thenAnswer(i -> { + attemptThreads.add(Thread.currentThread().getName()); + return CompletableFuture.completedFuture(response); + }); + return attemptThreads; + } + + /** + * Allows exactly one retry, with no backoff delay. + */ + private void stubOneRetry() { + AtomicBoolean first = new AtomicBoolean(); + when(mockRetryStrategy.refreshRetryTokenAsync(any())).thenAnswer(i -> { + if (first.compareAndSet(false, true)) { + return CompletableFuture.completedFuture(RefreshRetryTokenResponse.create(mockRetryToken, Duration.ZERO)); + } + return CompletableFutureUtils.failedFuture(new TokenAcquisitionFailedException("Acquire failed", + mockRetryToken, + null, + Duration.ZERO)); + }); + } + + private static SdkHttpFullRequest testRequest() { + return SdkHttpFullRequest.builder() + .method(SdkHttpMethod.GET) + .uri(URI.create("https://my-service.amazonaws.com")) + .build(); + } + private AsyncRetryableStage createRetryableStage(RetryStrategy retryStrategy, ScheduledExecutorService scheduler) { + return createRetryableStage(retryStrategy, scheduler, null); + } + + private AsyncRetryableStage createRetryableStage(RetryStrategy retryStrategy, + ScheduledExecutorService scheduler, + Executor futureCompletionExecutor) { SdkClientConfiguration clientConfig = SdkClientConfiguration.builder() .option(SdkClientOption.RETRY_STRATEGY, retryStrategy) .option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE, scheduler) + .option(SdkAdvancedAsyncClientOption + .FUTURE_COMPLETION_EXECUTOR, + futureCompletionExecutor) .build(); HttpClientDependencies deps = HttpClientDependencies.builder() diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/IdentityResolutionHelperTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/IdentityResolutionHelperTest.java new file mode 100644 index 000000000000..d4f1b7b0cb6b --- /dev/null +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/IdentityResolutionHelperTest.java @@ -0,0 +1,224 @@ +/* + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"). + * You may not use this file except in compliance with the License. + * A copy of the License is located at + * + * http://aws.amazon.com/apache2.0 + * + * or in the "license" file accompanying this file. This file is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either + * express or implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package software.amazon.awssdk.core.internal.http.pipeline.stages.utils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.Test; +import software.amazon.awssdk.core.SelectedAuthScheme; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.SdkExecutionAttribute; +import software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute; +import software.amazon.awssdk.core.metrics.CoreMetric; +import software.amazon.awssdk.http.auth.spi.scheme.AuthSchemeOption; +import software.amazon.awssdk.http.auth.spi.signer.HttpSigner; +import software.amazon.awssdk.identity.spi.AwsCredentialsIdentity; +import software.amazon.awssdk.identity.spi.Identity; +import software.amazon.awssdk.identity.spi.IdentityProperty; +import software.amazon.awssdk.identity.spi.IdentityProvider; +import software.amazon.awssdk.identity.spi.ResolveIdentityRequest; +import software.amazon.awssdk.metrics.MetricCollector; + +class IdentityResolutionHelperTest { + private static final IdentityProperty TEST_PROPERTY = + IdentityProperty.create(IdentityResolutionHelperTest.class, "TestProperty"); + + @Test + void reResolveIdentityForRetry_whenNoSelectedAuthScheme_doesNothing() { + ExecutionAttributes attributes = new ExecutionAttributes(); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + assertThat(attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME)).isNull(); + } + + @SuppressWarnings("deprecation") + @Test + void reResolveIdentityForRetry_whenNoIdentityProvider_keepsIdentity() { + SelectedAuthScheme scheme = + new SelectedAuthScheme<>(CompletableFuture.completedFuture(new TestIdentity("original")), + mockSigner(), + AuthSchemeOption.builder().schemeId("test").build()); + ExecutionAttributes attributes = attributesWith(scheme); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + assertThat(attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME)).isSameAs(scheme); + } + + @Test + void reResolveIdentityForRetry_replacesIdentityWithNewlyResolvedOne_andPreservesOtherFields() { + CountingIdentityProvider provider = new CountingIdentityProvider(); + HttpSigner signer = mockSigner(); + AuthSchemeOption option = AuthSchemeOption.builder().schemeId("test").build(); + SelectedAuthScheme scheme = SelectedAuthScheme.builder() + .identity(CompletableFuture.completedFuture( + new TestIdentity("original"))) + .signer(signer) + .authSchemeOption(option) + .identityProvider(provider) + .build(); + ExecutionAttributes attributes = attributesWith(scheme); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + SelectedAuthScheme updated = attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME); + assertThat(provider.resolveCount()).isEqualTo(1); + assertThat(((TestIdentity) updated.identity().join()).name()).isEqualTo("resolved-1"); + assertThat(updated.signer()).isSameAs(signer); + assertThat(updated.authSchemeOption()).isSameAs(option); + assertThat(updated.identityProvider()).isSameAs(provider); + } + + @Test + void reResolveIdentityForRetry_eachCallResolvesAgain() { + CountingIdentityProvider provider = new CountingIdentityProvider(); + ExecutionAttributes attributes = attributesWith(schemeWith(provider, AuthSchemeOption.builder().schemeId("test").build())); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + SelectedAuthScheme updated = attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME); + assertThat(provider.resolveCount()).isEqualTo(2); + assertThat(((TestIdentity) updated.identity().join()).name()).isEqualTo("resolved-2"); + } + + @Test + void reResolveIdentityForRetry_passesIdentityPropertiesOfAuthSchemeOption() { + CountingIdentityProvider provider = new CountingIdentityProvider(); + AuthSchemeOption option = AuthSchemeOption.builder() + .schemeId("test") + .putIdentityProperty(TEST_PROPERTY, "property-value") + .build(); + ExecutionAttributes attributes = attributesWith(schemeWith(provider, option)); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + assertThat(provider.requests()).hasSize(1); + assertThat(provider.requests().get(0).property(TEST_PROPERTY)).isEqualTo("property-value"); + } + + @Test + void reResolveIdentityForRetry_whenProviderThrows_storesFailureAsIdentity() { + RuntimeException failure = new RuntimeException("resolve failed"); + IdentityProvider provider = new CountingIdentityProvider() { + @Override + public CompletableFuture resolveIdentity(ResolveIdentityRequest request) { + throw failure; + } + }; + ExecutionAttributes attributes = attributesWith(schemeWith(provider, AuthSchemeOption.builder().schemeId("test").build())); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + SelectedAuthScheme updated = attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME); + assertThatThrownBy(() -> updated.identity().join()).isInstanceOf(CompletionException.class) + .hasCause(failure); + } + + @Test + void reResolveIdentityForRetry_reportsCredentialsFetchDurationToApiCallMetricCollector() { + IdentityProvider provider = new IdentityProvider() { + @Override + public Class identityType() { + return AwsCredentialsIdentity.class; + } + + @Override + public CompletableFuture resolveIdentity(ResolveIdentityRequest request) { + return CompletableFuture.completedFuture(AwsCredentialsIdentity.create("akid", "skid")); + } + }; + SelectedAuthScheme scheme = + SelectedAuthScheme.builder() + .identity(CompletableFuture.completedFuture(AwsCredentialsIdentity.create("old", "old"))) + .signer(mockSigner()) + .authSchemeOption(AuthSchemeOption.builder().schemeId("test").build()) + .identityProvider(provider) + .build(); + MetricCollector metricCollector = MetricCollector.create("ApiCall"); + ExecutionAttributes attributes = attributesWith(scheme); + attributes.putAttribute(SdkExecutionAttribute.API_CALL_METRIC_COLLECTOR, metricCollector); + + IdentityResolutionHelper.reResolveIdentityForRetry(attributes); + + assertThat(metricCollector.collect().metricValues(CoreMetric.CREDENTIALS_FETCH_DURATION)).hasSize(1); + } + + private static SelectedAuthScheme schemeWith(IdentityProvider provider, + AuthSchemeOption option) { + return SelectedAuthScheme.builder() + .identity(CompletableFuture.completedFuture(new TestIdentity("original"))) + .signer(mockSigner()) + .authSchemeOption(option) + .identityProvider(provider) + .build(); + } + + private static ExecutionAttributes attributesWith(SelectedAuthScheme scheme) { + ExecutionAttributes attributes = new ExecutionAttributes(); + attributes.putAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME, scheme); + return attributes; + } + + @SuppressWarnings("unchecked") + private static HttpSigner mockSigner() { + return (HttpSigner) mock(HttpSigner.class); + } + + private static final class TestIdentity implements Identity { + private final String name; + + TestIdentity(String name) { + this.name = name; + } + + String name() { + return name; + } + } + + private static class CountingIdentityProvider implements IdentityProvider { + private final AtomicInteger resolveCount = new AtomicInteger(); + private final List requests = new CopyOnWriteArrayList<>(); + + @Override + public Class identityType() { + return TestIdentity.class; + } + + @Override + public CompletableFuture resolveIdentity(ResolveIdentityRequest request) { + requests.add(request); + return CompletableFuture.completedFuture(new TestIdentity("resolved-" + resolveCount.incrementAndGet())); + } + + int resolveCount() { + return resolveCount.get(); + } + + List requests() { + return requests; + } + } +} diff --git a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelperTest.java b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelperTest.java index 9759ff75eb52..bbd57343e638 100644 --- a/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelperTest.java +++ b/core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/utils/RetryableStageHelperTest.java @@ -16,6 +16,7 @@ package software.amazon.awssdk.core.internal.http.pipeline.stages.utils; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -24,6 +25,7 @@ import java.net.URI; import java.time.Duration; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.stream.Stream; import org.junit.jupiter.api.BeforeEach; @@ -33,8 +35,10 @@ import org.junit.jupiter.params.provider.MethodSource; import org.mockito.ArgumentCaptor; import software.amazon.awssdk.core.SdkRequest; +import software.amazon.awssdk.core.SelectedAuthScheme; import software.amazon.awssdk.core.client.config.SdkClientConfiguration; import software.amazon.awssdk.core.client.config.SdkClientOption; +import software.amazon.awssdk.core.exception.SdkServiceException; import software.amazon.awssdk.core.http.ExecutionContext; import software.amazon.awssdk.core.interceptor.ExecutionAttributes; import software.amazon.awssdk.core.interceptor.SdkInternalExecutionAttribute; @@ -42,6 +46,11 @@ import software.amazon.awssdk.core.internal.http.RequestExecutionContext; import software.amazon.awssdk.http.SdkHttpFullRequest; import software.amazon.awssdk.http.SdkHttpMethod; +import software.amazon.awssdk.http.auth.spi.scheme.AuthSchemeOption; +import software.amazon.awssdk.http.auth.spi.signer.HttpSigner; +import software.amazon.awssdk.identity.spi.Identity; +import software.amazon.awssdk.identity.spi.IdentityProvider; +import software.amazon.awssdk.identity.spi.ResolveIdentityRequest; import software.amazon.awssdk.retries.api.AcquireInitialTokenResponse; import software.amazon.awssdk.retries.api.RefreshRetryTokenRequest; import software.amazon.awssdk.retries.api.RefreshRetryTokenResponse; @@ -177,6 +186,117 @@ void tryRefreshTokenAsync_refreshThrowsAcquireFailure__acquireFailureDurationRet assertThat(helper.tryRefreshTokenAsync(Duration.ZERO).join().right()).hasValue(failureAcquireDuration); } + @Test + void resolveIdentityForAttempt_onFirstAttempt_keepsIdentity() { + TrackingIdentityProvider provider = new TrackingIdentityProvider(CompletableFuture.completedFuture(null)); + SelectedAuthScheme scheme = schemeWith(provider); + ExecutionAttributes attributes = ExecutionAttributes.builder() + .put(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME, scheme) + .build(); + RetryableStageHelper helper = makeTestHelper(attributes); + + helper.startingAttempt(); + helper.resolveIdentityForAttempt(); + + assertThat(provider.resolveCount()).isZero(); + assertThat(attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME)).isSameAs(scheme); + } + + @Test + void resolveIdentityForAttempt_onEachRetryAttempt_resolvesIdentityAgain() { + TrackingIdentityProvider provider = new TrackingIdentityProvider(CompletableFuture.completedFuture(null)); + ExecutionAttributes attributes = ExecutionAttributes.builder() + .put(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME, + schemeWith(provider)) + .build(); + RetryableStageHelper helper = makeTestHelper(attributes); + + helper.startingAttempt(); + helper.resolveIdentityForAttempt(); + helper.startingAttempt(); + helper.resolveIdentityForAttempt(); + helper.startingAttempt(); + helper.resolveIdentityForAttempt(); + + assertThat(provider.resolveCount()).isEqualTo(2); + SelectedAuthScheme current = attributes.getAttribute(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME); + assertThat(current.identity().join()).isSameAs(provider.lastResolvedIdentity()); + } + + /** + * Invalidation is best-effort: an invalidation that never completes must not block the retry. + */ + @Test + void tryRefreshToken_onAuthError_invalidatesWithoutWaitingForInvalidationToComplete() { + CompletableFuture neverCompletingInvalidation = new CompletableFuture<>(); + TrackingIdentityProvider provider = new TrackingIdentityProvider(neverCompletingInvalidation); + RetryableStageHelper helper = makeTestHelper(ExecutionAttributes.builder() + .put(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME, + schemeWith(provider)) + .build()); + stubAcquireInitialToken(); + + RefreshRetryTokenResponse mockRefreshResponse = mock(RefreshRetryTokenResponse.class); + when(mockRefreshResponse.delay()).thenReturn(Duration.ofMillis(5)); + when(mockRetryStrategy.refreshRetryToken(any())).thenReturn(mockRefreshResponse); + + helper.acquireInitialToken(); + helper.setLastException(new AuthErrorException()); + + Either backoff = + assertTimeoutPreemptively(Duration.ofSeconds(5), () -> helper.tryRefreshToken(Duration.ZERO)); + + assertThat(backoff.left()).hasValue(Duration.ofMillis(5)); + assertThat(provider.invalidateCount()).isEqualTo(1); + assertThat(neverCompletingInvalidation).isNotDone(); + } + + /** + * Invalidation is best-effort: an invalidation that never completes must not delay the retry. + */ + @Test + void tryRefreshTokenAsync_onAuthError_invalidatesWithoutWaitingForInvalidationToComplete() { + CompletableFuture neverCompletingInvalidation = new CompletableFuture<>(); + TrackingIdentityProvider provider = new TrackingIdentityProvider(neverCompletingInvalidation); + RetryableStageHelper helper = makeTestHelper(ExecutionAttributes.builder() + .put(SdkInternalExecutionAttribute.SELECTED_AUTH_SCHEME, + schemeWith(provider)) + .build()); + stubAcquireInitialToken(); + + RefreshRetryTokenResponse mockRefreshResponse = mock(RefreshRetryTokenResponse.class); + when(mockRefreshResponse.delay()).thenReturn(Duration.ofMillis(5)); + when(mockRetryStrategy.refreshRetryTokenAsync(any())) + .thenReturn(CompletableFuture.completedFuture(mockRefreshResponse)); + + helper.acquireInitialToken(); + helper.setLastException(new AuthErrorException()); + + CompletableFuture> backoff = helper.tryRefreshTokenAsync(Duration.ZERO); + + assertThat(provider.invalidateCount()).isEqualTo(1); + assertThat(backoff).isCompleted(); + assertThat(backoff.join().left()).hasValue(Duration.ofMillis(5)); + assertThat(neverCompletingInvalidation).isNotDone(); + } + + private void stubAcquireInitialToken() { + AcquireInitialTokenResponse mockAcquireResponse = mock(AcquireInitialTokenResponse.class); + RetryToken token = mock(RetryToken.class); + when(mockAcquireResponse.token()).thenReturn(token); + when(mockRetryStrategy.acquireInitialToken(any())).thenReturn(mockAcquireResponse); + } + + @SuppressWarnings("unchecked") + private static SelectedAuthScheme schemeWith(IdentityProvider provider) { + return SelectedAuthScheme.builder() + .identity(CompletableFuture.completedFuture(new TestIdentity())) + .signer((HttpSigner) mock(HttpSigner.class)) + .authSchemeOption(AuthSchemeOption.builder().schemeId("test").build()) + .identityProvider(provider) + .build(); + } + RetryableStageHelper makeTestHelper(ExecutionAttributes executionAttributes) { SdkHttpFullRequest httpRequest = SdkHttpFullRequest.builder() .method(SdkHttpMethod.GET) @@ -220,4 +340,65 @@ private static Stream refreshBackoffTestParams() { Arguments.of(Duration.ofSeconds(1), null) ); } + + private static final class TestIdentity implements Identity { + } + + /** + * An identity provider that counts identity resolutions, and whose invalidation completes with a given future. + */ + private static final class TrackingIdentityProvider implements IdentityProvider { + private final CompletableFuture invalidation; + private final AtomicInteger resolveCount = new AtomicInteger(); + private final AtomicInteger invalidateCount = new AtomicInteger(); + private volatile TestIdentity lastResolvedIdentity; + + TrackingIdentityProvider(CompletableFuture invalidation) { + this.invalidation = invalidation; + } + + @Override + public Class identityType() { + return TestIdentity.class; + } + + @Override + public CompletableFuture resolveIdentity(ResolveIdentityRequest request) { + resolveCount.incrementAndGet(); + lastResolvedIdentity = new TestIdentity(); + return CompletableFuture.completedFuture(lastResolvedIdentity); + } + + @Override + public CompletableFuture invalidate(TestIdentity identity) { + invalidateCount.incrementAndGet(); + return invalidation; + } + + int resolveCount() { + return resolveCount.get(); + } + + int invalidateCount() { + return invalidateCount.get(); + } + + TestIdentity lastResolvedIdentity() { + return lastResolvedIdentity; + } + } + + /** + * A service exception that reports itself as an authentication error. + */ + private static final class AuthErrorException extends SdkServiceException { + AuthErrorException() { + super(SdkServiceException.builder().message("expired").statusCode(400)); + } + + @Override + public boolean isAuthenticationError() { + return true; + } + } } diff --git a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AuthErrorInvalidationFunctionalTest.java b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AuthErrorInvalidationFunctionalTest.java index 468da6b6b120..8ac8691a267b 100644 --- a/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AuthErrorInvalidationFunctionalTest.java +++ b/test/codegen-generated-classes-test/src/test/java/software/amazon/awssdk/services/retry/AuthErrorInvalidationFunctionalTest.java @@ -19,14 +19,28 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.net.URI; +import java.time.Duration; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; import software.amazon.awssdk.auth.credentials.AwsCredentials; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; import software.amazon.awssdk.awscore.exception.AwsServiceException; +import software.amazon.awssdk.core.exception.ApiCallTimeoutException; +import software.amazon.awssdk.core.interceptor.Context; +import software.amazon.awssdk.core.interceptor.ExecutionAttributes; +import software.amazon.awssdk.core.interceptor.ExecutionInterceptor; +import software.amazon.awssdk.core.retry.RetryMode; import software.amazon.awssdk.http.AbortableInputStream; import software.amazon.awssdk.http.HttpExecuteResponse; import software.amazon.awssdk.http.SdkHttpRequest; @@ -38,185 +52,320 @@ import software.amazon.awssdk.services.protocolrestjson.ProtocolRestJsonClient; import software.amazon.awssdk.testutils.service.http.MockAsyncHttpClient; import software.amazon.awssdk.testutils.service.http.MockSyncHttpClient; +import software.amazon.awssdk.utils.Pair; import software.amazon.awssdk.utils.StringInputStream; /** - * Integration test verifying the end-to-end flow: - * service returns ExpiredToken → credentials invalidated → next call uses fresh credentials. + * Functional tests for how the SDK handles a service rejecting a request's credentials, and how it resolves credentials on + * retries: + *

    + *
  • An {@code ExpiredToken} or {@code InvalidToken} rejection invalidates the provider's cached credentials, and the + * request is retried. The retry attempt resolves the credentials again, so it is signed with the refreshed credentials.
    • + *
  • Every retry attempt resolves the credentials again, whatever the reason for the retry.
    • + *
  • Authorization errors such as {@code AccessDenied} neither invalidate the credentials nor are retried.
    • + *
*/ public class AuthErrorInvalidationFunctionalTest { - /** - * End-to-end: ExpiredToken → invalidation is triggered → next call uses fresh credentials. - * - * ExpiredToken exceptions are not retryable, so the first API call fails immediately. - * However, invalidation is still triggered on the provider, ensuring the next API call - * resolves fresh credentials. - * - * Scenario: - * 1. First API call: signed with "old-key", service returns ExpiredToken (not retryable) - * 2. SDK calls invalidate() on the provider (switching it to return "new-key") - * 3. First call fails with ExpiredToken exception - * 4. Second API call: resolves fresh identity "new-key", succeeds - */ - @Test - public void expiredToken_invalidatesCredentials_nextCallUsesFreshCredentials() { - MockSyncHttpClient mockHttpClient = new MockSyncHttpClient(); - InvalidationTrackingCredentialsProvider credentialsProvider = - new InvalidationTrackingCredentialsProvider("old-key", "old-secret", "new-key", "new-secret"); + private static final int STANDARD_MAX_ATTEMPTS = 3; - try (ProtocolRestJsonClient client = ProtocolRestJsonClient.builder() - .credentialsProvider(credentialsProvider) - .region(Region.US_EAST_1) - .endpointOverride(URI.create("http://localhost")) - .httpClient(mockHttpClient) - .build()) { + @ParameterizedTest + @ValueSource(strings = {"ExpiredToken", "InvalidToken"}) + public void authError_invalidatesCredentials_andRetryUsesRefreshedCredentials(String errorCode) { + MockSyncHttpClient mockHttpClient = new MockSyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.refreshedOnInvalidate(); - // First API call: ExpiredToken (not retryable — fails immediately) - mockHttpClient.stubResponses(expiredTokenResponse()); + try (ProtocolRestJsonClient client = syncClient(mockHttpClient, credentialsProvider)) { + mockHttpClient.stubResponses(authErrorResponse(errorCode), successResponse()); - assertThatThrownBy(() -> client.allTypes()) - .isInstanceOf(AwsServiceException.class) - .satisfies(e -> assertThat(((AwsServiceException) e).awsErrorDetails().errorCode()) - .isEqualTo("ExpiredToken")); + client.allTypes(); - // Verify invalidate was called during the first API call assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(1); + assertThat(accessKeysUsed(mockHttpClient.getRequests())).containsExactly("key-0", "key-1"); + } + } - // Verify first call used "old-key" - List firstCallRequests = mockHttpClient.getRequests(); - assertThat(firstCallRequests).hasSize(1); - assertRequestUsedAccessKey(firstCallRequests.get(0), "old-key"); + @ParameterizedTest + @ValueSource(strings = {"ExpiredToken", "InvalidToken"}) + public void async_authError_invalidatesCredentials_andRetryUsesRefreshedCredentials(String errorCode) { + MockAsyncHttpClient mockHttpClient = new MockAsyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.refreshedOnInvalidate(); - // Now make a second API call — this should resolve fresh credentials - mockHttpClient.reset(); - mockHttpClient.stubResponses(successResponse()); + try (ProtocolRestJsonAsyncClient client = asyncClient(mockHttpClient, credentialsProvider)) { + mockHttpClient.stubResponses(authErrorResponse(errorCode), successResponse()); - client.allTypes(); + client.allTypes().join(); - // Second call should use new credentials (provider was invalidated) - List secondCallRequests = mockHttpClient.getRequests(); - assertThat(secondCallRequests).hasSize(1); - assertRequestUsedAccessKey(secondCallRequests.get(0), "new-key"); + assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(1); + assertThat(accessKeysUsed(mockHttpClient.getRequests())).containsExactly("key-0", "key-1"); + } finally { + mockHttpClient.close(); } } /** - * The async client must invalidate on an auth error just as the sync client does. The two client types drive - * retries through different helper methods, so this is exercised separately rather than assumed from the sync case. + * When invalidation cannot produce new credentials (for example, static credentials that have expired), every attempt is + * rejected, and the request fails once the retries are exhausted. */ @Test - public void async_expiredToken_invalidatesCredentials_nextCallUsesFreshCredentials() { + public void expiredToken_whenCredentialsAreNotRefreshed_failsAfterRetriesAreExhausted() { + MockSyncHttpClient mockHttpClient = new MockSyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.neverRefreshed(); + + try (ProtocolRestJsonClient client = + ProtocolRestJsonClient.builder() + .credentialsProvider(credentialsProvider) + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost")) + .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.retryStrategy(RetryMode.STANDARD)) + .build()) { + mockHttpClient.stubResponses(authErrorResponse("ExpiredToken")); + + assertThatThrownBy(client::allTypes) + .isInstanceOf(AwsServiceException.class) + .satisfies(e -> { + AwsServiceException serviceException = (AwsServiceException) e; + assertThat(serviceException.awsErrorDetails().errorCode()).isEqualTo("ExpiredToken"); + assertThat(serviceException.numAttempts()).isEqualTo(STANDARD_MAX_ATTEMPTS); + }); + + assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(STANDARD_MAX_ATTEMPTS); + assertThat(credentialsProvider.resolveCallCount()).isEqualTo(STANDARD_MAX_ATTEMPTS); + assertThat(accessKeysUsed(mockHttpClient.getRequests())).containsExactly("key-0", "key-0", "key-0"); + } + } + + @Test + public void async_expiredToken_whenCredentialsAreNotRefreshed_failsAfterRetriesAreExhausted() { MockAsyncHttpClient mockHttpClient = new MockAsyncHttpClient(); - InvalidationTrackingCredentialsProvider credentialsProvider = - new InvalidationTrackingCredentialsProvider("old-key", "old-secret", "new-key", "new-secret"); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.neverRefreshed(); + + try (ProtocolRestJsonAsyncClient client = + ProtocolRestJsonAsyncClient.builder() + .credentialsProvider(credentialsProvider) + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost")) + .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.retryStrategy(RetryMode.STANDARD)) + .build()) { + mockHttpClient.stubResponses(authErrorResponse("ExpiredToken")); - try (ProtocolRestJsonAsyncClient client = ProtocolRestJsonAsyncClient.builder() - .credentialsProvider(credentialsProvider) - .region(Region.US_EAST_1) - .endpointOverride(URI.create("http://localhost")) - .httpClient(mockHttpClient) - .build()) { + assertThatThrownBy(() -> client.allTypes().join()) + .isInstanceOf(CompletionException.class) + .satisfies(e -> { + AwsServiceException serviceException = (AwsServiceException) e.getCause(); + assertThat(serviceException.awsErrorDetails().errorCode()).isEqualTo("ExpiredToken"); + assertThat(serviceException.numAttempts()).isEqualTo(STANDARD_MAX_ATTEMPTS); + }); + + assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(STANDARD_MAX_ATTEMPTS); + assertThat(credentialsProvider.resolveCallCount()).isEqualTo(STANDARD_MAX_ATTEMPTS); + assertThat(accessKeysUsed(mockHttpClient.getRequests())).containsExactly("key-0", "key-0", "key-0"); + } finally { + mockHttpClient.close(); + } + } - // First API call: ExpiredToken (not retryable — fails immediately) - mockHttpClient.stubResponses(expiredTokenResponse()); + /** + * Credentials are resolved again on every retry attempt, not only after an authentication error. + */ + @Test + public void serverError_retryResolvesCredentialsAgain() { + MockSyncHttpClient mockHttpClient = new MockSyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.newCredentialsOnEveryResolve(); - assertThatThrownBy(() -> client.allTypes().join()) - .hasCauseInstanceOf(AwsServiceException.class) - .satisfies(e -> assertThat(((AwsServiceException) e.getCause()).awsErrorDetails().errorCode()) - .isEqualTo("ExpiredToken")); + try (ProtocolRestJsonClient client = syncClient(mockHttpClient, credentialsProvider)) { + mockHttpClient.stubResponses(serverErrorResponse(), successResponse()); - // Verify invalidate was called during the first API call - assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(1); + client.allTypes(); - // Verify first call used "old-key" - List firstCallRequests = mockHttpClient.getRequests(); - assertThat(firstCallRequests).hasSize(1); - assertRequestUsedAccessKey(firstCallRequests.get(0), "old-key"); + assertThat(credentialsProvider.invalidateCallCount()).isZero(); + assertThat(credentialsProvider.resolveCallCount()).isEqualTo(2); + assertThat(accessKeysUsed(mockHttpClient.getRequests())).containsExactly("key-0", "key-1"); + } + } - // Now make a second API call — this should resolve fresh credentials - mockHttpClient.reset(); - mockHttpClient.stubResponses(successResponse()); + @Test + public void async_serverError_retryResolvesCredentialsAgain() { + MockAsyncHttpClient mockHttpClient = new MockAsyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.newCredentialsOnEveryResolve(); + + try (ProtocolRestJsonAsyncClient client = asyncClient(mockHttpClient, credentialsProvider)) { + mockHttpClient.stubResponses(serverErrorResponse(), successResponse()); client.allTypes().join(); - // Second call should use new credentials (provider was invalidated) - List secondCallRequests = mockHttpClient.getRequests(); - assertThat(secondCallRequests).hasSize(1); - assertRequestUsedAccessKey(secondCallRequests.get(0), "new-key"); + assertThat(credentialsProvider.invalidateCallCount()).isZero(); + assertThat(credentialsProvider.resolveCallCount()).isEqualTo(2); + assertThat(accessKeysUsed(mockHttpClient.getRequests())).containsExactly("key-0", "key-1"); + } finally { + mockHttpClient.close(); } } /** - * Verify that AccessDenied does NOT trigger invalidation and the exception propagates. + * An async retry attempt resolves the credentials on the thread that runs the attempt, which is the thread that signs the + * request and runs the attempt's beforeTransmission interceptors. That is a future completion executor thread, not the + * scheduler thread that fired the backoff timer. The first attempt resolves them on the calling thread. */ @Test - public void accessDenied_doesNotInvalidateCredentials() { - MockSyncHttpClient mockHttpClient = new MockSyncHttpClient(); - InvalidationTrackingCredentialsProvider credentialsProvider = - new InvalidationTrackingCredentialsProvider("my-key", "my-secret", "new-key", "new-secret"); + public void async_retryResolvesCredentialsOnTheThreadThatRunsTheAttempt() { + MockAsyncHttpClient mockHttpClient = new MockAsyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.newCredentialsOnEveryResolve(); + List beforeTransmissionThreads = new CopyOnWriteArrayList<>(); + ExecutionInterceptor threadCapturingInterceptor = new ExecutionInterceptor() { + @Override + public void beforeTransmission(Context.BeforeTransmission context, ExecutionAttributes executionAttributes) { + beforeTransmissionThreads.add(Thread.currentThread()); + } + }; + + try (ProtocolRestJsonAsyncClient client = + ProtocolRestJsonAsyncClient.builder() + .credentialsProvider(credentialsProvider) + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost")) + .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.addExecutionInterceptor(threadCapturingInterceptor)) + .build()) { + mockHttpClient.stubResponses(serverErrorResponse(), successResponse()); + + client.allTypes().join(); - try (ProtocolRestJsonClient client = ProtocolRestJsonClient.builder() - .credentialsProvider(credentialsProvider) - .region(Region.US_EAST_1) - .endpointOverride(URI.create("http://localhost")) - .httpClient(mockHttpClient) - .build()) { + List resolveThreads = credentialsProvider.resolveThreads(); + assertThat(resolveThreads).hasSize(2); + assertThat(beforeTransmissionThreads).hasSize(2); - mockHttpClient.stubResponses(accessDeniedResponse()); + assertThat(resolveThreads.get(0)).isSameAs(Thread.currentThread()); + assertThat(resolveThreads.get(1)).isNotSameAs(Thread.currentThread()) + .isSameAs(beforeTransmissionThreads.get(1)); + assertThat(resolveThreads.get(1).getName()).startsWith("sdk-async-response"); + } finally { + mockHttpClient.close(); + } + } - assertThatThrownBy(() -> client.allTypes()) + /** + * A retry that blocks while refreshing credentials must not delay the timers of other requests on the client. The + * client's scheduler has a single thread, so a refresh running on it would hold back every timer until it finished. + */ + @Test + public void async_blockingCredentialRefreshOnRetry_doesNotDelayOtherRequestsTimeouts() { + MockAsyncHttpClient mockHttpClient = new MockAsyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = + TrackingCredentialsProvider.refreshedOnInvalidateBlockingFor(Duration.ofSeconds(2)); + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); + + try (ProtocolRestJsonAsyncClient client = + ProtocolRestJsonAsyncClient.builder() + .credentialsProvider(credentialsProvider) + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost")) + .httpClient(mockHttpClient) + .overrideConfiguration(o -> o.scheduledExecutorService(scheduler)) + .build()) { + // Responses are handed out in request order: the slow call's response takes 5 seconds, and the other call is + // rejected with ExpiredToken, then succeeds on retry after the blocking refresh. + mockHttpClient.stubResponses(Pair.of(successResponse(), Duration.ofSeconds(5)), + Pair.of(authErrorResponse("ExpiredToken"), Duration.ZERO), + Pair.of(successResponse(), Duration.ZERO)); + + long start = System.nanoTime(); + CompletableFuture slowCall = + client.allTypes(r -> r.overrideConfiguration(o -> o.apiCallTimeout(Duration.ofMillis(300)))); + CompletableFuture rejectedCall = client.allTypes(); + + assertThatThrownBy(slowCall::join).hasCauseInstanceOf(ApiCallTimeoutException.class); + assertThat(Duration.ofNanos(System.nanoTime() - start)).isLessThan(Duration.ofMillis(1500)); + + rejectedCall.join(); + assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(1); + } finally { + mockHttpClient.close(); + scheduler.shutdownNow(); + } + } + + @Test + public void accessDenied_doesNotInvalidateCredentials_andIsNotRetried() { + MockSyncHttpClient mockHttpClient = new MockSyncHttpClient(); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.refreshedOnInvalidate(); + + try (ProtocolRestJsonClient client = syncClient(mockHttpClient, credentialsProvider)) { + mockHttpClient.stubResponses(authErrorResponse("AccessDenied")); + + assertThatThrownBy(client::allTypes) .isInstanceOf(AwsServiceException.class) .satisfies(e -> assertThat(((AwsServiceException) e).awsErrorDetails().errorCode()) .isEqualTo("AccessDenied")); - // Verify invalidate was NOT called - assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(0); + assertThat(credentialsProvider.invalidateCallCount()).isZero(); + assertThat(mockHttpClient.getRequests()).hasSize(1); } } @Test - public void async_accessDenied_doesNotInvalidateCredentials() { + public void async_accessDenied_doesNotInvalidateCredentials_andIsNotRetried() { MockAsyncHttpClient mockHttpClient = new MockAsyncHttpClient(); - InvalidationTrackingCredentialsProvider credentialsProvider = - new InvalidationTrackingCredentialsProvider("my-key", "my-secret", "new-key", "new-secret"); + TrackingCredentialsProvider credentialsProvider = TrackingCredentialsProvider.refreshedOnInvalidate(); - try (ProtocolRestJsonAsyncClient client = ProtocolRestJsonAsyncClient.builder() - .credentialsProvider(credentialsProvider) - .region(Region.US_EAST_1) - .endpointOverride(URI.create("http://localhost")) - .httpClient(mockHttpClient) - .build()) { - - mockHttpClient.stubResponses(accessDeniedResponse()); + try (ProtocolRestJsonAsyncClient client = asyncClient(mockHttpClient, credentialsProvider)) { + mockHttpClient.stubResponses(authErrorResponse("AccessDenied")); assertThatThrownBy(() -> client.allTypes().join()) .hasCauseInstanceOf(AwsServiceException.class) .satisfies(e -> assertThat(((AwsServiceException) e.getCause()).awsErrorDetails().errorCode()) .isEqualTo("AccessDenied")); - // Verify invalidate was NOT called - assertThat(credentialsProvider.invalidateCallCount()).isEqualTo(0); + assertThat(credentialsProvider.invalidateCallCount()).isZero(); + assertThat(mockHttpClient.getRequests()).hasSize(1); + } finally { + mockHttpClient.close(); } } // --- Helper methods --- - private void assertRequestUsedAccessKey(SdkHttpRequest request, String expectedAccessKeyId) { - assertThat(request.firstMatchingHeader("Authorization")) - .isPresent() - .hasValueSatisfying(authHeader -> - assertThat(authHeader).contains("Credential=" + expectedAccessKeyId + "/") - ); + private static ProtocolRestJsonClient syncClient(MockSyncHttpClient httpClient, AwsCredentialsProvider credentialsProvider) { + return ProtocolRestJsonClient.builder() + .credentialsProvider(credentialsProvider) + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost")) + .httpClient(httpClient) + .build(); } - private HttpExecuteResponse expiredTokenResponse() { - String errorBody = "{\"__type\":\"ExpiredTokenException\",\"message\":\"The security token included in the request " - + "is expired\"}"; + private static ProtocolRestJsonAsyncClient asyncClient(MockAsyncHttpClient httpClient, + AwsCredentialsProvider credentialsProvider) { + return ProtocolRestJsonAsyncClient.builder() + .credentialsProvider(credentialsProvider) + .region(Region.US_EAST_1) + .endpointOverride(URI.create("http://localhost")) + .httpClient(httpClient) + .build(); + } + + private static List accessKeysUsed(List requests) { + return requests.stream() + .map(AuthErrorInvalidationFunctionalTest::accessKeyUsed) + .collect(Collectors.toList()); + } + + private static String accessKeyUsed(SdkHttpRequest request) { + String authorization = request.firstMatchingHeader("Authorization") + .orElseThrow(() -> new AssertionError("Request was not signed")); + String credentialPrefix = "Credential="; + int start = authorization.indexOf(credentialPrefix) + credentialPrefix.length(); + return authorization.substring(start, authorization.indexOf('/', start)); + } + + private static HttpExecuteResponse authErrorResponse(String errorCode) { + String errorBody = "{\"message\":\"The request was rejected\"}"; return HttpExecuteResponse.builder() .response(SdkHttpResponse.builder() .statusCode(403) - .putHeader("x-amzn-ErrorType", "ExpiredToken") + .putHeader("x-amzn-ErrorType", errorCode) .putHeader("content-length", String.valueOf(errorBody.length())) .build()) @@ -224,12 +373,12 @@ private HttpExecuteResponse expiredTokenResponse() { .build(); } - private HttpExecuteResponse accessDeniedResponse() { - String errorBody = "{\"__type\":\"AccessDeniedException\",\"message\":\"User is not authorized\"}"; + private static HttpExecuteResponse serverErrorResponse() { + String errorBody = "{\"message\":\"Internal failure\"}"; return HttpExecuteResponse.builder() .response(SdkHttpResponse.builder() - .statusCode(403) - .putHeader("x-amzn-ErrorType", "AccessDenied") + .statusCode(500) + .putHeader("x-amzn-ErrorType", "InternalFailure") .putHeader("content-length", String.valueOf(errorBody.length())) .build()) @@ -237,7 +386,7 @@ private HttpExecuteResponse accessDeniedResponse() { .build(); } - private HttpExecuteResponse successResponse() { + private static HttpExecuteResponse successResponse() { String body = "{}"; return HttpExecuteResponse.builder() .response(SdkHttpResponse.builder() @@ -252,31 +401,64 @@ private HttpExecuteResponse successResponse() { // --- Test doubles --- /** - * A credentials provider that tracks invalidation calls and switches to fresh credentials - * after invalidation is triggered. This simulates the behavior of a caching provider - * (like IMDS or Container provider) that serves stale credentials until invalidated. + * A credentials provider that vends credentials with access key {@code key-N}, where N is the current generation, and + * records every resolution and invalidation. Depending on how it is created, the generation advances when the + * provider is invalidated (simulating a caching provider that refreshes after invalidation), on every resolution + * (simulating a provider whose credentials change between attempts), or never (simulating static credentials). + * A provider can also block for a time on the first resolution after an invalidation, simulating a refresh from a slow + * credential source on the calling thread, as the SDK's caching providers do. */ - private static class InvalidationTrackingCredentialsProvider implements AwsCredentialsProvider { - private final AwsBasicCredentials oldCredentials; - private final AwsBasicCredentials newCredentials; - private final AtomicInteger invalidateCount = new AtomicInteger(0); - private volatile boolean invalidated = false; - - InvalidationTrackingCredentialsProvider(String oldAccessKey, String oldSecretKey, - String newAccessKey, String newSecretKey) { - this.oldCredentials = AwsBasicCredentials.create(oldAccessKey, oldSecretKey); - this.newCredentials = AwsBasicCredentials.create(newAccessKey, newSecretKey); + private static final class TrackingCredentialsProvider implements AwsCredentialsProvider { + private final boolean advanceOnInvalidate; + private final boolean advanceOnResolve; + private final Duration blockingRefreshDuration; + private final AtomicInteger generation = new AtomicInteger(); + private final AtomicInteger resolveCount = new AtomicInteger(); + private final AtomicInteger invalidateCount = new AtomicInteger(); + private final AtomicBoolean refreshPending = new AtomicBoolean(); + private final List resolveThreads = new CopyOnWriteArrayList<>(); + + private TrackingCredentialsProvider(boolean advanceOnInvalidate, boolean advanceOnResolve, + Duration blockingRefreshDuration) { + this.advanceOnInvalidate = advanceOnInvalidate; + this.advanceOnResolve = advanceOnResolve; + this.blockingRefreshDuration = blockingRefreshDuration; + } + + static TrackingCredentialsProvider refreshedOnInvalidate() { + return new TrackingCredentialsProvider(true, false, Duration.ZERO); + } + + static TrackingCredentialsProvider refreshedOnInvalidateBlockingFor(Duration blockingRefreshDuration) { + return new TrackingCredentialsProvider(true, false, blockingRefreshDuration); + } + + static TrackingCredentialsProvider newCredentialsOnEveryResolve() { + return new TrackingCredentialsProvider(false, true, Duration.ZERO); + } + + static TrackingCredentialsProvider neverRefreshed() { + return new TrackingCredentialsProvider(false, false, Duration.ZERO); } @Override public AwsCredentials resolveCredentials() { - return invalidated ? newCredentials : oldCredentials; + if (refreshPending.compareAndSet(true, false) && !blockingRefreshDuration.isZero()) { + try { + Thread.sleep(blockingRefreshDuration.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + int current = advanceOnResolve ? generation.getAndIncrement() : generation.get(); + return AwsBasicCredentials.create("key-" + current, "secret-" + current); } @Override public CompletableFuture resolveIdentity(ResolveIdentityRequest request) { - AwsCredentials creds = resolveCredentials(); - return CompletableFuture.completedFuture(creds); + resolveCount.incrementAndGet(); + resolveThreads.add(Thread.currentThread()); + return CompletableFuture.completedFuture(resolveCredentials()); } @Override @@ -287,12 +469,23 @@ public Class identityType() { @Override public CompletableFuture invalidate(AwsCredentialsIdentity identity) { invalidateCount.incrementAndGet(); - invalidated = true; + if (advanceOnInvalidate) { + generation.incrementAndGet(); + refreshPending.set(true); + } return CompletableFuture.completedFuture(null); } int invalidateCallCount() { return invalidateCount.get(); } + + int resolveCallCount() { + return resolveCount.get(); + } + + List resolveThreads() { + return resolveThreads; + } } }