diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index 075e15d..9dad185 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -41,6 +41,7 @@ def all_blueprints() -> list[Blueprint]: from .system_health import system_health_bp from .tags import tags_bp from .thumbnails import thumbnails_bp + from .workers import workers_bp return [ api_bp, attachments_bp, @@ -52,6 +53,7 @@ def all_blueprints() -> list[Blueprint]: showcase_bp, settings_bp, system_activity_bp, + workers_bp, system_health_bp, system_backup_bp, admin_bp, diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py index 0df6e5b..a50bd26 100644 --- a/backend/app/api/system_activity.py +++ b/backend/app/api/system_activity.py @@ -21,18 +21,22 @@ from ..config import get_config from ..extensions import get_session from ..models import TaskRun from ..services.scheduler_service import scheduler_status +from ..services.worker_lanes import LANES system_activity_bp = Blueprint( "system_activity", __name__, url_prefix="/api/system/activity", ) -# Canonical queue order — must match celery_app.task_routes. UI renders -# in this order; queues with no LLEN response show as null rather than -# absent. -_QUEUE_NAMES = ( - "default", "import", "thumbnail", "ml", - "download", "scan", "maintenance", "maintenance_long", -) +# Every queue, grouped by the lane that consumes it. DERIVED from +# `worker_lanes.LANES` (milestone 422 step 1) rather than written out: +# this was a hand-kept third copy of "which queues exist", alongside +# celery_app.task_routes and service_roster.ROLE_NAMES, and its own comment +# admitted the coupling — "must match celery_app.task_routes". +# +# The rendered ORDER changes with this: lane order rather than the previous +# hand-chosen one. That is the better grouping for a lane-oriented UI, and +# queues with no LLEN response still show as null rather than absent. +_QUEUE_NAMES = tuple(q for lane in LANES for q in lane.queues) # Cache module-level so all requests share the cache between polls. # Tests can reset via direct dict mutation if needed. diff --git a/backend/app/api/workers.py b/backend/app/api/workers.py new file mode 100644 index 0000000..b5f96f0 --- /dev/null +++ b/backend/app/api/workers.py @@ -0,0 +1,103 @@ +"""Worker lanes: what each is doing, and the dial that changes it. + +Milestone 422 step 2. The write half of a surface `api/system_activity.py` +only reads. + +## Why this is a separate blueprint + +`system_activity` says in its own first line that it is read-only, and it +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. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from quart import Blueprint, jsonify, request + +from ..extensions import get_session +from ..services.worker_control import LaneUpdateRefused, lane_view, set_lane +from ..services.worker_lanes import LANES_BY_NAME +from ._responses import error_response as _bad + +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. + + Response: {lanes: [...], fetched_at: iso8601} + + Deliberately NOT cached, unlike system_activity's 2s/5s caches. This is + the surface an operator watches while dragging a stepper, and a cached + reply would show them the value from before their own change and read as + the control having failed. + """ + async with get_session() as session: + lanes = await lane_view(session) + return jsonify({ + "lanes": lanes, + "fetched_at": datetime.now(UTC).isoformat(), + }) + + +@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. + + Partial: only the keys present are changed, so the UI's stepper can send + `{"slots": 3}` without restating the cap it did not touch. + + 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. + * **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". + """ + lane = LANES_BY_NAME.get(name) + if lane is None: + return _bad("unknown_lane", detail=name, known=sorted(LANES_BY_NAME)) + + body = await request.get_json() + 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 + if "enabled" in body: + if not isinstance(body["enabled"], bool): + return _bad("invalid_body", detail="enabled must be a boolean") + fields["enabled"] = body["enabled"] + + if not fields: + return _bad( + "invalid_body", + detail="give at least one of slots, slots_cap, enabled", + ) + + async with get_session() as session: + try: + result = await set_lane(session, lane, **fields) + except LaneUpdateRefused as exc: + return _bad("refused", detail=str(exc)) + return jsonify(result) diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index b9b7493..7adac19 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -47,10 +47,15 @@ autoscaler. from __future__ import annotations +import asyncio import logging from dataclasses import dataclass, field -from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import WorkerLane +from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane, derived_ceiling log = logging.getLogger(__name__) @@ -74,11 +79,22 @@ class LaneLiveState: present: bool = False replicas: int = 0 - # Per-process pool size. Equal across replicas unless one has drifted. - pool: int | None = None active: int = 0 reserved: int = 0 hostnames: list[str] = field(default_factory=list) + # Pool size PER HOSTNAME, not aggregated. The resize below computes each + # replica's own delta from its own current pool, so replicas that have + # drifted apart converge instead of being moved in lockstep from a shared + # baseline — which is what an aggregate here would silently reintroduce. + pools: dict[str, int] = field(default_factory=dict) + + @property + def pool(self) -> int | None: + """One number for the UI. `max` rather than a sum: `slots` means the + pool size of ONE process (see the module docstring), so the largest + replica is the honest answer to "what is this lane set to". None when + no replica reported — unknown, never zero.""" + return max(self.pools.values()) if self.pools else None def _lane_for_queues(queues: tuple[str, ...]) -> Lane | None: @@ -124,7 +140,7 @@ def inspect_lanes_sync() -> dict[str, LaneLiveState]: # answer, which leaves pool=None — unknown, not zero. pool = (stats.get(hostname) or {}).get("pool", {}).get("max-concurrency") if isinstance(pool, int): - state.pool = pool if state.pool is None else max(state.pool, pool) + state.pools[hostname] = pool for state in out.values(): state.hostnames.sort() @@ -150,16 +166,22 @@ def set_lane_slots_sync(lane: Lane, target: int) -> tuple[bool, str | None]: live = inspect_lanes_sync()[lane.name] if not live.present: return False, "lane is not running" - if live.pool is None: + if not live.pools: return False, "worker did not report its pool size" control = celery_app.control - for hostname in live.hostnames: - delta = target - live.pool + unreported = [h for h in live.hostnames if h not in live.pools] + for hostname, current in live.pools.items(): + delta = target - current if delta > 0: control.pool_grow(delta, destination=[hostname]) elif delta < 0: control.pool_shrink(-delta, destination=[hostname]) + if unreported: + # Resized what could be resized, and said which could not. Silence + # here would leave a replica running at a size the UI claims it is + # not, with nothing anywhere recording the gap. + return False, f"no pool size reported by {', '.join(sorted(unreported))}" return True, None except Exception as exc: # noqa: BLE001 — reported, never raised at a caller log.warning("worker_control: could not resize %s", lane.name, exc_info=True) @@ -194,3 +216,176 @@ def set_lane_enabled_sync(lane: Lane, enabled: bool) -> tuple[bool, str | None]: "enable" if enabled else "disable", lane.name, exc_info=True, ) return False, str(exc) + + +# --- the settings half, which is async ---------------------------------------- +# +# Sync celery control above, async DB below, in one module. Same split +# `service_roster` already runs (`_inspect_celery_sync` beside `touch_service`) +# — the boundary is the transport, not the concern, and "control the workers" +# is one concern. + + +async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]: + """Every lane's row, creating any that are missing from its LANES defaults. + + Self-heals rather than depending on a migration having run for a lane + added later: alembic 0103 seeded the four that existed on 2026-09-22, and + a fifth added to LANES afterwards gets its row the first time anything + asks. Without this, a new lane would read as absent and the UI would + simply not show it. + """ + rows = { + row.name: row + for row in (await session.execute(select(WorkerLane))).scalars() + } + 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, + ) + session.add(row) + rows[lane.name] = row + if missing: + await session.commit() + return rows + + +async def lane_view(session: AsyncSession) -> list[dict]: + """Every lane: what is configured, what is live, what it may grow to. + + One call rather than making the UI join three sources. `pending` is the + honest backlog — Redis depth PLUS reserved — because celery prefetches and + LLEN alone reads 0 while a worker holds tasks in memory. + """ + rows = await _rows_by_name(session) + live = await asyncio.to_thread(inspect_lanes_sync) + depths = await asyncio.to_thread(_queue_depths_sync) + + out = [] + for lane in LANES: + row = rows[lane.name] + state = live[lane.name] + # None for a queue the broker did not answer for, which must not be + # silently summed as zero — an unknown depth is not an empty one. + 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 + out.append({ + "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, + "memory_bound": lane.memory_bound, + "live": { + "present": state.present, + "replicas": state.replicas, + "pool": state.pool, + "active": state.active, + "reserved": state.reserved, + }, + "queue_depth": depth, + "pending": None if depth is None else depth + state.reserved, + }) + return out + + +def _queue_depths_sync() -> dict[str, int | None]: + """Redis LLEN per queue. None for one that did not answer — see lane_view. + + Sync; the caller threads it. A per-queue try/except so one bad queue does + not cost the whole report, matching `api/system_activity._read_queues_sync`. + """ + import redis + + from ..config import get_config + + out: dict[str, int | None] = {} + try: + client = redis.Redis.from_url(get_config().celery_broker_url) + except Exception: + log.warning("worker_control: no broker for queue depths", exc_info=True) + return {q: None for lane in LANES for q in lane.queues} + for lane in LANES: + for queue in lane.queues: + try: + out[queue] = int(client.llen(queue)) + except Exception: # noqa: BLE001 — a hiccup must not break the UI + out[queue] = None + return out + + +class LaneUpdateRefused(ValueError): + """A requested value is outside what the lane may hold. Carries the reason + 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, +) -> dict: + """Store the operator's choice, then push it to the running lane. + + 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. + + 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". + """ + 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_enabled = row.enabled if enabled is None else 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: + raise LaneUpdateRefused( + f"cap {new_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 + await session.commit() + + applied, error = True, None + if enabled is not None: + applied, error = await asyncio.to_thread( + set_lane_enabled_sync, lane, new_enabled, + ) + if applied and slots is not None: + applied, error = await asyncio.to_thread(set_lane_slots_sync, lane, new_slots) + + return { + "name": lane.name, + "slots": row.slots, + "slots_cap": row.slots_cap, + "ceiling": ceiling, + "enabled": row.enabled, + "applied": applied, + "apply_error": error, + } diff --git a/tests/test_api_workers.py b/tests/test_api_workers.py new file mode 100644 index 0000000..97bd728 --- /dev/null +++ b/tests/test_api_workers.py @@ -0,0 +1,172 @@ +"""/api/system/workers — the lane dial (milestone 422 step 2). + +Exercises the real endpoint against the real database. Only `celery inspect` +is stubbed, and only to keep the suite fast: an unstubbed inspect blocks for +its full 2s timeout per call with no workers to answer, which several writes +would turn into most of the lane's runtime. +""" + +import pytest +import pytest_asyncio +from sqlalchemy import select + +from backend.app.models import WorkerLane +from backend.app.services import worker_control as wc +from backend.app.services.worker_lanes import LANES + +pytestmark = pytest.mark.integration + + +@pytest_asyncio.fixture +async def no_live_workers(monkeypatch): + """Nothing is running — which is the CI lane's actual truth, asserted + rather than waited for. Makes every push fail, which is the interesting + half: the setting must still be stored.""" + monkeypatch.setattr( + wc, "inspect_lanes_sync", + lambda: {lane.name: wc.LaneLiveState() for lane in LANES}, + ) + + +async def _lane_row(db, name: str) -> WorkerLane: + return (await db.execute( + select(WorkerLane).where(WorkerLane.name == name) + )).scalar_one() + + +# --- reading ----------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_get_lists_every_lane_with_its_ceiling(client, no_live_workers): + resp = await client.get("/api/system/workers") + assert resp.status_code == 200 + body = await resp.get_json() + + by_name = {lane["name"]: lane for lane in body["lanes"]} + assert set(by_name) == {"worker", "scheduler", "maintenance_long", "ml"} + for lane in body["lanes"]: + assert lane["ceiling"] >= 0 + assert lane["slots"] <= lane["slots_cap"] + # 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_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.""" + 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 False + assert ml["slots"] == 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 +): + """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}) + assert resp.status_code == 200 + body = await resp.get_json() + + assert body["slots"] == 3 + assert body["applied"] is False + assert "not running" in body["apply_error"] + + assert (await _lane_row(db, "worker")).slots == 3 + + +@pytest.mark.asyncio +async def test_a_partial_update_leaves_the_other_fields_alone( + client, db, no_live_workers +): + """The stepper sends `{"slots": n}` without restating a cap it did not + touch.""" + before = await _lane_row(db, "worker") + original_cap = before.slots_cap + + await client.post("/api/system/workers/worker", json={"slots": 2}) + + await db.refresh(before) + assert before.slots == 2 + assert before.slots_cap == original_cap + + +# --- 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 +): + """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.""" + 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"] + + +@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}) + assert resp.status_code == 400 + body = await resp.get_json() + assert body["error"] == "invalid_body" + + +@pytest.mark.asyncio +async def test_an_unknown_lane_is_refused_and_names_the_known_ones( + client, no_live_workers +): + resp = await client.post("/api/system/workers/nonsense", json={"slots": 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 +): + """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" diff --git a/tests/test_worker_control.py b/tests/test_worker_control.py new file mode 100644 index 0000000..47df672 --- /dev/null +++ b/tests/test_worker_control.py @@ -0,0 +1,188 @@ +"""Changing a lane's slots on a running system (milestone 422 step 2). + +The celery control calls are stubbed: what is being tested is the DELTA +ARITHMETIC and the persist/push split, not that celery can resize its own +pool. A test that asserted celery's behaviour would be testing celery. +""" + +from __future__ import annotations + +import pytest + +from backend.app.services import worker_control as wc +from backend.app.services.worker_lanes import LANES_BY_NAME + +# --- what inspect reports ---------------------------------------------------- + + +class _Control: + """Records the control messages that were sent.""" + + def __init__(self): + self.grew: list[tuple[int, list[str]]] = [] + self.shrank: list[tuple[int, list[str]]] = [] + self.added: list[tuple[str, list[str]]] = [] + self.cancelled: list[tuple[str, list[str]]] = [] + + def pool_grow(self, n, destination=None): + self.grew.append((n, destination)) + + def pool_shrink(self, n, destination=None): + self.shrank.append((n, destination)) + + def add_consumer(self, queue, destination=None): + self.added.append((queue, destination)) + + def cancel_consumer(self, queue, destination=None): + self.cancelled.append((queue, destination)) + + +def _stub_live(monkeypatch, lane_name, *, pools, present=True, reserved=0): + state = wc.LaneLiveState( + present=present, + replicas=len(pools), + hostnames=sorted(pools), + pools=dict(pools), + reserved=reserved, + ) + monkeypatch.setattr( + wc, "inspect_lanes_sync", + lambda: {name: (state if name == lane_name else wc.LaneLiveState()) + for name in LANES_BY_NAME}, + ) + return state + + +def _stub_control(monkeypatch): + control = _Control() + + class _Celery: + pass + + celery = _Celery() + celery.control = control + import sys + import types + mod = types.ModuleType("backend.app.celery_app") + mod.celery = celery + monkeypatch.setitem(sys.modules, "backend.app.celery_app", mod) + return control + + +def test_pool_property_is_max_not_sum(): + """`slots` means the pool size of ONE process, so the aggregate shown to + the operator is the largest replica — not the total. A sum would report 8 + for two replicas of 4 and invite them to 'reduce it to 4', which would + halve the lane.""" + state = wc.LaneLiveState(pools={"a": 4, "b": 4}) + assert state.pool == 4 + + +def test_pool_is_none_when_nothing_reported(): + """Unknown, never zero — the distinction step 3's reconcile depends on.""" + assert wc.LaneLiveState(present=True).pool is None + + +# --- the delta arithmetic ---------------------------------------------------- + + +def test_each_replica_gets_its_own_delta(monkeypatch): + """The bug this exists to prevent: one delta computed from an aggregate + and applied to every replica. With replicas at 2 and 6 and a target of 4, + a shared delta moves both the same way and leaves them at 4 and 8 — or 0 + and 4 — depending on which aggregate was used. Per-replica deltas + converge both on 4. + """ + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={"host-a": 2, "host-b": 6}) + + applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) + + assert (applied, err) == (True, None) + assert control.grew == [(2, ["host-a"])] + assert control.shrank == [(2, ["host-b"])] + + +def test_a_replica_already_at_the_target_is_sent_nothing(monkeypatch): + """The fixed point step 3's reconcile needs. An enforcer that re-issues a + grow of zero every tick never converges and re-does its own work forever + (lesson #4183).""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={"host-a": 4}) + + applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) + + assert (applied, err) == (True, None) + assert control.grew == [] + assert control.shrank == [] + + +def test_resizing_an_absent_lane_reports_rather_than_raises(monkeypatch): + _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={}, present=False) + + applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) + + assert applied is False + assert "not running" in err + + +def test_a_replica_with_no_reported_pool_is_named_not_skipped_silently( + monkeypatch, +): + """Resize what can be resized, then say which could not. Silence would + leave a replica running at a size the UI claims it is not.""" + control = _stub_control(monkeypatch) + state = _stub_live(monkeypatch, "worker", pools={"host-a": 2}) + state.hostnames = ["host-a", "host-b"] # b answered inspect, not stats + state.replicas = 2 + + applied, err = wc.set_lane_slots_sync(LANES_BY_NAME["worker"], 4) + + assert control.grew == [(2, ["host-a"])] + assert applied is False + assert "host-b" in err + + +# --- enabling and disabling -------------------------------------------------- + + +def test_disabling_cancels_consumers_rather_than_killing_the_worker(monkeypatch): + """A cancelled consumer keeps the process alive and answering inspect, so + a disabled lane stays visible. A killed worker reads as ABSENT, which is + the same signal as a crash — and milestone 365 exists precisely so those + two do not look alike.""" + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "ml", pools={"host-a": 1}) + + applied, err = wc.set_lane_enabled_sync(LANES_BY_NAME["ml"], False) + + assert (applied, err) == (True, None) + assert control.cancelled == [("ml", ["host-a"])] + assert control.added == [] + + +def test_enabling_adds_a_consumer_for_every_queue_in_the_lane(monkeypatch): + control = _stub_control(monkeypatch) + _stub_live(monkeypatch, "worker", pools={"host-a": 1}) + + wc.set_lane_enabled_sync(LANES_BY_NAME["worker"], True) + + assert [q for q, _ in control.added] == list(LANES_BY_NAME["worker"].queues) + + +# --- lane identity ----------------------------------------------------------- + + +def test_queue_sets_map_back_to_their_lane_in_any_order(): + """celery does not guarantee the order it lists a worker's queues in, so + the lookup sorts. Unsorted, a lane would intermittently fail to match and + read as absent.""" + lane = LANES_BY_NAME["worker"] + assert wc._lane_for_queues(tuple(reversed(lane.queues))) is lane + + +def test_an_unknown_queue_set_maps_to_no_lane(): + """A deployment slicing CELERY_QUEUES differently has no lane row to + control. Honest rather than an error — the roster still reports it.""" + assert wc._lane_for_queues(("something", "else")) is None