Fix multipart integrity and reduce encrypted read overhead

Makefile

@@ -26,10 +26,13 @@ verify-passthrough:
 
 # Integration shards for parallel CI (make test-integration-shard SHARD=memory_usage)
 INTEGRATION_memory_usage_TESTS = tests/integration/test_memory_usage.py
+INTEGRATION_memory_usage_PYTEST_OPTS = -n0
+INTEGRATION_memory_usage_COMPOSE_OPTS = -f tests/docker-compose.oom.yml
 INTEGRATION_memory_leak_TESTS = tests/integration/test_memory_leak.py
 INTEGRATION_memory_copy_TESTS = tests/integration/test_copy_memory_governor.py tests/integration/test_copy_per_part_metrics.py tests/integration/test_upload_part_copy_passthrough_e2e.py
 INTEGRATION_memory_copy_PYTEST_OPTS = -n0
 INTEGRATION_core_TESTS = \
+	tests/integration/test_generation_roundtrip.py \
 	tests/integration/test_integration.py \
 	tests/integration/test_handlers.py \
 	tests/integration/test_concurrent_operations.py \
@@ -72,15 +75,15 @@ 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
+	@docker compose -f tests/docker-compose.yml $(INTEGRATION_$(SHARD)_COMPOSE_OPTS) down 2>/dev/null || true
+	@docker compose -f tests/docker-compose.yml $(INTEGRATION_$(SHARD)_COMPOSE_OPTS) 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; \
+		docker compose -f tests/docker-compose.yml $(INTEGRATION_$(SHARD)_COMPOSE_OPTS) down; \
 		exit $$EXIT_CODE
 
 # Run all tests with containers (unit + integration)

README.md

@@ -21,6 +21,10 @@
 
 ## Overview
 
