diff --git a/s3proxy/handlers/base.py b/s3proxy/handlers/base.py index fea2e89..7a61ac2 100644 --- a/s3proxy/handlers/base.py +++ b/s3proxy/handlers/base.py @@ -84,8 +84,19 @@ def is_retryable_source_error(exc: BaseException) -> bool: 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 + err = exc.response.get("Error", {}) + code = str(err.get("Code", "")) + if code in _RETRYABLE_S3_ERROR_CODES: + return True + # Hetzner sheds load under concurrency by returning a bare 400 with an + # empty body, which botocore surfaces as InvalidArgument with no Message. + # A real InvalidArgument always names the offending argument, so the + # message-less form is congestion, not a malformed request: 4/4 observed + # on UploadPart had Message=None, on legal 50MiB non-final parts, while + # the same parts succeeded on a quiet retry (24-part 16-way probe + # reproduced it as GatewayTimeout, p50 13.09s vs 1.79s unloaded). + if code in ("InvalidArgument", "BadRequest", "400") and not err.get("Message"): + return True return False diff --git a/s3proxy/handlers/multipart/upload_part.py b/s3proxy/handlers/multipart/upload_part.py index 02e8561..ae06e2e 100644 --- a/s3proxy/handlers/multipart/upload_part.py +++ b/s3proxy/handlers/multipart/upload_part.py @@ -257,13 +257,55 @@ async def _upload_part_with_retry( the same part number with the same ciphertext is idempotent, so retry here where the parsed error is actually visible. """ + started = time.monotonic() for attempt in range(1, base.SOURCE_READ_ATTEMPTS + 1): try: - return await client.upload_part( + result = await client.upload_part( bucket, key, upload_id, internal_part_num, ciphertext ) + if attempt > 1: + logger.warning( + "UPLOAD_PART_RECOVERED", + bucket=bucket, + key=key, + client_part=client_part_num, + internal_part=internal_part_num, + attempts=attempt, + elapsed_sec=round(time.monotonic() - started, 2), + ) + return result except Exception as exc: - if not is_retryable_source_error(exc) or attempt == base.SOURCE_READ_ATTEMPTS: + retryable = is_retryable_source_error(exc) + if not retryable or attempt == base.SOURCE_READ_ATTEMPTS: + # Distinguishes "gave up after N transient failures" from + # "rejected outright" -- the two are indistinguishable in the + # UPLOAD_PART_CLIENT_ERROR that follows, which cost hours of + # misdiagnosis chasing a retry gap that was really exhaustion. + logger.error( + "UPLOAD_PART_GAVE_UP", + bucket=bucket, + key=key, + client_part=client_part_num, + internal_part=internal_part_num, + part_size=len(ciphertext), + attempts=attempt, + max_attempts=base.SOURCE_READ_ATTEMPTS, + classified_retryable=retryable, + exhausted=retryable, + elapsed_sec=round(time.monotonic() - started, 2), + aws_error_code=( + exc.response.get("Error", {}).get("Code", "") + if isinstance(exc, ClientError) + else None + ), + aws_error_message=( + exc.response.get("Error", {}).get("Message") + if isinstance(exc, ClientError) + else None + ), + error_type=type(exc).__name__, + error=str(exc), + ) raise logger.warning( "UPLOAD_PART_RETRY", @@ -271,7 +313,15 @@ async def _upload_part_with_retry( key=key, client_part=client_part_num, internal_part=internal_part_num, + part_size=len(ciphertext), attempt=attempt, + max_attempts=base.SOURCE_READ_ATTEMPTS, + aws_error_code=( + exc.response.get("Error", {}).get("Code", "") + if isinstance(exc, ClientError) + else None + ), + error_type=type(exc).__name__, error=str(exc), ) await asyncio.sleep(base.SOURCE_READ_BACKOFF_SEC * (2 ** (attempt - 1))) diff --git a/tests/unit/test_source_read_resume.py b/tests/unit/test_source_read_resume.py index f650549..570596c 100644 --- a/tests/unit/test_source_read_resume.py +++ b/tests/unit/test_source_read_resume.py @@ -226,3 +226,51 @@ async def get_object(self, bucket, key, range_header=None): c = _Flaky() assert await read_source_bytes(c, "b", "k", "bytes=0-131071") == BODY assert c.calls == 2 + + +# --- message-less 400 retryability ----------------------------------------- +# +# Prod 2026-07-30: 4/4 UploadPart failures on main.companies arrived as +# InvalidArgument with Message=None, on legal 50MiB non-final parts (part 2-16, +# never part 1). A real InvalidArgument always names the offending argument, so +# the message-less form is Hetzner shedding load, not a malformed request. +# +# Reproduced: 24 x 50MiB parts at 16-way concurrency yielded 1 failure, surfaced +# as GatewayTimeout("The server did not respond in time"), with latency p50 +# 13.09s / max 52.32s -- against p50 1.79s for the same parts unloaded. Same +# congestion event, two different bodies depending on whether RGW managed to +# write one. + + +def _upload_part_error(code, message=None): + err = {"Code": code} + if message is not None: + err["Message"] = message + return ClientError({"Error": err}, "UploadPart") + + +@pytest.mark.parametrize("code", ["InvalidArgument", "BadRequest", "400"]) +def test_message_less_400_is_retryable(code): + assert base.is_retryable_source_error(_upload_part_error(code)) is True + assert base.is_retryable_source_error(_upload_part_error(code, "")) is True + + +def test_invalid_argument_with_reason_still_fails(): + """A genuine validation error must not be retried forever.""" + exc = _upload_part_error( + "InvalidArgument", "Part number must be an integer between 1 and 10000" + ) + assert base.is_retryable_source_error(exc) is False + + +@pytest.mark.parametrize( + "code,msg", + [ + ("AccessDenied", "Access Denied"), + ("NoSuchKey", "The specified key does not exist."), + ("EntityTooSmall", "Your proposed upload is smaller than the minimum allowed size"), + ("InvalidPart", "One or more of the specified parts could not be found"), + ], +) +def test_permanent_errors_with_messages_not_retried(code, msg): + assert base.is_retryable_source_error(_upload_part_error(code, msg)) is False diff --git a/tests/unit/test_upload_part_retry.py b/tests/unit/test_upload_part_retry.py index dbe4bb8..9772bc1 100644 --- a/tests/unit/test_upload_part_retry.py +++ b/tests/unit/test_upload_part_retry.py @@ -25,6 +25,7 @@ from botocore.exceptions import ClientError from s3proxy.handlers import base +from s3proxy.handlers.multipart import upload_part as upload_part_mod from s3proxy.handlers.multipart.upload_part import UploadPartMixin @@ -120,3 +121,113 @@ async def test_retryable_codes(code): client = _FlakyUploadClient(fail_times=1, error_code=code) await _upload(client) assert client.attempts == 2 + + +# --- message-less 400 + observability --------------------------------------- +# +# Prod 2026-07-30: the same `main.companies` upload failed again, this time as +# `InvalidArgument` with **Message=None** (4/4 occurrences) on legal 50MiB +# non-final parts (parts 2-16, never part 1). `GatewayTimeout` was already +# retryable from #133, but the message-less variant was not, so it fell through +# to raise on first sight. +# +# Both are the same congestion event: a 24 x 50MiB probe at 16-way concurrency +# reproduced 1 failure surfaced as GatewayTimeout("The server did not respond in +# time"), with latency p50 13.09s / max 52.32s against p50 1.79s unloaded -- +# Hetzner sheds load and only sometimes manages to write an error body. + + +class _MessagelessClient: + """Raises InvalidArgument with no Message, the prod 2026-07-30 shape.""" + + def __init__(self, fail_times: int = 1): + self._fail_times = fail_times + self.attempts = 0 + + async def upload_part(self, bucket, key, upload_id, part_number, body): + self.attempts += 1 + if self._fail_times > 0: + self._fail_times -= 1 + raise ClientError({"Error": {"Code": "InvalidArgument"}}, "UploadPart") + return {"ETag": f'"etag-{part_number}"'} + + +@pytest.mark.asyncio +async def test_retries_messageless_invalid_argument(): + client = _MessagelessClient(fail_times=1) + resp = await _upload(client) + assert resp["ETag"] == '"etag-7"' + assert client.attempts == 2 + + +@pytest.mark.asyncio +async def test_real_invalid_argument_is_not_retried(): + """A validation error that names its argument must fail immediately.""" + + class _Rejecting: + def __init__(self): + self.attempts = 0 + + async def upload_part(self, *a, **k): + self.attempts += 1 + raise _client_error( + "InvalidArgument", "Part number must be an integer between 1 and 10000" + ) + + client = _Rejecting() + with pytest.raises(ClientError): + await _upload(client) + assert client.attempts == 1 + + +@pytest.mark.asyncio +async def test_recovered_upload_is_logged(monkeypatch): + """A part that only lands after a retry must leave a trace.""" + events = [] + monkeypatch.setattr( + upload_part_mod.logger, + "warning", + lambda ev, **kw: events.append((ev, kw)), + ) + client = _FlakyUploadClient(fail_times=2) + await _upload(client) + names = [e for e, _ in events] + assert "UPLOAD_PART_RECOVERED" in names + recovered = next(kw for e, kw in events if e == "UPLOAD_PART_RECOVERED") + assert recovered["attempts"] == 3 + + +@pytest.mark.asyncio +async def test_exhaustion_distinguishable_from_rejection(monkeypatch): + """Exhausted transient retries log exhausted=True; outright rejection False.""" + events = [] + monkeypatch.setattr(upload_part_mod.logger, "error", lambda ev, **kw: events.append((ev, kw))) + monkeypatch.setattr(upload_part_mod.logger, "warning", lambda ev, **kw: None) + + client = _FlakyUploadClient(fail_times=99, error_code="InternalError") + with pytest.raises(ClientError): + await _upload(client) + ev, kw = next((e, k) for e, k in events if e == "UPLOAD_PART_GAVE_UP") + assert kw["exhausted"] is True + assert kw["classified_retryable"] is True + assert kw["attempts"] == base.SOURCE_READ_ATTEMPTS + assert client.attempts == base.SOURCE_READ_ATTEMPTS + + events.clear() + + class _Denied: + def __init__(self): + self.attempts = 0 + + async def upload_part(self, *a, **k): + self.attempts += 1 + raise _client_error("AccessDenied", "Access Denied") + + denied = _Denied() + with pytest.raises(ClientError): + await _upload(denied) + ev, kw = next((e, k) for e, k in events if e == "UPLOAD_PART_GAVE_UP") + assert kw["exhausted"] is False + assert kw["classified_retryable"] is False + assert kw["aws_error_code"] == "AccessDenied" + assert denied.attempts == 1