fix: per-part copy memory reservation (Scylla backup 503 storm)

.github/workflows/test.yml

@@ -34,7 +34,7 @@ jobs:
       fail-fast: false
       matrix:
         shard: [memory_usage, memory_leak, memory_copy, core, multipart, copy_range, misc]
-    timeout-minutes: ${{ contains(fromJSON('["memory_usage", "memory_leak"]'), matrix.shard) && 15 || 10 }}
+    timeout-minutes: ${{ matrix.shard == 'memory_copy' && 25 || (contains(fromJSON('["memory_usage", "memory_leak"]'), matrix.shard) && 15 || 10) }}
     steps:
       - uses: actions/checkout@v7
 

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test test-all test-unit test-integration test-integration-shard test-run test-oom e2e cluster lint
+.PHONY: test test-all test-unit test-integration test-integration-shard test-run test-oom verify-copy-memory e2e cluster lint
 
 # Lint: ruff check + format check
 lint:
@@ -12,10 +12,14 @@ test: test-unit
 test-unit:
 	uv run pytest -m "not e2e and not ha" -v -n auto
 
+# Pre-merge gate: subprocess Prometheus proof of per-part copy memory release.
+verify-copy-memory:
+	uv run pytest tests/integration/test_copy_per_part_metrics.py -m e2e -v -n0
+
 # Integration shards for parallel CI (make test-integration-shard SHARD=memory_usage)
 INTEGRATION_memory_usage_TESTS = tests/integration/test_memory_usage.py
 INTEGRATION_memory_leak_TESTS = tests/integration/test_memory_leak.py
-INTEGRATION_memory_copy_TESTS = tests/integration/test_copy_memory_governor.py
+INTEGRATION_memory_copy_TESTS = tests/integration/test_copy_memory_governor.py tests/integration/test_copy_per_part_metrics.py
 INTEGRATION_memory_copy_PYTEST_OPTS = -n0
 INTEGRATION_core_TESTS = \
 	tests/integration/test_integration.py \

s3proxy/crypto.py

@@ -261,31 +261,40 @@ def copy_internal_part_size(plaintext_size: int) -> int:
     return COPY_INTERNAL_PART_SIZE
 
 
+def copy_chunk_peak(chunk_plaintext_bytes: int) -> int:
+    """Peak memory while encrypting one internal copy chunk (streaming path).
+
+    The streaming copy path acquires/releases this amount per internal part so
+    memory is free between chunks and during S3 upload I/O. Reservation covers
+    read-buffer slack plus framed encrypt peak; ciphertext upload runs after
+    release (smaller RSS than encrypt peak).
+    """
+    framed = (
+        4 * chunk_plaintext_bytes
+        if chunk_plaintext_bytes <= FRAME_PLAINTEXT_SIZE
+        else 2 * chunk_plaintext_bytes + FRAME_PLAINTEXT_SIZE
+    )
+    peak = framed + 2 * MAX_BUFFER_SIZE
+    if chunk_plaintext_bytes > 32 * 1024 * 1024:
+        peak += chunk_plaintext_bytes // 7
+    return peak
+
+
+def copy_small_buffered_peak(plaintext_size: int) -> int:
+    """Peak for a small copy that buffers the whole object in RAM."""
+    return max(MAX_BUFFER_SIZE, 3 * plaintext_size)
+
+
 def copy_pipeline_peak(plaintext_size: int) -> int:
-    """Peak memory a server-side copy holds while decrypting + re-encrypting.
-
-    A copy request has no body, so the request-level limiter reserves ~nothing
-    for it -- yet the copy reads the source, decrypts it and re-encrypts it. The
-    streaming copy path frames exactly like the framed UploadPart path
-    (encrypt_frame per FRAME_PLAINTEXT_SIZE, one internal part at a time) but uses
-    a FIXED copy_internal_part_size, so the per-part peak -- and thus the whole
-    reservation -- is O(1) in object size (~90MB) rather than object_size/20
-    (~535MB for a 4.7GB copy, which monopolized the budget and deadlocked). It
-    adds one pipeline stage the body-fed upload path lacks: the source is read in
-    MAX_BUFFER_SIZE chunks into a _PlaintextReader buffer, so reserve two extra
-    buffers for it. Small copies buffer the whole object and re-encrypt it (~3x).
-
-    aiobotocore copies the ciphertext body for signing; tracemalloc shows
-    ~part/7 slack once parts exceed 32MB.
+    """Peak memory for one copy chunk (streaming) or a small buffered copy.
+
+    Streaming copies reserve per internal part via copy_chunk_peak; this returns
+    the per-chunk peak (O(1) in object size, ~88MB for 32MB parts). Small
+    copies hold the whole object under one reservation.
     """
     if plaintext_size > STREAMING_THRESHOLD:
-        part = copy_internal_part_size(plaintext_size)
-        framed = 4 * part if part <= FRAME_PLAINTEXT_SIZE else 2 * part + FRAME_PLAINTEXT_SIZE
-        peak = framed + 2 * MAX_BUFFER_SIZE
-        if part > 32 * 1024 * 1024:
-            peak += part // 7
-        return peak
-    return max(MAX_BUFFER_SIZE, 3 * plaintext_size)
+        return copy_chunk_peak(copy_internal_part_size(plaintext_size))
+    return copy_small_buffered_peak(plaintext_size)
 
 
 @dataclass(slots=True)

s3proxy/handlers/multipart/copy.py

@@ -219,27 +219,24 @@ async def _streaming_copy_part(
         plaintext_size: int,
     ) -> Response:
         # UploadPartCopy carries no request body, so the request-level limiter
-        # reserved ~nothing -- but this streams the source through decrypt +
-        # re-encrypt. Reserve the pipeline peak so concurrent copies are bounded
-        # (a dedup flood otherwise runs unbounded and OOMs the pod).
-        peak = crypto.copy_pipeline_peak(plaintext_size)
-        async with concurrency.reserve_copy_memory(peak):
-            return await self._streaming_copy_part_inner(
-                client,
-                bucket,
-                key,
-                upload_id,
-                part_num,
-                state,
-                src_bucket,
-                src_key,
-                copy_source_range,
-                src_wrapped_dek,
-                src_multipart_meta,
-                head_resp,
-                src_metadata,
-                plaintext_size,
-            )
+        # reserved ~nothing. Per-internal-part reservations in _pump_copy_chunks
+        # gate concurrent copies without holding memory for the whole object.
+        return await self._streaming_copy_part_inner(
+            client,
+            bucket,
+            key,
+            upload_id,
+            part_num,
+            state,
+            src_bucket,
+            src_key,
+            copy_source_range,
+            src_wrapped_dek,
+            src_multipart_meta,
+            head_resp,
+            src_metadata,
+            plaintext_size,
+        )
 
     async def _streaming_copy_part_inner(
         self,
@@ -397,11 +394,9 @@ async def _pump_copy_chunks(
         Mirrors UploadPartMixin._stream_and_upload_framed: reads FRAME_PLAINTEXT_SIZE
         plaintext frames from the source, encrypts each with encrypt_frame,
         accumulates one internal part's ciphertext, then uploads it before starting
-        the next. Peak memory is O(one internal part ciphertext + one frame) -- it
-        never holds the whole part's plaintext and ciphertext at once and runs one
-        part at a time, which is what makes copy_pipeline_peak == streaming_upload_peak.
-        Parts are byte-identical to the framed upload path and read back via
-        decrypt_framed / _iter_multipart_plaintext.
+        the next. Memory governor reservation is per internal part (acquire before
+        encrypt, release before upload) so a multi-GB copy does not hold ~88MB for
+        its entire duration and other copies can interleave during S3 I/O waits.
         """
         reader = _PlaintextReader(src_iter)
         md5 = hashlib.md5(usedforsecurity=False)
