diff --git a/alembic/versions/0103_worker_lane_settings.py b/alembic/versions/0103_worker_lane_settings.py new file mode 100644 index 0000000..5dca467 --- /dev/null +++ b/alembic/versions/0103_worker_lane_settings.py @@ -0,0 +1,121 @@ +"""worker_lane — settings-backed slots for each celery lane. + +Milestone 422 step 1. One row per lane, holding only what an operator can +change: how many slots it runs, the ceiling they have set for themselves, and +whether it consumes its queues at all. + +## What is deliberately not a column + +**The queues.** They are decided by `celery_app.py`'s `task_routes`, not by +preference, so a stored copy could contradict the routing table with nothing +to notice until a queue had no consumer. They live in +`services/worker_lanes.py`. + +**The derived ceiling.** Computed from the container's cgroup limits on every +read. A row written on a 32GB host and later run in a 4GB container must be +bounded by the 4GB; a stored ceiling would quietly authorise what the box can +no longer hold. + +## The seeded values + +Written out literally rather than imported from `worker_lanes.LANES`. A +migration is a statement about one moment in the schema's history — if it +imported the live defaults, changing them in 2027 would silently change what +this 2026 revision does on a fresh database. The two are allowed to diverge +afterwards, and that is correct: `LANES` supplies defaults for a lane added +later, this file records what was seeded today. + + lane slots cap enabled + worker 1 4 yes + scheduler 1 2 yes + maintenance_long 1 2 yes + ml 0 1 NO + +One of each, per the operator (2026-09-22: *"that starting value should be one +of each"*), and far below their own production numbers — worker 8 and ml 2 are +tuned for their hardware and are not a sane first boot for a stranger. + +**ml ships at zero and disabled**, which is milestone 422 step 6's requirement +arriving early: enabling the lane is what triggers the SigLIP download, and +rule 164 permits a runtime fetch only for a feature that is "optional and +clearly off". Seeding it on would make every fresh install reach HuggingFace. + +The caps start low on purpose. A cap that begins at the ceiling is a rubber +stamp; starting at 4/2/2/1 means raising slots within the cap is ordinary and +raising the cap is a deliberate act. + +## Existing installs + +Nothing is migrated FROM. The `CELERY_QUEUES` / `CELERY_CONCURRENCY` env vars +stay exactly as they are and remain the baseline each lane boots at; these +rows are the adjustment applied on top (step 3). So this migration changes no +behaviour on a running stack — it only makes the numbers storable. + +Revision ID: 0103 +Revises: 0102 +Create Date: 2026-09-22 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0103" +down_revision: Union[str, None] = "0102" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +# (name, slots, slots_cap, enabled) — see the docstring for why these are +# literals and not an import. +_SEED = ( + ("worker", 1, 4, True), + ("scheduler", 1, 2, True), + ("maintenance_long", 1, 2, True), + ("ml", 0, 1, False), +) + + +def upgrade() -> None: + worker_lane = op.create_table( + "worker_lane", + sa.Column("name", sa.String(length=32), nullable=False), + sa.Column("slots", sa.Integer(), nullable=False), + sa.Column("slots_cap", sa.Integer(), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column( + "updated_at", sa.DateTime(timezone=True), + server_default=sa.text("now()"), nullable=False, + ), + sa.PrimaryKeyConstraint("name", name=op.f("pk_worker_lane")), + # Bare constraint names: Base.metadata's naming convention prepends + # ck_worker_lane_, and pre-prefixing doubles it — the defect alembic + # 0088 had to rename four constraints for (#3275). op.f() marks these + # as already-final so autogenerate does not propose renaming them. + sa.CheckConstraint("slots >= 0", name=op.f("ck_worker_lane_slots_non_negative")), + sa.CheckConstraint("slots_cap >= 0", name=op.f("ck_worker_lane_cap_non_negative")), + # The invariant that makes the cap mean anything, in the database + # rather than only in the service: a row violating it is not a + # rejected request, it is a lane that step 3's reconcile will drive UP + # to a number the operator capped. + sa.CheckConstraint("slots <= slots_cap", name=op.f("ck_worker_lane_slots_within_cap")), + ) + # No index beyond the primary key, deliberately — four rows, forever. Same + # reasoning as service_seen, and the lesson of #3301, which removed seven + # indexes that were write cost buying nothing. + op.bulk_insert( + worker_lane, + [ + {"name": name, "slots": slots, "slots_cap": cap, "enabled": enabled} + for name, slots, cap, enabled in _SEED + ], + ) + + +def downgrade() -> None: + # The rows go with the table. They are settings with shipped defaults, not + # operator data that predates this revision — a downgrade returns the stack + # to reading its concurrency from env, which is where it reads it from + # today anyway. + op.drop_table("worker_lane") diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index bc0cf30..08f1d3f 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -44,6 +44,7 @@ from .tag_head import TagHead from .tag_positive_confirmation import TagPositiveConfirmation from .tag_suggestion_rejection import TagSuggestionRejection from .task_run import TaskRun +from .worker_lane import WorkerLane __all__ = [ "Base", @@ -94,4 +95,5 @@ __all__ = [ "TagPositiveConfirmation", "TagSuggestionRejection", "TaskRun", + "WorkerLane", ] diff --git a/backend/app/models/worker_lane.py b/backend/app/models/worker_lane.py new file mode 100644 index 0000000..ae3cc09 --- /dev/null +++ b/backend/app/models/worker_lane.py @@ -0,0 +1,86 @@ +"""worker_lane — how many slots the operator wants each worker lane to have. + +Milestone 422 step 1. One row per lane in `services/worker_lanes.LANES`. + +## What is NOT in here + +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. + +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. + +## The three numbers + + slots <= slots_cap <= derived_ceiling + (live) (this row) (computed) + +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 + +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. + +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. +""" + +from datetime import datetime + +from sqlalchemy import ( + Boolean, + CheckConstraint, + DateTime, + Integer, + String, + func, +) +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class WorkerLane(Base): + __tablename__ = "worker_lane" + __table_args__ = ( + # Bare names — 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 + # hostname. See models/service_seen.py for why: celery's worker names here + # 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) + + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + server_default=func.now(), + onupdate=func.now(), + ) diff --git a/backend/app/services/service_roster.py b/backend/app/services/service_roster.py index 5b16f3b..ff831b2 100644 --- a/backend/app/services/service_roster.py +++ b/backend/app/services/service_roster.py @@ -36,6 +36,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from ..models import ServiceSeen +from .worker_lanes import LANES log = logging.getLogger(__name__) @@ -53,13 +54,18 @@ INSPECT_TIMEOUT_SECONDS = 2.0 # Queue set -> the name an operator recognises. Sorted-tuple keys, because the # order celery reports them in is not guaranteed. # -# A deployment that slices CELERY_QUEUES differently falls through to the raw -# queue list rather than being given a name this table invented for it: a +# DERIVED from `worker_lanes.LANES` (milestone 422 step 1) rather than written +# out here. It was a hand-kept second copy of the same fact, and it had already +# drifted: `maintenance_long` is a live lane with four task routes pointing at +# it and a dedicated worker in the operator's stack, and this map did not know +# it — so the System tab labelled it `Worker (maintenance_long)`. One list of +# lanes now names them everywhere. +# +# A deployment that slices CELERY_QUEUES differently still falls through to the +# raw queue list rather than being given a name this code invented for it: a # wrong-but-confident label on a status page is worse than an ugly true one. ROLE_NAMES: dict[tuple[str, ...], str] = { - ("default", "download", "import", "thumbnail"): "Worker", - ("maintenance", "scan"): "Scheduler", - ("ml",): "ML worker", + lane.queue_key: lane.display_name for lane in LANES } diff --git a/backend/app/services/worker_lanes.py b/backend/app/services/worker_lanes.py new file mode 100644 index 0000000..a0d7444 --- /dev/null +++ b/backend/app/services/worker_lanes.py @@ -0,0 +1,281 @@ +"""The worker lanes: what they are, and how many slots each may be given. + +Milestone 422 step 1. This module is the ONE place that knows the lane set; +`models/worker_lane.py` holds only what the operator can change about them. + +## Why the queues are here and not in the table + +A lane's queue set is not a preference — it is decided by `celery_app.py`'s +`task_routes`, which is what puts a backup on `maintenance_long` and a +thumbnail on `thumbnail`. An operator cannot move a task to another lane, so +storing the queues as settings would create a row that can disagree with the +routing table, and nothing would notice until a queue had no consumer. + +So: queues and display names are code, slots and caps are data. The table +stores three numbers and a flag, and nothing that could contradict celery. + +This also collapses a duplicate rather than adding one. +`service_roster.ROLE_NAMES` was a second copy of "queue set -> the name an +operator recognises", and it had already drifted: `maintenance_long` is a +live lane with four task routes pointing at it, and the roster did not know +its name, so the System tab rendered it as `Worker (maintenance_long)`. That +map is now derived from `LANES` below, so a lane added here is named +everywhere at once. + +## Why the ceiling is derived rather than configured + +Consolidating the stack into one container (step 5) widens the OOM blast +radius: today an ml-worker that exhausts memory is killed by Docker on its +own, and web keeps serving. In one container the kernel picks a victim from +the whole cgroup, and it may pick hypercorn — so a tagging task can take the +UI down with it, on exactly the modest hardware least able to spare the +memory. + +Operator, 2026-09-22: *"ram isn't an issue for me but some users might run +this on weaker hardware and I don't want it to kill their servers."* + +So the maximum is computed from what the container actually has, and the +operator's own `slots_cap` must fit under it. Three numbers, not two, and the +ordering is the point: + + slots <= slots_cap <= derived_ceiling + (live) (operator) (this module) + +The operator can always lower their cap. They cannot raise it past what the +box can hold. The derived ceiling is never stored — a row that outlived a +change in container limits must not carry a stale one. +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass +from pathlib import Path + +log = logging.getLogger(__name__) + +# --- the lanes --------------------------------------------------------------- + + +@dataclass(frozen=True) +class Lane: + """A worker lane. `name` is the stable key the settings row is keyed on. + + Keyed on a lane NAME rather than a container hostname for the reason + `models/service_seen.py` gives at length: celery's worker names here are + `celery@` and are minted fresh on every deploy, so anything + keyed on them records a death and a birth every time the stack updates. + """ + + name: str + display_name: str + queues: tuple[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. + default_slots_cap: int + default_enabled: bool + # 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 + + @property + def queue_key(self) -> tuple[str, ...]: + """The sorted queue set, which is how `service_seen` identifies a + running worker. The join between what is configured here and what + `celery inspect` reports.""" + 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. +# +# 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. +LANES: tuple[Lane, ...] = ( + Lane( + name="worker", + display_name="Worker", + queues=("default", "import", "thumbnail", "download"), + default_slots=1, + default_slots_cap=4, + default_enabled=True, + ), + Lane( + name="scheduler", + display_name="Scheduler", + queues=("maintenance", "scan"), + default_slots=1, + default_slots_cap=2, + default_enabled=True, + ), + Lane( + name="maintenance_long", + display_name="Long maintenance", + queues=("maintenance_long",), + default_slots=1, + default_slots_cap=2, + default_enabled=True, + ), + Lane( + name="ml", + display_name="ML tagging", + queues=("ml",), + default_slots=0, + default_slots_cap=1, + default_enabled=False, + memory_bound=True, + ), +) + +LANES_BY_NAME: dict[str, Lane] = {lane.name: lane for lane in LANES} +LANES_BY_QUEUE_KEY: dict[tuple[str, ...], Lane] = { + lane.queue_key: lane for lane in LANES +} + + +# --- what the container actually has ----------------------------------------- + +# cgroup v2 first, then v1. A container started without an explicit memory +# limit reports "max" on v2 and a sentinel near 2**63 on v1; both mean "no +# limit", and the answer then is the host's RAM. +_CGROUP_V2_MEMORY = Path("/sys/fs/cgroup/memory.max") +_CGROUP_V1_MEMORY = Path("/sys/fs/cgroup/memory/memory.limit_in_bytes") +_CGROUP_V2_CPU = Path("/sys/fs/cgroup/cpu.max") +_CGROUP_V1_CPU_QUOTA = Path("/sys/fs/cgroup/cpu/cpu.cfs_quota_us") +_CGROUP_V1_CPU_PERIOD = Path("/sys/fs/cgroup/cpu/cpu.cfs_period_us") + +# A v1 "unlimited" is PAGE_SIZE-aligned LONG_MAX, not a round number, so it is +# recognised by magnitude rather than by equality. Anything claiming more than +# a petabyte is a sentinel, not a machine. +_UNLIMITED_ABOVE = 1 << 50 + +GIB = 1024 ** 3 + +# Memory one ML slot needs: the SigLIP so400m weights plus the runtime holding +# them. Prefork forks a child per slot and each child loads its own copy, so +# this multiplies — it is not a one-off cost. +# +# UNMEASURED AND DELIBERATELY CONSERVATIVE. This number decides whether a +# stranger's server survives enabling tagging, so it errs toward refusing a +# slot that would have fitted rather than granting one that will not. To +# replace it with a real figure: enable the lane on a container with a known +# limit, run one tagging task, and read the worker child's peak RSS +# (`grep VmHWM /proc//status`). Put the measurement in the commit +# message when you do. +ML_BYTES_PER_SLOT = 4 * GIB + +# Held back for hypercorn and the non-ML lanes before any ML slot is offered. +# In the consolidated container these share one cgroup with ML, and they are +# the processes an OOM kill must not take (see the module docstring). +RESERVED_BYTES = 2 * GIB + +# 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. +MIN_CEILING = 1 + +# What an unreadable limit yields. Low rather than unlimited, on purpose: not +# knowing how much memory there is must never read as "plenty". An unswept +# absence is not a verdict. +UNKNOWN_CEILING = 1 + + +def _read_int(path: Path) -> int | None: + try: + raw = path.read_text().strip() + except OSError: + return None + if raw == "max": + return None + try: + return int(raw) + except ValueError: + return None + + +def container_memory_bytes() -> int | None: + """The memory this container may use, or None when it cannot be read. + + None means UNKNOWN, never UNLIMITED. Every caller must treat it as the + conservative case — the whole point of the ceiling is to protect a machine + whose size we are unsure of. + """ + for path in (_CGROUP_V2_MEMORY, _CGROUP_V1_MEMORY): + value = _read_int(path) + if value is not None and value < _UNLIMITED_ABOVE: + return value + if value is not None: + # A sentinel: the cgroup exists but sets no limit, so the real + # bound is the host's. + break + try: + return os.sysconf("SC_PHYS_PAGES") * os.sysconf("SC_PAGE_SIZE") + except (ValueError, OSError, AttributeError): + return None + + +def container_cpu_count() -> int | None: + """Effective cores, honouring a cgroup CPU quota. + + `os.cpu_count()` reports the HOST's cores from inside a container, so a + quota of 2.0 on a 32-core host would otherwise offer 32 slots. The + operator's own stack sets `cpus: '4.0'` on ml-worker, so this is a real + configuration here and not a hypothetical. + """ + quota: float | None = None + try: + raw = _CGROUP_V2_CPU.read_text().strip().split() + if raw and raw[0] != "max": + quota = int(raw[0]) / int(raw[1]) + except (OSError, ValueError, IndexError, ZeroDivisionError): + pass + if quota is None: + q = _read_int(_CGROUP_V1_CPU_QUOTA) + p = _read_int(_CGROUP_V1_CPU_PERIOD) + if q is not None and p and q > 0: + quota = q / p + if quota is not None and quota > 0: + return max(1, int(quota)) + return os.cpu_count() + + +def derived_ceiling(lane: Lane) -> int: + """The most slots `lane` may be given on this container. + + Never stored. Recomputed on every read so a container whose limits changed + is bounded by what it has NOW rather than by what it had when its row was + written. + """ + if lane.memory_bound: + total = container_memory_bytes() + if total is None: + log.warning( + "worker_lanes: cannot read a memory limit; capping %s at %d", + lane.name, UNKNOWN_CEILING, + ) + return UNKNOWN_CEILING + usable = total - RESERVED_BYTES + if usable < ML_BYTES_PER_SLOT: + # Honestly zero. A box that cannot hold one model alongside the web + # process must be told it cannot run tagging, not sold a slot that + # will OOM the container the first time it is used. + return 0 + return int(usable // ML_BYTES_PER_SLOT) + + cores = container_cpu_count() + if cores is None: + return UNKNOWN_CEILING + return max(MIN_CEILING, cores) + + +def ceilings() -> dict[str, int]: + """Every lane's ceiling, for the settings API and the UI.""" + return {lane.name: derived_ceiling(lane) for lane in LANES} diff --git a/tests/test_worker_lanes.py b/tests/test_worker_lanes.py new file mode 100644 index 0000000..7482e4e --- /dev/null +++ b/tests/test_worker_lanes.py @@ -0,0 +1,258 @@ +"""Worker-lane definitions and the derived ceiling (milestone 422 step 1). + +Pure — no DB, no broker. The ceiling reads files under /sys/fs/cgroup, so the +tests point it at temporary files instead of at whatever the CI container +happens to have. +""" + +from __future__ import annotations + +import pytest + +from backend.app.services import worker_lanes as wl + + +# --- the lane definitions ---------------------------------------------------- + + +def test_lane_names_are_unique_and_stable(): + names = [lane.name for lane in wl.LANES] + assert len(names) == len(set(names)) + # The key every settings row and every API call uses. Renaming one orphans + # its row silently, so the set is pinned rather than merely counted. + assert set(names) == {"worker", "scheduler", "maintenance_long", "ml"} + + +def test_queue_sets_are_unique(): + """Two lanes sharing a queue set would be indistinguishable to + `celery inspect`, which groups by exactly that (see service_roster).""" + keys = [lane.queue_key for lane in wl.LANES] + assert len(keys) == len(set(keys)) + + +def test_every_routed_queue_has_a_lane_that_serves_it(): + """The check that actually matters: a queue celery routes tasks to, with + no lane consuming it, is work that queues forever and never runs. + + Read from `celery_app`'s real routing table rather than a list written + here, so adding a route without a lane fails this test. + """ + from backend.app.celery_app import celery as celery_app + + routes = celery_app.conf.task_routes or {} + routed = {spec["queue"] for spec in routes.values() if "queue" in spec} + served = {q for lane in wl.LANES for q in lane.queues} + assert routed - served == set(), ( + f"queues with no lane to consume them: {sorted(routed - served)}" + ) + + +def test_defaults_are_one_of_each_with_ml_off(): + """Operator, 2026-09-22: 'that starting value should be one of each.' + + 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. + """ + 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 + + +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.""" + for lane in wl.LANES: + assert lane.default_slots_cap >= lane.default_slots + assert lane.default_slots_cap >= 1 + + +def test_only_ml_is_memory_bound(): + """Which lane is memory-bound decides which ceiling formula applies. If + another lane ever becomes so, it needs its own per-slot cost — the ML + figure is about a SigLIP copy and means nothing for a thumbnailer.""" + assert [lane.name for lane in wl.LANES if lane.memory_bound] == ["ml"] + + +# --- reading the container's limits ------------------------------------------ + + +def _point_memory_at(monkeypatch, tmp_path, contents: str | None, *, v2=True): + target = tmp_path / "memory.max" + if contents is not None: + target.write_text(contents) + attr = "_CGROUP_V2_MEMORY" if v2 else "_CGROUP_V1_MEMORY" + monkeypatch.setattr(wl, attr, target) + # Neutralise the other tier so the test controls exactly one source. + other = "_CGROUP_V1_MEMORY" if v2 else "_CGROUP_V2_MEMORY" + monkeypatch.setattr(wl, other, tmp_path / "absent") + + +def test_memory_read_from_cgroup_v2(monkeypatch, tmp_path): + _point_memory_at(monkeypatch, tmp_path, str(8 * wl.GIB)) + assert wl.container_memory_bytes() == 8 * wl.GIB + + +def test_memory_read_from_cgroup_v1(monkeypatch, tmp_path): + target = tmp_path / "limit_in_bytes" + target.write_text(str(6 * wl.GIB)) + monkeypatch.setattr(wl, "_CGROUP_V2_MEMORY", tmp_path / "absent") + monkeypatch.setattr(wl, "_CGROUP_V1_MEMORY", target) + assert wl.container_memory_bytes() == 6 * wl.GIB + + +def test_v2_max_means_no_limit_not_zero(monkeypatch, tmp_path): + """'max' is a container with no memory limit set. Falling through to host + RAM is right; reading it as 0 or as an error would cap a large machine at + the unknown-ceiling.""" + _point_memory_at(monkeypatch, tmp_path, "max") + value = wl.container_memory_bytes() + assert value is not None and value > wl.GIB + + +def test_v1_sentinel_is_recognised_as_unlimited(monkeypatch, tmp_path): + """cgroup v1 spells 'no limit' as a PAGE_SIZE-aligned LONG_MAX rather than + a word, so it is recognised by magnitude. Taken literally it would be + petabytes and the ML ceiling would be nonsense.""" + target = tmp_path / "limit_in_bytes" + target.write_text(str(9223372036854771712)) + monkeypatch.setattr(wl, "_CGROUP_V2_MEMORY", tmp_path / "absent") + monkeypatch.setattr(wl, "_CGROUP_V1_MEMORY", target) + value = wl.container_memory_bytes() + assert value is not None + assert value < wl._UNLIMITED_ABOVE + + +# --- the derived ceiling ----------------------------------------------------- + + +def test_ml_ceiling_is_memory_divided_by_per_slot_cost(monkeypatch, tmp_path): + # 2 GiB reserved for web and the other lanes, then 4 GiB per model copy. + _point_memory_at(monkeypatch, tmp_path, str(14 * wl.GIB)) + assert wl.derived_ceiling(wl.LANES_BY_NAME["ml"]) == 3 + + +def test_a_small_box_is_told_it_cannot_run_tagging(monkeypatch, tmp_path): + """Honestly zero rather than a floor of one. A 4GB box cannot hold a model + alongside the web process, and offering a slot that OOMs the container the + first time it is used is precisely what the ceiling exists to prevent — + the operator's 'I don't want it to kill their servers'.""" + _point_memory_at(monkeypatch, tmp_path, str(4 * wl.GIB)) + assert wl.derived_ceiling(wl.LANES_BY_NAME["ml"]) == 0 + + +def test_unreadable_memory_limit_yields_a_low_ceiling_not_an_unlimited_one( + monkeypatch, tmp_path +): + """The failure DIRECTION is the point. Not knowing how much memory there + is must never read as 'plenty' — an unswept absence is not a verdict.""" + _point_memory_at(monkeypatch, tmp_path, None) + monkeypatch.setattr(wl.os, "sysconf", lambda _: (_ for _ in ()).throw(OSError)) + assert wl.derived_ceiling(wl.LANES_BY_NAME["ml"]) == wl.UNKNOWN_CEILING + + +def test_cpu_lanes_are_bounded_by_cores_not_memory(monkeypatch, tmp_path): + """A CPU-bound lane on a memory-starved box must still get slots — its + slots are processes, not model copies.""" + _point_memory_at(monkeypatch, tmp_path, str(2 * wl.GIB)) + monkeypatch.setattr(wl, "container_cpu_count", lambda: 8) + assert wl.derived_ceiling(wl.LANES_BY_NAME["worker"]) == 8 + + +def test_cpu_quota_beats_host_core_count(monkeypatch, tmp_path): + """`os.cpu_count()` reports the HOST's cores from inside a container, so a + 4-core quota on a 32-core host would otherwise offer 32 slots. The + operator's own stack sets `cpus: '4.0'` on ml-worker, so this is real + configuration rather than a hypothetical.""" + quota = tmp_path / "cpu.max" + quota.write_text("400000 100000") + monkeypatch.setattr(wl, "_CGROUP_V2_CPU", quota) + monkeypatch.setattr(wl.os, "cpu_count", lambda: 32) + assert wl.container_cpu_count() == 4 + + +def test_a_single_core_box_still_gets_a_slot(monkeypatch): + monkeypatch.setattr(wl, "container_cpu_count", lambda: 1) + assert wl.derived_ceiling(wl.LANES_BY_NAME["worker"]) >= wl.MIN_CEILING + + +def test_the_ceiling_is_computed_not_stored(monkeypatch, tmp_path): + """The property the design rests on: the same lane yields a different + ceiling when the container's limits change, with no row edit. A stored + ceiling would keep authorising what the box no longer has.""" + lane = wl.LANES_BY_NAME["ml"] + _point_memory_at(monkeypatch, tmp_path, str(34 * wl.GIB)) + big = wl.derived_ceiling(lane) + (tmp_path / "memory.max").write_text(str(10 * wl.GIB)) + small = wl.derived_ceiling(lane) + assert big > small + + +def test_ceilings_covers_every_lane(): + assert set(wl.ceilings()) == {lane.name for lane in wl.LANES} + + +# --- the duplicate this step collapsed --------------------------------------- + + +def test_role_names_is_derived_from_the_lane_table(): + """`service_roster.ROLE_NAMES` was a hand-kept second copy of 'queue set + -> display name' and had already drifted: maintenance_long is a live lane + with four task routes and a dedicated worker, and the roster did not know + its name, so the System tab rendered `Worker (maintenance_long)`. + + Asserting the derivation rather than the contents — a test listing the + names again would be a third copy. + """ + from backend.app.services.service_roster import ROLE_NAMES, role_display_name + + assert ROLE_NAMES == {lane.queue_key: lane.display_name for lane in wl.LANES} + assert role_display_name(("maintenance_long",)) == "Long maintenance" + + +def test_an_unrecognised_queue_set_still_gets_a_true_label(): + """A deployment slicing CELERY_QUEUES differently must not be handed a + name this code invented for it.""" + from backend.app.services.service_roster import role_display_name + + assert role_display_name(("nonsense",)) == "Worker (nonsense)" + + +# --- the model's invariant --------------------------------------------------- + + +@pytest.mark.parametrize( + "slots,cap,ok", + [ + (0, 0, True), + (1, 1, True), + (1, 4, True), + (5, 4, False), # slots above its own cap + (-1, 1, False), # negative slots + (1, -1, False), # negative cap + ], +) +def test_worker_lane_check_constraints(slots, cap, ok): + """The constraints live in the database, not only in the service, because + a row violating `slots <= slots_cap` is not a rejected request — it is a + lane the reconcile (step 3) will drive UP to a number the operator + capped.""" + from backend.app.models import WorkerLane + + constraints = { + c.name: str(c.sqltext) for c in WorkerLane.__table__.constraints + if hasattr(c, "sqltext") + } + assert "slots_within_cap" in constraints + assert "slots_non_negative" in constraints + assert "cap_non_negative" in constraints + + # Evaluate the same predicates the database will, so the parametrize table + # documents what is accepted rather than restating the SQL. + satisfied = slots >= 0 and cap >= 0 and slots <= cap + assert satisfied is ok