diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 0013a57..744c532 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -211,11 +211,10 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: state.reserved += len(reserved.get(hostname, [])) state.consuming.update(q["name"] for q in queues) - # `pool.max-concurrency` is the number pool_grow/pool_shrink move and - # the number the UI shows. Absent on a worker whose stats did not - # answer, which leaves pool=None — unknown, not zero. - pool = (stats.get(hostname) or {}).get("pool", {}).get("max-concurrency") - if isinstance(pool, int): + # Absent on a worker whose stats did not answer, which leaves + # pool=None — unknown, not zero. + pool = pool_size((stats.get(hostname) or {}).get("pool") or {}) + if pool is not None: state.pools[hostname] = pool for state in out.values(): @@ -223,6 +222,27 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: return out +def pool_size(pool_stats: dict) -> int | None: + """How many processes a prefork pool is running NOW, from `inspect stats`. + + The length of `processes`, not `max-concurrency`. Celery reports + `max-concurrency` as the pool's `limit`, set once at boot; `pool_grow` and + `pool_shrink` hand straight to billiard and never touch it (celery 5.6 + `concurrency/prefork.py`: `self.grow = P.grow`). So it read 1 forever on a + lane booted at 1, and the sizing sweep, computing `target - 1` on every + tick, grew a scheduler capped at 2 to six processes while the System tab + showed "6 / 1" (2026-09-24) — and could never shrink one, since 1 - 1 is 0. + + `max-concurrency` is the fallback only for a pool that lists no processes + (a non-prefork pool), where it is the best number there is. + """ + procs = pool_stats.get("processes") + if isinstance(procs, list): + return len(procs) + limit = pool_stats.get("max-concurrency") + return limit if isinstance(limit, int) else None + + def effective_slots(target: int) -> int: """What a pool can actually be set to. Never below one process. diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py index 8febdb9..b2f005f 100644 --- a/tests/test_worker_control.py +++ b/tests/test_worker_control.py @@ -671,3 +671,36 @@ def test_an_unknown_worker_is_still_ignored_rather_than_guessed_at(monkeypatch): live = wc.inspect_lanes_sync() assert all(not s.present for s in live.values()) + + +# --- the pool size is the processes running, not the boot limit --------------- +# +# The operator's System tab, 2026-09-24: Scheduler "6 / 1" at a cap of 2. Celery +# reports `max-concurrency` as the limit the pool BOOTED with; pool_grow and +# pool_shrink never update it. The sweep read 1 forever, sent `target - 1` on +# every tick, and grew the lane past its cap with no way to shrink it back. + + +def test_pool_size_counts_the_live_processes_not_the_boot_limit(): + """After two pool_grow calls on a pool booted at 1.""" + assert wc.pool_size({"max-concurrency": 1, "processes": [11, 12, 13]}) == 3 + + +def test_pool_size_falls_back_to_the_limit_without_a_process_list(): + assert wc.pool_size({"max-concurrency": 4}) == 4 + assert wc.pool_size({}) is None + + +def test_a_grown_pool_reads_at_its_real_size(monkeypatch): + """The read the sweep sizes from: a pool at its target must read AT its + target, or `target - current` never reaches zero and every tick grows it.""" + _stub_active_queues(monkeypatch, {"scheduler@h": []}) + import sys + insp = sys.modules["backend.app.celery_app"].celery.control.inspect + insp.stats = lambda self: { + "scheduler@h": {"pool": {"max-concurrency": 1, "processes": [1, 2]}}, + } + + live = wc.inspect_lanes_sync() + + assert live["scheduler"].pool == 2