diff --git a/backend/app/api/workers.py b/backend/app/api/workers.py index 10c03ee..78f7f1c 100644 --- a/backend/app/api/workers.py +++ b/backend/app/api/workers.py @@ -23,7 +23,13 @@ from datetime import UTC, datetime from quart import Blueprint, jsonify, request from ..extensions import get_session -from ..services.worker_control import LaneUpdateRefused, lane_view, set_lane +from ..services.worker_control import ( + LaneUpdateRefused, + lane_settings, + lane_view, + push_lane_cap, + store_lane_cap, +) from ..services.worker_lanes import LANES_BY_NAME from ._responses import error_response as _bad @@ -41,8 +47,14 @@ async def list_lanes(): reply would show them the value from before their own change and read as the control having failed. """ + # 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: - lanes = await lane_view(session) + settings = await lane_settings(session) + lanes = await lane_view(settings) return jsonify({ "lanes": lanes, "fetched_at": datetime.now(UTC).isoformat(), @@ -85,9 +97,13 @@ async def update_lane(name: str): if not isinstance(value, int) or isinstance(value, bool): return _bad("invalid_body", detail="slots_cap must be an integer") + # Store, close the session, THEN push. Same reason as the GET above, and + # more sharply here: a cap change could do three broker round trips, all + # of them previously with a connection held. async with get_session() as session: try: - result = await set_lane(session, lane, slots_cap=value) + was_cap = await store_lane_cap(session, lane, value) except LaneUpdateRefused as exc: return _bad("refused", detail=str(exc)) + result = await push_lane_cap(lane, value, was_cap=was_cap) return jsonify(result) diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 641c545..88d599f 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -334,14 +334,53 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]: return rows -async def lane_view(session: AsyncSession) -> list[dict]: +@dataclass +class LaneSettings: + """What the DATABASE knows about the lanes — read and finished with before + anything touches the broker. + + This exists because holding a Postgres connection across a celery round + trip is what made the System tab block the whole site (operator, + 2026-09-23: *"something about changing the cap number is blocking to the + website"*). + + `lane_view` used to take the session and keep it open through an inspect + whose budget is eleven seconds — and that page polls every fifteen. With a + lane not answering, every inspect ran to nearly its full budget, so each + poll pinned a connection for ten seconds. SQLAlchemy's default pool is + five connections plus ten overflow; a couple of browser tabs, the health + endpoint doing the same thing, and a cap change adding two more inspects + exhausts that, and every OTHER request then waits on a connection. + + Nothing was slow in itself. The slowness was a scarce resource held across + it, which is why it surfaced as the whole site stalling rather than as one + slow page. + """ + + caps: dict[str, int] + oldest_by_queue: dict[str, datetime] + + +async def lane_settings(session: AsyncSession) -> LaneSettings: + """Every DB read the lane view needs, in one short-lived session.""" + rows = await _rows_by_name(session) + return LaneSettings( + caps={name: row.slots_cap for name, row in rows.items()}, + oldest_by_queue=await _oldest_running_by_queue(session), + ) + + +async def lane_view(settings: LaneSettings) -> list[dict]: """Every lane: what is configured, what is live, what it may grow to. - One call rather than making the UI join three sources. `pending` is the - honest backlog — Redis depth PLUS reserved — because celery prefetches and - LLEN alone reads 0 while a worker holds tasks in memory. + 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. + + `pending` is the honest backlog — Redis depth PLUS reserved — because + celery prefetches and LLEN alone reads 0 while a worker holds tasks in + memory. """ - rows = await _rows_by_name(session) # 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 @@ -359,12 +398,12 @@ async def lane_view(session: AsyncSession) -> list[dict]: ) live = {lane.name: LaneLiveState() for lane in LANES} depths = await asyncio.to_thread(_queue_depths_sync) - oldest = await _oldest_running_by_queue(session) + oldest = settings.oldest_by_queue now = datetime.now(UTC) out = [] for lane in LANES: - row = rows[lane.name] + 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. @@ -376,11 +415,11 @@ async def lane_view(session: AsyncSession) -> list[dict]: "name": lane.name, "display_name": lane.display_name, "queues": list(lane.queues), - "slots_cap": row.slots_cap, + "slots_cap": cap, "ceiling": derived_ceiling(lane), # DERIVED, never stored. A cap of zero means no consumers, so # "off" and "may use no workers" cannot disagree. - "enabled": row.slots_cap > 0, + "enabled": cap > 0, "memory_bound": lane.memory_bound, "optional": lane.optional, # What raising this lane's cap will download, so the UI can say @@ -485,30 +524,14 @@ class LaneUpdateRefused(ValueError): the UI shows — a greyed control with no explanation reads as a bug.""" -async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict: - """Store the operator's cap for `lane`, then make the live lane obey it. +async def store_lane_cap( + session: AsyncSession, lane: Lane, slots_cap: int, +) -> int: + """Validate and store the cap. Returns the PREVIOUS cap. DB only. - ONE value, since 2026-09-23. It used to take `slots`, `slots_cap`, - `enabled` and `autoscale`, which was four ways of saying two things — and - two of them were the caller's job to keep in agreement with each other. - - ## What happens live, and what does not - - A cap CHANGE is pushed immediately in one direction only: lowering it - shrinks the pool now, because a cap the operator just lowered should not - be exceeded for up to a minute. Raising it does NOT grow the pool here — - a cap is permission, not a request, and growing on permission would put - slots on a lane with nothing to do. The sizing pass adds them on its next - tick if there is work, which is the whole point of it being always on. - - Consumers follow the cap in both directions and immediately: zero means - off, and off must take effect when it is asked for. - - A failed PUSH is not a failed setting. The value is saved either way and - the sizing pass carries it within a minute; the result says `applied: - false` with a reason so the UI can say "saved, not yet live" rather than - "that didn't work" (lesson #4202 — a live change that does not survive, - with nothing saying so). + Split from the live push for the reason `LaneSettings` gives at length: a + Postgres connection must not be held across a celery round trip. Everything + here is fast and finished with before `push_lane_cap` starts. """ rows = await _rows_by_name(session) row = rows[lane.name] @@ -522,11 +545,40 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict f"({ceiling} for {lane.display_name})" ) - was_on = row.slots_cap > 0 + was_cap = row.slots_cap row.slots_cap = slots_cap await session.commit() + # The previous value, because the push needs the DIRECTION: lowering a cap + # has to reach the running lane now, and raising one has nothing to say. + return was_cap - now_on = slots_cap > 0 + +async def push_lane_cap(lane: Lane, slots_cap: int, *, was_cap: int) -> dict: + """Make the running lane obey a cap that is already stored. NO database. + + ## What is pushed, and what is not + + Consumers follow the cap immediately in BOTH directions: zero means off, + and off must take effect when it is asked for rather than up to a minute + later. + + The pool is only ever pushed DOWNWARD. Raising a cap is permission, not a + request — growing on permission would put workers on a lane with nothing + to do — so the sizing pass spends it on its next tick if there is work. + That also makes the common case (raising a cap) free: no broker round trip + AT ALL, which is the difference between a control that answers instantly + and one that takes ten seconds. Keyed on the previous cap rather than on + "is it on" — the first cut only knew on/off, so it inspected on every + raise to find out whether the pool needed lowering, and the control it was + meant to make instant still waited out an inspect. + + A failed push is not a failed setting. The value is already stored and the + sizing pass carries it within a minute; `applied: false` with a reason + lets the UI say "saved, not yet live" rather than "that didn't work" + (lesson #4202 — a live change that does not survive, with nothing saying + so). + """ + was_on, now_on = was_cap > 0, slots_cap > 0 applied, error = True, None if now_on != was_on: applied, error = await asyncio.to_thread(set_lane_enabled_sync, lane, now_on) @@ -536,9 +588,10 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict applied, error = await asyncio.to_thread( set_lane_slots_sync, lane, MIN_POOL_SLOTS, ) - elif applied and now_on: - # Only DOWNWARD. See the docstring: raising a cap is permission, and - # the sizing pass decides whether there is work to spend it on. + elif applied and now_on and slots_cap < was_cap: + # 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) current = live[lane.name].pool if current is not None and current > slots_cap: @@ -561,8 +614,8 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict return { "name": lane.name, - "slots_cap": row.slots_cap, - "ceiling": ceiling, + "slots_cap": slots_cap, + "ceiling": derived_ceiling(lane), "enabled": now_on, "applied": applied, "apply_error": error, diff --git a/tests/test_api_workers.py b/tests/test_api_workers.py index 5fc61e2..90bcace 100644 --- a/tests/test_api_workers.py +++ b/tests/test_api_workers.py @@ -255,3 +255,81 @@ async def _refreshed(db, name: str, expected: int) -> None: row = await _lane_row(db, name) await db.refresh(row) assert row.slots_cap == expected, "a refused write must store nothing" + + +# --- no database connection is held across a broker round trip --------------- + + +@pytest.mark.asyncio +async def test_the_lane_read_holds_no_session_while_it_inspects(monkeypatch): + """Operator, 2026-09-23: *"something about changing the cap number is + blocking to the website... it shouldn't be"*. + + Nothing here was slow in itself. A Postgres connection was held across a + celery inspect whose budget is eleven seconds, on a page that polls every + fifteen — so with a lane not answering, each poll pinned a connection for + most of the interval. SQLAlchemy's default pool is five plus ten overflow; + two browser tabs, the health endpoint doing the same, and a cap change + adding more inspects exhausts it, and every OTHER request then waits on a + connection. It surfaced as the whole site stalling rather than as one slow + page, which is why it took a screenshot to find. + + Asserted STRUCTURALLY rather than by timing: `lane_view` must not accept a + session at all. A timing test would be flaky, and a mock-call-order test + would pass against a version that took the session and merely used it + early — the property that matters is that it CANNOT. + """ + import inspect as _inspect + + from backend.app.services.worker_control import lane_settings, lane_view + + assert "session" not in _inspect.signature(lane_view).parameters, ( + "lane_view takes a session again; the broker work must run with none held" + ) + # And the DB half still exists, so the split did not simply lose the reads. + assert "session" in _inspect.signature(lane_settings).parameters + + +@pytest.mark.asyncio +async def test_the_cap_write_holds_no_session_while_it_pushes(): + """The same property on the write path, where it was worse: a cap change + could make three broker round trips, each with a connection held.""" + import inspect as _inspect + + from backend.app.services.worker_control import push_lane_cap, store_lane_cap + + assert "session" in _inspect.signature(store_lane_cap).parameters + assert "session" not in _inspect.signature(push_lane_cap).parameters, ( + "push_lane_cap takes a session again; the push must run with none held" + ) + + +@pytest.mark.asyncio +async def test_raising_a_cap_costs_no_broker_round_trip_at_all( + client, db, no_live_workers, monkeypatch, +): + """The common case must be instant. Raising a cap is permission, not a + request — the sizing pass spends it — so there is nothing to tell the + broker, and the operator's `+` should answer immediately rather than + waiting out an inspect.""" + from backend.app.services import worker_control as wc + + calls = [] + monkeypatch.setattr( + wc, "inspect_lanes_sync", lambda: calls.append("inspect") or {}, + ) + monkeypatch.setattr( + wc, "set_lane_slots_sync", + lambda *a, **k: calls.append("resize") or (True, None), + ) + monkeypatch.setattr( + wc, "set_lane_enabled_sync", + lambda *a, **k: calls.append("consumers") or (True, None), + ) + + await client.post("/api/system/workers/worker", json={"slots_cap": 1}) + calls.clear() + resp = await client.post("/api/system/workers/worker", json={"slots_cap": 6}) + + assert resp.status_code == 200 + assert calls == [], f"raising a cap talked to the broker: {calls}"