fix: retry message-less 400 on UploadPart

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
 
 

s3proxy/handlers/multipart/upload_part.py

@@ -257,21 +257,71 @@ 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",
                     bucket=bucket,
                     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)))

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

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