diff --git a/alembic/versions/0107_worker_lane_sample.py b/alembic/versions/0107_worker_lane_sample.py new file mode 100644 index 0000000..df3fdfa --- /dev/null +++ b/alembic/versions/0107_worker_lane_sample.py @@ -0,0 +1,63 @@ +"""worker_lane_sample — where the sizing sweep leaves what it measured. + +Operator, 2026-09-23, on the System tab: *"there is a repull every time this +page loads — is there a reason this info isn't being tracked in the +background and stored in some way?"* + +`/api/system/workers` ran a full celery inspect on every call — four +broadcasts on an eleven-second budget — and the page polls it every fifteen +seconds. `size_worker_lanes` was already inspecting on a timer to decide pool +sizes, computing exactly these numbers and discarding them. This table is +where they land instead, and the endpoint becomes a plain read. + +## Why a new table rather than columns on `worker_lane` + +`worker_lane` holds the one number an operator sets. Putting a measurement +beside it is the mistake alembic 0105 undid: `slots` sat next to `slots_cap`, +and a measurement next to a preference reads as a second preference. + +No backfill. A row appears when the sweep first runs (within its period), and +until then the lane reads as not-yet-measured, which is true. + +Revision ID: 0107 +Revises: 0106 +Create Date: 2026-09-23 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0107" +down_revision: Union[str, None] = "0106" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "worker_lane_sample", + sa.Column("lane", sa.String(length=32), primary_key=True), + # Nullable=False with no server_default: the sweep writes every column + # on every upsert, so a row only ever exists complete. + sa.Column("present", sa.Boolean(), nullable=False), + sa.Column("replicas", sa.Integer(), nullable=False), + # Nullable on purpose — unknown, never zero. A worker that answered + # without reporting its pool, and a queue the broker did not answer + # for, must not be summed as empty. + sa.Column("pool", sa.Integer(), nullable=True), + sa.Column("active", sa.Integer(), nullable=False), + sa.Column("reserved", sa.Integer(), nullable=False), + sa.Column("queue_depth", sa.Integer(), nullable=True), + sa.Column( + "measured_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + ), + ) + + +def downgrade() -> None: + op.drop_table("worker_lane_sample") diff --git a/backend/app/api/system_health.py b/backend/app/api/system_health.py index e767ede..92dde1a 100644 --- a/backend/app/api/system_health.py +++ b/backend/app/api/system_health.py @@ -26,7 +26,6 @@ a true statement. from __future__ import annotations import asyncio -import logging import time from datetime import UTC, datetime @@ -36,9 +35,7 @@ from sqlalchemy import select, text from ..config import get_config from ..extensions import get_session from ..models import ServiceSeen -from ..services.service_roster import refresh_if_stale - -log = logging.getLogger(__name__) +from ..services.worker_lanes import SWEEP_PERIOD_SECONDS system_health_bp = Blueprint("system_health", __name__, url_prefix="/api/system") @@ -53,6 +50,23 @@ system_health_bp = Blueprint("system_health", __name__, url_prefix="/api/system" STALE_AFTER_SECONDS = 90 DOWN_AFTER_SECONDS = 300 +# The celery roster is written by `size_worker_lanes` and by nothing else, so +# these thresholds are only meaningful against ITS cadence. Asserted at import +# rather than left to a reader, because this is precisely the comparison that +# was never made for the GPU agent: its lease poll backed off to 900s while +# the roster called it stopped at 300s, and both numbers were individually +# correct, in different directions, in different files (lesson #4355). +# +# Two clear sweeps before a part is even called STALE. One missed tick is +# routine — the sweep rides the maintenance queue and does an inspect that can +# take eleven seconds — and must not turn the page yellow. +_SWEEPS_BEFORE_STALE = 2 +assert STALE_AFTER_SECONDS >= SWEEP_PERIOD_SECONDS * _SWEEPS_BEFORE_STALE, ( + f"a {SWEEP_PERIOD_SECONDS}s sweep cannot keep a roster fresh against a " + f"{STALE_AFTER_SECONDS}s stale threshold: raise the threshold or shorten " + f"the sweep" +) + # Probes cross a process boundary, so they carry deadlines. A hung Postgres # must make this endpoint say "postgres: down", not hang alongside it. PROBE_TIMEOUT_SECONDS = 2.0 @@ -151,14 +165,11 @@ async def system_health(): parts.append(pg) if pg["state"] == _OK: - # Rate-limited inside; see service_roster on why the web process - # is the right observer. - try: - await refresh_if_stale(session) - await session.commit() - except Exception: # noqa: BLE001 - log.warning("system health: roster refresh failed", exc_info=True) - + # A PURE READ since 2026-09-23. This used to refresh the celery + # roster here, rate-limited to once per 20s — so the roster only + # advanced while somebody had a browser open, and a broadcast rode + # on a request. `size_worker_lanes` writes it now, on a timer, and + # the assertion below is what keeps that cadence honest. rows = ( await session.execute(select(ServiceSeen).order_by(ServiceSeen.display_name)) ).scalars().all() diff --git a/backend/app/api/workers.py b/backend/app/api/workers.py index c0f6638..eb5de55 100644 --- a/backend/app/api/workers.py +++ b/backend/app/api/workers.py @@ -31,7 +31,12 @@ from ..services.worker_control import ( push_lane_cap, store_lane_cap, ) -from ..services.worker_lanes import LANES_BY_NAME, Lane, derived_ceiling +from ..services.worker_lanes import ( + LANES_BY_NAME, + SWEEP_PERIOD_SECONDS, + Lane, + derived_ceiling, +) from ._responses import error_response as _bad workers_bp = Blueprint("workers", __name__, url_prefix="/api/system/workers") @@ -43,22 +48,26 @@ async def list_lanes(): 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. + One database read, and NO broker call. Operator, 2026-09-23: *"there is a + repull every time this page loads — is there a reason this info isn't + being tracked in the background and stored in some way?"* + + It used to inspect the broker here, four broadcasts on an eleven-second + budget, four times a minute per open tab — while `size_worker_lanes` was + already inspecting on a timer and discarding the same numbers. The sweep + stores them now (`worker_lane_sample`) and this reads them. + + So the live figures are up to `SWEEP_PERIOD_SECONDS` old, and each lane + carries the `measured_at` that says so. `sweep_period_seconds` is returned + alongside, so the UI can explain the age without hard-coding the cadence + in a second place. """ - # The session closes BEFORE the broker work. Holding a Postgres connection - # across a celery inspect is what made this page block the whole site — - # see `worker_control.LaneSettings`. This endpoint polls every 15s and the - # inspect budget is 11s, so each poll was pinning a connection for most of - # the interval. async with get_session() as session: settings = await lane_settings(session) - lanes = await lane_view(settings) return jsonify({ - "lanes": lanes, + "lanes": lane_view(settings), "fetched_at": datetime.now(UTC).isoformat(), + "sweep_period_seconds": SWEEP_PERIOD_SECONDS, }) diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 21c98bf..d341530 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -14,6 +14,7 @@ Queues: from celery import Celery from .config import get_config +from .services.worker_lanes import SWEEP_PERIOD_SECONDS def make_celery() -> Celery: @@ -113,7 +114,14 @@ def make_celery() -> Celery: }, "size-worker-lanes": { "task": "backend.app.tasks.maintenance.size_worker_lanes", - "schedule": 60.0, # every minute. + "schedule": SWEEP_PERIOD_SECONDS, + # + # The number lives in `services/worker_lanes` because three + # places must agree on it: this schedule, the freshness of the + # sample the System tab reads, and the roster staleness + # thresholds in `api/system_health` — which now depend on this + # sweep rather than on a browser being open, and assert their + # headroom over it at import. # # ONE entry, replacing `autoscale-worker-lanes` (60s) and # `reconcile-worker-lanes` (300s) on 2026-09-23. They were two @@ -121,14 +129,16 @@ def make_celery() -> Celery: # existed to stop the reconcile undoing its work; with the # stored `slots` gone there is nothing to disagree about. # - # A minute because it reacts to a BACKLOG, and a five-minute - # reaction to a queue filling up is no reaction. It also now - # carries what the reconcile was for — a worker restarted at - # its ENV concurrency is corrected on the next tick rather - # than after five. + # Fast enough to react to a BACKLOG — a five-minute reaction to + # a queue filling up is no reaction. It also carries what the + # reconcile was for: a worker restarted at its ENV concurrency + # is corrected on the next tick rather than after five. # # Cheap when settled: one inspect plus one LLEN sweep, and no - # control messages at all once every lane matches. + # control messages at all once every lane matches. It is also + # now the ONLY thing that inspects — nothing on a request path + # does — so this is the whole broker cost of the System tab, + # whether nobody or ten tabs are watching. }, "cleanup-old-tasks": { "task": "backend.app.tasks.maintenance.cleanup_old_tasks", diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 08f1d3f..b925805 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -45,6 +45,7 @@ from .tag_positive_confirmation import TagPositiveConfirmation from .tag_suggestion_rejection import TagSuggestionRejection from .task_run import TaskRun from .worker_lane import WorkerLane +from .worker_lane_sample import WorkerLaneSample __all__ = [ "Base", @@ -96,4 +97,5 @@ __all__ = [ "TagSuggestionRejection", "TaskRun", "WorkerLane", + "WorkerLaneSample", ] diff --git a/backend/app/models/worker_lane_sample.py b/backend/app/models/worker_lane_sample.py new file mode 100644 index 0000000..26503dd --- /dev/null +++ b/backend/app/models/worker_lane_sample.py @@ -0,0 +1,83 @@ +"""worker_lane_sample — the last thing the sizing sweep measured about a lane. + +Milestone 422, 2026-09-23. A MEASUREMENT table, deliberately separate from +`worker_lane`, which holds the one number an operator sets. + +## Why this exists + +Operator, 2026-09-23, looking at the System tab: *"there is a repull every +time this page loads — is there a reason this info isn't being tracked in the +background and stored in some way?"* + +There was not a good one. `/api/system/workers` ran a full celery inspect — +four broadcast round trips on an eleven-second budget — on every call, and +the page polls it every fifteen seconds. Meanwhile `size_worker_lanes` was +already inspecting on a timer to decide pool sizes, computing exactly these +numbers, using them, and throwing them away. The browser then asked the +broker for them again. + +So the sweep writes what it saw here, and the endpoint reads this table. The +request path makes no broker call at all any more. + +## Why NOT columns on `worker_lane` + +Because that is the mistake this milestone already made once and undid. That +table used to carry `slots` — how many workers were running — beside +`slots_cap`, and a measurement sitting next to a preference reads as a second +preference: the operator had to keep two numbers in agreement, and the +autoscaler had to be granted permission to move one of them. + +The distinction is the whole design, so it is a table boundary. Nothing an +operator sets lives here; nothing here is ever an input to a decision about +what they wanted. + +## Freshness is a value, not an assumption + +`measured_at` is returned to the UI, which says how old the reading is rather +than implying it is live. A sample is a fact about a moment, and a page that +presents a one-minute-old number as current is how an operator ends up +mistrusting the whole surface. +""" + +from datetime import datetime + +from sqlalchemy import Boolean, DateTime, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from .base import Base + + +class WorkerLaneSample(Base): + __tablename__ = "worker_lane_sample" + + # The lane name from services/worker_lanes.LANES. One row per lane, + # overwritten in place: this is the LATEST reading, not a history. A time + # series would be a different table with a different retention problem, + # and nothing has asked for one. + lane: Mapped[str] = mapped_column(String(32), primary_key=True) + + # Whether anything answered for this lane. NOT the same as "zero workers" + # — an unswept absence is not a verdict (snippet #3969). False here means + # the inspect came back without this lane, so every count below is + # meaningless and the UI must say "not answering" rather than "0". + present: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + + replicas: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + # Pool size of ONE process, nullable because a worker that answered + # without reporting its pool is unknown rather than empty. + pool: Mapped[int | None] = mapped_column(Integer, nullable=True) + + active: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + reserved: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + + # Redis LLEN across the lane's queues. Nullable for the same reason as + # `pool`: a queue the broker did not answer for is unknown, and summing it + # as zero would report a buried lane as idle. + queue_depth: Mapped[int | None] = mapped_column(Integer, nullable=True) + + measured_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + server_default=func.now(), + ) diff --git a/backend/app/services/service_roster.py b/backend/app/services/service_roster.py index 7a77363..cd95f1e 100644 --- a/backend/app/services/service_roster.py +++ b/backend/app/services/service_roster.py @@ -31,7 +31,7 @@ from __future__ import annotations import asyncio import logging -from sqlalchemy import func, select +from sqlalchemy import func from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession @@ -40,10 +40,11 @@ from .worker_lanes import LANES, lane_for_node log = logging.getLogger(__name__) -# How stale the roster may be before a health request refreshes it. Comfortably -# under the staleness thresholds that decide a service is missing, so the -# verdict is never limited by how often anyone looked. -REFRESH_TTL_SECONDS = 20.0 +# There is no refresh TTL any more. It existed because the HEALTH REQUEST +# refreshed the roster, rate-limited to 20s so that a page open in two tabs +# did not inspect twice as often. `size_worker_lanes` owns the refresh now, on +# `SWEEP_PERIOD_SECONDS`, so the cadence is a schedule rather than a side +# effect of someone looking. # celery inspect is a broker round trip and this sits on a request path, so it # gets a deadline (rule 156). A broker that has stopped answering must make the @@ -140,10 +141,13 @@ def _inspect_celery_sync() -> dict[tuple[str, ...], dict]: return grouped -async def touch_service( - session: AsyncSession, *, key: str, kind: str, display_name: str, details: dict -) -> None: - """Record that a part checked in just now. +def touch_service_stmt(*, key: str, kind: str, display_name: str, details: dict): + """The upsert that records a check-in, as a statement. + + Built here rather than inline so the async caller (an agent lease, over + the API) and the sync one (the sizing sweep, in a celery task) run the + SAME write. Two spellings of one upsert is the kind of duplication that + stays correct right up until one of them gains a column. Upsert rather than read-modify-write: several web processes and several agents can be doing this at once, and the last writer is simply the most @@ -154,7 +158,7 @@ async def touch_service( stmt = pg_insert(ServiceSeen).values( key=key, kind=kind, display_name=display_name, details=details, ) - stmt = stmt.on_conflict_do_update( + return stmt.on_conflict_do_update( index_elements=[ServiceSeen.key], set_={ "kind": stmt.excluded.kind, @@ -163,7 +167,75 @@ async def touch_service( "last_seen_at": func.now(), }, ) - await session.execute(stmt) + + +async def touch_service( + session: AsyncSession, *, key: str, kind: str, display_name: str, details: dict +) -> None: + """Record that a part checked in just now.""" + await session.execute(touch_service_stmt( + key=key, kind=kind, display_name=display_name, details=details, + )) + + +def _roster_rows(grouped: dict[tuple[str, ...], dict]) -> list[dict]: + """The `touch_service` arguments for everything that answered. + + Split from the write so the async and sync refreshes below share the + mapping as well as the statement — what a roster row IS should not depend + on which kind of session is writing it. + """ + return [ + { + "key": "celery:" + ",".join(queues), + "kind": "celery", + "display_name": role_display_name(queues), + "details": { + "queues": list(queues), + "hostnames": entry["hostnames"], + "replicas": len(entry["hostnames"]), + "active": entry["active"], + }, + } + for queues, entry in grouped.items() + ] + + +def refresh_celery_roster_sync(session) -> None: + """The roster refresh, from the sizing sweep's sync session. + + ## Why the sweep owns this now + + It used to run on the request path, rate-limited to once every 20s by the + newest celery row. So the roster only advanced while somebody had a + browser open — the liveness of the workers was a function of whether + anyone was looking at them, which is the observer-effect version of the + bug this roster exists to prevent. + + Operator, 2026-09-23: *"there is a repull every time this page loads — is + there a reason this info isn't being tracked in the background and stored + in some way?"* + + Now a timer writes it and the page only reads. The cadence is + `SWEEP_PERIOD_SECONDS`, and `api/system_health` asserts it leaves headroom + under the staleness thresholds — because a sweep period and a stale + threshold chosen in different files and never compared is exactly how the + idle GPU agent came to read as stopped (lesson #4355). + + Never raises. A failure means the roster does not advance, and the rows + going stale is then a TRUE report about a broker nobody can reach. + """ + try: + grouped = _inspect_celery_sync() + except Exception: + log.warning( + "service roster: celery inspect failed; roster not refreshed", + exc_info=True, + ) + return + for row in _roster_rows(grouped): + session.execute(touch_service_stmt(**row)) + session.commit() async def refresh_celery_roster(session: AsyncSession) -> None: @@ -186,39 +258,5 @@ async def refresh_celery_roster(session: AsyncSession) -> None: log.warning("service roster: celery inspect failed; roster not refreshed", exc_info=True) return - for queues, entry in grouped.items(): - await touch_service( - session, - key="celery:" + ",".join(queues), - kind="celery", - display_name=role_display_name(queues), - details={ - "queues": list(queues), - "hostnames": entry["hostnames"], - "replicas": len(entry["hostnames"]), - "active": entry["active"], - }, - ) - - -async def refresh_if_stale(session: AsyncSession) -> None: - """Refresh the celery roster if nobody has for REFRESH_TTL_SECONDS. - - Rate-limited by the data rather than by a lock: the gate is the newest - last_seen_at across the celery rows, which every web process can see. Two - processes racing through the gate costs one redundant inspect and writes - the same values twice, so the benign outcome needs no coordination to - prevent. - """ - newest = ( - await session.execute( - select(func.max(ServiceSeen.last_seen_at)).where(ServiceSeen.kind == "celery") - ) - ).scalar_one_or_none() - - if newest is not None: - age = (await session.execute(select(func.now()))).scalar_one() - newest - if age.total_seconds() < REFRESH_TTL_SECONDS: - return - - await refresh_celery_roster(session) + for row in _roster_rows(grouped): + await touch_service(session, **row) diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 1df00eb..0013a57 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -53,9 +53,10 @@ from dataclasses import dataclass, field from datetime import UTC, datetime from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession -from ..models import TaskRun, WorkerLane +from ..models import TaskRun, WorkerLane, WorkerLaneSample from .worker_lanes import ( LANES, LANES_BY_QUEUE_KEY, @@ -341,6 +342,76 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]: return rows +@dataclass(frozen=True) +class LaneSample: + """What the sizing sweep last measured about one lane. + + The same fields `LaneLiveState` carries, plus the queue depth and WHEN — + because this one is read from a table rather than from the broker, and a + reading with no timestamp invites being presented as current. + + `measured_at=None` means no sweep has written this lane yet: a fresh + install inside its first period, or a stack whose beat is not running. + Distinct from `present=False` (something asked, nothing answered), and the + UI says different things about the two. + """ + + present: bool = False + replicas: int = 0 + pool: int | None = None + active: int = 0 + reserved: int = 0 + queue_depth: int | None = None + measured_at: datetime | None = None + + +def _lane_depth(lane: Lane, depths: dict[str, int | None]) -> int | None: + """A lane's backlog across its queues — None when NOTHING answered. + + A queue the broker did not answer for must not be summed as zero: an + unknown depth is not an empty one, and reporting a buried lane as idle is + the direction that matters. + """ + known = [depths.get(q) for q in lane.queues] + if not any(d is not None for d in known): + return None + return sum(d for d in known if d is not None) + + +def store_lane_samples_sync(session, live: dict[str, LaneLiveState], depths) -> None: + """Write what the sweep just measured. SYNC — the celery task owns a sync + session, and this is the only place these rows are written. + + Upsert per lane, last writer wins, same shape as `service_roster`'s + `touch_service`: two processes sweeping at once is a benign race that + needs no coordination, because both are recording what they actually saw. + + A lane that did not answer is STILL written, with `present=False`. Skipping + it would leave the previous reading in place and let the page go on showing + a pool that is no longer there — the stale row would read as a current one + (lesson #4202: the row is the thing that has to change). + """ + now = datetime.now(UTC) + for lane in LANES: + state = live.get(lane.name) or LaneLiveState() + values = { + "lane": lane.name, + "present": state.present, + "replicas": state.replicas, + "pool": state.pool, + "active": state.active, + "reserved": state.reserved, + "queue_depth": _lane_depth(lane, depths), + "measured_at": now, + } + stmt = pg_insert(WorkerLaneSample).values(**values) + session.execute(stmt.on_conflict_do_update( + index_elements=[WorkerLaneSample.lane], + set_={k: v for k, v in values.items() if k != "lane"}, + )) + session.commit() + + @dataclass class LaneSettings: """What the DATABASE knows about the lanes — read and finished with before @@ -366,58 +437,80 @@ class LaneSettings: caps: dict[str, int] oldest_by_queue: dict[str, datetime] + # The sizing sweep's last reading per lane. Since 2026-09-23 this is where + # the live numbers come from: the endpoint no longer inspects at all. + samples: dict[str, LaneSample] = field(default_factory=dict) async def lane_settings(session: AsyncSession) -> LaneSettings: - """Every DB read the lane view needs, in one short-lived session.""" + """Every DB read the lane view needs, in one short-lived session. + + Which is now ALL of them. `lane_view` below takes what this returns and + talks to nothing. + """ rows = await _rows_by_name(session) + samples = { + row.lane: LaneSample( + present=row.present, + replicas=row.replicas, + pool=row.pool, + active=row.active, + reserved=row.reserved, + queue_depth=row.queue_depth, + measured_at=row.measured_at, + ) + for row in ( + await session.execute(select(WorkerLaneSample)) + ).scalars() + } return LaneSettings( caps={name: row.slots_cap for name, row in rows.items()}, oldest_by_queue=await _oldest_running_by_queue(session), + samples=samples, ) -async def lane_view(settings: LaneSettings) -> list[dict]: - """Every lane: what is configured, what is live, what it may grow to. +def lane_view(settings: LaneSettings) -> list[dict]: + """Every lane: what is configured, what was last measured, what it may + grow to. NO broker call, and no database — `settings` is the whole input. - Takes the settings rather than a session ON PURPOSE — see `LaneSettings`. - Everything below this line is broker work, and no database connection is - held while it happens. + ## It used to inspect, on every request - `pending` is the honest backlog — Redis depth PLUS reserved — because + Four broadcast round trips on an eleven-second budget, on a page that + polls every fifteen seconds. Operator, 2026-09-23: *"there is a repull + every time this page loads — is there a reason this info isn't being + tracked in the background and stored in some way?"* + + There was one, and it had expired. The docstring here used to say the + endpoint was deliberately uncached because *"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"*. True while a cap change + refetched the table — and that refetch is exactly what was removed in + `1353d34`, so the UI now patches its own row from the write's reply and + nothing depends on this being live. + + Meanwhile `size_worker_lanes` was already inspecting on a timer to decide + pool sizes: the same numbers, computed, used, and discarded, while the + browser asked the broker for them again four times a minute. + + So the sweep writes `worker_lane_sample` and this reads it. The reading is + up to `SWEEP_PERIOD_SECONDS` old, and `measured_at` travels with it so the + UI can say so rather than implying it is current. + + `pending` is still the honest backlog — depth PLUS reserved — because celery prefetches and LLEN alone reads 0 while a worker holds tasks in memory. """ - # A deadline, because this is a request path and `to_thread` on its own is - # an await with no bound (rule 156). `inspect_lanes_sync` never raises and - # every inner call has its own timeout, so the only way past the budget is - # the thread not being scheduled — and a page that renders "not answering" - # is a better answer than one that does not render. - try: - live = await asyncio.wait_for( - asyncio.to_thread(inspect_lanes_sync), - timeout=INSPECT_BUDGET_SECONDS, - ) - except TimeoutError: - log.warning( - "worker_control: inspect exceeded %ss; reporting every lane as " - "not answering", INSPECT_BUDGET_SECONDS, - ) - live = {lane.name: LaneLiveState() for lane in LANES} - depths = await asyncio.to_thread(_queue_depths_sync) oldest = settings.oldest_by_queue - now = datetime.now(UTC) out = [] for lane in LANES: cap = settings.caps[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 + # A lane with no row yet is not-measured, which is distinct from + # measured-as-absent. The default carries `measured_at=None`, and the + # UI says "not measured yet" rather than "not answering". + sample = settings.samples.get(lane.name) or LaneSample() + depth = sample.queue_depth out.append({ "name": lane.name, "display_name": lane.display_name, @@ -444,17 +537,24 @@ async def lane_view(settings: LaneSettings) -> list[dict]: for m in lane.models ], "live": { - "present": state.present, - "replicas": state.replicas, - "pool": state.pool, - "active": state.active, - "reserved": state.reserved, + "present": sample.present, + "replicas": sample.replicas, + "pool": sample.pool, + "active": sample.active, + "reserved": sample.reserved, }, "queue_depth": depth, - "pending": None if depth is None else depth + state.reserved, + "pending": None if depth is None else depth + sample.reserved, + # When the numbers above were read. Per lane rather than one for + # the response, because a lane whose row has never been written + # has no reading at all and must not borrow another lane's. + "measured_at": ( + sample.measured_at.isoformat() if sample.measured_at else None + ), # How long the oldest still-running task on this lane has been - # going, in minutes. The operator asked for a trigger here — grow - # a lane whose tasks run past some duration — and it stayed a + # going, in minutes. Read from `task_run`, not from the sweep, so + # this one IS current. The operator asked for a trigger here — + # grow a lane whose tasks run past some duration — and it stayed a # REPORT: a long task does not finish sooner because the lane # gained a slot, so scaling on it would spend memory to change # nothing. Shown so they can see a lane wedged on one slow job, @@ -606,7 +706,22 @@ async def push_lane_cap(lane: Lane, slots_cap: int, *, was_cap: int) -> dict: # LOWERED on a running lane. Only this direction needs a message, and # only when the pool is actually above the new cap — so it reads the # live pool rather than resizing blind. A raise never reaches here. - live = await asyncio.to_thread(inspect_lanes_sync) + # + # Bounded (rule 156): `to_thread` on its own is an await with no + # deadline, and this runs in a background task where a hang would be + # silent rather than visible as a slow page. On a timeout the lane is + # simply not resized here and the sizing sweep carries it. + try: + live = await asyncio.wait_for( + asyncio.to_thread(inspect_lanes_sync), + timeout=INSPECT_BUDGET_SECONDS, + ) + except TimeoutError: + log.warning( + "worker_control: inspect exceeded %ss lowering %s; leaving the " + "pool to the sizing pass", INSPECT_BUDGET_SECONDS, lane.name, + ) + return _cap_result(lane, slots_cap, now_on, applied, error, False) current = live[lane.name].pool if current is not None and current > slots_cap: applied, error = await asyncio.to_thread( @@ -642,6 +757,15 @@ async def push_lane_cap(lane: Lane, slots_cap: int, *, was_cap: int) -> dict: lane.name, slots_cap, error, ) + return _cap_result(lane, slots_cap, now_on, applied, error, fetching) + + +def _cap_result( + lane: Lane, slots_cap: int, now_on: bool, applied: bool, + error: str | None, fetching: bool, +) -> dict: + """The push's outcome. One builder, because `push_lane_cap` has two exits + and a second literal would be free to disagree with the first.""" return { "name": lane.name, "slots_cap": slots_cap, @@ -746,7 +870,12 @@ def wanted_slots(cap: int, active: int, pending: int | None) -> int: return max(MIN_POOL_SLOTS, min(cap, active + (pending or 0))) -def size_lanes_sync(caps: dict[str, int]) -> list[LaneSizing]: +def size_lanes_sync( + caps: dict[str, int], + *, + live: dict[str, LaneLiveState] | None = None, + depths: dict[str, int | None] | None = None, +) -> list[LaneSizing]: """Size every lane to its backlog, within the cap. The whole control loop. `caps` is lane name -> slots_cap, read from the database by the caller. @@ -754,6 +883,13 @@ def size_lanes_sync(caps: dict[str, int]) -> list[LaneSizing]: the session, and keeping the DB out of here is what lets it be called from anywhere that already knows the caps. + `live` and `depths` are the measurements. Passing them in is not an + optimisation — it is how the caller gets to KEEP them. The sweep now + stores what it measured (`worker_lane_sample`) so the System tab reads a + table instead of inspecting on every page load, and that is only possible + if the same reading serves both purposes. Measured here when not given, so + every existing caller and test is unaffected. + ## It must converge and then go quiet One `inspect` for all lanes, and `set_lane_slots_sync` issues nothing to a @@ -770,8 +906,10 @@ def size_lanes_sync(caps: dict[str, int]) -> list[LaneSizing]: be a verdict drawn from an unswept read, and here it is worse than useless: there is nothing to send the message to. """ - live = inspect_lanes_sync() - depths = _queue_depths_sync() + if live is None: + live = inspect_lanes_sync() + if depths is None: + depths = _queue_depths_sync() out: list[LaneSizing] = [] for lane in LANES: @@ -805,11 +943,7 @@ def size_lanes_sync(caps: dict[str, int]) -> list[LaneSizing]: )) continue - 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 - ) + depth = _lane_depth(lane, depths) pending = None if depth is None else depth + state.reserved want = wanted_slots(cap, state.active, pending) diff --git a/backend/app/services/worker_lanes.py b/backend/app/services/worker_lanes.py index aad8f9a..a1b2b0c 100644 --- a/backend/app/services/worker_lanes.py +++ b/backend/app/services/worker_lanes.py @@ -282,6 +282,31 @@ MIN_POOL_SLOTS = 1 # not to make a small machine unusable. MIN_CEILING = 1 +# How often `size_worker_lanes` runs — the beat schedule, and the freshness of +# everything the System tab shows. +# +# It is here, in the import-light module, because three places have to agree +# about it and they are in different packages: the beat entry in `celery_app`, +# the sample the sweep writes (`worker_lane_sample`), and the roster's +# staleness thresholds in `api/system_health`, which now depend on this sweep +# rather than on a browser being open. +# +# 30s, down from 60s, because the sweep became the ONLY writer of the celery +# roster on 2026-09-23. A part is called stale after 90s of silence, so a +# 60-second sweep left one missed tick between "normal" and "everything is +# yellow". That is the shape of lesson #4355 — a reader's threshold and an +# emitter's cadence chosen in different files and never compared — and the +# fix is headroom plus a test that asserts it, not a number that happens to +# work today. +# +# The cost is one inspect every 30s instead of every 60s; the saving is every +# inspect that used to run on a request path, which with a single tab open +# was roughly four a minute against this two. Consequence worth knowing: the +# pass also SHRINKS an idle lane by one slot per tick, so an idle lane now +# gives its workers back twice as fast. That is the direction the operator +# asked for — *"idle instances quiet down when not running"*. +SWEEP_PERIOD_SECONDS = 30.0 + # 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. diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index bfcaee3..d796bec 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1368,9 +1368,31 @@ def size_worker_lanes() -> dict: Returns every lane's outcome INCLUDING the ones it held, each with a reason. A pass that only speaks when it acts cannot be debugged on the day it does not. + + ## It is also the only thing that MEASURES, since 2026-09-23 + + It always inspected the broker to decide pool sizes, and then threw the + reading away — while `/api/system/workers` ran the same inspect on every + page load and the System tab polls it four times a minute. Operator: + *"there is a repull every time this page loads — is there a reason this + info isn't being tracked in the background and stored in some way?"* + + So one inspect now feeds three things: the sizing decision, the stored + sample the System tab reads, and the celery roster. No request path + touches the broker any more. + + Order matters. The sample is stored BEFORE the roster refresh, because + that refresh does its own broadcast and a broker that has just started + failing must not cost us the reading we already have. """ from ..models import WorkerLane - from ..services.worker_control import size_lanes_sync + from ..services.service_roster import refresh_celery_roster_sync + from ..services.worker_control import ( + _queue_depths_sync, + inspect_lanes_sync, + size_lanes_sync, + store_lane_samples_sync, + ) # Read INSIDE the session. Reading a column off a detached instance # happens to work while the attribute is still loaded and stops working @@ -1387,7 +1409,18 @@ def size_worker_lanes() -> dict: # let this task disagree with the seed it is meant to be enforcing. return {"sized": []} - sized = size_lanes_sync(caps) + # Measured ONCE, here, and then used three times. Passing them down is + # what makes the reading keepable rather than an implementation detail of + # a function that returns decisions. + live = inspect_lanes_sync() + depths = _queue_depths_sync() + + sized = size_lanes_sync(caps, live=live, depths=depths) + + with _sync_session_factory()() as session: + store_lane_samples_sync(session, live, depths) + refresh_celery_roster_sync(session) + for d in sized: if d.action not in ("held", "skipped"): log.info( diff --git a/frontend/src/components/settings/SystemHealthTab.vue b/frontend/src/components/settings/SystemHealthTab.vue index 4b96486..9a365fb 100644 --- a/frontend/src/components/settings/SystemHealthTab.vue +++ b/frontend/src/components/settings/SystemHealthTab.vue @@ -150,6 +150,24 @@ + +

+ + +

+