From 4fcda8218d9c71fdc6fb804334f01500ffcf5c5e Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Thu, 16 Jul 2026 09:33:50 +0200 Subject: [PATCH 1/2] fix: retry transient source-read and segment-copy failures in UploadPartCopy A single truncated frame GET (backend drops a long-lived connection mid-body: ClientPayloadError 'received 8331264 of 8388636 bytes') or a 200-with-embedded-InternalError UploadPartCopy failed the whole client part. botocore's retry layer only covers a request up to the response headers, so body-read failures had no retry at all. During the 2026-07-16 Scylla backup this failed 12/13 hosts' upload jobs. - read_source_bytes(): full-body ranged GET with retry on transient transport/backend errors, used by frame reads and single-object reads - _stream_raw_source_with_resume(): raw copy-source streaming resumes with a ranged GET at the exact byte offset already yielded - passthrough copy_segment: retry upload_part_copy on retryable errors (same part number + range, safe to re-issue) - new env knobs: S3PROXY_SOURCE_READ_ATTEMPTS (4), S3PROXY_SOURCE_READ_BACKOFF (0.5s) --- s3proxy/handlers/base.py | 83 ++++++- s3proxy/handlers/multipart/copy.py | 109 +++++++-- tests/unit/test_source_read_retry.py | 338 +++++++++++++++++++++++++++ 3 files changed, 501 insertions(+), 29 deletions(-) create mode 100644 tests/unit/test_source_read_retry.py diff --git a/s3proxy/handlers/base.py b/s3proxy/handlers/base.py index 34740bb..b79dbd4 100644 --- a/s3proxy/handlers/base.py +++ b/s3proxy/handlers/base.py @@ -2,15 +2,25 @@ import asyncio import base64 +import os import re from collections.abc import AsyncIterator from datetime import datetime from typing import NoReturn from urllib.parse import parse_qs, unquote +import aiohttp import httpx import structlog -from botocore.exceptions import ClientError +from botocore.exceptions import ( + ClientError, + ConnectionClosedError, + ConnectTimeoutError, + EndpointConnectionError, + IncompleteReadError, + ReadTimeoutError, + ResponseStreamingError, +) from fastapi import Request, Response from structlog.stdlib import BoundLogger @@ -26,6 +36,67 @@ PATH_RE = re.compile(r"^/([^/]+)/(.+)$") BUCKET_RE = re.compile(r"^/([^/]+)/?$") # Handles /bucket and /bucket/ +# botocore's retry layer only covers a request up to the response headers; a +# body that dies mid-read (backend drops a long-lived connection, observed as +# ClientPayloadError at a fixed offset after minutes of sequential frame GETs) +# surfaces here with no retry and used to fail whole UploadPartCopy parts. +# Re-issuing the same ranged GET on a fresh connection is always safe. +SOURCE_READ_ATTEMPTS = int(os.environ.get("S3PROXY_SOURCE_READ_ATTEMPTS", "4")) +SOURCE_READ_BACKOFF_SEC = float(os.environ.get("S3PROXY_SOURCE_READ_BACKOFF", "0.5")) + +_RETRYABLE_S3_ERROR_CODES = frozenset( + {"InternalError", "SlowDown", "RequestTimeout", "ServiceUnavailable", "BadGateway", "500", "502", "503"} +) + +_RETRYABLE_TRANSPORT_ERRORS = ( + aiohttp.ClientPayloadError, + aiohttp.ClientConnectionError, + TimeoutError, + ConnectionClosedError, + ConnectTimeoutError, + EndpointConnectionError, + IncompleteReadError, + ReadTimeoutError, + ResponseStreamingError, +) + + +def is_retryable_source_error(exc: BaseException) -> bool: + """Transient transport/backend failures where re-issuing the request is safe.""" + if isinstance(exc, _RETRYABLE_TRANSPORT_ERRORS): + return True + if isinstance(exc, ClientError): + code = str(exc.response.get("Error", {}).get("Code", "")) + return code in _RETRYABLE_S3_ERROR_CODES + return False + + +async def read_source_bytes( + client: "S3Client", + bucket: str, + key: str, + range_header: str | None = None, +) -> bytes: + """GET an object (or range) and read the full body, retrying truncated reads.""" + for attempt in range(1, SOURCE_READ_ATTEMPTS + 1): + try: + resp = await client.get_object(bucket, key, range_header=range_header) + async with resp["Body"] as body: + return await body.read() + except Exception as exc: + if not is_retryable_source_error(exc) or attempt == SOURCE_READ_ATTEMPTS: + raise + logger.warning( + "SOURCE_READ_RETRY", + bucket=bucket, + key=key, + range=range_header, + attempt=attempt, + error=str(exc), + ) + await asyncio.sleep(SOURCE_READ_BACKOFF_SEC * (2 ** (attempt - 1))) + raise AssertionError("unreachable") + # Shared httpx client for connection reuse _http_client: httpx.AsyncClient | None = None _http_client_lock = asyncio.Lock() @@ -229,9 +300,7 @@ def _check_conditional_headers( async def _download_encrypted_single( self, client: S3Client, bucket: str, key: str, wrapped_dek_b64: str, kid: str = "" ) -> bytes: - resp = await client.get_object(bucket, key) - async with resp["Body"] as body: - ciphertext = await body.read() + ciphertext = await read_source_bytes(client, bucket, key) wrapped_dek = base64.b64decode(wrapped_dek_b64) return crypto.decrypt_object(ciphertext, wrapped_dek, self.keyring.key_by_id(kid)) @@ -283,9 +352,9 @@ async def _iter_multipart_plaintext( if in_range: ct_end = ct_offset + fsize - 1 - resp = await client.get_object(bucket, key, f"bytes={ct_offset}-{ct_end}") - async with resp["Body"] as body: - ciphertext = await body.read() + ciphertext = await read_source_bytes( + client, bucket, key, f"bytes={ct_offset}-{ct_end}" + ) plaintext = crypto.decrypt(ciphertext, dek) if range_start is not None: diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index c171e69..a53ade9 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -31,7 +31,8 @@ persist_upload_state, ) from ...utils import format_iso8601 -from ..base import BaseHandler +from .. import base +from ..base import BaseHandler, is_retryable_source_error, read_source_bytes from .upload_part import _PlaintextReader logger: BoundLogger = structlog.get_logger(__name__) @@ -876,14 +877,36 @@ async def copy_segment(idx: int, seg: _CiphertextSegment) -> InternalPartMetadat copy_range = f"bytes={seg.ct_offset}-{ct_end}" part_reserve = crypto.copy_passthrough_segment_peak(seg.plaintext_size) async with segment_sem, concurrency.reserve_copy_memory(part_reserve): - resp = await client.upload_part_copy( - bucket, - key, - upload_id, - internal_num, - copy_source_path, - copy_source_range=copy_range, - ) + # The backend can accept the copy, stream a 200, and still fail + # ("InternalError: The server did not respond in time" embedded + # in the body). Same part number + range makes a retry safe. + for attempt in range(1, base.SOURCE_READ_ATTEMPTS + 1): + try: + resp = await client.upload_part_copy( + bucket, + key, + upload_id, + internal_num, + copy_source_path, + copy_source_range=copy_range, + ) + break + except Exception as exc: + if ( + not is_retryable_source_error(exc) + or attempt == base.SOURCE_READ_ATTEMPTS + ): + raise + logger.warning( + "UPLOAD_PART_COPY_SEGMENT_RETRY", + bucket=bucket, + key=key, + part_num=part_num, + internal_part=internal_num, + attempt=attempt, + error=str(exc), + ) + await asyncio.sleep(base.SOURCE_READ_BACKOFF_SEC * (2 ** (attempt - 1))) return InternalPartMetadata( internal_part_number=internal_num, plaintext_size=seg.plaintext_size, @@ -1103,9 +1126,7 @@ async def _fetch_copy_plaintext( ) -> bytes: """Download the full plaintext of a copy source (small sources only).""" if not src_wrapped_dek and not src_multipart_meta: - resp = await client.get_object(src_bucket, src_key, range_header=copy_source_range) - async with resp["Body"] as body: - return await body.read() + return await read_source_bytes(client, src_bucket, src_key, copy_source_range) elif src_multipart_meta: range_start, range_end = self._parse_copy_source_range( copy_source_range, src_multipart_meta.total_plaintext_size @@ -1308,16 +1329,60 @@ async def _iter_copy_source( plaintext = plaintext[start : end + 1] yield plaintext else: - resp = await client.get_object(src_bucket, src_key, range_header=copy_source_range) - async with resp["Body"] as body: - # resp["Body"] enters as an aiohttp ClientResponse, whose read() - # takes no size arg; stream via its StreamReader in bounded chunks - # (body.read(n) raised TypeError and 500'd every passthrough copy). - while True: - chunk = await body.content.read(crypto.MAX_BUFFER_SIZE) - if not chunk: - break - yield chunk + async for chunk in self._stream_raw_source_with_resume( + client, src_bucket, src_key, copy_source_range + ): + yield chunk + + async def _stream_raw_source_with_resume( + self, + client: S3Client, + src_bucket: str, + src_key: str, + copy_source_range: str | None, + ) -> AsyncIterator[bytes]: + """Stream an unencrypted copy source, resuming with a ranged GET if the + body dies mid-read (backend dropping a long-lived connection). Bytes + already yielded are never re-fetched: the resume range starts at the + exact offset the previous attempt reached. + """ + if copy_source_range: + range_start, range_end = self._parse_raw_copy_source_range(copy_source_range) + else: + range_start, range_end = 0, None + sent = 0 + attempt = 1 + while True: + if sent == 0 and copy_source_range is None: + range_header = None + else: + end_part = "" if range_end is None else str(range_end) + range_header = f"bytes={range_start + sent}-{end_part}" + try: + resp = await client.get_object(src_bucket, src_key, range_header=range_header) + async with resp["Body"] as body: + # resp["Body"] enters as an aiohttp ClientResponse, whose read() + # takes no size arg; stream via its StreamReader in bounded chunks + # (body.read(n) raised TypeError and 500'd every passthrough copy). + while True: + chunk = await body.content.read(crypto.MAX_BUFFER_SIZE) + if not chunk: + return + sent += len(chunk) + yield chunk + except Exception as exc: + if not is_retryable_source_error(exc) or attempt >= base.SOURCE_READ_ATTEMPTS: + raise + logger.warning( + "COPY_SOURCE_STREAM_RESUME", + bucket=src_bucket, + key=src_key, + resumed_at=range_start + sent, + attempt=attempt, + error=str(exc), + ) + await asyncio.sleep(base.SOURCE_READ_BACKOFF_SEC * (2 ** (attempt - 1))) + attempt += 1 async def _pump_copy_chunks( self, diff --git a/tests/unit/test_source_read_retry.py b/tests/unit/test_source_read_retry.py new file mode 100644 index 0000000..edbded7 --- /dev/null +++ b/tests/unit/test_source_read_retry.py @@ -0,0 +1,338 @@ +"""Transient source-read failures must be retried, not fail the whole copy. + +Prod incident 2026-07-16: the backend (Hetzner) drops long-lived connections +mid-body, so a single frame GET died with ClientPayloadError ("received +8331264 of 8388636 bytes") after botocore's retry window (which only covers +the request up to response headers) had passed. One truncated read failed the +entire UploadPartCopy part, and through rclone's job retries, whole Scylla +backup runs. The same backend also fails UploadPartCopy after streaming a 200 +("InternalError: The server did not respond in time" embedded in the body). +""" + +import hashlib + +import aiohttp +import pytest +from botocore.exceptions import ClientError, ReadTimeoutError + +from s3proxy import crypto +from s3proxy.handlers import base as base_handler +from s3proxy.handlers.base import is_retryable_source_error, read_source_bytes +from s3proxy.handlers.multipart import MultipartHandlerMixin +from s3proxy.state import ( + InternalPartMetadata, + MultipartMetadata, + PartMetadata, + save_multipart_metadata, +) +from tests.unit.test_upload_part_copy_passthrough import ( + BUCKET, + _copy_handler, + _copy_part_request, + _patch_client, + _read, +) + + +@pytest.fixture(autouse=True) +def _no_backoff(monkeypatch): + monkeypatch.setattr(base_handler, "SOURCE_READ_BACKOFF_SEC", 0.0) + + +def _payload_error(): + return aiohttp.ClientPayloadError( + "Response payload is not completed: " + ) + + +def _client_error(code, operation="GetObject"): + return ClientError({"Error": {"Code": code, "Message": code}}, operation) + + +class _Body: + def __init__(self, data: bytes): + self._data = data + + async def read(self): + return self._data + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return None + + +class _TruncatingBody: + """read() dies mid-body, the exact prod failure shape.""" + + async def read(self): + raise _payload_error() + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return None + + +class _FlakyGetClient: + def __init__(self, data: bytes, body_failures: int = 0, call_exc: Exception | None = None): + self.data = data + self.body_failures = body_failures + self.call_exc = call_exc + self.calls: list[str | None] = [] + + async def get_object(self, bucket, key, range_header=None): + self.calls.append(range_header) + if self.call_exc is not None: + raise self.call_exc + if len(self.calls) <= self.body_failures: + return {"Body": _TruncatingBody()} + return {"Body": _Body(self.data)} + + +# --- read_source_bytes ------------------------------------------------------ + + +@pytest.mark.asyncio +async def test_read_source_bytes_retries_truncated_body(): + client = _FlakyGetClient(b"frame-bytes", body_failures=2) + assert await read_source_bytes(client, "b", "k", "bytes=0-10") == b"frame-bytes" + assert client.calls == ["bytes=0-10"] * 3 + + +@pytest.mark.asyncio +async def test_read_source_bytes_gives_up_after_max_attempts(): + client = _FlakyGetClient(b"", body_failures=10_000) + with pytest.raises(aiohttp.ClientPayloadError): + await read_source_bytes(client, "b", "k") + assert len(client.calls) == base_handler.SOURCE_READ_ATTEMPTS + + +@pytest.mark.asyncio +async def test_read_source_bytes_does_not_retry_non_retryable(): + client = _FlakyGetClient(b"", call_exc=_client_error("NoSuchKey")) + with pytest.raises(ClientError): + await read_source_bytes(client, "b", "k") + assert len(client.calls) == 1 + + +# --- error classification --------------------------------------------------- + + +def test_retryable_error_classification(): + assert is_retryable_source_error(_payload_error()) + assert is_retryable_source_error(ReadTimeoutError(endpoint_url="http://s3")) + assert is_retryable_source_error(TimeoutError()) + assert is_retryable_source_error(_client_error("InternalError")) + assert is_retryable_source_error(_client_error("SlowDown")) + assert not is_retryable_source_error(_client_error("NoSuchKey")) + assert not is_retryable_source_error(_client_error("AccessDenied")) + assert not is_retryable_source_error(ValueError("boom")) + + +# --- raw copy-source stream resume ------------------------------------------ + + +class _Stream: + def __init__(self, data: bytes, die_after: int | None = None): + self._buf = bytearray(data) + self._die_after = die_after + self._served = 0 + + @property + def content(self): + return self + + async def read(self, n: int = -1): + if self._die_after is not None and self._served >= self._die_after: + raise _payload_error() + limit = len(self._buf) if n == -1 else n + if self._die_after is not None: + limit = min(limit, self._die_after - self._served) + chunk = bytes(self._buf[:limit]) + del self._buf[:limit] + self._served += len(chunk) + if not chunk and self._die_after is not None: + raise _payload_error() + return chunk + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return None + + +class _StreamClient: + """Serves ranged GETs over `data`; the first GET truncates after `die_after` bytes.""" + + def __init__(self, data: bytes, die_after: int | None = None): + self.data = data + self.die_after = die_after + self.calls: list[str | None] = [] + + async def get_object(self, bucket, key, range_header=None): + self.calls.append(range_header) + start, end = 0, len(self.data) - 1 + if range_header: + spec = range_header.replace("bytes=", "") + s, e = spec.split("-") + start = int(s) + end = int(e) if e else len(self.data) - 1 + body = self.data[start : end + 1] + die_after = self.die_after if len(self.calls) == 1 else None + return {"Body": _Stream(body, die_after)} + + +def _mixin(): + return MultipartHandlerMixin.__new__(MultipartHandlerMixin) + + +@pytest.mark.asyncio +async def test_stream_resume_whole_object_resumes_at_offset(): + data = bytes(range(256)) * 4 + client = _StreamClient(data, die_after=100) + chunks = [ + c + async for c in _mixin()._stream_raw_source_with_resume(client, "b", "k", None) + ] + assert b"".join(chunks) == data + assert client.calls == [None, "bytes=100-"] + + +@pytest.mark.asyncio +async def test_stream_resume_preserves_original_range(): + data = bytes(range(256)) + client = _StreamClient(data, die_after=15) + chunks = [ + c + async for c in _mixin()._stream_raw_source_with_resume(client, "b", "k", "bytes=10-59") + ] + assert b"".join(chunks) == data[10:60] + assert client.calls == ["bytes=10-59", "bytes=25-59"] + + +@pytest.mark.asyncio +async def test_stream_resume_does_not_mask_non_retryable_errors(): + class _Boom(_StreamClient): + async def get_object(self, bucket, key, range_header=None): + self.calls.append(range_header) + raise _client_error("AccessDenied") + + client = _Boom(b"") + with pytest.raises(ClientError): + async for _ in _mixin()._stream_raw_source_with_resume(client, "b", "k", None): + pass + assert len(client.calls) == 1 + + +# --- end-to-end: passthrough copy survives transient backend failures -------- + + +class _FlakyS3: + """Wraps MockS3Client: first upload_part_copy 500s, first ranged GET truncates.""" + + def __init__(self, inner, upc_failures=1, get_failures=1): + self._inner = inner + self._upc_failures = upc_failures + self._get_failures = get_failures + self.upc_attempts = 0 + + def __getattr__(self, name): + return getattr(self._inner, name) + + async def upload_part_copy(self, *args, **kwargs): + self.upc_attempts += 1 + if self._upc_failures > 0: + self._upc_failures -= 1 + raise ClientError( + {"Error": {"Code": "InternalError", "Message": "The server did not respond in time."}}, + "UploadPartCopy", + ) + return await self._inner.upload_part_copy(*args, **kwargs) + + async def get_object(self, bucket, key, range_header=None): + if range_header is not None and self._get_failures > 0: + self._get_failures -= 1 + raise _payload_error() + return await self._inner.get_object(bucket, key, range_header=range_header) + + +@pytest.mark.asyncio +async def test_passthrough_copy_survives_transient_backend_failures( + mock_s3, settings, manager, credentials, monkeypatch +): + """One 500'd segment copy + one truncated frame GET must not fail the part.""" + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) + handler = _copy_handler(settings, manager) + flaky = _FlakyS3(mock_s3, upc_failures=1, get_failures=1) + _patch_client(handler, flaky) + await mock_s3.create_bucket(BUCKET) + + kid, kek = settings.keyring.key_for(credentials.access_key) + src_dek = crypto.generate_dek() + src_plaintext = b"S" * (crypto.MAX_BUFFER_SIZE * 4) + + ciphertext_blob = bytearray() + internal_parts_meta = [] + for i, start in enumerate(range(0, len(src_plaintext), crypto.MAX_BUFFER_SIZE), 1): + chunk = src_plaintext[start : start + crypto.MAX_BUFFER_SIZE] + ct = crypto.encrypt_frame(chunk, src_dek, "src-upload", i, 0) + internal_parts_meta.append( + InternalPartMetadata( + internal_part_number=i, + plaintext_size=len(chunk), + ciphertext_size=len(ct), + etag=hashlib.md5(ct, usedforsecurity=False).hexdigest(), + ) + ) + ciphertext_blob.extend(ct) + + src_meta = MultipartMetadata( + version=2, + part_count=1, + total_plaintext_size=len(src_plaintext), + parts=[ + PartMetadata( + part_number=1, + plaintext_size=len(src_plaintext), + ciphertext_size=len(ciphertext_blob), + etag="ignored", + md5=hashlib.md5(src_plaintext, usedforsecurity=False).hexdigest(), + internal_parts=internal_parts_meta, + ) + ], + wrapped_dek=crypto.wrap_key(src_dek, kek), + kid=kid, + ) + await mock_s3.put_object(BUCKET, "sst/big.db", bytes(ciphertext_blob)) + await save_multipart_metadata(mock_s3, BUCKET, "sst/big.db", src_meta) + + resp_create = await mock_s3.create_multipart_upload(BUCKET, "sst/big.db.snap") + upload_id = resp_create["UploadId"] + await manager.create_upload(BUCKET, "sst/big.db.snap", upload_id, crypto.generate_dek(), kid) + + resp = await handler.handle_upload_part_copy( + _copy_part_request( + f"/{BUCKET}/sst/big.db.snap", + f"/{BUCKET}/sst/big.db", + upload_id, + ), + credentials, + ) + body = await _read(resp) + assert b"" not in body + assert flaky.upc_attempts >= 2 # first attempt failed, retry landed + + updated = await manager.get_upload(BUCKET, "sst/big.db.snap", upload_id) + part = updated.parts[1] + assert part.plaintext_size == len(src_plaintext) + + recovered = bytearray() + for ip in sorted(part.internal_parts, key=lambda x: x.internal_part_number): + ct = mock_s3.multipart_uploads[upload_id]["Parts"][ip.internal_part_number]["Body"] + recovered.extend(crypto.decrypt_framed(ct, src_dek, ip.plaintext_size)) + assert bytes(recovered) == src_plaintext From 4d0e6cef4cfacb230f408afac6972a47c2cdb426 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Thu, 16 Jul 2026 09:35:08 +0200 Subject: [PATCH 2/2] style: satisfy ruff line length and formatting --- s3proxy/handlers/base.py | 6 ++++-- tests/unit/test_source_read_retry.py | 15 ++++++++------- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/s3proxy/handlers/base.py b/s3proxy/handlers/base.py index b79dbd4..43277f4 100644 --- a/s3proxy/handlers/base.py +++ b/s3proxy/handlers/base.py @@ -45,7 +45,8 @@ SOURCE_READ_BACKOFF_SEC = float(os.environ.get("S3PROXY_SOURCE_READ_BACKOFF", "0.5")) _RETRYABLE_S3_ERROR_CODES = frozenset( - {"InternalError", "SlowDown", "RequestTimeout", "ServiceUnavailable", "BadGateway", "500", "502", "503"} + {"InternalError", "SlowDown", "RequestTimeout", "ServiceUnavailable", "BadGateway"} + | {"500", "502", "503"} ) _RETRYABLE_TRANSPORT_ERRORS = ( @@ -72,7 +73,7 @@ def is_retryable_source_error(exc: BaseException) -> bool: async def read_source_bytes( - client: "S3Client", + client: S3Client, bucket: str, key: str, range_header: str | None = None, @@ -97,6 +98,7 @@ async def read_source_bytes( await asyncio.sleep(SOURCE_READ_BACKOFF_SEC * (2 ** (attempt - 1))) raise AssertionError("unreachable") + # Shared httpx client for connection reuse _http_client: httpx.AsyncClient | None = None _http_client_lock = asyncio.Lock() diff --git a/tests/unit/test_source_read_retry.py b/tests/unit/test_source_read_retry.py index edbded7..5c1fe6d 100644 --- a/tests/unit/test_source_read_retry.py +++ b/tests/unit/test_source_read_retry.py @@ -195,10 +195,7 @@ def _mixin(): async def test_stream_resume_whole_object_resumes_at_offset(): data = bytes(range(256)) * 4 client = _StreamClient(data, die_after=100) - chunks = [ - c - async for c in _mixin()._stream_raw_source_with_resume(client, "b", "k", None) - ] + chunks = [c async for c in _mixin()._stream_raw_source_with_resume(client, "b", "k", None)] assert b"".join(chunks) == data assert client.calls == [None, "bytes=100-"] @@ -208,8 +205,7 @@ async def test_stream_resume_preserves_original_range(): data = bytes(range(256)) client = _StreamClient(data, die_after=15) chunks = [ - c - async for c in _mixin()._stream_raw_source_with_resume(client, "b", "k", "bytes=10-59") + c async for c in _mixin()._stream_raw_source_with_resume(client, "b", "k", "bytes=10-59") ] assert b"".join(chunks) == data[10:60] assert client.calls == ["bytes=10-59", "bytes=25-59"] @@ -249,7 +245,12 @@ async def upload_part_copy(self, *args, **kwargs): if self._upc_failures > 0: self._upc_failures -= 1 raise ClientError( - {"Error": {"Code": "InternalError", "Message": "The server did not respond in time."}}, + { + "Error": { + "Code": "InternalError", + "Message": "The server did not respond in time.", + } + }, "UploadPartCopy", ) return await self._inner.upload_part_copy(*args, **kwargs)