feat: reconcile every running lane back to its stored slots (4293)

Milestone 422 step 3. `pool_grow` is not durable: a worker restarted by its
supervisor comes back at its ENV concurrency, silently below whatever the
operator set, and nothing on step 2's write path would ever notice. Storing
the value made it survivable; this makes it survive.

A BEAT TASK, NOT A HOOK IN WEB — a deliberate deviation from the step as
written, for a reason already recorded in this codebase. Step 3 said "web
applies the stored values after it starts". It cannot: service_roster.py
documents that hypercorn runs --workers 4, so anything in before_serving
becomes four concurrent loops per container hammering the broker forever.

service_roster's own answer — refresh on demand from whichever request
arrives — was also rejected, because the two solve different problems. A
stale ROSTER only misleads someone looking at it, so recomputing 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 see.

So: unattended, every 5 minutes, on the quick `maintenance` lane beside the
other recovery sweeps. Accepted cost — 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.

A BUG I WROTE AND CAUGHT BEFORE COMMITTING. The first version called
set_lane_enabled_sync unconditionally, so a settled system re-sent
add_consumer for every queue on every tick — forever. Harmless per call
(add_consumer on an already-consumed queue does nothing), unbounded in
aggregate, and completely invisible. That is lesson #4183's failure mode
exactly, in the very function whose docstring cites it.

Worse, my test would not have caught it: it asserted only on grew/shrank.
LaneLiveState now carries `consuming` — which queues a lane is actually
serving, distinct from the queues it was configured with — so the reconcile
compares before acting. The test now asserts ALL FOUR control families are
silent on a settled tick, plus a new case for an already-disabled lane, which
is the other half of the same fixed point.

An absent lane is SKIPPED, not corrected. present=False means nothing
answered, not zero slots; correcting it would be a conclusion from an unswept
read (snippet #3969), and there would be nothing to send the message to. One
lane failing does not stop the others.

One inspect serves every lane: the two setters now take an optional
pre-fetched LaneLiveState, so a tick costs one broker round trip rather than
one per lane.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

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}