fix: retry CompleteMultipartUpload on transient backend errors

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,

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