fix: hybrid passthrough for Scylla manifest mid-frame range tail

chart/values.yaml

@@ -16,7 +16,9 @@ server:
   noTls: true
 
 performance:
-  memoryLimitMb: 64
+  # Governor budget per pod. Prod runs 192MB on 1Gi pods; raise if MEMORY_REJECTED
+  # appears under backup load (Scylla manifest copies + barman multipart).
+  memoryLimitMb: 128
 
 # Redis holds only transient multipart-upload state (TTL'd, deleted on
 # completion). It is NOT a durable store — losing it only forces in-flight

s3proxy/app.py

@@ -25,6 +25,7 @@
 from .client import SigV4Verifier
 from .config import Settings
 from .errors import S3Error, get_s3_error_code
+from .request_context import get_request_context
 from .handlers import S3ProxyHandler
 from .handlers.base import close_http_client
 from .request_handler import handle_proxy_request
@@ -263,6 +264,14 @@ async def s3_exception_handler(request: Request, exc: HTTPException):
             error_code = get_s3_error_code(exc.status_code, exc.detail)
             message = exc.detail or "Unknown error"
 
+        logger.warning(
+            "S3_ERROR_RESPONSE",
+            status_code=exc.status_code,
+            error_code=error_code,
+            message=str(message),
+            **get_request_context(),
+        )
+
         error_xml = f"""<?xml version="1.0" encoding="UTF-8"?>
 <Error>
     <Code>{xml_escape(error_code)}</Code>

s3proxy/client/s3.py

@@ -57,7 +57,7 @@ def __init__(self, settings: Settings, credentials: S3Credentials):
             retries={"max_attempts": 3, "mode": "adaptive"},
             max_pool_connections=100,
             connect_timeout=10,
-            read_timeout=60,
+            read_timeout=300,
         )
         self._cached_client = None
         self._client_context = None

s3proxy/concurrency.py

@@ -20,6 +20,8 @@
 from s3proxy.errors import S3Error
 from s3proxy.metrics import MEMORY_LIMIT_BYTES, MEMORY_REJECTIONS, MEMORY_RESERVED_BYTES
 
+from .request_context import get_request_context
+
 logger = structlog.get_logger(__name__)
 
 # Constants
@@ -47,7 +49,7 @@ def _create_malloc_release() -> Callable[[], int] | None:
 _malloc_release = _create_malloc_release()
 
 
-BACKPRESSURE_TIMEOUT = int(os.environ.get("S3PROXY_BACKPRESSURE_TIMEOUT", "30"))
+BACKPRESSURE_TIMEOUT = int(os.environ.get("S3PROXY_BACKPRESSURE_TIMEOUT", "120"))
 
 
 class ConcurrencyLimiter:
@@ -153,6 +155,7 @@ async def try_acquire(self, bytes_needed: int, *, copy: bool = False) -> int:
                             requested_mb=round(request_mb, 2),
                             limit_mb=round(limit_mb, 2),
                             waited_sec=BACKPRESSURE_TIMEOUT,
+                            **get_request_context(),
                         )
                         MEMORY_REJECTIONS.inc()
                         raise S3Error.slow_down(
@@ -166,6 +169,7 @@ async def try_acquire(self, bytes_needed: int, *, copy: bool = False) -> int:
                             requested_mb=round(to_reserve / 1024 / 1024, 2),
                             limit_mb=round(self._limit_bytes / 1024 / 1024, 2),
                             remaining_sec=round(remaining, 1),
+                            **get_request_context(),
                         )
                         logged_backpressure = True
                     with contextlib.suppress(TimeoutError):

s3proxy/crypto.py

@@ -41,6 +41,8 @@
 # part (see state.manager), so a single client part may not expand into more
 # than this many internal parts without colliding into the next part's range.
 MAX_INTERNAL_PARTS_PER_CLIENT = 20
+# S3 multipart uploads allow at most 10,000 parts (AWS API limit).
+S3_MAX_PART_NUMBER = 10_000
 
 # Server-side copies use a FIXED internal part size instead of object_size/20.
 # A large single-part copy (Scylla dedup copies a 4.7GB SSTable as one
@@ -162,6 +164,28 @@ def calculate_optimal_part_size(content_length: int) -> int:
     return PART_SIZE
 
 
+def internal_part_range(client_part_number: int, count: int) -> tuple[int, int]:
+    """Inclusive internal part number range reserved for one client part."""
+    start = (client_part_number - 1) * MAX_INTERNAL_PARTS_PER_CLIENT + 1
+    return start, start + count - 1
+
+
+def validate_internal_part_allocation(client_part_number: int, count: int) -> tuple[int, int]:
+    """Return (start, end) or raise if the range exceeds S3's part-number ceiling."""
+    start, end = internal_part_range(client_part_number, count)
+    if end > S3_MAX_PART_NUMBER:
+        raise ValueError(
+            f"client part {client_part_number} needs internal parts {start}-{end} "
+            f"but S3 allows at most {S3_MAX_PART_NUMBER}"
+        )
+    if count > MAX_INTERNAL_PARTS_PER_CLIENT:
+        raise ValueError(
+            f"client part {client_part_number} needs {count} internal parts "
+            f"but at most {MAX_INTERNAL_PARTS_PER_CLIENT} are allowed per client part"
+        )
+    return start, end
+
+
 def memory_bounded_part_size(
     content_length: int, max_parts: int = MAX_INTERNAL_PARTS_PER_CLIENT
 ) -> int:

