"""Changing a lane's slots on a running system (milestone 422 step 2). The celery control calls are stubbed: what is being tested is the DELTA ARITHMETIC and the persist/push split, not that celery can resize its own pool. A test that asserted celery's behaviour would be testing celery. """ from __future__ import annotations import pytest from backend.app.services import worker_control as wc from backend.app.services.worker_lanes import LANES_BY_NAME # --- what inspect reports ---------------------------------------------------- class _Control: """Records the control messages that were sent.""" def __init__(self): self.grew: list[tuple[int, list[str]]] = [] self.shrank: list[tuple[int, list[str]]] = [] self.added: list[tuple[str, list[str]]] = [] self.cancelled: list[tuple[str, list[str]]] = [] def pool_grow(self, n, destination=None): self.grew.append((n, destination)) def pool_shrink(self, n, destination=None): self.shrank.append((n, destination)) def add_consumer(self, queue, destination=None): self.added.append((queue, destination)) def cancel_consumer(self, queue, destination=None): self.cancelled.append((queue, destination)) def _stub_live( monkeypatch, lane_name, *, pools, present=True, reserved=0, consuming=None, ): # `consuming` defaults to the lane's full queue set — i.e. an ENABLED # lane. Tests for the disabled case pass an empty set explicitly. if consuming is None: consuming = set(LANES_BY_NAME[lane_name].queues) state = wc.LaneLiveState( present=present, replicas=len(pools), hostnames=sorted(pools), pools=dict(pools), reserved=reserved, consuming=set(consuming), ) monkeypatch.setattr( wc, "inspect_lanes_sync", lambda: {name: (state if name == lane_name else wc.LaneLiveState()) for name in LANES_BY_NAME}, ) return state def _stub_control(monkeypatch): control = _Control() 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 control def test_pool_property_is_max_not_sum(): """`slots` means the pool size of ONE process, so the aggregate shown to the operator is the largest replica — not the total. A sum would report 8 for two replicas of 4 and invite them to 'reduce it to 4', which would halve the lane.""" state = wc.LaneLiveState(pools={"a": 4, "b": 4}) assert state.pool == 4 def test_pool_is_none_when_nothing_reported(): """Unknown, never zero — the distinction step 3's reconcile depends on.""" assert wc.LaneLiveState(present=True).pool is None # --- the delta arithmetic ---------------------------------------------------- def test_each_replica_gets_its_own_delta(monkeypatch): """The bug this exists to prevent: one delta computed from an aggregate and applied to every replica. With replicas at 2 and 6 and a target of 4, a shared delta moves both the same way and leaves them at 4 and 8 — or 0 and 4 — depending on which aggregate was used. Per-replica deltas converge both on 4. """ control = _stub_control(monkeypatch) _stub_live(monkeypatch, "worker", pools={"host-a": 2, "host-b": 6}) applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) assert (applied, err) == (True, None) assert control.grew == [(2, ["host-a"])] assert control.shrank == [(2, ["host-b"])] def test_a_replica_already_at_the_target_is_sent_nothing(monkeypatch): """The fixed point step 3's reconcile needs. An enforcer that re-issues a grow of zero every tick never converges and re-does its own work forever (lesson #4183).""" control = _stub_control(monkeypatch) _stub_live(monkeypatch, "worker", pools={"host-a": 4}) applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) assert (applied, err) == (True, None) assert control.grew == [] assert control.shrank == [] def test_resizing_an_absent_lane_reports_rather_than_raises(monkeypatch): _stub_control(monkeypatch) _stub_live(monkeypatch, "worker", pools={}, present=False) applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) assert applied is False assert "not running" in err def test_a_replica_with_no_reported_pool_is_named_not_skipped_silently( monkeypatch, ): """Resize what can be resized, then say which could not. Silence would leave a replica running at a size the UI claims it is not.""" control = _stub_control(monkeypatch) state = _stub_live(monkeypatch, "worker", pools={"host-a": 2}) state.hostnames = ["host-a", "host-b"] # b answered inspect, not stats state.replicas = 2 applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) assert control.grew == [(2, ["host-a"])] assert applied is False assert "host-b" in err # --- enabling and disabling -------------------------------------------------- def test_disabling_cancels_consumers_rather_than_killing_the_worker(monkeypatch): """A cancelled consumer keeps the process alive and answering inspect, so a disabled lane stays visible. A killed worker reads as ABSENT, which is the same signal as a crash — and milestone 365 exists precisely so those two do not look alike.""" control = _stub_control(monkeypatch) _stub_live(monkeypatch, "ml", pools={"host-a": 1}) applied, err = wc.set_lane_enabled_sync(LANES_BY_NAME["ml"], False) assert (applied, err) == (True, None) assert control.cancelled == [("ml", ["host-a"])] assert control.added == [] def test_enabling_adds_a_consumer_for_every_queue_in_the_lane(monkeypatch): control = _stub_control(monkeypatch) _stub_live(monkeypatch, "worker", pools={"host-a": 1}) wc.set_lane_enabled_sync(LANES_BY_NAME["worker"], True) assert [q for q, _ in control.added] == list(LANES_BY_NAME["worker"].queues) # --- lane identity ----------------------------------------------------------- def test_queue_sets_map_back_to_their_lane_in_any_order(): """celery does not guarantee the order it lists a worker's queues in, so the lookup sorts. Unsorted, a lane would intermittently fail to match and read as absent.""" lane = LANES_BY_NAME["worker"] assert wc._lane_for_queues(tuple(reversed(lane.queues))) is lane def test_an_unknown_queue_set_maps_to_no_lane(): """A deployment slicing CELERY_QUEUES differently has no lane row to control. Honest rather than an error — the roster still reports it.""" assert wc._lane_for_queues(("something", "else")) is None # --- 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}" ) # --- the sizing pass --------------------------------------------------------- # # One sweep replaced two on 2026-09-23. The reconcile drove the pool to a # stored `slots`; the autoscaler moved it away from that same number; and most # of the autoscaler's design existed to stop the reconcile undoing its work. # The stored number is gone, so there is nothing left to disagree about — and # these tests no longer have to assert that two sweeps get along. # # The fixtures deliberately set the live pool and the cap to DIFFERENT numbers # wherever the distinction matters. Lesson #4318: equal fixtures cannot tell # "reads live" from "reads stored", and that is exactly how the old autoscaler # shipped unable to do its job with every test green. def _sizing_live( monkeypatch, *, pools, active, reserved, depth, consuming=None, present=True, ): state = wc.LaneLiveState( present=present, replicas=len(pools), hostnames=sorted(pools), pools=dict(pools), active=active, reserved=reserved, consuming=set(LANES_BY_NAME["worker"].queues if consuming is None else consuming), ) monkeypatch.setattr( wc, "inspect_lanes_sync", lambda: {name: (state if name == "worker" else wc.LaneLiveState()) for name in LANES_BY_NAME}, ) monkeypatch.setattr( wc, "_queue_depths_sync", lambda: {q: depth for lane in LANES_BY_NAME.values() for q in lane.queues}, ) return state def _worker(sized): return next(d for d in sized if d.lane == "worker") # --- what a lane has work for ------------------------------------------------ def test_work_in_flight_and_waiting_each_justify_a_worker(): # Two running plus six queued is eight tasks, so eight workers — if the # cap allowed it. assert wc.wanted_slots(cap=10, active=2, pending=6) == 8 def test_it_never_exceeds_the_cap(): assert wc.wanted_slots(cap=2, active=2, pending=4000) == 2 def test_an_idle_lane_falls_to_one_and_not_to_zero(): """billiard will not run an empty pool, and the parked process is what `add_consumer` lands on when the cap goes back up.""" assert wc.wanted_slots(cap=8, active=0, pending=0) == 1 def test_a_lane_capped_at_zero_still_keeps_its_process(): """`cap 0` is expressed by cancelling CONSUMERS, not by emptying the pool. A lane with no process reads as absent, which is the same signal as a crash — and the roster exists to keep those two apart.""" assert wc.wanted_slots(cap=0, active=0, pending=0) == wc.MIN_POOL_SLOTS def test_an_unknown_backlog_contributes_nothing_rather_than_zero(): """The broker did not answer for these queues. Reading that as "empty" would shrink a lane on the strength of a failed read (snippet #3969) — and reading it as "huge" would grow one. It contributes nothing, and the work actually in flight still counts.""" assert wc.wanted_slots(cap=8, active=3, pending=None) == 3 # --- growing ----------------------------------------------------------------- def test_a_backlog_is_met_in_one_tick_not_one_slot_per_minute(monkeypatch): """The operator's condition for always-on autoscaling: *"always on"* is only pleasant if the ramp keeps up. Growing +1 per minute would take four minutes to answer a burst, and the old autoscaler did exactly that.""" control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=0, depth=4000) d = _worker(wc.size_lanes_sync({"worker": 4})) assert (d.action, d.slots) == ("grew", 4) assert control.grew == [(3, ["host-a"])] def test_a_prefetched_backlog_counts(monkeypatch): """THE case an LLEN-only implementation misses. Celery prefetches, so a lane can read queue depth 0 while holding thirty tasks in worker memory — a sizing pass watching LLEN alone sees an idle system and shrinks.""" control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=30, depth=0) assert _worker(wc.size_lanes_sync({"worker": 4})).action == "grew" assert control.grew == [(3, ["host-a"])] def test_a_restarted_worker_is_pulled_back_up(monkeypatch): """What the five-minute reconcile existed for, now done in one minute by the pass that was already running. `pool_grow` is not durable: a worker restarted by its supervisor comes back at its ENV concurrency, silently below what the lane should be running.""" control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=9, depth=0) assert _worker(wc.size_lanes_sync({"worker": 4})).slots == 4 assert control.grew == [(3, ["host-a"])] # --- shrinking --------------------------------------------------------------- def test_an_idle_lane_gives_a_worker_back_one_at_a_time(monkeypatch): """Operator: *"so that idle instances quiet down when not running."* One per tick on the way down, against immediate growth. Being one worker too large for a minute costs a sleeping process; being too small costs work not happening. For ML it matters most — every new slot reloads a multi-GB model, so a slow shrink is what stops a quiet patch from paying that cost again a minute later.""" control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 4}, active=0, reserved=0, depth=0) d = _worker(wc.size_lanes_sync({"worker": 4})) assert (d.action, d.slots) == ("shrank", 3) assert control.shrank == [(1, ["host-a"])] def test_it_stops_shrinking_at_one(monkeypatch): control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0) assert _worker(wc.size_lanes_sync({"worker": 4})).action == "held" assert control.shrank == [] def test_lowering_the_cap_pulls_a_lane_down(monkeypatch): """The cap is a ceiling on the live pool, not only on future growth.""" control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 4}, active=4, reserved=99, depth=0) assert _worker(wc.size_lanes_sync({"worker": 2})).slots == 3 assert control.shrank == [(1, ["host-a"])] # --- the fixed point --------------------------------------------------------- def test_a_settled_lane_sends_no_control_messages(monkeypatch): """THE property this pass lives or dies on. It runs every minute forever, so a converged tick must be silent — one inspect, one LLEN sweep, nothing else. An enforcer that re-issues a grow of zero churns forever and buries a real correction in its own noise (lesson #4183).""" control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 2}, active=2, reserved=0, depth=0) d = _worker(wc.size_lanes_sync({"worker": 4})) assert d.action == "held" # EVERY control family, not just the pool ones: an unconditional # add_consumer for four queues per tick is harmless per call and unbounded # in aggregate, and invisible. assert control.grew == [] assert control.shrank == [] assert control.added == [] assert control.cancelled == [] def test_an_absent_lane_is_skipped_not_corrected(monkeypatch): """`present=False` means nothing answered — a worker restarting, or an unreachable broker. It is NOT zero workers, and there is nothing to send a message to.""" control = _stub_control(monkeypatch) _sizing_live( monkeypatch, pools={}, active=0, reserved=0, depth=0, present=False, ) d = _worker(wc.size_lanes_sync({"worker": 4})) assert d.action == "skipped" assert control.grew == [] and control.shrank == [] def test_a_lane_with_no_row_is_left_alone(monkeypatch): control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=99) assert wc.size_lanes_sync({}) == [] assert control.grew == [] # --- the cap is also the switch ---------------------------------------------- def test_a_cap_of_zero_stops_the_lane_consuming(monkeypatch): control = _stub_control(monkeypatch) _sizing_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0) wc.size_lanes_sync({"worker": 0}) assert sorted(q for q, _ in control.cancelled) == sorted( LANES_BY_NAME["worker"].queues ) def test_a_cap_above_zero_starts_it_consuming(monkeypatch): control = _stub_control(monkeypatch) _sizing_live( monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0, consuming=set(), ) wc.size_lanes_sync({"worker": 2}) assert sorted(q for q, _ in control.added) == sorted( LANES_BY_NAME["worker"].queues ) def test_an_already_stopped_lane_is_not_cancelled_again(monkeypatch): """The other half of the fixed point. `cancel_consumer` on a queue that is not being consumed succeeds and does nothing, so an unconditional pass would churn here forever with no symptom.""" control = _stub_control(monkeypatch) _sizing_live( monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0, consuming=set(), ) wc.size_lanes_sync({"worker": 0}) assert control.cancelled == [] assert control.added == [] def test_a_stopped_lane_still_keeps_exactly_one_process(monkeypatch): """The live bug of 2026-09-23, from the other direction: the pass must not try to empty a pool billiard will not empty, and must not report having done so every tick.""" control = _stub_control(monkeypatch) _sizing_live( monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0, consuming=set(), ) assert _worker(wc.size_lanes_sync({"worker": 0})).action == "held" assert control.shrank == [] def test_the_sizing_task_is_registered_and_scheduled(): """A pass nothing schedules is a pass that never runs — and since this one replaced two entries, a stale name in the beat schedule would leave the lanes unmanaged with nothing red anywhere.""" from backend.app.celery_app import celery name = "backend.app.tasks.maintenance.size_worker_lanes" assert name in celery.tasks entries = celery.conf.beat_schedule assert any(e["task"] == name for e in entries.values()) # The two it replaced are gone, not merely unreferenced. tasks = {e["task"] for e in entries.values()} assert "backend.app.tasks.maintenance.reconcile_worker_lanes" not in tasks assert "backend.app.tasks.maintenance.autoscale_worker_lanes" not in tasks