@@ -410,28 +405,39 @@ async def _pump_copy_chunks(
         total_ciphertext = 0
         internal_part_num = internal_part_start
 
+        prefetch: bytes | None = None
         while True:
+            if prefetch is None:
+                prefetch = await reader.read(min(crypto.FRAME_PLAINTEXT_SIZE, chunk_size))
+                if not prefetch:
+                    break
+
+            part_reserve = crypto.copy_chunk_peak(chunk_size)
             ciphertext = bytearray()
             part_plaintext = 0
             frame_idx = 0
-            while part_plaintext < chunk_size:
-                frame_pt = await reader.read(
-                    min(crypto.FRAME_PLAINTEXT_SIZE, chunk_size - part_plaintext)
-                )
-                if not frame_pt:
-                    break
-                md5.update(frame_pt)
-                part_plaintext += len(frame_pt)
-                ciphertext.extend(
-                    crypto.encrypt_frame(
-                        frame_pt, state.dek, upload_id, internal_part_num, frame_idx
+            async with concurrency.reserve_copy_memory(part_reserve):
+                frame_pt = prefetch
+                prefetch = None
+                while True:
+                    md5.update(frame_pt)
+                    part_plaintext += len(frame_pt)
+                    ciphertext.extend(
+                        crypto.encrypt_frame(
+                            frame_pt, state.dek, upload_id, internal_part_num, frame_idx
+                        )
                     )
-                )
-                frame_idx += 1
-
-            if part_plaintext == 0:
-                break
+                    frame_idx += 1
+                    if part_plaintext >= chunk_size:
+                        break
+                    frame_pt = await reader.read(
+                        min(crypto.FRAME_PLAINTEXT_SIZE, chunk_size - part_plaintext)
+                    )
+                    if not frame_pt:
+                        break
 
+            # Reservation released before upload so other copies can encrypt while
+            # this one waits on S3 I/O (ciphertext RSS is smaller than encrypt peak).
             upload_start = time.monotonic()
             resp = await client.upload_part(bucket, key, upload_id, internal_part_num, ciphertext)
             del ciphertext

s3proxy/handlers/objects/misc.py

@@ -205,23 +205,27 @@ async def handle_copy_object(self, request: Request, creds: S3Credentials) -> Re
                     request,
                 )
 
-            # Encrypted source. A same-credential plain COPY needs no re-encrypt:
-            # the ciphertext is self-describing and key-independent (GCM AAD is
-            # None, the DEK is random and stored in metadata / the sidecar,
-            # nonces are embedded), so a native server-side CopyObject yields a
-            # byte-identical, decryptable destination and skips the
-            # download+re-encrypt+re-upload amplification (Scylla Manager dedup).
-            # A cross-credential copy must re-key under the copier's KEK, and
-            # REPLACE / oversized objects also fall back to the re-encrypt path.
+            # Encrypted source. A plain COPY needs no re-encrypt: the ciphertext
+            # is self-describing and key-independent (GCM AAD is None, the DEK is
+            # random and stored in metadata / the sidecar, nonces are embedded),
+            # so a native server-side CopyObject yields a byte-identical
+            # destination and skips the download+re-encrypt+re-upload
+            # amplification (Scylla Manager dedup versioning).
+            #
+            # Re-encrypt only to re-key to a *different* owning credential. A
+            # same-kid copy is byte-identical, and a legacy object with no kid
+            # cannot be decrypted to re-encrypt anyway (re-encrypt would fail with
+            # UnknownKidError) -- both must pass through as-is. REPLACE and
+            # objects above the single-op CopyObject limit still re-encrypt.
             if src_multipart_meta:
                 src_kid = src_multipart_meta.kid
             else:
                 src_kid = head_resp.get("Metadata", {}).get(self.settings.kidtag_name, "")
-            same_credential = bool(src_kid) and src_kid == client.credentials.access_key
+            needs_rekey = bool(src_kid) and src_kid != client.credentials.access_key
             ciphertext_size = head_resp.get("ContentLength", 0) or 0
             if (
                 metadata_directive == "COPY"
-                and same_credential
+                and not needs_rekey
                 and ciphertext_size <= MAX_SERVER_SIDE_COPY_BYTES
             ):
                 return await self._copy_passthrough_encrypted(

tests/integration/conftest.py

@@ -6,12 +6,95 @@
 import subprocess
 import sys
 import time
+import urllib.error
+import urllib.request
 from collections.abc import Generator
 from contextlib import contextmanager
 
 import boto3
 import pytest
 
+# === MINIO BACKEND ===
+
+
+def _is_minio(endpoint: str) -> bool:
+    try:
+        req = urllib.request.Request(f"{endpoint}/minio/health/live")
+        with urllib.request.urlopen(req, timeout=2) as resp:
+            return resp.status == 200
+    except urllib.error.URLError, TimeoutError, OSError:
+        return False
+
+
+def _wait_for_minio(endpoint: str, timeout: float = 30) -> bool:
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if _is_minio(endpoint):
+            return True
+        time.sleep(0.5)
+    return False
+
+
+def _find_free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+@contextlib.contextmanager
+def minio_backend() -> Generator[str]:
+    """Yield a MinIO HTTP endpoint, starting a throwaway container if needed."""
+    for port in (9000, 19000, 19001):
+        endpoint = f"http://localhost:{port}"
+        if _is_minio(endpoint):
+            yield endpoint
+            return
+
+    port = 19000
+    while port < 19100:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            if s.connect_ex(("localhost", port)) == 0:
+                port += 1
+                continue
+        break
+    else:
+        raise RuntimeError("no free port for temporary MinIO")
+
+    name = f"s3proxy-test-minio-{port}"
+    subprocess.run(["docker", "rm", "-f", name], capture_output=True, check=False)
+    proc = subprocess.Popen(
+        [
+            "docker",
+            "run",
+            "--rm",
+            "--name",
+            name,
+            "-p",
+            f"{port}:9000",
+            "-e",
+            "MINIO_ROOT_USER=minioadmin",
+            "-e",
+            "MINIO_ROOT_PASSWORD=minioadmin",
+            "minio/minio",
+            "server",
+            "/data",
+        ],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    endpoint = f"http://localhost:{port}"
+    try:
+        if not _wait_for_minio(endpoint):
+            raise RuntimeError(f"MinIO container failed to become healthy on port {port}")
+        yield endpoint
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            subprocess.run(["docker", "kill", name], capture_output=True, check=False)
+
+
 # === SHARED S3PROXY HELPER ===
 
 
@@ -61,7 +144,7 @@ def run_s3proxy(
 
     output = sys.stderr if log_output else subprocess.DEVNULL
     proc = subprocess.Popen(
-        ["python", "-m", "s3proxy.main"],
+        [sys.executable, "-m", "s3proxy.main"],
         env=env,
         stdout=output,
         stderr=output,