fix: stop copy clamp under-reserve for multi-GB manifest copies

.github/workflows/test.yml

@@ -30,7 +30,11 @@ jobs:
 
   integration:
     runs-on: ubuntu-latest
-    timeout-minutes: 20
+    strategy:
+      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 }}
     steps:
       - uses: actions/checkout@v7
 
@@ -46,5 +50,5 @@ jobs:
       - name: Install dependencies
         run: uv sync --extra dev
 
-      - name: Run integration tests
-        run: make test-integration
+      - name: Run integration tests (${{ matrix.shard }})
+        run: make test-integration-shard SHARD=${{ matrix.shard }}

Makefile

@@ -1,4 +1,4 @@
-.PHONY: test test-all test-unit test-integration test-run test-oom e2e cluster lint
+.PHONY: test test-all test-unit test-integration test-integration-shard test-run test-oom e2e cluster lint
 
 # Lint: ruff check + format check
 lint:
@@ -12,6 +12,38 @@ test: test-unit
 test-unit:
 	uv run pytest -m "not e2e and not ha" -v -n auto
 
+# 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_PYTEST_OPTS = -n0
+INTEGRATION_core_TESTS = \
+	tests/integration/test_integration.py \
+	tests/integration/test_handlers.py \
+	tests/integration/test_concurrent_operations.py \
+	tests/integration/test_per_key_encryption.py \
+	tests/integration/test_redis_coordination.py
+INTEGRATION_multipart_TESTS = \
+	tests/integration/test_upload_part_copy.py \
+	tests/integration/test_part_ordering.py \
+	tests/integration/test_sequential_part_numbering.py \
+	tests/integration/test_sequential_part_numbering_e2e.py \
+	tests/integration/test_large_file_streaming.py \
+	tests/integration/test_entity_too_small_errors.py \
+	tests/integration/test_entity_too_small_fix.py \
+	tests/integration/test_partial_complete_fix.py \
+	tests/integration/test_multipart_range_validation.py
+INTEGRATION_copy_range_TESTS = \
+	tests/integration/test_copy_passthrough.py \
+	tests/integration/test_download_range_requests.py \
+	tests/integration/test_get_prefetch.py \
+	tests/integration/test_elasticsearch_range_scenario.py \
+	tests/integration/test_state_recovery.py \
+	tests/integration/test_state_recovery_e2e.py
+INTEGRATION_misc_TESTS = \
+	tests/integration/test_delete_objects_errors.py \
+	tests/integration/test_metadata_and_errors.py
+
 # Run integration tests (needs minio/redis containers)
 test-integration:
 	@docker compose -f tests/docker-compose.yml down 2>/dev/null || true
@@ -22,6 +54,22 @@ test-integration:
 		docker compose -f tests/docker-compose.yml down; \
 		exit $$EXIT_CODE
 
+# Run one integration shard (CI matrix). SHARD=memory_usage|memory_leak|memory_copy|...
+test-integration-shard:
+ifndef SHARD
+	$(error SHARD is required, e.g. make test-integration-shard SHARD=memory)
+endif
+	@docker compose -f tests/docker-compose.yml down 2>/dev/null || true
+	@docker compose -f tests/docker-compose.yml up -d
+	@sleep 3
+	@AWS_ACCESS_KEY_ID=minioadmin AWS_SECRET_ACCESS_KEY=minioadmin \
+		uv run pytest -m "e2e" -v \
+		$(if $(INTEGRATION_$(SHARD)_PYTEST_OPTS),$(INTEGRATION_$(SHARD)_PYTEST_OPTS),-n auto --dist loadgroup) \
+		$(INTEGRATION_$(SHARD)_TESTS); \
+		EXIT_CODE=$$?; \
+		docker compose -f tests/docker-compose.yml down; \
+		exit $$EXIT_CODE
+
 # Run all tests with containers (unit + integration)
 test-all:
 	@docker compose -f tests/docker-compose.yml down 2>/dev/null || true

s3proxy/concurrency.py

@@ -12,7 +12,11 @@
 
 import structlog
 
-from s3proxy.crypto import governor_memory_footprint, streaming_governor_clamped_reserve
+from s3proxy.crypto import (
+    copy_governor_clamped_reserve,
+    governor_memory_footprint,
+    streaming_governor_clamped_reserve,
+)
 from s3proxy.errors import S3Error
 from s3proxy.metrics import MEMORY_LIMIT_BYTES, MEMORY_REJECTIONS, MEMORY_RESERVED_BYTES
 
@@ -80,7 +84,7 @@ def set_memory_limit(self, limit_mb: int) -> None:
         self._limit_bytes = limit_mb * 1024 * 1024
         MEMORY_LIMIT_BYTES.set(self._limit_bytes)
 
-    async def try_acquire(self, bytes_needed: int) -> int:
+    async def try_acquire(self, bytes_needed: int, *, copy: bool = False) -> int:
         """Reserve memory, waiting up to BACKPRESSURE_TIMEOUT if at capacity."""
         if self._limit_bytes <= 0:
             return 0
@@ -93,14 +97,23 @@ async def try_acquire(self, bytes_needed: int) -> int:
         # parts are not starved behind one clamped slot. Rare huge uploads may use
         # more RSS than reserved when run alone; the pod memory limit is the
         # backstop.
