diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 157b01b..d9c6cde 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -66,6 +66,23 @@ log = logging.getLogger(__name__) # report "not present", which is true, rather than hang the page. CONTROL_TIMEOUT_SECONDS = 2.0 +# The WORST case of `inspect_lanes_sync`, for callers that need a deadline. +# +# One broadcast plus three targeted reads. The targeted three normally return +# as soon as the named nodes answer; each can still cost a full timeout if a +# node disappears mid-read, so the bound stays four. +CONTROL_ROUND_TRIPS = 4 + +# Slack for the `asyncio.to_thread` handoff. A budget equal to the work is a +# budget that fails under load — the roster carried exactly that bug into the +# operator's first consolidated deploy and logged a TimeoutError per refresh +# while the inspect calls underneath were working fine. +CONTROL_SLACK_SECONDS = 3.0 + +INSPECT_BUDGET_SECONDS = ( + CONTROL_TIMEOUT_SECONDS * CONTROL_ROUND_TRIPS + CONTROL_SLACK_SECONDS +) + @dataclass class LaneLiveState: @@ -130,11 +147,36 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: try: from ..celery_app import celery as celery_app + # ONE broadcast, then three TARGETED reads. + # + # A broadcast with no `destination` cannot know how many replies to + # expect, so it waits out its whole timeout rather than returning on + # the last one. Four of those is four full timeouts — about eight + # seconds — and `lane_view` sits on the Settings card, so that was the + # load time of the Worker lanes page every time it was opened. + # + # Naming the destinations lets celery stop as soon as those nodes have + # answered, which for workers in this same container is milliseconds. + # The worst case is unchanged: a node that vanishes between the + # broadcast and the targeted reads costs a full timeout waiting for a + # reply that is not coming. insp = celery_app.control.inspect(timeout=CONTROL_TIMEOUT_SECONDS) active_queues = insp.active_queues() or {} - stats = insp.stats() or {} - active = insp.active() or {} - reserved = insp.reserved() or {} + + # Nothing answered — and the three reads below exist only to describe + # what did. Returning here also makes the broker-down case FAST + # (one timeout, not four), which is exactly when the healthcheck and + # the card need an answer rather than a long wait. + if not active_queues: + return out + + targeted = celery_app.control.inspect( + destination=sorted(active_queues), + timeout=CONTROL_TIMEOUT_SECONDS, + ) + stats = targeted.stats() or {} + active = targeted.active() or {} + reserved = targeted.reserved() or {} except Exception: log.warning("worker_control: celery inspect failed", exc_info=True) return out @@ -288,7 +330,22 @@ async def lane_view(session: AsyncSession) -> list[dict]: LLEN alone reads 0 while a worker holds tasks in memory. """ rows = await _rows_by_name(session) - live = await asyncio.to_thread(inspect_lanes_sync) + # A deadline, because this is a request path and `to_thread` on its own is + # an await with no bound (rule 156). `inspect_lanes_sync` never raises and + # every inner call has its own timeout, so the only way past the budget is + # the thread not being scheduled — and a page that renders "not answering" + # is a better answer than one that does not render. + try: + live = await asyncio.wait_for( + asyncio.to_thread(inspect_lanes_sync), + timeout=INSPECT_BUDGET_SECONDS, + ) + except TimeoutError: + log.warning( + "worker_control: inspect exceeded %ss; reporting every lane as " + "not answering", INSPECT_BUDGET_SECONDS, + ) + live = {lane.name: LaneLiveState() for lane in LANES} depths = await asyncio.to_thread(_queue_depths_sync) oldest = await _oldest_running_by_queue(session) diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py index d5d6d1b..8d27ff1 100644 --- a/tests/test_worker_control.py +++ b/tests/test_worker_control.py @@ -563,3 +563,106 @@ def test_the_autoscale_task_is_registered_and_scheduled(): name = "backend.app.tasks.maintenance.autoscale_worker_lanes" assert name in celery.tasks assert name in {e["task"] for e in celery.conf.beat_schedule.values()} + + +# --- the inspect round trips ------------------------------------------------- + + +class _Inspect: + """Records what was asked, and of whom.""" + + def __init__(self, calls, queues, destination=None): + self.calls = calls + self._queues = queues + self.destination = destination + + def active_queues(self): + self.calls.append(("active_queues", self.destination)) + return self._queues + + def stats(self): + self.calls.append(("stats", self.destination)) + return {} + + def active(self): + self.calls.append(("active", self.destination)) + return {} + + def reserved(self): + self.calls.append(("reserved", self.destination)) + return {} + + +def _stub_inspect(monkeypatch, queues): + calls = [] + + class _Control: + def inspect(self, timeout=None, destination=None): + return _Inspect(calls, queues, destination) + + class _Celery: + pass + + celery = _Celery() + celery.control = _Control() + import sys + import types + mod = types.ModuleType("backend.app.celery_app") + mod.celery = celery + monkeypatch.setitem(sys.modules, "backend.app.celery_app", mod) + return calls + + +def test_only_the_first_read_is_a_broadcast(monkeypatch): + """A broadcast cannot know how many replies to expect, so it waits out its + whole timeout. Four of those is ~8s, and lane_view sits on the Settings + card — that was the load time of the Worker lanes page. + + Naming the destinations lets celery stop as soon as those nodes answer. + """ + calls = _stub_inspect(monkeypatch, { + "worker@a": [{"name": q} for q in LANES_BY_NAME["worker"].queues], + }) + + wc.inspect_lanes_sync() + + assert calls[0] == ("active_queues", None), "the first read discovers nodes" + for name, destination in calls[1:]: + assert destination == ["worker@a"], ( + f"{name} broadcast instead of addressing the node that answered" + ) + + +def test_nothing_answering_costs_one_round_trip_not_four(monkeypatch): + """The three later reads exist only to describe what answered, so with an + empty roster they describe nothing — at three full timeouts. + + This is the broker-down case, which is exactly when the healthcheck and + the card need an answer rather than a long wait. + """ + calls = _stub_inspect(monkeypatch, {}) + + out = wc.inspect_lanes_sync() + + assert calls == [("active_queues", None)] + assert all(not state.present for state in out.values()) + + +def test_the_deadline_exceeds_the_worst_case_it_waits_for(monkeypatch): + """Same relation the roster needed after it shipped a budget equal to its + own worst case. Asserted as a relation, not a number: a fifth inspect call + is how this comes back.""" + work = wc.CONTROL_TIMEOUT_SECONDS * wc.CONTROL_ROUND_TRIPS + assert wc.INSPECT_BUDGET_SECONDS > work + assert wc.CONTROL_SLACK_SECONDS > 0 + + +def test_the_round_trip_bound_matches_the_reads_actually_made(): + import inspect as _inspect + + src = _inspect.getsource(wc.inspect_lanes_sync) + reads = src.count("insp.") + src.count("targeted.") + assert reads == wc.CONTROL_ROUND_TRIPS, ( + f"inspect_lanes_sync makes {reads} reads but CONTROL_ROUND_TRIPS " + f"says {wc.CONTROL_ROUND_TRIPS}" + )