diff --git a/backend/app/services/service_roster.py b/backend/app/services/service_roster.py index ff831b2..83de7db 100644 --- a/backend/app/services/service_roster.py +++ b/backend/app/services/service_roster.py @@ -51,6 +51,29 @@ REFRESH_TTL_SECONDS = 20.0 # page that exists to explain it. INSPECT_TIMEOUT_SECONDS = 2.0 +# How many broadcast round trips `_inspect_celery_sync` makes. Named, because +# the wrapper's budget is derived from it and the two must not drift. +# +# `active_queues()` and `active()` are separate broadcasts, and a broadcast +# with no `destination` cannot know how many replies to expect — so each one +# waits out its full timeout rather than returning on the last reply. The sync +# call therefore costs ~2 x INSPECT_TIMEOUT_SECONDS in the ordinary case, not +# once. +INSPECT_ROUND_TRIPS = 2 + +# Slack for the thread handoff. `asyncio.to_thread` hands work to the default +# executor, and on a loaded web process — the operator's showcase page pulling +# ninety thumbnails a second — the thread may not even be scheduled inside the +# budget, let alone finish. +# +# This exists because the wrapper used to allow `INSPECT_TIMEOUT_SECONDS * 2`, +# which LOOKS like a safety factor and is exactly the worst case with nothing +# left over. Observed on the operator's first consolidated deploy, 2026-09-23: +# a TimeoutError traceback per refresh while the two inspect calls were +# working perfectly. A budget equal to the work is a budget that fails under +# any load at all. +INSPECT_SLACK_SECONDS = 3.0 + # Queue set -> the name an operator recognises. Sorted-tuple keys, because the # order celery reports them in is not guaranteed. # @@ -86,6 +109,10 @@ def _inspect_celery_sync() -> dict[tuple[str, ...], dict]: from ..celery_app import celery as celery_app insp = celery_app.control.inspect(timeout=INSPECT_TIMEOUT_SECONDS) + # TWO broadcasts, each waiting out its own timeout — see + # INSPECT_ROUND_TRIPS, which the caller's budget is derived from. Adding a + # third call here without updating that constant puts the wrapper back + # under the work it is waiting for. active_queues = insp.active_queues() or {} active_tasks = insp.active() or {} @@ -137,7 +164,10 @@ async def refresh_celery_roster(session: AsyncSession) -> None: try: grouped = await asyncio.wait_for( asyncio.to_thread(_inspect_celery_sync), - timeout=INSPECT_TIMEOUT_SECONDS * 2, + timeout=( + INSPECT_TIMEOUT_SECONDS * INSPECT_ROUND_TRIPS + + INSPECT_SLACK_SECONDS + ), ) except Exception: log.warning("service roster: celery inspect failed; roster not refreshed", exc_info=True) diff --git a/tests/test_service_roster.py b/tests/test_service_roster.py new file mode 100644 index 0000000..372bee4 --- /dev/null +++ b/tests/test_service_roster.py @@ -0,0 +1,54 @@ +"""The roster's inspect budget must exceed the inspect work. + +Both tests here assert a RELATION between constants rather than their values. +The failure they exist for is not a wrong number — it is the budget and the +work drifting apart, which is invisible in each one read on its own. +""" + +from __future__ import annotations + +from backend.app.services import service_roster as sr + + +def test_the_wrapper_budget_exceeds_the_work_it_waits_for(): + """The bug the operator's first consolidated deploy surfaced (2026-09-23). + + `_inspect_celery_sync` makes INSPECT_ROUND_TRIPS broadcasts, and a + broadcast with no `destination` cannot know how many replies to expect, so + each waits out its full timeout rather than returning on the last reply. + The sync call therefore costs round_trips x INSPECT_TIMEOUT_SECONDS. + + The wrapper allowed `INSPECT_TIMEOUT_SECONDS * 2`, which reads like a + safety factor and is exactly that worst case with nothing left over. Live + result: a TimeoutError traceback on every refresh while the two inspect + calls were working perfectly, and a roster that stopped advancing. + + Asserted as a RELATION, not as a number. A third inspect call added to the + sync function is the way this silently comes back, and the only thing that + keeps the two honest is deriving one from the other. + """ + budget = ( + sr.INSPECT_TIMEOUT_SECONDS * sr.INSPECT_ROUND_TRIPS + + sr.INSPECT_SLACK_SECONDS + ) + work = sr.INSPECT_TIMEOUT_SECONDS * sr.INSPECT_ROUND_TRIPS + assert budget > work, ( + f"the wrapper allows {budget}s for {work}s of broadcasts — a budget " + f"equal to the work fails under any load at all" + ) + assert sr.INSPECT_SLACK_SECONDS > 0 + + +def test_the_round_trip_count_matches_the_calls_actually_made(): + """INSPECT_ROUND_TRIPS is only true if someone keeps it true, so read the + source rather than trusting the constant: the budget above is derived from + it, and a call added without updating it puts the wrapper back under the + work.""" + import inspect as _inspect + + src = _inspect.getsource(sr._inspect_celery_sync) + calls = src.count("insp.") + assert calls == sr.INSPECT_ROUND_TRIPS, ( + f"_inspect_celery_sync makes {calls} inspect calls but " + f"INSPECT_ROUND_TRIPS says {sr.INSPECT_ROUND_TRIPS}" + )