From 99b2eac3458e5d189a4640f1d8449e6d289c3bc0 Mon Sep 17 00:00:00 2001 From: serversidehannes Date: Tue, 7 Jul 2026 08:43:31 +0200 Subject: [PATCH] fix: make server-side copies O(1) memory so they stop 503-storming Root cause of the Scylla backup 503 storm (~50% of requests): a dedup UploadPartCopy of a 4.7GB SSTable (single client part, partNumber=1) sized its internal S3 parts as object_size/MAX_INTERNAL_PARTS_PER_CLIENT (=object/20 = 238MB), so the copy pipeline peaked at ~535MB -- larger than the 192MB governor budget. copy_governor_clamped_reserve then clamped the reservation to the WHOLE budget, making the copy exclusive: it could only be admitted when active_bytes hit exactly 0. Under steady backup traffic (baseline reservations of other in-flight requests keep active_bytes > 0) that never happens, so every large copy backpressured for the full timeout and returned 503 SlowDown. Confirmed in prod: MEMORY_CLAMPED_TO_BUDGET copy=true requested_mb=534, MEMORY_REJECTED at active_mb=0.06-0.31, 296x503 vs 304x200. Writer preference (#112) could not fix this: it holds back *new* small requests but cannot evict baseline reservations already held, and a single copy deadlocks even against its own request-gate reservation. Fix: copies frame at a FIXED internal part size (crypto.copy_internal_part_size, 32MB) instead of object_size/20, so the copy peak is O(1) in object size (~90MB) regardless of how large the object is. A copy now fits the budget outright, never clamps, never runs exclusive, and simply reserves ~90MB like any other request -- multiple copies run concurrently, bounded by the budget (so the multi-copy OOM that motivated exclusivity can't happen either, because each copy is light). A large copy needs many internal parts, so allocate them sequentially (client_part_number=0) rather than from the fixed 20-wide per-client window. Safe: copies are single-part uploads (verified 291/291 partNumber=1 in prod) and reassembly is metadata-driven (get.py sorts by internal_part_number and uses stored ciphertext sizes), so smaller/more parts and sequential numbering need no migration and read back byte-identically. Part size grows only past S3's 10k-part ceiling (>288GB objects, which backup SSTables never hit). Tests: rewrote the copy-governor tests that encoded the old 'monopolize the budget' design to assert the new invariants (peak is O(1) and size-independent, copy never clamps, N copies run concurrently within budget, copy coexists with uploads). Added test_large_copy_admitted_while_baseline_reservations_held which reproduces the exact prod deadlock and is verified to fail against the old O(size) sizing. Full unit suite green. --- s3proxy/crypto.py | 45 +++- s3proxy/handlers/multipart/copy.py | 13 +- tests/unit/test_copy_clamp_and_concurrency.py | 233 +++++++++--------- tests/unit/test_copy_memory_governing.py | 22 +- tests/unit/test_governor_peak_and_clamp.py | 19 +- tests/unit/test_streaming_copy.py | 14 +- 6 files changed, 190 insertions(+), 156 deletions(-) diff --git a/s3proxy/crypto.py b/s3proxy/crypto.py index 5d499b5..6f615e4 100644 --- a/s3proxy/crypto.py +++ b/s3proxy/crypto.py @@ -42,6 +42,20 @@ # than this many internal parts without colliding into the next part's range. MAX_INTERNAL_PARTS_PER_CLIENT = 20 +# Server-side copies use a FIXED internal part size instead of object_size/20. +# A large single-part copy (Scylla dedup copies a 4.7GB SSTable as one +# UploadPartCopy) otherwise picked 238MB internal parts -> ~535MB peak -> it had +# to reserve the whole governor budget and deadlocked against any concurrent +# request (active_bytes never hits 0). A fixed part keeps the copy peak ~90MB +# regardless of object size, so copies never monopolize the budget. Copies are +# single-part and their internal parts are allocated sequentially, so a large +# copy can use many internal parts without the per-client-range collision that +# limits streaming uploads. +COPY_INTERNAL_PART_SIZE = 32 * 1024 * 1024 # 32 MB +# Keep clear of S3's 10,000-part ceiling; only grow the part for objects so +# large a 32MB part would exceed it (>~288GB), which backup SSTables never hit. +COPY_MAX_INTERNAL_PARTS = 9000 + # Framed internal-part format. # S3's 10,000-part limit forces large internal parts for big objects, but we do # not want to hold a whole part (plaintext + ciphertext) in memory. So a part is @@ -233,24 +247,41 @@ def governor_memory_footprint(content_length: int) -> int: return min(honest, routine_cap) +def copy_internal_part_size(plaintext_size: int) -> int: + """Internal S3 part size for a server-side copy. + + Fixed at COPY_INTERNAL_PART_SIZE so a copy's peak memory is bounded and + independent of object size (the pump frames one internal part at a time). + Grown only when an object is large enough that a 32MB part would need more + than COPY_MAX_INTERNAL_PARTS parts. + """ + parts = -(-plaintext_size // COPY_INTERNAL_PART_SIZE) # ceil + if parts > COPY_MAX_INTERNAL_PARTS: + return -(-plaintext_size // COPY_MAX_INTERNAL_PARTS) # ceil, grow part + return COPY_INTERNAL_PART_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 now frames exactly like the framed UploadPart path - (encrypt_frame per FRAME_PLAINTEXT_SIZE, one internal part at a time, sized by - memory_bounded_part_size), so its per-part peak matches the upload path. It + 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). - Large internal parts (multi-GB manifest copies): aiobotocore copies the - ciphertext body for signing; tracemalloc shows ~part/7 slack at 238MB parts. + aiobotocore copies the ciphertext body for signing; tracemalloc shows + ~part/7 slack once parts exceed 32MB. """ if plaintext_size > STREAMING_THRESHOLD: - part = memory_bounded_part_size(plaintext_size) - peak = streaming_upload_peak(plaintext_size) + 2 * MAX_BUFFER_SIZE + 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 diff --git a/s3proxy/handlers/multipart/copy.py b/s3proxy/handlers/multipart/copy.py index 161df1b..3bdccca 100644 --- a/s3proxy/handlers/multipart/copy.py +++ b/s3proxy/handlers/multipart/copy.py @@ -259,14 +259,17 @@ async def _streaming_copy_part_inner( plaintext_size: int, ) -> Response: """Stream-decrypt the source and encrypt+upload each chunk as an internal S3 part.""" - # Same part sizing as the framed UploadPart path: bounds the internal part - # count to MAX_INTERNAL_PARTS_PER_CLIENT (so a large copy can never overflow - # the per-client internal-part range) and keeps parts small. - chunk_size = crypto.memory_bounded_part_size(plaintext_size) + # Fixed internal part size: the copy peak is O(1) in object size (~90MB) + # instead of object_size/20 (~535MB for a 4.7GB SSTable, which had to + # reserve the whole governor budget and deadlocked). A large copy needs + # many internal parts, so allocate them sequentially (client_part_number=0) + # rather than from the fixed 20-wide per-client window -- safe because + # copies are single-part uploads and reassembly is metadata-driven. + chunk_size = crypto.copy_internal_part_size(plaintext_size) estimated_parts = max(1, math.ceil(plaintext_size / chunk_size)) internal_part_start = await self.multipart_manager.allocate_internal_parts( - bucket, key, upload_id, estimated_parts, client_part_number=part_num + bucket, key, upload_id, estimated_parts, client_part_number=0 ) logger.info( diff --git a/tests/unit/test_copy_clamp_and_concurrency.py b/tests/unit/test_copy_clamp_and_concurrency.py index e1d8e10..4924be8 100644 --- a/tests/unit/test_copy_clamp_and_concurrency.py +++ b/tests/unit/test_copy_clamp_and_concurrency.py @@ -1,8 +1,22 @@ -"""Regression tests for copy-path governor clamp and concurrent manifest copies. - -Prod OOM: three concurrent 4.7GB Scylla .sm_ manifest UploadPartCopy ops on -one pod each reserved ~59MB (upload-style routine clamp) while holding ~175MB -ciphertext at copy.py:420 — governor active_mb ~178, RSS ~790, OOMKilled. +"""Server-side copy memory: bounded, concurrent, and no full-budget deadlock. + +History of this path: + * Copies decrypt+re-encrypt in RAM. A dedup flood of unthrottled copies OOMed + the pod, so copies were gated by the memory limiter (test_copy_memory_governing). + * The copy pipeline sized its internal parts as object_size/20, so a 4.7GB + Scylla `.sm_` dedup copy (a single UploadPartCopy, partNumber=1) peaked at + ~535MB -- larger than the 192MB governor budget. It therefore reserved the + WHOLE budget to run exclusively, and could only be admitted when active_bytes + hit exactly 0. Under steady backup traffic that never happens, so every large + copy backpressured for the full timeout and returned 503 (observed in prod: + ~50% of requests 503, all MEMORY_CLAMPED_TO_BUDGET copy=true requested_mb=534, + rejected at active_mb≈0.1). + +The fix: copies frame at a FIXED internal part size (crypto.copy_internal_part_size), +so the copy peak is O(1) in object size (~90MB). A copy never exceeds the budget, +never clamps, never runs exclusive -- it is just a normal ~90MB reservation. +Concurrency is then bounded by the budget like any other request, and the OOM +that motivated exclusivity can't happen because each copy is light. """ from __future__ import annotations @@ -18,82 +32,131 @@ MB = 1024 * 1024 -# Observed prod manifest copy size. +# Observed prod manifest copy size (single-part UploadPartCopy of a large SSTable). SCYLLA_MANIFEST_BYTES = 4_767 * MB -def test_copy_clamp_monopolizes_budget_not_routine_upload_peak(): - """4.7GB copy honest peak ~500MB at 192MB budget must reserve 192, not ~59.""" +def test_copy_peak_is_o1_across_object_sizes(): + """The core guarantee: copy peak does not grow with object size.""" + peaks = {mb: crypto.copy_pipeline_peak(mb * MB) for mb in (33, 64, 512, 4767, 20_000, 100_000)} + # All well under a 192MB budget, and identical until the 10k-part ceiling + # forces a larger part for enormous (>288GB) objects. + for mb, peak in peaks.items(): + assert peak < 128 * MB, f"{mb}MB copy peak {peak / MB:.0f}MB not bounded" + assert peaks[64] == peaks[4767] == peaks[100_000] # 100GB < 288GB ceiling + + +def test_copy_internal_part_size_fixed_until_part_ceiling(): + assert crypto.copy_internal_part_size(4767 * MB) == crypto.COPY_INTERNAL_PART_SIZE + assert crypto.copy_internal_part_size(100 * 1024 * MB) == crypto.COPY_INTERNAL_PART_SIZE + # >288GB: part grows to keep internal part count under S3's 10k ceiling. + huge = 400 * 1024 * MB + part = crypto.copy_internal_part_size(huge) + assert part > crypto.COPY_INTERNAL_PART_SIZE + assert -(-huge // part) <= crypto.COPY_MAX_INTERNAL_PARTS + + +def test_copy_does_not_monopolize_budget(): + """A 4.7GB copy fits the budget outright -- the governor clamp never engages.""" honest = crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) budget = 192 * MB - assert honest > budget - routine = crypto.streaming_upload_peak(crypto.STREAMING_GOVERNOR_CLIENT_PART_BYTES) - assert crypto.streaming_governor_clamped_reserve(honest, budget) == routine - assert crypto.copy_governor_clamped_reserve(honest, budget) == budget + assert honest < budget + # try_acquire only calls the clamp when honest > budget; here it must not, so + # the reservation is the honest peak, not the whole budget. + assert crypto.copy_governor_clamped_reserve(honest, budget) == honest + + +@pytest.mark.asyncio +async def test_large_copy_admitted_while_baseline_reservations_held(): + """Exact prod 503 regression. + + Pre-fix: a 4.7GB copy reserved the whole 192MB budget, so with any baseline + reservation held (other in-flight requests -> active_mb≈0.1) admission needed + active_bytes == 0 and never happened -> 30s backpressure -> 503. Post-fix the + copy reserves ~90MB and is admitted immediately alongside the baselines. + """ + original_timeout = concurrency.BACKPRESSURE_TIMEOUT + concurrency.BACKPRESSURE_TIMEOUT = 0 # if it can't admit *now*, it rejects + concurrency.reset_state() + concurrency.set_memory_limit(192) + try: + # Three other in-flight requests hold MIN_RESERVATION each (prod: ~0.18MB). + for _ in range(3): + await concurrency.try_acquire_memory(MIN_RESERVATION) + assert concurrency.get_active_memory() > 0 + + reserved = await concurrency.try_acquire_copy_memory( + crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) + ) + assert reserved < 128 * MB + assert concurrency.get_active_memory() <= 192 * MB + finally: + concurrency.BACKPRESSURE_TIMEOUT = original_timeout + concurrency.reset_state() @pytest.mark.asyncio -async def test_three_manifest_copies_do_not_run_concurrently_at_192mb(): - """Prod failure mode: 3× 4.7GB copies must not all hold slots on one pod.""" +async def test_multiple_copies_run_concurrently_bounded_by_budget(): + """Copies are no longer serialized to one; several run at once, capped by the + budget (each ~90MB), and total reserved never exceeds it -- so no OOM.""" original_timeout = concurrency.BACKPRESSURE_TIMEOUT concurrency.BACKPRESSURE_TIMEOUT = 0 concurrency.reset_state() concurrency.set_memory_limit(192) per_copy = crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) - assert per_copy > 192 * MB + assert per_copy < 192 * MB - max_inside = 0 inside = 0 + max_inside = 0 + peak_active = 0 admitted = 0 - rejected = 0 async def one_copy(): - nonlocal max_inside, inside, admitted, rejected + nonlocal inside, max_inside, peak_active, admitted try: async with concurrency.reserve_copy_memory(per_copy): inside += 1 - max_inside = max(max_inside, inside) admitted += 1 + max_inside = max(max_inside, inside) + peak_active = max(peak_active, concurrency.get_active_memory()) await asyncio.sleep(0.05) inside -= 1 except S3Error: - rejected += 1 + pass try: - await asyncio.gather(*[one_copy() for _ in range(3)]) + await asyncio.gather(*[one_copy() for _ in range(5)]) finally: concurrency.BACKPRESSURE_TIMEOUT = original_timeout concurrency.reset_state() - assert max_inside == 1, f"expected exclusive large copy, saw {max_inside} concurrent" - assert admitted == 1 - assert rejected == 2 + assert max_inside >= 2, "at least two ~90MB copies must run concurrently in 192MB" + assert peak_active <= 192 * MB, "governor must never overrun the budget (OOM guard)" + assert admitted >= 2 @pytest.mark.asyncio -async def test_manifest_copy_plus_scylla_upload_serialized_at_192mb(): - """Active ~25MB Scylla upload must block a budget-monopolizing manifest copy.""" - original_timeout = concurrency.BACKPRESSURE_TIMEOUT - concurrency.BACKPRESSURE_TIMEOUT = 0 +async def test_copy_coexists_with_scylla_upload(): + """A ~25MB Scylla upload and a ~90MB copy fit together (113MB < 192MB); the + copy no longer blocks the upload (or vice versa).""" concurrency.reset_state() concurrency.set_memory_limit(192) try: - scylla = crypto.governor_memory_footprint(50 * MB) - await concurrency.try_acquire_memory(scylla) - - with pytest.raises(S3Error): - await concurrency.try_acquire_copy_memory( - crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) - ) + scylla = await concurrency.try_acquire_memory(crypto.governor_memory_footprint(50 * MB)) + copy = await concurrency.try_acquire_copy_memory( + crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) + ) + assert scylla + copy < 192 * MB + assert concurrency.get_active_memory() == scylla + copy finally: - concurrency.BACKPRESSURE_TIMEOUT = original_timeout concurrency.reset_state() @pytest.mark.parametrize("mb", [512, 1024, 4767]) @pytest.mark.asyncio async def test_copy_reservation_bounds_real_peak_large_manifests(mb: int): - """copy_pipeline_peak must cover tracemalloc peak for prod-sized copies.""" + """copy_pipeline_peak must still cover the real tracemalloc peak of the (now + fixed-part) copy pipeline for prod-sized copies.""" plaintext_size = mb * MB real_peak = await _measure_peak(plaintext_size) reserved = crypto.copy_pipeline_peak(plaintext_size) @@ -103,103 +166,33 @@ async def test_copy_reservation_bounds_real_peak_large_manifests(mb: int): @pytest.mark.asyncio -async def test_old_upload_clamp_would_admit_three_concurrent_manifest_copies(): - """Regression guard: upload-style clamp is what caused prod OOM.""" - original_timeout = concurrency.BACKPRESSURE_TIMEOUT - concurrency.BACKPRESSURE_TIMEOUT = 0 - concurrency.reset_state() - concurrency.set_memory_limit(192) - per_copy = crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) - try: - - async def acquire_upload_clamp(): - return await concurrency.try_acquire_memory(per_copy) - - results = await asyncio.gather(*[acquire_upload_clamp() for _ in range(3)]) - assert all(r < 80 * MB for r in results) - assert sum(results) < 192 * MB - assert len(results) == 3 - finally: - concurrency.BACKPRESSURE_TIMEOUT = original_timeout - concurrency.reset_state() - - -@pytest.mark.asyncio -async def test_mixed_three_scylla_uploads_and_manifest_copy_limits_concurrency(): - """Prod replay: 3×50MB uploads + manifest copy — copy cannot take 4th slot.""" - original_timeout = concurrency.BACKPRESSURE_TIMEOUT - concurrency.BACKPRESSURE_TIMEOUT = 0 - concurrency.reset_state() - concurrency.set_memory_limit(192) - per_part = crypto.governor_memory_footprint(50 * MB) - manifest = crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) - try: - uploads = await asyncio.gather( - *[concurrency.try_acquire_memory(per_part) for _ in range(3)] - ) - assert sum(uploads) + 192 * MB > 192 * MB - with pytest.raises(S3Error): - await concurrency.try_acquire_copy_memory(manifest) - finally: - concurrency.BACKPRESSURE_TIMEOUT = original_timeout - concurrency.reset_state() - - -@pytest.mark.asyncio -async def test_pending_exclusive_copy_not_starved_by_small_requests(): - """Writer preference: a budget-monopolizing copy must not be starved by a - stream of small requests. - - The copy reserves the whole budget, so it can only be admitted when - active_bytes == 0. A small request that easily fits must NOT jump the queue - while the copy is waiting -- otherwise active_bytes never drains and the copy - backpressures forever (prod: requested_mb == limit_mb that never clears, then - 503). It must instead queue behind the pending copy. +async def test_writer_preference_still_protects_a_whole_budget_reservation(): + """Writer preference (kept as a safety net) must still hold back new small + requests while a genuinely exclusive whole-budget reservation waits, so it + is not starved. Copies are no longer exclusive, so drive it explicitly with a + reservation sized to the entire budget. """ original_timeout = concurrency.BACKPRESSURE_TIMEOUT concurrency.BACKPRESSURE_TIMEOUT = 5 concurrency.reset_state() concurrency.set_memory_limit(192) try: - # An in-flight ~25MB upload holds memory, so the copy cannot start yet. blocker = await concurrency.try_acquire_memory(crypto.governor_memory_footprint(50 * MB)) - assert 0 < blocker < 192 * MB - copy_task = asyncio.create_task( - concurrency.try_acquire_copy_memory(crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES)) - ) - # A tiny request that trivially fits alongside the blocker (25 + 0.06 << - # 192) but must be held back because an exclusive copy is pending. + # An exclusive (whole-budget) waiter. + exclusive_task = asyncio.create_task(concurrency.try_acquire_memory(192 * MB)) small_task = asyncio.create_task(concurrency.try_acquire_memory(MIN_RESERVATION)) await asyncio.sleep(0.05) + assert not exclusive_task.done() + assert not small_task.done(), "small request must queue behind the exclusive waiter" - assert not copy_task.done(), "copy must wait for exclusive access" - assert not small_task.done(), "small request must queue behind the pending copy" - - # Drain the blocker: active_bytes -> 0 -> the copy wins the slot, not the - # queued small request. await concurrency.release_memory(blocker) - reserved = await asyncio.wait_for(copy_task, timeout=2) - assert reserved == 192 * MB - assert concurrency.get_active_memory() == 192 * MB - assert not small_task.done(), "copy holds the whole budget; small still waits" + got = await asyncio.wait_for(exclusive_task, timeout=2) + assert got == 192 * MB + assert not small_task.done() - # Copy releases -> the small request finally proceeds. - await concurrency.release_memory(reserved) + await concurrency.release_memory(got) assert await asyncio.wait_for(small_task, timeout=2) == MIN_RESERVATION finally: concurrency.BACKPRESSURE_TIMEOUT = original_timeout concurrency.reset_state() - - -@pytest.mark.asyncio -async def test_copy_acquire_uses_copy_clamp_not_upload_clamp(): - concurrency.reset_state() - concurrency.set_memory_limit(192) - try: - honest = crypto.copy_pipeline_peak(SCYLLA_MANIFEST_BYTES) - reserved = await concurrency.try_acquire_copy_memory(honest) - assert reserved == 192 * MB - assert reserved > crypto.streaming_governor_clamped_reserve(honest, 192 * MB) - finally: - concurrency.reset_state() diff --git a/tests/unit/test_copy_memory_governing.py b/tests/unit/test_copy_memory_governing.py index a2a28c1..ea65179 100644 --- a/tests/unit/test_copy_memory_governing.py +++ b/tests/unit/test_copy_memory_governing.py @@ -21,17 +21,17 @@ MB = 1024 * 1024 -def test_copy_pipeline_peak_matches_framed_upload(): - # The streaming copy path now frames exactly like the framed UploadPart path - # (one internal part at a time), so its peak is identical: the old - # size-independent 4*MAX_BUFFER_SIZE under-counted the real peak ~6x -- when - # it un-framed a whole part -- and OOMed the pod. - for size in (64 * MB, 512 * MB, 5 * 1024 * MB): - part = crypto.memory_bounded_part_size(size) - base = crypto.streaming_upload_peak(size) + 2 * crypto.MAX_BUFFER_SIZE - if part > 32 * MB: - base += part // 7 - assert crypto.copy_pipeline_peak(size) == base +def test_copy_pipeline_peak_is_bounded_regardless_of_object_size(): + # Copies frame at a FIXED internal part size, so the peak is O(1) in object + # size instead of object_size/20. A 4.7GB SSTable dedup copy used to peak at + # ~535MB -- larger than the whole governor budget -- so it reserved the + # entire budget and deadlocked against any concurrent request. It now stays + # ~90MB, identical to a 64MB copy. + peaks = [crypto.copy_pipeline_peak(s) for s in (64 * MB, 512 * MB, 4767 * MB, 5 * 1024 * MB)] + assert all(p == peaks[0] for p in peaks), peaks + part = crypto.COPY_INTERNAL_PART_SIZE + assert peaks[0] == 2 * part + crypto.FRAME_PLAINTEXT_SIZE + 2 * crypto.MAX_BUFFER_SIZE + assert peaks[0] < 128 * MB def test_copy_pipeline_peak_small_is_three_x(): diff --git a/tests/unit/test_governor_peak_and_clamp.py b/tests/unit/test_governor_peak_and_clamp.py index 1c69612..0d827ae 100644 --- a/tests/unit/test_governor_peak_and_clamp.py +++ b/tests/unit/test_governor_peak_and_clamp.py @@ -378,15 +378,14 @@ async def one(): reset_state() -def test_honest_copy_peak_exceeds_upload_governor_cap(): - """copy_pipeline_peak uses per-chunk honest peak, not governor_memory_footprint. - - Upload gate caps multi-GB Content-Length at the routine workload peak; copies - reserve separately via reserve_copy_memory with copy_governor_clamped_reserve. +def test_copy_peak_is_bounded_and_independent_of_object_size(): + """Copies frame at a fixed internal part size, so copy_pipeline_peak is O(1) + in object size -- it no longer tracks streaming_upload_peak (object/20). This + is what stops a multi-GB copy monopolizing the governor budget/deadlocking. """ huge = 5 * 1024 * MB - assert crypto.copy_pipeline_peak(huge) > estimate_memory_footprint("PUT", huge) - part = crypto.memory_bounded_part_size(huge) - assert crypto.copy_pipeline_peak(huge) == ( - crypto.streaming_upload_peak(huge) + 2 * crypto.MAX_BUFFER_SIZE + part // 7 - ) + peak = crypto.copy_pipeline_peak(huge) + assert peak == crypto.copy_pipeline_peak(64 * MB) # size-independent + assert peak < 128 * MB + # far below the old object/20 peak that forced full-budget clamping + assert peak < crypto.streaming_upload_peak(huge) // 4 diff --git a/tests/unit/test_streaming_copy.py b/tests/unit/test_streaming_copy.py index 8e15b46..1aaae46 100644 --- a/tests/unit/test_streaming_copy.py +++ b/tests/unit/test_streaming_copy.py @@ -143,9 +143,13 @@ async def test_small_source_uses_simple_path(self, mock_s3, settings, manager, c @pytest.mark.asyncio async def test_large_unencrypted_source_splits_into_multiple_parts( - self, mock_s3, settings, manager, credentials + self, mock_s3, settings, manager, credentials, monkeypatch ): """Source > STREAMING_THRESHOLD → multiple internal parts.""" + # Pin the fixed copy part size to one buffer so parts stay single-frame + # and this test's per-part crypto.decrypt roundtrip applies; the 32MB + # multi-frame path is covered by test_streamed_parts_are_framed_and_roundtrip. + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) handler = _make_handler(settings, manager) await mock_s3.create_bucket("bucket") @@ -250,9 +254,10 @@ async def test_streamed_parts_are_framed_and_roundtrip( @pytest.mark.asyncio async def test_large_unencrypted_source_with_range( - self, mock_s3, settings, manager, credentials + self, mock_s3, settings, manager, credentials, monkeypatch ): """Range request on a large unencrypted source streams only the range bytes.""" + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) handler = _make_handler(settings, manager) await mock_s3.create_bucket("bucket") @@ -302,8 +307,11 @@ async def test_large_unencrypted_source_with_range( assert bytes(recovered) == plaintext[:range_size] @pytest.mark.asyncio - async def test_large_multipart_encrypted_source(self, mock_s3, settings, manager, credentials): + async def test_large_multipart_encrypted_source( + self, mock_s3, settings, manager, credentials, monkeypatch + ): """Large multipart-encrypted source → iterates source parts, re-encrypts in chunks.""" + monkeypatch.setattr(crypto, "COPY_INTERNAL_PART_SIZE", crypto.MAX_BUFFER_SIZE) handler = _make_handler(settings, manager) await mock_s3.create_bucket("bucket")