diff --git a/backend/app/api/workers.py b/backend/app/api/workers.py index 78f7f1c..c0f6638 100644 --- a/backend/app/api/workers.py +++ b/backend/app/api/workers.py @@ -19,8 +19,9 @@ second into the first would make a read-only module a write one. from __future__ import annotations from datetime import UTC, datetime +from functools import partial -from quart import Blueprint, jsonify, request +from quart import Blueprint, current_app, jsonify, request from ..extensions import get_session from ..services.worker_control import ( @@ -30,7 +31,7 @@ from ..services.worker_control import ( push_lane_cap, store_lane_cap, ) -from ..services.worker_lanes import LANES_BY_NAME +from ..services.worker_lanes import LANES_BY_NAME, Lane, derived_ceiling from ._responses import error_response as _bad workers_bp = Blueprint("workers", __name__, url_prefix="/api/system/workers") @@ -63,22 +64,38 @@ async def list_lanes(): @workers_bp.route("/", methods=["POST"]) async def update_lane(name: str): - """Set a lane's cap. Stores it, then makes the live lane obey it. + """Set a lane's cap. Stores it, answers, and makes the lane follow after. ONE field, since 2026-09-23. It used to take `slots`, `slots_cap`, `enabled` and `autoscale`; how many workers are running is now a measurement the sizing pass owns, and `enabled` is `cap > 0`. - Two failure kinds, deliberately different statuses: + ## The reply does not wait for the lane + + Operator, 2026-09-23: *"when the number is changed the change should be + queued so that it isn't blocking of the webui or the system itself. we + shouldn't have to wait for the validation live."* + + So the request does exactly one thing that can be slow — a row update — + and hands the broker work to a background task. Turning a lane off is + four `cancel_consumer` messages and a resize; lowering a cap is an + `inspect` on an eleven-second budget. Both used to happen between the + click and the response, with the stepper disabled the whole time. + + Nothing is lost by not waiting: the cap in the database is what the + system obeys, the sizing pass re-reads it every minute, and the table + polls, so the live columns catch up on their own. If the web process dies + before the background task runs, that sweep is the backstop — which is + the same guarantee the awaited version had, since a push could fail + there too. + + Refusals still happen inline, because they are decided from the value and + the machine's ceiling alone and never touch the broker: * **400** — the value is not allowed (negative, or above what this container can hold). Nothing was stored. The body carries `detail`, which is the sentence the UI shows; a refused control with no reason reads as a bug. - * **200 with `applied: false`** — the value WAS stored but could not be - pushed, because the lane is not currently answering. Not an error: the - sizing pass carries it within a minute, and the UI should say "saved, - not yet live" rather than "that didn't work". """ lane = LANES_BY_NAME.get(name) if lane is None: @@ -97,13 +114,48 @@ 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. + # Store, close the session, THEN hand off. The session must not be held + # across broker work — that is what made this page block the whole site + # (see `worker_control.LaneSettings`) — and now the request does not wait + # for that work either. async with get_session() as session: try: 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) + + _schedule_push(lane, value, was_cap) + + return jsonify({ + "name": lane.name, + "slots_cap": value, + "ceiling": derived_ceiling(lane), + "enabled": value > 0, + # The value is stored; the live lane is being told separately. The UI + # patches its row from this and lets the next poll bring the live + # columns, rather than refetching and paying for an inspect it just + # avoided. + "queued": True, + # Raising the cap off zero is what downloads the model (step 6), and + # the background task does it. Reported here so the UI can say a + # download has started rather than leaving the operator to wonder why + # a lane they just turned on is busy. + "fetching_models": value > 0 and was_cap == 0 and bool(lane.models), + }) + + +def _schedule_push(lane: Lane, slots_cap: int, was_cap: int) -> None: + """Run the live push after the response has gone out. + + A seam, not an abstraction: it is one call, and it exists so the tests can + hold the push still — a background task that outlived a test's patches + would reach the real broker during teardown. + + Quart tracks the task on the app and awaits it at shutdown, so an + in-flight push survives a graceful restart. `partial` rather than passing + `was_cap=` through `add_background_task`, so nothing depends on how that + forwards keyword arguments. + """ + current_app.add_background_task( + partial(push_lane_cap, lane, slots_cap, was_cap=was_cap) + ) diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index c4c1d7d..1df00eb 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -563,6 +563,13 @@ async def store_lane_cap( 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. + Runs OFF the request path since 2026-09-23 — the endpoint stores the cap, + answers, and hands this to a background task (operator: *"the change + should be queued so that it isn't blocking of the webui"*). Nothing here + changed as a result except who waits for it: the return value is now read + by the log rather than by a browser, and every branch below already + treated failure as "the sizing pass will carry it". + ## What is pushed, and what is not Consumers follow the cap immediately in BOTH directions: zero means off, @@ -612,13 +619,29 @@ async def push_lane_cap(lane: Lane, slots_cap: int, *, was_cap: int) -> dict: # feature that is optional and clearly OFF. # # On the TRANSITION, so re-saving a cap on a lane already running does not - # re-enqueue. And only when the consumer change landed: enqueueing onto a - # queue nothing is consuming would leave the task pending with no - # explanation until the lane returns. + # re-enqueue. + # + # NOT gated on the consumer change having landed, which it was until + # 2026-09-23. The reasoning then was that enqueueing onto a queue nothing + # consumes leaves the task pending — true, and it is the right place for + # it to wait. Gated, a cap raised while the lane was restarting stored the + # cap, let the sizing pass start the consumers a minute later, and left + # the lane running with no model, because nothing else ever asks for one. + # A task parked on the `ml` queue is picked up the moment that happens. fetching = False - if now_on and not was_on and lane.models and applied: + if now_on and not was_on and lane.models: fetching = _enqueue_model_fetch() + # Nobody is waiting on this any more, so the log is where a push that did + # not land has to be visible. Not an error: the value is stored and the + # sizing pass carries it within a minute. + if not applied: + log.info( + "worker_control: %s cap %s stored, not pushed (%s); " + "the sizing pass will carry it", + lane.name, slots_cap, error, + ) + return { "name": lane.name, "slots_cap": slots_cap, diff --git a/frontend/src/components/settings/SystemHealthTab.vue b/frontend/src/components/settings/SystemHealthTab.vue index ae24f91..4ae4b34 100644 --- a/frontend/src/components/settings/SystemHealthTab.vue +++ b/frontend/src/components/settings/SystemHealthTab.vue @@ -299,16 +299,13 @@ async function apply(lane, fields) { text: `${lane.display_name} is on. Downloading its model now — watch ` + 'progress under Activity. It only happens once.', } - } else if (reply && reply.applied === false) { - // Saved but not pushed — the lane is restarting, or the broker blipped. - // NOT an error: the reconcile carries it when the lane answers again, - // and saying "failed" would invite setting it a second time. - notice.value = { - type: 'info', - text: `Saved. ${lane.display_name} is not answering right now — ` - + 'it will pick this up within a few minutes.', - } } + // There is no "saved but not applied" message any more, because there is + // nothing to report yet: the endpoint answers once the cap is STORED and + // tells the running lane afterwards, so the press cannot wait on a broker + // round trip (operator: "we shouldn't have to wait for the validation + // live"). Whether the lane has caught up is what its row says, and the + // row updates on the next poll a few seconds later. } catch (e) { // The endpoint's `detail` is written to be read by a person ("cap 10000 is // above what this container can hold"). Surface it rather than a status diff --git a/frontend/src/stores/systemActivity.js b/frontend/src/stores/systemActivity.js index ca9c05b..efb2617 100644 --- a/frontend/src/stores/systemActivity.js +++ b/frontend/src/stores/systemActivity.js @@ -88,17 +88,37 @@ export const useSystemActivityStore = defineStore('systemActivity', () => { } } - // Change one lane. Returns the endpoint's reply so the caller can tell a - // stored-but-not-yet-live change (`applied: false`) from a live one — the - // difference between "saved, the lane is restarting" and "that failed", - // which the UI must not collapse into one message. + // Change one lane, and PATCH the row from the reply. + // + // It refetched until 2026-09-23, on the reasoning that the server's whole + // table beats a local guess. The cost of that was the operator's: + // + // "something about changing the cap number is blocking to the website... + // it shouldn't be" / "the change should be queued so that it isn't + // blocking of the webui or the system itself. we shouldn't have to wait + // for the validation live." + // + // GET /api/system/workers costs a celery inspect — an eleven-second budget + // — so every press of `+` disabled the stepper until a broker round trip + // the press did not need had finished. The endpoint now answers as soon as + // the cap is stored and tells the lane separately. + // + // Only the three fields the reply actually decides are copied. The live + // columns (pool, active, pending) are measurements this reply does not + // carry and must not invent — the 15s poller brings them, a beat behind, + // which is what they are anyway. // // Deliberately NOT swallowing the error: a refused value (400) carries the // sentence explaining why, and the card shows it. Returning null on failure // would leave the operator with a control that silently did nothing. async function setLane(name, fields) { const reply = await api.post(`/api/system/workers/${name}`, { body: fields }) - await loadLanes() + const row = lanes.value?.lanes?.find((l) => l.name === name) + if (row && reply) { + if (reply.slots_cap !== undefined) row.slots_cap = reply.slots_cap + if (reply.ceiling !== undefined) row.ceiling = reply.ceiling + if (reply.enabled !== undefined) row.enabled = reply.enabled + } return reply } diff --git a/frontend/test/workerLanes.spec.js b/frontend/test/workerLanes.spec.js index b594853..22981de 100644 --- a/frontend/test/workerLanes.spec.js +++ b/frontend/test/workerLanes.spec.js @@ -5,11 +5,18 @@ import { laneStuckFor, useSystemActivityStore } from '../src/stores/systemActivi // Milestone 422 step 4. Covers the store half of the worker-lane dial — the // part that decides what the card can tell the operator. // -// The distinction being protected: a change that was STORED but not pushed -// (`applied: false`, because the lane is restarting) is not a failure, and a -// REFUSED value (400) is. Collapsing those two into one message is how a -// control stops being trustworthy — one invites waiting, the other invites -// changing what you asked for. +// What is protected here since 2026-09-23: pressing the dial must not wait on +// a broker round trip. Operator: *"when the number is changed the change +// should be queued so that it isn't blocking of the webui or the system +// itself. we shouldn't have to wait for the validation live."* The endpoint +// answers once the cap is STORED, so the store patches its row from that +// reply and lets the 15s poll bring the live columns — it used to refetch, +// and GET /api/system/workers costs a celery inspect on an eleven-second +// budget. +// +// And still: a REFUSED value (400) is a failure and must reach the operator +// as one. A control that silently does nothing is worse than one that +// refuses out loud. function stubFetch(handler) { globalThis.fetch = vi.fn(async (url, init) => { @@ -70,7 +77,7 @@ describe('worker lanes store', () => { stubFetch((url, init) => { calls.push({ url, init }) if (init?.method === 'POST') { - return { status: 200, body: { name: 'worker', slots_cap: 2, applied: true } } + return { status: 200, body: { name: 'worker', slots_cap: 2, queued: true } } } return { status: 200, body: LANES_BODY } }) @@ -82,38 +89,73 @@ describe('worker lanes store', () => { expect(JSON.parse(post.init.body)).toEqual({ slots_cap: 2 }) }) - it('setLane refetches so the card shows the server truth, not the guess', async () => { - // The reply is one lane; the table renders all of them plus live pool - // and pending. Patching the local row from the reply would leave every - // other column stale and eventually wrong. + it('setLane does not refetch — that refetch is what blocked the press', async () => { + // It refetched until 2026-09-23, so that the card showed server truth + // rather than a local guess. The cost was the operator's: GET + // /api/system/workers runs a celery inspect, so every press of `+` sat + // with the stepper disabled through a broker round trip the press did not + // need. let gets = 0 stubFetch((url, init) => { - if (init?.method === 'POST') return { status: 200, body: { applied: true } } + if (init?.method === 'POST') { + return { status: 200, body: { slots_cap: 2, ceiling: 8, enabled: true } } + } gets += 1 return { status: 200, body: LANES_BODY } }) const s = useSystemActivityStore() + await s.loadLanes() + gets = 0 await s.setLane('worker', { slots_cap: 2 }) - expect(gets).toBe(1) + + expect(gets).toBe(0) }) - it('a stored-but-unapplied change comes back as applied:false, not an error', async () => { - // The lane is restarting. The value IS saved and the reconcile will carry - // it — so this must reach the card as information, not as a failure that - // invites the operator to set it again. + it('patches the row from the reply, so the new number shows at once', async () => { stubFetch((url, init) => { if (init?.method === 'POST') { return { status: 200, - body: { applied: false, apply_error: 'lane is not running', slots_cap: 2 }, + body: { name: 'worker', slots_cap: 2, ceiling: 8, enabled: true, queued: true }, } } return { status: 200, body: LANES_BODY } }) const s = useSystemActivityStore() - const reply = await s.setLane('worker', { slots_cap: 2 }) - expect(reply.applied).toBe(false) - expect(reply.apply_error).toContain('not running') + await s.loadLanes() + await s.setLane('worker', { slots_cap: 2 }) + + const worker = s.lanes.lanes.find((l) => l.name === 'worker') + expect(worker.slots_cap).toBe(2) + }) + + it('does not invent the live columns the reply cannot know', async () => { + // The reply is decided from the stored cap and the machine's ceiling + // alone — it never asked a worker anything. Pool, active and pending are + // MEASUREMENTS, and the poll a few seconds later is what carries them. + // Zeroing or guessing them here would make the table lie in the direction + // that reads as "the lane stopped". + stubFetch((url, init) => { + if (init?.method === 'POST') { + return { status: 200, body: { slots_cap: 2, ceiling: 8, enabled: true } } + } + return { status: 200, body: LANES_BODY } + }) + const s = useSystemActivityStore() + await s.loadLanes() + const before = { ...s.lanes.lanes[0].live } + await s.setLane('worker', { slots_cap: 2 }) + + expect(s.lanes.lanes[0].live).toEqual(before) + expect(s.lanes.lanes[0].pending).toBe(8) + }) + + it('survives a reply arriving for a lane it has not loaded yet', async () => { + // First paint, or a lane added by a newer build. Patching must not be the + // thing that throws inside the click handler. + stubFetch(() => ({ status: 200, body: { slots_cap: 2 } })) + const s = useSystemActivityStore() + await expect(s.setLane('worker', { slots_cap: 2 })).resolves.toBeTruthy() }) it('a refused value throws so the card can show the reason', async () => { diff --git a/tests/test_api_workers.py b/tests/test_api_workers.py index 90bcace..1f247db 100644 --- a/tests/test_api_workers.py +++ b/tests/test_api_workers.py @@ -10,6 +10,7 @@ import pytest import pytest_asyncio from sqlalchemy import select +from backend.app.api import workers as workers_api from backend.app.models import WorkerLane from backend.app.services import worker_control as wc from backend.app.services.worker_lanes import LANES @@ -28,6 +29,39 @@ async def no_live_workers(monkeypatch): ) +@pytest.fixture(autouse=True) +def queued_pushes(monkeypatch): + """Hold the background push still, and hand the test the hand-off. + + Since 2026-09-23 the endpoint stores the cap, answers, and gives the live + push to a Quart background task — operator: *"the change should be queued + so that it isn't blocking of the webui or the system itself."* A task that + outlived a test's monkeypatches would reach the real broker during + teardown and wait out its timeout there, so every test in this module + captures the hand-off instead, and the ones that care about what the push + DOES run it deliberately with `_run_pushes`. + + Autouse, because a test that forgets is not a test that fails — it is a + test that leaks a 2s broker call into whichever test runs next. + """ + scheduled: list[tuple] = [] + monkeypatch.setattr( + workers_api, "_schedule_push", + lambda lane, slots_cap, was_cap: scheduled.append((lane, slots_cap, was_cap)), + ) + return scheduled + + +async def _run_pushes(scheduled: list[tuple]) -> list[dict]: + """Run what the endpoint queued, in order, and clear the queue.""" + out = [ + await wc.push_lane_cap(lane, cap, was_cap=was_cap) + for lane, cap, was_cap in scheduled + ] + scheduled.clear() + return out + + async def _lane_row(db, name: str) -> WorkerLane: return (await db.execute( select(WorkerLane).where(WorkerLane.name == name) @@ -90,11 +124,11 @@ async def test_the_other_lanes_ship_at_one(client, no_live_workers): @pytest.mark.asyncio async def test_raising_a_cap_stores_it_and_pushes_nothing( - client, db, no_live_workers, + client, db, no_live_workers, queued_pushes, ): """A cap is PERMISSION, not a request. Raising it must not grow the pool - here — that would put workers on a lane with nothing to do — so there is - nothing to push and `applied` is vacuously true. + — that would put workers on a lane with nothing to do — so the queued + push has nothing to say to the broker. This asserted `applied is False` until run 7367, carried over from when the number meant "run this many". The code was right and the test was @@ -105,32 +139,33 @@ async def test_raising_a_cap_stores_it_and_pushes_nothing( assert resp.status_code == 200 body = await resp.get_json() assert body["slots_cap"] == 3 - assert body["applied"] is True assert (await _lane_row(db, "worker")).slots_cap == 3 + assert [r["applied"] for r in await _run_pushes(queued_pushes)] == [True] @pytest.mark.asyncio async def test_turning_a_lane_off_is_stored_even_when_it_cannot_be_pushed( - client, db, no_live_workers, + client, db, no_live_workers, queued_pushes, ): """The direction that DOES push. Consumers follow the cap immediately in both directions — off must take effect when it is asked for — so with nothing answering, the push fails. - That is NOT a failed setting: the value is saved and the sizing pass - carries it within a minute (lesson #4202 — a live change that does not - survive, with nothing saying so). The UI says "saved, not yet live" - rather than "that didn't work", which is the distinction `applied` - exists to carry. + That is NOT a failed setting, and it is why the reply does not wait for + it: the value is saved and the sizing pass carries it within a minute + (lesson #4202 — a live change that does not survive, with nothing saying + so). The push reports the failure to the log, where an operator can find + it; the table shows the lane not answering either way. """ resp = await client.post("/api/system/workers/worker", json={"slots_cap": 0}) assert resp.status_code == 200 - body = await resp.get_json() - assert body["applied"] is False - assert "not running" in body["apply_error"] assert (await _lane_row(db, "worker")).slots_cap == 0 + pushed = await _run_pushes(queued_pushes) + assert pushed[0]["applied"] is False + assert "not running" in pushed[0]["apply_error"] + # --- the cap is the switch --------------------------------------------------- @@ -158,7 +193,7 @@ async def test_raising_it_off_zero_turns_the_lane_on(client, db, no_live_workers @pytest.mark.asyncio async def test_the_model_fetch_fires_on_the_transition_not_on_every_write( - client, db, no_live_workers, monkeypatch, + client, db, no_live_workers, queued_pushes, monkeypatch, ): """Raising the cap off zero downloads SigLIP, once. A second nudge of the same dial must not re-enqueue a multi-GB download — and the trigger must @@ -176,15 +211,42 @@ async def test_the_model_fetch_fires_on_the_transition_not_on_every_write( first = await (await client.post( "/api/system/workers/ml", json={"slots_cap": 1}, )).get_json() + await _run_pushes(queued_pushes) second = await (await client.post( "/api/system/workers/ml", json={"slots_cap": 2}, )).get_json() + await _run_pushes(queued_pushes) + # The reply promises it, the queued push performs it. Both halves are + # checked, because the reply is what the UI says out loud. assert first["fetching_models"] is True assert second["fetching_models"] is False assert fired == [1] +@pytest.mark.asyncio +async def test_the_model_fetch_is_enqueued_even_if_the_lane_is_not_answering( + client, no_live_workers, queued_pushes, monkeypatch, +): + """Turning ML on while it is restarting must still fetch the model. + + It was gated on the consumer change having landed until 2026-09-23, on + the reasoning that a task enqueued onto a queue nothing consumes just sits + there. It does — and that is the right place for it to wait. Gated, this + path stored the cap, let the sizing pass start the consumers a minute + later, and left the lane running with no model, because nothing else ever + asks for one. + """ + fired = [] + monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True) + + await client.post("/api/system/workers/ml", json={"slots_cap": 1}) + pushed = await _run_pushes(queued_pushes) + + assert pushed[0]["applied"] is False, "the fixture must leave the push failing" + assert fired == [1] + + # --- what is refused --------------------------------------------------------- @@ -306,14 +368,12 @@ async def test_the_cap_write_holds_no_session_while_it_pushes(): @pytest.mark.asyncio async def test_raising_a_cap_costs_no_broker_round_trip_at_all( - client, db, no_live_workers, monkeypatch, + client, db, no_live_workers, queued_pushes, 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 - + """Raising a cap is permission, not a request — the sizing pass spends it + — so there is nothing to tell the broker AT ALL. Asserted on the queued + push rather than on the request, because the request no longer waits for + it either way and would pass this vacuously.""" calls = [] monkeypatch.setattr( wc, "inspect_lanes_sync", lambda: calls.append("inspect") or {}, @@ -328,8 +388,48 @@ async def test_raising_a_cap_costs_no_broker_round_trip_at_all( ) await client.post("/api/system/workers/worker", json={"slots_cap": 1}) + await _run_pushes(queued_pushes) calls.clear() resp = await client.post("/api/system/workers/worker", json={"slots_cap": 6}) + await _run_pushes(queued_pushes) assert resp.status_code == 200 assert calls == [], f"raising a cap talked to the broker: {calls}" + + +@pytest.mark.asyncio +async def test_the_reply_never_waits_for_the_broker_in_any_direction( + client, db, no_live_workers, queued_pushes, monkeypatch, +): + """Operator, 2026-09-23: *"when the number is changed the change should be + queued so that it isn't blocking of the webui or the system itself. we + shouldn't have to wait for the validation live."* + + Raising a cap was already free. The rest were not: turning a lane off is + four `cancel_consumer` messages, and lowering a cap reads the live pool + first — an `inspect` on an eleven-second budget — all of it between the + click and the response, with the stepper disabled throughout. + + Asserted by making any broker call from the request path RAISE. A timing + assertion would be flaky, and counting calls afterwards would pass against + a version that made them and was merely quick about it. + """ + def boom(*args, **kwargs): + raise AssertionError("the request path talked to the broker") + + monkeypatch.setattr(wc, "inspect_lanes_sync", boom) + monkeypatch.setattr(wc, "set_lane_slots_sync", boom) + monkeypatch.setattr(wc, "set_lane_enabled_sync", boom) + + # Every direction: on, up, down, off. + for cap in (1, 4, 2, 0): + resp = await client.post( + "/api/system/workers/worker", json={"slots_cap": cap}, + ) + assert resp.status_code == 200, cap + assert (await resp.get_json())["queued"] is True, cap + assert (await _lane_row(db, "worker")).slots_cap == cap + # The work was handed off rather than skipped — a control that answers + # instantly by doing nothing is the failure this could become. + assert queued_pushes, f"cap {cap} queued no push" + queued_pushes.clear()