fix: hybrid passthrough defer tail + UploadPartCopy keepalive (2026.7.17)

.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:

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:

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]

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

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)

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: