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)
 

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}"
+    )