From a01165365b182d84e37b0adc6c6961db875116de Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 22 Sep 2026 10:05:43 -0400 Subject: [PATCH] feat: a saturated lane can grow itself, within the cap the operator set (4297) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Milestone 422 step 7 — the one sweep in this milestone that decides rather than obeys, so it is off until a lane is opted in, bounded by the operator's cap, floored at the operator's value, and it reports every decision including the ones where it did nothing. Growth needs BOTH halves: all slots busy AND a backlog. Depth alone means celery is about to pick those up and growing would add idle children (#1253 is that bug in the GPU agent); saturation alone means the lane is busy with exactly as much work as exists. The backlog is depth PLUS reserved, because celery prefetches and LLEN reads 0 while a worker holds thirty tasks in memory — the case an LLEN-only autoscaler misses entirely, and the reason step 2 plumbed `reserved` through. The two sweeps had to be taught not to fight. The reconcile drives every lane to its stored slots every five minutes, which would have reverted each grow on the next tick: grow, revert, grow, revert, forever. For an autoscaling lane the stored value is now a FLOOR — restored when a lane falls below it, never taken back above it. The operator's "a task that runs for x concurrent time" idea stays a UI warning rather than a trigger: a long task does not finish sooner because the lane gained a slot, so scaling on it would spend memory to change nothing. Read from `task_run` on our own wall clock, not celery's `time_start`, which is the WORKER's monotonic clock and would produce a duration that is meaningless in the direction that matters — plausible. Caught while reading it back: the first version read the stored slots as the CURRENT pool. The autoscaler never writes that row, so every tick would have proposed floor+1 — resizing nothing, reporting `grew` anyway (a replica already past the target is issued no message and reports success), and capping the lane one slot above its floor forever while claiming otherwise. It now reads the live pool and keeps the stored value purely as the floor, and the tests fix the two to different numbers so an equal-fixture pass cannot hide it again. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- .../versions/0104_worker_lane_autoscale.py | 51 ++++ backend/app/api/workers.py | 11 +- backend/app/celery_app.py | 8 + backend/app/models/worker_lane.py | 12 + backend/app/services/worker_control.py | 255 +++++++++++++++++- backend/app/services/worker_lanes.py | 5 + backend/app/tasks/maintenance.py | 65 ++++- .../components/settings/WorkerLanesCard.vue | 48 +++- frontend/src/stores/systemActivity.js | 23 ++ frontend/test/workerLanes.spec.js | 53 +++- tests/test_worker_control.py | 242 +++++++++++++++++ 11 files changed, 754 insertions(+), 19 deletions(-) create mode 100644 alembic/versions/0104_worker_lane_autoscale.py diff --git a/alembic/versions/0104_worker_lane_autoscale.py b/alembic/versions/0104_worker_lane_autoscale.py new file mode 100644 index 0000000..38e74d0 --- /dev/null +++ b/alembic/versions/0104_worker_lane_autoscale.py @@ -0,0 +1,51 @@ +"""worker_lane.autoscale — may this lane grow itself? + +Milestone 422 step 7. One boolean, defaulting FALSE on every existing row and +on every new one. + +## Why the default is false and not "sensible" + +This is the only part of the milestone that acts without anyone watching. The +manual dial (step 4) and the reconcile (step 3) both do exactly what someone +asked for; this one decides. Shipping it on would mean every install starts +with a process that changes its own resource usage based on a heuristic tuned +against nobody's workload. + +Off also makes the failure mode benign: if the signal is wrong, nothing +happens until an operator opts a lane in, and they opted in while watching. + +## Why per lane and not one global switch + +The lanes are not alike in what a slot costs. A `worker` slot is a process; +an `ml` slot is another copy of a ~3.5GB model. A global switch would enable +growth on a lane whose behaviour under load nobody has observed, and the one +it would hurt most is the one whose cost is least visible. + +Revision ID: 0104 +Revises: 0103 +Create Date: 2026-09-22 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0104" +down_revision: Union[str, None] = "0103" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "worker_lane", + sa.Column( + "autoscale", sa.Boolean(), + server_default=sa.text("false"), nullable=False, + ), + ) + + +def downgrade() -> None: + op.drop_column("worker_lane", "autoscale") diff --git a/backend/app/api/workers.py b/backend/app/api/workers.py index b5f96f0..678c061 100644 --- a/backend/app/api/workers.py +++ b/backend/app/api/workers.py @@ -84,15 +84,16 @@ async def update_lane(name: str): if not isinstance(value, int) or isinstance(value, bool): return _bad("invalid_body", detail=f"{key} must be an integer") fields[key] = value - if "enabled" in body: - if not isinstance(body["enabled"], bool): - return _bad("invalid_body", detail="enabled must be a boolean") - fields["enabled"] = body["enabled"] + for key in ("enabled", "autoscale"): + if key in body: + if not isinstance(body[key], bool): + return _bad("invalid_body", detail=f"{key} must be a boolean") + fields[key] = body[key] if not fields: return _bad( "invalid_body", - detail="give at least one of slots, slots_cap, enabled", + detail="give at least one of slots, slots_cap, enabled, autoscale", ) async with get_session() as session: diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index 894fcd1..e509f8f 100644 --- a/backend/app/celery_app.py +++ b/backend/app/celery_app.py @@ -111,6 +111,14 @@ def make_celery() -> Celery: "task": "backend.app.tasks.maintenance.recover_interrupted_tasks", "schedule": 300.0, # every 5 minutes }, + "autoscale-worker-lanes": { + "task": "backend.app.tasks.maintenance.autoscale_worker_lanes", + "schedule": 60.0, # every minute — it reacts to a BACKLOG, and + # a five-minute reaction to a queue filling up is no reaction. + # Cheap when nothing opted in (one DB read, then nothing) and + # cheap when settled (one inspect + one LLEN sweep, no control + # messages), so the short interval costs little. + }, "reconcile-worker-lanes": { "task": "backend.app.tasks.maintenance.reconcile_worker_lanes", "schedule": 300.0, # every 5 minutes — the window in which a diff --git a/backend/app/models/worker_lane.py b/backend/app/models/worker_lane.py index ae3cc09..94daf45 100644 --- a/backend/app/models/worker_lane.py +++ b/backend/app/models/worker_lane.py @@ -78,6 +78,18 @@ class WorkerLane(Base): slots_cap: Mapped[int] = mapped_column(Integer, nullable=False) enabled: Mapped[bool] = mapped_column(Boolean, nullable=False) + # Whether the autoscaler may raise this lane's slots on its own, up to + # slots_cap and never past it (milestone 422 step 7). + # + # OFF by default and per-lane rather than global. It is the one part of + # this milestone that acts unattended, so it is opt-in per lane the + # operator has actually thought about — a global switch would turn it on + # for lanes whose behaviour under load nobody has watched, including `ml`, + # where every extra slot is another copy of a multi-GB model. + autoscale: Mapped[bool] = mapped_column( + Boolean, nullable=False, server_default="false", + ) + updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, diff --git a/backend/app/services/worker_control.py b/backend/app/services/worker_control.py index 5047e9e..157b01b 100644 --- a/backend/app/services/worker_control.py +++ b/backend/app/services/worker_control.py @@ -50,11 +50,12 @@ from __future__ import annotations import asyncio import logging from dataclasses import dataclass, field +from datetime import UTC, datetime -from sqlalchemy import select +from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from ..models import WorkerLane +from ..models import TaskRun, WorkerLane from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane, derived_ceiling log = logging.getLogger(__name__) @@ -103,6 +104,16 @@ class LaneLiveState: no replica reported — unknown, never zero.""" return max(self.pools.values()) if self.pools else None + @property + def capacity(self) -> int: + """Total slots across replicas — how many tasks this lane can run at + once. Distinct from `pool`, and the two must not be confused: `pool` + is the DIAL (one process's size, what grow/shrink move), `capacity` is + the CAPABILITY. Asking "is this lane saturated" compares `active`, + which is summed across replicas, against this — against `pool` it + would call two half-busy replicas of 4 saturated at 4 active.""" + return sum(self.pools.values()) + def _lane_for_queues(queues: tuple[str, ...]) -> Lane | None: return LANES_BY_QUEUE_KEY.get(tuple(sorted(queues))) @@ -260,6 +271,7 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]: slots=lane.default_slots, slots_cap=lane.default_slots_cap, enabled=lane.default_enabled, + autoscale=lane.default_autoscale, ) session.add(row) rows[lane.name] = row @@ -278,7 +290,9 @@ async def lane_view(session: AsyncSession) -> list[dict]: rows = await _rows_by_name(session) live = await asyncio.to_thread(inspect_lanes_sync) depths = await asyncio.to_thread(_queue_depths_sync) + oldest = await _oldest_running_by_queue(session) + now = datetime.now(UTC) out = [] for lane in LANES: row = rows[lane.name] @@ -297,6 +311,7 @@ async def lane_view(session: AsyncSession) -> list[dict]: "slots_cap": row.slots_cap, "ceiling": derived_ceiling(lane), "enabled": row.enabled, + "autoscale": row.autoscale, "memory_bound": lane.memory_bound, "optional": lane.optional, # What enabling this lane will download, so the UI can say WHICH @@ -321,10 +336,55 @@ async def lane_view(session: AsyncSession) -> list[dict]: }, "queue_depth": depth, "pending": None if depth is None else depth + state.reserved, + # 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 + # 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, + # which is the genuinely useful half of the idea. + "oldest_running_minutes": _minutes_since( + min( + (oldest[q] for q in lane.queues if q in oldest), + default=None, + ), + now, + ), }) return out +async def _oldest_running_by_queue(session: AsyncSession) -> dict[str, datetime]: + """When the longest-running unfinished task on each queue started. + + Read from `task_run`, which is OUR OWN table on OUR OWN wall clock, and + deliberately not from celery's `inspect active()`. Those entries carry a + `time_start` taken from the WORKER's `time.monotonic()` — a clock with an + arbitrary origin per process. Subtracting it from this process's wall + clock produces a number that looks like a duration and is meaningless, and + it would be meaningless in the direction that matters: plausible. + + `task_run` also already carries the per-queue staleness thresholds the + recovery sweep uses, so a row still `running` here is one the system + itself considers legitimately in flight rather than abandoned. + """ + result = await session.execute( + select(TaskRun.queue, func.min(TaskRun.started_at)) + .where(TaskRun.status == "running", TaskRun.finished_at.is_(None)) + .group_by(TaskRun.queue) + ) + return {queue: started for queue, started in result if started is not None} + + +def _minutes_since(started: datetime | None, now: datetime) -> int | None: + """Whole minutes, or None when nothing is running. Never negative: a row + written by a container whose clock is a few seconds ahead must read as 0 + rather than as a task that starts in the future.""" + if started is None: + return None + return max(0, int((now - started).total_seconds() // 60)) + + def _queue_depths_sync() -> dict[str, int | None]: """Redis LLEN per queue. None for one that did not answer — see lane_view. @@ -362,6 +422,7 @@ async def set_lane( slots: int | None = None, slots_cap: int | None = None, enabled: bool | None = None, + autoscale: bool | None = None, ) -> dict: """Store the operator's choice, then push it to the running lane. @@ -383,6 +444,7 @@ 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 ceiling = derived_ceiling(lane) if new_cap < 0 or new_slots < 0: @@ -398,6 +460,7 @@ async def set_lane( row.slots_cap = new_cap row.slots = new_slots row.enabled = new_enabled + row.autoscale = new_autoscale await session.commit() applied, error = True, None @@ -428,6 +491,7 @@ async def set_lane( "slots_cap": row.slots_cap, "ceiling": ceiling, "enabled": row.enabled, + "autoscale": row.autoscale, "applied": applied, "apply_error": error, # Tells the card to say a download has started rather than leaving the @@ -458,11 +522,28 @@ def _enqueue_model_fetch() -> bool: return False -def reconcile_lanes_sync(desired: dict[str, tuple[int, bool]]) -> dict: +def reconcile_lanes_sync( + desired: dict[str, tuple[int, bool]], + autoscaling: frozenset[str] = frozenset(), +) -> dict: """Drive every RUNNING lane to its stored slots and enabled flag. `desired` is lane name -> (slots, enabled), read from the database by the - caller. This function touches no database: the celery task that schedules + caller. `autoscaling` names the lanes the autoscaler is allowed to move. + + ## For an autoscaling lane the stored value is a FLOOR, not a target + + Step 7's autoscaler raises a saturated lane's live pool without changing + its row — the row holds what the OPERATOR set. If this pass treated that + row as an exact target it would shrink the lane back on the very next + tick, and the two sweeps would fight forever at five-minute intervals: + grow, revert, grow, revert. That is lesson #4183's failure arriving + between two enforcers rather than inside one. + + So for those lanes the target becomes `max(stored, current)` — this pass + still restores a lane that came back from a restart below what the + operator set, and never takes back what the autoscaler added. Bringing it + down is the autoscaler's job, and it does so only to that same floor. This function touches no database: the celery task that schedules it owns the sync session, and keeping the DB out of here is what lets the same code be called from anywhere that already knows the target. @@ -521,17 +602,177 @@ def reconcile_lanes_sync(desired: dict[str, tuple[int, bool]]) -> dict: changed.append(lane.name) current = state.pool - if current is not None and current == slots: + # The floor, for a lane the autoscaler manages. See the docstring. + target_slots = slots + if lane.name in autoscaling and current is not None: + target_slots = max(slots, current) + if current is not None and current == target_slots: continue - ok, err = set_lane_slots_sync(lane, slots, live=state) + ok, err = set_lane_slots_sync(lane, target_slots, live=state) if ok: if lane.name not in changed: changed.append(lane.name) log.info( "worker_control: %s reconciled %s -> %s slots", - lane.name, current, slots, + lane.name, current, target_slots, ) else: failed[lane.name] = err or "could not resize" return {"changed": changed, "skipped": skipped, "failed": failed} + + +# --- the autoscaler (step 7) -------------------------------------------------- +# +# The only part of this milestone that acts without anyone asking. Everything +# above does what an operator pressed; this decides. So it is off by default, +# opted into per lane, bounded by the cap the operator set, and it reports what +# it did rather than moving numbers silently. + +# ## Why these three numbers are not in Settings +# +# Rule 25 puts anything an operator might want to tune in the UI, and the +# knobs that decide what this does ARE there: whether a lane autoscales at +# all, its cap, and its floor — all DB-backed, all changeable without a +# restart. What is left here is the POLICY's internals, and exposing them +# would add four numbers per lane to a card whose whole value is being +# readable at a glance, to tune a decision the operator has a better lever +# for. If growth turns out to be too eager or too shy in practice, that is a +# reason to change these values for everyone, not to ask each operator to +# discover them. + +# All slots busy AND this many tasks waiting before a lane may grow. +# +# The AND is the design. Depth with free slots means nothing — celery is about +# to pick those up, and growing the pool would add idle children. Saturation +# with an empty queue means nothing either: the lane is busy with exactly as +# much work as exists. Only both together say "there is more work than this +# lane can reach". +AUTOSCALE_BACKLOG_THRESHOLD = 10 + +# Grow by one slot per tick, never to the cap in one jump. A lane that is +# saturated because of one slow burst settles a slot or two above where it +# started rather than at its ceiling, and the next tick re-measures rather +# than committing to a guess made once. +AUTOSCALE_STEP = 1 + +# Hysteresis: shrink only when the backlog is well BELOW the grow threshold, +# not merely under it. Equal thresholds flap — one task arriving and leaving +# would grow and shrink the lane forever at the tick interval, which is +# lesson #4183's churn arriving through a different door. +AUTOSCALE_SHRINK_BELOW = 2 + + +@dataclass +class AutoscaleDecision: + """What the autoscaler did to one lane, and why — in the operator's terms. + + A reason string on every outcome including "nothing", because an + autoscaler that only speaks when it acts is one nobody can debug when it + does not. + """ + + lane: str + action: str # "grew" | "shrank" | "held" + slots: int + reason: str + + +def autoscale_lanes_sync( + lanes: dict[str, tuple[int, int, bool]], +) -> list[AutoscaleDecision]: + """Decide and apply one round of autoscaling. + + `lanes` is name -> (slots_cap, configured_slots, autoscale_on), read from + the database by the caller — this function touches no database, for the + same reason `reconcile_lanes_sync` does not. + + ## What is current, and what is the floor + + The value this moves is the LIVE pool, read from `inspect`. The stored + `configured_slots` is what the operator set and is only the FLOOR: growth + goes above it and a shrink returns to it, never below. + + The two are deliberately not the same number, and reading the stored value + as "current" is the mistake that makes this function useless in a way no + unit test of a single tick would show. The autoscaler never writes the + row, so the stored value never moves; a tick that computed `stored + 1` + would propose the same target forever, cap the lane one slot above the + floor no matter the load, and — because resizing a replica already at the + target issues nothing and reports success — claim `grew` on every tick + while nothing changed. Lesson #4183's non-convergence, arriving with a + success message attached. + + So: `current = state.pool`, `configured` is the floor, and both `grew` and + `shrank` mean the live pool actually moved. + """ + live = inspect_lanes_sync() + depths = _queue_depths_sync() + out: list[AutoscaleDecision] = [] + + for lane in LANES: + target = lanes.get(lane.name) + if target is None: + continue + cap, configured, on = target + if not on: + continue + + state = live[lane.name] + if not state.present or state.pool is None: + # Nothing answered. Not "idle" — unknown, and a decision drawn + # from an unswept read is exactly what snippet #3969 warns about. + # `configured` is reported because there is no live number to + # report; it is what the lane will come back at. + out.append(AutoscaleDecision( + lane.name, "held", configured, "lane is not answering", + )) + continue + + current = state.pool + known = [depths.get(q) for q in lane.queues] + if all(d is None for d in known): + out.append(AutoscaleDecision( + lane.name, "held", current, "queue depth unavailable", + )) + continue + backlog = sum(d for d in known if d is not None) + state.reserved + # Against CAPACITY, not against the dial: `active` is summed across + # replicas, so comparing it to one replica's pool size would call two + # half-busy replicas of 4 saturated at 4 active and grow a lane that + # has idle slots. + saturated = state.active >= state.capacity > 0 + busy = backlog >= AUTOSCALE_BACKLOG_THRESHOLD + + if saturated and busy and current < cap: + new = min(cap, current + AUTOSCALE_STEP) + ok, err = set_lane_slots_sync(lane, new, live=state) + out.append(AutoscaleDecision( + lane.name, "grew" if ok else "held", new if ok else current, + f"{backlog} waiting and all {state.capacity} slots busy" + if ok else f"could not grow: {err}", + )) + elif saturated and busy: + # At the cap with work still waiting. Said out loud rather than + # held silently: this is the operator's own ceiling doing its job, + # and it is the moment they would want to know they set it. + out.append(AutoscaleDecision( + lane.name, "held", current, + f"{backlog} waiting but the cap is {cap}", + )) + elif current > configured and backlog <= AUTOSCALE_SHRINK_BELOW: + new = max(configured, current - AUTOSCALE_STEP) + ok, err = set_lane_slots_sync(lane, new, live=state) + out.append(AutoscaleDecision( + lane.name, "shrank" if ok else "held", new if ok else current, + f"backlog cleared, back toward {configured}" + if ok else f"could not shrink: {err}", + )) + else: + # The fixed point. A settled lane sends nothing and says so — + # the tick is one inspect and one LLEN sweep, no control messages. + out.append(AutoscaleDecision( + lane.name, "held", current, + f"{backlog} waiting, {state.active} busy", + )) + return out diff --git a/backend/app/services/worker_lanes.py b/backend/app/services/worker_lanes.py index 66521a2..c36ec52 100644 --- a/backend/app/services/worker_lanes.py +++ b/backend/app/services/worker_lanes.py @@ -136,6 +136,11 @@ class Lane: # rubber stamp and protects nobody. default_slots_cap: int default_enabled: bool + # Whether a lane arrives with the autoscaler on. False for every lane, and + # the field exists anyway: this module is the one place that describes a + # lane, and leaving one operator-settable default to the column's DDL + # default would make it the only setting you cannot read here. + default_autoscale: bool = False # True when a slot costs a copy of the ML model rather than just a process. # The only lane whose ceiling is decided by memory instead of by cores. memory_bound: bool = False diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index c7c8fe7..b13ac68 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1387,14 +1387,69 @@ def reconcile_worker_lanes() -> dict: from ..models import WorkerLane from ..services.worker_control import reconcile_lanes_sync + # Both dicts built INSIDE the session. Reading a column off a detached + # instance happens to work while the attribute is still loaded and stops + # working the moment anything expires it — a failure that would appear + # long after this line, in a sweep nobody is watching. with _sync_session_factory()() as session: - desired = { - row.name: (row.slots, row.enabled) - for row in session.execute(select(WorkerLane)).scalars() - } + rows = list(session.execute(select(WorkerLane)).scalars()) + desired = {row.name: (row.slots, row.enabled) for row in rows} + # Lanes the autoscaler may move. For these the stored value is a + # FLOOR: this sweep restores a lane that fell below it and never takes + # back what the autoscaler added, or the two would fight every five + # minutes. + autoscaling = frozenset(row.name for row in rows if row.autoscale) if not desired: # Migration 0103 seeds these, so an empty table means it has not run # yet. Nothing to assert — and inventing defaults here would let this # task disagree with the seed it is supposed to be enforcing. return {"changed": [], "skipped": [], "failed": {}} - return reconcile_lanes_sync(desired) + return reconcile_lanes_sync(desired, autoscaling) + + +@celery.task(name="backend.app.tasks.maintenance.autoscale_worker_lanes") +def autoscale_worker_lanes() -> dict: + """Grow a saturated lane, within the cap the operator set. + + Milestone 422 step 7, and the only sweep in this milestone that decides + rather than obeys. Everything else applies what someone pressed. + + OFF unless a lane opts in. A no-op costs one `celery inspect` and one LLEN + sweep and sends no control messages — the same fixed point the reconcile + holds, and for the same reason: this runs forever, so a settled system has + to be silent or the real signal drowns in its own heartbeat. + + Returns every lane's decision INCLUDING the ones it held, each with a + reason. An autoscaler that only speaks when it acts cannot be debugged on + the day it does not. + """ + from ..models import WorkerLane + from ..services.worker_control import autoscale_lanes_sync + + with _sync_session_factory()() as session: + lanes = { + # The stored `slots` is passed ONLY as the floor. What the + # autoscaler moves is the live pool, which it reads itself — this + # row is never written, so using it as the current value would pin + # every decision to the same number forever. + row.name: (row.slots_cap, row.slots, True) + for row in session.execute(select(WorkerLane)).scalars() + if row.autoscale + } + if not lanes: + return {"decisions": []} + + decisions = autoscale_lanes_sync(lanes) + for d in decisions: + if d.action != "held": + log.info( + "autoscale: %s %s to %s slots — %s", + d.lane, d.action, d.slots, d.reason, + ) + return { + "decisions": [ + {"lane": d.lane, "action": d.action, "slots": d.slots, + "reason": d.reason} + for d in decisions + ], + } diff --git a/frontend/src/components/settings/WorkerLanesCard.vue b/frontend/src/components/settings/WorkerLanesCard.vue index 14999f4..dce4fd8 100644 --- a/frontend/src/components/settings/WorkerLanesCard.vue +++ b/frontend/src/components/settings/WorkerLanesCard.vue @@ -37,6 +37,7 @@ Pending Busy Slots + Auto On @@ -62,6 +63,15 @@
{{ lane.live.replicas }} replicas
+ +
+ all slots busy for {{ laneStuckFor(lane) }} +
{{ lane.queues.join(', ') }} @@ -106,6 +116,31 @@ + + + +
+ raise the cap +
+ + import { computed, ref } from 'vue' -import { useSystemActivityStore } from '../../stores/systemActivity.js' +import { laneStuckFor, useSystemActivityStore } from '../../stores/systemActivity.js' import { formatRelative } from '../../utils/date.js' import CardHeading from '../common/CardHeading.vue' @@ -238,6 +273,17 @@ function step(lane, delta) { function toggle(lane, value) { return apply(lane, { enabled: Boolean(value) }) } + +// Room to grow into. The autoscaler moves the LIVE pool, but the floor it +// starts from is the stored value, so a lane whose floor already sits at its +// cap has nowhere to go and enabling it would do nothing at all. +function canGrow(lane) { + return lane.slots_cap > lane.slots +} + +function setAutoscale(lane, value) { + return apply(lane, { autoscale: Boolean(value) }) +}