diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index fad86a3..a59ba4f 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -9,7 +9,7 @@ on:
jobs:
unit:
runs-on: ubuntu-latest
- timeout-minutes: 10
+ timeout-minutes: 15
steps:
- uses: actions/checkout@v7
diff --git a/chart/values.yaml b/chart/values.yaml
index fc43fb5..441ddb9 100644
--- a/chart/values.yaml
+++ b/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
diff --git a/s3proxy/app.py b/s3proxy/app.py
index c95f1e5..1547fa8 100644
--- a/s3proxy/app.py
+++ b/s3proxy/app.py
@@ -27,6 +27,7 @@
from .errors import S3Error, get_s3_error_code
from .handlers import S3ProxyHandler
from .handlers.base import close_http_client
+from .request_context import get_request_context
from .request_handler import handle_proxy_request
from .state import MultipartStateManager, close_redis, create_state_store, init_redis
@@ -263,20 +264,44 @@ 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_escape(error_code)}
{xml_escape(str(message))}
{request_id}
"""
+ headers = {
+ "x-amz-request-id": request_id,
+ "x-amz-id-2": request_id,
+ }
+ # A small request rejected before its body was read can leave the body
+ # bytes unconsumed on the keep-alive connection, and the client's next
+ # request on it gets a raw uvicorn 400 (e.g. AbortMultipartUpload after
+ # a rejected UploadPart). Drain small bodies so the connection stays
+ # clean. Large bodies are left alone: uvicorn discards them after the
+ # response, and closing the connection instead resets clients that
+ # finish sending before they read (presigned PUT uploaders).
+ if request.method in ("PUT", "POST"):
+ try:
+ content_length = int(request.headers.get("content-length", "0"))
+ except ValueError:
+ content_length = 0
+ if 0 < content_length <= concurrency.MAX_BUFFER_SIZE:
+ with contextlib.suppress(Exception):
+ await request.body()
return Response(
content=error_xml,
status_code=exc.status_code,
media_type="application/xml",
- headers={
- "x-amz-request-id": request_id,
- "x-amz-id-2": request_id,
- },
+ headers=headers,
)
diff --git a/s3proxy/client/s3.py b/s3proxy/client/s3.py
index ac294c9..3b82b7c 100644
--- a/s3proxy/client/s3.py
+++ b/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
diff --git a/s3proxy/concurrency.py b/s3proxy/concurrency.py
index 4abbba0..b1a638e 100644
--- a/s3proxy/concurrency.py
+++ b/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):
diff --git a/s3proxy/crypto.py b/s3proxy/crypto.py
index 7dfccf7..05213c4 100644
--- a/s3proxy/crypto.py
+++ b/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:
diff --git a/s3proxy/errors.py b/s3proxy/errors.py
index de662f3..11c3e04 100644
--- a/s3proxy/errors.py
+++ b/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,6 +262,18 @@ 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
@@ -234,6 +281,7 @@ 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:
diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py
index 333084c..3160dd4 100644
--- a/s3proxy/handlers/multipart/copy.py
+++ b/s3proxy/handlers/multipart/copy.py
@@ -2,22 +2,25 @@
import asyncio
import base64
+import contextlib
import hashlib
import math
import os
import time
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Coroutine
from dataclasses import dataclass
from datetime import UTC, datetime
from urllib.parse import quote
import structlog
+from botocore.exceptions import ClientError
from fastapi import Request, Response
+from fastapi.responses import StreamingResponse
from structlog.stdlib import BoundLogger
from ... import concurrency, crypto, xml_responses
from ...client import S3Client, S3Credentials
-from ...errors import S3Error
+from ...errors import S3Error, raise_for_client_error, raise_for_exception
from ...state import (
InternalPartMetadata,
MultipartMetadata,
@@ -39,6 +42,19 @@
MAX_PARALLEL_COPY_PIPELINES = int(os.environ.get("S3PROXY_MAX_PARALLEL_COPIES", "2"))
_copy_pipeline_semaphore = asyncio.Semaphore(MAX_PARALLEL_COPY_PIPELINES)
+# A multi-GB UploadPartCopy can take longer than the client's idle timeout
+# (rclone in scylla-manager-agent gives up after 5 minutes with no response
+# bytes). Like AWS S3, commit to 200 OK and trickle whitespace while the copy
+# runs, then send CopyPartResult -- or an document -- as the body.
+COPY_KEEPALIVE_INTERVAL = float(os.environ.get("S3PROXY_COPY_KEEPALIVE_INTERVAL", "5"))
+
+# Backend UploadPartCopy calls per passthrough copy run concurrently; a 4.7GB
+# Scylla part is ~570 x 8.33MB segments and sequential calls alone can exceed
+# the client timeout.
+PASSTHROUGH_SEGMENT_CONCURRENCY = int(
+ os.environ.get("S3PROXY_PASSTHROUGH_SEGMENT_CONCURRENCY", "8")
+)
+
def reset_copy_pipeline_semaphore(limit: int | None = None) -> None:
"""Reset the global copy pipeline semaphore (testing only)."""
@@ -57,6 +73,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 +102,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", {})
@@ -121,13 +155,56 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
passthrough_blocked_reason=passthrough_block,
)
- if plaintext_size <= crypto.STREAMING_THRESHOLD:
- # Small copies buffer the whole object + re-encrypt it; gate them
- # by the limiter too (they carry no body, so the request-level
- # reservation was ~nothing and a small-object flood ran unbounded).
+ # Validation and routing are done with real HTTP status semantics; the
+ # copy work itself runs while the response streams keepalive whitespace,
+ # so a copy that outlives the client's idle timeout still succeeds.
+ async def run_copy() -> bytes:
+ async with self._client(creds) as work_client:
+ if plaintext_size <= crypto.STREAMING_THRESHOLD:
+ # Small copies buffer the whole object + re-encrypt it; gate
+ # them by the limiter too (they carry no body, so the
+ # request-level reservation was ~nothing and a small-object
+ # flood ran unbounded).
+ if passthrough_block is None:
+ resp = await self._gated_passthrough_copy_part(
+ work_client,
+ bucket,
+ key,
+ upload_id,
+ part_num,
+ state,
+ src_bucket,
+ src_key,
+ copy_source,
+ head_resp,
+ src_metadata,
+ src_wrapped_dek,
+ src_multipart_meta,
+ plaintext_size,
+ copy_source_range,
+ )
+ return resp.body
+ peak = crypto.copy_pipeline_peak(plaintext_size)
+ async with concurrency.reserve_copy_memory(peak):
+ resp = await self._simple_copy_part(
+ work_client,
+ bucket,
+ key,
+ upload_id,
+ part_num,
+ state,
+ src_bucket,
+ src_key,
+ copy_source_range,
+ head_resp,
+ src_metadata,
+ src_wrapped_dek,
+ src_multipart_meta,
+ )
+ return resp.body
if passthrough_block is None:
- return await self._gated_passthrough_copy_part(
- client,
+ resp = await self._gated_passthrough_copy_part(
+ work_client,
bucket,
key,
upload_id,
@@ -143,26 +220,9 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
plaintext_size,
copy_source_range,
)
- peak = crypto.copy_pipeline_peak(plaintext_size)
- async with concurrency.reserve_copy_memory(peak):
- return await self._simple_copy_part(
- client,
- bucket,
- key,
- upload_id,
- part_num,
- state,
- src_bucket,
- src_key,
- copy_source_range,
- head_resp,
- src_metadata,
- src_wrapped_dek,
- src_multipart_meta,
- )
- if passthrough_block is None:
- return await self._gated_passthrough_copy_part(
- client,
+ return resp.body
+ resp = await self._streaming_copy_part(
+ work_client,
bucket,
key,
upload_id,
@@ -170,30 +230,93 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
state,
src_bucket,
src_key,
- copy_source,
- head_resp,
- src_metadata,
+ copy_source_range,
src_wrapped_dek,
src_multipart_meta,
+ head_resp,
+ src_metadata,
plaintext_size,
- copy_source_range,
)
- return await self._streaming_copy_part(
- client,
- bucket,
- key,
- upload_id,
- part_num,
- state,
- src_bucket,
- src_key,
- copy_source_range,
- src_wrapped_dek,
- src_multipart_meta,
- head_resp,
- src_metadata,
- plaintext_size,
- )
+ return resp.body
+
+ return StreamingResponse(
+ self._keepalive_copy_stream(run_copy(), bucket=bucket, key=key, part_num=part_num),
+ media_type="application/xml",
+ )
+
+ async def _keepalive_copy_stream(
+ self,
+ work: Coroutine[None, None, bytes],
+ *,
+ bucket: str,
+ key: str,
+ part_num: int,
+ ) -> AsyncIterator[bytes]:
+ """Run the copy while trickling whitespace, then emit the result XML.
+
+ Cancels the copy if the client disconnects, so an abandoned request does
+ not keep burning memory budget and backend bandwidth as a zombie.
+ """
+ task = asyncio.create_task(work)
+ start = time.monotonic()
+ keepalives = 0
+ try:
+ while True:
+ done, _ = await asyncio.wait({task}, timeout=COPY_KEEPALIVE_INTERVAL)
+ if done:
+ break
+ keepalives += 1
+ yield b" "
+ try:
+ body = task.result()
+ except Exception as e:
+ code, message = self._copy_failure_code_message(e, bucket, key)
+ logger.error(
+ "UPLOAD_PART_COPY_FAILED_AFTER_200",
+ bucket=bucket,
+ key=key,
+ part_num=part_num,
+ error_code=code,
+ error=message,
+ keepalives=keepalives,
+ elapsed_sec=f"{time.monotonic() - start:.2f}s",
+ )
+ yield xml_responses.error_document(code, message).encode()
+ return
+ if keepalives:
+ logger.info(
+ "UPLOAD_PART_COPY_KEEPALIVE",
+ bucket=bucket,
+ key=key,
+ part_num=part_num,
+ keepalives=keepalives,
+ elapsed_sec=f"{time.monotonic() - start:.2f}s",
+ )
+ yield xml_responses.without_xml_declaration(body.decode()).encode()
+ finally:
+ if not task.done():
+ task.cancel()
+ with contextlib.suppress(BaseException):
+ await task
+ logger.warning(
+ "UPLOAD_PART_COPY_CLIENT_GONE",
+ bucket=bucket,
+ key=key,
+ part_num=part_num,
+ keepalives=keepalives,
+ elapsed_sec=f"{time.monotonic() - start:.2f}s",
+ )
+
+ def _copy_failure_code_message(self, exc: Exception, bucket: str, key: str) -> tuple[str, str]:
+ """Map a copy failure to an S3 error code/message for the 200-body Error."""
+ if isinstance(exc, S3Error):
+ return exc.code, exc.message
+ try:
+ if isinstance(exc, ClientError):
+ raise_for_client_error(exc, bucket, key)
+ raise_for_exception(exc)
+ except S3Error as mapped:
+ return mapped.code, mapped.message
def _copy_plaintext_size(
self,
@@ -263,17 +386,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 +410,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 +476,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 +712,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 +736,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,
)
@@ -609,6 +770,9 @@ async def _passthrough_copy_part(
src_metadata,
)
etag = md5.hexdigest()
+ # Store the BACKEND part etag; CompleteMultipartUpload must present
+ # it to S3, and the synthetic plaintext etag returned to the client
+ # would be rejected there (InvalidPart).
await self.multipart_manager.add_part(
bucket,
key,
@@ -617,7 +781,7 @@ async def _passthrough_copy_part(
part_number=part_num,
plaintext_size=seg.plaintext_size,
ciphertext_size=seg.ciphertext_size,
- etag=etag,
+ etag=resp["CopyPartResult"]["ETag"].strip('"'),
md5=etag,
),
)
@@ -629,19 +793,22 @@ 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] = []
- total_plaintext = 0
- total_ciphertext = 0
+ start = time.monotonic()
+ segment_sem = asyncio.Semaphore(PASSTHROUGH_SEGMENT_CONCURRENCY)
- for i, seg in enumerate(segments):
- internal_num = internal_part_start + i
+ async def copy_segment(idx: int, seg: _CiphertextSegment) -> InternalPartMetadata:
+ internal_num = internal_part_start + idx
ct_end = seg.ct_offset + seg.ciphertext_size - 1
copy_range = f"bytes={seg.ct_offset}-{ct_end}"
part_reserve = crypto.copy_passthrough_segment_peak(seg.plaintext_size)
- async with concurrency.reserve_copy_memory(part_reserve):
+ async with segment_sem, concurrency.reserve_copy_memory(part_reserve):
resp = await client.upload_part_copy(
bucket,
key,
@@ -650,29 +817,72 @@ async def _passthrough_copy_part(
copy_source_path,
copy_source_range=copy_range,
)
- etag_part = resp["CopyPartResult"]["ETag"].strip('"')
- internal_parts.append(
- InternalPartMetadata(
- internal_part_number=internal_num,
- plaintext_size=seg.plaintext_size,
- ciphertext_size=seg.ciphertext_size,
- etag=etag_part,
- )
+ return InternalPartMetadata(
+ internal_part_number=internal_num,
+ plaintext_size=seg.plaintext_size,
+ ciphertext_size=seg.ciphertext_size,
+ etag=resp["CopyPartResult"]["ETag"].strip('"'),
)
- total_plaintext += seg.plaintext_size
- total_ciphertext += seg.ciphertext_size
- md5 = await self._md5_plaintext_from_copy_source(
- client,
- src_bucket,
- src_key,
- copy_source_range,
- src_wrapped_dek,
- src_multipart_meta,
- head_resp,
- src_metadata,
- )
- etag = md5.hexdigest()
+ async def upload_tail() -> InternalPartMetadata | None:
+ if not (streaming_tail and src_multipart_meta):
+ return None
+ 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)
+ 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",
+ )
+ return InternalPartMetadata(
+ internal_part_number=internal_num,
+ plaintext_size=len(tail_plaintext),
+ ciphertext_size=len(tail_ct),
+ etag=resp["ETag"].strip('"'),
+ )
+
+ # The synthetic-ETag MD5 pass reads the whole source; run it alongside
+ # the segment copies and the tail instead of serially after them --
+ # sequential, the three phases pushed a 4.7GB part past the client's
+ # idle timeout.
+ try:
+ async with asyncio.TaskGroup() as tg:
+ seg_tasks = [tg.create_task(copy_segment(i, seg)) for i, seg in enumerate(segments)]
+ tail_task = tg.create_task(upload_tail())
+ md5_task = tg.create_task(
+ self._md5_plaintext_from_copy_source(
+ client,
+ src_bucket,
+ src_key,
+ copy_source_range,
+ src_wrapped_dek,
+ src_multipart_meta,
+ head_resp,
+ src_metadata,
+ )
+ )
+ except BaseExceptionGroup as eg:
+ exc: BaseException = eg
+ while isinstance(exc, BaseExceptionGroup):
+ exc = exc.exceptions[0]
+ raise exc from eg
+
+ internal_parts = [t.result() for t in seg_tasks]
+ if (tail_part := tail_task.result()) is not None:
+ internal_parts.append(tail_part)
+ total_plaintext = sum(p.plaintext_size for p in internal_parts)
+ total_ciphertext = sum(p.ciphertext_size for p in internal_parts)
+ etag = md5_task.result().hexdigest()
await self.multipart_manager.add_part(
bucket,
key,
@@ -695,6 +905,7 @@ async def _passthrough_copy_part(
internal_parts=len(internal_parts),
plaintext_mb=f"{total_plaintext / 1024 / 1024:.2f}MB",
copy_source_range=copy_source_range,
+ elapsed_sec=f"{time.monotonic() - start:.2f}s",
)
return Response(
content=xml_responses.upload_part_copy_result(etag, format_iso8601(datetime.now(UTC))),
diff --git a/s3proxy/handlers/multipart/lifecycle.py b/s3proxy/handlers/multipart/lifecycle.py
index 40a6a5f..0e2cec0 100644
--- a/s3proxy/handlers/multipart/lifecycle.py
+++ b/s3proxy/handlers/multipart/lifecycle.py
@@ -262,10 +262,18 @@ def _build_s3_parts(
}
)
else:
+ # Use the stored backend etag, not the client-echoed one:
+ # single-segment passthrough copies return a synthetic
+ # plaintext etag to the client that S3 would reject.
+ etag = (
+ f'"{part_meta.etag}"'
+ if not part_meta.etag.startswith('"')
+ else part_meta.etag
+ )
s3_parts.append(
{
"PartNumber": client_part_num,
- "ETag": cp["ETag"],
+ "ETag": etag,
}
)
else:
diff --git a/s3proxy/handlers/multipart/upload_part.py b/s3proxy/handlers/multipart/upload_part.py
index 1f21977..9b298dc 100644
--- a/s3proxy/handlers/multipart/upload_part.py
+++ b/s3proxy/handlers/multipart/upload_part.py
@@ -150,6 +150,16 @@ async def handle_upload_part(self, request: Request, creds: S3Credentials) -> Re
estimated_parts,
client_part_number=part_num,
)
+ internal_part_end = internal_part_start + estimated_parts - 1
+ logger.info(
+ "UPLOAD_PART_INTERNAL_RANGE",
+ bucket=bucket,
+ key=key,
+ part_number=part_num,
+ internal_part_start=internal_part_start,
+ internal_part_end=internal_part_end,
+ estimated_internal_parts=estimated_parts,
+ )
try:
# Known-length direct streams (unsigned or large signed, e.g. barman
@@ -168,6 +178,7 @@ async def handle_upload_part(self, request: Request, creds: S3Credentials) -> Re
content_length,
internal_part_size,
internal_part_start,
+ estimated_parts,
)
else:
result = await self._stream_and_upload(
@@ -392,6 +403,7 @@ async def _stream_and_upload_framed(
content_length: int,
optimal_part_size: int,
internal_part_start: int,
+ estimated_internal_parts: int,
) -> dict[str, str | int]:
"""Memory-bounded UploadPart for known-length direct streams.
@@ -431,6 +443,23 @@ async def _stream_and_upload_framed(
)
frame_idx += 1
+ if remaining > 0:
+ bytes_read = part_pt_size - remaining
+ logger.error(
+ "UPLOAD_PART_STREAM_SHORT",
+ bucket=bucket,
+ key=key,
+ client_part=part_num,
+ internal_part=ipn,
+ expected_bytes=part_pt_size,
+ received_bytes=bytes_read,
+ content_length=content_length,
+ )
+ raise S3Error.bad_request(
+ f"Upload body ended early: got {bytes_read}B of {part_pt_size}B "
+ f"for client part {part_num} internal part {ipn}"
+ )
+
upload_start = time.monotonic()
resp = await client.upload_part(bucket, key, upload_id, ipn, ciphertext)
del ciphertext
@@ -457,6 +486,16 @@ async def _stream_and_upload_framed(
internal_part_num += 1
remaining_total -= part_pt_size
+ if internal_part_num - 1 > internal_part_start + estimated_internal_parts - 1:
+ logger.warning(
+ "UPLOAD_PART_INTERNAL_RANGE_EXCEEDED",
+ bucket=bucket,
+ key=key,
+ client_part=part_num,
+ allocated_end=internal_part_start + estimated_internal_parts - 1,
+ actual_end=internal_part_num - 1,
+ )
+
client_etag = md5_hash.hexdigest()
part_meta = PartMetadata(
part_number=part_num,
@@ -617,14 +656,47 @@ def _check_upload_results(
bucket=bucket,
key=key,
upload_id=upload_id,
+ client_part=part_num,
)
raise S3Error.no_such_upload(upload_id)
+ if isinstance(result, ClientError):
+ error_code = result.response.get("Error", {}).get("Code", "")
+ error_msg = result.response.get("Error", {}).get("Message", str(result))
+ logger.error(
+ "UPLOAD_PART_INTERNAL_FAILED",
+ bucket=bucket,
+ key=key,
+ client_part=part_num,
+ upload_id=upload_id[:20] + "...",
+ aws_error_code=error_code,
+ aws_error_message=error_msg,
+ )
+ else:
+ logger.error(
+ "UPLOAD_PART_INTERNAL_FAILED",
+ bucket=bucket,
+ key=key,
+ client_part=part_num,
+ upload_id=upload_id[:20] + "...",
+ error_type=exc_name,
+ error=str(result),
+ )
raise result
def _handle_client_error(
self, e: ClientError, bucket: str, key: str, part_num: int, upload_id: str
) -> NoReturn:
- logger.error("UPLOAD_PART_CLIENT_ERROR", bucket=bucket, key=key, part_num=part_num)
+ error_code = e.response.get("Error", {}).get("Code", "")
+ error_msg = e.response.get("Error", {}).get("Message", str(e))
+ logger.error(
+ "UPLOAD_PART_CLIENT_ERROR",
+ bucket=bucket,
+ key=key,
+ part_num=part_num,
+ upload_id=upload_id[:20] + "...",
+ aws_error_code=error_code,
+ aws_error_message=error_msg,
+ )
raise_for_client_error(e, bucket, key)
def _handle_generic_error(
@@ -635,6 +707,7 @@ def _handle_generic_error(
bucket=bucket,
key=key,
part_num=part_num,
+ upload_id=upload_id[:20] + "...",
error=str(e),
error_type=type(e).__name__,
)
diff --git a/s3proxy/request_context.py b/s3proxy/request_context.py
new file mode 100644
index 0000000..fb48f02
--- /dev/null
+++ b/s3proxy/request_context.py
@@ -0,0 +1,42 @@
+"""Per-request context for structured logging across async call stacks."""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+from typing import Any
+from urllib.parse import parse_qs
+
+_request_ctx: ContextVar[dict[str, Any] | None] = ContextVar("s3proxy_request_ctx", default=None)
+
+
+def bind_request(
+ *,
+ method: str,
+ path: str,
+ query: str = "",
+ content_length: int | None = None,
+) -> None:
+ """Attach request fields visible to nested handlers (memory governor, errors)."""
+ ctx: dict[str, Any] = {"method": method, "path": path}
+ if content_length is not None:
+ ctx["content_length"] = content_length
+ if query:
+ params = parse_qs(query, keep_blank_values=True)
+ upload_ids = params.get("uploadId") or params.get("uploadid")
+ part_nums = params.get("partNumber") or params.get("partnumber")
+ if upload_ids:
+ ctx["upload_id"] = upload_ids[0]
+ if part_nums:
+ try:
+ ctx["part_number"] = int(part_nums[0])
+ except ValueError:
+ ctx["part_number"] = part_nums[0]
+ _request_ctx.set(ctx)
+
+
+def clear_request() -> None:
+ _request_ctx.set({})
+
+
+def get_request_context() -> dict[str, Any]:
+ return dict(_request_ctx.get() or {})
diff --git a/s3proxy/request_handler.py b/s3proxy/request_handler.py
index fe17de8..4b34490 100644
--- a/s3proxy/request_handler.py
+++ b/s3proxy/request_handler.py
@@ -24,6 +24,7 @@
REQUESTS_IN_FLIGHT,
get_operation_name,
)
+from .request_context import bind_request, clear_request, get_request_context
from .routing import RequestDispatcher
pod_name = os.environ.get("HOSTNAME", "unknown")
@@ -133,6 +134,8 @@ async def handle_proxy_request(
REQUESTS_IN_FLIGHT.labels(method=method).inc()
+ bind_request(method=method, path=path, query=query, content_length=None)
+
# Check memory limit BEFORE reading body data - reject if at capacity
reserved_memory = 0
needs_limit = method in ("PUT", "POST", "GET")
@@ -143,6 +146,7 @@ async def handle_proxy_request(
content_length = int(request.headers.get("content-length", "0"))
except ValueError:
content_length = 0
+ bind_request(method=method, path=path, query=query, content_length=content_length)
memory_needed = concurrency.estimate_memory_footprint(method, content_length)
logger.info(
@@ -182,11 +186,37 @@ async def handle_proxy_request(
return response
except HTTPException as e:
status_code = e.status_code
+ if isinstance(e, S3Error):
+ logger.warning(
+ "REQUEST_S3_ERROR",
+ status_code=e.status_code,
+ code=e.code,
+ message=e.message,
+ active_mb=round(concurrency.get_active_memory() / 1024 / 1024, 2),
+ **get_request_context(),
+ )
+ else:
+ logger.warning(
+ "REQUEST_HTTP_ERROR",
+ status_code=e.status_code,
+ detail=e.detail,
+ active_mb=round(concurrency.get_active_memory() / 1024 / 1024, 2),
+ **get_request_context(),
+ )
raise
- except Exception:
+ except Exception as e:
status_code = 500
+ logger.error(
+ "REQUEST_UNHANDLED_EXCEPTION",
+ error_type=type(e).__name__,
+ error=str(e),
+ active_mb=round(concurrency.get_active_memory() / 1024 / 1024, 2),
+ **get_request_context(),
+ exc_info=True,
+ )
raise
finally:
+ clear_request()
# Record metrics
duration = time.perf_counter() - start_time
REQUESTS_IN_FLIGHT.labels(method=method).dec()
diff --git a/s3proxy/state/manager.py b/s3proxy/state/manager.py
index 201d97d..42b28ab 100644
--- a/s3proxy/state/manager.py
+++ b/s3proxy/state/manager.py
@@ -3,7 +3,8 @@
import structlog
from structlog.stdlib import BoundLogger
-from ..crypto import MAX_INTERNAL_PARTS_PER_CLIENT
+from ..crypto import MAX_INTERNAL_PARTS_PER_CLIENT, validate_internal_part_allocation
+from ..errors import S3Error
from .models import (
MultipartUploadState,
PartMetadata,
@@ -230,7 +231,19 @@ async def allocate_internal_parts(
internal part numbers to avoid conflicts.
"""
if client_part_number > 0:
- start = (client_part_number - 1) * MAX_INTERNAL_PARTS_PER_CLIENT + 1
+ try:
+ start, end = validate_internal_part_allocation(client_part_number, count)
+ except ValueError as e:
+ logger.error(
+ "INTERNAL_PART_ALLOCATION_REJECTED",
+ bucket=bucket,
+ key=key,
+ client_part=client_part_number,
+ requested=count,
+ max_per_client=MAX_INTERNAL_PARTS_PER_CLIENT,
+ error=str(e),
+ )
+ raise S3Error.invalid_part(str(e)) from e
if count > MAX_INTERNAL_PARTS_PER_CLIENT:
logger.warning(
@@ -242,14 +255,14 @@ async def allocate_internal_parts(
max=MAX_INTERNAL_PARTS_PER_CLIENT,
)
- logger.debug(
+ logger.info(
"ALLOCATE_INTERNAL_PARTS",
bucket=bucket,
key=key,
client_part=client_part_number,
count=count,
start=start,
- end=start + count - 1,
+ end=end,
)
return start
diff --git a/s3proxy/state/storage.py b/s3proxy/state/storage.py
index 1b047d1..64ac1cb 100644
--- a/s3proxy/state/storage.py
+++ b/s3proxy/state/storage.py
@@ -6,6 +6,7 @@
from __future__ import annotations
+import asyncio
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import TYPE_CHECKING
@@ -22,7 +23,8 @@
Updater = Callable[[bytes], bytes]
# Maximum retries for Redis optimistic locking (WATCH/MULTI/EXEC)
-MAX_WATCH_RETRIES = 5
+MAX_WATCH_RETRIES = 25
+WATCH_RETRY_BASE_DELAY_SEC = 0.005
class StateStore(ABC):
@@ -141,9 +143,17 @@ async def get_and_delete(self, key: str) -> bytes | None:
"REDIS_WATCH_RETRIES_EXHAUSTED",
key=key,
operation="get_and_delete",
+ attempts=MAX_WATCH_RETRIES,
)
raise
- logger.debug("REDIS_WATCH_RETRY", key=key, attempt=attempt + 1)
+ logger.warning(
+ "REDIS_WATCH_RETRY",
+ key=key,
+ attempt=attempt + 1,
+ max_attempts=MAX_WATCH_RETRIES,
+ operation="get_and_delete",
+ )
+ await asyncio.sleep(WATCH_RETRY_BASE_DELAY_SEC * (2**attempt))
return None
async def update(self, key: str, updater: Updater, ttl_seconds: int) -> bytes | None:
@@ -168,7 +178,19 @@ async def update(self, key: str, updater: Updater, ttl_seconds: int) -> bytes |
return new_data
except redis.WatchError:
if attempt == MAX_WATCH_RETRIES - 1:
- logger.error("REDIS_WATCH_RETRIES_EXHAUSTED", key=key, operation="update")
+ logger.error(
+ "REDIS_WATCH_RETRIES_EXHAUSTED",
+ key=key,
+ operation="update",
+ attempts=MAX_WATCH_RETRIES,
+ )
raise
- logger.debug("REDIS_WATCH_RETRY", key=key, attempt=attempt + 1)
+ logger.warning(
+ "REDIS_WATCH_RETRY",
+ key=key,
+ attempt=attempt + 1,
+ max_attempts=MAX_WATCH_RETRIES,
+ operation="update",
+ )
+ await asyncio.sleep(WATCH_RETRY_BASE_DELAY_SEC * (2**attempt))
return None
diff --git a/s3proxy/xml_responses.py b/s3proxy/xml_responses.py
index 0af5383..a71eaa3 100644
--- a/s3proxy/xml_responses.py
+++ b/s3proxy/xml_responses.py
@@ -370,3 +370,21 @@ def get_tagging(tags: list[dict]) -> str:
def upload_part_copy_result(etag: str, last_modified: str) -> str:
return copy_result(etag, last_modified, is_part=True)
+
+
+def without_xml_declaration(xml: str) -> str:
+ """Strip the declaration so the document stays well-formed when
+ keepalive whitespace has already been sent before it (a declaration is only
+ legal at byte 0 of the entity)."""
+ if xml.startswith("", 1)[1].lstrip("\n")
+ return xml
+
+
+def error_document(code: str, message: str) -> str:
+ """Error body emitted after a 200 OK was already committed (AWS S3 pattern
+ for long-running copies); no declaration for the same reason as above."""
+ return f"""
+ {escape(code)}
+ {escape(message)}
+"""
diff --git a/tests/unit/storage_watch_retry_fakes.py b/tests/unit/storage_watch_retry_fakes.py
new file mode 100644
index 0000000..3db8101
--- /dev/null
+++ b/tests/unit/storage_watch_retry_fakes.py
@@ -0,0 +1,51 @@
+"""Shared fakes for RedisStateStore WATCH/MULTI/EXEC retry unit tests."""
+
+from __future__ import annotations
+
+from redis.asyncio import WatchError
+
+
+class FakeWatchPipe:
+ def __init__(self, parent: FakeWatchRedis) -> None:
+ self._parent = parent
+
+ async def __aenter__(self) -> FakeWatchPipe:
+ return self
+
+ async def __aexit__(self, *exc) -> bool:
+ return False
+
+ async def watch(self, *keys) -> None:
+ pass
+
+ async def unwatch(self) -> None:
+ pass
+
+ def multi(self) -> None:
+ pass
+
+ def delete(self, *args) -> None:
+ pass
+
+ def set(self, *args, **kwargs) -> None:
+ pass
+
+ async def execute(self):
+ if self._parent.fail_remaining > 0:
+ self._parent.fail_remaining -= 1
+ raise WatchError("simulated optimistic-lock conflict")
+ return [True]
+
+
+class FakeWatchRedis:
+ def __init__(self, value: bytes | None, fail_times: int) -> None:
+ self._value = value
+ self.fail_remaining = fail_times
+ self.attempts = 0
+
+ async def get(self, key):
+ return self._value
+
+ def pipeline(self, transaction: bool = True) -> FakeWatchPipe:
+ self.attempts += 1
+ return FakeWatchPipe(self)
diff --git a/tests/unit/test_error_body_drain.py b/tests/unit/test_error_body_drain.py
new file mode 100644
index 0000000..9d2624e
--- /dev/null
+++ b/tests/unit/test_error_body_drain.py
@@ -0,0 +1,40 @@
+"""Error responses must not leave unread request bodies on keep-alive connections.
+
+A small request rejected before its body is read (e.g. UploadPart with an
+out-of-range part number) leaves the body bytes unconsumed; the client's next
+request on that connection then gets a raw uvicorn 400 (this is how
+test_part_number_out_of_range broke in CI: the abort following the rejected
+upload failed). The error handler drains small bodies. Large bodies are left
+for uvicorn to discard -- actively closing the connection instead resets
+clients that send the whole body before reading the response (covered by
+tests/integration/test_presigned_put_e2e.py::test_bad_signature_rejected).
+The end-to-end connection-reuse semantics are integration-tested; these tests
+pin the handler behavior itself.
+"""
+
+import xml.etree.ElementTree as ET
+
+from fastapi.testclient import TestClient
+
+from s3proxy import concurrency
+from s3proxy.app import create_app
+
+
+def _client(settings):
+ return TestClient(create_app(settings), raise_server_exceptions=False)
+
+
+def test_error_with_small_body_is_drained_and_keeps_connection_alive(settings):
+ with _client(settings) as client:
+ resp = client.put("/bucket/key", content=b"x" * 16)
+ assert resp.status_code == 403
+ assert "connection" not in resp.headers
+ assert ET.fromstring(resp.content).tag == "Error"
+
+
+def test_error_with_large_body_is_not_slurped_and_not_closed(settings):
+ with _client(settings) as client:
+ resp = client.put("/bucket/key", content=b"x" * (concurrency.MAX_BUFFER_SIZE + 1))
+ assert resp.status_code == 403
+ assert "connection" not in resp.headers
+ assert ET.fromstring(resp.content).tag == "Error"
diff --git a/tests/unit/test_error_mapping.py b/tests/unit/test_error_mapping.py
new file mode 100644
index 0000000..67d14a2
--- /dev/null
+++ b/tests/unit/test_error_mapping.py
@@ -0,0 +1,44 @@
+"""S3 error mapping and internal part allocation validation."""
+
+from __future__ import annotations
+
+import pytest
+from botocore.exceptions import ClientError
+
+from s3proxy import crypto
+from s3proxy.errors import S3Error, raise_for_client_error
+
+
+def _client_error(code: str, message: str = "msg") -> ClientError:
+ return ClientError(
+ {"Error": {"Code": code, "Message": message}, "ResponseMetadata": {"HTTPStatusCode": 400}},
+ "UploadPart",
+ )
+
+
+@pytest.mark.parametrize(
+ ("code", "status", "s3_code"),
+ [
+ ("InvalidPart", 400, "InvalidPart"),
+ ("InvalidPartNumber", 400, "InvalidPart"),
+ ("EntityTooSmall", 400, "EntityTooSmall"),
+ ("NoSuchUpload", 404, "NoSuchUpload"),
+ ],
+)
+def test_raise_for_client_error_maps_known_codes(code, status, s3_code):
+ with pytest.raises(S3Error) as exc:
+ raise_for_client_error(_client_error(code), bucket="b", key="k")
+ assert exc.value.status_code == status
+ assert exc.value.code == s3_code
+
+
+def test_validate_internal_part_allocation_rejects_over_s3_limit():
+ # Client part 501 needs internal parts starting at 10001.
+ with pytest.raises(ValueError, match="10001"):
+ crypto.validate_internal_part_allocation(501, 1)
+
+
+def test_validate_internal_part_allocation_allows_part_418():
+ start, end = crypto.validate_internal_part_allocation(418, 20)
+ assert start == 8341
+ assert end == 8360
diff --git a/tests/unit/test_governor_peak_and_clamp.py b/tests/unit/test_governor_peak_and_clamp.py
index 0d827ae..c650743 100644
--- a/tests/unit/test_governor_peak_and_clamp.py
+++ b/tests/unit/test_governor_peak_and_clamp.py
@@ -82,6 +82,7 @@ async def _measure_framed_peak(content_length: int) -> int:
h = _handler()
state = MultipartUploadState(dek=crypto.generate_dek(), bucket="b", key="k", upload_id="u")
part_size = crypto.memory_bounded_part_size(content_length)
+ estimated_parts = max(1, -(-content_length // part_size))
tracemalloc.start()
base = tracemalloc.get_traced_memory()[0]
await h._stream_and_upload_framed(
@@ -95,6 +96,7 @@ async def _measure_framed_peak(content_length: int) -> int:
content_length,
part_size,
1,
+ estimated_parts,
)
peak = tracemalloc.get_traced_memory()[1]
tracemalloc.stop()
diff --git a/tests/unit/test_storage_watch_retry.py b/tests/unit/test_storage_watch_retry.py
index 9717ab2..35c1c71 100644
--- a/tests/unit/test_storage_watch_retry.py
+++ b/tests/unit/test_storage_watch_retry.py
@@ -1,69 +1,14 @@
-"""RedisStateStore optimistic-locking retry (issue #66 item 3).
-
-The WATCH/MULTI/EXEC retry was rewritten from self-recursion (with a leaking
-``_retries`` kwarg) to a bounded ``for`` loop. These tests drive the loop with a
-fake pipeline whose ``execute`` raises ``WatchError`` a controlled number of
-times, pinning that it retries-then-succeeds and gives up after
-``MAX_WATCH_RETRIES`` attempts.
-"""
+"""RedisStateStore optimistic-locking retry — success after transient conflicts."""
from __future__ import annotations
-import pytest
-from redis.asyncio import WatchError
-
-from s3proxy.state.storage import MAX_WATCH_RETRIES, RedisStateStore
-
-
-class _FakePipe:
- def __init__(self, parent: _FakeRedis) -> None:
- self._parent = parent
-
- async def __aenter__(self) -> _FakePipe:
- return self
-
- async def __aexit__(self, *exc) -> bool:
- return False
-
- async def watch(self, *keys) -> None:
- pass
-
- async def unwatch(self) -> None:
- pass
-
- def multi(self) -> None:
- pass
-
- def delete(self, *args) -> None:
- pass
-
- def set(self, *args, **kwargs) -> None:
- pass
-
- async def execute(self):
- if self._parent.fail_remaining > 0:
- self._parent.fail_remaining -= 1
- raise WatchError("simulated optimistic-lock conflict")
- return [True]
-
-
-class _FakeRedis:
- def __init__(self, value: bytes | None, fail_times: int) -> None:
- self._value = value
- self.fail_remaining = fail_times
- self.attempts = 0
-
- async def get(self, key):
- return self._value
-
- def pipeline(self, transaction: bool = True) -> _FakePipe:
- self.attempts += 1
- return _FakePipe(self)
+from s3proxy.state.storage import RedisStateStore
+from tests.unit.storage_watch_retry_fakes import FakeWatchRedis
class TestWatchRetryLoop:
async def test_get_and_delete_retries_then_succeeds(self):
- fake = _FakeRedis(value=b"payload", fail_times=2)
+ fake = FakeWatchRedis(value=b"payload", fail_times=2)
store = RedisStateStore(fake)
result = await store.get_and_delete("k")
@@ -71,29 +16,11 @@ async def test_get_and_delete_retries_then_succeeds(self):
assert result == b"payload"
assert fake.attempts == 3 # two conflicts + one success
- async def test_get_and_delete_gives_up_after_max_retries(self):
- fake = _FakeRedis(value=b"payload", fail_times=MAX_WATCH_RETRIES + 5)
- store = RedisStateStore(fake)
-
- with pytest.raises(WatchError):
- await store.get_and_delete("k")
-
- assert fake.attempts == MAX_WATCH_RETRIES
-
async def test_update_retries_then_succeeds(self):
- fake = _FakeRedis(value=b"abc", fail_times=1)
+ fake = FakeWatchRedis(value=b"abc", fail_times=1)
store = RedisStateStore(fake)
result = await store.update("k", lambda data: data + b"-z", ttl_seconds=10)
assert result == b"abc-z"
assert fake.attempts == 2
-
- async def test_update_gives_up_after_max_retries(self):
- fake = _FakeRedis(value=b"abc", fail_times=MAX_WATCH_RETRIES + 5)
- store = RedisStateStore(fake)
-
- with pytest.raises(WatchError):
- await store.update("k", lambda data: data, ttl_seconds=10)
-
- assert fake.attempts == MAX_WATCH_RETRIES
diff --git a/tests/unit/test_storage_watch_retry_exhaustion.py b/tests/unit/test_storage_watch_retry_exhaustion.py
new file mode 100644
index 0000000..f6c9c7b
--- /dev/null
+++ b/tests/unit/test_storage_watch_retry_exhaustion.py
@@ -0,0 +1,42 @@
+"""RedisStateStore optimistic-locking retry — give up after MAX_WATCH_RETRIES.
+
+These drive every retry attempt; production exponential backoff would sleep for
+hours if left unmocked (0.005 * 2**attempt across 25 attempts).
+"""
+
+from __future__ import annotations
+
+import pytest
+from redis.asyncio import WatchError
+
+from s3proxy.state import storage as storage_module
+from s3proxy.state.storage import MAX_WATCH_RETRIES, RedisStateStore
+from tests.unit.storage_watch_retry_fakes import FakeWatchRedis
+
+
+@pytest.fixture(autouse=True)
+def _no_watch_retry_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
+ async def _instant(_seconds: float) -> None:
+ return None
+
+ monkeypatch.setattr(storage_module.asyncio, "sleep", _instant)
+
+
+class TestWatchRetryExhaustion:
+ async def test_get_and_delete_gives_up_after_max_retries(self):
+ fake = FakeWatchRedis(value=b"payload", fail_times=MAX_WATCH_RETRIES + 5)
+ store = RedisStateStore(fake)
+
+ with pytest.raises(WatchError):
+ await store.get_and_delete("k")
+
+ assert fake.attempts == MAX_WATCH_RETRIES
+
+ async def test_update_gives_up_after_max_retries(self):
+ fake = FakeWatchRedis(value=b"abc", fail_times=MAX_WATCH_RETRIES + 5)
+ store = RedisStateStore(fake)
+
+ with pytest.raises(WatchError):
+ await store.update("k", lambda data: data, ttl_seconds=10)
+
+ assert fake.attempts == MAX_WATCH_RETRIES
diff --git a/tests/unit/test_streaming_framed_upload.py b/tests/unit/test_streaming_framed_upload.py
index 9d5ce64..b76846e 100644
--- a/tests/unit/test_streaming_framed_upload.py
+++ b/tests/unit/test_streaming_framed_upload.py
@@ -52,6 +52,7 @@ async def upload_part(self, bucket, key, upload_id, part_number, body):
return {"ETag": f'"{part_number:032x}"'}
mgr = _Manager()
+ estimated_parts = max(1, -(-len(plaintext) // optimal))
result = await _handler(mgr)._stream_and_upload_framed(
_Request(plaintext),
_Client(),
@@ -63,6 +64,7 @@ async def upload_part(self, bucket, key, upload_id, part_number, body):
len(plaintext),
optimal,
1,
+ estimated_parts,
)
# Three internal parts of the planned sizes.
@@ -104,6 +106,8 @@ async def stream(self):
async def _measure_framed_peak(client_part_size: int) -> int:
handler = _handler(_Manager())
+ part_size = crypto.memory_bounded_part_size(client_part_size)
+ estimated_parts = max(1, -(-client_part_size // part_size))
tracemalloc.start()
tracemalloc.reset_peak()
await handler._stream_and_upload_framed(
@@ -115,8 +119,9 @@ async def _measure_framed_peak(client_part_size: int) -> int:
1,
types.SimpleNamespace(dek=crypto.generate_dek()),
client_part_size,
- crypto.memory_bounded_part_size(client_part_size), # size the handler picks
+ part_size,
1,
+ estimated_parts,
)
_, peak = tracemalloc.get_traced_memory()
tracemalloc.stop()
diff --git a/tests/unit/test_upload_part_copy_keepalive.py b/tests/unit/test_upload_part_copy_keepalive.py
new file mode 100644
index 0000000..509ea07
--- /dev/null
+++ b/tests/unit/test_upload_part_copy_keepalive.py
@@ -0,0 +1,335 @@
+"""UploadPartCopy keepalive streaming, work cancellation, and copy parallelism.
+
+Prod failure mode these guard against: a multi-GB Scylla part copy produced no
+response bytes for longer than rclone's 5-minute idle timeout, the client hung
+up at exactly 300s, the proxy kept the doomed copy running as a zombie, and
+scylla-manager retried forever. The response must start streaming keepalive
+whitespace while the copy runs, backend work must be concurrent so the copy
+finishes fast, and a client disconnect must cancel the in-flight work.
+"""
+
+import asyncio
+import contextlib
+import hashlib
+import time
+import xml.etree.ElementTree as ET
+
+import pytest
+
+from s3proxy import crypto
+from s3proxy.handlers.multipart import MultipartHandlerMixin
+from s3proxy.handlers.multipart import copy as copy_mod
+from s3proxy.state import (
+ InternalPartMetadata,
+ MultipartMetadata,
+ PartMetadata,
+ save_multipart_metadata,
+)
+
+BUCKET = "backups"
+SRC_KEY = "sst/source-Data.db"
+DST_KEY = "sst/source-Data.db.sm_manifest"
+
+
+def _copy_handler(settings, manager):
+ return MultipartHandlerMixin(settings, {}, manager)
+
+
+def _patch_client(handler, mock_s3):
+ @contextlib.asynccontextmanager
+ async def _fake_client(creds):
+ yield mock_s3
+
+ handler._client = _fake_client
+
+
+def _copy_part_request(upload_id, part_number=1, copy_source_range=None):
+ from unittest.mock import MagicMock
+
+ req = MagicMock()
+ req.url.path = f"/{BUCKET}/{DST_KEY}"
+ req.url.query = f"uploadId={upload_id}&partNumber={part_number}"
+ req.headers = {"x-amz-copy-source": f"/{BUCKET}/{SRC_KEY}"}
+ if copy_source_range is not None:
+ req.headers["x-amz-copy-source-range"] = copy_source_range
+ return req
+
+
+async def _build_encrypted_source(mock_s3, settings, credentials, *, frames, frame_size):
+ """Encrypted multipart source with one internal part per frame."""
+ kid, kek = settings.keyring.key_for(credentials.access_key)
+ src_dek = crypto.generate_dek()
+ plaintext = bytes(i % 251 for i in range(frames * frame_size))
+
+ ciphertext_blob = bytearray()
+ internal_parts_meta = []
+ for i, start in enumerate(range(0, len(plaintext), frame_size), 1):
+ chunk = plaintext[start : start + frame_size]
+ ct = crypto.encrypt_frame(chunk, src_dek, "src-upload", i, 0)
+ internal_parts_meta.append(
+ InternalPartMetadata(
+ internal_part_number=i,
+ plaintext_size=len(chunk),
+ ciphertext_size=len(ct),
+ etag=hashlib.md5(ct, usedforsecurity=False).hexdigest(),
+ )
+ )
+ ciphertext_blob.extend(ct)
+
+ src_meta = MultipartMetadata(
+ version=2,
+ part_count=1,
+ total_plaintext_size=len(plaintext),
+ parts=[
+ PartMetadata(
+ part_number=1,
+ plaintext_size=len(plaintext),
+ ciphertext_size=len(ciphertext_blob),
+ etag="ignored",
+ md5=hashlib.md5(plaintext, usedforsecurity=False).hexdigest(),
+ internal_parts=internal_parts_meta,
+ )
+ ],
+ wrapped_dek=crypto.wrap_key(src_dek, kek),
+ kid=kid,
+ )
+ await mock_s3.create_bucket(BUCKET)
+ await mock_s3.put_object(BUCKET, SRC_KEY, bytes(ciphertext_blob))
+ await save_multipart_metadata(mock_s3, BUCKET, SRC_KEY, src_meta)
+ return src_dek, plaintext, kid
+
+
+async def _start_upload(mock_s3, manager, dek, kid):
+ resp_create = await mock_s3.create_multipart_upload(BUCKET, DST_KEY)
+ upload_id = resp_create["UploadId"]
+ await manager.create_upload(BUCKET, DST_KEY, upload_id, dek, kid)
+ return upload_id
+
+
+@pytest.mark.asyncio
+async def test_keepalive_bytes_stream_while_copy_is_still_running(
+ mock_s3, settings, manager, credentials, monkeypatch
+):
+ """First response bytes must arrive BEFORE the backend copy finishes.
+
+ This is the regression that broke prod: no bytes for the whole copy
+ duration, so rclone's idle timeout killed every large part copy.
+ """
+ monkeypatch.setattr(copy_mod, "COPY_KEEPALIVE_INTERVAL", 0.02)
+ handler = _copy_handler(settings, manager)
+ _patch_client(handler, mock_s3)
+ src_dek, plaintext, kid = await _build_encrypted_source(
+ mock_s3, settings, credentials, frames=4, frame_size=256 * 1024
+ )
+ upload_id = await _start_upload(mock_s3, manager, crypto.generate_dek(), kid)
+
+ work_done = asyncio.Event()
+ orig_copy = mock_s3.upload_part_copy
+
+ async def slow_copy(*args, **kwargs):
+ await asyncio.sleep(0.1)
+ result = await orig_copy(*args, **kwargs)
+ work_done.set()
+ return result
+
+ mock_s3.upload_part_copy = slow_copy
+
+ resp = await handler.handle_upload_part_copy(_copy_part_request(upload_id), credentials)
+ assert resp.status_code == 200
+
+ stream = resp.body_iterator
+ first = await anext(stream)
+ assert first == b" "
+ assert not work_done.is_set(), "keepalive byte must be sent while the copy is in flight"
+ body = first + b"".join([c async for c in stream])
+ assert work_done.is_set()
+
+ # Whitespace + XML exactly as sent over the wire must parse.
+ root = ET.fromstring(body)
+ assert root.tag.endswith("CopyPartResult")
+ etag = root.find("{*}ETag")
+ assert etag.text.strip('"') == hashlib.md5(plaintext, usedforsecurity=False).hexdigest()
+
+ state = await manager.get_upload(BUCKET, DST_KEY, upload_id)
+ assert state.parts[1].plaintext_size == len(plaintext)
+
+
+@pytest.mark.asyncio
+async def test_copy_failure_after_200_reports_error_document(
+ mock_s3, settings, manager, credentials, monkeypatch
+):
+ """A copy that fails mid-flight must yield a parseable body, not hang."""
+ monkeypatch.setattr(copy_mod, "COPY_KEEPALIVE_INTERVAL", 0.02)
+ handler = _copy_handler(settings, manager)
+ _patch_client(handler, mock_s3)
+ _, _, kid = await _build_encrypted_source(
+ mock_s3, settings, credentials, frames=3, frame_size=256 * 1024
+ )
+ upload_id = await _start_upload(mock_s3, manager, crypto.generate_dek(), kid)
+
+ async def failing_copy(*args, **kwargs):
+ await asyncio.sleep(0.05)
+ raise RuntimeError("backend exploded")
+
+ mock_s3.upload_part_copy = failing_copy
+
+ resp = await handler.handle_upload_part_copy(_copy_part_request(upload_id), credentials)
+ body = b"".join([c async for c in resp.body_iterator])
+
+ assert resp.status_code == 200
+ root = ET.fromstring(body)
+ assert root.tag == "Error"
+ assert root.find("Code").text == "InternalError"
+ assert "backend exploded" in root.find("Message").text
+
+
+@pytest.mark.asyncio
+async def test_client_disconnect_cancels_inflight_copy_work(
+ mock_s3, settings, manager, credentials, monkeypatch
+):
+ """Closing the response stream must cancel backend work (no zombie copies)."""
+ monkeypatch.setattr(copy_mod, "COPY_KEEPALIVE_INTERVAL", 0.02)
+ handler = _copy_handler(settings, manager)
+ _patch_client(handler, mock_s3)
+ _, _, kid = await _build_encrypted_source(
+ mock_s3, settings, credentials, frames=3, frame_size=256 * 1024
+ )
+ upload_id = await _start_upload(mock_s3, manager, crypto.generate_dek(), kid)
+
+ started = asyncio.Event()
+ cancelled = asyncio.Event()
+
+ async def hanging_copy(*args, **kwargs):
+ started.set()
+ try:
+ await asyncio.sleep(3600)
+ except asyncio.CancelledError:
+ cancelled.set()
+ raise
+ raise AssertionError("copy was not cancelled")
+
+ mock_s3.upload_part_copy = hanging_copy
+
+ resp = await handler.handle_upload_part_copy(_copy_part_request(upload_id), credentials)
+ stream = resp.body_iterator
+ assert await anext(stream) == b" "
+ await asyncio.wait_for(started.wait(), 1)
+
+ await stream.aclose()
+
+ await asyncio.wait_for(cancelled.wait(), 1)
+ state = await manager.get_upload(BUCKET, DST_KEY, upload_id)
+ assert 1 not in state.parts, "aborted copy must not record a part"
+
+
+@pytest.mark.asyncio
+async def test_passthrough_segments_copy_concurrently(mock_s3, settings, manager, credentials):
+ """Segment copies must overlap; sequential backend calls pushed prod copies
+ past the client timeout even on the passthrough route."""
+ handler = _copy_handler(settings, manager)
+ _patch_client(handler, mock_s3)
+ _, plaintext, kid = await _build_encrypted_source(
+ mock_s3, settings, credentials, frames=12, frame_size=256 * 1024
+ )
+ upload_id = await _start_upload(mock_s3, manager, crypto.generate_dek(), kid)
+
+ in_flight = 0
+ max_in_flight = 0
+ orig_copy = mock_s3.upload_part_copy
+
+ async def tracking_copy(*args, **kwargs):
+ nonlocal in_flight, max_in_flight
+ in_flight += 1
+ max_in_flight = max(max_in_flight, in_flight)
+ try:
+ await asyncio.sleep(0.02)
+ return await orig_copy(*args, **kwargs)
+ finally:
+ in_flight -= 1
+
+ mock_s3.upload_part_copy = tracking_copy
+
+ resp = await handler.handle_upload_part_copy(_copy_part_request(upload_id), credentials)
+ body = b"".join([c async for c in resp.body_iterator])
+
+ assert ET.fromstring(body).tag.endswith("CopyPartResult")
+ assert max_in_flight >= 2, "segment copies ran sequentially"
+ assert max_in_flight <= copy_mod.PASSTHROUGH_SEGMENT_CONCURRENCY
+ state = await manager.get_upload(BUCKET, DST_KEY, upload_id)
+ assert state.parts[1].plaintext_size == len(plaintext)
+
+
+@pytest.mark.asyncio
+async def test_md5_source_pass_overlaps_segment_copies(mock_s3, settings, manager, credentials):
+ """The synthetic-ETag read of the source must run alongside the segment
+ copies, not serially after them (it re-reads the whole object)."""
+ handler = _copy_handler(settings, manager)
+ _patch_client(handler, mock_s3)
+ _, _, kid = await _build_encrypted_source(
+ mock_s3, settings, credentials, frames=8, frame_size=256 * 1024
+ )
+ upload_id = await _start_upload(mock_s3, manager, crypto.generate_dek(), kid)
+
+ first_md5_read: float | None = None
+ last_copy_done: float | None = None
+ orig_copy = mock_s3.upload_part_copy
+ orig_get = mock_s3.get_object
+
+ async def slow_copy(*args, **kwargs):
+ nonlocal last_copy_done
+ await asyncio.sleep(0.03)
+ result = await orig_copy(*args, **kwargs)
+ last_copy_done = time.monotonic()
+ return result
+
+ async def tracked_get(*args, **kwargs):
+ nonlocal first_md5_read
+ if first_md5_read is None:
+ first_md5_read = time.monotonic()
+ return await orig_get(*args, **kwargs)
+
+ mock_s3.upload_part_copy = slow_copy
+ mock_s3.get_object = tracked_get
+
+ resp = await handler.handle_upload_part_copy(_copy_part_request(upload_id), credentials)
+ b"".join([c async for c in resp.body_iterator])
+
+ assert first_md5_read is not None and last_copy_done is not None
+ assert first_md5_read < last_copy_done, "MD5 pass started only after all copies finished"
+
+
+@pytest.mark.asyncio
+async def test_hybrid_tail_roundtrip_with_parallel_segments(
+ mock_s3, settings, manager, credentials
+):
+ """Mid-frame range: parallel passthrough segments + re-encrypted tail must
+ reassemble to exactly the requested plaintext range, in order."""
+ handler = _copy_handler(settings, manager)
+ _patch_client(handler, mock_s3)
+ frame_size = 8 * 1024 * 1024
+ src_dek, plaintext, kid = await _build_encrypted_source(
+ mock_s3, settings, credentials, frames=6, frame_size=frame_size
+ )
+ upload_id = await _start_upload(mock_s3, manager, crypto.generate_dek(), kid)
+
+ range_end = 4 * frame_size + frame_size // 2 - 1 # ends mid 5th frame
+ assert range_end + 1 > crypto.STREAMING_THRESHOLD
+
+ resp = await handler.handle_upload_part_copy(
+ _copy_part_request(upload_id, copy_source_range=f"bytes=0-{range_end}"),
+ credentials,
+ )
+ body = b"".join([c async for c in resp.body_iterator])
+ assert ET.fromstring(body).tag.endswith("CopyPartResult")
+
+ state = await manager.get_upload(BUCKET, DST_KEY, upload_id)
+ assert state.dek == src_dek # destination adopted the source DEK
+ part = state.parts[1]
+ assert part.plaintext_size == range_end + 1
+ assert len(part.internal_parts) == 5 # 4 passthrough + 1 tail
+
+ recovered = bytearray()
+ for ip in sorted(part.internal_parts, key=lambda x: x.internal_part_number):
+ ct = mock_s3.multipart_uploads[upload_id]["Parts"][ip.internal_part_number]["Body"]
+ recovered.extend(crypto.decrypt_framed(bytes(ct), src_dek, ip.plaintext_size))
+ assert bytes(recovered) == plaintext[: range_end + 1]
diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py
index 0fae68f..8b81cf6 100644
--- a/tests/unit/test_upload_part_copy_passthrough.py
+++ b/tests/unit/test_upload_part_copy_passthrough.py
@@ -123,7 +123,7 @@ async def test_multipart_encrypted_upload_part_copy_is_server_side_passthrough(
await manager.create_upload(BUCKET, "sst/big.db.snap", upload_id, dst_dek, kid)
mark = len(mock_s3.call_history)
- await handler.handle_upload_part_copy(
+ resp = await handler.handle_upload_part_copy(
_copy_part_request(
f"/{BUCKET}/sst/big.db.snap",
f"/{BUCKET}/sst/big.db",
@@ -131,6 +131,7 @@ async def test_multipart_encrypted_upload_part_copy_is_server_side_passthrough(
),
credentials,
)
+ await _read(resp)
during = mock_s3.call_history[mark:]
assert any(c[0] == "upload_part_copy" for c in during)
@@ -204,10 +205,11 @@ async def test_upload_part_copy_passthrough_roundtrips_via_get(
)
upload_id = _extract_upload_id(create_resp.body)
- await handler.handle_upload_part_copy(
+ copy_resp = await handler.handle_upload_part_copy(
_copy_part_request(f"/{BUCKET}/sst/dest.db", f"/{BUCKET}/sst/source.db", upload_id),
credentials,
)
+ await _read(copy_resp)
part_state = await handler.multipart_manager.get_upload(BUCKET, "sst/dest.db", upload_id)
complete_body = (
@@ -269,7 +271,7 @@ async def test_range_copy_still_reencrypts(mock_s3, settings, manager, credentia
req.headers["x-amz-copy-source-range"] = f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}"
mark = len(mock_s3.call_history)
- await handler.handle_upload_part_copy(req, credentials)
+ await _read(await handler.handle_upload_part_copy(req, credentials))
during = mock_s3.call_history[mark:]
assert not any(c[0] == "upload_part_copy" for c in during)
@@ -306,17 +308,34 @@ def test_segments_for_plaintext_range_selects_aligned_prefix(settings, manager):
assert handler._segments_for_plaintext_range(segments, 0, chunk) is None
+def test_split_plaintext_range_allows_hybrid_tail(settings, manager):
+ from s3proxy.handlers.multipart.copy import _CiphertextSegment
+
+ handler = _copy_handler(settings, manager)
+ chunk = 1_048_576 # 1MB internal frames (prod uses ~8.33MB; same logic)
+ ct = chunk + 64
+ segments = [_CiphertextSegment(chunk, ct, i * ct) for i in range(10)]
+ # 9.5MB range ends mid 10th segment → 9 passthrough + tail.
+ split = handler._split_plaintext_range_on_segments(segments, 0, chunk * 10 - 1 - chunk // 2)
+ assert split is not None
+ assert len(split.passthrough_segments) == 9
+ assert split.streaming_tail == (chunk * 9, chunk * 10 - 1 - chunk // 2)
+
+
def _build_scylla_prod_shape_source(
- kid, kek, *, num_internal_parts: int, metadata_inflate_ratio: float
+ kid,
+ kek,
+ *,
+ num_internal_parts: int,
+ metadata_inflate_ratio: float,
+ chunk_size: int | None = None,
+ scylla_range_end: int | None = None,
):
- """Prod shape: metadata total_plaintext > Scylla manifest copy range.
-
- Scylla uploads ~122 client parts (~6GB metadata total) but manifest copies
- ~149 internal chunks (~4.7GB). Simulate by inflating metadata total while
- keeping only the first ``num_internal_parts`` segments real.
- """
+ """Prod shape: 8.33MB internal frames, metadata total > Scylla manifest range."""
+ if chunk_size is None:
+ # 50MB client parts / 6 internal frames (prod INTERNAL_PART_UPLOADED ~8.33MB).
+ chunk_size = (50 * 1024 * 1024) // 6
src_dek = crypto.generate_dek()
- chunk_size = crypto.MAX_BUFFER_SIZE
src_plaintext = b"M" * (chunk_size * num_internal_parts)
ciphertext_blob = bytearray()
@@ -352,28 +371,42 @@ def _build_scylla_prod_shape_source(
wrapped_dek=crypto.wrap_key(src_dek, kek),
kid=kid,
)
- return src_dek, src_plaintext, bytes(ciphertext_blob), src_meta, inflated_total
+ return (
+ src_dek,
+ src_plaintext,
+ bytes(ciphertext_blob),
+ src_meta,
+ inflated_total,
+ scylla_range_end if scylla_range_end is not None else len(src_plaintext) - 1,
+ )
@pytest.mark.asyncio
async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough(
mock_s3, settings, manager, credentials, monkeypatch
):
- """Regression: range bytes=0-(N-1) with N < metadata total must passthrough, not stream."""
+ """Regression: prod range ends mid internal frame → hybrid passthrough + tail."""
monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE)
handler = _copy_handler(settings, manager)
_patch_client(handler, mock_s3)
await mock_s3.create_bucket(BUCKET)
kid, kek = settings.keyring.key_for(credentials.access_key)
- num_parts = 149 # prod manifest internal part count
- src_dek, src_plaintext, ciphertext_blob, src_meta, inflated_total = (
+ prod_range_end = 4_999_341_931
+ chunk_size = (50 * 1024 * 1024) // 6
+ num_parts = (prod_range_end // chunk_size) + 2
+ src_dek, _, ciphertext_blob, src_meta, inflated_total, range_end = (
_build_scylla_prod_shape_source(
- kid, kek, num_internal_parts=num_parts, metadata_inflate_ratio=1.27
+ kid,
+ kek,
+ num_internal_parts=num_parts,
+ metadata_inflate_ratio=1.27,
+ chunk_size=chunk_size,
+ scylla_range_end=prod_range_end,
)
)
- assert inflated_total > len(src_plaintext)
- assert len(src_plaintext) > crypto.STREAMING_THRESHOLD
+ assert inflated_total > prod_range_end
+ assert prod_range_end + 1 > crypto.STREAMING_THRESHOLD
await mock_s3.put_object(BUCKET, "sst/big-Data.db", ciphertext_blob)
await save_multipart_metadata(mock_s3, BUCKET, "sst/big-Data.db", src_meta)
@@ -382,9 +415,9 @@ async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough(
upload_id = resp_create["UploadId"]
await manager.create_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id, src_dek, kid)
- scylla_range = f"bytes=0-{len(src_plaintext) - 1}"
+ scylla_range = f"bytes=0-{range_end}"
mark = len(mock_s3.call_history)
- await handler.handle_upload_part_copy(
+ resp = await handler.handle_upload_part_copy(
_copy_part_request(
f"/{BUCKET}/sst/big-Data.db.sm_manifest",
f"/{BUCKET}/sst/big-Data.db",
@@ -393,26 +426,36 @@ async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough(
),
credentials,
)
+ await _read(resp)
during = mock_s3.call_history[mark:]
- assert any(c[0] == "upload_part_copy" for c in during)
- assert not any(c[0] == "upload_part" for c in during)
+ copy_ops = [c for c in during if c[0] == "upload_part_copy"]
+ tail_ops = [c for c in during if c[0] == "upload_part"]
+ assert len(copy_ops) >= 500
+ assert len(tail_ops) == 1
updated = await manager.get_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id)
part = updated.parts[1]
- assert part.plaintext_size == len(src_plaintext)
- assert len(part.internal_parts) == num_parts
+ assert part.plaintext_size == prod_range_end + 1
+ assert len(part.internal_parts) == len(copy_ops) + len(tail_ops)
def test_prod_shape_passthrough_eligibility_and_route(settings, manager, credentials):
- """Unit check for prod mismatch: metadata total > range plaintext, still eligible."""
+ """Unit check for prod mismatch: metadata total > range, mid-frame tail still eligible."""
handler = _copy_handler(settings, manager)
kid, kek = settings.keyring.key_for(credentials.access_key)
- num_parts = 149
- _, src_plaintext, _, src_meta, inflated_total = _build_scylla_prod_shape_source(
- kid, kek, num_internal_parts=num_parts, metadata_inflate_ratio=1.27
+ prod_range_end = 4_999_341_931
+ chunk_size = (50 * 1024 * 1024) // 6
+ num_parts = (prod_range_end // chunk_size) + 2
+ _, _, _, src_meta, inflated_total, range_end = _build_scylla_prod_shape_source(
+ kid,
+ kek,
+ num_internal_parts=num_parts,
+ metadata_inflate_ratio=1.27,
+ chunk_size=chunk_size,
+ scylla_range_end=prod_range_end,
)
- scylla_range = f"bytes=0-{len(src_plaintext) - 1}"
+ scylla_range = f"bytes=0-{range_end}"
block = handler._passthrough_block_reason(
scylla_range,
scylla_range,
@@ -420,11 +463,11 @@ def test_prod_shape_passthrough_eligibility_and_route(settings, manager, credent
src_meta,
{},
credentials,
- len(src_plaintext),
+ prod_range_end + 1,
{},
)
assert block is None
- assert inflated_total > len(src_plaintext)
+ assert inflated_total > prod_range_end
assert (
handler._normalize_copy_source_range(scylla_range, inflated_total, {}, "wrapped-dek")
== scylla_range
@@ -547,7 +590,7 @@ async def test_scylla_manifest_full_range_uses_passthrough_not_streaming(
full_range = f"bytes=0-{len(src_plaintext) - 1}"
mark = len(mock_s3.call_history)
- await handler.handle_upload_part_copy(
+ resp = await handler.handle_upload_part_copy(
_copy_part_request(
f"/{BUCKET}/sst/big-Data.db.sm_manifest",
f"/{BUCKET}/sst/big-Data.db",
@@ -556,6 +599,7 @@ async def test_scylla_manifest_full_range_uses_passthrough_not_streaming(
),
credentials,
)
+ await _read(resp)
during = mock_s3.call_history[mark:]
assert any(c[0] == "upload_part_copy" for c in during)
@@ -605,8 +649,8 @@ async def blocked_streaming(*args, **kwargs):
handler._streaming_copy_part = blocked_streaming # type: ignore[method-assign]
- passthrough_task = asyncio.create_task(
- handler.handle_upload_part_copy(
+ async def run_passthrough():
+ resp = await handler.handle_upload_part_copy(
_copy_part_request(
f"/{BUCKET}/sst/big-Data.db.sm_manifest",
f"/{BUCKET}/sst/big-Data.db",
@@ -615,7 +659,9 @@ async def blocked_streaming(*args, **kwargs):
),
credentials,
)
- )
+ return await _read(resp)
+
+ passthrough_task = asyncio.create_task(run_passthrough())
for _ in range(200):
if passthrough_task.done():
@@ -627,6 +673,80 @@ async def blocked_streaming(*args, **kwargs):
await passthrough_task
+@pytest.mark.asyncio
+async def test_single_segment_passthrough_complete_presents_backend_etag(
+ mock_s3, settings, credentials
+):
+ """Regression: CompleteMultipartUpload must send the BACKEND part etag to S3.
+
+ The single-segment passthrough copy returns a synthetic plaintext etag to
+ the client; forwarding that echoed etag to the backend gets InvalidPart
+ from real S3/MinIO (masked in CI for a while because the resulting 500 was
+ retried by boto3 and recovered via state reconstruction).
+ """
+ import base64
+
+ handler = _handler(settings, mock_s3, credentials)
+ await mock_s3.create_bucket(BUCKET)
+
+ kid, kek = settings.keyring.key_for(credentials.access_key)
+ plaintext = b"E" * (2 * 1024 * 1024)
+ enc = crypto.encrypt_object(plaintext, kek)
+ await mock_s3.put_object(
+ BUCKET,
+ "single/src.bin",
+ enc.ciphertext,
+ metadata={
+ settings.dektag_name: base64.b64encode(enc.wrapped_dek).decode(),
+ settings.kidtag_name: kid,
+ "plaintext-size": str(len(plaintext)),
+ },
+ )
+
+ create_resp = await handler.handle_create_multipart_upload(
+ MagicMock(url=MagicMock(path=f"/{BUCKET}/single/dst.bin"), headers={}),
+ credentials,
+ )
+ upload_id = _extract_upload_id(create_resp.body)
+
+ copy_resp = await handler.handle_upload_part_copy(
+ _copy_part_request(f"/{BUCKET}/single/dst.bin", f"/{BUCKET}/single/src.bin", upload_id),
+ credentials,
+ )
+ import xml.etree.ElementTree as ET
+
+ client_etag = ET.fromstring(await _read(copy_resp)).find("{*}ETag").text.strip('"')
+ backend_etag = mock_s3.multipart_uploads[upload_id]["Parts"][1]["ETag"]
+ assert client_etag == hashlib.md5(plaintext, usedforsecurity=False).hexdigest()
+ assert client_etag != backend_etag
+
+ sent_parts = []
+ orig_complete = mock_s3.complete_multipart_upload
+
+ async def capturing_complete(bucket, key, upload_id_, parts):
+ sent_parts.extend(parts)
+ return await orig_complete(bucket, key, upload_id_, parts)
+
+ mock_s3.complete_multipart_upload = capturing_complete
+
+ complete_req = MagicMock()
+ complete_req.url.path = f"/{BUCKET}/single/dst.bin"
+ complete_req.url.query = f"uploadId={upload_id}"
+ complete_req.headers = {}
+ complete_req.body = AsyncMock(
+ return_value=(
+ f"1"
+ f'"{client_etag}"'
+ ).encode()
+ )
+ await handler.handle_complete_multipart_upload(complete_req, credentials)
+
+ assert sent_parts == [{"PartNumber": 1, "ETag": f'"{backend_etag}"'}]
+
+ resp = await handler.handle_get_object(_get_request(f"/{BUCKET}/single/dst.bin"), credentials)
+ assert await _read(resp) == plaintext
+
+
def _extract_upload_id(xml_body: bytes) -> str:
import xml.etree.ElementTree as ET
diff --git a/tests/unit/test_upload_reservation_vs_real.py b/tests/unit/test_upload_reservation_vs_real.py
index 2f35bcb..22c9fc7 100644
--- a/tests/unit/test_upload_reservation_vs_real.py
+++ b/tests/unit/test_upload_reservation_vs_real.py
@@ -59,6 +59,7 @@ async def _measure_peak(content_length):
h = _handler()
state = MultipartUploadState(dek=crypto.generate_dek(), bucket="b", key="k", upload_id="u")
part_size = crypto.memory_bounded_part_size(content_length)
+ estimated_parts = max(1, -(-content_length // part_size))
tracemalloc.start()
base = tracemalloc.get_traced_memory()[0]
await h._stream_and_upload_framed(
@@ -72,6 +73,7 @@ async def _measure_peak(content_length):
content_length,
part_size,
1,
+ estimated_parts,
)
peak = tracemalloc.get_traced_memory()[1]
tracemalloc.stop()