diff --git a/alembic/versions/0105_worker_lane_one_cap.py b/alembic/versions/0105_worker_lane_one_cap.py new file mode 100644 index 0000000..01b4813 --- /dev/null +++ b/alembic/versions/0105_worker_lane_one_cap.py @@ -0,0 +1,123 @@ +"""worker_lane — one number: the cap. `slots`, `enabled` and `autoscale` go. + +Milestone 422, reshaped by the operator 2026-09-23: + + "auto should be always on, not a setting, so that idle instances quiet + down when not running. the number that is visible and something the user + can tweak and manage should be the cap itself the number of running + workers is handled by the autoscaling function which is always on." + +## What each dropped column was, and why it is not needed + +**`slots`** — how many workers the lane should run. That is a MEASUREMENT, +not a preference: the autoscaler moves the live pool between one and the cap +according to the backlog, and reads it back from the worker every minute. +Storing it made it look like something to keep in agreement with the cap, +which is exactly what the operator had to do. + +**`autoscale`** — whether the lane was allowed to size itself. It gated the +mechanism behind a per-lane opt-in, so a lane nobody enabled simply never +gave its slots back. Always on now, which is the only way "idle instances +quiet down" can be true of an install nobody has configured. + +**`enabled`** — whether the lane consumes its queues. Derived from `cap > 0`. +It and `slots = 0` were two spellings of one fact and were free to disagree; +this migration picks the one an operator can see. + +## Why the caps are rewritten rather than preserved + +The old defaults were 4 / 2 / 2 / 1, chosen when the number meant "the most +you may raise SLOTS to" — a bound on a manual control, deliberately loose +because moving within it was the ordinary act. The number now means "the most +workers this lane may actually use", which is a different promise, and +carrying the old figure over would silently quadruple the worker lane on +every existing install at the moment this deploys. + +So every row is reset to the new defaults: **one for each required lane, zero +for ML.** That loses whatever an operator had set — which is the honest +trade, because what they set was an answer to a different question. The UI +now tells a busy lane's operator to raise its cap, which is how the number +gets back up on an install that needs it. + +ML at zero also keeps rule 164's carve-out intact: no consumers, so no model +download until someone raises the cap. + +Revision ID: 0105 +Revises: 0104 +Create Date: 2026-09-23 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0105" +down_revision: Union[str, None] = "0104" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +# (name, cap) — the same values `services/worker_lanes.LANES` declares. Seeded +# here as literals rather than imported: a migration must describe the schema +# at ITS point in history, and importing the live table would make this file +# change meaning every time that table does. +_CAPS = ( + ("worker", 1), + ("scheduler", 1), + ("maintenance_long", 1), + ("ml", 0), +) + + +def upgrade() -> None: + # The constraint goes first: it names `slots`, so dropping the column out + # from under it fails on Postgres. + op.drop_constraint("ck_worker_lane_slots_within_cap", "worker_lane", type_="check") + op.drop_constraint( + "ck_worker_lane_slots_non_negative", "worker_lane", type_="check", + ) + op.drop_column("worker_lane", "slots") + op.drop_column("worker_lane", "enabled") + op.drop_column("worker_lane", "autoscale") + + # Reset to the new meaning. See the docstring: the old value answered a + # different question, and carrying it over would raise every lane. + for name, cap in _CAPS: + op.execute( + sa.text("UPDATE worker_lane SET slots_cap = :cap WHERE name = :name") + .bindparams(cap=cap, name=name) + ) + + # A lane the old seed never wrote — or one an operator added by hand — is + # left alone rather than guessed at. `_rows_by_name` creates any missing + # row at the lane's default on first read. + + +def downgrade() -> None: + op.add_column( + "worker_lane", + sa.Column("slots", sa.Integer(), nullable=False, server_default="1"), + ) + op.add_column( + "worker_lane", + sa.Column( + "enabled", sa.Boolean(), nullable=False, server_default=sa.text("true"), + ), + ) + op.add_column( + "worker_lane", + sa.Column( + "autoscale", sa.Boolean(), nullable=False, server_default=sa.text("false"), + ), + ) + # Restore the pre-0105 invariants. `slots` comes back as 1 everywhere and + # the caps are 1/1/1/0, so a lane at cap 0 would violate `slots <= cap` — + # hence the clamp before the constraint is added. + op.execute(sa.text("UPDATE worker_lane SET slots = 0 WHERE slots_cap = 0")) + op.execute(sa.text("UPDATE worker_lane SET enabled = (slots_cap > 0)")) + op.create_check_constraint( + op.f("ck_worker_lane_slots_non_negative"), "worker_lane", "slots >= 0", + ) + op.create_check_constraint( + op.f("ck_worker_lane_slots_within_cap"), "worker_lane", "slots <= slots_cap", + ) diff --git a/backend/app/api/workers.py b/backend/app/api/workers.py index 678c061..10c03ee 100644 --- a/backend/app/api/workers.py +++ b/backend/app/api/workers.py @@ -10,10 +10,10 @@ answers a different question: its `/workers` is keyed on celery HOSTNAME and reports which nodes answered. That stays as it is — the existing SystemActivityTab consumes it. -This is keyed on LANE, joins the stored settings to the live pool, and -accepts writes. Two endpoints answering "which celery processes exist" and -"how much work is each lane allowed to do" are not the same endpoint, and -folding the second into the first would make a read-only module a write one. +This is keyed on LANE, joins the stored cap to the live pool, and accepts +writes. Two endpoints answering "which celery processes exist" and "how much +work is each lane allowed to do" are not the same endpoint, and folding the +second into the first would make a read-only module a write one. """ from __future__ import annotations @@ -32,7 +32,7 @@ workers_bp = Blueprint("workers", __name__, url_prefix="/api/system/workers") @workers_bp.route("", methods=["GET"]) async def list_lanes(): - """Every lane: configured slots, the cap, the ceiling, and live state. + """Every lane: its cap, the ceiling above it, and what is live. Response: {lanes: [...], fetched_at: iso8601} @@ -51,20 +51,22 @@ async def list_lanes(): @workers_bp.route("/", methods=["POST"]) async def update_lane(name: str): - """Set a lane's slots, cap and/or enabled flag. Stores, then pushes live. + """Set a lane's cap. Stores it, then makes the live lane obey it. - Partial: only the keys present are changed, so the UI's stepper can send - `{"slots": 3}` without restating the cap it did not touch. + ONE field, since 2026-09-23. It used to take `slots`, `slots_cap`, + `enabled` and `autoscale`; how many workers are running is now a + measurement the sizing pass owns, and `enabled` is `cap > 0`. Two failure kinds, deliberately different statuses: - * **400** — the value is not allowed (above the cap, above the ceiling, - negative). Nothing was stored. The body carries `detail`, which is the - sentence the UI shows; a refused control with no reason reads as a bug. + * **400** — the value is not allowed (negative, or above what this + container can hold). Nothing was stored. The body carries `detail`, + which is the sentence the UI shows; a refused control with no reason + reads as a bug. * **200 with `applied: false`** — the value WAS stored but could not be - pushed, because the lane is not currently answering. That is not an - error: step 3's reconcile carries it when the lane comes back, and the - UI should say "saved, not yet live" rather than "that didn't work". + pushed, because the lane is not currently answering. Not an error: the + sizing pass carries it within a minute, and the UI should say "saved, + not yet live" rather than "that didn't work". """ lane = LANES_BY_NAME.get(name) if lane is None: @@ -74,31 +76,18 @@ async def update_lane(name: str): if not isinstance(body, dict): return _bad("invalid_body", detail="body must be a JSON object") - fields: dict = {} - for key in ("slots", "slots_cap"): - if key in body: - value = body[key] - # Rejected rather than coerced: `True` is an int in Python, and - # silently reading it as 1 slot would be a control that appears to - # work and sets something nobody asked for. - if not isinstance(value, int) or isinstance(value, bool): - return _bad("invalid_body", detail=f"{key} must be an integer") - fields[key] = value - for key in ("enabled", "autoscale"): - if key in body: - if not isinstance(body[key], bool): - return _bad("invalid_body", detail=f"{key} must be a boolean") - fields[key] = body[key] - - if not fields: - return _bad( - "invalid_body", - detail="give at least one of slots, slots_cap, enabled, autoscale", - ) + if "slots_cap" not in body: + return _bad("invalid_body", detail="give slots_cap") + value = body["slots_cap"] + # Rejected rather than coerced: `True` is an int in Python, and silently + # reading it as a cap of 1 would be a control that appears to work and + # sets something nobody asked for. + if not isinstance(value, int) or isinstance(value, bool): + return _bad("invalid_body", detail="slots_cap must be an integer") async with get_session() as session: try: - result = await set_lane(session, lane, **fields) + result = await set_lane(session, lane, slots_cap=value) except LaneUpdateRefused as exc: return _bad("refused", detail=str(exc)) return jsonify(result) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index e509f8f..21c98bf 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -111,21 +111,24 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.recover_interrupted_tasks", "schedule": 300.0, # every 5 minutes }, - "autoscale-worker-lanes": { - "task": "backend.app.tasks.maintenance.autoscale_worker_lanes", - "schedule": 60.0, # every minute — it reacts to a BACKLOG, and - # a five-minute reaction to a queue filling up is no reaction. - # Cheap when nothing opted in (one DB read, then nothing) and - # cheap when settled (one inspect + one LLEN sweep, no control - # messages), so the short interval costs little. - }, - "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. + "size-worker-lanes": { + "task": "backend.app.tasks.maintenance.size_worker_lanes", + "schedule": 60.0, # every minute. + # + # ONE entry, replacing `autoscale-worker-lanes` (60s) and + # `reconcile-worker-lanes` (300s) on 2026-09-23. They were two + # sweeps over one number and most of the autoscaler's design + # existed to stop the reconcile undoing its work; with the + # stored `slots` gone there is nothing to disagree about. + # + # A minute because it reacts to a BACKLOG, and a five-minute + # reaction to a queue filling up is no reaction. It also now + # carries what the reconcile was for — a worker restarted at + # its ENV concurrency is corrected on the next tick rather + # than after five. + # + # Cheap when settled: one inspect plus one LLEN sweep, and no + # control messages at all once every lane matches. }, "cleanup-old-tasks": { "task": "backend.app.tasks.maintenance.cleanup_old_tasks", diff --git a/backend/app/models/worker_lane.py b/backend/app/models/worker_lane.py index 94daf45..6504656 100644 --- a/backend/app/models/worker_lane.py +++ b/backend/app/models/worker_lane.py @@ -1,48 +1,49 @@ -"""worker_lane — how many slots the operator wants each worker lane to have. +"""worker_lane — the most workers the operator will let each lane use. -Milestone 422 step 1. One row per lane in `services/worker_lanes.LANES`. +Milestone 422 step 1, reshaped 2026-09-23. One row per lane in +`services/worker_lanes.LANES`, and ONE COLUMN an operator sets. -## What is NOT in here +## Why there is only one number now -The lane's queues and its display name. Those are decided by `celery_app.py`'s -`task_routes` — an operator cannot move a backup off `maintenance_long` — so -storing them would be a row that can contradict the routing table, with -nothing to notice until a queue had no consumer. They live in -`services/worker_lanes.py`; this table holds only what an operator may change. +There were three — `slots`, `slots_cap` and `autoscale` — because the manual +dial was built first and the autoscaler arrived last, beside a control that +already existed rather than in place of it. -The DERIVED CEILING is also absent, and that is deliberate rather than an -omission. It is computed from the container's cgroup limits on every read, so -a row written on a 32GB host and later run in a 4GB container is bounded by -the 4GB — a stored ceiling would quietly authorise what the box can no longer -hold. +Operator: *"auto should be always on, not a setting, so that idle instances +quiet down when not running. the number that is visible and something the +user can tweak and manage should be the cap itself the number of running +workers is handled by the autoscaling function which is always on."* -## The three numbers +So `slots` is gone. How many workers a lane is running right now is a +MEASUREMENT — read live from the worker, moved by the autoscaler, never +stored. Storing it made it look like a preference, which meant the operator +had to keep two numbers in agreement and the autoscaler had to be told it was +allowed to touch one of them. - slots <= slots_cap <= derived_ceiling - (live) (this row) (computed) +`autoscale` is gone for the same reason: it gated the mechanism behind a +choice, and a lane nobody opted in simply never gave its slots back. -Operator's distinction, 2026-09-22: the derived value is *a cap on the cap*. -`slots_cap` is theirs and is always lowerable; it simply may not exceed what -the container can hold. The CHECK constraint below enforces the left half, -which is a fact about the row; the right half is enforced at write, because -it depends on a value no database column holds. +`enabled` is gone too, and is now DERIVED: a cap of zero means no consumers. +"Off" and "may use no workers" were two spellings of one fact, free to +disagree. -## enabled +## What remains -Whether the lane consumes its queues at all. This is how ML ships off -(milestone 422 step 6): `enabled=false` with `slots=0`, so a fresh install -never loads a model or reaches HuggingFace, and turning tagging on in -Settings is what triggers the fetch. + 1 <= live pool <= slots_cap <= derived_ceiling + (autoscaler) (this row) (computed) -Not a substitute for `slots=0`. A lane can be enabled with zero slots while -it is being resized, and the two answer different questions: `enabled` is -intent, `slots` is capacity. +The floor is one PROCESS, not zero: billiard will not run an empty pool, and +the parked process is what `add_consumer` lands on when the cap goes back up. + +The DERIVED CEILING is deliberately absent from this table. It is computed +from the container's cgroup limits on every read, so a row written on a 32GB +host and later run in a 4GB container is bounded by the 4GB — a stored +ceiling would quietly authorise what the box can no longer hold. """ from datetime import datetime from sqlalchemy import ( - Boolean, CheckConstraint, DateTime, Integer, @@ -57,16 +58,10 @@ from .base import Base class WorkerLane(Base): __tablename__ = "worker_lane" __table_args__ = ( - # Bare names — Base.metadata's naming convention prepends + # Bare name — Base.metadata's naming convention prepends # ck_worker_lane_. Pre-prefixing here doubles it, which is what # alembic 0088 had to rename four constraints for (#3275). - CheckConstraint("slots >= 0", name="slots_non_negative"), CheckConstraint("slots_cap >= 0", name="cap_non_negative"), - # The invariant that makes the cap mean anything. Enforced in the - # database rather than only in the service, because a row that - # violates it is not a rejected request — it is a lane that will be - # reconciled UP to a value the operator capped. - CheckConstraint("slots <= slots_cap", name="slots_within_cap"), ) # The lane name from services/worker_lanes.LANES — never a container @@ -74,21 +69,13 @@ class WorkerLane(Base): # are `celery@`, minted fresh on every deploy. name: Mapped[str] = mapped_column(String(32), primary_key=True) - slots: Mapped[int] = mapped_column(Integer, nullable=False) - slots_cap: Mapped[int] = mapped_column(Integer, nullable=False) - enabled: Mapped[bool] = mapped_column(Boolean, nullable=False) - - # Whether the autoscaler may raise this lane's slots on its own, up to - # slots_cap and never past it (milestone 422 step 7). + # The most workers this lane may use. Zero means off — no consumers, so + # the lane takes no work and (for ML) downloads no model. # - # OFF by default and per-lane rather than global. It is the one part of - # this milestone that acts unattended, so it is opt-in per lane the - # operator has actually thought about — a global switch would turn it on - # for lanes whose behaviour under load nobody has watched, including `ml`, - # where every extra slot is another copy of a multi-GB model. - autoscale: Mapped[bool] = mapped_column( - Boolean, nullable=False, server_default="false", - ) + # There is no upper CHECK here, because the bound it would need is the + # derived ceiling, and no column holds that: it depends on the cgroup the + # container is running in right now. Enforced at write instead. + slots_cap: Mapped[int] = mapped_column(Integer, nullable=False) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), diff --git a/backend/app/scripts/gen_supervisord.py b/backend/app/scripts/gen_supervisord.py index 447e66b..e8c92fc 100644 --- a/backend/app/scripts/gen_supervisord.py +++ b/backend/app/scripts/gen_supervisord.py @@ -54,7 +54,7 @@ import argparse import shlex import sys -from ..services.worker_lanes import LANES, Lane +from ..services.worker_lanes import LANES, MIN_POOL_SLOTS, Lane # One number for the whole container, and it must cover the SLOWEST lane — # docker gives the container a single stop timeout, where compose today gives @@ -212,7 +212,7 @@ def render() -> str: # worker there is nothing for `add_consumer` to reach, so enabling the # lane from the UI could not work at all — the process has to exist for # the switch to have something to switch. - parts.append(_program(lane, slots=max(1, lane.default_slots))) + parts.append(_program(lane, slots=MIN_POOL_SLOTS)) return "\n".join(parts) diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 7e31f09..641c545 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -56,7 +56,13 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import TaskRun, WorkerLane -from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane, derived_ceiling +from .worker_lanes import ( + LANES, + LANES_BY_QUEUE_KEY, + MIN_POOL_SLOTS, + Lane, + derived_ceiling, +) log = logging.getLogger(__name__) @@ -66,28 +72,6 @@ log = logging.getLogger(__name__) # report "not present", which is true, rather than hang the page. CONTROL_TIMEOUT_SECONDS = 2.0 -# A celery prefork pool cannot have ZERO processes, and a lane at zero slots -# is expressed by cancelling its consumers rather than by emptying its pool. -# -# The generator already starts every lane at one process for exactly this -# reason: `add_consumer` needs something to reach, so enabling a lane from the -# UI would be impossible if no process existed. The consequence was missed -# until the operator's live deploy, 2026-09-23: -# -# [scheduler] worker_control: ml reconciled 1 -> 0 slots -# [ml] pidbox command error: ValueError("Can't shrink pool. All processes -# busy!") -# -# billiard refuses to remove the last worker, so ml could never reach 0. And -# `set_lane_slots_sync` returns True on SENDING the control message — the -# failure happens later, on the worker — so the reconcile logged success and -# reported `changed: ['ml']` on every single tick. An enforcer with no -# reachable fixed point, re-doing its own work forever and saying it worked: -# lesson #4183, in production, on the default configuration of every install. -# -# So the floor is one process. Zero slots still means zero WORK, because the -# consumers are cancelled — the idle process is the switch's landing pad. -MIN_POOL_SLOTS = 1 # The WORST case of `inspect_lanes_sync`, for callers that need a deadline. # @@ -342,13 +326,7 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]: } missing = [lane for lane in LANES if lane.name not in rows] for lane in missing: - row = WorkerLane( - name=lane.name, - slots=lane.default_slots, - slots_cap=lane.default_slots_cap, - enabled=lane.default_enabled, - autoscale=lane.default_autoscale, - ) + row = WorkerLane(name=lane.name, slots_cap=lane.default_slots_cap) session.add(row) rows[lane.name] = row if missing: @@ -398,17 +376,18 @@ async def lane_view(session: AsyncSession) -> list[dict]: "name": lane.name, "display_name": lane.display_name, "queues": list(lane.queues), - "slots": row.slots, "slots_cap": row.slots_cap, "ceiling": derived_ceiling(lane), - "enabled": row.enabled, - "autoscale": row.autoscale, + # DERIVED, never stored. A cap of zero means no consumers, so + # "off" and "may use no workers" cannot disagree. + "enabled": row.slots_cap > 0, "memory_bound": lane.memory_bound, "optional": lane.optional, - # What enabling this lane will download, so the UI can say WHICH - # model and how big BEFORE the switch is thrown rather than after - # a multi-GB fetch has started. `measured` travels with the - # numbers: the card must not present an estimate as a fact. + # What raising this lane's cap will download, so the UI can say + # WHICH model and how big BEFORE the first slot is asked for + # rather than after a multi-GB fetch has started. `measured` + # travels with the numbers: the UI must not present an estimate + # as a fact. "models": [ { "repo": m.repo, @@ -506,119 +485,89 @@ class LaneUpdateRefused(ValueError): the UI shows — a greyed control with no explanation reads as a bug.""" -async def set_lane( - session: AsyncSession, - lane: Lane, - *, - slots: int | None = None, - slots_cap: int | None = None, - enabled: bool | None = None, - autoscale: bool | None = None, -) -> dict: - """Store the operator's choice, then push it to the running lane. +async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict: + """Store the operator's cap for `lane`, then make the live lane obey it. - BOTH, in one call, and the order matters. `pool_grow`/`pool_shrink` are - not durable — a restart drops every lane back to its env concurrency — so - a UI that only pushed would have its setting evaporate on the next deploy - with nothing to show for it (lesson #4202: the live change does not - survive, and nothing says so). Storing alone would be a number that - describes nothing until something restarts. + ONE value, since 2026-09-23. It used to take `slots`, `slots_cap`, + `enabled` and `autoscale`, which was four ways of saying two things — and + two of them were the caller's job to keep in agreement with each other. + + ## What happens live, and what does not + + A cap CHANGE is pushed immediately in one direction only: lowering it + shrinks the pool now, because a cap the operator just lowered should not + be exceeded for up to a minute. Raising it does NOT grow the pool here — + a cap is permission, not a request, and growing on permission would put + slots on a lane with nothing to do. The sizing pass adds them on its next + tick if there is work, which is the whole point of it being always on. + + Consumers follow the cap in both directions and immediately: zero means + off, and off must take effect when it is asked for. A failed PUSH is not a failed setting. The value is saved either way and - step 3's reconcile carries it when the lane answers again; the result says - `applied: false` with a reason so the UI can say "saved, not yet live" - rather than "that didn't work". + the sizing pass carries it within a minute; the result says `applied: + false` with a reason so the UI can say "saved, not yet live" rather than + "that didn't work" (lesson #4202 — a live change that does not survive, + with nothing saying so). """ rows = await _rows_by_name(session) row = rows[lane.name] - new_cap = row.slots_cap if slots_cap is None else slots_cap - new_slots = row.slots if slots is None else slots - new_autoscale = row.autoscale if autoscale is None else autoscale - - # THE DIAL IS THE SWITCH. A lane at zero slots is a lane that is off, and - # there is no second control saying so. - # - # Operator, 2026-09-23, on the card that had both: *"there's nothing to - # describe what 'auto' means or why their needs to be or should be on/off - # toggles. almost all of it always needs to run there's only one optional - # piece and it is killed by moving the 'cap' to zero."* They are right. Of - # four lanes, three must run for the application to work at all, so a - # switch beside each of them offered a choice that was never real — and - # for the one lane that IS optional, "off" and "zero slots" were two ways - # of saying the same thing that could disagree with each other. - # - # `enabled` stays in the model and on the API. It is still the mechanism: - # a disabled lane keeps its process and cancels its consumers, which is - # what makes it visible in the roster instead of looking like a crash. It - # is now DERIVED from the number the operator actually sets, rather than - # being a second thing for them to keep in agreement with it. - was_enabled = row.enabled - if enabled is not None: - new_enabled = enabled - elif slots is not None: - new_enabled = new_slots > 0 - else: - new_enabled = row.enabled - ceiling = derived_ceiling(lane) - if new_cap < 0 or new_slots < 0: - raise LaneUpdateRefused("slots and cap cannot be negative") - if new_cap > ceiling: + if slots_cap < 0: + raise LaneUpdateRefused("a cap cannot be negative") + if slots_cap > ceiling: raise LaneUpdateRefused( - f"cap {new_cap} is above what this container can hold " + f"a cap of {slots_cap} is above what this container can hold " f"({ceiling} for {lane.display_name})" ) - if new_slots > new_cap: - raise LaneUpdateRefused(f"slots {new_slots} is above the cap {new_cap}") - row.slots_cap = new_cap - row.slots = new_slots - row.enabled = new_enabled - row.autoscale = new_autoscale + was_on = row.slots_cap > 0 + row.slots_cap = slots_cap await session.commit() + now_on = slots_cap > 0 applied, error = True, None - # On the CHANGE, not on the field being present. Now that `enabled` is - # derived, every slots write would otherwise re-send a consumer command - # that changes nothing — the churn lesson #4183 keeps producing, arriving - # here through the new derivation. - if new_enabled != was_enabled: + if now_on != was_on: + applied, error = await asyncio.to_thread(set_lane_enabled_sync, lane, now_on) + if applied and not now_on: + # Down to the floor at once. The pool cannot be emptied, so "off" is + # one parked process with its consumers cancelled. applied, error = await asyncio.to_thread( - set_lane_enabled_sync, lane, new_enabled, + set_lane_slots_sync, lane, MIN_POOL_SLOTS, ) - if applied and slots is not None: - applied, error = await asyncio.to_thread(set_lane_slots_sync, lane, new_slots) + elif applied and now_on: + # Only DOWNWARD. See the docstring: raising a cap is permission, and + # the sizing pass decides whether there is work to spend it on. + live = await asyncio.to_thread(inspect_lanes_sync) + current = live[lane.name].pool + if current is not None and current > slots_cap: + applied, error = await asyncio.to_thread( + set_lane_slots_sync, lane, slots_cap, live=live[lane.name], + ) - # Enabling a lane that needs models is what triggers the fetch (milestone + # Raising the cap off zero is what triggers the model download (milestone # 422 step 6). Never at boot: that made every start of the ML role reach # HuggingFace for ~3.5GB, and rule 164 permits a runtime fetch only for a # feature that is optional and clearly OFF. # - # Only on the TRANSITION from off to on, so re-saving slots on a lane that - # is already running does not re-enqueue. This used to test `enabled is - # True` — the field having been sent — which stopped meaning "came on" the - # moment the dial became the switch: the UI no longer sends `enabled` at - # all, so the fetch that makes the ML lane usable would never have fired. - # - # And only when the consumer change landed: enqueueing onto a queue - # nothing is consuming would leave the task pending with no explanation - # until the lane returns. + # On the TRANSITION, so re-saving a cap on a lane already running does not + # re-enqueue. And only when the consumer change landed: enqueueing onto a + # queue nothing is consuming would leave the task pending with no + # explanation until the lane returns. fetching = False - if new_enabled and not was_enabled and lane.models and applied: + if now_on and not was_on and lane.models and applied: fetching = _enqueue_model_fetch() return { "name": lane.name, - "slots": row.slots, "slots_cap": row.slots_cap, "ceiling": ceiling, - "enabled": row.enabled, - "autoscale": row.autoscale, + "enabled": now_on, "applied": applied, "apply_error": error, - # Tells the card to say a download has started rather than leaving the - # operator to wonder why a freshly enabled lane is busy. + # Tells the UI to say a download has started rather than leaving the + # operator to wonder why a lane they just turned on is busy. "fetching_models": fetching, } @@ -645,263 +594,166 @@ def _enqueue_model_fetch() -> bool: return False -def reconcile_lanes_sync( - desired: dict[str, tuple[int, bool]], - autoscaling: frozenset[str] = frozenset(), -) -> 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. `autoscaling` names the lanes the autoscaler is allowed to move. - - ## For an autoscaling lane the stored value is a FLOOR, not a target - - Step 7's autoscaler raises a saturated lane's live pool without changing - its row — the row holds what the OPERATOR set. If this pass treated that - row as an exact target it would shrink the lane back on the very next - tick, and the two sweeps would fight forever at five-minute intervals: - grow, revert, grow, revert. That is lesson #4183's failure arriving - between two enforcers rather than inside one. - - So for those lanes the target becomes `max(stored, current)` — this pass - still restores a lane that came back from a restart below what the - operator set, and never takes back what the autoscaler added. Bringing it - down is the autoscaler's job, and it does so only to that same floor. 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 - # The floor, for a lane the autoscaler manages. See the docstring. - target_slots = slots - if lane.name in autoscaling and current is not None: - target_slots = max(slots, current) - # And the floor every lane has: a pool cannot be emptied. Applied to - # the COMPARISON, not just the send — against the raw 0 this sees a - # difference no control message can close and re-sends it every tick. - target_slots = effective_slots(target_slots) - if current is not None and current == target_slots: - continue - ok, err = set_lane_slots_sync(lane, target_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, target_slots, - ) - else: - failed[lane.name] = err or "could not resize" - - return {"changed": changed, "skipped": skipped, "failed": failed} - - -# --- the autoscaler (step 7) -------------------------------------------------- +# --- the sizing pass: one sweep, always on ------------------------------------ # -# The only part of this milestone that acts without anyone asking. Everything -# above does what an operator pressed; this decides. So it is off by default, -# opted into per lane, bounded by the cap the operator set, and it reports what -# it did rather than moving numbers silently. - -# ## Why these three numbers are not in Settings +# This replaced BOTH `reconcile_lanes_sync` (step 3) and `autoscale_lanes_sync` +# (step 7) on 2026-09-23. They were two enforcers over one number, and the +# whole of step 7's hardest reasoning — a stored value that is a FLOOR, a +# target of `max(stored, current)` so the reconcile does not undo what the +# autoscaler added — existed only to stop them fighting. Delete one of them and +# the problem is not solved, it is absent. # -# Rule 25 puts anything an operator might want to tune in the UI, and the -# knobs that decide what this does ARE there: whether a lane autoscales at -# all, its cap, and its floor — all DB-backed, all changeable without a -# restart. What is left here is the POLICY's internals, and exposing them -# would add four numbers per lane to a card whose whole value is being -# readable at a glance, to tune a decision the operator has a better lever -# for. If growth turns out to be too eager or too shy in practice, that is a -# reason to change these values for everyone, not to ask each operator to -# discover them. - -# All slots busy AND this many tasks waiting before a lane may grow. +# Operator: *"auto should be always on, not a setting, so that idle instances +# quiet down when not running. the number that is visible and something the +# user can tweak and manage should be the cap itself the number of running +# workers is handled by the autoscaling function which is always on."* # -# The AND is the design. Depth with free slots means nothing — celery is about -# to pick those up, and growing the pool would add idle children. Saturation -# with an empty queue means nothing either: the lane is busy with exactly as -# much work as exists. Only both together say "there is more work than this -# lane can reach". -AUTOSCALE_BACKLOG_THRESHOLD = 10 +# So there is one pass, it runs every minute, it reads the live pool rather +# than any stored number, and the only thing it obeys is the cap. +# +# It also subsumes what the reconcile existed for. `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. This pass reads the live +# pool every minute and sizes from the backlog, so that worker is corrected on +# the next tick — sooner than the five-minute reconcile managed, and without a +# second sweep that could disagree with this one. -# Grow by one slot per tick, never to the cap in one jump. A lane that is -# saturated because of one slow burst settles a slot or two above where it -# started rather than at its ceiling, and the next tick re-measures rather -# than committing to a guess made once. -AUTOSCALE_STEP = 1 - -# Hysteresis: shrink only when the backlog is well BELOW the grow threshold, -# not merely under it. Equal thresholds flap — one task arriving and leaving -# would grow and shrink the lane forever at the tick interval, which is -# lesson #4183's churn arriving through a different door. -AUTOSCALE_SHRINK_BELOW = 2 +# How much work justifies a slot. `pending` is depth + reserved, so it already +# counts what celery has prefetched into worker memory — one task, one slot. +# +# Growth is IMMEDIATE and shrink is one slot per tick, deliberately asymmetric. +# A backlog of four thousand should not take an hour to reach the cap, and a +# lane that idles for one minute should not drop every process it has: the +# cost of being one slot too large for a minute is a sleeping process, and the +# cost of being too small is work not happening. For ML the asymmetry matters +# most — every new slot reloads a multi-GB model, so the slow shrink is what +# stops a quiet patch from paying that cost again a minute later. +SHRINK_STEP = 1 @dataclass -class AutoscaleDecision: - """What the autoscaler did to one lane, and why — in the operator's terms. +class LaneSizing: + """What the pass did to one lane, and why — in the operator's terms. - A reason string on every outcome including "nothing", because an - autoscaler that only speaks when it acts is one nobody can debug when it - does not. + A reason on every outcome including "held", because a sizing pass that + only speaks when it acts is one nobody can debug when it does not. """ lane: str - action: str # "grew" | "shrank" | "held" + action: str # "grew" | "shrank" | "held" | "skipped" slots: int reason: str -def autoscale_lanes_sync( - lanes: dict[str, tuple[int, int, bool]], -) -> list[AutoscaleDecision]: - """Decide and apply one round of autoscaling. +def wanted_slots(cap: int, active: int, pending: int | None) -> int: + """How many workers this lane has work for right now, within its cap. - `lanes` is name -> (slots_cap, configured_slots, autoscale_on), read from - the database by the caller — this function touches no database, for the - same reason `reconcile_lanes_sync` does not. + One slot per task in flight or waiting, floored at one process and + ceilinged by the cap. `pending` of None means the broker did not answer + for this lane's queues — an unknown backlog is not an empty one (snippet + #3969), so it contributes nothing rather than being read as zero. - ## What is current, and what is the floor + A cap of zero still returns one: billiard cannot run an empty pool, and + the parked process is what `add_consumer` lands on when the cap goes back + up. "Off" is expressed by cancelling consumers, not by emptying the pool. + """ + if cap <= 0: + return MIN_POOL_SLOTS + return max(MIN_POOL_SLOTS, min(cap, active + (pending or 0))) - The value this moves is the LIVE pool, read from `inspect`. The stored - `configured_slots` is what the operator set and is only the FLOOR: growth - goes above it and a shrink returns to it, never below. - The two are deliberately not the same number, and reading the stored value - as "current" is the mistake that makes this function useless in a way no - unit test of a single tick would show. The autoscaler never writes the - row, so the stored value never moves; a tick that computed `stored + 1` - would propose the same target forever, cap the lane one slot above the - floor no matter the load, and — because resizing a replica already at the - target issues nothing and reports success — claim `grew` on every tick - while nothing changed. Lesson #4183's non-convergence, arriving with a - success message attached. +def size_lanes_sync(caps: dict[str, int]) -> list[LaneSizing]: + """Size every lane to its backlog, within the cap. The whole control loop. - So: `current = state.pool`, `configured` is the floor, and both `grew` and - `shrank` mean the live pool actually moved. + `caps` is lane name -> slots_cap, read from the database by the caller. + This function touches no database: the celery task that schedules it owns + the session, and keeping the DB out of here is what lets it be called from + anywhere that already knows the caps. + + ## It must converge and then go quiet + + One `inspect` for all lanes, and `set_lane_slots_sync` issues nothing to a + replica already at its target. A settled system therefore performs one + broker round trip plus one LLEN sweep per tick and sends no control + messages at all — 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. + + ## An absent lane is SKIPPED, not corrected + + `present=False` means nothing answered — a worker restarting, or an + unreachable broker. It does NOT mean zero slots. Deciding from that would + be a verdict drawn from an unswept read, and here it is worse than + useless: there is nothing to send the message to. """ live = inspect_lanes_sync() depths = _queue_depths_sync() - out: list[AutoscaleDecision] = [] + out: list[LaneSizing] = [] for lane in LANES: - target = lanes.get(lane.name) - if target is None: + cap = caps.get(lane.name) + if cap is None: continue - cap, configured, on = target - if not on: + state = live[lane.name] + if not state.present: + out.append(LaneSizing(lane.name, "skipped", 0, "lane is not answering")) continue - state = live[lane.name] - if not state.present or state.pool is None: - # Nothing answered. Not "idle" — unknown, and a decision drawn - # from an unswept read is exactly what snippet #3969 warns about. - # `configured` is reported because there is no live number to - # report; it is what the lane will come back at. - out.append(AutoscaleDecision( - lane.name, "held", configured, "lane is not answering", - )) - continue + # Consumers first, and only when they DISAGREE. Sending add_consumer + # for every queue on every tick of a settled system is the exact churn + # above, and invisible: add_consumer on a queue already consumed is + # harmless and reports success. + should_consume = cap > 0 + if should_consume != state.consuming.issuperset(lane.queues): + ok, err = set_lane_enabled_sync(lane, should_consume, live=state) + if not ok: + out.append(LaneSizing( + lane.name, "held", state.pool or 0, + f"could not {'start' if should_consume else 'stop'} " + f"consuming: {err}", + )) + continue current = state.pool - known = [depths.get(q) for q in lane.queues] - if all(d is None for d in known): - out.append(AutoscaleDecision( - lane.name, "held", current, "queue depth unavailable", + if current is None: + out.append(LaneSizing( + lane.name, "held", 0, "worker did not report its pool size", )) continue - backlog = sum(d for d in known if d is not None) + state.reserved - # Against CAPACITY, not against the dial: `active` is summed across - # replicas, so comparing it to one replica's pool size would call two - # half-busy replicas of 4 saturated at 4 active and grow a lane that - # has idle slots. - saturated = state.active >= state.capacity > 0 - busy = backlog >= AUTOSCALE_BACKLOG_THRESHOLD - if saturated and busy and current < cap: - new = min(cap, current + AUTOSCALE_STEP) - ok, err = set_lane_slots_sync(lane, new, live=state) - out.append(AutoscaleDecision( - lane.name, "grew" if ok else "held", new if ok else current, - f"{backlog} waiting and all {state.capacity} slots busy" - if ok else f"could not grow: {err}", - )) - elif saturated and busy: - # At the cap with work still waiting. Said out loud rather than - # held silently: this is the operator's own ceiling doing its job, - # and it is the moment they would want to know they set it. - out.append(AutoscaleDecision( - lane.name, "held", current, - f"{backlog} waiting but the cap is {cap}", - )) - elif current > effective_slots(configured) and ( - backlog <= AUTOSCALE_SHRINK_BELOW - ): - new = max(effective_slots(configured), current - AUTOSCALE_STEP) - ok, err = set_lane_slots_sync(lane, new, live=state) - out.append(AutoscaleDecision( - lane.name, "shrank" if ok else "held", new if ok else current, - f"backlog cleared, back toward {configured}" - if ok else f"could not shrink: {err}", - )) + known = [depths.get(q) for q in lane.queues] + depth = ( + sum(d for d in known if d is not None) + if any(d is not None for d in known) else None + ) + pending = None if depth is None else depth + state.reserved + want = wanted_slots(cap, state.active, pending) + + if want > current: + new = want + verb = "grew" + elif want < current: + # One at a time on the way down. See SHRINK_STEP. + new = max(want, current - SHRINK_STEP) + verb = "shrank" else: - # The fixed point. A settled lane sends nothing and says so — - # the tick is one inspect and one LLEN sweep, no control messages. - out.append(AutoscaleDecision( + out.append(LaneSizing( lane.name, "held", current, - f"{backlog} waiting, {state.active} busy", + f"{pending if pending is not None else '?'} waiting, " + f"{state.active} busy, cap {cap}", )) + continue + + ok, err = set_lane_slots_sync(lane, new, live=state) + if not ok: + out.append(LaneSizing( + lane.name, "held", current, f"could not resize: {err}", + )) + continue + out.append(LaneSizing( + lane.name, verb, new, + f"{pending if pending is not None else '?'} waiting, " + f"{state.active} busy, cap {cap}", + )) return out diff --git a/backend/app/services/worker_lanes.py b/backend/app/services/worker_lanes.py index c36ec52..7694696 100644 --- a/backend/app/services/worker_lanes.py +++ b/backend/app/services/worker_lanes.py @@ -129,18 +129,15 @@ class Lane: # with CELERY_QUEUES=maintenance_long). Recorded here so the generated # supervisord config and the compose file cannot disagree about it. entrypoint_role: str - default_slots: int - # The cap a lane STARTS with, which is not the ceiling. Set low enough - # that raising slots within it is an ordinary adjustment, and raising the - # cap itself is a deliberate act — a cap that begins at the ceiling is a - # rubber stamp and protects nobody. + # THE cap a lane starts with — and, since 2026-09-23, the only number an + # operator sets for it. How many workers actually run is the autoscaler's + # job; this is the most it may use. Zero means the lane is off. + # + # One, and zero for ML. Deliberately far below the operator's own + # production numbers, which are tuned for their hardware and are not a + # sane first boot for a stranger — and low enough that a busy instance + # tells them to raise it rather than quietly consuming the machine. default_slots_cap: int - default_enabled: bool - # Whether a lane arrives with the autoscaler on. False for every lane, and - # the field exists anyway: this module is the one place that describes a - # lane, and leaving one operator-settable default to the column's DDL - # default would make it the only setting you cannot read here. - default_autoscale: bool = False # True when a slot costs a copy of the ML model rather than just a process. # The only lane whose ceiling is decided by memory instead of by cores. memory_bound: bool = False @@ -159,51 +156,58 @@ class Lane: return tuple(sorted(self.queues)) -# Defaults are ONE OF EACH, with ML off — operator, 2026-09-22: *"that -# starting value should be one of each."* Deliberately far below the -# operator's own production numbers (worker 8, ml 2), which are tuned for -# their hardware and are not a sane first boot for a stranger. +# ONE CAP PER LANE, and that is the whole of what an operator sets. # -# ML ships disabled because enabling it is what triggers the SigLIP download -# (milestone 422 step 6) — rule 164 allows a feature that needs a fetch only -# when it is "optional and clearly off", and off-by-default is also what keeps -# a small box from loading a multi-GB model it was never asked to load. +# Operator, 2026-09-23: *"auto should be always on, not a setting, so that +# idle instances quiet down when not running. the number that is visible and +# something the user can tweak and manage should be the cap itself the number +# of running workers is handled by the autoscaling function which is always +# on."* +# +# Until then a lane had THREE operator values — `slots`, `slots_cap` and +# `autoscale` — because the manual dial was built first (steps 2-4) and the +# autoscaler arrived last (step 7) as an opt-in beside a control that already +# existed. Nothing ever asked whether the dial should still exist once +# something could move it automatically. It should not: "how many are running +# right now" is a measurement, not a preference. +# +# One of each, and ML at zero. ML at zero is also rule 164's carve-out: a cap +# of zero means no consumers, so a fresh install never loads a model or +# reaches HuggingFace, and raising the cap is what triggers the fetch. +# +# These are far below the operator's own production numbers, and deliberately +# so — they are what a stranger's first boot should do, not what a tuned +# machine can. The UI is what closes that gap: a lane sitting at its cap with +# a backlog says so, and says raising the cap is the fix. Without that a +# conservative default is just a slow instance nobody knows how to speed up. LANES: tuple[Lane, ...] = ( Lane( name="worker", display_name="Worker", queues=("default", "import", "thumbnail", "download"), entrypoint_role="worker", - default_slots=1, - default_slots_cap=4, - default_enabled=True, + default_slots_cap=1, ), Lane( name="scheduler", display_name="Scheduler", queues=("maintenance", "scan"), entrypoint_role="scheduler", - default_slots=1, - default_slots_cap=2, - default_enabled=True, + default_slots_cap=1, ), Lane( name="maintenance_long", display_name="Long maintenance", queues=("maintenance_long",), entrypoint_role="worker", - default_slots=1, - default_slots_cap=2, - default_enabled=True, + default_slots_cap=1, ), Lane( name="ml", display_name="ML tagging", queues=("ml",), entrypoint_role="ml-worker", - default_slots=0, - default_slots_cap=1, - default_enabled=False, + default_slots_cap=0, memory_bound=True, models=(SIGLIP_MODEL,), optional=True, @@ -242,6 +246,22 @@ ML_BYTES_PER_SLOT = SIGLIP_MODEL.approx_resident_bytes # the processes an OOM kill must not take (see the module docstring). RESERVED_BYTES = 2 * GIB +# The smallest pool a lane can actually run: ONE process, never zero. +# +# billiard refuses to remove the last worker in a pool, so a lane asked to +# shrink to nothing gets `ValueError("Can't shrink pool. All processes +# busy!")` and the sizing pass re-sends the doomed message forever. Found on +# the operator's live deploy, 2026-09-23. +# +# It is also what makes "off" expressible: a lane at cap 0 keeps this one +# parked process with its consumers cancelled, so it still answers `inspect` +# (and so reads as present rather than crashed), and `add_consumer` has +# something to reach when the cap goes back up. +# +# Lives HERE rather than in `worker_control` because `gen_supervisord` needs +# it at container boot and must not import the models package to get it. +MIN_POOL_SLOTS = 1 + # The floor a cores-derived ceiling never goes below. A single-core box still # needs to be able to run its lanes; the ceiling exists to stop absurd values, # not to make a small machine unusable. diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index b13ac68..bfcaee3 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1349,107 +1349,55 @@ def sync_memberships() -> str: parts.append(f"suggested={res['suggested']}") return " ".join(parts) or "no platforms" +@celery.task(name="backend.app.tasks.maintenance.size_worker_lanes") +def size_worker_lanes() -> dict: + """Size every lane to its backlog, within the cap the operator set. -@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. + ONE sweep, replacing `reconcile_worker_lanes` and `autoscale_worker_lanes` + (2026-09-23). They were two enforcers over one number: the reconcile drove + the pool to a stored `slots`, the autoscaler moved it away from that same + value, and most of the autoscaler's design existed to keep the reconcile + from undoing its work. Deleting the stored number deletes the conflict. - 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. + It still does what the reconcile existed for. `pool_grow` is not durable — + a worker restarted by its supervisor comes back at its ENV concurrency, + silently below what the lane should run — and this reads the LIVE pool + every minute, so that worker is corrected on the next tick rather than + after five. - ## 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. + Returns every lane's outcome INCLUDING the ones it held, each with a + reason. A pass that only speaks when it acts cannot be debugged on the day + it does not. """ from ..models import WorkerLane - from ..services.worker_control import reconcile_lanes_sync + from ..services.worker_control import size_lanes_sync - # Both dicts built INSIDE the session. Reading a column off a detached - # instance happens to work while the attribute is still loaded and stops - # working the moment anything expires it — a failure that would appear - # long after this line, in a sweep nobody is watching. + # Read INSIDE the session. Reading a column off a detached instance + # happens to work while the attribute is still loaded and stops working + # the moment anything expires it — a failure that would appear long after + # this line, in a sweep nobody is watching. with _sync_session_factory()() as session: - rows = list(session.execute(select(WorkerLane)).scalars()) - desired = {row.name: (row.slots, row.enabled) for row in rows} - # Lanes the autoscaler may move. For these the stored value is a - # FLOOR: this sweep restores a lane that fell below it and never takes - # back what the autoscaler added, or the two would fight every five - # minutes. - autoscaling = frozenset(row.name for row in rows if row.autoscale) - 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, autoscaling) - - -@celery.task(name="backend.app.tasks.maintenance.autoscale_worker_lanes") -def autoscale_worker_lanes() -> dict: - """Grow a saturated lane, within the cap the operator set. - - Milestone 422 step 7, and the only sweep in this milestone that decides - rather than obeys. Everything else applies what someone pressed. - - OFF unless a lane opts in. A no-op costs one `celery inspect` and one LLEN - sweep and sends no control messages — the same fixed point the reconcile - holds, and for the same reason: this runs forever, so a settled system has - to be silent or the real signal drowns in its own heartbeat. - - Returns every lane's decision INCLUDING the ones it held, each with a - reason. An autoscaler that only speaks when it acts cannot be debugged on - the day it does not. - """ - from ..models import WorkerLane - from ..services.worker_control import autoscale_lanes_sync - - with _sync_session_factory()() as session: - lanes = { - # The stored `slots` is passed ONLY as the floor. What the - # autoscaler moves is the live pool, which it reads itself — this - # row is never written, so using it as the current value would pin - # every decision to the same number forever. - row.name: (row.slots_cap, row.slots, True) + caps = { + row.name: row.slots_cap for row in session.execute(select(WorkerLane)).scalars() - if row.autoscale } - if not lanes: - return {"decisions": []} + if not caps: + # Migration 0103/0105 seed these, so an empty table means they have + # not run yet. Nothing to assert — and inventing defaults here would + # let this task disagree with the seed it is meant to be enforcing. + return {"sized": []} - decisions = autoscale_lanes_sync(lanes) - for d in decisions: - if d.action != "held": + sized = size_lanes_sync(caps) + for d in sized: + if d.action not in ("held", "skipped"): log.info( - "autoscale: %s %s to %s slots — %s", + "worker lanes: %s %s to %s slots — %s", d.lane, d.action, d.slots, d.reason, ) return { - "decisions": [ + "sized": [ {"lane": d.lane, "action": d.action, "slots": d.slots, "reason": d.reason} - for d in decisions + for d in sized ], } diff --git a/frontend/src/components/settings/SystemHealthTab.vue b/frontend/src/components/settings/SystemHealthTab.vue index d5e4364..ae24f91 100644 --- a/frontend/src/components/settings/SystemHealthTab.vue +++ b/frontend/src/components/settings/SystemHealthTab.vue @@ -55,10 +55,9 @@ Part Queues - Pending - Busy - Slots - Auto + Waiting + Workers + Max workers @@ -85,7 +84,15 @@ memory to change nothing. Seeing a lane wedged on one slow job is the useful half. -->
- all slots busy for {{ row.stuckFor }} + all workers busy for {{ row.stuckFor }} +
+ +
+ mdi-arrow-up-bold-outline + {{ row.advice }}
@@ -98,6 +105,9 @@ {{ row.lane ? (row.lane.pending ?? '—') : '—' }} + — - - - - — - - +

