diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index d49c8ef..894fcd1 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -111,6 +111,14 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.recover_interrupted_tasks", "schedule": 300.0, # every 5 minutes }, + "reconcile-worker-lanes": { + "task": "backend.app.tasks.maintenance.reconcile_worker_lanes", + "schedule": 300.0, # every 5 minutes — the window in which a + # restarted worker runs at its ENV concurrency rather than the + # slots the operator set (milestone 422 step 3). A no-op once + # every lane matches: one broker round trip, no control + # messages, nothing logged. + }, "cleanup-old-tasks": { "task": "backend.app.tasks.maintenance.cleanup_old_tasks", "schedule": 86400.0, # daily diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 7adac19..97e70f8 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -82,6 +82,13 @@ class LaneLiveState: active: int = 0 reserved: int = 0 hostnames: list[str] = field(default_factory=list) + # The queues this lane is actually consuming right now, across replicas. + # Distinct from the lane's CONFIGURED queues: `cancel_consumer` stops a + # worker consuming one without changing what it was started with, which + # is how `enabled=false` is implemented. The reconcile needs this to tell + # "already disabled" from "needs disabling" — without it, it would re-send + # add_consumer for every queue on every tick forever (lesson #4183). + consuming: set[str] = field(default_factory=set) # Pool size PER HOSTNAME, not aggregated. The resize below computes each # replica's own delta from its own current pool, so replicas that have # drifted apart converge instead of being moved in lockstep from a shared @@ -134,6 +141,7 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: state.hostnames.append(hostname) state.active += len(active.get(hostname, [])) 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 @@ -147,7 +155,9 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: return out -def set_lane_slots_sync(lane: Lane, target: int) -> tuple[bool, str | None]: +def set_lane_slots_sync( + lane: Lane, target: int, live: LaneLiveState | None = None, +) -> tuple[bool, str | None]: """Drive every replica of `lane` to `target` slots. Returns (applied, err). Per-replica deltas rather than one shared delta: see the module docstring. @@ -163,7 +173,8 @@ def set_lane_slots_sync(lane: Lane, target: int) -> tuple[bool, str | None]: try: from ..celery_app import celery as celery_app - live = inspect_lanes_sync()[lane.name] + if live is None: + live = inspect_lanes_sync()[lane.name] if not live.present: return False, "lane is not running" if not live.pools: @@ -188,7 +199,9 @@ def set_lane_slots_sync(lane: Lane, target: int) -> tuple[bool, str | None]: return False, str(exc) -def set_lane_enabled_sync(lane: Lane, enabled: bool) -> tuple[bool, str | None]: +def set_lane_enabled_sync( + lane: Lane, enabled: bool, live: LaneLiveState | None = None, +) -> tuple[bool, str | None]: """Start or stop `lane` consuming its queues, without killing the process. `cancel_consumer` rather than a shutdown: a stopped consumer keeps its @@ -200,7 +213,8 @@ def set_lane_enabled_sync(lane: Lane, enabled: bool) -> tuple[bool, str | None]: try: from ..celery_app import celery as celery_app - live = inspect_lanes_sync()[lane.name] + if live is None: + live = inspect_lanes_sync()[lane.name] if not live.present: return False, "lane is not running" control = celery_app.control @@ -389,3 +403,82 @@ async def set_lane( "applied": applied, "apply_error": error, } + + +def reconcile_lanes_sync(desired: dict[str, tuple[int, bool]]) -> dict: + """Drive every RUNNING lane to its stored slots and enabled flag. + + `desired` is lane name -> (slots, enabled), read from the database by the + caller. This function touches no database: the celery task that schedules + it owns the sync session, and keeping the DB out of here is what lets the + same code be called from anywhere that already knows the target. + + ## Why this exists at all + + `pool_grow` is not durable. A worker that dies and is restarted by its + supervisor comes back at its ENV concurrency — silently below whatever the + operator set — and nothing in step 2's path would ever notice. Storing the + value made it survivable; this is what makes it actually survive. + + ## It must converge and then go quiet + + One `inspect` for all lanes, and `set_lane_slots_sync` issues nothing at + all to a replica already at its target. So a settled system performs one + broker round trip per tick and sends no control messages — the reachable + fixed point lesson #4183 is about. An enforcer that re-sent a grow of zero + every tick would churn forever and bury a real correction in its own noise, + which is why `changed` below counts only lanes that actually moved. + + ## An absent lane is SKIPPED, not corrected + + `present=False` means nothing answered — a worker restarting, or a broker + that is unreachable. It does NOT mean zero slots. Correcting an absence + would be drawing a conclusion from an unswept read (snippet #3969), and + here it would be worse than useless: there is nothing to send the message + to. The lane is reported as skipped and picked up on a later tick. + """ + live = inspect_lanes_sync() + changed: list[str] = [] + skipped: list[str] = [] + failed: dict[str, str] = {} + + for lane in LANES: + target = desired.get(lane.name) + if target is None: + continue + slots, enabled = target + state = live[lane.name] + if not state.present: + skipped.append(lane.name) + continue + + # Enabled first: a lane being turned on should be consuming before + # its pool is sized, so the slots it gains have work to pick up. + # + # Only when it DISAGREES. Calling this unconditionally would send + # add_consumer for every queue on every tick of a settled system — + # the exact churn lesson #4183 describes, and invisible because + # add_consumer on a queue already consumed is harmless. + consuming_all = state.consuming.issuperset(lane.queues) + if enabled != consuming_all: + ok, err = set_lane_enabled_sync(lane, enabled, live=state) + if not ok: + failed[lane.name] = err or "could not set consumers" + continue + changed.append(lane.name) + + current = state.pool + if current is not None and current == slots: + continue + ok, err = set_lane_slots_sync(lane, slots, live=state) + if ok: + if lane.name not in changed: + changed.append(lane.name) + log.info( + "worker_control: %s reconciled %s -> %s slots", + lane.name, current, slots, + ) + else: + failed[lane.name] = err or "could not resize" + + return {"changed": changed, "skipped": skipped, "failed": failed} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 3a87924..c7c8fe7 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1348,3 +1348,53 @@ def sync_memberships() -> str: if res.get("suggested") is not None: parts.append(f"suggested={res['suggested']}") return " ".join(parts) or "no platforms" + + +@celery.task(name="backend.app.tasks.maintenance.reconcile_worker_lanes") +def reconcile_worker_lanes() -> dict: + """Drive every running lane back to the slots the operator set. + + Milestone 422 step 3. `pool_grow` is not durable: a worker restarted by + its supervisor comes back at its ENV concurrency, silently below whatever + was configured, and nothing on step 2's write path would ever notice. + + ## Why a beat task and not a hook in web + + Step 3 was written as "web applies the stored values after it starts". It + cannot, and `services/service_roster.py` already records why: hypercorn + runs `--workers 4`, so anything in `before_serving` becomes FOUR + concurrent loops per container, all hammering the broker forever. + + The other option was service_roster's own answer — refresh on demand from + whichever request happens to arrive. Rejected here because the two are + solving different problems. A stale ROSTER only misleads someone who is + looking at it, so recomputing it when they look is exactly right. A lane + running at the wrong size is doing less work than it was told to whether + or not anyone is watching — and the case that matters is a deploy at 3am + followed by a backlog nobody is awake to notice. + + So: unattended, on the quick `maintenance` lane (its module routes it + there), beside the other recovery sweeps. The accepted cost is that a dead + scheduler stops reconciliation — but a dead scheduler already stops every + other sweep, and the roster reports it, so this adds no new blind spot. + + ## Quiet when settled + + One broker round trip per tick, and no control messages at all once every + lane matches. Returns the lanes it actually moved, so the log shows a + correction rather than a heartbeat. + """ + from ..models import WorkerLane + from ..services.worker_control import reconcile_lanes_sync + + with _sync_session_factory()() as session: + desired = { + row.name: (row.slots, row.enabled) + for row in session.execute(select(WorkerLane)).scalars() + } + if not desired: + # Migration 0103 seeds these, so an empty table means it has not run + # yet. Nothing to assert — and inventing defaults here would let this + # task disagree with the seed it is supposed to be enforcing. + return {"changed": [], "skipped": [], "failed": {}} + return reconcile_lanes_sync(desired) diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py index 47df672..77f1b63 100644 --- a/tests/test_worker_control.py +++ b/tests/test_worker_control.py @@ -37,13 +37,20 @@ class _Control: self.cancelled.append((queue, destination)) -def _stub_live(monkeypatch, lane_name, *, pools, present=True, reserved=0): +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", @@ -186,3 +193,131 @@ 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 reconcile (step 3) -------------------------------------------------- + + +def test_a_settled_system_sends_no_control_messages(monkeypatch): + """THE property this sweep lives or dies on. It runs every 5 minutes + forever, so a converged tick must be silent — one broker round trip and + 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) + _stub_live(monkeypatch, "worker", pools={"host-a": 4}) + + result = wc.reconcile_lanes_sync({"worker": (4, True)}) + + # EVERY control family, not just the pool ones. The first version of this + # test asserted only grew/shrank and would have passed while the reconcile + # re-sent add_consumer for all four queues on every tick — harmless per + # call, unbounded churn in aggregate, and invisible. + assert control.grew == [] + assert control.shrank == [] + assert control.added == [] + assert control.cancelled == [] + assert result["changed"] == [] + + +def test_a_worker_back_at_its_env_concurrency_is_corrected(monkeypatch): + """The failure this exists for: a restart drops the pool to CELERY_ + CONCURRENCY, silently below what the operator set.""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={"host-a": 2}) + + result = wc.reconcile_lanes_sync({"worker": (6, True)}) + + assert control.grew == [(4, ["host-a"])] + assert result["changed"] == ["worker"] + + +def test_an_absent_lane_is_skipped_not_corrected(monkeypatch): + """`present=False` is 'nothing answered', not 'zero slots'. Correcting it + would be a conclusion drawn from an unswept read (snippet #3969) — and + there is nothing to send the message to anyway.""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={}, present=False) + + result = wc.reconcile_lanes_sync({"worker": (6, True)}) + + assert result["skipped"] == ["worker"] + assert result["changed"] == [] + assert control.grew == [] + assert control.shrank == [] + + +def test_one_lane_failing_does_not_stop_the_others(monkeypatch): + """A broker blip on one lane must not leave the rest un-reconciled for + another five minutes.""" + control = _stub_control(monkeypatch) + present = wc.LaneLiveState( + present=True, replicas=1, hostnames=["host-a"], pools={"host-a": 1}, + ) + absent = wc.LaneLiveState() + monkeypatch.setattr( + wc, "inspect_lanes_sync", + lambda: {"worker": absent, "scheduler": present, + "maintenance_long": absent, "ml": absent}, + ) + + result = wc.reconcile_lanes_sync({ + "worker": (4, True), "scheduler": (3, True), + }) + + assert result["skipped"] == ["worker"] + assert result["changed"] == ["scheduler"] + assert control.grew == [(2, ["host-a"])] + + +def test_a_lane_disabled_in_settings_but_still_consuming_is_stopped(monkeypatch): + """The case that made `consuming` necessary: a lane turned off while its + worker was down comes back consuming, and must be stopped when it + returns.""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "ml", pools={"host-a": 0}, consuming={"ml"}) + + result = wc.reconcile_lanes_sync({"ml": (0, False)}) + + assert control.cancelled == [("ml", ["host-a"])] + assert result["changed"] == ["ml"] + + +def test_an_already_disabled_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 + reconcile would churn here forever with no symptom.""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "ml", pools={"host-a": 0}, consuming=set()) + + result = wc.reconcile_lanes_sync({"ml": (0, False)}) + + assert control.cancelled == [] + assert control.added == [] + assert result["changed"] == [] + + +def test_a_lane_with_no_stored_row_is_left_alone(monkeypatch): + """A lane in LANES but not in `desired` is one whose row has not been + seeded. Inventing a target here would let the sweep enforce a number that + disagrees with the migration it is supposed to be upholding.""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={"host-a": 2}) + + result = wc.reconcile_lanes_sync({}) + + assert result["changed"] == [] + assert control.grew == [] + + +def test_the_reconcile_task_is_registered_and_scheduled(): + """A task name only enters `celery.tasks` when its module is imported, and + a beat entry naming a task that is not registered fails at tick time + rather than at import — silently, every five minutes.""" + import backend.app.tasks.maintenance # noqa: F401 + from backend.app.celery_app import celery + + name = "backend.app.tasks.maintenance.reconcile_worker_lanes" + assert name in celery.tasks + scheduled = {e["task"] for e in celery.conf.beat_schedule.values()} + assert name in scheduled