From e37e2657df2aeb2696e70c0c3adc601055e8e97b Mon Sep 17 00:00:00 2001 From: Alex Shovlin Date: Tue, 8 Sep 2026 10:41:04 -0400 Subject: [PATCH 1/3] Support following S3 bucket region redirects when the CRT client is enabled --- .../next-release/enhancement-s3-42614.json | 5 + awscli/botocore/utils.py | 94 ++-- awscli/customizations/s3/factory.py | 24 +- awscli/s3transfer/crt.py | 479 ++++++++++++++++-- awscli/topics/s3-config.rst | 3 - tests/functional/s3transfer/test_crt.py | 204 +++++++- tests/integration/s3transfer/test_crt.py | 36 +- tests/unit/customizations/s3/test_factory.py | 32 ++ tests/unit/s3transfer/test_crt.py | 283 ++++++++++- 9 files changed, 1042 insertions(+), 118 deletions(-) create mode 100644 .changes/next-release/enhancement-s3-42614.json diff --git a/.changes/next-release/enhancement-s3-42614.json b/.changes/next-release/enhancement-s3-42614.json new file mode 100644 index 000000000000..42e801c8284d --- /dev/null +++ b/.changes/next-release/enhancement-s3-42614.json @@ -0,0 +1,5 @@ +{ + "type": "enhancement", + "category": "``s3```", + "description": "Follow bucket region redirects for the ``s3`` commands when CRT is enabled." +} diff --git a/awscli/botocore/utils.py b/awscli/botocore/utils.py index a3ac03e745e8..8a43ecaac75a 100644 --- a/awscli/botocore/utils.py +++ b/awscli/botocore/utils.py @@ -1576,7 +1576,7 @@ class S3RegionRedirectorv2: """ def __init__(self, endpoint_bridge, client, cache=None): - self._cache = cache or {} + self._cache = {} if cache is None else cache self._client = weakref.proxy(client) def register(self, event_emitter=None): @@ -1616,49 +1616,9 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs): ) return - error = response[1].get('Error', {}) - error_code = error.get('Code') - response_metadata = response[1].get('ResponseMetadata', {}) - - # We have to account for 400 responses because - # if we sign a Head* request with the wrong region, - # we'll get a 400 Bad Request but we won't get a - # body saying it's an "AuthorizationHeaderMalformed". - is_special_head_object = ( - error_code in ('301', '400') and operation.name == 'HeadObject' - ) - is_special_head_bucket = ( - error_code in ('301', '400') - and operation.name == 'HeadBucket' - and 'x-amz-bucket-region' - in response_metadata.get('HTTPHeaders', {}) - ) - is_wrong_signing_region = ( - error_code == 'AuthorizationHeaderMalformed' and 'Region' in error - ) - is_redirect_status = response[0] is not None and response[ - 0 - ].status_code in (301, 302, 307) - is_permanent_redirect = error_code == 'PermanentRedirect' - is_opt_in_region_redirect = ( - error_code == 'IllegalLocationConstraintException' - and operation.name != 'CreateBucket' - ) - if not any( - [ - is_special_head_object, - is_wrong_signing_region, - is_permanent_redirect, - is_special_head_bucket, - is_redirect_status, - is_opt_in_region_redirect, - ] - ): - return - - bucket = request_dict['context']['s3_redirect']['bucket'] - client_region = request_dict['context'].get('client_region') - new_region = self.get_bucket_region(bucket, response) + bucket = redirect_ctx.get('bucket') + client_region = request_dict.get('context', {}).get('client_region') + new_region = self.get_redirect_region(bucket, response, operation) if new_region is None: logger.debug( @@ -1702,6 +1662,52 @@ def redirect_from_error(self, request_dict, response, operation, **kwargs): # Return 0 so it doesn't wait to retry return 0 + def get_redirect_region(self, bucket, response, operation): + if bucket is None: + return None + if ArnParser.is_arn(bucket): + return None + error = response[1].get('Error', {}) + error_code = error.get('Code') + response_metadata = response[1].get('ResponseMetadata', {}) + + # We have to account for 400 responses because + # if we sign a Head* request with the wrong region, + # we'll get a 400 Bad Request but we won't get a + # body saying it's an "AuthorizationHeaderMalformed". + is_special_head_object = ( + error_code in ('301', '400') and operation.name == 'HeadObject' + ) + is_special_head_bucket = ( + error_code in ('301', '400') + and operation.name == 'HeadBucket' + and 'x-amz-bucket-region' + in response_metadata.get('HTTPHeaders', {}) + ) + is_wrong_signing_region = ( + error_code == 'AuthorizationHeaderMalformed' and 'Region' in error + ) + is_redirect_status = response[0] is not None and response[ + 0 + ].status_code in (301, 302, 307) + is_permanent_redirect = error_code == 'PermanentRedirect' + is_opt_in_region_redirect = ( + error_code == 'IllegalLocationConstraintException' + and operation.name != 'CreateBucket' + ) + if not any( + [ + is_special_head_object, + is_wrong_signing_region, + is_permanent_redirect, + is_special_head_bucket, + is_redirect_status, + is_opt_in_region_redirect, + ] + ): + return None + return self.get_bucket_region(bucket, response) + def get_bucket_region(self, bucket, response): """ There are multiple potential sources for the new region to redirect to, diff --git a/awscli/customizations/s3/factory.py b/awscli/customizations/s3/factory.py index 50ba57294400..ab8b6da703fa 100644 --- a/awscli/customizations/s3/factory.py +++ b/awscli/customizations/s3/factory.py @@ -20,6 +20,7 @@ BotocoreCRTRequestSerializer, CRTTransferManager, acquire_crt_s3_process_lock, + create_crt_client_bootstrap, create_s3_crt_client, ) from s3transfer.manager import TransferManager @@ -114,16 +115,28 @@ def _acquire_crt_s3_process_lock(self): def _create_crt_transfer_manager(self, params, runtime_config): self._acquire_crt_s3_process_lock() + region = self._resolve_region(params) + bootstrap = create_crt_client_bootstrap() return CRTTransferManager( - self._create_crt_client(params, runtime_config), - self._create_crt_request_serializer(params), + crt_client_factory=lambda client_region=None: ( + self._create_crt_client( + params, + runtime_config, + region=client_region or region, + bootstrap=bootstrap, + ) + ), + crt_request_serializer=self._create_crt_request_serializer(params), ) - def _create_crt_client(self, params, runtime_config): + def _create_crt_client( + self, params, runtime_config, region=None, bootstrap=None + ): config_file_params = self._session.get_scoped_config().get('s3', {}) create_crt_client_kwargs = { - 'region': self._resolve_region(params), + 'region': region or self._resolve_region(params), 'verify': self._resolve_verify(params), + 'bootstrap': bootstrap, } endpoint_url = params.get('endpoint_url') if endpoint_url and urlparse.urlparse(endpoint_url).scheme == 'http': @@ -161,6 +174,9 @@ def _create_crt_request_serializer(self, params): 'region_name': self._resolve_region(params), 'endpoint_url': params.get('endpoint_url'), }, + region_redirect_client_factory=lambda: ( + self._botocore_client_factory.create_client(params) + ), ) def _create_classic_transfer_manager( diff --git a/awscli/s3transfer/crt.py b/awscli/s3transfer/crt.py index ae4e5195a4b2..2cad7bfc8d23 100644 --- a/awscli/s3transfer/crt.py +++ b/awscli/s3transfer/crt.py @@ -13,6 +13,7 @@ import logging import re import threading +from concurrent.futures import CancelledError, Future from io import BytesIO import awscrt.http @@ -44,7 +45,13 @@ from botocore.config import Config from botocore.exceptions import InvalidConfigError, NoCredentialsError from botocore.useragent import register_feature_id -from botocore.utils import ArnParser, InvalidArnException, is_s3express_bucket +from botocore.utils import ( + ArnParser, + InvalidArnException, + S3RegionRedirectorv2, + is_s3express_bucket, +) +from s3transfer.compat import seekable from s3transfer.constants import FULL_OBJECT_CHECKSUM_ARGS, MB from s3transfer.exceptions import TransferNotDoneError from s3transfer.futures import BaseTransferFuture, BaseTransferMeta @@ -89,6 +96,7 @@ def create_s3_crt_client( use_ssl=True, verify=None, fio_options=None, + bootstrap=None, ): """ :type region: str @@ -135,11 +143,14 @@ def create_s3_crt_client( :type fio_options: Optional[dict] :param fio_options: Kwargs to use to build an `awscrt.s3.S3FileIoOptions`. + + :type bootstrap: Optional[awscrt.io.ClientBootstrap] + :param bootstrap: Shared I/O bootstrap to use for the client. If not + provided, a new bootstrap is created. """ - event_loop_group = EventLoopGroup(num_threads) - host_resolver = DefaultHostResolver(event_loop_group) - bootstrap = ClientBootstrap(event_loop_group, host_resolver) + if bootstrap is None: + bootstrap = create_crt_client_bootstrap(num_threads) tls_connection_options = None tls_mode = ( @@ -183,6 +194,12 @@ def create_s3_crt_client( ) +def create_crt_client_bootstrap(num_threads=None): + event_loop_group = EventLoopGroup(num_threads) + host_resolver = DefaultHostResolver(event_loop_group) + return ClientBootstrap(event_loop_group, host_resolver) + + def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None): if provided_throughput_target_bytes is None: target_gbps = get_recommended_throughput_target_gbps() @@ -201,17 +218,106 @@ def _get_crt_throughput_target_gbps(provided_throughput_target_bytes=None): return target_gbps +class CRTS3RegionRedirectPolicy: + """Decides which region a CRT transfer to a bucket should use. + + Region discovery is delegated to the request serializer, which reuses + botocore's ``S3RegionRedirectorv2`` classification. This class owns only + the conditions under which a redirect may be attempted at all. + """ + + def __init__(self, crt_request_serializer): + self._crt_request_serializer = crt_request_serializer + + def get_cached_bucket_region(self, bucket): + """Return a region already discovered for a bucket, if any.""" + region = self._crt_request_serializer.get_cached_bucket_region(bucket) + if region is not None: + logger.debug( + 'Using cached region %s for S3 bucket %s', region, bucket + ) + return region + + def get_retry_region( + self, + bucket, + transfer_type, + error, + is_retry, + bytes_transferred, + cancelled, + is_replayable, + ): + """Return the region to retry a failed request in, or ``None``. + + A returned region has been cached, so both the retried request and + later transfers to the same bucket use it. + """ + if error is None: + return None + if is_retry: + logger.debug( + 'Transfer for bucket %s was already redirected, not ' + 'redirecting again.', + bucket, + ) + return None + if cancelled: + return None + if not is_replayable: + logger.debug( + 'Not redirecting transfer for bucket %s because its stream ' + 'cannot be replayed.', + bucket, + ) + return None + if bytes_transferred: + # Replaying a request that moved data would either duplicate + # bytes or double-count progress. + logger.debug( + 'Not redirecting transfer for bucket %s because it already ' + 'transferred %s bytes.', + bucket, + bytes_transferred, + ) + return None + if is_s3express_bucket(bucket): + return None + try: + new_region = self._crt_request_serializer.get_bucket_region( + bucket, transfer_type, error + ) + except Exception as redirect_error: + logger.debug( + 'Unable to determine S3 redirect region.', + exc_info=redirect_error, + ) + return None + if new_region is None: + return None + logger.debug( + 'Redirecting CRT S3 transfer for bucket %s to region %s', + bucket, + new_region, + ) + self._crt_request_serializer.cache_bucket_region(bucket, new_region) + return new_region + + class CRTTransferManager: - def __init__(self, crt_s3_client, crt_request_serializer, osutil=None): + def __init__( + self, crt_client_factory, crt_request_serializer, osutil=None + ): """A transfer manager interface for Amazon S3 on CRT s3 client. - :type crt_s3_client: awscrt.s3.S3Client - :param crt_s3_client: The CRT s3 client, handling all the - HTTP requests and functions under then hood + :type crt_client_factory: + Callable[[Optional[str]], awscrt.s3.S3Client] + :param crt_client_factory: Creates a CRT client. ``None`` selects the + configured region; a region string selects a redirected region. :type crt_request_serializer: s3transfer.crt.BaseCRTRequestSerializer - :param crt_request_serializer: Serializer, generates unsigned crt HTTP - request. + :param crt_request_serializer: Serializer, generates unsigned CRT HTTP + requests. :type osutil: s3transfer.utils.OSUtils :param osutil: OSUtils object to use for os-related behavior when @@ -219,13 +325,18 @@ def __init__(self, crt_s3_client, crt_request_serializer, osutil=None): """ if osutil is None: self._osutil = OSUtils() - self._crt_s3_client = crt_s3_client self._s3_args_creator = S3ClientArgsCreator( crt_request_serializer, self._osutil ) self._crt_exception_translator = ( crt_request_serializer.translate_crt_exception ) + self._crt_client_factory = crt_client_factory + self._crt_clients = {} + self._crt_client_lock = threading.Lock() + self._region_redirect_policy = CRTS3RegionRedirectPolicy( + crt_request_serializer + ) self._future_coordinators = [] self._semaphore = threading.Semaphore(128) # not configurable # A counter to create unique id's for each transfer submitted. @@ -327,12 +438,25 @@ def _shutdown(self, cancel=False): def _release_semaphore(self, **kwargs): self._semaphore.release() + def get_crt_client(self, region=None): + with self._crt_client_lock: + crt_client = self._crt_clients.get(region) + if crt_client is None: + logger.debug( + 'Creating CRT S3 client for region %s', + region if region is not None else 'default', + ) + crt_client = self._crt_client_factory(region) + self._crt_clients[region] = crt_client + return crt_client + def _submit_transfer(self, request_type, call_args): register_feature_id('S3_TRANSFER') on_done_after_calls = [self._release_semaphore] coordinator = CRTTransferCoordinator( transfer_id=self._id_counter, exception_translator=self._crt_exception_translator, + completion_future=Future(), ) components = { 'meta': CRTTransferMeta(self._id_counter, call_args), @@ -341,6 +465,22 @@ def _submit_transfer(self, request_type, call_args): future = CRTTransferFuture(**components) afterdone = AfterDoneHandler(coordinator) on_done_after_calls.append(afterdone) + # Serialization can rewrite an ARN in call_args, so retain the + # caller-provided bucket for redirect eligibility and caching. + bucket = call_args.bucket + # Record the current stream position and if its replayable, + # in the event of a region redirect, we need to reset first + is_replayable = True + upload_stream_position = None + if request_type == 'put_object' and not isinstance( + call_args.fileobj, str + ): + try: + is_replayable = seekable(call_args.fileobj) + if is_replayable: + upload_stream_position = call_args.fileobj.tell() + except (AttributeError, OSError, ValueError): + is_replayable = False try: self._semaphore.acquire() @@ -348,22 +488,38 @@ def _submit_transfer(self, request_type, call_args): future, 'queued' ) on_queued() - crt_callargs = self._s3_args_creator.get_make_request_args( + + def create_request(is_retry): + # Reset the stream if we're redirecting due to bucket region + if is_retry and upload_stream_position is not None: + call_args.fileobj.seek(upload_stream_position) + crt_callargs = self._s3_args_creator.get_make_request_args( + request_type, + call_args, + coordinator, + future, + on_done_after_calls, + ) + region = self._region_redirect_policy.get_cached_bucket_region( + bucket + ) + crt_client = self.get_crt_client(region) + return crt_client, crt_callargs + + coordinator.submit( + create_request, + self._region_redirect_policy, + bucket, request_type, - call_args, - coordinator, - future, - on_done_after_calls, + is_replayable=is_replayable, ) - crt_s3_request = self._crt_s3_client.make_request(**crt_callargs) except Exception as e: coordinator.set_exception(e, True) on_done = self._s3_args_creator.get_crt_callback( future, 'done', after_subscribers=on_done_after_calls ) + coordinator.complete(e) on_done(error=e) - else: - coordinator.set_s3_request(crt_s3_request) self._future_coordinators.append(coordinator) self._id_counter += 1 @@ -448,8 +604,34 @@ def translate_crt_exception(self, exception): raise NotImplementedError('translate_crt_exception()') +class LazyHeadBucketClient: + """ + A boto client that is lazily initialized when needing to call head_bucket. + The existing region redirect logic supports falling back to this when the + region cannot be parsed from the S3 exception. We use this boto client even + when CRT is being used for transfers + """ + + def __init__(self, client_factory): + self._client_factory = client_factory + self._client = None + self._lock = threading.Lock() + + def head_bucket(self, **kwargs): + with self._lock: + if self._client is None: + self._client = self._client_factory() + client = self._client + return client.head_bucket(**kwargs) + + class BotocoreCRTRequestSerializer(BaseCRTRequestSerializer): - def __init__(self, session, client_kwargs=None): + def __init__( + self, + session, + client_kwargs=None, + region_redirect_client_factory=None, + ): """Serialize CRT HTTP request using botocore logic It also takes into account configuration from both the session and any keyword arguments that could be passed to @@ -460,12 +642,43 @@ def __init__(self, session, client_kwargs=None): :type client_kwargs: Optional[Dict[str, str]]) :param client_kwargs: The kwargs for the botocore s3 client initialization. + + :type region_redirect_client_factory: Optional[Callable] + :param region_redirect_client_factory: Creates a real botocore S3 + client if a CRT redirect response requires a HeadBucket fallback. """ self._session = session if client_kwargs is None: client_kwargs = {} + client_kwargs = client_kwargs.copy() + + # Preserve the caller's client configuration for the real botocore + # client used only by the HeadBucket fallback. The serializer client + # below is reconfigured as unsigned and never sends its request. + if region_redirect_client_factory is None: + redirect_client_kwargs = client_kwargs.copy() + redirect_client_kwargs['service_name'] = 's3' + region_redirect_client_factory = lambda: session.create_client( + **redirect_client_kwargs + ) + + # Build the botocore client that converts API arguments into the + # unsigned HTTP requests consumed by CRT. self._resolve_client_config(session, client_kwargs) self._client = session.create_client(**client_kwargs) + + # Share the bucket-region cache between redirect discovery and later + # request serialization. The real fallback client remains lazy. + self._region_redirect_client = LazyHeadBucketClient( + region_redirect_client_factory + ) + self._region_cache = {} + self._region_redirector = S3RegionRedirectorv2( + None, self._region_redirect_client, cache=self._region_cache + ) + + # Convert normal botocore calls into serialized requests without + # allowing the serializer client to perform network I/O. self._client.meta.events.register( 'request-created.s3.*', self._capture_http_request ) @@ -478,6 +691,10 @@ def __init__(self, session, client_kwargs=None): self._client.meta.events.register( 'before-call.s3.*', self._remove_checksum_context ) + self._client.meta.events.register( + 'before-endpoint-resolution.s3', + self._region_redirector.redirect_from_cache, + ) def _resolve_client_config(self, session, client_kwargs): user_provided_config = None @@ -612,6 +829,45 @@ def _remove_checksum_context(self, params, **kwargs): if "checksum" in request_context: del request_context["checksum"] + def cache_bucket_region(self, bucket, region): + self._region_cache[bucket] = region + + def get_cached_bucket_region(self, bucket): + return self._region_cache.get(bucket) + + def get_bucket_region(self, bucket, transfer_type, error): + """Extract a redirect region from a CRT response error. + This adapts the CRT error for S3RegionRedirectorv2. + """ + if not isinstance(error, S3ResponseError): + return None + translated_error = self._translate_crt_s3_response_error(error) + if translated_error is None: + return None + + operation_name = ( + translated_error.operation_name + or self._client.meta.method_to_api_mapping[transfer_type] + ) + operation = self._client.meta.service_model.operation_model( + operation_name + ) + http_response = botocore.awsrequest.AWSResponse( + None, + error.status_code, + dict(error.headers or []), + FakeRawResponse(error.body or b''), + ) + response = (http_response, translated_error.response) + # The redirector checks the CRT response first. It only uses this + # serializer's real botocore client for HeadBucket when the response + # identifies a redirect but omits the target region. + return self._region_redirector.get_redirect_region( + bucket, + response, + operation, + ) + class FakeRawResponse(BytesIO): def stream(self, amt=1024, decode_content=None): @@ -642,18 +898,30 @@ def _get_credentials(self): class CRTTransferCoordinator: - """A helper class for managing CRTTransferFuture""" + """ + Coordinates one logical transfer across its native CRT request(s), which + can make two if following a bucket region redirect + """ def __init__( - self, transfer_id=None, s3_request=None, exception_translator=None + self, + transfer_id=None, + s3_request=None, + exception_translator=None, + completion_future=None, ): self.transfer_id = transfer_id self._exception_translator = exception_translator self._s3_request = s3_request self._lock = threading.Lock() self._exception = None - self._crt_future = None + # This future represents the entire transfer, + # which could include a retry for a region redirect. + self._completion_future = completion_future or Future() + self._completion_started = False self._done_event = threading.Event() + self._cancelled = False + self._redirect_retry_started = False @property def s3_request(self): @@ -671,17 +939,25 @@ def set_exception(self, exception, override=False): self._exception = exception def cancel(self): - if self._s3_request: - self._s3_request.cancel() + with self._lock: + self._cancelled = True + s3_request = self._s3_request + if s3_request: + s3_request.cancel() + + @property + def cancelled(self): + with self._lock: + return self._cancelled def result(self, timeout=None): if self._exception: raise self._exception try: - self._crt_future.result(timeout) + self._completion_future.result(timeout) except KeyboardInterrupt: self.cancel() - self._crt_future.result(timeout) + self._completion_future.result(timeout) raise except Exception as e: self.handle_exception(e) @@ -705,16 +981,101 @@ def handle_exception(self, exc): raise exc def done(self): - if self._crt_future is None: - return False - return self._crt_future.done() + return self._completion_future.done() - def set_s3_request(self, s3_request): - self._s3_request = s3_request - self._crt_future = self._s3_request.finished_future + def submit( + self, + request_factory, + region_redirect_policy, + bucket, + transfer_type, + is_replayable=True, + is_retry=False, + ): + """ + Submits a single CRT request which can either be our first attempt or + a second attempt with a region redirect. + """ + with self._lock: + if self._cancelled: + raise CancelledError() + if is_retry: + self._redirect_retry_started = True + crt_client, crt_callargs = request_factory(is_retry) + on_done = crt_callargs['on_done'] + on_progress = crt_callargs['on_progress'] + bytes_transferred = 0 + + def track_progress(transferred): + nonlocal bytes_transferred + bytes_transferred += transferred + on_progress(transferred) + + def request_done(error=None, **kwargs): + new_region = region_redirect_policy.get_retry_region( + bucket=bucket, + transfer_type=transfer_type, + error=error, + is_retry=is_retry, + bytes_transferred=bytes_transferred, + cancelled=self.cancelled, + is_replayable=is_replayable, + ) + if new_region is not None: + try: + self.submit( + request_factory, + region_redirect_policy, + bucket, + transfer_type, + is_replayable=is_replayable, + is_retry=True, + ) + return + except Exception as retry_error: + retry_error.__cause__ = error + error = retry_error + self.set_exception(retry_error, True) + self.complete(error) + on_done(error=error, **kwargs) + + crt_callargs['on_done'] = request_done + crt_callargs['on_progress'] = track_progress + s3_request = crt_client.make_request(**crt_callargs) + self.set_s3_request(s3_request, is_retry=is_retry) + + def set_s3_request(self, s3_request, is_retry=False): + with self._lock: + if not is_retry and self._redirect_retry_started: + # The retry is already active. The original request completed + # and redirected before make_request() returned. + return + if is_retry: + self._redirect_retry_started = True + self._s3_request = s3_request + cancelled = self._cancelled + if cancelled: + s3_request.cancel() + + def complete(self, error=None): + """Complete the logical transfer after the original or retry request. + + This is separate from the native CRT request completion callbacks. + """ + with self._lock: + if self._completion_started or self._completion_future.done(): + return + self._completion_started = True + completion_future = self._completion_future + if error is None: + completion_future.set_result(None) + else: + completion_future.set_exception(error) class S3ClientArgsCreator: + _DOWNLOAD_TEMP_FILENAME = '_crt_download_temp_filename' + def __init__(self, crt_request_serializer, os_utils): self._request_serializer = crt_request_serializer self._os_utils = os_utils @@ -779,15 +1140,23 @@ def _get_make_request_args_put_object( call_args.extra_args["Body"] = call_args.fileobj checksum_config = None + provided_checksum_algorithm = None if not any( checksum_arg in call_args.extra_args for checksum_arg in FULL_OBJECT_CHECKSUM_ARGS ): - checksum_algorithm = call_args.extra_args.pop( - 'ChecksumAlgorithm', 'CRC64NVME' + # CRT applies this checksum itself, so we hide it from botocore + # while serializing but store it for a possible redirected attempt + provided_checksum_algorithm = call_args.extra_args.pop( + 'ChecksumAlgorithm', None + ) + applied_checksum_algorithm = ( + provided_checksum_algorithm or 'CRC64NVME' ).upper() checksum_config = awscrt.s3.S3ChecksumConfig( - algorithm=awscrt.s3.S3ChecksumAlgorithm[checksum_algorithm], + algorithm=awscrt.s3.S3ChecksumAlgorithm[ + applied_checksum_algorithm + ], location=awscrt.s3.S3ChecksumLocation.TRAILER, ) # Suppress botocore's automatic MD5 calculation by setting an override @@ -796,14 +1165,20 @@ def _get_make_request_args_put_object( # automatically add trailing checksums to its uploads. call_args.extra_args["ContentMD5"] = "override-to-be-removed" - make_request_args = self._default_get_make_request_args( - request_type=request_type, - call_args=call_args, - coordinator=coordinator, - future=future, - on_done_before_calls=on_done_before_calls, - on_done_after_calls=on_done_after_calls, - ) + try: + make_request_args = self._default_get_make_request_args( + request_type=request_type, + call_args=call_args, + coordinator=coordinator, + future=future, + on_done_before_calls=on_done_before_calls, + on_done_after_calls=on_done_after_calls, + ) + finally: + if provided_checksum_algorithm is not None: + call_args.extra_args['ChecksumAlgorithm'] = ( + provided_checksum_algorithm + ) make_request_args['send_filepath'] = send_filepath make_request_args['checksum_config'] = checksum_config return make_request_args @@ -822,7 +1197,21 @@ def _get_make_request_args_get_object( checksum_config = awscrt.s3.S3ChecksumConfig(validate_response=True) if isinstance(call_args.fileobj, str): final_filepath = call_args.fileobj - recv_filepath = self._os_utils.get_temp_filename(final_filepath) + # A redirected download creates more than one CRT request for the + # same logical transfer. Keep one temp path so the final done + # callback handles the file used by every attempt. + recv_filepath = future.meta.user_context.get( + self._DOWNLOAD_TEMP_FILENAME + ) + if recv_filepath is None: + # Store the path before the first request so a redirected + # attempt reuses it. + recv_filepath = self._os_utils.get_temp_filename( + final_filepath + ) + future.meta.user_context[self._DOWNLOAD_TEMP_FILENAME] = ( + recv_filepath + ) on_done_before_calls.append( RenameTempFileHandler( coordinator, final_filepath, recv_filepath, self._os_utils diff --git a/awscli/topics/s3-config.rst b/awscli/topics/s3-config.rst index defa8b2da5ec..3e74fc322362 100644 --- a/awscli/topics/s3-config.rst +++ b/awscli/topics/s3-config.rst @@ -376,9 +376,6 @@ files to and from S3. Valid choices are: * S3 to S3 copies - Falls back to using the ``classic`` transfer client - * Region redirects - Transfers fail for requests sent to a region that does - not match the region of the targeted S3 bucket. - * ``max_concurrent_requests``, ``max_queue_size``, ``multipart_threshold``, and ``max_bandwidth`` configuration values - Ignores these configuration values. diff --git a/tests/functional/s3transfer/test_crt.py b/tests/functional/s3transfer/test_crt.py index d092982f082c..88b397e6640b 100644 --- a/tests/functional/s3transfer/test_crt.py +++ b/tests/functional/s3transfer/test_crt.py @@ -16,6 +16,7 @@ import time from concurrent.futures import Future +from botocore.exceptions import ClientError from botocore.session import Session from s3transfer.subscribers import BaseSubscriber @@ -49,16 +50,20 @@ class RecordingSubscriber(BaseSubscriber): def __init__(self): self.on_queued_called = False self.on_done_called = False + self.on_queued_calls = 0 + self.on_done_calls = 0 self.bytes_transferred = 0 self.on_queued_future = None self.on_done_future = None def on_queued(self, future, **kwargs): self.on_queued_called = True + self.on_queued_calls += 1 self.on_queued_future = future def on_done(self, future, **kwargs): self.on_done_called = True + self.on_done_calls += 1 self.on_done_future = future @@ -97,8 +102,9 @@ def setUp(self): self.request_serializer = s3transfer.crt.BotocoreCRTRequestSerializer( self.session ) + self.crt_client_factory = mock.Mock(return_value=self.s3_crt_client) self.transfer_manager = s3transfer.crt.CRTTransferManager( - crt_s3_client=self.s3_crt_client, + crt_client_factory=self.crt_client_factory, crt_request_serializer=self.request_serializer, ) self.record_subscriber = RecordingSubscriber() @@ -218,7 +224,7 @@ def _invoke_done_callbacks(self, **kwargs): callargs = self.s3_crt_client.make_request.call_args callargs_kwargs = callargs[1] on_done = callargs_kwargs["on_done"] - on_done(error=None) + on_done(error=kwargs.get('error')) def _simulate_file_download(self, recv_filepath): self.files.create_file( @@ -236,6 +242,42 @@ def _simulate_make_request_side_effect(self, **kwargs): self._invoke_done_callbacks() return self.s3_request + def _create_redirect_error(self, region=None): + headers = [] if region is None else [('x-amz-bucket-region', region)] + return awscrt.s3.S3ResponseError( + code=14343, + name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS', + message='Invalid response status from request', + status_code=301, + headers=headers, + body=b'PermanentRedirect', + operation_name='PutObject', + ) + + def _create_redirect_transfer_manager( + self, initial_client, client_factory + ): + def create_client(region=None): + if region is None: + return initial_client + return client_factory(region) + + return s3transfer.crt.CRTTransferManager( + crt_client_factory=create_client, + crt_request_serializer=self.request_serializer, + ) + + def _fail_make_request(self, error): + def make_request(**kwargs): + kwargs['on_done'](error=error) + return mock.Mock(awscrt.s3.S3Request) + + return make_request + + def _succeed_make_request(self, **kwargs): + kwargs['on_done'](error=None) + return mock.Mock(awscrt.s3.S3Request) + def test_upload(self): future = self.transfer_manager.upload( self.filename, self.bucket, self.key, {}, [self.record_subscriber] @@ -262,6 +304,153 @@ def test_upload(self): ) self._assert_subscribers_called(future) + def test_upload_redirects_and_reuses_cached_region(self): + redirected_region = 'eu-central-1' + redirect_error = self._create_redirect_error(redirected_region) + initial_client = mock.Mock(awscrt.s3.S3Client) + redirected_client = mock.Mock(awscrt.s3.S3Client) + initial_client.make_request.side_effect = self._fail_make_request( + redirect_error + ) + redirected_client.make_request.side_effect = self._succeed_make_request + client_factory = mock.Mock(return_value=redirected_client) + transfer_manager = self._create_redirect_transfer_manager( + initial_client, client_factory + ) + + first_subscriber = RecordingSubscriber() + first_future = transfer_manager.upload( + self.filename, + self.bucket, + self.key, + {}, + [first_subscriber], + ) + first_future.result() + + self.assertEqual(initial_client.make_request.call_count, 1) + self.assertEqual(redirected_client.make_request.call_count, 1) + client_factory.assert_called_once_with(redirected_region) + initial_call = initial_client.make_request.call_args_list[0].kwargs + redirected_call = redirected_client.make_request.call_args_list[ + 0 + ].kwargs + self.assertEqual( + initial_call['request'].headers.get('host'), + f's3.{self.region}.amazonaws.com', + ) + self.assertEqual( + redirected_call['request'].headers.get('host'), + f's3.{redirected_region}.amazonaws.com', + ) + # The redirect is internal to one logical transfer, so subscribers + # only see it once. + self.assertEqual(first_subscriber.on_queued_calls, 1) + self.assertEqual(first_subscriber.on_done_calls, 1) + + second_subscriber = RecordingSubscriber() + second_future = transfer_manager.upload( + self.filename, + self.bucket, + self.key, + {}, + [second_subscriber], + ) + second_future.result() + + self.assertEqual(initial_client.make_request.call_count, 1) + self.assertEqual(redirected_client.make_request.call_count, 2) + client_factory.assert_called_once_with(redirected_region) + self.assertEqual(second_subscriber.on_queued_calls, 1) + self.assertEqual(second_subscriber.on_done_calls, 1) + + def test_upload_redirect_restores_seekable_stream_position(self): + redirected_region = 'eu-central-1' + redirect_error = self._create_redirect_error(redirected_region) + initial_client = mock.Mock(awscrt.s3.S3Client) + redirected_client = mock.Mock(awscrt.s3.S3Client) + attempt_bodies = [] + + def consume_body_and_finish(error): + def make_request(**kwargs): + attempt_bodies.append( + kwargs['request'].body_stream._stream.read() + ) + kwargs['on_done'](error=error) + return mock.Mock(awscrt.s3.S3Request) + + return make_request + + initial_client.make_request.side_effect = consume_body_and_finish( + redirect_error + ) + redirected_client.make_request.side_effect = consume_body_and_finish( + None + ) + client_factory = mock.Mock(return_value=redirected_client) + transfer_manager = self._create_redirect_transfer_manager( + initial_client, client_factory + ) + + future = transfer_manager.upload( + io.BytesIO(self.expected_content), self.bucket, self.key, {}, [] + ) + future.result() + + self.assertEqual( + attempt_bodies, [self.expected_content, self.expected_content] + ) + + def test_upload_does_not_redirect_nonseekable_stream(self): + redirect_error = self._create_redirect_error('eu-central-1') + initial_client = mock.Mock(awscrt.s3.S3Client) + redirected_client = mock.Mock(awscrt.s3.S3Client) + initial_client.make_request.side_effect = self._fail_make_request( + redirect_error + ) + client_factory = mock.Mock(return_value=redirected_client) + transfer_manager = self._create_redirect_transfer_manager( + initial_client, client_factory + ) + + future = transfer_manager.upload( + NonSeekableReader(self.expected_content), + self.bucket, + self.key, + {}, + [], + ) + + with self.assertRaises(ClientError): + future.result() + client_factory.assert_not_called() + redirected_client.make_request.assert_not_called() + + def test_upload_does_not_redirect_after_progress(self): + redirect_error = self._create_redirect_error('eu-central-1') + initial_client = mock.Mock(awscrt.s3.S3Client) + redirected_client = mock.Mock(awscrt.s3.S3Client) + + def fail_after_progress(**kwargs): + kwargs['on_progress'](1) + kwargs['on_done'](error=redirect_error) + return mock.Mock(awscrt.s3.S3Request) + + initial_client.make_request.side_effect = fail_after_progress + client_factory = mock.Mock(return_value=redirected_client) + transfer_manager = self._create_redirect_transfer_manager( + initial_client, client_factory + ) + + future = transfer_manager.upload( + self.filename, self.bucket, self.key, {}, [] + ) + + with self.assertRaises(ClientError): + future.result() + client_factory.assert_not_called() + redirected_client.make_request.assert_not_called() + def test_upload_from_seekable_stream(self): with open(self.filename, 'rb') as f: future = self.transfer_manager.upload( @@ -667,13 +856,14 @@ def test_blocks_when_max_requests_processes_reached(self): def _cancel_function(self): self.cancel_called = True - self.s3_request.finished_future.set_exception( - awscrt.exceptions.from_code(0) - ) - self._invoke_done_callbacks() + error = awscrt.exceptions.from_code(0) + self.s3_request.finished_future.set_exception(error) + self._invoke_done_callbacks(error=error) def test_cancel(self): self.s3_request.finished_future = Future() + self.s3_crt_client.make_request.side_effect = None + self.s3_crt_client.make_request.return_value = self.s3_request self.cancel_called = False self.s3_request.cancel = self._cancel_function try: @@ -701,7 +891,7 @@ def serialize_http_request(self, transfer_type, future): not_impl_serializer = ExceptionRaisingSerializer() transfer_manager = s3transfer.crt.CRTTransferManager( - crt_s3_client=self.s3_crt_client, + crt_client_factory=self.crt_client_factory, crt_request_serializer=not_impl_serializer, ) future = transfer_manager.upload( diff --git a/tests/integration/s3transfer/test_crt.py b/tests/integration/s3transfer/test_crt.py index 2603cfeac6ac..7dba76a7089c 100644 --- a/tests/integration/s3transfer/test_crt.py +++ b/tests/integration/s3transfer/test_crt.py @@ -58,17 +58,25 @@ def setUp(self): self.s3_key = 's3key.txt' self.download_path = os.path.join(self.files.rootdir, 'download.txt') - def _create_s3_transfer(self): + def _create_s3_transfer(self, region=None): + if region is None: + region = self.region self.request_serializer = s3transfer.crt.BotocoreCRTRequestSerializer( - self.session, client_kwargs={'region_name': self.region} - ) - self.s3_crt_client = s3transfer.crt.create_s3_crt_client( - self.region, self._get_crt_credentials_provider() + self.session, client_kwargs={'region_name': region} ) + credentials_provider = self._get_crt_credentials_provider() + bootstrap = s3transfer.crt.create_crt_client_bootstrap() self.record_subscriber = RecordingSubscriber() self.osutil = OSUtils() return s3transfer.crt.CRTTransferManager( - self.s3_crt_client, self.request_serializer + crt_client_factory=lambda client_region=None: ( + s3transfer.crt.create_s3_crt_client( + client_region or region, + credentials_provider, + bootstrap=bootstrap, + ) + ), + crt_request_serializer=self.request_serializer, ) def _get_crt_credentials_provider(self): @@ -150,6 +158,22 @@ def test_upload_below_multipart_chunksize(self): self.assertTrue(self.object_exists('foo.txt')) self._assert_subscribers_called(file_size) + def test_upload_redirects_from_wrong_region(self): + wrong_region = ( + 'us-east-1' if self.region != 'us-east-1' else 'us-west-2' + ) + transfer = self._create_s3_transfer(region=wrong_region) + filename = self.files.create_file('redirect.txt', 'content') + self.addCleanup(self.delete_object, 'redirect.txt') + + with transfer: + future = transfer.upload( + filename, self.bucket_name, 'redirect.txt' + ) + future.result() + + self.assertTrue(self.object_exists('redirect.txt')) + def test_upload_above_multipart_chunksize(self): transfer = self._create_s3_transfer() file_size = 20 * 1024 * 1024 diff --git a/tests/unit/customizations/s3/test_factory.py b/tests/unit/customizations/s3/test_factory.py index 9fafd4a9d588..b62abd1a912a 100644 --- a/tests/unit/customizations/s3/test_factory.py +++ b/tests/unit/customizations/s3/test_factory.py @@ -205,6 +205,8 @@ def assert_is_classic_manager(self, manager): def assert_is_crt_manager(self, manager): self.assertIsInstance(manager, CRTTransferManager) + # The CRT client is initialized lazily once the region is known + manager.get_crt_client() def assert_expected_throughput_target_gbps( self, mock_crt_client, expected_throughput_target_gbps @@ -284,6 +286,36 @@ def test_uses_region_parameter_for_crt_manager(self, mock_crt_client): 'param-region', ) + @mock.patch('s3transfer.crt.S3Client') + def test_creates_crt_client_for_redirected_region(self, mock_crt_client): + self.runtime_config = self.get_runtime_config( + preferred_transfer_client='crt' + ) + transfer_manager = self.factory.create_transfer_manager( + self.params, self.runtime_config + ) + + # Client creation is lazy and each selected region is cached. + self.assertEqual(mock_crt_client.call_count, 0) + self.assertIs( + transfer_manager.get_crt_client(), + transfer_manager.get_crt_client(), + ) + self.assertIs( + transfer_manager.get_crt_client('eu-central-1'), + transfer_manager.get_crt_client('eu-central-1'), + ) + + self.assertEqual(mock_crt_client.call_count, 2) + self.assertEqual( + mock_crt_client.call_args_list[0].kwargs['region'], + 'us-west-2', + ) + self.assertEqual( + mock_crt_client.call_args_list[1].kwargs['region'], + 'eu-central-1', + ) + @mock.patch('s3transfer.crt.S3Client') def test_falls_back_to_session_region_for_crt_manager( self, mock_crt_client diff --git a/tests/unit/s3transfer/test_crt.py b/tests/unit/s3transfer/test_crt.py index 740131c8be81..d08db2de5b6a 100644 --- a/tests/unit/s3transfer/test_crt.py +++ b/tests/unit/s3transfer/test_crt.py @@ -11,12 +11,14 @@ # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import io +from concurrent.futures import Future import pytest from botocore.credentials import Credentials, ReadOnlyCredentials from botocore.exceptions import ( ClientError, InvalidConfigError, + InvalidRegionError, NoCredentialsError, ) from botocore.session import Session @@ -169,14 +171,10 @@ def test_delete_request(self): self.assertIsNone(crt_request.headers.get("Authorization")) def _create_crt_response_error( - self, status_code, body, operation_name=None + self, status_code, body, operation_name=None, headers=None ): - return awscrt.s3.S3ResponseError( - code=14343, - name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS', - message='Invalid response status from request', - status_code=status_code, - headers=[ + if headers is None: + headers = [ ('x-amz-request-id', 'QSJHJJZR2EDYD4GQ'), ( 'x-amz-id-2', @@ -186,11 +184,26 @@ def _create_crt_response_error( ('Transfer-Encoding', 'chunked'), ('Date', 'Fri, 10 Nov 2023 23:22:47 GMT'), ('Server', 'AmazonS3'), - ], + ] + return awscrt.s3.S3ResponseError( + code=14343, + name='AWS_ERROR_S3_INVALID_RESPONSE_STATUS', + message='Invalid response status from request', + status_code=status_code, + headers=headers, body=body, operation_name=operation_name, ) + def _create_serializer_with_redirect_client(self): + redirect_client = mock.Mock() + client_factory = mock.Mock(return_value=redirect_client) + serializer = s3transfer.crt.BotocoreCRTRequestSerializer( + self.session, + region_redirect_client_factory=client_factory, + ) + return serializer, redirect_client, client_factory + def test_translate_get_object_404(self): body = ( b'\n' @@ -222,6 +235,98 @@ def test_translate_unknown_operation_404(self): boto_err = self.request_serializer.translate_crt_exception(crt_exc) self.assertIsInstance(boto_err, ClientError) + def test_cached_bucket_region_changes_serialized_endpoint(self): + self.request_serializer.cache_bucket_region( + self.bucket, 'eu-central-1' + ) + callargs = CallArgs( + bucket=self.bucket, + key=self.key, + fileobj=self.filename, + extra_args={}, + subscribers=[], + ) + coordinator = s3transfer.crt.CRTTransferCoordinator() + future = s3transfer.crt.CRTTransferFuture( + s3transfer.crt.CRTTransferMeta(call_args=callargs), coordinator + ) + + crt_request = self.request_serializer.serialize_http_request( + "get_object", future + ) + + self.assertEqual( + crt_request.headers.get("host"), + "s3.eu-central-1.amazonaws.com", + ) + + def test_redirect_region_does_not_create_fallback_client(self): + serializer, redirect_client, client_factory = ( + self._create_serializer_with_redirect_client() + ) + error = self._create_crt_response_error( + 301, + None, + operation_name='GetObject', + headers=[('x-amz-bucket-region', 'eu-central-1')], + ) + + region = serializer.get_bucket_region(self.bucket, 'get_object', error) + + self.assertEqual(region, 'eu-central-1') + client_factory.assert_not_called() + redirect_client.head_bucket.assert_not_called() + + def test_redirect_region_creates_fallback_client(self): + serializer, redirect_client, client_factory = ( + self._create_serializer_with_redirect_client() + ) + redirect_client.head_bucket.return_value = { + 'ResponseMetadata': { + 'HTTPHeaders': {'x-amz-bucket-region': 'eu-central-1'} + } + } + error = self._create_crt_response_error( + 301, + b'PermanentRedirect', + operation_name='DeleteObject', + ) + + region = serializer.get_bucket_region( + self.bucket, 'delete_object', error + ) + + self.assertEqual(region, 'eu-central-1') + client_factory.assert_called_once_with() + redirect_client.head_bucket.assert_called_once_with(Bucket=self.bucket) + + def test_redirect_region_rejects_invalid_region(self): + error = self._create_crt_response_error( + 301, + b'PermanentRedirect', + operation_name='GetObject', + headers=[('x-amz-bucket-region', 'invalid region!')], + ) + with self.assertRaises(InvalidRegionError): + self.request_serializer.get_bucket_region( + self.bucket, 'get_object', error + ) + + def test_redirect_region_ignores_arn_bucket(self): + error = self._create_crt_response_error( + 301, + b'PermanentRedirect', + operation_name='GetObject', + headers=[('x-amz-bucket-region', 'eu-central-1')], + ) + bucket = 'arn:aws:s3:us-west-2:123456789012:accesspoint/myendpoint' + + region = self.request_serializer.get_bucket_region( + bucket, 'get_object', error + ) + + self.assertIsNone(region) + @requires_crt_pytest class TestBotocoreCRTCredentialsWrapper: @@ -301,7 +406,9 @@ def setUp(self): self.mock_s3_request = mock.Mock(awscrt.s3.S3RequestType) self.mock_crt_future = mock.Mock(awscrt.s3.Future) self.mock_s3_request.finished_future = self.mock_crt_future - self.coordinator = s3transfer.crt.CRTTransferCoordinator() + self.coordinator = s3transfer.crt.CRTTransferCoordinator( + completion_future=self.mock_crt_future + ) self.coordinator.set_s3_request(self.mock_s3_request) self.future = s3transfer.crt.CRTTransferFuture( coordinator=self.coordinator @@ -324,6 +431,164 @@ def test_set_exception_can_override_previous_exception(self): self.future.result() +@requires_crt_pytest +class TestCRTTransferCoordinator: + def setup_method(self): + self.completion_future = Future() + self.coordinator = s3transfer.crt.CRTTransferCoordinator( + completion_future=self.completion_future + ) + + def create_s3_request(self): + s3_request = mock.Mock(awscrt.s3.S3Request) + s3_request.finished_future = Future() + return s3_request + + def test_set_s3_request(self): + s3_request = self.create_s3_request() + self.coordinator.set_s3_request(s3_request) + assert self.coordinator.s3_request is s3_request + + def test_original_request_cannot_replace_retry(self): + first_request = self.create_s3_request() + second_request = self.create_s3_request() + # The retry started before the original request registered its native + # request, so the original request must not become active. + self.coordinator.set_s3_request(second_request, is_retry=True) + self.coordinator.set_s3_request(first_request) + + assert self.coordinator.s3_request is second_request + + def test_cancel_cancels_retry_request(self): + first_request = self.create_s3_request() + second_request = self.create_s3_request() + self.coordinator.set_s3_request(first_request) + self.coordinator.set_s3_request(second_request, is_retry=True) + + self.coordinator.cancel() + + second_request.cancel.assert_called_once_with() + first_request.cancel.assert_not_called() + + def test_cancel_before_request_cancels_request(self): + self.coordinator.cancel() + s3_request = self.create_s3_request() + + self.coordinator.set_s3_request(s3_request) + + assert self.coordinator.cancelled + s3_request.cancel.assert_called_once_with() + + def test_default_completion_future_can_complete_without_request(self): + coordinator = s3transfer.crt.CRTTransferCoordinator() + + coordinator.complete() + + assert coordinator.done() + assert coordinator.result() is None + + def test_complete_resolves_completion_future(self): + self.coordinator.complete() + assert self.completion_future.done() + assert self.coordinator.result() is None + + def test_complete_with_error(self): + self.coordinator.complete(CustomFutureException()) + with pytest.raises(CustomFutureException): + self.coordinator.result() + + def test_complete_is_idempotent(self): + self.coordinator.complete() + self.coordinator.complete(CustomFutureException()) + assert self.coordinator.result() is None + + def test_not_done_until_completed(self): + s3_request = self.create_s3_request() + self.coordinator.set_s3_request(s3_request) + # A native CRT request failing does not complete the transfer, + # since it may still be redirected to another region. + s3_request.finished_future.set_exception(CustomFutureException()) + + assert not self.coordinator.done() + + self.coordinator.complete() + assert self.coordinator.done() + + +@requires_crt_pytest +class TestS3RegionRedirectPolicy: + def setup_method(self): + self.bucket = 'mybucket' + self.error = Exception('wrong region') + self.serializer = mock.Mock( + s3transfer.crt.BotocoreCRTRequestSerializer + ) + self.serializer.get_cached_bucket_region.return_value = None + self.serializer.get_bucket_region.return_value = 'eu-central-1' + self.policy = s3transfer.crt.CRTS3RegionRedirectPolicy(self.serializer) + + def get_retry_region(self, **overrides): + kwargs = { + 'bucket': self.bucket, + 'transfer_type': 'put_object', + 'error': self.error, + 'is_retry': False, + 'bytes_transferred': 0, + 'cancelled': False, + 'is_replayable': True, + } + kwargs.update(overrides) + return self.policy.get_retry_region(**kwargs) + + def test_returns_and_caches_discovered_region(self): + assert self.get_retry_region() == 'eu-central-1' + self.serializer.cache_bucket_region.assert_called_once_with( + self.bucket, 'eu-central-1' + ) + + def test_returns_none_when_region_not_discovered(self): + self.serializer.get_bucket_region.return_value = None + assert self.get_retry_region() is None + self.serializer.cache_bucket_region.assert_not_called() + + def test_returns_none_for_no_error(self): + assert self.get_retry_region(error=None) is None + self.serializer.get_bucket_region.assert_not_called() + + def test_returns_none_after_redirect(self): + assert self.get_retry_region(is_retry=True) is None + self.serializer.get_bucket_region.assert_not_called() + + def test_returns_none_after_bytes_transferred(self): + assert self.get_retry_region(bytes_transferred=1) is None + self.serializer.get_bucket_region.assert_not_called() + + def test_returns_none_when_cancelled(self): + assert self.get_retry_region(cancelled=True) is None + self.serializer.get_bucket_region.assert_not_called() + + def test_returns_none_when_stream_is_not_replayable(self): + assert self.get_retry_region(is_replayable=False) is None + self.serializer.get_bucket_region.assert_not_called() + + def test_returns_none_for_s3express_bucket(self): + region = self.get_retry_region(bucket='mybucket--usw2-az5--x-s3') + assert region is None + self.serializer.get_bucket_region.assert_not_called() + + def test_returns_none_when_discovery_raises(self): + self.serializer.get_bucket_region.side_effect = InvalidRegionError( + region_name='not a region!' + ) + assert self.get_retry_region() is None + self.serializer.cache_bucket_region.assert_not_called() + + def test_get_cached_bucket_region(self): + self.serializer.get_cached_bucket_region.return_value = 'eu-west-1' + region = self.policy.get_cached_bucket_region(self.bucket) + assert region == 'eu-west-1' + + @requires_crt() class TestOnBodyFileObjWriter(unittest.TestCase): def test_call(self): From a526dd0c4fcac52e713d6a561f185747896bb319 Mon Sep 17 00:00:00 2001 From: Alex Shovlin Date: Tue, 8 Sep 2026 21:12:06 -0400 Subject: [PATCH 2/3] Fix test hang --- tests/functional/s3/__init__.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/functional/s3/__init__.py b/tests/functional/s3/__init__.py index 76bb8d076ccc..f9a6cff6e6b8 100644 --- a/tests/functional/s3/__init__.py +++ b/tests/functional/s3/__init__.py @@ -505,6 +505,7 @@ def simulate_make_request_side_effect(self, *args, **kwargs): s3_request = FakeCRTS3Request( future=FakeCRTFuture(kwargs.get('on_done')) ) + s3_request.finished_future.result() return s3_request def simulate_file_download(self, recv_filepath): From c9feb205331ca32389a4b9d0bc908970ef02bf5e Mon Sep 17 00:00:00 2001 From: Alex Shovlin Date: Thu, 10 Sep 2026 11:05:31 -0400 Subject: [PATCH 3/3] Fix correctness and threading issues in CRT S3 region redirects --- .../next-release/enhancement-s3-42614.json | 2 +- awscli/customizations/s3/factory.py | 8 +- awscli/s3transfer/crt.py | 325 ++++++++++++++---- awscli/topics/s3-config.rst | 7 + tests/functional/s3transfer/test_crt.py | 311 ++++++++++++++--- tests/unit/customizations/s3/test_factory.py | 23 +- tests/unit/s3transfer/test_crt.py | 101 ++++-- 7 files changed, 616 insertions(+), 161 deletions(-) diff --git a/.changes/next-release/enhancement-s3-42614.json b/.changes/next-release/enhancement-s3-42614.json index 42e801c8284d..ba821821fb0c 100644 --- a/.changes/next-release/enhancement-s3-42614.json +++ b/.changes/next-release/enhancement-s3-42614.json @@ -1,5 +1,5 @@ { "type": "enhancement", - "category": "``s3```", + "category": "``s3``", "description": "Follow bucket region redirects for the ``s3`` commands when CRT is enabled." } diff --git a/awscli/customizations/s3/factory.py b/awscli/customizations/s3/factory.py index ab8b6da703fa..5a618a121f93 100644 --- a/awscli/customizations/s3/factory.py +++ b/awscli/customizations/s3/factory.py @@ -117,7 +117,7 @@ def _create_crt_transfer_manager(self, params, runtime_config): self._acquire_crt_s3_process_lock() region = self._resolve_region(params) bootstrap = create_crt_client_bootstrap() - return CRTTransferManager( + transfer_manager = CRTTransferManager( crt_client_factory=lambda client_region=None: ( self._create_crt_client( params, @@ -128,6 +128,12 @@ def _create_crt_transfer_manager(self, params, runtime_config): ), crt_request_serializer=self._create_crt_request_serializer(params), ) + # Clients for redirected regions are created on demand, but create the + # one for the configured region now. Otherwise invalid client + # configuration is not reported until a transfer is submitted, which + # reports it once per object instead of once for the command. + transfer_manager.get_crt_client() + return transfer_manager def _create_crt_client( self, params, runtime_config, region=None, bootstrap=None diff --git a/awscli/s3transfer/crt.py b/awscli/s3transfer/crt.py index 2cad7bfc8d23..6d9d25622bb7 100644 --- a/awscli/s3transfer/crt.py +++ b/awscli/s3transfer/crt.py @@ -10,6 +10,7 @@ # 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. +import contextlib import logging import re import threading @@ -228,49 +229,61 @@ class CRTS3RegionRedirectPolicy: def __init__(self, crt_request_serializer): self._crt_request_serializer = crt_request_serializer + # Bucket regions are held under one lock so that a burst of transfers + # failing at once shares a single lookup instead of each paying for + # its own, potentially a HeadBucket request each, and so that a + # request cannot be built for a region that changes while it is built. + # One lock covers every bucket because a command generally transfers + # to or from a single one. + self._region_lock = threading.Lock() def get_cached_bucket_region(self, bucket): """Return a region already discovered for a bucket, if any.""" - region = self._crt_request_serializer.get_cached_bucket_region(bucket) - if region is not None: - logger.debug( - 'Using cached region %s for S3 bucket %s', region, bucket - ) - return region + return self._crt_request_serializer.get_cached_bucket_region(bucket) - def get_retry_region( + @contextlib.contextmanager + def locked_bucket_region(self, bucket): + """Hold a bucket's region steady while a request is built for it. + + Serializing a request resolves its endpoint from the region cached for + its bucket. The caller has to select a client for that same region, so + the region must not change in between. + """ + with self._region_lock: + yield self.get_cached_bucket_region(bucket) + + def is_error_redirect_candidate( self, bucket, - transfer_type, - error, - is_retry, + is_region_redirect, bytes_transferred, cancelled, is_replayable, ): - """Return the region to retry a failed request in, or ``None``. + """Return whether a failed request may be worth redirecting. - A returned region has been cached, so both the retried request and - later transfers to the same bucket use it. + Callers only ask this about a request that failed, so this inspects + the state of the transfer rather than the error itself. These checks + are cheap and never do I/O, so a caller running on a CRT completion + thread can use them to decide whether discovering a region is worth + handing off to another thread. """ - if error is None: - return None - if is_retry: + if is_region_redirect: logger.debug( 'Transfer for bucket %s was already redirected, not ' 'redirecting again.', bucket, ) - return None + return False if cancelled: - return None + return False if not is_replayable: logger.debug( 'Not redirecting transfer for bucket %s because its stream ' 'cannot be replayed.', bucket, ) - return None + return False if bytes_transferred: # Replaying a request that moved data would either duplicate # bytes or double-count progress. @@ -280,9 +293,58 @@ def get_retry_region( bucket, bytes_transferred, ) - return None + return False if is_s3express_bucket(bucket): + return False + return True + + def get_retry_region( + self, bucket, transfer_type, error, request_region=None + ): + """Return the region to retry a failed request in, or ``None``. + + A returned region has been cached, so both the retried request and + later transfers to the same bucket use it. This discovers the region + of a bucket, which may require an additional ``HeadBucket`` request, + so it must not be called from a CRT completion thread. + + :type request_region: Optional[str] + :param request_region: The region the failed request was made in, or + ``None`` if it used the configured region. + """ + retry_region = self._get_cached_retry_region(bucket, request_region) + if retry_region is not None: + return retry_region + with self._region_lock: + # Another transfer may have discovered the region while this one + # waited for the lock. + retry_region = self._get_cached_retry_region( + bucket, request_region + ) + if retry_region is not None: + return retry_region + return self._discover_bucket_region( + bucket, transfer_type, error, request_region + ) + + def _get_cached_retry_region(self, bucket, request_region): + """Return an already discovered region the failed request did not use. + + A region another transfer discovered is worth retrying in, but the one + the request just failed in is not. + """ + cached_region = self.get_cached_bucket_region(bucket) + if cached_region is None: + return None + if cached_region == request_region: + # The failed request already used this region, so the cached + # region is stale and retrying there would fail the same way. return None + return cached_region + + def _discover_bucket_region( + self, bucket, transfer_type, error, request_region=None + ): try: new_region = self._crt_request_serializer.get_bucket_region( bucket, transfer_type, error @@ -295,6 +357,20 @@ def get_retry_region( return None if new_region is None: return None + if new_region == ( + request_region + or self._crt_request_serializer.get_configured_region() + ): + # The failed request was already made in this region, so retrying + # it there would fail the same way. Leaving it out of the cache + # also keeps later transfers on the client they already use. + logger.debug( + 'Not redirecting transfer for bucket %s because it was ' + 'already made in region %s.', + bucket, + new_region, + ) + return None logger.debug( 'Redirecting CRT S3 transfer for bucket %s to region %s', bucket, @@ -489,22 +565,29 @@ def _submit_transfer(self, request_type, call_args): ) on_queued() - def create_request(is_retry): + def create_request(is_region_redirect): # Reset the stream if we're redirecting due to bucket region - if is_retry and upload_stream_position is not None: + if is_region_redirect and upload_stream_position is not None: call_args.fileobj.seek(upload_stream_position) - crt_callargs = self._s3_args_creator.get_make_request_args( - request_type, - call_args, - coordinator, - future, - on_done_after_calls, - ) - region = self._region_redirect_policy.get_cached_bucket_region( - bucket - ) - crt_client = self.get_crt_client(region) - return crt_client, crt_callargs + policy = self._region_redirect_policy + with policy.locked_bucket_region(bucket) as region: + if region is not None: + logger.debug( + 'Using cached region %s for S3 bucket %s', + region, + bucket, + ) + crt_callargs = ( + self._s3_args_creator.get_make_request_args( + request_type, + call_args, + coordinator, + future, + on_done_after_calls, + ) + ) + crt_client = self.get_crt_client(region) + return crt_client, crt_callargs, region coordinator.submit( create_request, @@ -603,6 +686,26 @@ def serialize_http_request(self, transfer_type, future): def translate_crt_exception(self, exception): raise NotImplementedError('translate_crt_exception()') + def get_cached_bucket_region(self, bucket): + """Return the region already discovered for a bucket, if any. + + Serializers that do not support bucket region redirects never have a + region to report, which keeps their transfers in the configured + region. + """ + return None + + def cache_bucket_region(self, bucket, region): + """Remember the region a bucket was found in.""" + + def get_bucket_region(self, bucket, transfer_type, error): + """Return the region a failed request should be retried in, if any.""" + return None + + def get_configured_region(self): + """Return the region requests are serialized for by default.""" + return None + class LazyHeadBucketClient: """ @@ -835,6 +938,9 @@ def cache_bucket_region(self, bucket, region): def get_cached_bucket_region(self, bucket): return self._region_cache.get(bucket) + def get_configured_region(self): + return self._client.meta.region_name + def get_bucket_region(self, bucket, transfer_type, error): """Extract a redirect region from a CRT response error. This adapts the CRT error for S3RegionRedirectorv2. @@ -922,6 +1028,12 @@ def __init__( self._done_event = threading.Event() self._cancelled = False self._redirect_retry_started = False + # Set by submit(), and the same for every request the transfer makes. + self._request_factory = None + self._region_redirect_policy = None + self._bucket = None + self._transfer_type = None + self._is_replayable = True @property def s3_request(self): @@ -990,18 +1102,29 @@ def submit( bucket, transfer_type, is_replayable=True, - is_retry=False, ): + """Submit the transfer's CRT request. + + A request that failed because it was made in the wrong region for its + bucket is resubmitted in the bucket's region, which makes a second + request for the same transfer. """ - Submits a single CRT request which can either be our first attempt or - a second attempt with a region redirect. - """ + self._request_factory = request_factory + self._region_redirect_policy = region_redirect_policy + self._bucket = bucket + self._transfer_type = transfer_type + self._is_replayable = is_replayable + self._start_request(is_region_redirect=False) + + def _start_request(self, is_region_redirect): with self._lock: if self._cancelled: raise CancelledError() - if is_retry: + if is_region_redirect: self._redirect_retry_started = True - crt_client, crt_callargs = request_factory(is_retry) + crt_client, crt_callargs, request_region = self._request_factory( + is_region_redirect + ) on_done = crt_callargs['on_done'] on_progress = crt_callargs['on_progress'] bytes_transferred = 0 @@ -1011,46 +1134,106 @@ def track_progress(transferred): bytes_transferred += transferred on_progress(transferred) - def request_done(error=None, **kwargs): - new_region = region_redirect_policy.get_retry_region( - bucket=bucket, - transfer_type=transfer_type, - error=error, - is_retry=is_retry, - bytes_transferred=bytes_transferred, - cancelled=self.cancelled, - is_replayable=is_replayable, - ) - if new_region is not None: - try: - self.submit( - request_factory, - region_redirect_policy, - bucket, - transfer_type, - is_replayable=is_replayable, - is_retry=True, - ) - return - except Exception as retry_error: - retry_error.__cause__ = error - error = retry_error - self.set_exception(retry_error, True) + def finish(error, kwargs): self.complete(error) on_done(error=error, **kwargs) + def redirect_and_finish(error, kwargs): + # Any failure deciding on or starting a redirect must still + # complete the transfer. Otherwise the transfer is never marked + # done and anything waiting on its result blocks forever. + try: + new_region = self._region_redirect_policy.get_retry_region( + self._bucket, + self._transfer_type, + error, + request_region, + ) + if new_region is not None: + try: + self._start_request(is_region_redirect=True) + return + except Exception as retry_error: + retry_error.__cause__ = error + error = retry_error + self.set_exception(retry_error, True) + except Exception as redirect_error: + logger.debug( + 'Unable to determine whether to redirect transfer for ' + 'bucket %s.', + self._bucket, + exc_info=redirect_error, + ) + if error is None: + error = redirect_error + self.set_exception(redirect_error, True) + finish(error, kwargs) + + def request_done(error=None, **kwargs): + if error is not None and self._can_redirect( + is_region_redirect, bytes_transferred + ): + # Discovering a region and serializing the retry can both + # block, and this runs on a CRT completion thread, where + # blocking stalls every other transfer sharing the event loop. + self._dispatch_redirect(redirect_and_finish, error, kwargs) + return + # Nothing to discover, so finish on this thread rather than paying + # for a handoff on every completed transfer. + finish(error, kwargs) + crt_callargs['on_done'] = request_done crt_callargs['on_progress'] = track_progress s3_request = crt_client.make_request(**crt_callargs) - self.set_s3_request(s3_request, is_retry=is_retry) + self.set_s3_request(s3_request, is_region_redirect=is_region_redirect) + + def _can_redirect(self, is_region_redirect, bytes_transferred): + try: + return self._region_redirect_policy.is_error_redirect_candidate( + bucket=self._bucket, + is_region_redirect=is_region_redirect, + bytes_transferred=bytes_transferred, + cancelled=self.cancelled, + is_replayable=self._is_replayable, + ) + except Exception as redirect_error: + logger.debug( + 'Unable to determine whether transfer for bucket %s can be ' + 'redirected.', + self._bucket, + exc_info=redirect_error, + ) + return False + + def _dispatch_redirect(self, fn, *args): + """Run a region redirect off of the CRT completion thread. + + Discovering a region and serializing the retry can both block, which + would stall the event loop shared by every in-flight transfer. A + transfer is redirected at most once and only when it fails, so these + threads are few and short lived. + """ + try: + threading.Thread( + target=fn, args=args, name='crt-s3-region-redirect' + ).start() + except RuntimeError as thread_error: + # The OS refused a new thread. Blocking this thread is still + # better than stranding the transfer. + logger.debug( + 'Unable to hand off S3 region redirect, handling it inline.', + exc_info=thread_error, + ) + fn(*args) - def set_s3_request(self, s3_request, is_retry=False): + def set_s3_request(self, s3_request, is_region_redirect=False): + """Make a CRT request the one the transfer acts on.""" with self._lock: - if not is_retry and self._redirect_retry_started: - # The retry is already active. The original request completed - # and redirected before make_request() returned. + if not is_region_redirect and self._redirect_retry_started: + # The redirect is already active. The original request + # completed and redirected before make_request() returned. return - if is_retry: + if is_region_redirect: self._redirect_retry_started = True self._s3_request = s3_request cancelled = self._cancelled diff --git a/awscli/topics/s3-config.rst b/awscli/topics/s3-config.rst index 3e74fc322362..47ce57dea195 100644 --- a/awscli/topics/s3-config.rst +++ b/awscli/topics/s3-config.rst @@ -376,6 +376,13 @@ files to and from S3. Valid choices are: * S3 to S3 copies - Falls back to using the ``classic`` transfer client + * Some region redirects - Transfers to a region that does not match the + region of the targeted S3 bucket are retried in the bucket's region, but + only if the transfer can be replayed from the start. A transfer fails + instead of being redirected if it is an upload whose source cannot be + rewound or if any of its data was already transferred. The ``classic`` + transfer client is able to redirect in both of those cases. + * ``max_concurrent_requests``, ``max_queue_size``, ``multipart_threshold``, and ``max_bandwidth`` configuration values - Ignores these configuration values. diff --git a/tests/functional/s3transfer/test_crt.py b/tests/functional/s3transfer/test_crt.py index 88b397e6640b..64053183eccb 100644 --- a/tests/functional/s3transfer/test_crt.py +++ b/tests/functional/s3transfer/test_crt.py @@ -34,6 +34,10 @@ import awscrt import s3transfer.crt +# Bound on waiting for a transfer that completes from another thread, so a +# transfer that never completes fails the test instead of hanging it. +RESULT_TIMEOUT = 20 + class submitThread(threading.Thread): def __init__(self, transfer_manager, futures, callargs): @@ -108,6 +112,7 @@ def setUp(self): crt_request_serializer=self.request_serializer, ) self.record_subscriber = RecordingSubscriber() + self.completion_threads = [] def tearDown(self): self.files.remove_all() @@ -267,6 +272,39 @@ def create_client(region=None): crt_request_serializer=self.request_serializer, ) + def _create_redirecting_transfer_manager( + self, initial_make_request, redirected_make_request=None + ): + """Create a manager whose initial region and redirected region differ. + + The clients for both regions and the factory that creates the + redirected one are recorded as ``self.initial_client``, + ``self.redirected_client``, and ``self.redirected_client_factory``. + """ + self.initial_client = mock.Mock(awscrt.s3.S3Client) + self.initial_client.make_request.side_effect = initial_make_request + self.redirected_client = mock.Mock(awscrt.s3.S3Client) + self.redirected_client.make_request.side_effect = ( + redirected_make_request or self._succeed_make_request + ) + self.redirected_client_factory = mock.Mock( + return_value=self.redirected_client + ) + return self._create_redirect_transfer_manager( + self.initial_client, self.redirected_client_factory + ) + + def _upload_and_wait(self, transfer_manager, subscribers=None): + future = transfer_manager.upload( + self.filename, + self.bucket, + self.key, + {}, + subscribers if subscribers is not None else [], + ) + future.result(timeout=RESULT_TIMEOUT) + return future + def _fail_make_request(self, error): def make_request(**kwargs): kwargs['on_done'](error=error) @@ -274,6 +312,27 @@ def make_request(**kwargs): return make_request + def _fail_make_request_on_other_thread(self, error): + """Fail a request from another thread, like a CRT completion thread. + + The thread the request completed on is recorded in + ``self.completion_threads``. + """ + + def complete_request(on_done): + self.completion_threads.append(threading.get_ident()) + on_done(error=error) + + def make_request(**kwargs): + thread = threading.Thread( + target=complete_request, args=(kwargs['on_done'],) + ) + self.addCleanup(thread.join) + thread.start() + return mock.Mock(awscrt.s3.S3Request) + + return make_request + def _succeed_make_request(self, **kwargs): kwargs['on_done'](error=None) return mock.Mock(awscrt.s3.S3Request) @@ -306,16 +365,10 @@ def test_upload(self): def test_upload_redirects_and_reuses_cached_region(self): redirected_region = 'eu-central-1' - redirect_error = self._create_redirect_error(redirected_region) - initial_client = mock.Mock(awscrt.s3.S3Client) - redirected_client = mock.Mock(awscrt.s3.S3Client) - initial_client.make_request.side_effect = self._fail_make_request( - redirect_error - ) - redirected_client.make_request.side_effect = self._succeed_make_request - client_factory = mock.Mock(return_value=redirected_client) - transfer_manager = self._create_redirect_transfer_manager( - initial_client, client_factory + transfer_manager = self._create_redirecting_transfer_manager( + self._fail_make_request( + self._create_redirect_error(redirected_region) + ) ) first_subscriber = RecordingSubscriber() @@ -328,11 +381,15 @@ def test_upload_redirects_and_reuses_cached_region(self): ) first_future.result() - self.assertEqual(initial_client.make_request.call_count, 1) - self.assertEqual(redirected_client.make_request.call_count, 1) - client_factory.assert_called_once_with(redirected_region) - initial_call = initial_client.make_request.call_args_list[0].kwargs - redirected_call = redirected_client.make_request.call_args_list[ + self.assertEqual(self.initial_client.make_request.call_count, 1) + self.assertEqual(self.redirected_client.make_request.call_count, 1) + self.redirected_client_factory.assert_called_once_with( + redirected_region + ) + initial_call = self.initial_client.make_request.call_args_list[ + 0 + ].kwargs + redirected_call = self.redirected_client.make_request.call_args_list[ 0 ].kwargs self.assertEqual( @@ -358,17 +415,17 @@ def test_upload_redirects_and_reuses_cached_region(self): ) second_future.result() - self.assertEqual(initial_client.make_request.call_count, 1) - self.assertEqual(redirected_client.make_request.call_count, 2) - client_factory.assert_called_once_with(redirected_region) + self.assertEqual(self.initial_client.make_request.call_count, 1) + self.assertEqual(self.redirected_client.make_request.call_count, 2) + self.redirected_client_factory.assert_called_once_with( + redirected_region + ) self.assertEqual(second_subscriber.on_queued_calls, 1) self.assertEqual(second_subscriber.on_done_calls, 1) def test_upload_redirect_restores_seekable_stream_position(self): redirected_region = 'eu-central-1' redirect_error = self._create_redirect_error(redirected_region) - initial_client = mock.Mock(awscrt.s3.S3Client) - redirected_client = mock.Mock(awscrt.s3.S3Client) attempt_bodies = [] def consume_body_and_finish(error): @@ -381,15 +438,9 @@ def make_request(**kwargs): return make_request - initial_client.make_request.side_effect = consume_body_and_finish( - redirect_error - ) - redirected_client.make_request.side_effect = consume_body_and_finish( - None - ) - client_factory = mock.Mock(return_value=redirected_client) - transfer_manager = self._create_redirect_transfer_manager( - initial_client, client_factory + transfer_manager = self._create_redirecting_transfer_manager( + consume_body_and_finish(redirect_error), + consume_body_and_finish(None), ) future = transfer_manager.upload( @@ -401,16 +452,39 @@ def make_request(**kwargs): attempt_bodies, [self.expected_content, self.expected_content] ) - def test_upload_does_not_redirect_nonseekable_stream(self): - redirect_error = self._create_redirect_error('eu-central-1') - initial_client = mock.Mock(awscrt.s3.S3Client) - redirected_client = mock.Mock(awscrt.s3.S3Client) - initial_client.make_request.side_effect = self._fail_make_request( - redirect_error + def test_successful_upload_does_not_consult_redirect_policy(self): + # A transfer that did not fail is never a redirect candidate. + with mock.patch.object( + self.transfer_manager._region_redirect_policy, + 'is_error_redirect_candidate', + ) as is_error_redirect_candidate: + future = self.transfer_manager.upload( + self.filename, self.bucket, self.key, {}, [] + ) + future.result(timeout=RESULT_TIMEOUT) + + is_error_redirect_candidate.assert_not_called() + + def test_upload_does_not_redirect_to_configured_region(self): + # A redirect naming the region the request already used, e.g. from an + # accelerate or dualstack endpoint, is not worth retrying. + transfer_manager = self._create_redirecting_transfer_manager( + self._fail_make_request(self._create_redirect_error(self.region)) ) - client_factory = mock.Mock(return_value=redirected_client) - transfer_manager = self._create_redirect_transfer_manager( - initial_client, client_factory + + with self.assertRaises(ClientError): + self._upload_and_wait(transfer_manager) + + # No duplicate client for a region the transfer already used, and no + # retry that would just fail again. + self.redirected_client_factory.assert_not_called() + self.assertEqual(self.initial_client.make_request.call_count, 1) + + def test_upload_does_not_redirect_nonseekable_stream(self): + transfer_manager = self._create_redirecting_transfer_manager( + self._fail_make_request( + self._create_redirect_error('eu-central-1') + ) ) future = transfer_manager.upload( @@ -423,33 +497,170 @@ def test_upload_does_not_redirect_nonseekable_stream(self): with self.assertRaises(ClientError): future.result() - client_factory.assert_not_called() - redirected_client.make_request.assert_not_called() + self.redirected_client_factory.assert_not_called() + self.redirected_client.make_request.assert_not_called() def test_upload_does_not_redirect_after_progress(self): redirect_error = self._create_redirect_error('eu-central-1') - initial_client = mock.Mock(awscrt.s3.S3Client) - redirected_client = mock.Mock(awscrt.s3.S3Client) def fail_after_progress(**kwargs): kwargs['on_progress'](1) kwargs['on_done'](error=redirect_error) return mock.Mock(awscrt.s3.S3Request) - initial_client.make_request.side_effect = fail_after_progress - client_factory = mock.Mock(return_value=redirected_client) - transfer_manager = self._create_redirect_transfer_manager( - initial_client, client_factory + transfer_manager = self._create_redirecting_transfer_manager( + fail_after_progress + ) + + with self.assertRaises(ClientError): + self._upload_and_wait(transfer_manager) + self.redirected_client_factory.assert_not_called() + self.redirected_client.make_request.assert_not_called() + + def test_concurrent_redirects_discover_region_once(self): + # Transfers redirected at the same time share one region lookup, and + # each request is sent on a client for the region it was signed for. + redirected_region = 'eu-central-1' + redirect_error = self._create_redirect_error(redirected_region) + release = threading.Event() + + def fail_when_released(**kwargs): + def complete_request(): + release.wait(RESULT_TIMEOUT) + kwargs['on_done'](error=redirect_error) + + thread = threading.Thread(target=complete_request) + self.addCleanup(thread.join) + thread.start() + return mock.Mock(awscrt.s3.S3Request) + + transfer_manager = self._create_redirecting_transfer_manager( + fail_when_released + ) + + with mock.patch.object( + self.request_serializer, + 'get_bucket_region', + wraps=self.request_serializer.get_bucket_region, + ) as discover_region: + # Both transfers are in flight before either has a region to + # reuse, then both fail with a redirect at once. + futures = [ + transfer_manager.upload( + self.filename, self.bucket, f'{self.key}-{i}', {}, [] + ) + for i in range(2) + ] + release.set() + for future in futures: + future.result(timeout=RESULT_TIMEOUT) + + # The region is discovered once and reused, rather than every + # redirected transfer paying for its own lookup. + self.assertEqual(discover_region.call_count, 1) + self.assertEqual(self.redirected_client.make_request.call_count, 2) + # Sending a request signed for one region on a client configured for + # another fails with SignatureDoesNotMatch, so every request has to + # agree with the client it was sent on. + for call in self.initial_client.make_request.call_args_list: + self.assertEqual( + call.kwargs['request'].headers.get('host'), + self.expected_host, + ) + for call in self.redirected_client.make_request.call_args_list: + self.assertEqual( + call.kwargs['request'].headers.get('host'), + f's3.{redirected_region}.amazonaws.com', + ) + + def test_upload_redirect_does_not_block_completion_thread(self): + # Redirecting must not run on the CRT thread that reported the + # failure, since it can block on the network. + redirect_threads = [] + + def succeed_and_record_thread(**kwargs): + redirect_threads.append(threading.get_ident()) + return self._succeed_make_request(**kwargs) + + transfer_manager = self._create_redirecting_transfer_manager( + self._fail_make_request_on_other_thread( + self._create_redirect_error('eu-central-1') + ), + succeed_and_record_thread, + ) + + self._upload_and_wait(transfer_manager) + + # Discovering the region and serializing the retry can both block on + # the network, so they must not run on the thread the CRT completed + # the original request on. + self.assertEqual(len(redirect_threads), 1) + self.assertEqual(len(self.completion_threads), 1) + self.assertNotEqual(redirect_threads[0], self.completion_threads[0]) + + def test_cancel_cancels_retry_started_before_original_request_returned( + self, + ): + # A redirect can start before the original request registers, so a + # cancel has to reach the retry rather than the finished request. + redirect_error = self._create_redirect_error('eu-central-1') + original_request = mock.Mock(awscrt.s3.S3Request) + retry_request = mock.Mock(awscrt.s3.S3Request) + retry_started = threading.Event() + retry_callbacks = {} + + def start_retry(**kwargs): + # Leave the retry in flight so it is the request a cancel has to + # reach. + retry_callbacks['on_done'] = kwargs['on_done'] + retry_started.set() + return retry_request + + def redirect_before_returning(**kwargs): + kwargs['on_done'](error=redirect_error) + # The redirect is handled on another thread, so wait for the retry + # to register before this request reports its own native request. + self.assertTrue(retry_started.wait(RESULT_TIMEOUT)) + return original_request + + transfer_manager = self._create_redirecting_transfer_manager( + redirect_before_returning, start_retry ) future = transfer_manager.upload( self.filename, self.bucket, self.key, {}, [] ) + future.cancel() - with self.assertRaises(ClientError): - future.result() - client_factory.assert_not_called() - redirected_client.make_request.assert_not_called() + # The original request completed and was replaced by the retry, so + # cancelling must not target the request that already finished. + retry_request.cancel.assert_called_once_with() + original_request.cancel.assert_not_called() + + retry_callbacks['on_done'](error=None) + future.result(timeout=RESULT_TIMEOUT) + + def test_upload_completes_when_redirect_decision_raises(self): + # A redirect decision that raises must still finish the transfer. + # The CRT invokes on_done from one of its own threads, so raising + # there strands the transfer instead of failing make_request(). + transfer_manager = self._create_redirecting_transfer_manager( + self._fail_make_request_on_other_thread( + self._create_redirect_error('eu-central-1') + ) + ) + with mock.patch.object( + transfer_manager._region_redirect_policy, + 'get_retry_region', + side_effect=RuntimeError('Unexpected redirect failure'), + ): + # The transfer must still finish, and surface the error from the + # transfer itself instead of the one from the redirect decision. + with self.assertRaises(ClientError): + self._upload_and_wait(transfer_manager) + transfer_manager.shutdown() + + self.redirected_client_factory.assert_not_called() def test_upload_from_seekable_stream(self): with open(self.filename, 'rb') as f: diff --git a/tests/unit/customizations/s3/test_factory.py b/tests/unit/customizations/s3/test_factory.py index b62abd1a912a..23c66580934d 100644 --- a/tests/unit/customizations/s3/test_factory.py +++ b/tests/unit/customizations/s3/test_factory.py @@ -16,6 +16,7 @@ from awscrt.s3 import S3FileIoOptions, S3RequestTlsMode from botocore.config import Config from botocore.credentials import Credentials +from botocore.exceptions import InvalidConfigError from botocore.httpsession import DEFAULT_CA_BUNDLE from botocore.session import Session from s3transfer.crt import CRTTransferManager @@ -205,8 +206,6 @@ def assert_is_classic_manager(self, manager): def assert_is_crt_manager(self, manager): self.assertIsInstance(manager, CRTTransferManager) - # The CRT client is initialized lazily once the region is known - manager.get_crt_client() def assert_expected_throughput_target_gbps( self, mock_crt_client, expected_throughput_target_gbps @@ -295,8 +294,9 @@ def test_creates_crt_client_for_redirected_region(self, mock_crt_client): self.params, self.runtime_config ) - # Client creation is lazy and each selected region is cached. - self.assertEqual(mock_crt_client.call_count, 0) + # The client for the configured region is created up front, and each + # selected region is cached. + self.assertEqual(mock_crt_client.call_count, 1) self.assertIs( transfer_manager.get_crt_client(), transfer_manager.get_crt_client(), @@ -424,6 +424,21 @@ def test_disable_botocore_credentials_for_crt_manager( self.session.get_credentials.assert_not_called() self.assertIsNone(mock_crt_client.call_args[1]['credential_provider']) + @mock.patch('s3transfer.crt.S3Client') + def test_invalid_client_config_raises_when_creating_crt_manager( + self, mock_crt_client + ): + self.runtime_config = self.get_runtime_config( + preferred_transfer_client='crt' + ) + self.params['verify_ssl'] = ' ' + + # Reported once here rather than once per submitted transfer. + with self.assertRaises(InvalidConfigError): + self.factory.create_transfer_manager( + self.params, self.runtime_config + ) + @mock.patch('s3transfer.crt.S3Client') @mock.patch('s3transfer.crt.ClientTlsContext') def test_use_verify_ssl_parameter_for_crt_manager( diff --git a/tests/unit/s3transfer/test_crt.py b/tests/unit/s3transfer/test_crt.py index d08db2de5b6a..01362f0ed778 100644 --- a/tests/unit/s3transfer/test_crt.py +++ b/tests/unit/s3transfer/test_crt.py @@ -449,21 +449,21 @@ def test_set_s3_request(self): self.coordinator.set_s3_request(s3_request) assert self.coordinator.s3_request is s3_request - def test_original_request_cannot_replace_retry(self): + def test_original_request_cannot_replace_redirect(self): first_request = self.create_s3_request() second_request = self.create_s3_request() - # The retry started before the original request registered its native - # request, so the original request must not become active. - self.coordinator.set_s3_request(second_request, is_retry=True) + # The redirect started before the original request registered its + # native request, so the original request must not become active. + self.coordinator.set_s3_request(second_request, is_region_redirect=True) self.coordinator.set_s3_request(first_request) assert self.coordinator.s3_request is second_request - def test_cancel_cancels_retry_request(self): + def test_cancel_cancels_redirected_request(self): first_request = self.create_s3_request() second_request = self.create_s3_request() self.coordinator.set_s3_request(first_request) - self.coordinator.set_s3_request(second_request, is_retry=True) + self.coordinator.set_s3_request(second_request, is_region_redirect=True) self.coordinator.cancel() @@ -525,20 +525,24 @@ def setup_method(self): ) self.serializer.get_cached_bucket_region.return_value = None self.serializer.get_bucket_region.return_value = 'eu-central-1' + self.serializer.get_configured_region.return_value = 'us-west-2' self.policy = s3transfer.crt.CRTS3RegionRedirectPolicy(self.serializer) - def get_retry_region(self, **overrides): + def is_error_redirect_candidate(self, **overrides): kwargs = { 'bucket': self.bucket, - 'transfer_type': 'put_object', - 'error': self.error, - 'is_retry': False, + 'is_region_redirect': False, 'bytes_transferred': 0, 'cancelled': False, 'is_replayable': True, } kwargs.update(overrides) - return self.policy.get_retry_region(**kwargs) + return self.policy.is_error_redirect_candidate(**kwargs) + + def get_retry_region(self, request_region=None): + return self.policy.get_retry_region( + self.bucket, 'put_object', self.error, request_region + ) def test_returns_and_caches_discovered_region(self): assert self.get_retry_region() == 'eu-central-1' @@ -551,37 +555,66 @@ def test_returns_none_when_region_not_discovered(self): assert self.get_retry_region() is None self.serializer.cache_bucket_region.assert_not_called() - def test_returns_none_for_no_error(self): - assert self.get_retry_region(error=None) is None - self.serializer.get_bucket_region.assert_not_called() + def test_returns_none_when_discovery_raises(self): + self.serializer.get_bucket_region.side_effect = InvalidRegionError( + region_name='not a region!' + ) + assert self.get_retry_region() is None + self.serializer.cache_bucket_region.assert_not_called() - def test_returns_none_after_redirect(self): - assert self.get_retry_region(is_retry=True) is None - self.serializer.get_bucket_region.assert_not_called() + def test_returns_none_when_discovered_region_is_configured_region(self): + # Retrying in the region the request already used would fail the same + # way, and caching it would build a duplicate client for that region. + self.serializer.get_bucket_region.return_value = 'us-west-2' + assert self.get_retry_region() is None + self.serializer.cache_bucket_region.assert_not_called() - def test_returns_none_after_bytes_transferred(self): - assert self.get_retry_region(bytes_transferred=1) is None - self.serializer.get_bucket_region.assert_not_called() + def test_returns_none_when_discovered_region_is_request_region(self): + self.serializer.get_bucket_region.return_value = 'eu-west-1' + assert self.get_retry_region(request_region='eu-west-1') is None + self.serializer.cache_bucket_region.assert_not_called() - def test_returns_none_when_cancelled(self): - assert self.get_retry_region(cancelled=True) is None + def test_reuses_region_discovered_by_another_transfer(self): + # A transfer that failed in the configured region does not need to + # rediscover a region another transfer already cached. + self.serializer.get_cached_bucket_region.return_value = 'eu-west-1' + assert self.get_retry_region() == 'eu-west-1' self.serializer.get_bucket_region.assert_not_called() - def test_returns_none_when_stream_is_not_replayable(self): - assert self.get_retry_region(is_replayable=False) is None - self.serializer.get_bucket_region.assert_not_called() + def test_rediscovers_region_when_cached_region_failed(self): + # The failed request already used the cached region, so the cache is + # stale and retrying there again would just fail the same way. + self.serializer.get_cached_bucket_region.return_value = 'eu-west-1' + assert self.get_retry_region(request_region='eu-west-1') == ( + 'eu-central-1' + ) + self.serializer.get_bucket_region.assert_called_once_with( + self.bucket, 'put_object', self.error + ) - def test_returns_none_for_s3express_bucket(self): - region = self.get_retry_region(bucket='mybucket--usw2-az5--x-s3') - assert region is None - self.serializer.get_bucket_region.assert_not_called() + def test_is_candidate_for_failed_replayable_transfer(self): + assert self.is_error_redirect_candidate() - def test_returns_none_when_discovery_raises(self): - self.serializer.get_bucket_region.side_effect = InvalidRegionError( - region_name='not a region!' + def test_not_candidate_after_redirect(self): + assert not self.is_error_redirect_candidate(is_region_redirect=True) + + def test_not_candidate_after_bytes_transferred(self): + assert not self.is_error_redirect_candidate(bytes_transferred=1) + + def test_not_candidate_when_cancelled(self): + assert not self.is_error_redirect_candidate(cancelled=True) + + def test_not_candidate_when_stream_is_not_replayable(self): + assert not self.is_error_redirect_candidate(is_replayable=False) + + def test_not_candidate_for_s3express_bucket(self): + assert not self.is_error_redirect_candidate( + bucket='mybucket--usw2-az5--x-s3' ) - assert self.get_retry_region() is None - self.serializer.cache_bucket_region.assert_not_called() + + def test_candidate_checks_do_not_discover_region(self): + self.is_error_redirect_candidate() + self.serializer.get_bucket_region.assert_not_called() def test_get_cached_bucket_region(self): self.serializer.get_cached_bucket_region.return_value = 'eu-west-1'