fix: passthrough Scylla manifest UploadPartCopy with full-object range

s3proxy/handlers/multipart/copy.py

@@ -63,7 +63,7 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
         async with self._client(creds) as client:
             upload_id, part_num = self._extract_multipart_params(request)
             copy_source = request.headers.get("x-amz-copy-source", "")
-            copy_source_range = request.headers.get("x-amz-copy-source-range")
+            raw_copy_source_range = request.headers.get("x-amz-copy-source-range")
             src_bucket, src_key = self._parse_copy_source(copy_source)
 
             state = await self.multipart_manager.get_upload(bucket, key, upload_id)
@@ -84,6 +84,12 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
             src_wrapped_dek = src_metadata.get(self.settings.dektag_name)
             src_multipart_meta = await load_multipart_metadata(client, src_bucket, src_key)
 
+            total_plaintext = self._copy_plaintext_size(
+                head_resp, None, src_wrapped_dek, src_multipart_meta
+            )
+            copy_source_range = self._normalize_copy_source_range(
+                raw_copy_source_range, total_plaintext
+            )
             plaintext_size = self._copy_plaintext_size(
                 head_resp, copy_source_range, src_wrapped_dek, src_multipart_meta
             )
@@ -198,6 +204,22 @@ def _copy_plaintext_size(
         start, end = self._parse_copy_source_range(copy_source_range, total)
         return end - start + 1
 
+    def _normalize_copy_source_range(
+        self, copy_source_range: str | None, total_plaintext_size: int
+    ) -> str | None:
+        """Treat a range spanning the entire object as 'copy whole object'.
+
+        Scylla Manager often sends ``bytes=0-(size-1)`` on manifest UploadPartCopy
+        even when copying the full SST. That must not force the streaming re-encrypt
+        path (which queues behind the copy pipeline and exceeds the 300s client timeout).
+        """
+        if not copy_source_range:
+            return None
+        start, end = self._parse_copy_source_range(copy_source_range, total_plaintext_size)
+        if start == 0 and end == total_plaintext_size - 1:
+            return None
+        return copy_source_range
+
     def _can_passthrough_part_copy(
         self,
         copy_source_range: str | None,
@@ -340,25 +362,7 @@ async def _gated_passthrough_copy_part(
         src_multipart_meta: MultipartMetadata | None,
         plaintext_size: int,
     ) -> Response:
-        """Passthrough with the same pipeline cap as the re-encrypt path."""
-        if plaintext_size > crypto.STREAMING_THRESHOLD:
-            async with _copy_pipeline_semaphore:
-                return await self._passthrough_copy_part(
-                    client,
-                    bucket,
-                    key,
-                    upload_id,
-                    part_num,
-                    state,
-                    src_bucket,
-                    src_key,
-                    copy_source,
-                    head_resp,
-                    src_metadata,
-                    src_wrapped_dek,
-                    src_multipart_meta,
-                    plaintext_size,
-                )
+        """Server-side ciphertext copy; does not use the streaming pipeline semaphore."""
         return await self._passthrough_copy_part(
             client,
             bucket,

tests/integration/passthrough_verify.py

@@ -304,14 +304,35 @@ def check_dek_adoption(ctx: RunContext, source: str, dest: str) -> None:
 
 
 def check_reencrypt_control(ctx: RunContext, source: str, dest: str, size: int) -> None:
+    # Partial range forces streaming re-encrypt; full-object range is passthrough.
+    partial_end = min(size - 1, 32 * MB - 1)
     peak, enc = poll_during(
         ctx,
-        lambda: upload_part_copy(ctx, dest, source, byte_range=f"bytes=0-{size - 1}"),
+        lambda: upload_part_copy(ctx, dest, source, byte_range=f"bytes=0-{partial_end}"),
     )
-    ctx.ok("encrypts full object", enc >= size * 0.5, f"{enc / MB:.0f}MB")
+    ctx.ok("encrypts partial range", enc >= (partial_end + 1) * 0.5, f"{enc / MB:.0f}MB")
     ctx.ok("high peak memory", peak >= CHUNK_PEAK * 0.5, f"{peak / MB:.2f}MB")
 
 
+def check_scylla_manifest_full_range_passthrough(
+    ctx: RunContext, source: str, dest: str, size: int
+) -> None:
+    """Scylla sends bytes=0-(size-1) on manifest UploadPartCopy; must passthrough."""
+    peak, enc = poll_during(
+        ctx,
+        lambda: upload_part_copy(ctx, dest, source, byte_range=f"bytes=0-{size - 1}"),
+    )
+    ctx.ok("full-range zero bytes_encrypted", enc <= 2 * MB, f"{enc / MB:.2f}MB")
+    ctx.ok("full-range low peak memory", peak <= CHUNK_PEAK * 0.5, f"{peak / MB:.2f}MB")
+    src_ct, dest_ct = raw_ciphertext(ctx, source), raw_ciphertext(ctx, dest)
+    ctx.ok("full-range ciphertext identical", src_ct == dest_ct)
+    ctx.ok(
+        "UPLOAD_PART_COPY_PASSTHROUGH logged",
+        log_has_passthrough(ctx.log_path),
+        str(ctx.log_path),
+    )
+
+
 def check_scylla_scale(ctx: RunContext, source: str, dest: str) -> None:
     t0 = time.monotonic()
     peak, enc = poll_during(ctx, lambda: upload_part_copy(ctx, dest, source))

tests/integration/test_upload_part_copy_passthrough_e2e.py

@@ -17,6 +17,7 @@
     check_concurrent_flood,
     check_dek_adoption,
     check_reencrypt_control,
+    check_scylla_manifest_full_range_passthrough,
     check_scylla_scale,
     log_has_passthrough,
     poll_during,
@@ -78,9 +79,21 @@ def test_range_copy_reencrypts_with_high_memory(self, passthrough_env):
         ctx.failures.clear()
         check_reencrypt_control(ctx, "sst/source-96mb.bin", "sst/dest-d.bin", QUICK_SIZE)
         assert proc.poll() is None
-        _assert_check(ctx, "encrypts full object")
+        _assert_check(ctx, "encrypts partial range")
         _assert_check(ctx, "high peak memory")
 
+    def test_scylla_manifest_full_range_passthrough(self, passthrough_env):
+        ctx, proc, _ = passthrough_env
+        ctx.failures.clear()
+        check_scylla_manifest_full_range_passthrough(
+            ctx, "sst/source-1280mb.bin", "sst/dest-1280mb-fullrange.bin", SCYLLA_SSTABLE_SIZE
+        )
+        assert proc.poll() is None
+        _assert_check(ctx, "full-range zero bytes_encrypted")
+        _assert_check(ctx, "full-range low peak memory")
+        _assert_check(ctx, "full-range ciphertext identical")
+        _assert_check(ctx, "UPLOAD_PART_COPY_PASSTHROUGH logged")
+
     def test_passthrough_log_line(self, passthrough_env):
         _, proc, log_path = passthrough_env
         assert proc.poll() is None
@@ -116,17 +129,18 @@ def test_passthrough_encrypt_delta_vs_reencrypt(self, passthrough_env):
         peak_pt, enc_pt = poll_during(
             ctx, lambda: upload_part_copy(ctx, "sst/dest-compare-pt.bin", "sst/source-96mb.bin")
         )
+        partial_end = min(QUICK_SIZE - 1, 32 * MB - 1)
         peak_re, enc_re = poll_during(
             ctx,
             lambda: upload_part_copy(
                 ctx,
                 "sst/dest-compare-re.bin",
                 "sst/source-96mb.bin",
-                byte_range=f"bytes=0-{QUICK_SIZE - 1}",
+                byte_range=f"bytes=0-{partial_end}",
             ),
         )
         assert proc.poll() is None
         assert enc_pt <= 1 * MB
-        assert enc_re >= QUICK_SIZE * 0.5
+        assert enc_re >= (partial_end + 1) * 0.5
         assert peak_pt <= CHUNK_PEAK * 0.5
         assert peak_re >= CHUNK_PEAK * 0.5

tests/unit/test_upload_part_copy_passthrough.py

@@ -39,13 +39,15 @@ async def _fake_client(creds):
     handler._client = _fake_client
 
 
-def _copy_part_request(dest_path, copy_source, upload_id, part_number=1):
+def _copy_part_request(dest_path, copy_source, upload_id, part_number=1, copy_source_range=None):
     req = MagicMock()
     req.url.path = dest_path
     req.url.query = f"uploadId={upload_id}&partNumber={part_number}"
     req.headers = {
         "x-amz-copy-source": copy_source,
     }
+    if copy_source_range is not None:
+        req.headers["x-amz-copy-source-range"] = copy_source_range
     return req
 
 
@@ -274,6 +276,166 @@ async def test_range_copy_still_reencrypts(mock_s3, settings, manager, credentia
     assert any(c[0] == "upload_part" for c in during)
 
 
+@pytest.mark.asyncio
+async def test_normalize_copy_source_range_treats_full_object_as_whole(
+    settings, manager, credentials
+):
+    handler = _copy_handler(settings, manager)
+    total = crypto.STREAMING_THRESHOLD + crypto.MAX_BUFFER_SIZE
+    assert handler._normalize_copy_source_range(None, total) is None
+    assert handler._normalize_copy_source_range(f"bytes=0-{total - 1}", total) is None
+    assert handler._normalize_copy_source_range(f"bytes=0-{total + 9999}", total) is None
+    partial = handler._normalize_copy_source_range(f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}", total)
+    assert partial == f"bytes=0-{crypto.MAX_BUFFER_SIZE - 1}"
+
+
+def _build_large_multipart_source(mock_s3, kid, kek, *, num_internal_parts: int):
+    """Encrypted multipart source above STREAMING_THRESHOLD (Scylla manifest shape)."""
+    src_dek = crypto.generate_dek()
+    chunk_size = crypto.MAX_BUFFER_SIZE
+    src_plaintext = b"M" * (chunk_size * num_internal_parts)
+
+    ciphertext_blob = bytearray()
+    internal_parts_meta = []
+    for i, start in enumerate(range(0, len(src_plaintext), chunk_size), 1):
+        chunk = src_plaintext[start : start + chunk_size]
+        ct = crypto.encrypt_frame(chunk, src_dek, "src-upload", i, 0)
+        internal_parts_meta.append(
+            InternalPartMetadata(
+                internal_part_number=i,
+                plaintext_size=len(chunk),
+                ciphertext_size=len(ct),
+                etag=hashlib.md5(ct, usedforsecurity=False).hexdigest(),
+            )
+        )
+        ciphertext_blob.extend(ct)
+
+    src_meta = MultipartMetadata(
+        version=2,
+        part_count=1,
+        total_plaintext_size=len(src_plaintext),
+        parts=[
+            PartMetadata(
+                part_number=1,
+                plaintext_size=len(src_plaintext),
+                ciphertext_size=len(ciphertext_blob),
+                etag="ignored",
+                md5=hashlib.md5(src_plaintext, usedforsecurity=False).hexdigest(),
+                internal_parts=internal_parts_meta,
+            )
+        ],
+        wrapped_dek=crypto.wrap_key(src_dek, kek),
+        kid=kid,
+    )
+    return src_dek, src_plaintext, bytes(ciphertext_blob), src_meta
+
+
+@pytest.mark.asyncio
+async def test_scylla_manifest_full_range_uses_passthrough_not_streaming(
+    mock_s3, settings, manager, credentials, monkeypatch
+):
+    """Full-object CopySourceRange must not fall into UPLOAD_PART_COPY_STREAMING."""
+    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 = (crypto.STREAMING_THRESHOLD // crypto.MAX_BUFFER_SIZE) + 2
+    src_dek, src_plaintext, ciphertext_blob, src_meta = _build_large_multipart_source(
+        mock_s3, kid, kek, num_internal_parts=num_parts
+    )
+    assert len(src_plaintext) > 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)
+
+    resp_create = await mock_s3.create_multipart_upload(BUCKET, "sst/big-Data.db.sm_manifest")
+    upload_id = resp_create["UploadId"]
+    await manager.create_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id, src_dek, kid)
+
+    full_range = f"bytes=0-{len(src_plaintext) - 1}"
+    mark = len(mock_s3.call_history)
+    await handler.handle_upload_part_copy(
+        _copy_part_request(
+            f"/{BUCKET}/sst/big-Data.db.sm_manifest",
+            f"/{BUCKET}/sst/big-Data.db",
+            upload_id,
+            copy_source_range=full_range,
+        ),
+        credentials,
+    )
+    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)
+
+    updated = await manager.get_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id)
+    assert updated.dek == src_dek
+    part = updated.parts[1]
+    assert part.plaintext_size == len(src_plaintext)
+    assert len(part.internal_parts) == num_parts
+
+
+@pytest.mark.asyncio
+async def test_large_passthrough_not_blocked_by_pipeline_semaphore(
+    mock_s3, settings, manager, credentials, monkeypatch
+):
+    """Passthrough must not queue behind the streaming copy pipeline semaphore."""
+    import asyncio
+
+    from s3proxy.handlers.multipart.copy import reset_copy_pipeline_semaphore
+
+    monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE)
+    reset_copy_pipeline_semaphore(1)
+
+    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 = (crypto.STREAMING_THRESHOLD // crypto.MAX_BUFFER_SIZE) + 2
+    src_dek, src_plaintext, ciphertext_blob, src_meta = _build_large_multipart_source(
+        mock_s3, kid, kek, num_internal_parts=num_parts
+    )
+
+    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)
+
+    resp_create = await mock_s3.create_multipart_upload(BUCKET, "sst/big-Data.db.sm_manifest")
+    upload_id = resp_create["UploadId"]
+    await manager.create_upload(BUCKET, "sst/big-Data.db.sm_manifest", upload_id, src_dek, kid)
+
+    gate = asyncio.Event()
+
+    async def blocked_streaming(*args, **kwargs):
+        gate.set()
+        await asyncio.Event().wait()
+
+    handler._streaming_copy_part = blocked_streaming  # type: ignore[method-assign]
+
+    passthrough_task = asyncio.create_task(
+        handler.handle_upload_part_copy(
+            _copy_part_request(
+                f"/{BUCKET}/sst/big-Data.db.sm_manifest",
+                f"/{BUCKET}/sst/big-Data.db",
+                upload_id,
+                copy_source_range=f"bytes=0-{len(src_plaintext) - 1}",
+            ),
+            credentials,
+        )
+    )
+
+    for _ in range(200):
+        if passthrough_task.done():
+            break
+        await asyncio.sleep(0.01)
+
+    assert passthrough_task.done()
+    assert not gate.is_set()
+    await passthrough_task
+
+
 def _extract_upload_id(xml_body: bytes) -> str:
     import xml.etree.ElementTree as ET