- Slots is how many tasks a lane runs at once. Changes - reach the running worker immediately and survive a restart. - Zero slots turns the lane off — it keeps its process - and stops taking work, so it stays listed here rather than looking like - a crash. Three of the four lanes need to be running for FabledCurator to - work at all; ML tagging is the one that is genuinely optional. The - of N beneath each dial is the most that lane may have on this - machine. + Max workers is the only thing you set. How many + workers a lane is actually running at any moment is decided for you — + it grows to meet a backlog and gives the workers back when the queue + empties, so an idle instance settles down to one of each without being + told to. The cap is the ceiling on that, never a target, so raising it + costs nothing until there is work to spend it on.

- Auto lets a lane add slots by itself when its - queue is backed up and every slot it has is busy — one at a - time, never past of N — and hand them back once the backlog - clears. Off means the lane stays at exactly the number you set. It is - off by default, per lane, because this is the only thing on this page - that acts without being asked. A lane already dialled to its maximum has - nowhere to grow, so its switch stays unavailable until you leave it some - room. + A cap of zero turns the lane off. It keeps one parked + process and stops taking work, so it stays listed here rather than + looking like a crash. Three of the four lanes need to be running for + FabledCurator to work at all; ML tagging is the one that is genuinely + optional, and it ships at zero because switching it on downloads a + model. Up to N beneath each dial is what this machine can + hold — memory for ML tagging, processor cores for the rest — and it is + recalculated from the container's real limits every time this page + loads. +

