From afa065e1985398401b52be7c217b677745019b22 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 15:53:56 -0500 Subject: [PATCH 1/6] perf: delete expired Firestore states concurrently under a bound cleanup_old_states() deleted every expired document in a sequential await loop, despite a "# Delete in batch" comment claiming otherwise. Cleanup therefore cost N network round-trips and scaled linearly with the size of the expired backlog. Deletes are now fanned out with asyncio.gather() under a semaphore bounded by CLEANUP_DELETE_CONCURRENCY (16), so a large backlog cannot flood Firestore with unbounded in-flight RPCs. return_exceptions=True keeps a single failing delete from abandoning deletes already in flight, and the returned count now reflects deletes that actually succeeded rather than being lost to a propagating exception. Adds three regression tests, each verified to fail against the previous sequential implementation: - overlap test (old peak in-flight was 1) - boundedness test (peak never exceeds the configured limit) - failure-isolation test (old code propagated and skipped the rest) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../services/cloud/firestore_state.py | 39 ++++++++-- tests/unit/test_firestore_state.py | 78 +++++++++++++++++++ 2 files changed, 112 insertions(+), 5 deletions(-) diff --git a/src/youtube_extension/services/cloud/firestore_state.py b/src/youtube_extension/services/cloud/firestore_state.py index 7531aeb76..754999824 100644 --- a/src/youtube_extension/services/cloud/firestore_state.py +++ b/src/youtube_extension/services/cloud/firestore_state.py @@ -7,6 +7,7 @@ Replaces in-memory caching for cloud-native, scalable deployment. """ +import asyncio import logging import os from dataclasses import asdict, dataclass @@ -26,6 +27,11 @@ logger = logging.getLogger(__name__) +# Upper bound on concurrent delete RPCs issued by cleanup_old_states(). Cleanup +# can match an unbounded number of expired documents, so deletes are fanned out +# under a semaphore rather than dispatched all at once. +CLEANUP_DELETE_CONCURRENCY = 16 + @dataclass class VideoProcessingState: @@ -320,11 +326,34 @@ async def cleanup_old_states(self, days: int = 7) -> int: query = collection.where('created_at', '<', cutoff_date) docs = await query.get() - # Delete in batch - count = 0 - for doc in docs: - await doc.reference.delete() - count += 1 + if not docs: + logger.info(f"Cleaned up 0 old states (>{days} days)") + return 0 + + # Delete concurrently. Each delete is an independent network round-trip, + # so issuing them sequentially made cleanup cost O(n) round-trips. The + # semaphore bounds in-flight RPCs so a large backlog cannot flood + # Firestore, and return_exceptions keeps one failure from abandoning + # deletes that are already in flight. + semaphore = asyncio.Semaphore(CLEANUP_DELETE_CONCURRENCY) + + async def _delete_one(doc: Any) -> None: + async with semaphore: + await doc.reference.delete() + + results = await asyncio.gather( + *(_delete_one(doc) for doc in docs), + return_exceptions=True, + ) + + failures = [r for r in results if isinstance(r, BaseException)] + count = len(results) - len(failures) + + if failures: + logger.warning( + f"Failed to delete {len(failures)} of {len(results)} old states; " + f"first error: {failures[0]!r}" + ) logger.info(f"Cleaned up {count} old states (>{days} days)") return count diff --git a/tests/unit/test_firestore_state.py b/tests/unit/test_firestore_state.py index 5ae852a9a..57f9af763 100644 --- a/tests/unit/test_firestore_state.py +++ b/tests/unit/test_firestore_state.py @@ -15,6 +15,7 @@ from __future__ import annotations +import asyncio import sys from datetime import datetime, timedelta, timezone from pathlib import Path @@ -777,6 +778,83 @@ async def test_cleanup_custom_days_parameter(self): count = await svc.cleanup_old_states(days=30) assert count == 0 + @staticmethod + def _tracking_docs(n: int) -> tuple[list, dict]: + """Build n mock docs whose deletes record peak overlap.""" + stats = {"in_flight": 0, "peak": 0} + + async def _tracked() -> None: + stats["in_flight"] += 1 + stats["peak"] = max(stats["peak"], stats["in_flight"]) + # Yield so sibling deletes get a chance to start. Under the previous + # sequential loop nothing else could be running here. + await asyncio.sleep(0) + stats["in_flight"] -= 1 + + docs = [] + for _ in range(n): + doc = MagicMock() + doc.reference = MagicMock() + doc.reference.delete = AsyncMock(side_effect=_tracked) + docs.append(doc) + return docs, stats + + async def test_cleanup_deletes_overlap_instead_of_running_sequentially(self): + """Deletes must overlap. The sequential loop this replaced had peak overlap 1.""" + svc, _, _, _, _, coll_ref, query = _make_service_with_db() + docs, stats = self._tracking_docs(8) + query.get = AsyncMock(return_value=docs) + + count = await svc.cleanup_old_states(days=7) + + assert count == 8 + assert stats["peak"] > 1, ( + f"deletes never overlapped (peak={stats['peak']}); " + "cleanup is still issuing one round-trip at a time" + ) + + async def test_cleanup_bounds_in_flight_deletes(self): + """A large backlog must not fan out unbounded concurrent RPCs.""" + svc, _, _, _, _, coll_ref, query = _make_service_with_db() + limit = _mod.CLEANUP_DELETE_CONCURRENCY + docs, stats = self._tracking_docs(limit * 3) + query.get = AsyncMock(return_value=docs) + + count = await svc.cleanup_old_states(days=7) + + assert count == limit * 3 + assert stats["peak"] <= limit, ( + f"peak in-flight deletes {stats['peak']} exceeded the " + f"CLEANUP_DELETE_CONCURRENCY bound of {limit}" + ) + + async def test_cleanup_failure_does_not_abandon_remaining_deletes(self): + """One failing delete must not strand the rest, and must not be counted.""" + svc, _, _, _, _, coll_ref, query = _make_service_with_db() + + ok_before = MagicMock() + ok_before.reference = MagicMock() + ok_before.reference.delete = AsyncMock() + + failing = MagicMock() + failing.reference = MagicMock() + failing.reference.delete = AsyncMock(side_effect=RuntimeError("firestore boom")) + + ok_after = MagicMock() + ok_after.reference = MagicMock() + ok_after.reference.delete = AsyncMock() + + query.get = AsyncMock(return_value=[ok_before, failing, ok_after]) + + count = await svc.cleanup_old_states(days=7) + + # Only the two successful deletes are reported. + assert count == 2 + # The delete queued after the failure still ran; the sequential loop + # would have propagated the error and skipped it entirely. + ok_before.reference.delete.assert_awaited_once() + ok_after.reference.delete.assert_awaited_once() + # =========================================================================== # Module-level singleton helpers From db083b637feb8a2ff1f0c6cfc90e9399059278e5 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:02:38 -0500 Subject: [PATCH 2/6] perf: bound delete task allocation with a worker pool Review correctly identified that asyncio.gather() over a comprehension allocates one task per document *before* the semaphore can gate anything. Since the cleanup query has no limit, a large expired backlog would cost unbounded task and event-loop memory even though only 16 deletes reached Firestore at a time. Replaces the gather-plus-semaphore fan-out with a fixed pool of CLEANUP_DELETE_CONCURRENCY workers pulling from a shared iterator over the documents. Pulling with next() is safe without a lock because the event loop is single-threaded and there is no await between taking a document and using it. Worker count is min(CONCURRENCY, len(docs)), so both in-flight RPCs and allocated tasks are bounded by the same constant. Failure isolation is preserved: each worker tallies its own exceptions instead of relying on return_exceptions, so one bad delete still cannot abandon the remaining backlog. Adds test_cleanup_bounds_allocated_delete_tasks_not_just_rpcs, which measures peak len(asyncio.all_tasks()) during cleanup. Against the previous gather implementation it reports 49 concurrent tasks for 48 documents; with the worker pool it stays within the bound. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../services/cloud/firestore_state.py | 49 ++++++++++++------- tests/unit/test_firestore_state.py | 28 ++++++++++- 2 files changed, 59 insertions(+), 18 deletions(-) diff --git a/src/youtube_extension/services/cloud/firestore_state.py b/src/youtube_extension/services/cloud/firestore_state.py index 754999824..73dd9ec3f 100644 --- a/src/youtube_extension/services/cloud/firestore_state.py +++ b/src/youtube_extension/services/cloud/firestore_state.py @@ -330,28 +330,43 @@ async def cleanup_old_states(self, days: int = 7) -> int: logger.info(f"Cleaned up 0 old states (>{days} days)") return 0 - # Delete concurrently. Each delete is an independent network round-trip, - # so issuing them sequentially made cleanup cost O(n) round-trips. The - # semaphore bounds in-flight RPCs so a large backlog cannot flood - # Firestore, and return_exceptions keeps one failure from abandoning - # deletes that are already in flight. - semaphore = asyncio.Semaphore(CLEANUP_DELETE_CONCURRENCY) - - async def _delete_one(doc: Any) -> None: - async with semaphore: - await doc.reference.delete() - - results = await asyncio.gather( - *(_delete_one(doc) for doc in docs), - return_exceptions=True, + # Delete with a fixed pool of workers pulling from a shared iterator. + # Deleting sequentially made cleanup latency scale with the size of the + # expired backlog. The pool overlaps up to CLEANUP_DELETE_CONCURRENCY + # deletes at a time while keeping *both* the in-flight RPCs and the + # number of pending task objects bounded -- gather() over every document + # would allocate one task per document up front, which is unsafe for a + # query whose result set has no limit. Failures are tallied rather than + # raised so one bad delete cannot abandon the rest of the backlog. + pending = iter(docs) + succeeded = 0 + failures: list[Exception] = [] + + async def _delete_worker() -> None: + nonlocal succeeded + while True: + try: + doc = next(pending) + except StopIteration: + return + try: + await doc.reference.delete() + succeeded += 1 + except Exception as exc: # noqa: BLE001 - tallied and logged below + failures.append(exc) + + await asyncio.gather( + *( + _delete_worker() + for _ in range(min(CLEANUP_DELETE_CONCURRENCY, len(docs))) + ) ) - failures = [r for r in results if isinstance(r, BaseException)] - count = len(results) - len(failures) + count = succeeded if failures: logger.warning( - f"Failed to delete {len(failures)} of {len(results)} old states; " + f"Failed to delete {len(failures)} of {len(docs)} old states; " f"first error: {failures[0]!r}" ) diff --git a/tests/unit/test_firestore_state.py b/tests/unit/test_firestore_state.py index 57f9af763..fd15f9c64 100644 --- a/tests/unit/test_firestore_state.py +++ b/tests/unit/test_firestore_state.py @@ -781,11 +781,14 @@ async def test_cleanup_custom_days_parameter(self): @staticmethod def _tracking_docs(n: int) -> tuple[list, dict]: """Build n mock docs whose deletes record peak overlap.""" - stats = {"in_flight": 0, "peak": 0} + stats = {"in_flight": 0, "peak": 0, "peak_tasks": 0} async def _tracked() -> None: stats["in_flight"] += 1 stats["peak"] = max(stats["peak"], stats["in_flight"]) + # Task count reflects how many coroutines were *allocated*, which is + # a stricter bound than how many RPCs are in flight. + stats["peak_tasks"] = max(stats["peak_tasks"], len(asyncio.all_tasks())) # Yield so sibling deletes get a chance to start. Under the previous # sequential loop nothing else could be running here. await asyncio.sleep(0) @@ -828,6 +831,29 @@ async def test_cleanup_bounds_in_flight_deletes(self): f"CLEANUP_DELETE_CONCURRENCY bound of {limit}" ) + async def test_cleanup_bounds_allocated_delete_tasks_not_just_rpcs(self): + """A worker pool must not allocate one task per document. + + Gathering over every document would schedule len(docs) tasks up front, + so a large backlog would cost unbounded task/event-loop memory even + though only CLEANUP_DELETE_CONCURRENCY RPCs are in flight. The query + driving this has no limit, so that allocation must stay bounded too. + """ + svc, _, _, _, _, coll_ref, query = _make_service_with_db() + limit = _mod.CLEANUP_DELETE_CONCURRENCY + docs, stats = self._tracking_docs(limit * 3) + query.get = AsyncMock(return_value=docs) + + count = await svc.cleanup_old_states(days=7) + + assert count == limit * 3 + # Allow a small margin for the enclosing test task and gather bookkeeping. + assert stats["peak_tasks"] <= limit + 3, ( + f"cleanup allocated {stats['peak_tasks']} concurrent tasks for " + f"{limit * 3} documents; delete tasks are not bounded by the " + f"CLEANUP_DELETE_CONCURRENCY worker pool of {limit}" + ) + async def test_cleanup_failure_does_not_abandon_remaining_deletes(self): """One failing delete must not strand the rest, and must not be counted.""" svc, _, _, _, _, coll_ref, query = _make_service_with_db() From 7558f0341f8da99452c7ca969b67635358db4662 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:13:25 -0500 Subject: [PATCH 3/6] test: make cleanup failure-isolation test exercise worker continuation The failure-isolation test used 3 documents against a pool of CLEANUP_DELETE_CONCURRENCY=16, so min(16, 3) = 3 workers each handled exactly one document. The `while True` continuation path was never taken, which made the test vacuous: a worker that returned on its first exception instead of continuing to drain the shared iterator would still have satisfied every assertion. Narrow the pool to a single worker against a 5-document backlog so the worker whose delete raises must keep pulling. Verified non-vacuous: with `return` added to the worker's except branch the test fails with `assert 0 == (5 - 1)`. Also correct the CLEANUP_DELETE_CONCURRENCY comment, which still described a semaphore-gated fan-out rather than the fixed worker pool that replaced it. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../services/cloud/firestore_state.py | 8 +-- tests/unit/test_firestore_state.py | 49 ++++++++++++------- 2 files changed, 35 insertions(+), 22 deletions(-) diff --git a/src/youtube_extension/services/cloud/firestore_state.py b/src/youtube_extension/services/cloud/firestore_state.py index 73dd9ec3f..ef5db6a05 100644 --- a/src/youtube_extension/services/cloud/firestore_state.py +++ b/src/youtube_extension/services/cloud/firestore_state.py @@ -27,9 +27,11 @@ logger = logging.getLogger(__name__) -# Upper bound on concurrent delete RPCs issued by cleanup_old_states(). Cleanup -# can match an unbounded number of expired documents, so deletes are fanned out -# under a semaphore rather than dispatched all at once. +# Size of the worker pool that drains expired documents in cleanup_old_states(). +# Cleanup can match an unbounded number of documents, so deletes are pulled from +# a shared iterator by this many workers rather than dispatched all at once. +# Sizing the pool -- rather than gating a full fan-out -- bounds the in-flight +# delete RPCs and the number of allocated task objects by the same constant. CLEANUP_DELETE_CONCURRENCY = 16 diff --git a/tests/unit/test_firestore_state.py b/tests/unit/test_firestore_state.py index fd15f9c64..cbc2544d3 100644 --- a/tests/unit/test_firestore_state.py +++ b/tests/unit/test_firestore_state.py @@ -855,31 +855,42 @@ async def test_cleanup_bounds_allocated_delete_tasks_not_just_rpcs(self): ) async def test_cleanup_failure_does_not_abandon_remaining_deletes(self): - """One failing delete must not strand the rest, and must not be counted.""" + """One failing delete must not strand the rest, and must not be counted. + + The pool is deliberately narrowed to a single worker while the backlog is + larger, so the *same* worker whose delete raises has to keep pulling from + the shared iterator. With a pool wider than the backlog every worker + handles exactly one document and the continuation path is never taken -- + the assertions below would then hold even for a worker that returned on + its first exception. + """ svc, _, _, _, _, coll_ref, query = _make_service_with_db() - ok_before = MagicMock() - ok_before.reference = MagicMock() - ok_before.reference.delete = AsyncMock() - - failing = MagicMock() - failing.reference = MagicMock() - failing.reference.delete = AsyncMock(side_effect=RuntimeError("firestore boom")) + pool_size = 1 + doc_count = 5 - ok_after = MagicMock() - ok_after.reference = MagicMock() - ok_after.reference.delete = AsyncMock() + docs = [] + for i in range(doc_count): + doc = MagicMock(name=f"doc{i}") + doc.reference = MagicMock() + # Fail on the very first document the lone worker touches. + doc.reference.delete = AsyncMock( + side_effect=RuntimeError("firestore boom") if i == 0 else None + ) + docs.append(doc) - query.get = AsyncMock(return_value=[ok_before, failing, ok_after]) + query.get = AsyncMock(return_value=docs) - count = await svc.cleanup_old_states(days=7) + with patch.object(_mod, "CLEANUP_DELETE_CONCURRENCY", pool_size): + count = await svc.cleanup_old_states(days=7) - # Only the two successful deletes are reported. - assert count == 2 - # The delete queued after the failure still ran; the sequential loop - # would have propagated the error and skipped it entirely. - ok_before.reference.delete.assert_awaited_once() - ok_after.reference.delete.assert_awaited_once() + # Only the successful deletes are reported. + assert count == doc_count - 1 + # Every document was attempted exactly once. The four after the failure + # were reachable only because the worker continued draining the queue; + # the original sequential loop propagated the error and skipped them. + for doc in docs: + doc.reference.delete.assert_awaited_once() # =========================================================================== From 1777291740fdf001acc20d9b4c1beb41d1622681 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:22:14 -0500 Subject: [PATCH 4/6] fix: bound Firestore cleanup controls with deployment configuration --- .../services/cloud/firestore_state.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/src/youtube_extension/services/cloud/firestore_state.py b/src/youtube_extension/services/cloud/firestore_state.py index ef5db6a05..723ea8f4c 100644 --- a/src/youtube_extension/services/cloud/firestore_state.py +++ b/src/youtube_extension/services/cloud/firestore_state.py @@ -30,9 +30,14 @@ # Size of the worker pool that drains expired documents in cleanup_old_states(). # Cleanup can match an unbounded number of documents, so deletes are pulled from # a shared iterator by this many workers rather than dispatched all at once. -# Sizing the pool -- rather than gating a full fan-out -- bounds the in-flight -# delete RPCs and the number of allocated task objects by the same constant. -CLEANUP_DELETE_CONCURRENCY = 16 +# Both controls are configurable so operators can tune cleanup independently of +# an application deployment; invalid numeric values fail fast during startup. +CLEANUP_DELETE_CONCURRENCY = max( + 1, int(os.getenv("CLEANUP_DELETE_CONCURRENCY", "16")) +) +CLEANUP_DELETE_TIMEOUT_SECONDS = max( + 0.001, float(os.getenv("CLEANUP_DELETE_TIMEOUT_SECONDS", "30")) +) @dataclass @@ -352,7 +357,7 @@ async def _delete_worker() -> None: except StopIteration: return try: - await doc.reference.delete() + await doc.reference.delete(timeout=CLEANUP_DELETE_TIMEOUT_SECONDS) succeeded += 1 except Exception as exc: # noqa: BLE001 - tallied and logged below failures.append(exc) From 57fc58d137ecb718980ff62b2de59e7aa4edc65b Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:22:15 -0500 Subject: [PATCH 5/6] test: pin Firestore cleanup delete timeout behavior --- tests/unit/test_firestore_state.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_firestore_state.py b/tests/unit/test_firestore_state.py index cbc2544d3..2a7b6fa60 100644 --- a/tests/unit/test_firestore_state.py +++ b/tests/unit/test_firestore_state.py @@ -760,6 +760,19 @@ async def test_cleanup_deletes_docs_and_returns_count(self): doc1.reference.delete.assert_awaited_once() doc2.reference.delete.assert_awaited_once() + async def test_cleanup_passes_configured_delete_timeout(self): + svc, _, _, _, _, coll_ref, query = _make_service_with_db() + doc = MagicMock() + doc.reference = MagicMock() + doc.reference.delete = AsyncMock() + query.get = AsyncMock(return_value=[doc]) + + with patch.object(_mod, "CLEANUP_DELETE_TIMEOUT_SECONDS", 12.5): + count = await svc.cleanup_old_states(days=7) + + assert count == 1 + doc.reference.delete.assert_awaited_once_with(timeout=12.5) + async def test_cleanup_queries_with_where_clause(self): svc, _, _, _, _, coll_ref, query = _make_service_with_db() query.get = AsyncMock(return_value=[]) @@ -783,7 +796,8 @@ def _tracking_docs(n: int) -> tuple[list, dict]: """Build n mock docs whose deletes record peak overlap.""" stats = {"in_flight": 0, "peak": 0, "peak_tasks": 0} - async def _tracked() -> None: + async def _tracked(*, timeout: float) -> None: + assert timeout > 0 stats["in_flight"] += 1 stats["peak"] = max(stats["peak"], stats["in_flight"]) # Task count reflects how many coroutines were *allocated*, which is From f6c7910e4232403a121e6fd7bd5378b13640eca3 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sat, 1 Aug 2026 16:26:57 -0500 Subject: [PATCH 6/6] fix: reject non-finite cleanup timeout overrides instead of clamping CLEANUP_DELETE_TIMEOUT_SECONDS was parsed as `max(0.001, float(os.getenv(...)))`. `float()` accepts `inf`, `-inf` and `nan`, and `max(0.001, inf)` returns `inf` unchanged, so setting CLEANUP_DELETE_TIMEOUT_SECONDS=inf silently removed the per-delete deadline (or would be rejected downstream by gRPC timeout validation). `nan` was equally unsafe: it compares false against every bound, so it was silently swallowed by max() rather than reported. Replace the clamping with two explicit parsers: - `_positive_int_env` - integer >= 1 - `_positive_finite_float_env` - finite float > 0 (math.isfinite) Out-of-range values now raise at import rather than being silently coerced, so an operator typo surfaces at startup instead of quietly changing cleanup behaviour. Blank values (a common artifact of a compose/Helm template rendering an empty string) fall back to the default instead of raising. Adds TestCleanupConfigEnvParsing covering inf/Infinity/-inf/nan/0/ negative/malformed/blank/unset/valid for both parsers, plus an invariant check on the module defaults. Non-vacuity proven by mutation: restoring the `max(0.001, value)` clamping inside the helper fails all 7 non-finite/non-positive cases with "DID NOT RAISE ValueError". Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../services/cloud/firestore_state.py | 53 ++++++++++++--- tests/unit/test_firestore_state.py | 65 +++++++++++++++++++ 2 files changed, 110 insertions(+), 8 deletions(-) diff --git a/src/youtube_extension/services/cloud/firestore_state.py b/src/youtube_extension/services/cloud/firestore_state.py index 723ea8f4c..c16074772 100644 --- a/src/youtube_extension/services/cloud/firestore_state.py +++ b/src/youtube_extension/services/cloud/firestore_state.py @@ -9,6 +9,7 @@ import asyncio import logging +import math import os from dataclasses import asdict, dataclass from datetime import datetime, timezone @@ -27,16 +28,52 @@ logger = logging.getLogger(__name__) -# Size of the worker pool that drains expired documents in cleanup_old_states(). +def _positive_int_env(name: str, default: int) -> int: + """Read a positive integer override, failing fast on invalid configuration. + + An unset or blank variable falls back to ``default`` (blank is common when a + compose/Helm template renders an empty value). Anything else must parse to an + integer >= 1; out-of-range values raise rather than being silently clamped, + so an operator typo surfaces at startup instead of changing behaviour quietly. + """ + raw = os.getenv(name) + if raw is None or not raw.strip(): + return default + value = int(raw.strip()) + if value < 1: + raise ValueError(f"{name} must be >= 1, got {raw!r}") + return value + + +def _positive_finite_float_env(name: str, default: float) -> float: + """Read a positive, finite float override, failing fast on invalid configuration. + + ``float()`` happily accepts ``inf``/``-inf``/``nan``. An infinite timeout would + silently remove the per-delete deadline (or be rejected downstream by gRPC + timeout validation), and ``nan`` compares false against every bound, so + non-finite values are rejected outright rather than clamped into range. + """ + raw = os.getenv(name) + if raw is None or not raw.strip(): + return default + value = float(raw.strip()) + if not math.isfinite(value) or value <= 0: + raise ValueError( + f"{name} must be a positive, finite number of seconds, got {raw!r}" + ) + return value + + +# Worker-pool size and per-delete deadline used by cleanup_old_states(). # Cleanup can match an unbounded number of documents, so deletes are pulled from # a shared iterator by this many workers rather than dispatched all at once. -# Both controls are configurable so operators can tune cleanup independently of -# an application deployment; invalid numeric values fail fast during startup. -CLEANUP_DELETE_CONCURRENCY = max( - 1, int(os.getenv("CLEANUP_DELETE_CONCURRENCY", "16")) -) -CLEANUP_DELETE_TIMEOUT_SECONDS = max( - 0.001, float(os.getenv("CLEANUP_DELETE_TIMEOUT_SECONDS", "30")) +# Sizing the pool -- rather than gating a full fan-out -- bounds the in-flight +# delete RPCs and the number of allocated task objects by the same constant. +# Both controls are overridable so operators can tune cleanup independently of an +# application deployment; invalid values fail fast during import. +CLEANUP_DELETE_CONCURRENCY = _positive_int_env("CLEANUP_DELETE_CONCURRENCY", 16) +CLEANUP_DELETE_TIMEOUT_SECONDS = _positive_finite_float_env( + "CLEANUP_DELETE_TIMEOUT_SECONDS", 30.0 ) diff --git a/tests/unit/test_firestore_state.py b/tests/unit/test_firestore_state.py index 2a7b6fa60..440783cef 100644 --- a/tests/unit/test_firestore_state.py +++ b/tests/unit/test_firestore_state.py @@ -16,6 +16,8 @@ from __future__ import annotations import asyncio +import math +import os import sys from datetime import datetime, timedelta, timezone from pathlib import Path @@ -728,6 +730,69 @@ async def test_list_states_applies_order_and_limit(self): query.limit.assert_called_once_with(50) +# =========================================================================== +# Module-level cleanup configuration parsing +# =========================================================================== + + +class TestCleanupConfigEnvParsing: + """Tests for the env-override parsers backing the cleanup constants. + + ``float()`` accepts ``inf``/``-inf``/``nan``, so an unvalidated timeout + override could silently remove the per-delete deadline. These tests pin that + non-finite and non-positive values are rejected rather than clamped. + """ + + @pytest.mark.parametrize("raw", ["inf", "Infinity", "-inf", "nan", "0", "-1", "0.0"]) + def test_float_env_rejects_non_finite_and_non_positive(self, raw): + with patch.dict(os.environ, {"CLEANUP_TEST_TIMEOUT": raw}, clear=False): + with pytest.raises(ValueError, match="positive, finite"): + _mod._positive_finite_float_env("CLEANUP_TEST_TIMEOUT", 30.0) + + def test_float_env_rejects_malformed_value(self): + with patch.dict(os.environ, {"CLEANUP_TEST_TIMEOUT": "abc"}, clear=False): + with pytest.raises(ValueError): + _mod._positive_finite_float_env("CLEANUP_TEST_TIMEOUT", 30.0) + + @pytest.mark.parametrize("raw", ["", " "]) + def test_float_env_blank_falls_back_to_default(self, raw): + with patch.dict(os.environ, {"CLEANUP_TEST_TIMEOUT": raw}, clear=False): + assert _mod._positive_finite_float_env("CLEANUP_TEST_TIMEOUT", 30.0) == 30.0 + + def test_float_env_unset_falls_back_to_default(self): + os.environ.pop("CLEANUP_TEST_TIMEOUT", None) + assert _mod._positive_finite_float_env("CLEANUP_TEST_TIMEOUT", 30.0) == 30.0 + + def test_float_env_parses_valid_override(self): + with patch.dict(os.environ, {"CLEANUP_TEST_TIMEOUT": " 12.5 "}, clear=False): + assert _mod._positive_finite_float_env("CLEANUP_TEST_TIMEOUT", 30.0) == 12.5 + + @pytest.mark.parametrize("raw", ["0", "-4"]) + def test_int_env_rejects_non_positive(self, raw): + with patch.dict(os.environ, {"CLEANUP_TEST_POOL": raw}, clear=False): + with pytest.raises(ValueError, match=">= 1"): + _mod._positive_int_env("CLEANUP_TEST_POOL", 16) + + @pytest.mark.parametrize("raw", ["abc", "1.5", "inf"]) + def test_int_env_rejects_malformed_value(self, raw): + with patch.dict(os.environ, {"CLEANUP_TEST_POOL": raw}, clear=False): + with pytest.raises(ValueError): + _mod._positive_int_env("CLEANUP_TEST_POOL", 16) + + def test_int_env_blank_falls_back_to_default(self): + with patch.dict(os.environ, {"CLEANUP_TEST_POOL": ""}, clear=False): + assert _mod._positive_int_env("CLEANUP_TEST_POOL", 16) == 16 + + def test_int_env_parses_valid_override(self): + with patch.dict(os.environ, {"CLEANUP_TEST_POOL": " 4 "}, clear=False): + assert _mod._positive_int_env("CLEANUP_TEST_POOL", 16) == 4 + + def test_module_defaults_are_positive_and_finite(self): + assert _mod.CLEANUP_DELETE_CONCURRENCY >= 1 + assert math.isfinite(_mod.CLEANUP_DELETE_TIMEOUT_SECONDS) + assert _mod.CLEANUP_DELETE_TIMEOUT_SECONDS > 0 + + # =========================================================================== # FirestoreStateService — cleanup_old_states # ===========================================================================