From 7d427fc5fdbd683b11d6886bbbf53a8ea686838a Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Wed, 22 Jul 2026 10:18:52 +0200 Subject: [PATCH] fix: retry CompleteMultipartUpload on transient backend errors Hetzner streams a 200 for CompleteMultipartUpload then can fail mid-response with an embedded InternalError ("The server did not respond in time.") - a documented S3-family quirk for this operation. Nothing retried it: botocore's own retry checker only treats HTTP 5xx status and a small error-code allowlist as transient (neither covers a 200 status with an error body), and the handler bare-raised everything except EntityTooSmall. Scylla backups saw these as failed multi-GB SSTable uploads and had to re-upload the whole file. Add a bounded retry (reusing base.is_retryable_source_error) around complete_multipart_upload. A plain retry isn't safe by itself: if the backend actually finished assembling the object before the response died, the upload_id is already invalidated, so the retry surfaces NoSuchUpload even though the object is fine. Guard that case with a head_object check before treating NoSuchUpload-after-retry as a real failure. --- s3proxy/handlers/multipart/lifecycle.py | 86 +++++++++- tests/unit/test_complete_multipart_retry.py | 176 ++++++++++++++++++++ 2 files changed, 258 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_complete_multipart_retry.py diff --git a/s3proxy/handlers/multipart/lifecycle.py b/s3proxy/handlers/multipart/lifecycle.py index f7af824..ecfd264 100644 --- a/s3proxy/handlers/multipart/lifecycle.py +++ b/s3proxy/handlers/multipart/lifecycle.py @@ -6,8 +6,9 @@ import base64 import contextlib import hashlib +import os import xml.etree.ElementTree as ET -from typing import NoReturn +from typing import Any, NoReturn import structlog from botocore.exceptions import ClientError @@ -28,10 +29,19 @@ synthetic_multipart_etag, ) from ...xml_utils import find_elements, get_element_text -from ..base import BaseHandler +from ..base import BaseHandler, is_retryable_source_error logger: BoundLogger = structlog.get_logger(__name__) +# Backend (Hetzner) can stream a 200 for CompleteMultipartUpload and still fail +# server-side mid-response, embedding an InternalError in the body - a documented +# S3-family quirk for this operation specifically. Retrying the same upload_id +# and part list is safe *unless* the prior attempt actually finished assembling +# the object before the response died, in which case the upload_id is already +# invalidated and a retry surfaces NoSuchUpload - see _verify_already_completed. +COMPLETE_RETRY_ATTEMPTS = int(os.environ.get("S3PROXY_COMPLETE_RETRY_ATTEMPTS", "4")) +COMPLETE_RETRY_BACKOFF_SEC = float(os.environ.get("S3PROXY_COMPLETE_RETRY_BACKOFF", "0.5")) + class LifecycleMixin(BaseHandler): async def _recover_upload_state( @@ -181,8 +191,8 @@ async def handle_complete_multipart_upload( # Complete in S3 try: - complete_resp = await client.complete_multipart_upload( - bucket, key, upload_id, s3_parts + complete_resp = await self._complete_multipart_upload_with_retry( + client, bucket, key, upload_id, s3_parts, completed_parts ) except ClientError as e: await self._handle_complete_error( @@ -323,6 +333,74 @@ def _build_s3_parts( s3_parts.sort(key=lambda p: p["PartNumber"]) return s3_parts, completed_parts, total_plaintext + async def _complete_multipart_upload_with_retry( + self, + client: S3Client, + bucket: str, + key: str, + upload_id: str, + s3_parts: list[dict[str, int | str]], + completed_parts: list[PartMetadata], + ) -> dict[str, Any]: + expected_ciphertext_size = sum(p.ciphertext_size for p in completed_parts) + last_exc: ClientError | None = None + + for attempt in range(1, COMPLETE_RETRY_ATTEMPTS + 1): + try: + return await client.complete_multipart_upload(bucket, key, upload_id, s3_parts) + except ClientError as e: + error_code = e.response.get("Error", {}).get("Code", "") + + # A previous attempt may have actually finished assembling the + # object before its response died - the upload_id would then + # already be invalidated. Confirm before treating this as failed. + if error_code == "NoSuchUpload" and attempt > 1: + verified = await self._verify_already_completed( + client, bucket, key, expected_ciphertext_size + ) + if verified is not None: + logger.warning( + "COMPLETE_MULTIPART_ALREADY_DONE", + bucket=bucket, + key=key, + upload_id=upload_id[:20] + "...", + attempt=attempt, + ) + return verified + raise + + if not is_retryable_source_error(e) or attempt == COMPLETE_RETRY_ATTEMPTS: + raise + + logger.warning( + "COMPLETE_MULTIPART_RETRY", + bucket=bucket, + key=key, + upload_id=upload_id[:20] + "...", + attempt=attempt, + aws_error_code=error_code, + error=str(e), + ) + last_exc = e + await asyncio.sleep(COMPLETE_RETRY_BACKOFF_SEC * (2 ** (attempt - 1))) + + assert last_exc is not None # loop always returns or raises before exhausting + raise last_exc + + async def _verify_already_completed( + self, client: S3Client, bucket: str, key: str, expected_ciphertext_size: int + ) -> dict[str, Any] | None: + """Check whether a retried CompleteMultipartUpload's NoSuchUpload means the + prior attempt actually succeeded (backend assembled the object, then the + response delivery died before the client saw it).""" + try: + head = await client.head_object(bucket, key) + except ClientError: + return None + if head.get("ContentLength") == expected_ciphertext_size: + return {"ETag": head.get("ETag", "")} + return None + async def _handle_complete_error( self, e: ClientError, diff --git a/tests/unit/test_complete_multipart_retry.py b/tests/unit/test_complete_multipart_retry.py new file mode 100644 index 0000000..2d30d88 --- /dev/null +++ b/tests/unit/test_complete_multipart_retry.py @@ -0,0 +1,176 @@ +"""CompleteMultipartUpload retry + idempotency. + +Prod incident 2026-07-22: Hetzner streams a 200 for CompleteMultipartUpload +then fails mid-response with an embedded ``InternalError`` ("The server did +not respond in time.") - a documented S3-family quirk for this operation. +Nothing retried it: botocore's own retry checker only treats HTTP 5xx status +codes and a small error-code allowlist as transient, and neither covers a 200 +response with an error body, and the handler bare-``raise``d everything except +``EntityTooSmall``. rclone saw the failure and had to re-upload whole +multi-GB SSTables. + +A plain retry isn't safe on its own: if the backend actually finished +assembling the object before the response died, the ``upload_id`` is already +invalidated, so retrying surfaces ``NoSuchUpload`` even though the object is +fine. These tests pin both halves: the retry-on-transient-error behavior and +the head_object-based check that stops a retry from turning a successful +upload into a false failure. +""" + +from __future__ import annotations + +import pytest +from botocore.exceptions import ClientError + +from s3proxy.handlers.multipart import lifecycle +from s3proxy.state import PartMetadata + +COMPLETED_PARTS = [ + PartMetadata( + part_number=1, plaintext_size=100, ciphertext_size=128, etag="etag-1", md5="etag-1" + ), + PartMetadata( + part_number=2, plaintext_size=100, ciphertext_size=128, etag="etag-2", md5="etag-2" + ), +] +S3_PARTS = [{"PartNumber": 1, "ETag": '"etag-1"'}, {"PartNumber": 2, "ETag": '"etag-2"'}] +EXPECTED_CIPHERTEXT_SIZE = sum(p.ciphertext_size for p in COMPLETED_PARTS) + + +@pytest.fixture(autouse=True) +def _no_backoff(monkeypatch): + monkeypatch.setattr(lifecycle, "COMPLETE_RETRY_BACKOFF_SEC", 0.0) + + +def _client_error(code: str, message: str | None = None) -> ClientError: + return ClientError( + {"Error": {"Code": code, "Message": message or code}}, "CompleteMultipartUpload" + ) + + +class _FlakyCompleteClient: + """Fake backend client for exercising complete_multipart_upload retry paths. + + fail_times: number of leading calls that raise a transient error without + actually assembling the object. + error_code: the transient error code those calls raise. + phantom_success_on_first: the first call raises InternalError *after* the + backend actually finished assembling the object server-side (the "200 + then dies mid-response" case) - the retry then hits NoSuchUpload + because the upload_id is already gone. + abort_after_first_failure: the object is never assembled and the upload + is separately aborted/expired before the retry - retry hits + NoSuchUpload for real. + """ + + def __init__( + self, + *, + fail_times: int = 0, + error_code: str = "InternalError", + phantom_success_on_first: bool = False, + abort_after_first_failure: bool = False, + ) -> None: + self.calls = 0 + self.head_calls = 0 + self.fail_times = fail_times + self.error_code = error_code + self.phantom_success_on_first = phantom_success_on_first + self.abort_after_first_failure = abort_after_first_failure + self.object_assembled = False + + async def complete_multipart_upload(self, bucket, key, upload_id, parts): + self.calls += 1 + + if self.phantom_success_on_first: + if self.calls == 1: + self.object_assembled = True + raise _client_error("InternalError", "The server did not respond in time.") + raise _client_error("NoSuchUpload", "The specified upload does not exist") + + if self.abort_after_first_failure: + if self.calls == 1: + raise _client_error("InternalError", "The server did not respond in time.") + raise _client_error("NoSuchUpload", "The specified upload does not exist") + + if self.calls <= self.fail_times: + raise _client_error(self.error_code) + + self.object_assembled = True + return {"ETag": '"final-etag"'} + + async def head_object(self, bucket, key): + self.head_calls += 1 + if self.object_assembled: + return {"ContentLength": EXPECTED_CIPHERTEXT_SIZE, "ETag": '"final-etag"'} + raise _client_error("NoSuchKey", "Not Found") + + +@pytest.mark.asyncio +async def test_retries_transient_error_then_succeeds(handler): + client = _FlakyCompleteClient(fail_times=2) + + resp = await handler._complete_multipart_upload_with_retry( + client, "bucket", "key", "upload-1", S3_PARTS, COMPLETED_PARTS + ) + + assert resp == {"ETag": '"final-etag"'} + assert client.calls == 3 + + +@pytest.mark.asyncio +async def test_gives_up_after_max_attempts(handler): + client = _FlakyCompleteClient(fail_times=lifecycle.COMPLETE_RETRY_ATTEMPTS + 10) + + with pytest.raises(ClientError) as exc: + await handler._complete_multipart_upload_with_retry( + client, "bucket", "key", "upload-1", S3_PARTS, COMPLETED_PARTS + ) + + assert exc.value.response["Error"]["Code"] == "InternalError" + assert client.calls == lifecycle.COMPLETE_RETRY_ATTEMPTS + assert client.object_assembled is False + + +@pytest.mark.asyncio +async def test_does_not_retry_non_retryable_error(handler): + client = _FlakyCompleteClient(fail_times=1, error_code="AccessDenied") + + with pytest.raises(ClientError) as exc: + await handler._complete_multipart_upload_with_retry( + client, "bucket", "key", "upload-1", S3_PARTS, COMPLETED_PARTS + ) + + assert exc.value.response["Error"]["Code"] == "AccessDenied" + assert client.calls == 1 + + +@pytest.mark.asyncio +async def test_recovers_via_head_object_when_prior_attempt_already_finished(handler): + """The exact prod failure: backend finished the assembly, the client only saw + the error, and a naive retry would otherwise report a false failure.""" + client = _FlakyCompleteClient(phantom_success_on_first=True) + + resp = await handler._complete_multipart_upload_with_retry( + client, "bucket", "key", "upload-1", S3_PARTS, COMPLETED_PARTS + ) + + assert resp == {"ETag": '"final-etag"'} + assert client.calls == 2 + assert client.head_calls == 1 + + +@pytest.mark.asyncio +async def test_does_not_mask_a_real_failure_as_success(handler): + """NoSuchUpload after a retry must still fail if the object was never + actually assembled - the head_object check must not paper over genuine + failures.""" + client = _FlakyCompleteClient(abort_after_first_failure=True) + + with pytest.raises(ClientError) as exc: + await handler._complete_multipart_upload_with_retry( + client, "bucket", "key", "upload-1", S3_PARTS, COMPLETED_PARTS + ) + + assert exc.value.response["Error"]["Code"] == "NoSuchUpload" + assert client.head_calls == 1