diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index a59ba4f..3d5f97c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -28,6 +28,27 @@ jobs: - name: Run unit tests run: make test-unit + unit-slow: + runs-on: ubuntu-latest + timeout-minutes: 30 + steps: + - uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v6 + with: + python-version: '3.14' + cache: 'pip' + + - name: Install uv + run: pip install uv + + - name: Install dependencies + run: uv sync --extra dev + + - name: Run slow unit tests + run: make test-unit-slow + integration: runs-on: ubuntu-latest strategy: diff --git a/Makefile b/Makefile index 0fa418f..4f85617 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test test-all test-unit test-integration test-integration-shard test-run test-oom verify-copy-memory verify-passthrough e2e cluster lint +.PHONY: test test-all test-unit test-unit-slow test-integration test-integration-shard test-run test-oom verify-copy-memory verify-passthrough e2e cluster lint # Lint: ruff check + format check lint: @@ -8,9 +8,13 @@ lint: # Default: run unit tests only (no containers needed) test: test-unit -# Run unit tests (excludes e2e and ha tests) +# Run unit tests (excludes e2e, ha, and slow tests) test-unit: - uv run pytest -m "not e2e and not ha" -v -n auto + uv run pytest -m "not e2e and not ha and not slow" -v -n auto + +# Prod-scale passthrough tests (~500+ segments); serial to avoid xdist + CI timeouts. +test-unit-slow: + uv run pytest -m "slow" -v -n0 # Pre-merge gate: subprocess Prometheus proof of per-part copy memory release. verify-copy-memory: diff --git a/pyproject.toml b/pyproject.toml index 97795ba..7286d94 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -76,6 +76,7 @@ testpaths = ["tests"] markers = [ "e2e: End-to-end integration tests requiring real S3/MinIO (deselect with '-m \"not e2e\"')", "ha: HA tests with multiple s3proxy pods and real Redis (deselect with '-m \"not ha\"')", + "slow: Large mock-S3 tests (500+ segment passthrough); run serially with '-m slow -n0'", ] [tool.mypy] diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index 3160dd4..ef03c0c 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -55,6 +55,10 @@ os.environ.get("S3PROXY_PASSTHROUGH_SEGMENT_CONCURRENCY", "8") ) +# Scylla/rclone manifest part 1 is ~4.7GB; only defer a sub-5MB hybrid tail when +# part 1 is large enough that a client part 2 will follow (avoids EntityTooSmall). +HYBRID_TAIL_DEFER_MIN_CLIENT_PART = 1024 * 1024 * 1024 # 1 GiB + def reset_copy_pipeline_semaphore(limit: int | None = None) -> None: """Reset the global copy pipeline semaphore (testing only).""" @@ -421,6 +425,47 @@ def _split_plaintext_range_on_segments( return None return _PlaintextRangeSplit(tuple(selected), streaming_tail) + def _source_plaintext_end(self, segments: list[_CiphertextSegment]) -> int: + if not segments: + return -1 + return sum(seg.plaintext_size for seg in segments) - 1 + + def _should_defer_hybrid_tail( + self, + streaming_tail: tuple[int, int] | None, + range_end: int, + all_segments: list[_CiphertextSegment], + *, + client_part_plaintext_size: int, + part_num: int, + ) -> bool: + """Defer a sub-5MB hybrid tail when a large client part 1 precedes part 2. + + S3 requires every internal part except the last to be >= 5MB. Scylla + manifest part 1 (~4.7GB) often ends mid internal frame (~1MB tail) + before part 2. Small single-part partial copies still upload the tail + immediately so the client part is self-contained. + """ + if streaming_tail is None or part_num != 1: + return False + if range_end >= self._source_plaintext_end(all_segments): + return False + tail_bytes = streaming_tail[1] - streaming_tail[0] + 1 + if tail_bytes >= crypto.MIN_PART_SIZE: + return False + defer = client_part_plaintext_size >= HYBRID_TAIL_DEFER_MIN_CLIENT_PART + logger.info( + "HYBRID_TAIL_DEFER_DECISION", + defer=defer, + part_num=part_num, + tail_bytes=tail_bytes, + client_part_plaintext_mb=f"{client_part_plaintext_size / 1024 / 1024:.2f}MB", + range_end=range_end, + source_end=self._source_plaintext_end(all_segments), + min_client_part_mb=f"{HYBRID_TAIL_DEFER_MIN_CLIENT_PART / 1024 / 1024:.0f}MB", + ) + return defer + def _segments_for_plaintext_range( self, segments: list[_CiphertextSegment], @@ -708,6 +753,8 @@ async def _passthrough_copy_part( if not segments: raise S3Error.invalid_request("Copy source has no ciphertext segments") + all_segments = segments + range_end = 0 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 @@ -717,8 +764,16 @@ 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 + defer_tail = self._should_defer_hybrid_tail( + streaming_tail, + range_end, + all_segments, + client_part_plaintext_size=plaintext_size, + part_num=part_num, + ) else: streaming_tail = None + defer_tail = False copy_source_path = copy_source or f"/{src_bucket}/{quote(src_key, safe='/')}" tail_mb = ( @@ -737,6 +792,7 @@ async def _passthrough_copy_part( plaintext_mb=f"{plaintext_size / 1024 / 1024:.2f}MB", segments=len(segments), streaming_tail_mb=tail_mb, + defer_tail=defer_tail, copy_source_range=copy_source_range, ) @@ -792,16 +848,27 @@ async def _passthrough_copy_part( media_type="application/xml", ) + 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) + (1 if streaming_tail else 0), + len(segments) + tail_internal_slots, client_part_number=0, ) + logger.info( + "UPLOAD_PART_COPY_PASSTHROUGH_ALLOC", + bucket=bucket, + key=key, + part_num=part_num, + passthrough_segments=len(segments), + 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 async def copy_segment(idx: int, seg: _CiphertextSegment) -> InternalPartMetadata: internal_num = internal_part_start + idx @@ -825,12 +892,23 @@ async def copy_segment(idx: int, seg: _CiphertextSegment) -> InternalPartMetadat ) async def upload_tail() -> InternalPartMetadata | None: + nonlocal deferred_tail_bytes 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 ) + if defer_tail: + deferred_tail_bytes = tail_plaintext + logger.info( + "UPLOAD_PART_COPY_PASSTHROUGH_TAIL_DEFERRED", + bucket=bucket, + key=key, + part_num=part_num, + tail_plaintext_mb=f"{len(tail_plaintext) / 1024 / 1024:.2f}MB", + ) + return None 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)) @@ -880,7 +958,13 @@ async def upload_tail() -> InternalPartMetadata | None: internal_parts = [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: + await self.multipart_manager.set_deferred_copy_tail( + bucket, key, upload_id, deferred_tail_bytes + ) total_plaintext = sum(p.plaintext_size for p in internal_parts) + if deferred_tail_bytes: + total_plaintext += len(deferred_tail_bytes) total_ciphertext = sum(p.ciphertext_size for p in internal_parts) etag = md5_task.result().hexdigest() await self.multipart_manager.add_part( @@ -903,6 +987,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, plaintext_mb=f"{total_plaintext / 1024 / 1024:.2f}MB", copy_source_range=copy_source_range, elapsed_sec=f"{time.monotonic() - start:.2f}s", @@ -912,6 +997,46 @@ async def upload_tail() -> InternalPartMetadata | None: media_type="application/xml", ) + async def _flush_deferred_copy_tail_for_complete( + self, + client: S3Client, + bucket: str, + key: str, + upload_id: str, + state: MultipartUploadState, + ) -> MultipartUploadState: + """Upload a leftover deferred hybrid tail as the final internal S3 part.""" + tail = state.deferred_copy_tail + if not tail: + return state + + internal_num = state.next_internal_part_number + tail_ct = crypto.encrypt_frame(tail, state.dek, upload_id, internal_num, 0) + resp = await client.upload_part(bucket, key, upload_id, internal_num, tail_ct) + ip = InternalPartMetadata( + internal_part_number=internal_num, + plaintext_size=len(tail), + ciphertext_size=len(tail_ct), + etag=resp["ETag"].strip('"'), + ) + last_pn = max(state.parts) + last_part = state.parts[last_pn] + last_part.internal_parts.append(ip) + last_part.ciphertext_size += len(tail_ct) + state.deferred_copy_tail = b"" + state.next_internal_part_number = internal_num + 1 + + logger.info( + "DEFERRED_COPY_TAIL_FLUSHED_ON_COMPLETE", + bucket=bucket, + key=key, + upload_id=upload_id[:20] + "...", + client_part=last_pn, + internal_part=internal_num, + tail_plaintext_mb=f"{len(tail) / 1024 / 1024:.2f}MB", + ) + return state + async def _simple_copy_part( self, client: S3Client, @@ -1058,6 +1183,17 @@ async def _streaming_copy_part_inner( chunk_size = crypto.copy_internal_part_size(plaintext_size) estimated_parts = max(1, math.ceil(plaintext_size / chunk_size)) + deferred_tail = await self.multipart_manager.take_deferred_copy_tail(bucket, key, upload_id) + if deferred_tail: + logger.info( + "UPLOAD_PART_COPY_CONSUME_DEFERRED_TAIL", + bucket=bucket, + key=key, + part_num=part_num, + tail_plaintext_mb=f"{len(deferred_tail) / 1024 / 1024:.2f}MB", + ) + estimated_parts = max(1, math.ceil((plaintext_size + len(deferred_tail)) / chunk_size)) + internal_part_start = await self.multipart_manager.allocate_internal_parts( bucket, key, upload_id, estimated_parts, client_part_number=0 ) @@ -1068,8 +1204,10 @@ async def _streaming_copy_part_inner( key=key, part_num=part_num, plaintext_mb=f"{plaintext_size / 1024 / 1024:.2f}MB", + deferred_tail_bytes=len(deferred_tail), chunk_size_mb=f"{chunk_size / 1024 / 1024:.2f}MB", estimated_parts=estimated_parts, + internal_part_start=internal_part_start, ) src_iter = self._iter_copy_source( @@ -1093,6 +1231,7 @@ async def _streaming_copy_part_inner( src_iter, chunk_size, internal_part_start, + leading_plaintext=deferred_tail or b"", ) etag = md5.hexdigest() @@ -1117,6 +1256,11 @@ async def _streaming_copy_part_inner( part_num=part_num, plaintext_mb=f"{total_plaintext / 1024 / 1024:.2f}MB", internal_parts=len(internal_parts), + first_internal_plaintext_mb=( + f"{internal_parts[0].plaintext_size / 1024 / 1024:.2f}MB" + if internal_parts + else None + ), ) return Response( content=xml_responses.upload_part_copy_result(etag, format_iso8601(datetime.now(UTC))), @@ -1181,6 +1325,8 @@ async def _pump_copy_chunks( src_iter: AsyncIterator[bytes], chunk_size: int, internal_part_start: int, + *, + leading_plaintext: bytes = b"", ) -> tuple[list[InternalPartMetadata], int, int, object]: """Frame-encrypt the copy source into internal S3 parts, one at a time. @@ -1192,7 +1338,7 @@ async def _pump_copy_chunks( matches what the limiter tracks and a multi-GB copy does not hold one reservation for its entire duration. """ - reader = _PlaintextReader(src_iter) + reader = _PlaintextReader(src_iter, prefix=leading_plaintext) md5 = hashlib.md5(usedforsecurity=False) internal_parts: list[InternalPartMetadata] = [] total_plaintext = 0 diff --git a/s3proxy/handlers/multipart/lifecycle.py b/s3proxy/handlers/multipart/lifecycle.py index 0e2cec0..e6763ae 100644 --- a/s3proxy/handlers/multipart/lifecycle.py +++ b/s3proxy/handlers/multipart/lifecycle.py @@ -146,6 +146,18 @@ async def handle_complete_multipart_upload( client, bucket, key, upload_id, context="for complete" ) + if state.deferred_copy_tail: + logger.info( + "COMPLETE_MULTIPART_DEFERRED_TAIL_PENDING", + bucket=bucket, + key=key, + upload_id=upload_id[:20] + "...", + tail_bytes=len(state.deferred_copy_tail), + ) + state = await self._flush_deferred_copy_tail_for_complete( + client, bucket, key, upload_id, state + ) + # Parse client's part list body = await request.body() client_parts = self._parse_client_parts(body) diff --git a/s3proxy/handlers/multipart/upload_part.py b/s3proxy/handlers/multipart/upload_part.py index 9b298dc..c49623f 100644 --- a/s3proxy/handlers/multipart/upload_part.py +++ b/s3proxy/handlers/multipart/upload_part.py @@ -68,9 +68,9 @@ class _PlaintextReader: frames never accumulates the whole part. """ - def __init__(self, source: AsyncIterator[bytes]) -> None: + def __init__(self, source: AsyncIterator[bytes], *, prefix: bytes = b"") -> None: self._it = aiter(source) - self._buf = bytearray() + self._buf = bytearray(prefix) self._eof = False async def read(self, n: int) -> bytes: diff --git a/s3proxy/state/manager.py b/s3proxy/state/manager.py index 42b28ab..981b9da 100644 --- a/s3proxy/state/manager.py +++ b/s3proxy/state/manager.py @@ -305,6 +305,68 @@ def updater(data: bytes) -> bytes: return start + async def set_deferred_copy_tail( + self, + bucket: str, + key: str, + upload_id: str, + tail: bytes, + ) -> None: + """Store a hybrid-passthrough plaintext suffix for the next client part.""" + sk = self._storage_key(bucket, key, upload_id) + + def updater(data: bytes) -> bytes: + state = deserialize_upload_state(data) + if state is None: + raise StateMissingError(f"Upload state corrupted for {bucket}/{key}") + state.deferred_copy_tail = tail + return serialize_upload_state(state) + + result = await self._store.update(sk, updater, self._ttl) + if result is None: + raise StateMissingError(f"Upload state missing for {bucket}/{key}/{upload_id}") + + logger.info( + "DEFERRED_COPY_TAIL_SET", + bucket=bucket, + key=key, + upload_id=self._truncate_id(upload_id), + tail_bytes=len(tail), + ) + + async def take_deferred_copy_tail( + self, + bucket: str, + key: str, + upload_id: str, + ) -> bytes: + """Return and clear any deferred hybrid-passthrough plaintext suffix.""" + sk = self._storage_key(bucket, key, upload_id) + taken = b"" + + def updater(data: bytes) -> bytes: + nonlocal taken + state = deserialize_upload_state(data) + if state is None: + raise StateMissingError(f"Upload state corrupted for {bucket}/{key}") + taken = state.deferred_copy_tail + state.deferred_copy_tail = b"" + return serialize_upload_state(state) + + result = await self._store.update(sk, updater, self._ttl) + if result is None: + raise StateMissingError(f"Upload state missing for {bucket}/{key}/{upload_id}") + + if taken: + logger.info( + "DEFERRED_COPY_TAIL_TAKEN", + bucket=bucket, + key=key, + upload_id=self._truncate_id(upload_id), + tail_bytes=len(taken), + ) + return taken + @staticmethod def _truncate_id(upload_id: str, max_len: int = 20) -> str: """Truncate upload ID for logging.""" diff --git a/s3proxy/state/models.py b/s3proxy/state/models.py index bcb0338..d9a6d0d 100644 --- a/s3proxy/state/models.py +++ b/s3proxy/state/models.py @@ -52,6 +52,9 @@ class MultipartUploadState: total_plaintext_size: int = 0 next_internal_part_number: int = 1 # Next S3 part number to use kid: str = "" # Key id that wraps this upload's DEK ("" = default key) + # Hybrid passthrough may leave a sub-5MB plaintext suffix here when the + # client part range ends mid internal frame but more client parts follow. + deferred_copy_tail: bytes = b"" @dataclass(slots=True) diff --git a/s3proxy/state/serialization.py b/s3proxy/state/serialization.py index 5612626..9857852 100644 --- a/s3proxy/state/serialization.py +++ b/s3proxy/state/serialization.py @@ -37,6 +37,9 @@ def serialize_upload_state(state: MultipartUploadState) -> bytes: "total_plaintext_size": state.total_plaintext_size, "next_internal_part_number": state.next_internal_part_number, "kid": state.kid, + "deferred_copy_tail": base64.b64encode(state.deferred_copy_tail).decode() + if state.deferred_copy_tail + else "", "parts": { str(pn): { "part_number": p.part_number, @@ -134,6 +137,9 @@ def deserialize_upload_state(data: bytes) -> MultipartUploadState | None: total_plaintext_size=obj.get("total_plaintext_size", 0), next_internal_part_number=obj.get("next_internal_part_number", 1), kid=obj.get("kid", ""), + deferred_copy_tail=base64.b64decode(obj["deferred_copy_tail"]) + if obj.get("deferred_copy_tail") + else b"", ) except (KeyError, TypeError, ValueError) as e: logger.error( diff --git a/tests/unit/test_copy_per_part_memory.py b/tests/unit/test_copy_per_part_memory.py index 860e729..358afe4 100644 --- a/tests/unit/test_copy_per_part_memory.py +++ b/tests/unit/test_copy_per_part_memory.py @@ -46,6 +46,12 @@ async def allocate_internal_parts(self, bucket, key, upload_id, count, client_pa async def add_part(self, *a, **k): return None + async def take_deferred_copy_tail(self, bucket, key, upload_id): + return b"" + + async def set_deferred_copy_tail(self, bucket, key, upload_id, tail): + return None + def _streaming_copy_inner(handler, client, plaintext_size: int): state = MultipartUploadState(dek=crypto.generate_dek(), bucket="b", key="k", upload_id="u") diff --git a/tests/unit/test_copy_reservation_vs_real.py b/tests/unit/test_copy_reservation_vs_real.py index b53be39..d57a495 100644 --- a/tests/unit/test_copy_reservation_vs_real.py +++ b/tests/unit/test_copy_reservation_vs_real.py @@ -33,6 +33,12 @@ async def allocate_internal_parts(self, bucket, key, upload_id, count, client_pa async def add_part(self, *a, **k): return None + async def take_deferred_copy_tail(self, bucket, key, upload_id): + return b"" + + async def set_deferred_copy_tail(self, bucket, key, upload_id, tail): + return None + class _Body: """aiohttp-like streaming body: `async with body`, `body.content.read(n)`.""" diff --git a/tests/unit/test_upload_part_copy_passthrough.py b/tests/unit/test_upload_part_copy_passthrough.py index 8b81cf6..50f1f6b 100644 --- a/tests/unit/test_upload_part_copy_passthrough.py +++ b/tests/unit/test_upload_part_copy_passthrough.py @@ -381,11 +381,17 @@ def _build_scylla_prod_shape_source( ) +@pytest.mark.slow @pytest.mark.asyncio async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough( mock_s3, settings, manager, credentials, monkeypatch ): - """Regression: prod range ends mid internal frame → hybrid passthrough + tail.""" + """Regression: prod range ends mid internal frame → hybrid passthrough + deferred tail. + + Two-part complete (part 2 consumes tail, no EntityTooSmall) is covered by + test_two_part_hybrid_defer_tail_completes_fast — the full prod-shape two-part + path OOMs CI runners (~5GB mock source). + """ monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) handler = _copy_handler(settings, manager) _patch_client(handler, mock_s3) @@ -432,12 +438,190 @@ async def test_scylla_prod_shape_range_smaller_than_metadata_uses_passthrough( 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 + assert len(tail_ops) == 0 updated = await manager.get_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id) part = updated.parts[1] assert part.plaintext_size == prod_range_end + 1 - assert len(part.internal_parts) == len(copy_ops) + len(tail_ops) + assert len(part.internal_parts) == len(copy_ops) + assert len(updated.deferred_copy_tail) > 0 + assert len(updated.deferred_copy_tail) < crypto.MIN_PART_SIZE + + +def test_should_defer_hybrid_tail_when_more_client_parts_follow(settings, manager): + from s3proxy.handlers.multipart.copy import _CiphertextSegment + + handler = _copy_handler(settings, manager) + chunk = (50 * 1024 * 1024) // 6 + ct = chunk + 64 + segments = [_CiphertextSegment(chunk, ct, i * ct) for i in range(600)] + range_end = 4_999_341_931 + split = handler._split_plaintext_range_on_segments(segments, 0, range_end) + assert split is not None + assert split.streaming_tail is not None + assert handler._should_defer_hybrid_tail( + split.streaming_tail, + range_end, + segments, + client_part_plaintext_size=range_end + 1, + part_num=1, + ) + # Last client part: tail may stay on S3 as the final internal part. + source_end = handler._source_plaintext_end(segments) + assert not handler._should_defer_hybrid_tail( + split.streaming_tail, + source_end, + segments, + client_part_plaintext_size=source_end + 1, + part_num=1, + ) + + +def test_should_not_defer_hybrid_tail_for_small_client_part(settings, manager): + """Single-part partial copies must upload the tail immediately (self-contained).""" + 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(20)] + range_end = chunk * 10 - 1 - chunk // 2 + split = handler._split_plaintext_range_on_segments(segments, 0, range_end) + assert split is not None + assert split.streaming_tail is not None + tail_bytes = split.streaming_tail[1] - split.streaming_tail[0] + 1 + assert tail_bytes < crypto.MIN_PART_SIZE + # 10MB client part 1 — below 1 GiB threshold, tail uploads on part 1. + assert not handler._should_defer_hybrid_tail( + split.streaming_tail, + range_end, + segments, + client_part_plaintext_size=range_end + 1, + part_num=1, + ) + + +@pytest.mark.asyncio +async def test_deferred_copy_tail_state_roundtrip(manager): + """Deferred tail survives serialize/deserialize between client parts.""" + bucket, key, upload_id = BUCKET, "defer-key", "upload-defer" + dek = crypto.generate_dek() + tail = b"T" * (crypto.MIN_PART_SIZE // 2) + await manager.create_upload(bucket, key, upload_id, dek, kid="kid") + await manager.set_deferred_copy_tail(bucket, key, upload_id, tail) + + loaded = await manager.get_upload(bucket, key, upload_id) + assert loaded.deferred_copy_tail == tail + + taken = await manager.take_deferred_copy_tail(bucket, key, upload_id) + assert taken == tail + after = await manager.get_upload(bucket, key, upload_id) + assert not after.deferred_copy_tail + + +@pytest.mark.asyncio +async def test_two_part_hybrid_defer_tail_completes_fast( + mock_s3, settings, manager, credentials, monkeypatch +): + """Fast regression: defer sub-5MB tail on part 1, fold into part 2, complete cleanly.""" + from s3proxy.handlers.multipart import copy as copy_mod + + monkeypatch.setattr(copy_mod, "HYBRID_TAIL_DEFER_MIN_CLIENT_PART", 5 * 1024 * 1024) + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MIN_PART_SIZE) + handler = _handler(settings, mock_s3, credentials) + await mock_s3.create_bucket(BUCKET) + + 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. + num_segments = 12 + part1_range_end = chunk * 4 + chunk // 2 # 36MB, ~4MB deferred tail + src_dek, src_plaintext, ciphertext_blob, src_meta, inflated_total, _ = ( + _build_scylla_prod_shape_source( + kid, + kek, + num_internal_parts=num_segments, + metadata_inflate_ratio=1.2, + chunk_size=chunk, + scylla_range_end=part1_range_end, + ) + ) + assert inflated_total > part1_range_end + 1 + + await mock_s3.put_object(BUCKET, "sst/small-Data.db", ciphertext_blob) + await save_multipart_metadata(mock_s3, BUCKET, "sst/small-Data.db", src_meta) + + resp_create = await mock_s3.create_multipart_upload(BUCKET, "sst/small-Data.db.sm_manifest") + upload_id = resp_create["UploadId"] + await handler.multipart_manager.create_upload( + BUCKET, "sst/small-Data.db.sm_manifest", upload_id, src_dek, kid + ) + + mark = len(mock_s3.call_history) + resp1 = await handler.handle_upload_part_copy( + _copy_part_request( + f"/{BUCKET}/sst/small-Data.db.sm_manifest", + f"/{BUCKET}/sst/small-Data.db", + upload_id, + part_number=1, + copy_source_range=f"bytes=0-{part1_range_end}", + ), + credentials, + ) + await _read(resp1) + part1_ops = mock_s3.call_history[mark:] + assert not [c for c in part1_ops if c[0] == "upload_part"], ( + "deferred tail must not upload on part 1" + ) + + after_part1 = await handler.multipart_manager.get_upload( + BUCKET, "sst/small-Data.db.sm_manifest", upload_id + ) + deferred = after_part1.deferred_copy_tail + assert deferred + assert len(deferred) < crypto.MIN_PART_SIZE + + part2_start = part1_range_end + 1 + resp2 = await handler.handle_upload_part_copy( + _copy_part_request( + f"/{BUCKET}/sst/small-Data.db.sm_manifest", + f"/{BUCKET}/sst/small-Data.db", + upload_id, + part_number=2, + copy_source_range=f"bytes={part2_start}-{inflated_total - 1}", + ), + credentials, + ) + await _read(resp2) + + after_part2 = await handler.multipart_manager.get_upload( + BUCKET, "sst/small-Data.db.sm_manifest", upload_id + ) + assert not after_part2.deferred_copy_tail + part2_internal = after_part2.parts[2].internal_parts + assert part2_internal[0].plaintext_size >= crypto.MIN_PART_SIZE + + all_internal = [] + for pn in sorted(after_part2.parts): + all_internal.extend(after_part2.parts[pn].internal_parts) + all_internal.sort(key=lambda ip: ip.internal_part_number) + for ip in all_internal[:-1]: + assert ip.plaintext_size >= crypto.MIN_PART_SIZE + + complete_body = ( + "" + f'1"{after_part2.parts[1].etag}"' + f'2"{after_part2.parts[2].etag}"' + "" + ).encode() + complete_req = MagicMock() + complete_req.url.path = f"/{BUCKET}/sst/small-Data.db.sm_manifest" + complete_req.url.query = f"uploadId={upload_id}" + complete_req.headers = {} + complete_req.body = AsyncMock(return_value=complete_body) + await handler.handle_complete_multipart_upload(complete_req, credentials) def test_prod_shape_passthrough_eligibility_and_route(settings, manager, credentials):