+

+ The shipped caps are one of each, which is right for a first boot and + wrong for a busy library. A lane that is running everything its cap + allows while work piles up will say so in its row, and + raising that cap is the answer when it does.

A part is called stale after @@ -204,7 +208,7 @@ you are not running the GPU agent, which does the same work faster.

-

Giving it a slot downloads, once:

+

Raising its cap above zero downloads, once:

  • {{ m.repo }} — @@ -214,8 +218,8 @@

- Each slot loads its own copy, which is why this machine allows it at - most {{ lane.ceiling }}. + Each worker loads its own copy, which is why this machine allows it + at most {{ lane.ceiling }}. @@ -230,7 +234,7 @@ import { computed, onMounted, onUnmounted, ref } from 'vue' import { laneStuckFor, useSystemActivityStore } from '../../stores/systemActivity.js' import { useSystemHealthStore } from '../../stores/systemHealth.js' import { formatRelative } from '../../utils/date.js' -import { mergeParts } from '../../utils/systemParts.js' +import { laneAdvice, mergeParts } from '../../utils/systemParts.js' const store = useSystemHealthStore() const lanesStore = useSystemActivityStore() @@ -266,11 +270,11 @@ onUnmounted(() => { clearInterval(pollId) }) // that removed it. const rows = computed(() => mergeParts( store.parts, lanesStore.lanes?.lanes ?? [], laneStuckFor, -)) +).map((row) => ({ ...row, advice: laneAdvice(row.lane) }))) const offOptionalLanes = computed(() => (lanesStore.lanes?.lanes ?? []).filter( - (l) => l.optional && l.slots === 0 && l.models?.length, + (l) => l.optional && l.slots_cap === 0 && l.models?.length, ), ) @@ -319,23 +323,12 @@ async function apply(lane, fields) { } } -// The dial IS the switch — the API derives `enabled` from the number, so -// stepping to zero turns the lane off and stepping off zero turns it on. -// Nothing here sends `enabled`, and there is no second control that could -// disagree with the number on screen. +// The cap is the only thing an operator sets, and it doubles as the switch: +// zero means no consumers. How many workers actually run is the sizing pass's +// business — always on, reading the live pool every minute — so nothing here +// sends a worker count. function step(lane, delta) { - return apply(lane, { slots: lane.slots + delta }) -} - -// Room to grow into. The autoscaler moves the LIVE pool, but the floor it -// starts from is the stored value, so a lane already dialled to its cap has -// nowhere to go and turning this on would do nothing at all. -function canGrow(lane) { - return lane.slots_cap > lane.slots -} - -function setAutoscale(lane, value) { - return apply(lane, { autoscale: Boolean(value) }) + return apply(lane, { slots_cap: lane.slots_cap + delta }) } @@ -390,6 +383,10 @@ function setAutoscale(lane, value) { font-size: 0.8rem; color: rgb(var(--v-theme-on-surface) / 0.72); padding-left: 17px; } +.fc-parts__advice { + font-size: 0.8rem; padding-left: 17px; margin-top: 2px; + color: rgb(var(--v-theme-accent)); +} .fc-parts__queues { font-size: 0.78rem; color: rgb(var(--v-theme-on-surface) / 0.6); } diff --git a/frontend/src/utils/systemParts.js b/frontend/src/utils/systemParts.js index e222edf..7eb6510 100644 --- a/frontend/src/utils/systemParts.js +++ b/frontend/src/utils/systemParts.js @@ -54,10 +54,10 @@ export function mergeParts(parts, lanes, stuckFor = () => null) { name: part.name, kindLabel: lane?.optional ? 'optional lane' : kindLabel(part.kind), state: part.state, - // A lane dialled to zero is OFF, not broken. Say so, rather than let the + // A lane capped at zero is OFF, not broken. Say so, rather than let the // roster's heartbeat sentence report the operator's own choice as a // fault — the roster cannot know the difference, and the lane can. - detail: lane && lane.slots === 0 ? 'off — no slots' : part.detail, + detail: lane && lane.slots_cap === 0 ? 'off — cap is zero' : part.detail, queues: (part.queues || []).join(', '), lane, stuckFor: lane ? stuckFor(lane) : null, @@ -80,7 +80,7 @@ export function mergeParts(parts, lanes, stuckFor = () => null) { } function laneRow(lane, stuckFor) { - const on = lane.slots > 0 + const on = lane.slots_cap > 0 let state = 'unknown' if (lane.live?.present) state = on ? (stuckFor(lane) ? 'stale' : 'ok') : 'unknown' else if (on) state = 'down' @@ -90,7 +90,7 @@ function laneRow(lane, stuckFor) { kindLabel: lane.optional ? 'optional lane' : 'worker lane', state, detail: lane.live?.present - ? (on ? 'running' : 'off — no slots') + ? (on ? 'running' : 'off — cap is zero') : 'has not checked in yet', queues: (lane.queues || []).join(', '), lane, @@ -98,3 +98,47 @@ function laneRow(lane, stuckFor) { severity: SEVERITY[state], } } + + +// How much has to be waiting before we tell someone to raise a cap. +// +// Not "anything at all". A lane at its cap with three items queued is working +// normally and will be empty in a moment; a notice there is one people learn +// to scroll past, and at that point it is worse than not having it. +export const ADVISE_BACKLOG = 10 + +/** + * The sentence that tells an operator the cap is now the limiting factor. + * + * Operator, 2026-09-23: *"there needs to be something that tells you user to + * bump those numbers to improve processing rate or they'd never know the + * controls exist."* The defaults are deliberately one-of-each, so on a busy + * instance the shipped configuration IS the bottleneck — and a conservative + * default nobody knows how to raise is just a slow product. + * + * Only fires when raising the cap would actually help: there is real work + * waiting, every worker the cap allows is already running, and the cap is + * below what this machine can hold. A lane already at its ceiling gets + * nothing, because there is nothing it could be told to do. + */ +export function laneAdvice(lane) { + if (!lane) return null + const pending = lane.pending + if (pending == null || pending < ADVISE_BACKLOG) return null + + if (lane.slots_cap === 0) { + return `${pending.toLocaleString()} waiting, and this lane is off. ` + + 'Raise its cap to start working through them.' + } + if (lane.ceiling <= lane.slots_cap) { + // At the machine's limit, not the operator's. Saying "raise the cap" + // here would be advice they cannot take. + return null + } + if (!lane.live?.present || (lane.live.pool ?? 0) < lane.slots_cap) return null + + return `${pending.toLocaleString()} waiting and all ${lane.slots_cap} ` + + `worker${lane.slots_cap === 1 ? '' : 's'} busy. ` + + `Raise the cap to run more at once — this machine allows up to ` + + `${lane.ceiling}.` +} diff --git a/frontend/test/systemParts.spec.js b/frontend/test/systemParts.spec.js index 894b09d..7879b01 100644 --- a/frontend/test/systemParts.spec.js +++ b/frontend/test/systemParts.spec.js @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' -import { mergeParts, queueKey } from '../src/utils/systemParts.js' +import { laneAdvice, mergeParts, queueKey } from '../src/utils/systemParts.js' // The System tab's one table (operator 2026-09-23: "combine the two sections // into a single table"). What is pinned here is the JOIN, because its failure @@ -25,9 +25,8 @@ const LANE = { // sorted. If the join ever compares these two lists directly rather than as // sets, this fixture is what catches it. queues: ['default', 'import', 'thumbnail', 'download'], - slots: 2, slots_cap: 4, - autoscale: false, + ceiling: 8, live: { present: true, replicas: 1, pool: 2, active: 0 }, pending: 0, } @@ -71,7 +70,7 @@ describe('mergeParts', () => { // is exactly the one with no roster entry. const ml = { ...LANE, name: 'ml', display_name: 'ML tagging', queues: ['ml'], - slots: 0, optional: true, + slots_cap: 0, optional: true, live: { present: false, replicas: 0, pool: null, active: 0 }, } const rows = mergeParts([POSTGRES], [ml]) @@ -81,15 +80,15 @@ describe('mergeParts', () => { expect(row.kindLabel).toBe('optional lane') }) - it('does not call a lane at zero slots broken', () => { + it('does not call a lane capped at zero broken', () => { // The roster only knows a heartbeat age, so it goes on saying "is running" // for a lane the operator deliberately dialled to nothing. The lane knows // the difference; reporting the operator's own choice as a fault is how an // indicator stops being read. - const off = { ...LANE, slots: 0 } + const off = { ...LANE, slots_cap: 0 } const row = mergeParts([PART], [off])[0] - expect(row.detail).toBe('off — no slots') + expect(row.detail).toBe('off — cap is zero') }) it('puts the broken thing first, whatever it is', () => { @@ -126,3 +125,68 @@ describe('mergeParts', () => { expect(mergeParts(undefined, undefined)).toEqual([]) }) }) + + +// The nudge. Operator, 2026-09-23: *"there needs to be something that tells +// you user to bump those numbers to improve processing rate or they'd never +// know the controls exist."* +// +// The caps ship at one of each, so on a busy instance the SHIPPED +// CONFIGURATION is the bottleneck. What is pinned here is that it fires only +// when raising the cap would actually help — a notice on a lane that is +// keeping up, or one already at the machine's limit, is a notice people learn +// to scroll past, and at that point it is worse than not having it. + +describe('laneAdvice', () => { + const busyAtCap = { + slots_cap: 1, ceiling: 7, pending: 4060, + live: { present: true, pool: 1, active: 1 }, + } + + it('speaks when the cap is the limiting factor', () => { + const advice = laneAdvice(busyAtCap) + expect(advice).toContain('4,060 waiting') + expect(advice).toContain('Raise the cap') + // The headroom, so it is an instruction rather than a complaint. + expect(advice).toContain('7') + }) + + it('says something different about a lane that is switched off', () => { + const advice = laneAdvice({ ...busyAtCap, slots_cap: 0 }) + expect(advice).toContain('this lane is off') + }) + + it('stays quiet when the lane is keeping up', () => { + expect(laneAdvice({ ...busyAtCap, pending: 2 })).toBeNull() + }) + + it('stays quiet when the cap is not the thing holding it back', () => { + // Four allowed, two running — the sizing pass has room it has not taken, + // so the queue is not the cap's fault. + expect(laneAdvice({ + ...busyAtCap, slots_cap: 4, live: { present: true, pool: 2, active: 2 }, + })).toBeNull() + }) + + it('stays quiet at the machine ceiling, where the advice is untakeable', () => { + // The one case where saying "raise the cap" would send someone to a + // control that will refuse them. + expect(laneAdvice({ ...busyAtCap, ceiling: 1 })).toBeNull() + }) + + it('stays quiet about a lane that is not answering', () => { + // An unswept read is not a verdict: nothing replied, so "all workers + // busy" is a claim nobody made. + expect(laneAdvice({ + ...busyAtCap, live: { present: false, pool: null, active: 0 }, + })).toBeNull() + }) + + it('stays quiet when the backlog is unknown rather than reading it as huge', () => { + expect(laneAdvice({ ...busyAtCap, pending: null })).toBeNull() + }) + + it('says nothing about a row that is not a lane at all', () => { + expect(laneAdvice(undefined)).toBeNull() + }) +}) diff --git a/frontend/test/workerLanes.spec.js b/frontend/test/workerLanes.spec.js index f553e57..b594853 100644 --- a/frontend/test/workerLanes.spec.js +++ b/frontend/test/workerLanes.spec.js @@ -28,13 +28,13 @@ const LANES_BODY = { { name: 'worker', display_name: 'Worker', queues: ['default', 'import', 'thumbnail', 'download'], - slots: 1, slots_cap: 4, ceiling: 8, enabled: true, memory_bound: false, + slots_cap: 1, ceiling: 8, enabled: true, memory_bound: false, live: { present: true, replicas: 1, pool: 1, active: 0, reserved: 3 }, queue_depth: 5, pending: 8, }, { name: 'ml', display_name: 'ML tagging', queues: ['ml'], - slots: 0, slots_cap: 1, ceiling: 2, enabled: false, memory_bound: true, + slots_cap: 0, ceiling: 2, enabled: false, memory_bound: true, live: { present: false, replicas: 0, pool: null, active: 0, reserved: 0 }, queue_depth: null, pending: null, }, @@ -64,23 +64,22 @@ describe('worker lanes store', () => { }) it('setLane posts only the fields it was given', async () => { - // Partial update: the stepper sends slots without restating a cap it did - // not touch. Sending the whole row back would make two operators editing - // different fields clobber each other. + // One field: the cap. How many workers are running is a measurement the + // sizing pass owns, so there is nothing else for the UI to send. const calls = [] stubFetch((url, init) => { calls.push({ url, init }) if (init?.method === 'POST') { - return { status: 200, body: { name: 'worker', slots: 2, applied: true } } + return { status: 200, body: { name: 'worker', slots_cap: 2, applied: true } } } return { status: 200, body: LANES_BODY } }) const s = useSystemActivityStore() - await s.setLane('worker', { slots: 2 }) + await s.setLane('worker', { slots_cap: 2 }) const post = calls.find((c) => c.init?.method === 'POST') expect(post.url).toContain('/api/system/workers/worker') - expect(JSON.parse(post.init.body)).toEqual({ slots: 2 }) + expect(JSON.parse(post.init.body)).toEqual({ slots_cap: 2 }) }) it('setLane refetches so the card shows the server truth, not the guess', async () => { @@ -94,7 +93,7 @@ describe('worker lanes store', () => { return { status: 200, body: LANES_BODY } }) const s = useSystemActivityStore() - await s.setLane('worker', { slots: 2 }) + await s.setLane('worker', { slots_cap: 2 }) expect(gets).toBe(1) }) @@ -106,13 +105,13 @@ describe('worker lanes store', () => { if (init?.method === 'POST') { return { status: 200, - body: { applied: false, apply_error: 'lane is not running', slots: 2 }, + body: { applied: false, apply_error: 'lane is not running', slots_cap: 2 }, } } return { status: 200, body: LANES_BODY } }) const s = useSystemActivityStore() - const reply = await s.setLane('worker', { slots: 2 }) + const reply = await s.setLane('worker', { slots_cap: 2 }) expect(reply.applied).toBe(false) expect(reply.apply_error).toContain('not running') }) @@ -182,7 +181,7 @@ describe('laneStuckFor', () => { }) it('says nothing when the pool is zero', () => { - // A lane sized to zero is not "fully busy at zero" — it is off. + // A lane with no workers is not "fully busy at zero" — it is off. expect(laneStuckFor(lane({ live: { present: true, pool: 0, active: 0 } }))) .toBeNull() }) diff --git a/tests/test_api_workers.py b/tests/test_api_workers.py index d36e1fd..d39f4fe 100644 --- a/tests/test_api_workers.py +++ b/tests/test_api_workers.py @@ -47,237 +47,186 @@ async def test_get_lists_every_lane_with_its_ceiling(client, no_live_workers): assert set(by_name) == {"worker", "scheduler", "maintenance_long", "ml"} for lane in body["lanes"]: assert lane["ceiling"] >= 0 - assert lane["slots"] <= lane["slots_cap"] + assert lane["slots_cap"] <= lane["ceiling"] # Nothing is running, so live state must say so rather than report # zeroes that read like a healthy idle lane. assert lane["live"]["present"] is False +@pytest.mark.asyncio +async def test_enabled_is_derived_from_the_cap_and_never_stored( + client, no_live_workers, +): + """The reshape of 2026-09-23. "Off" and "may use no workers" were two + spellings of one fact, stored separately and free to disagree.""" + body = await (await client.get("/api/system/workers")).get_json() + for lane in body["lanes"]: + assert lane["enabled"] == (lane["slots_cap"] > 0), lane["name"] + + @pytest.mark.asyncio async def test_ml_ships_off(client, no_live_workers): - """Rule 164's carve-out and the weak-hardware default in one row: enabling - the lane is what triggers the SigLIP download, so a fresh install must not - find it on.""" + """Rule 164's carve-out and the weak-hardware default in one row: raising + the cap is what triggers the SigLIP download, so a fresh install must not + find it above zero.""" body = await (await client.get("/api/system/workers")).get_json() ml = next(lane for lane in body["lanes"] if lane["name"] == "ml") + assert ml["slots_cap"] == 0 assert ml["enabled"] is False - assert ml["slots"] == 0 + + +@pytest.mark.asyncio +async def test_the_other_lanes_ship_at_one(client, no_live_workers): + """Operator: *"the cap defaults should be 1 and 0 for the ml-worker."*""" + body = await (await client.get("/api/system/workers")).get_json() + caps = {lane["name"]: lane["slots_cap"] for lane in body["lanes"]} + assert caps == { + "worker": 1, "scheduler": 1, "maintenance_long": 1, "ml": 0, + } # --- the persist / push split ------------------------------------------------ @pytest.mark.asyncio -async def test_a_value_is_stored_even_when_it_cannot_be_pushed( - client, db, no_live_workers +async def test_a_cap_is_stored_even_when_it_cannot_be_pushed( + client, db, no_live_workers, ): - """THE test for this step. `pool_grow` is not durable and the lane is not - answering, so the push fails — and the setting must survive anyway, or a - UI that saved while a worker was restarting would silently lose it - (lesson #4202). 200 with `applied: false`, not an error. - """ - resp = await client.post("/api/system/workers/worker", json={"slots": 3}) + """Nothing is answering, so the live push fails. That is NOT a failed + setting: the value is saved and the sizing pass carries it within a + minute (lesson #4202 — a live change that does not survive, with nothing + saying so).""" + resp = await client.post("/api/system/workers/worker", json={"slots_cap": 3}) + assert resp.status_code == 200 body = await resp.get_json() - - assert body["slots"] == 3 + assert body["slots_cap"] == 3 assert body["applied"] is False assert "not running" in body["apply_error"] - assert (await _lane_row(db, "worker")).slots == 3 + assert (await _lane_row(db, "worker")).slots_cap == 3 + + +# --- the cap is the switch --------------------------------------------------- @pytest.mark.asyncio -async def test_a_partial_update_leaves_the_other_fields_alone( - client, db, no_live_workers +async def test_a_cap_of_zero_turns_the_lane_off(client, db, no_live_workers): + await client.post("/api/system/workers/worker", json={"slots_cap": 0}) + + row = await _lane_row(db, "worker") + assert row.slots_cap == 0 + body = await (await client.get("/api/system/workers")).get_json() + worker = next(lane for lane in body["lanes"] if lane["name"] == "worker") + assert worker["enabled"] is False + + +@pytest.mark.asyncio +async def test_raising_it_off_zero_turns_the_lane_on(client, db, no_live_workers): + await client.post("/api/system/workers/ml", json={"slots_cap": 1}) + + assert (await _lane_row(db, "ml")).slots_cap == 1 + body = await (await client.get("/api/system/workers")).get_json() + ml = next(lane for lane in body["lanes"] if lane["name"] == "ml") + assert ml["enabled"] is True + + +@pytest.mark.asyncio +async def test_the_model_fetch_fires_on_the_transition_not_on_every_write( + client, db, no_live_workers, monkeypatch, ): - """The stepper sends `{"slots": n}` without restating a cap it did not - touch.""" - before = await _lane_row(db, "worker") - original_cap = before.slots_cap + """Raising the cap off zero downloads SigLIP, once. A second nudge of the + same dial must not re-enqueue a multi-GB download — and the trigger must + be the TRANSITION rather than "a field was sent", which is what it tested + before the UI stopped sending `enabled` at all.""" + monkeypatch.setattr( + wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None), + ) + monkeypatch.setattr( + wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None), + ) + fired = [] + monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True) - await client.post("/api/system/workers/worker", json={"slots": 2}) + first = await (await client.post( + "/api/system/workers/ml", json={"slots_cap": 1}, + )).get_json() + second = await (await client.post( + "/api/system/workers/ml", json={"slots_cap": 2}, + )).get_json() - await db.refresh(before) - assert before.slots == 2 - assert before.slots_cap == original_cap + assert first["fetching_models"] is True + assert second["fetching_models"] is False + assert fired == [1] # --- what is refused --------------------------------------------------------- -@pytest.mark.asyncio -async def test_slots_above_the_cap_are_refused_and_nothing_is_stored( - client, db, no_live_workers -): - row = await _lane_row(db, "worker") - resp = await client.post( - "/api/system/workers/worker", json={"slots": row.slots_cap + 1}, - ) - assert resp.status_code == 400 - body = await resp.get_json() - assert body["error"] == "refused" - # The sentence the UI shows. A greyed control with no reason reads as a bug. - assert "cap" in body["detail"] - - await db.refresh(row) - assert row.slots <= row.slots_cap - - @pytest.mark.asyncio async def test_a_cap_above_the_derived_ceiling_is_refused( - client, db, no_live_workers + client, db, no_live_workers, ): - """The operator's cap is theirs to set, but not past what the container - can hold — the whole point of deriving a ceiling rather than typing one.""" + """The ceiling is the machine's, not the operator's, and it is the one + bound they cannot lower themselves past. The detail is written to be read + by a person — a refused control with no reason reads as a bug.""" + before = (await _lane_row(db, "ml")).slots_cap + resp = await client.post( "/api/system/workers/ml", json={"slots_cap": 10_000}, ) + assert resp.status_code == 400 body = await resp.get_json() - assert body["error"] == "refused" assert "container can hold" in body["detail"] + await _refreshed(db, "ml", before) @pytest.mark.asyncio -async def test_a_boolean_is_not_accepted_as_a_slot_count(client, no_live_workers): - """`True` is an int in Python. Coerced, it would silently set one slot — - a control that appears to work and does something nobody asked for.""" - resp = await client.post("/api/system/workers/worker", json={"slots": True}) +async def test_a_negative_cap_is_refused(client, db, no_live_workers): + resp = await client.post("/api/system/workers/worker", json={"slots_cap": -1}) assert resp.status_code == 400 - body = await resp.get_json() - assert body["error"] == "invalid_body" + + +@pytest.mark.asyncio +async def test_a_boolean_is_not_accepted_as_a_cap(client, no_live_workers): + """`True` is an int in Python. Reading it as a cap of 1 would be a control + that appears to work and sets something nobody asked for.""" + resp = await client.post("/api/system/workers/worker", json={"slots_cap": True}) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_the_retired_fields_are_no_longer_accepted(client, no_live_workers): + """`slots`, `enabled` and `autoscale` are gone. A client still sending one + must be told, not silently ignored — a POST that returns 200 having + changed nothing is the worst of the three outcomes.""" + for field in ("slots", "enabled", "autoscale"): + resp = await client.post( + "/api/system/workers/worker", json={field: 2}, + ) + assert resp.status_code == 400, field @pytest.mark.asyncio async def test_an_unknown_lane_is_refused_and_names_the_known_ones( - client, no_live_workers + client, no_live_workers, ): - resp = await client.post("/api/system/workers/nonsense", json={"slots": 1}) + resp = await client.post("/api/system/workers/nope", json={"slots_cap": 1}) assert resp.status_code == 400 body = await resp.get_json() - assert body["error"] == "unknown_lane" assert "worker" in body["known"] @pytest.mark.asyncio async def test_an_empty_body_is_refused_rather_than_treated_as_a_no_op( - client, no_live_workers + client, no_live_workers, ): - """A POST that changes nothing and returns 200 is indistinguishable from - one that worked, which is how a broken UI control goes unnoticed.""" resp = await client.post("/api/system/workers/worker", json={}) assert resp.status_code == 400 - body = await resp.get_json() - assert body["error"] == "invalid_body" -# --- the dial is the switch -------------------------------------------------- -# -# Operator, 2026-09-23: *"almost all of it always needs to run there's only one -# optional piece and it is killed by moving the 'cap' to zero."* So `enabled` -# is derived from the number rather than being a second control the operator -# has to keep in agreement with it. It stays on the API — these assert that it -# still does, because it is the mechanism the reconcile and the healthcheck -# read. - - -@pytest.mark.asyncio -async def test_dialling_a_lane_to_zero_turns_it_off(client, db, no_live_workers): - await client.post("/api/system/workers/worker", json={"slots": 0}) - - row = await _lane_row(db, "worker") - assert row.slots == 0 - assert row.enabled is False - - -@pytest.mark.asyncio -async def test_dialling_it_back_up_turns_it_on(client, db, no_live_workers): - await client.post("/api/system/workers/ml", json={"slots": 1}) - - row = await _lane_row(db, "ml") - assert row.slots == 1 - assert row.enabled is True, "the lane the operator just asked for work from" - - -@pytest.mark.asyncio -async def test_an_explicit_enabled_still_wins(client, db, no_live_workers): - """The field is not removed, only derived when absent. Something that - genuinely wants a lane holding its process with consumers cancelled — a - drain before a restart — must still be able to say so without having to - destroy the operator's slot count to express it.""" - await client.post( - "/api/system/workers/worker", json={"slots": 3, "enabled": False}, - ) - - row = await _lane_row(db, "worker") - assert (row.slots, row.enabled) == (3, False) - - -@pytest.mark.asyncio -async def test_a_cap_only_write_does_not_decide_the_switch( - client, db, no_live_workers, -): - """Only the SLOTS dial derives it. A cap is a ceiling, not a request for - work, and letting it flip the lane would make raising a ceiling start - something.""" - before = await _lane_row(db, "ml") - assert before.enabled is False - - await client.post("/api/system/workers/ml", json={"slots_cap": 1}) - - await db.refresh(before) - assert (before.slots, before.enabled) == (0, False) - - -@pytest.mark.asyncio -async def test_the_model_fetch_fires_on_the_transition_not_on_the_field( - client, db, no_live_workers, monkeypatch, -): - """The trap the derivation set, caught here rather than in production. - - The fetch used to be conditioned on `enabled is True` — the FIELD having - been sent. The UI no longer sends it at all, so the download that makes - the ML lane usable would simply never have fired, and the lane would have - come on and sat there consuming a queue it had no model for. - """ - fired = [] - monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True) - monkeypatch.setattr( - wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None), - ) - monkeypatch.setattr( - wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None), - ) - - body = await (await client.post( - "/api/system/workers/ml", json={"slots": 1}, - )).get_json() - - assert body["fetching_models"] is True - assert fired == [1] - - -@pytest.mark.asyncio -async def test_it_does_not_fire_again_on_a_lane_already_running( - client, db, no_live_workers, monkeypatch, -): - """The other half. A second nudge of the dial on a lane that is already on - must not re-enqueue a multi-GB download.""" - monkeypatch.setattr( - wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None), - ) - monkeypatch.setattr( - wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None), - ) - await client.post("/api/system/workers/ml", json={"slots": 1}) - - fired = [] - monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True) - - body = await (await client.post( - "/api/system/workers/ml", json={"slots": 1}, - )).get_json() - - assert body["fetching_models"] is False - assert fired == [] +async def _refreshed(db, name: str, expected: int) -> None: + row = await _lane_row(db, name) + await db.refresh(row) + assert row.slots_cap == expected, "a refused write must store nothing" diff --git a/tests/test_gen_supervisord.py b/tests/test_gen_supervisord.py index f89812d..0eb1957 100644 --- a/tests/test_gen_supervisord.py +++ b/tests/test_gen_supervisord.py @@ -54,13 +54,13 @@ def test_every_lane_gets_a_program(): assert programs == expected -def test_the_ml_lane_runs_even_though_it_ships_disabled(): +def test_the_ml_lane_runs_even_though_it_ships_off(): """It holds a PROCESS and no model. `add_consumer` needs a running worker - to reach, so without this the UI switch would have nothing to switch — + to reach, so without this raising the cap would have nothing to reach — and nothing is downloaded by starting it, which is what lets rule 164 permit the fetch at all.""" assert _parse().has_section("program:ml") - assert LANES_BY_NAME["ml"].default_enabled is False + assert LANES_BY_NAME["ml"].default_slots_cap == 0 # --- the coupling this generator exists to guarantee ------------------------- @@ -152,12 +152,12 @@ def test_no_program_waits_longer_than_the_compose_stop_grace_period(): # --- what runs, and how much ------------------------------------------------ -def test_a_zero_slot_lane_still_gets_a_running_process(): - """ML ships at 0 slots and disabled — but `add_consumer` needs something to - reach. With no process there would be nothing for the UI switch to switch, - and enabling tagging could not work at all.""" +def test_a_lane_capped_at_zero_still_gets_a_running_process(): + """ML ships at cap 0 — but `add_consumer` needs something to reach. With + no process there would be nothing for the cap to switch back on, and + enabling tagging could not work at all.""" cp = _parse() - assert LANES_BY_NAME["ml"].default_slots == 0 + assert LANES_BY_NAME["ml"].default_slots_cap == 0 env = cp.get("program:ml", "environment") assert "CELERY_CONCURRENCY=1" in env @@ -275,29 +275,28 @@ def test_supervisorctl_can_reach_supervisord(): def test_each_program_starts_at_the_smallest_pool_the_control_path_allows(): """The two ends of the same floor, asserted together. - `gen_supervisord` starts every lane at `max(1, default_slots)` because - billiard will not run a pool of zero. `worker_control` has the same floor - for the opposite reason: it cannot SHRINK to zero either — + `gen_supervisord` starts every lane at one process because billiard will + not run a pool of zero. The sizing pass has the same floor for the + opposite reason: it cannot SHRINK to zero either — [ml] pidbox command error: ValueError("Can't shrink pool. All processes busy!") - Live, 2026-09-23. ML starts at one process and stores zero, so the + Live, 2026-09-23. ML started at one process and stored zero, so the reconcile tried 1 -> 0 on every tick, billiard refused, and `set_lane_slots_sync` — which returns True on SENDING the message — reported the lane changed forever (lesson #4183, on the default configuration of every install). - Two constants, in two files, that must agree or the container cannot - settle. Asserted through `effective_slots` rather than against a literal - 1, so raising the floor moves both ends at once. + One constant now, read from the same module by both, rather than two that + must agree. Asserted through it rather than against a literal 1, so + raising the floor moves both ends at once. """ - from backend.app.services.worker_control import effective_slots + from backend.app.services.worker_lanes import MIN_POOL_SLOTS cp = _parse() for lane in LANES: env = cp.get(f"program:{lane.name}", "environment") - want = effective_slots(lane.default_slots) - assert f"CELERY_CONCURRENCY={want}," in env, ( + assert f"CELERY_CONCURRENCY={MIN_POOL_SLOTS}," in env, ( f"{lane.name} starts at a size the control path cannot reach" ) diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py index e2e31f2..c98133d 100644 --- a/tests/test_worker_control.py +++ b/tests/test_worker_control.py @@ -195,388 +195,6 @@ def test_an_unknown_queue_set_maps_to_no_lane(): 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. - - Its live pool is ONE — zero slots means zero WORK, not an empty pool. - billiard will not run one, and the process is what the enable switch lands - on. - """ - control = _stub_control(monkeypatch) - _stub_live(monkeypatch, "ml", pools={"host-a": 1}, 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. - - The live pool is ONE, not zero. This fixture said zero until the floor - landed, and it was describing a container that cannot exist — billiard - will not run an empty pool, and the generator starts ml at one process for - exactly that reason. A fixture in an impossible state passes for the wrong - reason and then fails on the correct fix, which is what it did. - """ - control = _stub_control(monkeypatch) - _stub_live(monkeypatch, "ml", pools={"host-a": 1}, 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 - - -# --- the autoscaler (step 7) -------------------------------------------------- -# -# Every case below fixes the LIVE pool and the stored floor to DIFFERENT -# numbers wherever it can. That is deliberate: the first version of this -# function read the stored value as the current one, and with the two equal -# — which is what a single tick of a settled system looks like — every test -# here still passed while the autoscaler could not grow past floor+1 or shrink -# at all. Equal fixtures cannot see that bug. - - -def _autoscale_live( - monkeypatch, *, pools, active, reserved, depth, 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), - ) - 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(decisions): - return next(d for d in decisions if d.lane == "worker") - - -def test_a_prefetched_backlog_still_triggers_growth(monkeypatch): - """THE case an LLEN-only implementation misses, and the reason `reserved` - was plumbed through in step 2. Celery prefetches, so a lane can read queue - depth 0 while holding thirty tasks in worker memory — an autoscaler - watching LLEN alone sees an idle system and never grows.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 2}, active=2, reserved=30, depth=0) - - d = _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})) - - assert d.action == "grew" - assert d.slots == 3 - assert control.grew == [(1, ["host-a"])] - - -def test_it_keeps_climbing_past_the_floor_on_later_ticks(monkeypatch): - """The regression test for reading the stored value as the current one. - - The row still says 2 — the autoscaler never writes it — but the live pool - is already 5 from earlier ticks. The target must be 6. Computing from the - stored 2 would propose 3, which resizes nothing (the replica is past it), - reports success, and pins the lane one slot above its floor forever while - claiming to grow on every tick.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 5}, active=5, reserved=40, depth=0) - - d = _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})) - - assert d.slots == 6 - assert control.grew == [(1, ["host-a"])] - - -def test_backlog_with_free_slots_does_nothing(monkeypatch): - """Depth alone is not a signal — celery is about to pick those up, and - growing would add children that idle.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 8}, active=1, reserved=0, depth=50) - - assert _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})).action == "held" - assert control.grew == [] - - -def test_saturation_is_measured_against_every_replica(monkeypatch): - """`active` is summed across replicas and `pool` is one replica's size, so - comparing them calls two half-busy replicas of 4 saturated at 4 active. - Capacity is 8 here and 4 tasks are running: half the lane is idle.""" - control = _stub_control(monkeypatch) - _autoscale_live( - monkeypatch, pools={"host-a": 4, "host-b": 4}, - active=4, reserved=40, depth=0, - ) - - assert _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})).action == "held" - assert control.grew == [] - - -def test_a_saturated_lane_with_no_backlog_does_nothing(monkeypatch): - """The long-running-task case. One slow task holding every slot with an - empty queue needs no extra slots — growing does not make it finish sooner, - which is why the operator's 'runs for x time' idea became a UI warning - rather than a trigger.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 2}, active=2, reserved=0, depth=0) - - assert _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})).action == "held" - assert control.grew == [] - - -def test_growth_stops_at_the_cap_and_says_so(monkeypatch): - """The autoscaler gets no authority the operator does not already have. - And it SAYS it is capped — that is the moment they would want to know they - set one.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 4}, active=4, reserved=99, depth=0) - - d = _worker(wc.autoscale_lanes_sync({"worker": (4, 2, True)})) - - assert d.action == "held" - assert "cap is 4" in d.reason - assert control.grew == [] - - -def test_a_cleared_backlog_returns_the_lane_to_the_operator_s_value(monkeypatch): - """Down toward `configured`, one step at a time.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 5}, active=0, reserved=0, depth=0) - - d = _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})) - - assert d.action == "shrank" - assert d.slots == 4 - assert control.shrank == [(1, ["host-a"])] - - -def test_it_never_shrinks_below_what_the_operator_set(monkeypatch): - """The stored value is the operator's and the autoscaler must not eat it — - there would be nothing left to restore to.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 2}, active=0, reserved=0, depth=0) - - assert _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})).action == "held" - assert control.shrank == [] - - -def test_hysteresis_keeps_a_middling_backlog_from_flapping(monkeypatch): - """Between the two thresholds nothing happens. Equal thresholds would grow - and shrink the lane forever as one task arrives and leaves — lesson - #4183's churn arriving through a different door.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 5}, active=5, reserved=5, depth=0) - - assert _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})).action == "held" - assert control.grew == [] - assert control.shrank == [] - - -def test_a_lane_that_did_not_opt_in_is_never_touched(monkeypatch): - """Off by default, per lane. This is the only sweep that decides rather - than obeys, so it acts only where someone said it may — and it does not - even report on a lane it was not given.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 2}, active=2, reserved=99, depth=0) - - assert wc.autoscale_lanes_sync({"worker": (8, 2, False)}) == [] - assert control.grew == [] - - -def test_an_absent_lane_holds_rather_than_guessing(monkeypatch): - """`present=False` is 'nothing answered', not 'idle'. Deciding from an - unswept read is snippet #3969's shape.""" - control = _stub_control(monkeypatch) - _autoscale_live( - monkeypatch, pools={}, active=0, reserved=0, depth=0, present=False, - ) - - d = _worker(wc.autoscale_lanes_sync({"worker": (8, 2, True)})) - - assert d.action == "held" - assert "not answering" in d.reason - assert control.grew == [] - - -def test_a_settled_lane_sends_no_control_messages(monkeypatch): - """The fixed point, asserted as a property rather than inferred from the - reported action: this runs every minute forever, so a settled system has - to be silent or a real correction drowns in the heartbeat.""" - control = _stub_control(monkeypatch) - _autoscale_live(monkeypatch, pools={"host-a": 2}, active=1, reserved=0, depth=1) - - for _ in range(5): - assert _worker( - wc.autoscale_lanes_sync({"worker": (8, 2, True)}) - ).action == "held" - assert control.grew == [] - assert control.shrank == [] - - -# --- the two sweeps must not fight ------------------------------------------- - - -def test_the_reconcile_does_not_undo_what_the_autoscaler_added(monkeypatch): - """Otherwise the two would fight every five minutes: grow, revert, grow, - revert. For an autoscaling lane the stored slots are a FLOOR.""" - control = _stub_control(monkeypatch) - _stub_live(monkeypatch, "worker", pools={"host-a": 6}) - - wc.reconcile_lanes_sync({"worker": (2, True)}, frozenset({"worker"})) - - assert control.shrank == [] - - -def test_the_reconcile_still_restores_an_autoscaling_lane_that_fell_below( - monkeypatch, -): - """A floor is still a floor. A restart drops the lane to its env value and - this must bring it back to what the operator set.""" - control = _stub_control(monkeypatch) - _stub_live(monkeypatch, "worker", pools={"host-a": 1}) - - wc.reconcile_lanes_sync({"worker": (4, True)}, frozenset({"worker"})) - - assert control.grew == [(3, ["host-a"])] - - -def test_a_non_autoscaling_lane_is_still_driven_down_to_its_value(monkeypatch): - """The floor applies only to lanes that opted in — otherwise turning - autoscale off would leave the lane stuck at whatever it had grown to.""" - control = _stub_control(monkeypatch) - _stub_live(monkeypatch, "worker", pools={"host-a": 6}) - - wc.reconcile_lanes_sync({"worker": (2, True)}, frozenset()) - - assert control.shrank == [(4, ["host-a"])] - - -def test_the_autoscale_task_is_registered_and_scheduled(): - import backend.app.tasks.maintenance # noqa: F401 - from backend.app.celery_app import celery - - 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 ------------------------------------------------- @@ -680,49 +298,275 @@ def test_the_round_trip_bound_matches_the_reads_actually_made(): ) -# --- a pool cannot be emptied ------------------------------------------------ -def test_a_lane_at_zero_slots_is_never_shrunk_to_an_empty_pool(monkeypatch): - """The error in the operator's live log, 2026-09-23: +# --- 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. - [scheduler] worker_control: ml reconciled 1 -> 0 slots - [ml] pidbox command error: - ValueError("Can't shrink pool. All processes busy!") - billiard refuses to remove the last worker, so a lane at zero stored slots - could never reach its target — and `set_lane_slots_sync` returns True on - SENDING the message, so the reconcile logged success and reported the lane - as `changed` on every tick, forever. Lesson #4183 in production, on the - default configuration of every install. - """ +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) - _stub_live(monkeypatch, "ml", pools={"host-a": 1}, consuming=set()) + _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=0, depth=4000) - wc.reconcile_lanes_sync({"ml": (0, False)}) + d = _worker(wc.size_lanes_sync({"worker": 4})) - assert control.shrank == [], "tried to empty a pool billiard will not empty" + assert (d.action, d.slots) == ("grew", 4) + assert control.grew == [(3, ["host-a"])] -def test_the_floor_applies_to_a_direct_resize_too(monkeypatch): - """Not only the reconcile — the UI dial and the autoscaler go through the - same function, so the clamp belongs there rather than at each caller.""" +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) - state = _stub_live(monkeypatch, "worker", pools={"host-a": 3}) + _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=30, depth=0) - wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 0, live=state) - - assert control.shrank == [(2, ["host-a"])], "should stop at one, not zero" + assert _worker(wc.size_lanes_sync({"worker": 4})).action == "grew" + assert control.grew == [(3, ["host-a"])] -def test_the_autoscaler_will_not_shrink_into_an_empty_pool(monkeypatch): - """A lane whose operator value is 0 and whose live pool is the floor has - nowhere to shrink to — and saying `shrank` every minute is the same - non-convergence wearing a different hat.""" +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) - _autoscale_live(monkeypatch, pools={"host-a": 1}, active=0, reserved=0, depth=0) + _sizing_live(monkeypatch, pools={"host-a": 1}, active=1, reserved=9, depth=0) - d = _worker(wc.autoscale_lanes_sync({"worker": (8, 0, True)})) + 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 diff --git a/tests/test_worker_lanes.py b/tests/test_worker_lanes.py index 4627ad6..b3ba4fc 100644 --- a/tests/test_worker_lanes.py +++ b/tests/test_worker_lanes.py @@ -46,29 +46,50 @@ def test_every_routed_queue_has_a_lane_that_serves_it(): ) -def test_defaults_are_one_of_each_with_ml_off(): - """Operator, 2026-09-22: 'that starting value should be one of each.' +def test_the_shipped_caps_are_one_of_each_with_ml_at_zero(): + """Operator, 2026-09-22: *"that starting value should be one of each."* + Restated 2026-09-23 when the cap became the ONLY number: *"the cap + defaults should be 1 and 0 for the ml-worker."* - ML at zero AND disabled is milestone 422 step 6's requirement arriving - early: enabling the lane is what triggers the SigLIP download, and rule - 164 allows a runtime fetch only for a feature that is optional and clearly - off. A default of 1 here would make every fresh install reach HuggingFace. + ML at zero is milestone 422 step 6's requirement: a cap of zero means no + consumers, and raising it is what triggers the SigLIP download. Rule 164 + allows a runtime fetch only for a feature that is optional and clearly + off, so a default of 1 here would make every fresh install reach + HuggingFace. """ by_name = wl.LANES_BY_NAME for name in ("worker", "scheduler", "maintenance_long"): - assert by_name[name].default_slots == 1 - assert by_name[name].default_enabled is True - assert by_name["ml"].default_slots == 0 - assert by_name["ml"].default_enabled is False + assert by_name[name].default_slots_cap == 1 + assert by_name["ml"].default_slots_cap == 0 -def test_default_caps_leave_room_but_are_not_the_ceiling(): - """A cap that starts at the ceiling is a rubber stamp. Each lane's cap - must be at least its default slots (or the CHECK constraint rejects the - seeded row) and must leave somewhere to grow.""" +def test_a_lane_has_exactly_one_operator_setting(): + """The whole point of the 2026-09-23 reshape. `slots`, `enabled` and + `autoscale` are gone: how many workers run is a measurement the sizing + pass owns, and "off" is a cap of zero. + + Asserted against the dataclass's own fields rather than by name, so a + second knob added later fails here instead of quietly reappearing in the + UI beside the cap — which is exactly how three settings accumulated the + first time.""" + import dataclasses + + settable = { + f.name for f in dataclasses.fields(wl.Lane) if f.name.startswith("default_") + } + assert settable == {"default_slots_cap"}, ( + f"a lane has more than one operator default: {sorted(settable)}" + ) + + +def test_the_shipped_cap_is_never_the_ceiling_on_a_real_machine(): + """A cap that starts at the ceiling is a rubber stamp: there is nowhere to + raise it to, so the nudge that tells an operator to raise it would have + nothing to say. One-of-each leaves room on anything bigger than a + single-core box.""" for lane in wl.LANES: - assert lane.default_slots_cap >= lane.default_slots - assert lane.default_slots_cap >= 1 + assert lane.default_slots_cap >= 0 + assert lane.default_slots_cap <= 1 def test_only_ml_is_memory_bound():