+        #
+        # Server-side copies use copy_governor_clamped_reserve instead: their
+        # honest peak already reflects real chunk work and must not be crushed to
+        # the routine upload peak (~59MB) or several multi-GB manifest copies run
+        # concurrently and OOM the pod.
         if to_reserve > self._limit_bytes:
             honest = to_reserve
-            to_reserve = streaming_governor_clamped_reserve(honest, self._limit_bytes)
+            if copy:
+                to_reserve = copy_governor_clamped_reserve(honest, self._limit_bytes)
+            else:
+                to_reserve = streaming_governor_clamped_reserve(honest, self._limit_bytes)
             logger.info(
                 "MEMORY_CLAMPED_TO_BUDGET",
                 requested_mb=round(honest / 1024 / 1024, 2),
                 reserved_mb=round(to_reserve / 1024 / 1024, 2),
                 limit_mb=round(self._limit_bytes / 1024 / 1024, 2),
+                copy=copy,
             )
 
         async with self._condition:
@@ -197,6 +210,10 @@ async def try_acquire_memory(bytes_needed: int) -> int:
     return await _default.try_acquire(bytes_needed)
 
 
+async def try_acquire_copy_memory(bytes_needed: int) -> int:
+    return await _default.try_acquire(bytes_needed, copy=True)
+
+
 @contextlib.asynccontextmanager
 async def reserve_memory(bytes_needed: int):
     """Reserve memory for the duration of a block, releasing on exit.
@@ -213,6 +230,17 @@ async def reserve_memory(bytes_needed: int):
             await _default.release(reserved)
 
 
+@contextlib.asynccontextmanager
+async def reserve_copy_memory(bytes_needed: int):
+    """Reserve memory for a server-side copy (upload-style clamp does not apply)."""
+    reserved = await _default.try_acquire(bytes_needed, copy=True)
+    try:
+        yield
+    finally:
+        if reserved > 0:
+            await _default.release(reserved)
+
+
 async def release_memory(bytes_reserved: int) -> None:
     await _default.release(bytes_reserved)
 

s3proxy/crypto.py

@@ -207,6 +207,17 @@ def streaming_governor_clamped_reserve(honest_peak: int, budget_bytes: int) -> i
     return min(honest_peak, routine_peak, budget_bytes)
 
 
+def copy_governor_clamped_reserve(honest_peak: int, budget_bytes: int) -> int:
+    """Reservation when a copy's honest peak exceeds the governor budget.
+
+    Unlike uploads, a multi-GB copy's honest peak reflects real per-chunk work
+    (238MB internal parts for a 4.7GB Scylla manifest). Clamping to the routine
+    upload peak (~59MB) under-reserves and admits several concurrent copies that
+    each need ~500MB+ RSS. Monopolize the budget slot instead.
+    """
+    return min(honest_peak, budget_bytes)
+
+
 def governor_memory_footprint(content_length: int) -> int:
     """Memory to reserve for a framed upload at the request gate.
 
@@ -233,9 +244,16 @@ def copy_pipeline_peak(plaintext_size: int) -> int:
     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).
+
+    Large internal parts (multi-GB manifest copies): aiobotocore copies the
+    ciphertext body for signing; tracemalloc shows ~part/7 slack at 238MB parts.
     """
     if plaintext_size > STREAMING_THRESHOLD:
-        return streaming_upload_peak(plaintext_size) + 2 * MAX_BUFFER_SIZE
+        part = memory_bounded_part_size(plaintext_size)
+        peak = streaming_upload_peak(plaintext_size) + 2 * MAX_BUFFER_SIZE
+        if part > 32 * 1024 * 1024:
+            peak += part // 7
+        return peak
     return max(MAX_BUFFER_SIZE, 3 * plaintext_size)
 
 

s3proxy/handlers/multipart/copy.py

@@ -62,7 +62,8 @@ async def handle_upload_part_copy(self, request: Request, creds: S3Credentials)
                 # Small copies buffer the whole object + re-encrypt it; gate them
                 # by the limiter too (they carry no body, so the request-level
                 # reservation was ~nothing and a small-object flood ran unbounded).
-                async with concurrency.reserve_memory(crypto.copy_pipeline_peak(plaintext_size)):
+                peak = crypto.copy_pipeline_peak(plaintext_size)
+                async with concurrency.reserve_copy_memory(peak):
                     return await self._simple_copy_part(
                         client,
                         bucket,
@@ -221,7 +222,8 @@ async def _streaming_copy_part(
         # 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).
-        async with concurrency.reserve_memory(crypto.copy_pipeline_peak(plaintext_size)):
+        peak = crypto.copy_pipeline_peak(plaintext_size)
+        async with concurrency.reserve_copy_memory(peak):
             return await self._streaming_copy_part_inner(
                 client,
                 bucket,

s3proxy/handlers/objects/misc.py

@@ -386,7 +386,7 @@ async def _copy_encrypted(
         else:
             _s = head_resp.get("Metadata", {}).get("plaintext-size")
             pt_size = int(_s) if _s else crypto.plaintext_size(head_resp.get("ContentLength", 0))
-        async with concurrency.reserve_memory(crypto.copy_pipeline_peak(pt_size)):
+        async with concurrency.reserve_copy_memory(crypto.copy_pipeline_peak(pt_size)):
             return await self._copy_encrypted_inner(
                 client,
                 bucket,