diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 9db7d23..7e31f09 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -534,9 +534,33 @@ async def set_lane( new_cap = row.slots_cap if slots_cap is None else slots_cap new_slots = row.slots if slots is None else slots - new_enabled = row.enabled if enabled is None else enabled new_autoscale = row.autoscale if autoscale is None else autoscale + # THE DIAL IS THE SWITCH. A lane at zero slots is a lane that is off, and + # there is no second control saying so. + # + # Operator, 2026-09-23, on the card that had both: *"there's nothing to + # describe what 'auto' means or why their needs to be or should be on/off + # toggles. almost all of it always needs to run there's only one optional + # piece and it is killed by moving the 'cap' to zero."* They are right. Of + # four lanes, three must run for the application to work at all, so a + # switch beside each of them offered a choice that was never real — and + # for the one lane that IS optional, "off" and "zero slots" were two ways + # of saying the same thing that could disagree with each other. + # + # `enabled` stays in the model and on the API. It is still the mechanism: + # a disabled lane keeps its process and cancels its consumers, which is + # what makes it visible in the roster instead of looking like a crash. It + # is now DERIVED from the number the operator actually sets, rather than + # being a second thing for them to keep in agreement with it. + was_enabled = row.enabled + if enabled is not None: + new_enabled = enabled + elif slots is not None: + new_enabled = new_slots > 0 + else: + new_enabled = row.enabled + ceiling = derived_ceiling(lane) if new_cap < 0 or new_slots < 0: raise LaneUpdateRefused("slots and cap cannot be negative") @@ -555,7 +579,11 @@ async def set_lane( await session.commit() applied, error = True, None - if enabled is not None: + # On the CHANGE, not on the field being present. Now that `enabled` is + # derived, every slots write would otherwise re-send a consumer command + # that changes nothing — the churn lesson #4183 keeps producing, arriving + # here through the new derivation. + if new_enabled != was_enabled: applied, error = await asyncio.to_thread( set_lane_enabled_sync, lane, new_enabled, ) @@ -567,13 +595,17 @@ async def set_lane( # HuggingFace for ~3.5GB, and rule 164 permits a runtime fetch only for a # feature that is optional and clearly OFF. # - # Only when the lane actually came on — `enabled is True` rather than - # `new_enabled`, so re-saving slots on an already-enabled lane does not - # re-enqueue. And only when the consumer change landed: enqueueing a task - # onto a queue nothing is consuming would leave it pending with no - # explanation until the lane returns. + # Only on the TRANSITION from off to on, so re-saving slots on a lane that + # is already running does not re-enqueue. This used to test `enabled is + # True` — the field having been sent — which stopped meaning "came on" the + # moment the dial became the switch: the UI no longer sends `enabled` at + # all, so the fetch that makes the ML lane usable would never have fired. + # + # 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. fetching = False - if enabled is True and lane.models and applied: + if new_enabled and not was_enabled and lane.models and applied: fetching = _enqueue_model_fetch() return { diff --git a/frontend/src/components/settings/SystemHealthTab.vue b/frontend/src/components/settings/SystemHealthTab.vue index 753d2d2..d5e4364 100644 --- a/frontend/src/components/settings/SystemHealthTab.vue +++ b/frontend/src/components/settings/SystemHealthTab.vue @@ -1,8 +1,22 @@ + + diff --git a/frontend/src/components/settings/WorkerLanesCard.vue b/frontend/src/components/settings/WorkerLanesCard.vue deleted file mode 100644 index 247e9db..0000000 --- a/frontend/src/components/settings/WorkerLanesCard.vue +++ /dev/null @@ -1,342 +0,0 @@ - - - - - diff --git a/frontend/src/styles/app.css b/frontend/src/styles/app.css index 93c8d7a..fbd9380 100644 --- a/frontend/src/styles/app.css +++ b/frontend/src/styles/app.css @@ -132,50 +132,3 @@ backdrop-filter: blur(2px); -webkit-backdrop-filter: blur(2px); } - -/* The System tab's row idiom (operator 2026-09-23: "improve the view to be - more inline with other UI elements"). - - The roster (milestone 365) and the worker lanes (milestone 422) sit on the - same tab and answer the two halves of one question — what is running, and - how hard. They were built months apart and looked it: one a flat list of - dotted rows, the other a bordered card wrapping a v-table. Lifting these out - of SystemHealthTab's scoped block is what lets the second pane BE the first - one's idiom rather than imitate it — a copy would drift the first time - either was touched. - - `fc-sys__` rather than a new prefix because the roster's markup already uses - these names; renaming would have been churn in the file that is not - changing. */ -.fc-sys__lede, .fc-sys__muted, .fc-sys__checked, .fc-sys__foot { - color: rgb(var(--v-theme-on-surface) / 0.66); -} -.fc-sys__checked { font-size: 0.78rem; } -.fc-sys__card { background: rgb(var(--v-theme-on-surface) / 0.04); } - -.fc-sys__row { - display: flex; align-items: center; gap: 12px; - padding: 12px 16px; - border-bottom: 1px solid rgb(var(--v-theme-on-surface) / 0.08); -} -.fc-sys__row:last-child { border-bottom: 0; } - -.fc-sys__dot { width: 9px; height: 9px; border-radius: 50%; flex: 0 0 auto; } -.fc-sys__dot--ok { background: rgb(var(--v-theme-success)); } -.fc-sys__dot--stale { background: rgb(var(--v-theme-warning)); } -.fc-sys__dot--down { background: rgb(var(--v-theme-error)); } -.fc-sys__dot--unknown { background: rgb(var(--v-theme-on-surface) / 0.35); } - -.fc-sys__body { min-width: 0; flex: 1 1 auto; } -.fc-sys__name { font-weight: 600; } -.fc-sys__kind { - margin-left: 8px; font-weight: 400; font-size: 0.72rem; text-transform: uppercase; - letter-spacing: 0.04em; color: rgb(var(--v-theme-on-surface) / 0.5); -} -.fc-sys__detail { font-size: 0.82rem; color: rgb(var(--v-theme-on-surface) / 0.72); } - -.fc-sys__meta { - text-align: right; font-size: 0.75rem; flex: 0 0 auto; - font-variant-numeric: tabular-nums; color: rgb(var(--v-theme-on-surface) / 0.6); -} -.fc-sys__queues { opacity: 0.75; } diff --git a/frontend/src/utils/systemParts.js b/frontend/src/utils/systemParts.js new file mode 100644 index 0000000..e222edf --- /dev/null +++ b/frontend/src/utils/systemParts.js @@ -0,0 +1,100 @@ +// Joining the roster to the worker lanes, for the System tab's one table. +// +// Extracted from the component rather than left inline because the failure +// this can have is SILENT and is exactly the thing the merge exists to fix: if +// a lane stops matching its roster part, nothing throws — the table simply +// grows a second row for the same worker, one with controls and one without, +// which is the duplication the operator asked to be rid of, returned by the +// code that removed it. +// +// Operator, 2026-09-23: "I feel that we can probably combine the two +// sections into a single table." + +// A learned roster part and a lane are the same thing seen from two sides, and +// the QUEUES are what identify it — `service_roster.refresh_celery_roster` +// keys a celery part on exactly `"celery:" + ",".join(sorted(queues))`. +// +// Matched on the sorted set rather than on that string so the join survives a +// change to how the key is spelled, and so neither side has to agree about +// ORDER: the lane table lists a lane's queues in the order the role reads them +// (`default, import, thumbnail, download`) while the roster sorts them +// (`default, download, import, thumbnail`). +export function queueKey(queues) { + return [...(queues || [])].sort().join(',') +} + +// Worst first. A stopped datastore is why someone opened this tab. +export const SEVERITY = { down: 3, stale: 2, unknown: 1, ok: 0 } + +export function kindLabel(kind) { + if (kind === 'celery') return 'worker lane' + if (kind === 'agent') return 'GPU agent' + if (kind === 'datastore') return 'datastore' + return kind +} + +/** + * One row per moving part, with a lane attached where there is one. + * + * @param parts the roster's parts, as /api/system/health returns them + * @param lanes the lane rows, as /api/system/workers returns them + * @param stuckFor a lane -> "40 minutes" | null reporter (laneStuckFor) + */ +export function mergeParts(parts, lanes, stuckFor = () => null) { + const unmatched = {} + for (const lane of lanes || []) unmatched[queueKey(lane.queues)] = lane + + const out = [] + for (const part of parts || []) { + const key = queueKey(part.queues) + const lane = part.kind === 'celery' ? unmatched[key] : undefined + if (lane) delete unmatched[key] + out.push({ + key: part.key, + name: part.name, + kindLabel: lane?.optional ? 'optional lane' : kindLabel(part.kind), + state: part.state, + // A lane dialled to zero is OFF, not broken. Say so, rather than let the + // roster's heartbeat sentence report the operator's own choice as a + // fault — the roster cannot know the difference, and the lane can. + detail: lane && lane.slots === 0 ? 'off — no slots' : part.detail, + queues: (part.queues || []).join(', '), + lane, + stuckFor: lane ? stuckFor(lane) : null, + severity: SEVERITY[part.state] ?? SEVERITY.unknown, + }) + } + + // A lane the roster has not learned yet. Parts appear only once they have + // checked in, while the lane table is known up front — so without this, the + // lane an operator most needs to find (an optional one, never yet started) + // would be the only one missing from the table. + for (const lane of Object.values(unmatched)) out.push(laneRow(lane, stuckFor)) + + // Severity leads; then lanes ahead of everything else, because they are the + // rows you can actually do something about; then by name. + return out.sort((a, b) => + b.severity - a.severity + || Number(Boolean(b.lane)) - Number(Boolean(a.lane)) + || a.name.localeCompare(b.name)) +} + +function laneRow(lane, stuckFor) { + const on = lane.slots > 0 + let state = 'unknown' + if (lane.live?.present) state = on ? (stuckFor(lane) ? 'stale' : 'ok') : 'unknown' + else if (on) state = 'down' + return { + key: `lane:${lane.name}`, + name: lane.display_name, + kindLabel: lane.optional ? 'optional lane' : 'worker lane', + state, + detail: lane.live?.present + ? (on ? 'running' : 'off — no slots') + : 'has not checked in yet', + queues: (lane.queues || []).join(', '), + lane, + stuckFor: stuckFor(lane), + severity: SEVERITY[state], + } +} diff --git a/frontend/test/systemParts.spec.js b/frontend/test/systemParts.spec.js new file mode 100644 index 0000000..894b09d --- /dev/null +++ b/frontend/test/systemParts.spec.js @@ -0,0 +1,128 @@ +import { describe, expect, it } from 'vitest' + +import { mergeParts, queueKey } from '../src/utils/systemParts.js' + +// The System tab's one table (operator 2026-09-23: "combine the two sections +// into a single table"). What is pinned here is the JOIN, because its failure +// is silent and is precisely the thing the merge exists to fix: a lane that +// stops matching its roster part does not throw — the table grows a SECOND row +// for the same worker, one with controls and one without. + +const PART = { + key: 'celery:default,download,import,thumbnail', + kind: 'celery', + name: 'Worker', + state: 'ok', + detail: 'Worker is running', + // The roster SORTS a celery part's queues into its key. + queues: ['default', 'download', 'import', 'thumbnail'], +} + +const LANE = { + name: 'worker', + display_name: 'Worker', + // The lane table lists them in the order the role reads them, which is NOT + // sorted. If the join ever compares these two lists directly rather than as + // sets, this fixture is what catches it. + queues: ['default', 'import', 'thumbnail', 'download'], + slots: 2, + slots_cap: 4, + autoscale: false, + live: { present: true, replicas: 1, pool: 2, active: 0 }, + pending: 0, +} + +const POSTGRES = { + key: 'postgres', kind: 'datastore', name: 'PostgreSQL', + state: 'ok', detail: 'answering', latency_ms: 2.5, +} + +describe('queueKey', () => { + it('does not care what order either side lists its queues in', () => { + expect(queueKey(LANE.queues)).toBe(queueKey(PART.queues)) + }) + + it('survives a part that has no queues at all', () => { + // A datastore, and also the stale `Worker ()` row a previous deployment + // left in the roster with an empty queue set. Neither must match a lane. + expect(queueKey(undefined)).toBe('') + expect(queueKey([])).toBe('') + }) +}) + +describe('mergeParts', () => { + it('gives a worker ONE row, carrying its controls', () => { + const rows = mergeParts([PART, POSTGRES], [LANE]) + + expect(rows).toHaveLength(2) + const worker = rows.find((r) => r.name === 'Worker') + expect(worker.lane).toBe(LANE) + expect(rows.filter((r) => r.name === 'Worker')).toHaveLength(1) + }) + + it('leaves a datastore without a lane rather than guessing one', () => { + const pg = mergeParts([PART, POSTGRES], [LANE]).find((r) => r.key === 'postgres') + expect(pg.lane).toBeUndefined() + }) + + it('still lists a lane the roster has never seen', () => { + // Parts are learned as they appear; the lane table is known up front. The + // lane an operator most needs to find — an optional one, never started — + // is exactly the one with no roster entry. + const ml = { + ...LANE, name: 'ml', display_name: 'ML tagging', queues: ['ml'], + slots: 0, optional: true, + live: { present: false, replicas: 0, pool: null, active: 0 }, + } + const rows = mergeParts([POSTGRES], [ml]) + + const row = rows.find((r) => r.name === 'ML tagging') + expect(row.lane).toBe(ml) + expect(row.kindLabel).toBe('optional lane') + }) + + it('does not call a lane at zero slots broken', () => { + // The roster only knows a heartbeat age, so it goes on saying "is running" + // for a lane the operator deliberately dialled to nothing. The lane knows + // the difference; reporting the operator's own choice as a fault is how an + // indicator stops being read. + const off = { ...LANE, slots: 0 } + const row = mergeParts([PART], [off])[0] + + expect(row.detail).toBe('off — no slots') + }) + + it('puts the broken thing first, whatever it is', () => { + const down = { ...POSTGRES, state: 'down', detail: 'not answering' } + const rows = mergeParts([PART, down], [LANE]) + + expect(rows[0].name).toBe('PostgreSQL') + }) + + it('otherwise puts the rows you can act on first', () => { + const rows = mergeParts([POSTGRES, PART], [LANE]) + + expect(rows.map((r) => r.name)).toEqual(['Worker', 'PostgreSQL']) + }) + + it('reports a wedged lane, and only through the reporter it was given', () => { + // `laneStuckFor` is passed in rather than imported, so this file does not + // re-test the store's rule — it tests that the merge asks. + const asked = [] + const rows = mergeParts([PART], [LANE], (lane) => { + asked.push(lane.name) + return '40 minutes' + }) + + expect(asked).toEqual(['worker']) + expect(rows[0].stuckFor).toBe('40 minutes') + // The roster still owns a matched row's state — `stuckFor` is a note + // beside it, not a verdict that overrides the heartbeat. + expect(rows[0].state).toBe('ok') + }) + + it('handles an empty everything without inventing rows', () => { + expect(mergeParts([], [])).toEqual([]) + expect(mergeParts(undefined, undefined)).toEqual([]) + }) +}) diff --git a/frontend/test/workerLanes.spec.js b/frontend/test/workerLanes.spec.js index 787d7f0..f553e57 100644 --- a/frontend/test/workerLanes.spec.js +++ b/frontend/test/workerLanes.spec.js @@ -54,8 +54,9 @@ describe('worker lanes store', () => { }) it('a load failure records the error rather than throwing at the caller', async () => { - // The card polls this every 3s. An unhandled rejection per tick would - // drown the console and stop the other pollers in the same function. + // The System tab polls this every 15s. An unhandled rejection per tick + // would drown the console and stop the other pollers in the same + // function. stubFetch(() => ({ status: 500, body: { error: 'boom' } })) const s = useSystemActivityStore() await expect(s.loadLanes()).resolves.toBeUndefined() @@ -83,9 +84,9 @@ describe('worker lanes store', () => { }) it('setLane refetches so the card shows the server truth, not the guess', async () => { - // The reply is one lane; the card 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. + // 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. let gets = 0 stubFetch((url, init) => { if (init?.method === 'POST') return { status: 200, body: { applied: true } } diff --git a/tests/test_api_workers.py b/tests/test_api_workers.py index 97bd728..d36e1fd 100644 --- a/tests/test_api_workers.py +++ b/tests/test_api_workers.py @@ -170,3 +170,114 @@ async def test_an_empty_body_is_refused_rather_than_treated_as_a_no_op( assert resp.status_code == 400 body = await resp.get_json() assert body["error"] == "invalid_body" + + +# --- the dial is the switch -------------------------------------------------- +# +# Operator, 2026-09-23: *"almost all of it always needs to run there's only one +# optional piece and it is killed by moving the 'cap' to zero."* So `enabled` +# is derived from the number rather than being a second control the operator +# has to keep in agreement with it. It stays on the API — these assert that it +# still does, because it is the mechanism the reconcile and the healthcheck +# read. + + +@pytest.mark.asyncio +async def test_dialling_a_lane_to_zero_turns_it_off(client, db, no_live_workers): + await client.post("/api/system/workers/worker", json={"slots": 0}) + + row = await _lane_row(db, "worker") + assert row.slots == 0 + assert row.enabled is False + + +@pytest.mark.asyncio +async def test_dialling_it_back_up_turns_it_on(client, db, no_live_workers): + await client.post("/api/system/workers/ml", json={"slots": 1}) + + row = await _lane_row(db, "ml") + assert row.slots == 1 + assert row.enabled is True, "the lane the operator just asked for work from" + + +@pytest.mark.asyncio +async def test_an_explicit_enabled_still_wins(client, db, no_live_workers): + """The field is not removed, only derived when absent. Something that + genuinely wants a lane holding its process with consumers cancelled — a + drain before a restart — must still be able to say so without having to + destroy the operator's slot count to express it.""" + await client.post( + "/api/system/workers/worker", json={"slots": 3, "enabled": False}, + ) + + row = await _lane_row(db, "worker") + assert (row.slots, row.enabled) == (3, False) + + +@pytest.mark.asyncio +async def test_a_cap_only_write_does_not_decide_the_switch( + client, db, no_live_workers, +): + """Only the SLOTS dial derives it. A cap is a ceiling, not a request for + work, and letting it flip the lane would make raising a ceiling start + something.""" + before = await _lane_row(db, "ml") + assert before.enabled is False + + await client.post("/api/system/workers/ml", json={"slots_cap": 1}) + + await db.refresh(before) + assert (before.slots, before.enabled) == (0, False) + + +@pytest.mark.asyncio +async def test_the_model_fetch_fires_on_the_transition_not_on_the_field( + client, db, no_live_workers, monkeypatch, +): + """The trap the derivation set, caught here rather than in production. + + The fetch used to be conditioned on `enabled is True` — the FIELD having + been sent. The UI no longer sends it at all, so the download that makes + the ML lane usable would simply never have fired, and the lane would have + come on and sat there consuming a queue it had no model for. + """ + fired = [] + monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True) + monkeypatch.setattr( + wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None), + ) + monkeypatch.setattr( + wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None), + ) + + body = await (await client.post( + "/api/system/workers/ml", json={"slots": 1}, + )).get_json() + + assert body["fetching_models"] is True + assert fired == [1] + + +@pytest.mark.asyncio +async def test_it_does_not_fire_again_on_a_lane_already_running( + client, db, no_live_workers, monkeypatch, +): + """The other half. A second nudge of the dial on a lane that is already on + must not re-enqueue a multi-GB download.""" + monkeypatch.setattr( + wc, "set_lane_enabled_sync", lambda lane, enabled, live=None: (True, None), + ) + monkeypatch.setattr( + wc, "set_lane_slots_sync", lambda lane, target, live=None: (True, None), + ) + await client.post("/api/system/workers/ml", json={"slots": 1}) + + fired = [] + monkeypatch.setattr(wc, "_enqueue_model_fetch", lambda: fired.append(1) or True) + + body = await (await client.post( + "/api/system/workers/ml", json={"slots": 1}, + )).get_json() + + assert body["fetching_models"] is False + assert fired == []