+For the generation-format upgrade, migration requirements and measured performance, see
+[Generation-bound writes and streaming changes](docs/GENERATION_FORMAT.md).
+
+
 S3's server-side encryption is great, but your cloud provider holds the keys. S3Proxy sits between your app and S3, encrypting everything **before** it leaves your infrastructure.
 
 ```

docs/GENERATION_FORMAT.md

@@ -0,0 +1,63 @@
+# Generation-bound writes and streaming changes
+
+This change implements the correctness findings and performance work from [the code review](CODE_REVIEW.md). It changes the format of newly written objects. Read this document before upgrading a running cluster.
+
+## Publication and retries
+
+New buffered PUTs carry `s3proxy-format=single-v3`. Streaming PUTs, multipart uploads and streaming copies carry `s3proxy-format=multipart-v3` and an immutable generation pointer. Required manifests are stored at `.s3proxy-internal/generations/<generation>.meta` **before** publishing ciphertext. The generation is derived from the initial wrapped random upload key; it remains the same if the first copy selects a source key before any writer starts. Readers resolve the pointer from the current HEAD response. A missing, corrupt or unavailable required manifest is an error, never a signal to return ciphertext as plaintext.
+
+Each UploadPart attempt writes a private staging object at `.s3proxy-internal/attempts/<generation>/<attempt>`. Hash/signature validation and completion of that staging object precede publication in upload state. A rejected replacement cannot modify an accepted attempt. When a state write has an uncertain outcome, its completed staging object is retained because the state write may already have succeeded.
+
+Complete reads the accepted client-part mapping, validates the client's ordered ETag list, and assembles the selected staging objects in client order with server-side copies. State remains available until completion succeeds. The manifest records the originating bucket, key, upload ID and client ETag, so size alone cannot prove that a retry succeeded. Redis completion locks renew their leases; loss of a lease interrupts the operation. In-memory lock entries disappear after the last waiter finishes.
+
+The first writer atomically freezes the upload's DEK. A whole-object UploadPartCopy may select its source DEK at that point and snapshot the ciphertext using a source ETag precondition. Subsequent copies using a different key, partial ranges and ordinary uploads encrypt with the already frozen key. No active writer can change that key. New encryption uses fresh random nonces; transport retries reuse the already sealed bytes. Full CopyObject retains native copying and publishes the corresponding manifest first.
+
+## Compatibility and deployment
+
+- Existing single-seal and framed objects remain readable. The existing 8 MiB encryption frame boundary is unchanged. Legacy multipart ETags retain their historical fallback where no stored client ETag exists.
+- Upgrade all readers before allowing v3 writes. Old releases do not understand the new generation pointers. Do not run old and new writers against the same keys. Use a maintenance window to drain existing uploads and switch the fleet together.
+- Drain or restart legacy in-flight multipart uploads. UploadPart and UploadPartCopy reject the old active layout. Do not rely on completing legacy active state after the upgrade.
+- Use persistent Redis for uploads that must survive process restarts or move between pods. Configure its TTL above the maximum permitted upload duration. Loss of v3 upload state fails closed: accepted attempts are not reconstructed by guessing from backend parts. Committed objects do not depend on Redis.
+- Keep the manifest namespace available to the same backend credentials that operate on objects. The existing internal-prefix filter hides it from proxy listings. Backend permissions now also need multipart creation/copy, listing and deletion for the attempt prefix.
+- `If-None-Match: *` is passed to the backend publication operation, including CompleteMultipartUpload for streaming PUT. Verify support on the target S3-compatible service. Failed preconditions return 412.
+- Supported signed streaming mode is `STREAMING-AWS4-HMAC-SHA256-PAYLOAD`. Its entire signature chain, terminal chunk and decoded length are checked. Unsupported trailer/signature modes are explicitly rejected. Clients using trailer checksums must select a supported encoding until that protocol is implemented.
+- Control request bodies are bounded at 8 MiB. Client parts are bounded at 5 GiB; assembled uploads remain subject to backend object-size and 10,000-part limits. Manifest JSON is bounded at 10 MiB and rejected before publication when it exceeds the reader limit.
+
+## Storage cleanup
+
+Successful completion and abort attempt to delete all completed staging objects for that upload generation. Replaced attempts remain immutable until terminal cleanup to avoid racing a concurrent completion snapshot. Configure a backend lifecycle rule for `.s3proxy-internal/attempts/` to expire completed orphan attempts and abort incomplete staging MPUs after a period **longer than the maximum supported upload duration and retry window**. For example, seven days is appropriate only if the deployment prohibits uploads lasting that long. This PR does not install or modify bucket lifecycle policies.
+
+Crashes, cancelled writes and failed cleanup can leave attempts or unpublished manifests. Generation manifests are intentionally retained: native copies and backend object versions may reference them. Do not apply an age-only deletion policy to `.s3proxy-internal/generations/`. Reclaiming those manifests requires checking all retained object versions, copy references and active uploads. Automatic generation garbage collection is outside this change; retaining metadata is the safe default.
+
+## Read path, resources and performance
+
+GET, HEAD, LIST and CopyObject share an object descriptor for plaintext size, ETag and manifest interpretation. GET reuses the initial HEAD. Consecutive frames share backend range requests of up to 64 MiB while buffering and authenticating one frame at a time. Retry resumes at the first unpublished frame. Output is emitted in bounded chunks. Legacy large single seals authenticate to a temporary spool before any plaintext is exposed; configure sufficient local temporary disk for concurrent legacy reads.
+
+The frame reader allocates exact-size buffers instead of repeatedly growing bytearrays. This was necessary to prevent allocator fragmentation observed during repeated real HTTP reads. GC and allocator trimming no longer run synchronously on each memory release. GET reserves 32 MiB for the working set, and copy operations own their reservation instead of nesting it inside another reservation.
+
+An application-owned pool retains up to 32 credential-isolated S3 clients, evicts idle entries, and waits for active leases on shutdown. Response ownership keeps bodies, clients and reservations alive until streaming finishes, including failure before the body starts. Metrics finish with the stream; dashboard metric failures cannot prevent reservation release. Concurrent listing lookups for the same object/ETag coalesce and recheck the existing bounded attribute cache. Cold listings still need metadata reads for previously unseen objects; this is not a bucket-wide metadata index.
+
+Staging adds temporary storage and server-side copy work to ordinary multipart writes. Full compatible copies avoid re-encryption; partial copies and copies after an incompatible key selection use bounded re-encryption. The change prioritizes verified publication over overwriting unverified backend parts. It does not promise that every write workload gets faster.
+
+## Measured results
+
+A local macOS/Python 3.14.7 test compared `d45732d` with this implementation against isolated Docker MinIO. Each process wrote a 32 MiB object, performed one warm-up GET, then ten sequential GETs. A fresh process was used for each version. Requests used the same client configuration and verified every returned byte.
+
+| Metric | Before | After |
+| --- | ---: | ---: |
+| Median GET latency | 317.12 ms | 289.26 ms |
+| Maximum observed latency | 332.85 ms | 331.64 ms |
+| Median transfer rate | 100.91 MiB/s | 110.63 MiB/s |
+| Highest sampled process RSS | 179.23 MiB | 175.06 MiB |
+
+This is about 8.8% lower median latency in this small local sample, not a production forecast. [Raw results](read-benchmark.json) are included. A separate 50-request run stayed below 176 MiB sampled RSS. Sampling occurred after requests and can miss transient peaks; the governor budget is not a hard process RSS limit. No p99 claim or AWS/Ceph throughput claim is made.
+
+Regression tests separately demonstrate eight consecutive frames using one backend GET, frame-aligned recovery without duplicate plaintext, AWS-published streaming-signature vectors, corrupt legacy seals, cancelled publication, failed completion retry, immutable key selection, coalesced listings and resource cleanup on failed response headers.
+
+## Validation and remaining rollout checks
+
+Local validation: the CI unit selection passed 779 tests, including 110 mock integration tests (two upstream deprecation warnings). The separate full unit-directory run passed 670 tests, including its slow case. Real-backend validation passed 27 compatibility tests, 11 copy/concurrency tests and nine native-copy tests covering 1,280 MiB objects and concurrent copies. Ruff lint and formatting checks passed. CI repeats the Linux integration shards; the memory-usage shard runs serially, removes all data from each unique test bucket and uses disk-backed MinIO. Its previous 4 GiB tmpfs filled during the multi-gigabyte staging workload; proxy memory limits and memory assertions are unchanged.
+
+Run `uv run pytest tests/unit -q`, `uv run ruff check .` and `uv run ruff format --check .`. `tests/integration/test_generation_roundtrip.py` exercises real HTTP and MinIO, including hash tampering below/at/above the buffering threshold, generation overwrite, conditional writes, ListParts, native copying and UploadPartCopy. Set `S3PROXY_TEST_REDIS_URL` to a dedicated test Redis to repeat the scenario with durable state. `S3PROXY_TEST_BACKEND` selects the backend for the shared integration fixtures.
+
+Before production rollout, run the target backend's compatibility suite and workload benchmarks with its real latency, upload sizes, concurrency, lifecycle policy and pod limits. Existing benchmark/compatibility tests for legacy copy internals remain explicitly separate from the public v3 route. No production deployment or bucket policy changes are part of this PR.

docs/read-benchmark.json

@@ -0,0 +1,18 @@
+{
+  "baseline": {
+    "samples": 10,
+    "bytes": 33554432,
+    "median_ms": 317.12,
+    "max_ms": 332.85,
+    "median_MiB_s": 100.91,
+    "sampled_peak_RSS_MiB": 179.23
+  },
+  "changed": {
+    "samples": 10,
+    "bytes": 33554432,
+    "median_ms": 289.26,
+    "max_ms": 331.64,
+    "median_MiB_s": 110.63,
+    "sampled_peak_RSS_MiB": 175.06
+  }
+}

s3proxy/app.py

@@ -200,14 +200,16 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
         tracemalloc_task = _maybe_start_tracemalloc()
 
-        yield
-
-        if tracemalloc_task is not None:
-            tracemalloc_task.cancel()
-        await stats_store.aclose()  # flush buffered samples before Redis closes
-        await close_redis()
-        await close_http_client()
-        logger.info("Shutting down")
+        try:
+            yield
+        finally:
+            if tracemalloc_task is not None:
+                tracemalloc_task.cancel()
+            await stats_store.aclose()  # flush buffered samples before Redis closes
+            await close_redis()
+            await close_http_client()
+            await handler.client_pool.close()
+            logger.info("Shutting down")
 
     return lifespan
 

s3proxy/client/pool.py

@@ -0,0 +1,51 @@
+"""Bounded credential-isolated S3 client pool owned by one application lifespan."""
+
+import asyncio
+from contextlib import asynccontextmanager
+
+from ..errors import S3Error
+from .s3 import S3Client
+
+
+class S3ClientPool:
+    def __init__(self, settings, max_clients=32):
+        self.settings = settings
+        self.max_clients = max_clients
+        self.entries = {}
+        self.condition = asyncio.Condition()
+        self.closed = False
+
+    @asynccontextmanager
+    async def acquire(self, credentials):
+        key = (credentials.access_key, credentials.secret_key, credentials.region)
+        async with self.condition:
+            if self.closed:
+                raise S3Error.slow_down("S3 client pool is shutting down")
+            if key not in self.entries:
+                if len(self.entries) >= self.max_clients:
+                    idle = next((k for k, (_, refs) in self.entries.items() if refs == 0), None)
+                    if idle is None:
+                        raise S3Error.slow_down("S3 client pool is busy")
+                    client, _ = self.entries.pop(idle)
+                    await client.__aexit__(None, None, None)
+                client = S3Client(self.settings, credentials)
+                await client.__aenter__()
+                self.entries[key] = [client, 0]
+            entry = self.entries[key]
+            entry[1] += 1
+        try:
+            yield entry[0]
+        finally:
+            async with self.condition:
+                entry[1] -= 1
+                self.condition.notify_all()
+
+    async def close(self):
+        async with self.condition:
+            self.closed = True
+            await self.condition.wait_for(
+                lambda: all(refs == 0 for _, refs in self.entries.values())
+            )
+            entries, self.entries = self.entries, {}
+        for client, _ in entries.values():
+            await client.__aexit__(None, None, None)

s3proxy/client/s3.py

@@ -41,7 +41,7 @@ class S3Client:
 
     Memory management:
     - Uses a shared aioboto3 Session to avoid repeated JSON model loading
-    - Creates fresh clients per request for proper connection cleanup
+    - Clients are leased from the credential-isolated application pool
     - Each session load costs ~30-150MB (botocore service definitions)
 
     See: https://github.com/boto/boto3/issues/1670
@@ -118,6 +118,7 @@ async def put_object(
         tagging: str | None = None,
         cache_control: str | None = None,
         expires: str | None = None,
+        if_none_match: str | None = None,
     ) -> dict[str, Any]:
         """Put object to S3."""
         kwargs: dict[str, Any] = {"Bucket": bucket, "Key": key, "Body": body}
@@ -128,6 +129,7 @@ async def put_object(
             Tagging=tagging,
             CacheControl=cache_control,
             Expires=expires,
+            IfNoneMatch=if_none_match,
         )
         return await self._cached_client.put_object(**kwargs)
 
@@ -213,6 +215,7 @@ async def complete_multipart_upload(
         key: str,
         upload_id: str,
         parts: list[dict[str, Any]],
+        if_none_match: str | None = None,
     ) -> dict[str, Any]:
         """Complete multipart upload."""
         start = time.monotonic()
@@ -221,6 +224,7 @@ async def complete_multipart_upload(
             Key=key,
             UploadId=upload_id,
             MultipartUpload={"Parts": parts},
+            **({"IfNoneMatch": if_none_match} if if_none_match is not None else {}),
         )
         duration = time.monotonic() - start
         logger.info(
@@ -284,6 +288,9 @@ async def copy_object(
         content_type: str | None = None,
         tagging_directive: str | None = None,
         tagging: str | None = None,
+        copy_source_if_match: str | None = None,
+        cache_control: str | None = None,
+        expires=None,
     ) -> dict[str, Any]:
         """Copy object within S3."""
         kwargs: dict[str, Any] = {
@@ -300,6 +307,12 @@ async def copy_object(
             kwargs["TaggingDirective"] = tagging_directive
         if tagging and tagging_directive == "REPLACE":
             kwargs["Tagging"] = tagging
+        _add_optional_kwargs(
+            kwargs,
+            CopySourceIfMatch=copy_source_if_match,
+            CacheControl=cache_control,
+            Expires=expires,
+        )
         return await self._cached_client.copy_object(**kwargs)
 
     async def delete_objects(
@@ -399,6 +412,7 @@ async def upload_part_copy(
         part_number: int,
         copy_source: str,
         copy_source_range: str | None = None,
+        copy_source_if_match: str | None = None,
     ) -> dict[str, Any]:
         """Copy a part from another object."""
         kwargs: dict[str, Any] = {
@@ -408,5 +422,7 @@ async def upload_part_copy(
             "PartNumber": part_number,
             "CopySource": copy_source,
         }
-        _add_optional_kwargs(kwargs, CopySourceRange=copy_source_range)
+        _add_optional_kwargs(
+            kwargs, CopySourceRange=copy_source_range, CopySourceIfMatch=copy_source_if_match
+        )
         return await self._cached_client.upload_part_copy(**kwargs)