s3proxy/errors.py

@@ -2,11 +2,16 @@
 
 from __future__ import annotations
 
-from typing import NoReturn
+from typing import Any, NoReturn
 
+import structlog
 from botocore.exceptions import ClientError
 from fastapi import HTTPException
 
+from .request_context import get_request_context
+
+logger = structlog.get_logger(__name__)
+
 # S3 Error Code mappings
 # https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html
 S3_ERROR_CODES = {
@@ -206,6 +211,35 @@ def get_s3_error_code(status_code: int, detail: str | None = None) -> str:
     return S3_ERROR_CODES.get(status_code, "InternalError")
 
 
+def _log_upstream_failure(
+    *,
+    source: str,
+    exc: Exception,
+    bucket: str | None = None,
+    key: str | None = None,
+    extra: dict[str, Any] | None = None,
+) -> None:
+    fields: dict[str, Any] = {
+        "event": "UPSTREAM_FAILURE",
+        "source": source,
+        "error_type": type(exc).__name__,
+        "error": str(exc),
+        **get_request_context(),
+    }
+    if bucket is not None:
+        fields["bucket"] = bucket
+    if key is not None:
+        fields["key"] = key
+    if extra:
+        fields.update(extra)
+    if isinstance(exc, ClientError):
+        err = exc.response.get("Error", {})
+        fields["aws_error_code"] = err.get("Code", "")
+        fields["aws_error_message"] = err.get("Message", "")
+        fields["http_status"] = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
+    logger.error(**fields)
+
+
 def raise_for_client_error(
     e: ClientError,
     bucket: str | None = None,
@@ -214,6 +248,7 @@ def raise_for_client_error(
     """Convert botocore ClientError to S3Error. Always raises."""
     code = e.response.get("Error", {}).get("Code", "")
     msg = e.response.get("Error", {}).get("Message", str(e))
+    _log_upstream_failure(source="client_error", exc=e, bucket=bucket, key=key)
 
     if code == "NoSuchUpload":
         raise S3Error.no_such_upload(msg) from e
@@ -227,13 +262,26 @@ def raise_for_client_error(
         raise S3Error.bucket_already_exists(bucket) from e
     if code == "BucketAlreadyOwnedByYou":
         raise S3Error.bucket_already_owned_by_you(bucket) from e
+    if code in ("InvalidPart", "InvalidPartNumber"):
+        raise S3Error.invalid_part(msg or code) from e
+    if code == "EntityTooSmall":
+        raise S3Error.entity_too_small(msg or code) from e
+    if code == "EntityTooLarge":
+        raise S3Error.entity_too_large(512) from e
+    if code in ("InvalidArgument", "MalformedXML"):
+        raise S3Error.invalid_argument(msg or code) from e
+    if code in ("RequestTimeout", "RequestTimeTooSkewed"):
+        raise S3Error.bad_request(msg or code) from e
+    if code in ("ServiceUnavailable", "SlowDown", "Throttling", "ThrottlingException"):
+        raise S3Error.slow_down(msg or code) from e
     raise S3Error.internal_error(msg) from e
 
 
 def raise_for_exception(e: Exception) -> NoReturn:
     """Convert generic exception to S3Error. Always raises."""
     exc_name = type(e).__name__
     error_str = str(e)
+    _log_upstream_failure(source="exception", exc=e)
 
     # Handle NoSuchUpload that may come as a non-ClientError
     if exc_name == "NoSuchUpload" or "NoSuchUpload" in error_str:

s3proxy/handlers/multipart/copy.py

@@ -57,6 +57,14 @@ class _CiphertextSegment:
     ct_offset: int
 
 
+@dataclass(frozen=True, slots=True)
+class _PlaintextRangeSplit:
+    """Passthrough-safe prefix segments plus optional plaintext tail to re-encrypt."""
+
+    passthrough_segments: tuple[_CiphertextSegment, ...]
+    streaming_tail: tuple[int, int] | None  # inclusive plaintext [start, end]
+
+
 class CopyPartMixin(BaseHandler):
     async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) -> Response:
         bucket, key = self._parse_path(request.url.path)
@@ -78,6 +86,16 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
             try:
                 head_resp = await client.head_object(src_bucket, src_key)
             except Exception as e:
+                logger.error(
+                    "UPLOAD_PART_COPY_HEAD_FAILED",
+                    bucket=bucket,
+                    key=key,
+                    client_part=part_num,
+                    src_bucket=src_bucket,
+                    src_key=src_key,
+                    error_type=type(e).__name__,
+                    error=str(e),
+                )
                 raise S3Error.no_such_key(src_key) from e
 
             src_metadata = head_resp.get("Metadata", {})
@@ -263,17 +281,22 @@ def _normalize_copy_source_range(
             return None
         return copy_source_range
 
-    def _segments_for_plaintext_range(
+    def _split_plaintext_range_on_segments(
         self,
         segments: list[_CiphertextSegment],
         range_start: int,
         range_end: int,
-    ) -> list[_CiphertextSegment] | None:
-        """Segments fully inside [range_start, range_end], or None if range splits a segment."""
+    ) -> _PlaintextRangeSplit | None:
+        """Split a plaintext range into ciphertext-passthrough prefix + optional streaming tail.
+
+        Scylla manifest ranges often end mid internal frame (~8.33MB from 50MB uploads).
+        Copy every fully covered segment server-side and re-encrypt only the trailing suffix.
+        """
         if range_start > range_end:
-            return []
+            return _PlaintextRangeSplit((), None)
         selected: list[_CiphertextSegment] = []
         pt_offset = 0
+        streaming_tail: tuple[int, int] | None = None
         for seg in segments:
             seg_start = pt_offset
             seg_end = pt_offset + seg.plaintext_size - 1
@@ -282,11 +305,30 @@ def _segments_for_plaintext_range(
                 continue
             if seg_start > range_end:
                 break
-            if seg_start < range_start or seg_end > range_end:
+            if seg_start < range_start:
                 return None
+            if seg_end > range_end:
+                streaming_tail = (seg_start, range_end)
+                break
             selected.append(seg)
             pt_offset += seg.plaintext_size
-        return selected
+        if streaming_tail is None and pt_offset <= range_end:
+            return None
+        return _PlaintextRangeSplit(tuple(selected), streaming_tail)
+
+    def _segments_for_plaintext_range(
+        self,
+        segments: list[_CiphertextSegment],
+        range_start: int,
+        range_end: int,
+    ) -> list[_CiphertextSegment] | None:
+        """Segments fully inside [range_start, range_end], or None if range splits a segment."""
+        split = self._split_plaintext_range_on_segments(segments, range_start, range_end)
+        if split is None:
+            return None
+        if split.streaming_tail is not None:
+            return None
+        return list(split.passthrough_segments)
 
     def _passthrough_block_reason(
         self,
@@ -329,11 +371,16 @@ def _passthrough_block_reason(
         segments = self._source_ciphertext_segments(
             src_multipart_meta, head_resp, src_wrapped_dek, src_metadata
         )
-        filtered = self._segments_for_plaintext_range(segments, range_start, range_end)
-        if filtered is None:
+        split = self._split_plaintext_range_on_segments(segments, range_start, range_end)
+        if split is None:
             return "ranged_copy_segment_misaligned"
-        if not filtered:
+        if not split.passthrough_segments and not split.streaming_tail:
             return "ranged_copy_empty_segments"
+        if not split.passthrough_segments and split.streaming_tail:
+            tail_bytes = split.streaming_tail[1] - split.streaming_tail[0] + 1
+            if tail_bytes <= crypto.STREAMING_THRESHOLD:
+                return "small_ranged_copy"
+            return "ranged_copy_segment_misaligned"
         return None
 
     def _log_upload_part_copy_route(
@@ -560,12 +607,20 @@ async def _passthrough_copy_part(
             range_start, range_end = self._parse_copy_source_range(
                 copy_source_range, src_multipart_meta.total_plaintext_size
             )
-            filtered = self._segments_for_plaintext_range(segments, range_start, range_end)
-            if filtered is None:
+            split = self._split_plaintext_range_on_segments(segments, range_start, range_end)
+            if split is None:
                 raise S3Error.invalid_request("Copy range splits an encrypted segment")
-            segments = filtered
+            segments = list(split.passthrough_segments)
+            streaming_tail = split.streaming_tail
+        else:
+            streaming_tail = None
 
         copy_source_path = copy_source or f"/{src_bucket}/{quote(src_key, safe='/')}"
+        tail_mb = (
+            f"{(streaming_tail[1] - streaming_tail[0] + 1) / 1024 / 1024:.2f}MB"
+            if streaming_tail
+            else None
+        )
 
         logger.info(
             "UPLOAD_PART_COPY_PASSTHROUGH",
@@ -576,6 +631,7 @@ async def _passthrough_copy_part(
             src_key=src_key,
             plaintext_mb=f"{plaintext_size / 1024 / 1024:.2f}MB",
             segments=len(segments),
+            streaming_tail_mb=tail_mb,
             copy_source_range=copy_source_range,
         )
 
@@ -629,7 +685,11 @@ async def _passthrough_copy_part(
             )
 
         internal_part_start = await self.multipart_manager.allocate_internal_parts(
-            bucket, key, upload_id, len(segments), client_part_number=0
+            bucket,
+            key,
+            upload_id,
+            len(segments) + (1 if streaming_tail else 0),
+            client_part_number=0,
         )
 
         internal_parts: list[InternalPartMetadata] = []
@@ -662,6 +722,35 @@ async def _passthrough_copy_part(
             total_plaintext += seg.plaintext_size
             total_ciphertext += seg.ciphertext_size
 
+        if streaming_tail and src_multipart_meta:
+            tail_start, tail_end = streaming_tail
+            tail_plaintext = await self._download_encrypted_multipart(
+                client, src_bucket, src_key, src_multipart_meta, tail_start, tail_end
+            )
+            internal_num = internal_part_start + len(segments)
+            tail_ct = crypto.encrypt_frame(tail_plaintext, state.dek, upload_id, internal_num, 0)
+            part_reserve = crypto.copy_chunk_peak(len(tail_plaintext))
+            async with concurrency.reserve_copy_memory(part_reserve):
+                resp = await client.upload_part(bucket, key, upload_id, internal_num, tail_ct)
+            internal_parts.append(
+                InternalPartMetadata(
+                    internal_part_number=internal_num,
+                    plaintext_size=len(tail_plaintext),
+                    ciphertext_size=len(tail_ct),
+                    etag=resp["ETag"].strip('"'),
+                )
+            )
+            total_plaintext += len(tail_plaintext)
+            total_ciphertext += len(tail_ct)
+            logger.info(
+                "UPLOAD_PART_COPY_PASSTHROUGH_TAIL",
+                bucket=bucket,
+                key=key,
+                part_num=part_num,
+                internal_part=internal_num,
+                tail_plaintext_mb=f"{len(tail_plaintext) / 1024 / 1024:.2f}MB",
+            )
+
         md5 = await self._md5_plaintext_from_copy_source(
             client,
             src_bucket,