From a55e5871ecd7a8e231fc462ee533c8e3c5bbab41 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Mon, 20 Jul 2026 13:23:38 +0200 Subject: [PATCH 1/3] fix: hybrid passthrough for Scylla backup UploadPartCopy part 2 Part 1 already passthrough-copied aligned ciphertext and deferred mid-frame tails, but part 2 was blocked at ranged_copy_nonzero_start and fell back to frame-by-frame streaming (~190 Hetzner GETs), causing InvalidPart under load. Mirror part 1's hybrid path for nonzero ranges: consume the deferred tail, re-encrypt only the misaligned head, and server-side copy the aligned bulk. Co-authored-by: Cursor --- s3proxy/handlers/multipart/copy.py | 117 ++++++++++++++---- .../unit/test_upload_part_copy_passthrough.py | 89 ++++++++++++- 2 files changed, 176 insertions(+), 30 deletions(-) diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index a53ade9..af57a80 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -80,10 +80,11 @@ class _CiphertextSegment: @dataclass(frozen=True, slots=True) class _PlaintextRangeSplit: - """Passthrough-safe prefix segments plus optional plaintext tail to re-encrypt.""" + """Passthrough-safe segments plus optional plaintext head/tail to re-encrypt.""" passthrough_segments: tuple[_CiphertextSegment, ...] streaming_tail: tuple[int, int] | None # inclusive plaintext [start, end] + streaming_head: tuple[int, int] | None = None # inclusive plaintext [start, end] class CopyPartMixin(BaseHandler): @@ -397,16 +398,21 @@ def _split_plaintext_range_on_segments( range_start: int, range_end: int, ) -> _PlaintextRangeSplit | None: - """Split a plaintext range into ciphertext-passthrough prefix + optional streaming tail. + """Split a plaintext range into passthrough segments + optional head/tail to re-encrypt. - 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. + Scylla manifest part 1 often ends mid internal frame (~8.33MB from 50MB uploads): + passthrough aligned prefix, defer or stream the trailing suffix. Part 2 starts + mid-frame after part 1: stream/defer the head prefix, passthrough the rest. """ + if not segments: + return None + range_end = min(range_end, self._source_plaintext_end(segments)) if range_start > range_end: - return _PlaintextRangeSplit((), None) + return _PlaintextRangeSplit((), None, None) selected: list[_CiphertextSegment] = [] pt_offset = 0 streaming_tail: tuple[int, int] | None = None + streaming_head: tuple[int, int] | None = None for seg in segments: seg_start = pt_offset seg_end = pt_offset + seg.plaintext_size - 1 @@ -416,15 +422,22 @@ def _split_plaintext_range_on_segments( if seg_start > range_end: break if seg_start < range_start: - return None + head_end = min(seg_end, range_end) + streaming_head = (range_start, head_end) + if head_end >= range_end: + return _PlaintextRangeSplit((), None, streaming_head) + pt_offset += seg.plaintext_size + continue if seg_end > range_end: streaming_tail = (seg_start, range_end) break selected.append(seg) pt_offset += seg.plaintext_size - if streaming_tail is None and pt_offset <= range_end: + if streaming_tail is None and streaming_head is None and pt_offset <= range_end: + return None + if not selected and not streaming_tail and not streaming_head: return None - return _PlaintextRangeSplit(tuple(selected), streaming_tail) + return _PlaintextRangeSplit(tuple(selected), streaming_tail, streaming_head) def _source_plaintext_end(self, segments: list[_CiphertextSegment]) -> int: if not segments: @@ -477,7 +490,7 @@ def _segments_for_plaintext_range( 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: + if split.streaming_tail is not None or split.streaming_head is not None: return None return list(split.passthrough_segments) @@ -516,8 +529,6 @@ def _passthrough_block_reason( total = src_multipart_meta.total_plaintext_size range_start, range_end = self._parse_copy_source_range(copy_source_range, total) - if range_start != 0: - return "ranged_copy_nonzero_start" segments = self._source_ciphertext_segments( src_multipart_meta, head_resp, src_wrapped_dek, src_metadata @@ -525,13 +536,18 @@ def _passthrough_block_reason( split = self._split_plaintext_range_on_segments(segments, range_start, range_end) if split is None: return "ranged_copy_segment_misaligned" - if not split.passthrough_segments and not split.streaming_tail: + if not split.passthrough_segments and not split.streaming_tail and not split.streaming_head: 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: + if not split.passthrough_segments: + edge_bytes = 0 + if split.streaming_tail: + edge_bytes += split.streaming_tail[1] - split.streaming_tail[0] + 1 + if split.streaming_head: + edge_bytes += split.streaming_head[1] - split.streaming_head[0] + 1 + if edge_bytes <= crypto.STREAMING_THRESHOLD: return "small_ranged_copy" - return "ranged_copy_segment_misaligned" + if split.streaming_head and split.streaming_tail: + return "ranged_copy_segment_misaligned" return None def _log_upload_part_copy_route( @@ -756,6 +772,10 @@ async def _passthrough_copy_part( all_segments = segments range_end = 0 + streaming_head: tuple[int, int] | None = None + deferred_tail_bytes = await self.multipart_manager.take_deferred_copy_tail( + bucket, key, upload_id + ) if copy_source_range and src_multipart_meta: range_start, range_end = self._parse_copy_source_range( copy_source_range, src_multipart_meta.total_plaintext_size @@ -765,6 +785,7 @@ async def _passthrough_copy_part( raise S3Error.invalid_request("Copy range splits an encrypted segment") segments = list(split.passthrough_segments) streaming_tail = split.streaming_tail + streaming_head = split.streaming_head defer_tail = self._should_defer_hybrid_tail( streaming_tail, range_end, @@ -782,6 +803,11 @@ async def _passthrough_copy_part( if streaming_tail else None ) + head_mb = ( + f"{(streaming_head[1] - streaming_head[0] + 1) / 1024 / 1024:.2f}MB" + if streaming_head + else None + ) logger.info( "UPLOAD_PART_COPY_PASSTHROUGH", @@ -792,6 +818,8 @@ async def _passthrough_copy_part( src_key=src_key, plaintext_mb=f"{plaintext_size / 1024 / 1024:.2f}MB", segments=len(segments), + streaming_head_mb=head_mb, + deferred_head_bytes=len(deferred_tail_bytes) if deferred_tail_bytes else 0, streaming_tail_mb=tail_mb, defer_tail=defer_tail, copy_source_range=copy_source_range, @@ -849,12 +877,13 @@ async def _passthrough_copy_part( media_type="application/xml", ) + head_internal_slots = 1 if (streaming_head or deferred_tail_bytes) else 0 tail_internal_slots = 0 if defer_tail else (1 if streaming_tail else 0) internal_part_start = await self.multipart_manager.allocate_internal_parts( bucket, key, upload_id, - len(segments) + tail_internal_slots, + len(segments) + head_internal_slots + tail_internal_slots, client_part_number=0, ) logger.info( @@ -863,16 +892,48 @@ async def _passthrough_copy_part( key=key, part_num=part_num, passthrough_segments=len(segments), + head_internal_slots=head_internal_slots, tail_internal_slots=tail_internal_slots, internal_part_start=internal_part_start, ) start = time.monotonic() segment_sem = asyncio.Semaphore(PASSTHROUGH_SEGMENT_CONCURRENCY) - deferred_tail_bytes: bytes | None = None + deferred_tail_for_next: bytes | None = None + + async def upload_head() -> InternalPartMetadata | None: + if not streaming_head and not deferred_tail_bytes: + return None + head_suffix = b"" + if streaming_head: + head_start, head_end = streaming_head + head_suffix = await self._download_encrypted_multipart( + client, src_bucket, src_key, src_multipart_meta, head_start, head_end + ) + head_plaintext = (deferred_tail_bytes or b"") + head_suffix + internal_num = internal_part_start + head_ct = crypto.encrypt_frame(head_plaintext, state.dek, upload_id, internal_num, 0) + part_reserve = crypto.copy_chunk_peak(len(head_plaintext)) + async with concurrency.reserve_copy_memory(part_reserve): + resp = await client.upload_part(bucket, key, upload_id, internal_num, head_ct) + logger.info( + "UPLOAD_PART_COPY_PASSTHROUGH_HEAD", + bucket=bucket, + key=key, + part_num=part_num, + internal_part=internal_num, + head_plaintext_mb=f"{len(head_plaintext) / 1024 / 1024:.2f}MB", + deferred_head_bytes=len(deferred_tail_bytes) if deferred_tail_bytes else 0, + ) + return InternalPartMetadata( + internal_part_number=internal_num, + plaintext_size=len(head_plaintext), + ciphertext_size=len(head_ct), + etag=resp["ETag"].strip('"'), + ) async def copy_segment(idx: int, seg: _CiphertextSegment) -> InternalPartMetadata: - internal_num = internal_part_start + idx + internal_num = internal_part_start + head_internal_slots + 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) @@ -915,7 +976,7 @@ async def copy_segment(idx: int, seg: _CiphertextSegment) -> InternalPartMetadat ) async def upload_tail() -> InternalPartMetadata | None: - nonlocal deferred_tail_bytes + nonlocal deferred_tail_for_next if not (streaming_tail and src_multipart_meta): return None tail_start, tail_end = streaming_tail @@ -923,7 +984,7 @@ async def upload_tail() -> InternalPartMetadata | None: client, src_bucket, src_key, src_multipart_meta, tail_start, tail_end ) if defer_tail: - deferred_tail_bytes = tail_plaintext + deferred_tail_for_next = tail_plaintext logger.info( "UPLOAD_PART_COPY_PASSTHROUGH_TAIL_DEFERRED", bucket=bucket, @@ -932,7 +993,7 @@ async def upload_tail() -> InternalPartMetadata | None: tail_plaintext_mb=f"{len(tail_plaintext) / 1024 / 1024:.2f}MB", ) return None - internal_num = internal_part_start + len(segments) + internal_num = internal_part_start + head_internal_slots + 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): @@ -958,6 +1019,7 @@ async def upload_tail() -> InternalPartMetadata | None: # idle timeout. try: async with asyncio.TaskGroup() as tg: + head_task = tg.create_task(upload_head()) 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( @@ -978,12 +1040,15 @@ async def upload_tail() -> InternalPartMetadata | None: exc = exc.exceptions[0] raise exc from eg - internal_parts = [t.result() for t in seg_tasks] + internal_parts: list[InternalPartMetadata] = [] + if (head_part := head_task.result()) is not None: + internal_parts.append(head_part) + internal_parts.extend(t.result() for t in seg_tasks) if (tail_part := tail_task.result()) is not None: internal_parts.append(tail_part) - if deferred_tail_bytes: + if deferred_tail_for_next: await self.multipart_manager.set_deferred_copy_tail( - bucket, key, upload_id, deferred_tail_bytes + bucket, key, upload_id, deferred_tail_for_next ) # A deferred tail is NOT part of this client part's stored bytes: it is # folded into the next part's stream (take_deferred_copy_tail) or flushed @@ -1014,7 +1079,7 @@ async def upload_tail() -> InternalPartMetadata | None: key=key, part_num=part_num, internal_parts=len(internal_parts), - deferred_tail_bytes=len(deferred_tail_bytes) if deferred_tail_bytes else 0, + deferred_tail_bytes=len(deferred_tail_for_next) if deferred_tail_for_next else 0, plaintext_mb=f"{total_plaintext / 1024 / 1024:.2f}MB", copy_source_range=copy_source_range, elapsed_sec=f"{time.monotonic() - start:.2f}s", diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py index 5630828..de0f798 100644 --- a/tests/unit/test_upload_part_copy_passthrough.py +++ b/tests/unit/test_upload_part_copy_passthrough.py @@ -320,6 +320,79 @@ def test_split_plaintext_range_allows_hybrid_tail(settings, manager): assert split is not None assert len(split.passthrough_segments) == 9 assert split.streaming_tail == (chunk * 9, chunk * 10 - 1 - chunk // 2) + assert split.streaming_head is None + + +def test_split_plaintext_range_allows_hybrid_head(settings, manager): + """Part 2 starts mid internal frame: stream head, passthrough the rest.""" + from s3proxy.handlers.multipart.copy import _CiphertextSegment + + handler = _copy_handler(settings, manager) + chunk = 1_048_576 + ct = chunk + 64 + segments = [_CiphertextSegment(chunk, ct, i * ct) for i in range(10)] + part1_end = chunk * 5 + chunk // 2 + part2_start = part1_end + 1 + part2_end = chunk * 10 - 1 + split = handler._split_plaintext_range_on_segments(segments, part2_start, part2_end) + assert split is not None + assert split.streaming_head == (part2_start, chunk * 6 - 1) + assert len(split.passthrough_segments) == 4 + assert split.streaming_tail is None + + +def test_prod_shape_part2_split_has_head_and_bulk_passthrough(settings, manager): + """Prod byte offsets: part 2 starts mid-frame, bulk is passthrough-safe.""" + from s3proxy.handlers.multipart.copy import _CiphertextSegment + + handler = _copy_handler(settings, manager) + chunk_size = (50 * 1024 * 1024) // 6 + ct = chunk_size + 28 + prod_part1_end = 4_999_341_931 + prod_part2_start = prod_part1_end + 1 + num_segments = (prod_part2_start // chunk_size) + 200 + segments = [_CiphertextSegment(chunk_size, ct, i * ct) for i in range(num_segments)] + source_end = num_segments * chunk_size - 1 + split = handler._split_plaintext_range_on_segments(segments, prod_part2_start, source_end) + assert split is not None + assert split.streaming_head is not None + assert len(split.passthrough_segments) > 100 + head_bytes = split.streaming_head[1] - split.streaming_head[0] + 1 + assert head_bytes < chunk_size + assert prod_part2_start % chunk_size != 0 + + +def test_prod_shape_part2_nonzero_start_eligible_for_passthrough(settings, manager, credentials): + """Regression: prod part 2 must not hit ranged_copy_nonzero_start.""" + handler = _copy_handler(settings, manager) + kid, kek = settings.keyring.key_for(credentials.access_key) + prod_part1_end = 4_999_341_931 + prod_part2_start = prod_part1_end + 1 + chunk_size = (50 * 1024 * 1024) // 6 + num_parts = (prod_part1_end // chunk_size) + 3 + _, src_plaintext, _, src_meta, inflated_total, _ = _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_part1_end, + ) + # Prod part 2 range uses HEAD/metadata size (inflated), not only stored segments. + part2_range = f"bytes={prod_part2_start}-{inflated_total - 1}" + plaintext_size = inflated_total - prod_part2_start + assert plaintext_size > crypto.STREAMING_THRESHOLD + block = handler._passthrough_block_reason( + part2_range, + part2_range, + "wrapped-dek", + src_meta, + {}, + credentials, + plaintext_size, + {}, + ) + assert block is None, f"expected passthrough, got {block}" def _build_scylla_prod_shape_source( @@ -537,8 +610,8 @@ async def test_two_part_hybrid_defer_tail_completes_fast( kid, kek = settings.keyring.key_for(credentials.access_key) chunk = 8 * 1024 * 1024 # Part 1 must exceed STREAMING_THRESHOLD (32MB) for hybrid passthrough; part 2 - # must also exceed it so streaming consumes the deferred tail. Internal frames - # must be >= MIN_PART_SIZE so passthrough segments are valid S3 parts. + # must also exceed it so hybrid passthrough consumes the deferred head. Internal + # frames must be >= MIN_PART_SIZE so passthrough segments are valid S3 parts. num_segments = 12 part1_range_end = chunk * 4 + chunk // 2 # 36MB, ~4MB deferred tail src_dek, src_plaintext, ciphertext_blob, src_meta, inflated_total, _ = ( @@ -587,6 +660,7 @@ async def test_two_part_hybrid_defer_tail_completes_fast( assert len(deferred) < crypto.MIN_PART_SIZE part2_start = part1_range_end + 1 + mark2 = len(mock_s3.call_history) resp2 = await handler.handle_upload_part_copy( _copy_part_request( f"/{BUCKET}/sst/small-Data.db.sm_manifest", @@ -598,6 +672,13 @@ async def test_two_part_hybrid_defer_tail_completes_fast( credentials, ) await _read(resp2) + part2_ops = mock_s3.call_history[mark2:] + copy_ops = [c for c in part2_ops if c[0] == "upload_part_copy"] + upload_ops = [c for c in part2_ops if c[0] == "upload_part"] + assert len(copy_ops) >= 3, "part 2 bulk must passthrough aligned segments" + assert len(upload_ops) == 1, "part 2 should upload one hybrid head, not stream every frame" + # MD5 etag computation still frame-reads the range in parallel; the regression + # is avoiding a streaming-only path that upload_part's every internal chunk. after_part2 = await handler.multipart_manager.get_upload( BUCKET, "sst/small-Data.db.sm_manifest", upload_id @@ -648,8 +729,8 @@ async def test_two_part_defer_tail_total_plaintext_not_double_counted( ): """Regression: rclone 2-part copy (>copy-cutoff SSTable) reported size+tail. - Part 1 defers its sub-5MB tail; part 2's streaming copy folds the tail into - its own stream and counts it. Part 1's metadata must therefore NOT count the + Part 1 defers its sub-5MB tail; part 2's hybrid passthrough folds the tail into + the head internal part and counts it. Part 1's metadata must therefore NOT count the tail too, or HEAD Content-Length comes out exactly one tail too large and rclone fails the copy with "corrupted on transfer: sizes differ" (398 files x +1,129,664 bytes, backup run sm_20260709080013UTC). From 5437324140701e20c4c64913ea596bec94a167ab Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Mon, 20 Jul 2026 13:30:29 +0200 Subject: [PATCH 2/3] chore: add hybrid split diagnostics for UploadPartCopy routing Log hybrid_mode, passthrough segment counts, and estimated frame-read savings on UPLOAD_PART_COPY_ROUTE. Warn when part 2 still falls back to streaming so prod validation is obvious from logs alone. Co-authored-by: Cursor --- s3proxy/handlers/multipart/copy.py | 89 +++++++++++++++++++ .../unit/test_upload_part_copy_passthrough.py | 31 +++++++ 2 files changed, 120 insertions(+) diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index af57a80..280b1a2 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -149,6 +149,13 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) if passthrough_block is None else ("streaming" if plaintext_size > crypto.STREAMING_THRESHOLD else "simple") ) + hybrid_split = self._hybrid_split_log_fields( + copy_source_range, + src_wrapped_dek, + src_multipart_meta, + src_metadata, + head_resp, + ) self._log_upload_part_copy_route( bucket=bucket, key=key, @@ -159,7 +166,23 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials) range_plaintext_size=plaintext_size, route=route, passthrough_blocked_reason=passthrough_block, + hybrid_split=hybrid_split, ) + if ( + route == "streaming" + and part_num > 1 + and plaintext_size > crypto.STREAMING_THRESHOLD + ): + logger.warning( + "UPLOAD_PART_COPY_PART2_STREAMING_FALLBACK", + bucket=bucket, + key=key, + part_num=part_num, + raw_copy_source_range=raw_copy_source_range, + passthrough_blocked_reason=passthrough_block, + range_plaintext_mb=f"{plaintext_size / 1024 / 1024:.2f}MB", + **(hybrid_split or {}), + ) # Validation and routing are done with real HTTP status semantics; the # copy work itself runs while the response streams keepalive whitespace, @@ -550,6 +573,61 @@ def _passthrough_block_reason( return "ranged_copy_segment_misaligned" return None + def _hybrid_split_log_fields( + self, + copy_source_range: str | None, + src_wrapped_dek: str | None, + src_multipart_meta: MultipartMetadata | None, + src_metadata: dict, + head_resp: dict, + ) -> dict[str, str | int] | None: + """Structured split breakdown for route/passthrough diagnostics.""" + if not copy_source_range or not src_multipart_meta: + return None + total = src_multipart_meta.total_plaintext_size + range_start, range_end = self._parse_copy_source_range(copy_source_range, total) + segments = self._source_ciphertext_segments( + src_multipart_meta, head_resp, src_wrapped_dek, src_metadata + ) + split = self._split_plaintext_range_on_segments(segments, range_start, range_end) + if split is None: + return { + "hybrid_mode": "split_failed", + "range_start": range_start, + "range_end": range_end, + } + head_bytes = ( + split.streaming_head[1] - split.streaming_head[0] + 1 if split.streaming_head else 0 + ) + tail_bytes = ( + split.streaming_tail[1] - split.streaming_tail[0] + 1 if split.streaming_tail else 0 + ) + passthrough_bytes = sum(seg.plaintext_size for seg in split.passthrough_segments) + if split.streaming_head and split.streaming_tail: + mode = "head_and_tail" + elif split.streaming_head: + mode = "head_and_passthrough" + elif split.streaming_tail: + mode = "passthrough_and_tail" + elif split.passthrough_segments: + mode = "passthrough_only" + else: + mode = "stream_only" + chunk = split.passthrough_segments[0].plaintext_size if split.passthrough_segments else 0 + frame_reads_if_streaming = ( + (range_end - range_start + 1) // chunk + 1 if chunk else 0 + ) + return { + "hybrid_mode": mode, + "range_start": range_start, + "range_end": range_end, + "passthrough_segments": len(split.passthrough_segments), + "passthrough_mb": f"{passthrough_bytes / 1024 / 1024:.2f}MB", + "streaming_head_mb": f"{head_bytes / 1024 / 1024:.2f}MB", + "streaming_tail_mb": f"{tail_bytes / 1024 / 1024:.2f}MB", + "estimated_frame_reads_if_streaming": frame_reads_if_streaming, + } + def _log_upload_part_copy_route( self, *, @@ -562,6 +640,7 @@ def _log_upload_part_copy_route( range_plaintext_size: int, route: str, passthrough_blocked_reason: str | None, + hybrid_split: dict[str, str | int] | None = None, ) -> None: logger.info( "UPLOAD_PART_COPY_ROUTE", @@ -574,6 +653,7 @@ def _log_upload_part_copy_route( range_plaintext_mb=f"{range_plaintext_size / 1024 / 1024:.2f}MB", route=route, passthrough_blocked_reason=passthrough_blocked_reason, + **(hybrid_split or {}), ) def _can_passthrough_part_copy( @@ -1073,15 +1153,24 @@ async def upload_tail() -> InternalPartMetadata | None: ), ) + passthrough_segment_count = len(segments) + head_plaintext_bytes = internal_parts[0].plaintext_size if head_internal_slots else 0 logger.info( "UPLOAD_PART_COPY_PASSTHROUGH_COMPLETE", bucket=bucket, key=key, part_num=part_num, internal_parts=len(internal_parts), + passthrough_segments=passthrough_segment_count, + head_plaintext_mb=f"{head_plaintext_bytes / 1024 / 1024:.2f}MB", deferred_tail_bytes=len(deferred_tail_for_next) if deferred_tail_for_next else 0, plaintext_mb=f"{total_plaintext / 1024 / 1024:.2f}MB", copy_source_range=copy_source_range, + hybrid_mode=( + "head_and_passthrough" + if streaming_head + else ("passthrough_and_tail" if streaming_tail else "passthrough_only") + ), elapsed_sec=f"{time.monotonic() - start:.2f}s", ) return Response( diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py index de0f798..7253b75 100644 --- a/tests/unit/test_upload_part_copy_passthrough.py +++ b/tests/unit/test_upload_part_copy_passthrough.py @@ -362,6 +362,37 @@ def test_prod_shape_part2_split_has_head_and_bulk_passthrough(settings, manager) assert prod_part2_start % chunk_size != 0 +def test_hybrid_split_log_fields_part2_shape(settings, manager, credentials): + """Route diagnostics must show head+passthrough for prod part 2 offsets.""" + handler = _copy_handler(settings, manager) + kid, kek = settings.keyring.key_for(credentials.access_key) + prod_part1_end = 4_999_341_931 + prod_part2_start = prod_part1_end + 1 + chunk_size = (50 * 1024 * 1024) // 6 + num_parts = (prod_part1_end // chunk_size) + 3 + _, src_plaintext, _, src_meta, inflated_total, _ = _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_part1_end, + ) + part2_range = f"bytes={prod_part2_start}-{inflated_total - 1}" + fields = handler._hybrid_split_log_fields( + part2_range, + "wrapped-dek", + src_meta, + {}, + {}, + ) + assert fields is not None + assert fields["hybrid_mode"] == "head_and_passthrough" + assert fields["range_start"] == prod_part2_start + assert fields["passthrough_segments"] >= 1 + assert int(fields["estimated_frame_reads_if_streaming"]) > int(fields["passthrough_segments"]) + + def test_prod_shape_part2_nonzero_start_eligible_for_passthrough(settings, manager, credentials): """Regression: prod part 2 must not hit ranged_copy_nonzero_start.""" handler = _copy_handler(settings, manager) From d90b5f750df08eab2a7604a3610d3475c54825b3 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Mon, 20 Jul 2026 13:32:41 +0200 Subject: [PATCH 3/3] style: ruff format copy.py (fix CI) Co-authored-by: Cursor --- s3proxy/handlers/multipart/copy.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index 280b1a2..f796de6 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -614,9 +614,7 @@ def _hybrid_split_log_fields( else: mode = "stream_only" chunk = split.passthrough_segments[0].plaintext_size if split.passthrough_segments else 0 - frame_reads_if_streaming = ( - (range_end - range_start + 1) // chunk + 1 if chunk else 0 - ) + frame_reads_if_streaming = (range_end - range_start + 1) // chunk + 1 if chunk else 0 return { "hybrid_mode": mode, "range_start": range_start,