feat: worker lanes become rows — slots, a settable cap, a derived ceiling (4291)
CI / lint (push) Failing after 2s
CI / extension-version (push) Successful in 2s
Build images / sign-extension (push) Successful in 4s
Build images / build-agent (push) Successful in 5s
CI / frontend-build (push) Successful in 24s
CI / backend-lint-and-test (push) Failing after 32s
Build images / build-web (push) Successful in 58s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 1m45s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m13s

Milestone 422 step 1. The data model the rest of the milestone reads. No
behaviour change: nothing consumes these rows yet, and every lane still boots
at its CELERY_CONCURRENCY env value.

Three numbers, not two, per the operator's distinction — the derived value is
a cap ON the cap:

    slots  <=  slots_cap  <=  derived_ceiling
    (live)     (operator)     (computed)

They can always lower their own cap; they cannot raise it past what the
container can hold. The ceiling is never stored, so a row written on a 32GB
host and later run in a 4GB container is bounded by the 4GB.

`services/worker_lanes.py` is the one place that knows the lane set.
`models/worker_lane.py` holds only what an operator may change.

Two deviations from the step as written, both deliberate:

QUEUES ARE NOT A COLUMN. The step body said the row carries its `-Q` list,
but a lane's queues are decided by celery_app's task_routes, not by
preference — an operator cannot move a backup off maintenance_long. Storing
them would create a row that can contradict the routing table, with nothing
to notice until a queue had no consumer. So queues are code, slots are data.
`test_every_routed_queue_has_a_lane_that_serves_it` reads the real routing
table and fails if a route is ever added without a lane.

ROLE_NAMES IS NOW DERIVED, not left alone. It 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 in the operator's
stack, and the roster did not know its name — so the System tab labelled it
`Worker (maintenance_long)`. Adding a lane table beside it would have made
three copies.

The ceiling honours cgroup limits rather than the host's. `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 — and the operator's own stack
sets `cpus: '4.0'` on ml-worker, so that is real configuration, not a
hypothetical. Memory reads cgroup v2 then v1, and recognises v1's
PAGE_SIZE-aligned LONG_MAX sentinel by magnitude rather than treating it as
petabytes.

Every uncertain case fails LOW. An unreadable limit yields UNKNOWN_CEILING,
never unlimited — not knowing how much memory there is must not read as
plenty. A box too small to hold one model beside the web process gets an ML
ceiling of 0 rather than a floor of 1: offering a slot that OOMs the
container the first time it is used is exactly what this exists to prevent.

ML_BYTES_PER_SLOT is 4 GiB and is UNMEASURED — flagged as such in the code,
with the method for replacing it with a real figure. It decides whether a
stranger's server survives enabling tagging, so it errs toward refusing a
slot that would have fitted.

Seeded one-of-each with ml at 0 and disabled (alembic 0103). ML off is 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. The seed values are literals rather than an import
of LANES — a migration is a statement about one moment, and importing the
live defaults would silently change what this revision does on a fresh
database in 2027.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR
This commit is contained in:
2026-09-22 07:48:25 -04:00
co-authored by Claude Opus 5
parent aa2bb665b9
commit 84f13135ce
6 changed files with 759 additions and 5 deletions
@@ -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")
+2
View File
@@ -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",
]
+86
View File
@@ -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@<container id>`, 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(),
)
+11 -5
View File
@@ -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
}
+281
View File
@@ -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@<container id>` 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/<child pid>/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}
+258
View File
@@ -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