From 2a92e6451505ad74ea3479196b7a20029ef2c5f9 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Wed, 8 Jul 2026 22:38:24 +0200 Subject: [PATCH 1/2] fix: hybrid passthrough when Scylla range ends mid internal frame 2026.7.15 logged passthrough_blocked_reason=ranged_copy_segment_misaligned: manifest ranges (bytes=0-4999341931) end ~1.1MB into an 8.33MB encrypted frame. Copy all complete ciphertext segments server-side and re-encrypt only the trailing suffix instead of falling back to full streaming. Co-authored-by: Cursor --- s3proxy/handlers/multipart/copy.py | 105 +++++++++++++++--- .../unit/test_upload_part_copy_passthrough.py | 94 +++++++++++----- 2 files changed, 159 insertions(+), 40 deletions(-) diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index 333084c..5ab9533 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/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) @@ -263,17 +271,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 +295,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 +361,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 +597,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 +621,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 +675,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 +712,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, diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py index 0fae68f..a19a474 100644 --- a/tests/unit/test_upload_part_copy_passthrough.py +++ b/tests/unit/test_upload_part_copy_passthrough.py @@ -306,17 +306,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 +369,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,7 +413,7 @@ 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( _copy_part_request( @@ -395,24 +426,33 @@ async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough( ) 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 +460,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 From ae392445e2d35b169045185272e38424ffdf60a5 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Wed, 8 Jul 2026 22:55:42 +0200 Subject: [PATCH 2/2] fix: structured error logging and harden multipart upload failures Add request-scoped context (path, uploadId, partNumber) to MEMORY_*, REQUEST_* and S3_ERROR_RESPONSE logs so prod 4xx/5xx are diagnosable without guessing. Harden the postgres-style UploadPart path: map InvalidPart/EntityTooSmall ClientErrors to 400, reject internal part ranges past S3's 10k ceiling, detect truncated upload bodies, retry Redis WATCH conflicts with backoff (25 attempts), and raise S3 read_timeout to 300s for slow seaweed uploads. Co-authored-by: Cursor --- chart/values.yaml | 4 +- s3proxy/app.py | 9 +++ s3proxy/client/s3.py | 2 +- s3proxy/concurrency.py | 6 +- s3proxy/crypto.py | 24 ++++++++ s3proxy/errors.py | 50 ++++++++++++++- s3proxy/handlers/multipart/copy.py | 10 +++ s3proxy/handlers/multipart/upload_part.py | 75 ++++++++++++++++++++++- s3proxy/request_context.py | 42 +++++++++++++ s3proxy/request_handler.py | 32 +++++++++- s3proxy/state/manager.py | 21 +++++-- s3proxy/state/storage.py | 30 +++++++-- tests/unit/test_error_mapping.py | 44 +++++++++++++ 13 files changed, 335 insertions(+), 14 deletions(-) create mode 100644 s3proxy/request_context.py create mode 100644 tests/unit/test_error_mapping.py 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..087a5c9 100644 --- a/s3proxy/app.py +++ b/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_escape(error_code)} 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 5ab9533..cdfb3fc 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -86,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", {}) 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..7f32e00 --- /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]] = ContextVar("s3proxy_request_ctx", default={}) + + +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()) 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/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