diff --git a/README.md b/README.md index 6702b8c..2c04ee3 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,24 @@ as intuitive as Dockge, as capable as Portainer for Compose workflows. > (Auto-update) + Phase 23 (Secrets & configs) + Phase 24 (Design System v2) > complete. +## Upgrading to 0.47.0 — nothing to do + +Three pieces of state moved out of process memory and into the database, and +the live-stats endpoint got a cache. No configuration changes, no migration +steps; the new tables are created on first start. + +- **Stacks can only run one compose operation at a time.** A second `start` / + `update` / `down` on a busy stack answers `409` instead of racing the first + one over the same containers. Auto-update skips a stack you are already + deploying and picks it up next cycle. +- **`GET /api/stacks/stats` is cached for four seconds.** It sampled every + running container on every call, and the dashboard and the stacks list both + poll it every five seconds — two open tabs on a 40-container host meant a + sustained ~16 daemon calls a second. +- **The update cache and the login rate limiter persist.** Both used to reset on + restart; the rate limiter also used to multiply by the worker count, so + neither behaved as documented with `--workers` set. + ## Upgrading to 0.46.0 — everyone is signed out once Sessions now hold their access token in memory and the refresh token in an @@ -86,6 +104,10 @@ it is what your saved destination credentials are encrypted with. stop / restart / pull / update` via `docker compose`. - **Live status** — running / partial / stopped / error / updating, computed from Docker container labels. +- **One operation per stack** — a lifecycle call takes a lock (a row, so it + holds across workers and across a restart) and a second one gets `409` while + it is held; auto-update skips a stack somebody is already deploying. Locks + carry an expiry, so a worker killed mid-deploy does not strand a stack. - **Real-time logs** — streamed over WebSocket, color-coded per service. - **Live deploy console** — deploying from the editor streams `compose up` output (image pulls, container creation) over a WebSocket in real time instead @@ -133,6 +155,9 @@ it is what your saved destination credentials are encrypted with. - **Image update checker**: background task compares the local manifest digest with the registry (Docker Hub / ghcr / lscr / private v2 with token auth); update badges on the Images page + an "updates available" banner on the dashboard. + Results are cached in the database, so a restart shows the badges immediately + instead of blanking them until the next sweep — and does not re-announce + updates it already notified about. - **Port conflict detector**: pre-deploy check against host-bound ports (`/proc/net/tcp[6]`) and running container bindings, with a confirm dialog. - **Resource limits**: CPU/memory sliders in the editor → `deploy.resources.limits`. @@ -538,7 +563,7 @@ Same three commands the CI runs — `build-and-push` only starts once they pass. ```bash cd backend pip install -r requirements-dev.txt -pytest # 698 tests, no Docker daemon needed +pytest # 729 tests, no Docker daemon needed ruff check . cd ../frontend && npx tsc --noEmit -p tsconfig.json ``` @@ -564,6 +589,14 @@ HTTP. `tests/test_schema_migration.py` builds a database with the *old* user table and asserts the added column is backfilled rather than left NULL, which is what would otherwise have signed out every user on every install. +`tests/test_stack_locking.py` and `tests/test_runtime_state.py` cover the state +that moved into the database: that a busy stack answers 409 without ever +reaching Docker, that an expired lock is taken over rather than stranding the +stack, that the stats cache serves repeat callers from one sweep, and that the +update cache and rate limiter survive a restart. One of them asserts that +`update_service` never imports the database — it is shared with the agent, +which has none, so persistence has to stay opt-in. + `tests/test_bundled_templates.py` covers the 83 shipped templates: each must parse, name an image per service, keep `.env.example` in sync with the variables compose actually reads, ship every file it bind-mounts, and never come with a diff --git a/backend/main.py b/backend/main.py index 01a2c71..42a6fd2 100644 --- a/backend/main.py +++ b/backend/main.py @@ -38,7 +38,9 @@ from routers import ( ) from services import ( backup_destination_service, + image_status_store, schedule_service, + stack_lock_service, template_service, update_service, ) @@ -71,6 +73,20 @@ async def lifespan(app: FastAPI): logger.info("Migrated %d custom template(s) from the database to folders", moved) except Exception as exc: # noqa: BLE001 logger.warning("Legacy template migration failed: %s", exc) + # Runtime state that used to live in module dicts and was lost on restart. + try: + with Session(engine) as session: + stale = stack_lock_service.prune_expired(session) + if stale: + logger.info("Cleared %d stale stack lock(s) from a previous run", stale) + except Exception as exc: # noqa: BLE001 + logger.warning("Could not prune stack locks: %s", exc) + try: + restored = image_status_store.install() + logger.info("Restored %d cached image update status(es)", restored) + except Exception as exc: # noqa: BLE001 + logger.warning("Could not restore the image update cache: %s", exc) + update_task = asyncio.create_task(update_service.background_loop()) schedule_task = asyncio.create_task(schedule_service.scheduler_loop()) logger.info("StackPilot backend ready on port %s", settings.PORT) diff --git a/backend/models/__init__.py b/backend/models/__init__.py index 68f5ca0..1145e8f 100644 --- a/backend/models/__init__.py +++ b/backend/models/__init__.py @@ -4,6 +4,7 @@ from models.audit import AuditLog from models.auto_update import AutoUpdate from models.backup_destination import BackupDestination from models.backup_schedule import BackupSchedule +from models.runtime_state import ImageStatus, LoginAttempt, StackLock from models.setting import Setting, Webhook from models.stack import Stack from models.user import User @@ -11,4 +12,5 @@ from models.user import User __all__ = [ "User", "Stack", "AuditLog", "Setting", "Webhook", "Agent", "BackupDestination", "BackupSchedule", "AutoUpdate", + "StackLock", "ImageStatus", "LoginAttempt", ] diff --git a/backend/models/runtime_state.py b/backend/models/runtime_state.py new file mode 100644 index 0000000..fa2797a --- /dev/null +++ b/backend/models/runtime_state.py @@ -0,0 +1,76 @@ +"""Runtime state that used to live in module-level dicts. + +Three things were kept in process memory: which stacks are mid-deploy, the +registry digests behind the "update available" badges, and the login rate +limiter's counters. All three assumed exactly one uvicorn worker — nothing said +so, and ``--workers 2`` would have silently given each worker its own copy — +and all three were lost on restart. + +They are tables now. SQLite is already here; this needs no new dependency. + +Note that ``services/compose_service.py`` and ``services/update_service.py`` +are shared with the agent, which has no database at all — so these tables are +only ever touched from the central app's own routers and background loops. +""" +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Optional + +from sqlmodel import Field, SQLModel + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +class StackLock(SQLModel, table=True): + """A stack is mid-operation and must not be touched concurrently. + + ``docker compose`` has no locking of its own, so two simultaneous ``update`` + calls — two browser tabs, or auto-update racing a manual click — would both + run ``pull`` and ``up`` against the same project and fight over recreating + containers. + + ``expires_at`` is what keeps a crashed worker from locking a stack forever: + an expired row is simply taken over by the next caller. + """ + + stack_id: str = Field(primary_key=True) + action: str # "update", "start", "backup", … + #: Free-form owner, for the log when a lock is stolen. Not a security control. + owner: str = "" + acquired_at: datetime = Field(default_factory=_now) + expires_at: datetime + + +class ImageStatus(SQLModel, table=True): + """Cached result of one image's registry digest check. + + Persisted so a restart does not blank every update badge until the next + background sweep (up to an hour), and so ``notified`` survives with it — + otherwise every restart re-announced the same pending updates. + """ + + image: str = Field(primary_key=True) + update_available: bool = False + current_digest: Optional[str] = None + remote_digest: Optional[str] = None + checked_at: float = 0.0 + error: Optional[str] = None + #: Whether an "update available" notification already went out for this + #: image at its current state. + notified: bool = False + + +class LoginAttempt(SQLModel, table=True): + """One login attempt, for the rate limiter. + + In memory this reset on every restart, so an attacker could clear their own + budget by getting the process to restart — and with more than one worker the + limit multiplied by the worker count. Rows are pruned as they age out. + """ + + id: Optional[int] = Field(default=None, primary_key=True) + ip: str = Field(index=True) + at: datetime = Field(default_factory=_now, index=True) diff --git a/backend/routers/auth.py b/backend/routers/auth.py index 9dfea10..8dcedb2 100644 --- a/backend/routers/auth.py +++ b/backend/routers/auth.py @@ -1,14 +1,14 @@ """Authentication routes + first-launch setup wizard.""" from __future__ import annotations -import time -from collections import defaultdict, deque +from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Request, Response, status -from sqlmodel import Session, select +from sqlmodel import Session, delete, select import auth as auth_mod from database import get_session +from models.runtime_state import LoginAttempt from models.user import ( LoginRequest, RefreshRequest, @@ -23,23 +23,38 @@ from services import audit_service router = APIRouter(prefix="/api/auth", tags=["auth"]) -# Simple in-memory rate limiter for login (max 10 / minute / IP). -_LOGIN_HITS: dict[str, deque] = defaultdict(deque) +# Login rate limit: max 10 attempts per minute per client IP. +# +# Kept in the database rather than a module dict. In memory it reset on every +# restart — so an attacker could clear their own budget by getting the process +# to restart — and with more than one uvicorn worker each worker enforced its +# own limit, multiplying the real allowance by the worker count. +# +# The IP is only meaningful because uvicorn runs with --proxy-headers; without +# that every request looks like it comes from the frontend container and this +# would throttle all users together. _RATE_LIMIT = 10 -_RATE_WINDOW = 60.0 +_RATE_WINDOW = timedelta(seconds=60) +#: Attempts older than this are deleted while we are in the table anyway. +_RATE_RETENTION = timedelta(hours=1) -def _check_rate_limit(ip: str) -> None: - now = time.monotonic() - hits = _LOGIN_HITS[ip] - while hits and now - hits[0] > _RATE_WINDOW: - hits.popleft() - if len(hits) >= _RATE_LIMIT: +def _check_rate_limit(session: Session, ip: str) -> None: + now = datetime.now(timezone.utc) + session.exec(delete(LoginAttempt).where(LoginAttempt.at < now - _RATE_RETENTION)) + recent = session.exec( + select(LoginAttempt).where( + LoginAttempt.ip == ip, LoginAttempt.at >= now - _RATE_WINDOW + ) + ).all() + if len(recent) >= _RATE_LIMIT: + session.commit() raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail="Too many login attempts, slow down.", ) - hits.append(now) + session.add(LoginAttempt(ip=ip, at=now)) + session.commit() #: The refresh cookie is scoped to the two endpoints that consume it, so it is @@ -122,7 +137,7 @@ def login( session: Session = Depends(get_session), ) -> TokenPair: ip = request.client.host if request.client else "unknown" - _check_rate_limit(ip) + _check_rate_limit(session, ip) user = auth_mod.authenticate(session, body.username, body.password) if not user: raise HTTPException( diff --git a/backend/routers/stacks.py b/backend/routers/stacks.py index f9a5d5d..d1015d8 100644 --- a/backend/routers/stacks.py +++ b/backend/routers/stacks.py @@ -27,7 +27,15 @@ from models.setting import ( ) from models.auto_update import AutoUpdateRead, AutoUpdateWrite from models.user import User -from services import audit_service, auto_update_service, compose_service, notify_service, stats_service, update_service +from services import ( + audit_service, + auto_update_service, + compose_service, + notify_service, + stack_lock_service, + stats_service, + update_service, +) from services.convert_service import convert_docker_run router = APIRouter(prefix="/api/stacks", tags=["stacks"]) @@ -59,13 +67,17 @@ def _get_stack_or_404(session: Session, stack_id: str) -> Stack: return stack -def _stack_summary(stack: Stack, summaries: dict | None = None) -> dict: +def _stack_summary( + stack: Stack, summaries: dict | None = None, busy: dict[str, str] | None = None +) -> dict: """Build a list-row summary. - Pass ``summaries`` (from :func:`compose_service.stack_status_summaries`) to - serve the whole stacks list from a single Docker call. Without it (single + Pass ``summaries`` (from :func:`compose_service.stack_status_summaries`) and + ``busy`` (from :func:`stack_lock_service.active`) to serve the whole stacks + list from one Docker call and one query. Without them (single create/update/clone responses), fall back to one direct query for this stack. """ + busy = busy or {} if summaries is None: try: containers = compose_service.containers_for_stack(stack.id) @@ -79,7 +91,7 @@ def _stack_summary(stack: Stack, summaries: dict | None = None) -> dict: info = summaries.get(stack.id) total = info["total"] if info else 0 running = info["running"] if info else 0 - if compose_service.is_busy(stack.id): + if stack.id in busy: status = "updating" else: status = info["status"] if info else "stopped" @@ -111,7 +123,8 @@ def list_stacks( summaries = compose_service.stack_status_summaries() except DockerError: summaries = {} - return [_stack_summary(s, summaries) for s in stacks] + busy = stack_lock_service.active(session) + return [_stack_summary(s, summaries, busy) for s in stacks] @router.post("", status_code=201) @@ -296,7 +309,17 @@ async def _notify_lifecycle(action_name: str, stack_id: str, ok: bool, detail: s async def _lifecycle(action_fn, action_name, stack_id, request, session, user): _get_stack_or_404(session, stack_id) - result = await action_fn(stack_id) + # One compose operation per stack. Without this two tabs (or auto-update + # landing on a stack somebody just clicked) both run pull + up -d against + # the same project and race over recreating containers. + try: + with stack_lock_service.hold(session, stack_id, action_name, user.username): + result = await action_fn(stack_id) + except stack_lock_service.StackBusy as exc: + raise HTTPException( + status_code=409, + detail=f"Stack '{stack_id}' is busy: {exc.action} in progress", + ) from exc audit_service.record( session, user=user.username, action=f"stack.{action_name}", target=stack_id, detail=f"rc={result.get('returncode')}", ip=_client_ip(request), diff --git a/backend/routers/ws.py b/backend/routers/ws.py index dde8121..bf07d3d 100644 --- a/backend/routers/ws.py +++ b/backend/routers/ws.py @@ -22,6 +22,7 @@ from services import ( compose_service, exec_service, notify_service, + stack_lock_service, update_service, ) @@ -145,6 +146,18 @@ async def ws_deploy( rc: int | None = None disconnected = False + # Same guard the REST lifecycle uses — the deploy console runs the very + # same `compose up`, so it has to queue behind an in-flight operation + # rather than race it. + lock_session = Session(engine) + try: + stack_lock_service.acquire(lock_session, stack_id, "start", username) + except stack_lock_service.StackBusy as exc: + lock_session.close() + with contextlib.suppress(Exception): + await websocket.send_text(json.dumps({"type": "error", "detail": str(exc)})) + await websocket.close(code=4409) + return compose_service.mark_busy(stack_id) try: async for kind, payload in compose_service.stream_up(stack_id): @@ -162,6 +175,9 @@ async def ws_deploy( await websocket.send_text(json.dumps({"type": "error", "detail": str(exc)})) finally: compose_service.clear_busy(stack_id) + with contextlib.suppress(Exception): + stack_lock_service.release(lock_session, stack_id) + lock_session.close() ok = rc in (0, None) try: @@ -275,6 +291,15 @@ async def ws_update( rc: int | None = None disconnected = False + lock_session = Session(engine) + try: + stack_lock_service.acquire(lock_session, stack_id, "update", username) + except stack_lock_service.StackBusy as exc: + lock_session.close() + with contextlib.suppress(Exception): + await websocket.send_text(json.dumps({"type": "error", "detail": str(exc)})) + await websocket.close(code=4409) + return compose_service.mark_busy(stack_id) try: async for kind, payload in compose_service.stream_update(stack_id): @@ -291,6 +316,9 @@ async def ws_update( await websocket.send_text(json.dumps({"type": "error", "detail": str(exc)})) finally: compose_service.clear_busy(stack_id) + with contextlib.suppress(Exception): + stack_lock_service.release(lock_session, stack_id) + lock_session.close() ok = rc in (0, None) try: diff --git a/backend/services/auto_update_service.py b/backend/services/auto_update_service.py index a00032e..03ef951 100644 --- a/backend/services/auto_update_service.py +++ b/backend/services/auto_update_service.py @@ -20,7 +20,13 @@ from database import engine from models.agent import Agent from models.auto_update import AutoUpdate from models.setting import EVENT_PULL_FAILED, EVENT_STACK_AUTO_UPDATED -from services import agent_service, compose_service, notify_service, update_service +from services import ( + agent_service, + compose_service, + notify_service, + stack_lock_service, + update_service, +) logger = logging.getLogger("stackpilot.autoupdate") @@ -104,8 +110,15 @@ async def _run_local(session: Session, policy: AutoUpdate) -> None: prev = policy.last_status if policy.redeploy: try: - await compose_service.pull(stack_id) - await compose_service.up(stack_id) + # Never redeploy underneath somebody: if a user is mid-deploy on + # this stack, skip and pick it up next cycle rather than racing + # them over the same containers. + with stack_lock_service.hold(session, stack_id, "auto-update", "auto-update"): + await compose_service.pull(stack_id) + await compose_service.up(stack_id) + except stack_lock_service.StackBusy as exc: + _record(session, policy, "skipped", str(exc)) + return except Exception as exc: # noqa: BLE001 _record(session, policy, "error", str(exc)) await _safe_notify(EVENT_PULL_FAILED, f"Auto-update of '{stack_id}' failed", str(exc), session) diff --git a/backend/services/image_status_store.py b/backend/services/image_status_store.py new file mode 100644 index 0000000..5f4b04f --- /dev/null +++ b/backend/services/image_status_store.py @@ -0,0 +1,80 @@ +"""Persistence for the image update cache. + +``update_service`` holds the registry-digest results in a module dict because +it is shared with the agent, which has no database. This module is the central +app's half: it seeds that dict at startup and mirrors every write back into +SQLite, wired up in ``main.lifespan``. + +What it buys: after a restart the update badges are there immediately instead +of blank until the next background sweep (up to an hour), and the "already +notified" marks come back with them, so a restart no longer re-announces +updates the user has already seen. +""" +from __future__ import annotations + +import logging + +from sqlmodel import Session, select + +from database import engine +from models.runtime_state import ImageStatus +from services import update_service + +logger = logging.getLogger("stackpilot.image_status") + + +def _to_status(row: ImageStatus) -> update_service.UpdateStatus: + return update_service.UpdateStatus( + image=row.image, + update_available=row.update_available, + current_digest=row.current_digest, + remote_digest=row.remote_digest, + checked_at=row.checked_at, + error=row.error, + ) + + +def save(status: update_service.UpdateStatus, notified: bool) -> None: + """Upsert one image's status. Opens its own session — the caller is the + background loop, which has none.""" + with Session(engine) as session: + row = session.get(ImageStatus, status.image) + if row is None: + row = ImageStatus(image=status.image) + row.update_available = status.update_available + row.current_digest = status.current_digest + row.remote_digest = status.remote_digest + row.checked_at = status.checked_at + row.error = status.error + row.notified = notified + session.add(row) + session.commit() + + +def install() -> int: + """Seed the in-memory cache from the database and start mirroring writes. + + Returns how many entries were restored. + """ + with Session(engine) as session: + rows = session.exec(select(ImageStatus)).all() + update_service.restore_cache([(_to_status(r), r.notified) for r in rows]) + update_service.set_persist_callback(save, prune) + return len(rows) + + +def prune(keep: set[str]) -> int: + """Drop rows for images that are no longer used by any stack. + + Without this the table grows for the life of the install, one row per image + tag that was ever running. + """ + removed = 0 + with Session(engine) as session: + for row in session.exec(select(ImageStatus)).all(): + if row.image not in keep: + session.delete(row) + removed += 1 + if removed: + session.commit() + return removed diff --git a/backend/services/stack_lock_service.py b/backend/services/stack_lock_service.py new file mode 100644 index 0000000..100b171 --- /dev/null +++ b/backend/services/stack_lock_service.py @@ -0,0 +1,148 @@ +"""One compose operation per stack at a time. + +``docker compose`` does no locking. Two ``update`` calls against the same +project — two open browser tabs, or the auto-update pass landing on a stack +somebody just clicked — both run ``pull`` and then ``up -d``, and race each +other recreating the same containers. + +There *was* a busy flag (``compose_service._BUSY``), but it only ever fed the +status column: no lifecycle handler consulted it before acting. This module is +the actual guard, and it lives in the database so it holds across workers and +across a restart. + +``compose_service`` keeps its in-process set because it is shared with the +agent, which has no database. The agent is a single process managing one host, +and the central app holds this lock before calling it, so the two do not +conflict. +""" +from __future__ import annotations + +import logging +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from typing import Optional + +from sqlalchemy.exc import IntegrityError +from sqlmodel import Session, delete, select + +from models.runtime_state import StackLock + +logger = logging.getLogger("stackpilot.stack_lock") + +#: Long enough to outlast the slowest legitimate operation (compose commands +#: time out at 600s, a full pull of a large stack can chain several), short +#: enough that a lock orphaned by a killed worker clears itself within an hour. +DEFAULT_TTL = timedelta(minutes=30) + + +class StackBusy(Exception): + """The stack is already running an operation.""" + + def __init__(self, stack_id: str, action: str): + self.stack_id = stack_id + self.action = action + super().__init__(f"Stack '{stack_id}' is busy: {action} in progress") + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _aware(value: Optional[datetime]) -> Optional[datetime]: + """SQLite hands datetimes back naive; compare them as UTC.""" + if value is not None and value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +def acquire( + session: Session, + stack_id: str, + action: str, + owner: str = "", + ttl: timedelta = DEFAULT_TTL, +) -> None: + """Take the lock for ``stack_id`` or raise :class:`StackBusy`. + + An expired lock is taken over — that is the recovery path for a worker that + died mid-deploy, which would otherwise leave the stack unusable. + """ + now = _now() + existing = session.get(StackLock, stack_id) + if existing is not None: + if (_aware(existing.expires_at) or now) > now: + raise StackBusy(stack_id, existing.action) + logger.warning( + "Taking over an expired %s lock on '%s' (held by %r since %s)", + existing.action, stack_id, existing.owner, existing.acquired_at, + ) + session.delete(existing) + session.commit() + + session.add( + StackLock( + stack_id=stack_id, + action=action, + owner=owner, + acquired_at=now, + expires_at=now + ttl, + ) + ) + try: + session.commit() + except IntegrityError as exc: + # Another worker inserted between our check and our commit. The primary + # key is what actually makes this safe; the read above is only there to + # give a useful error and to clear stale rows. + session.rollback() + raise StackBusy(stack_id, action) from exc + + +def release(session: Session, stack_id: str) -> None: + """Drop the lock. Safe to call when it is not held.""" + existing = session.get(StackLock, stack_id) + if existing is not None: + session.delete(existing) + session.commit() + + +@contextmanager +def hold(session: Session, stack_id: str, action: str, owner: str = ""): + """Hold the lock for the duration of the block. + + Raises :class:`StackBusy` if somebody else has it. Always releases, so a + failed deploy does not leave the stack locked. + """ + acquire(session, stack_id, action, owner) + try: + yield + finally: + try: + release(session, stack_id) + except Exception: # noqa: BLE001 - never mask the original error + logger.exception("Failed to release the lock on '%s'", stack_id) + + +def active(session: Session) -> dict[str, str]: + """``{stack_id: action}`` for every lock still in force. + + One query for the whole stacks list, rather than a lookup per row. + """ + now = _now() + return { + lock.stack_id: lock.action + for lock in session.exec(select(StackLock)).all() + if (_aware(lock.expires_at) or now) > now + } + + +def is_busy(session: Session, stack_id: str) -> bool: + lock = session.get(StackLock, stack_id) + return lock is not None and (_aware(lock.expires_at) or _now()) > _now() + + +def prune_expired(session: Session) -> int: + """Drop locks that have timed out. Called at startup and by the scheduler.""" + result = session.exec(delete(StackLock).where(StackLock.expires_at < _now())) + session.commit() + return result.rowcount or 0 diff --git a/backend/services/stats_service.py b/backend/services/stats_service.py index 3d8ffd5..8157217 100644 --- a/backend/services/stats_service.py +++ b/backend/services/stats_service.py @@ -3,15 +3,33 @@ Reads a one-shot ``docker stats`` sample per running container (the daemon includes ``precpu_stats`` so a single read yields a usable CPU delta) and sums them by ``com.docker.compose.project`` label, which equals the stack id. + +Sampling is not free: it is one blocking call to the daemon *per running +container*, and the dashboard and the stacks list both poll this every five +seconds. Two open tabs on a 40-container host meant a sustained ~16 samples a +second. Results are therefore cached for :data:`CACHE_TTL`, the same shape +``dashboard_service`` already uses for its fleet aggregate — one sweep serves +every reader in the window, and the numbers stay well inside what a +five-second poll can show. """ from __future__ import annotations +import threading +import time from concurrent.futures import ThreadPoolExecutor from docker_client import DockerError, get_client, safe_call COMPOSE_PROJECT_LABEL = "com.docker.compose.project" +#: Slightly under the frontend's 5s poll, so a refresh usually gets fresh +#: numbers while concurrent readers still share one sweep. +CACHE_TTL = 4.0 + +_cache: dict = {"data": None, "ts": 0.0} +# Held across the sample so N simultaneous callers trigger one sweep, not N. +_lock = threading.Lock() + def _container_stats(container) -> dict | None: try: @@ -60,12 +78,30 @@ def _container_stats(container) -> dict | None: } -def stack_stats() -> dict: +def stack_stats(refresh: bool = False) -> dict: """Return {stack_id: {cpu_used, cpu_limit, mem_used, mem_limit, containers}}. Limits are the summed assigned limits across the stack's containers, or null - when none of them have that limit set. + when none of them have that limit set. Served from a short-lived cache + unless ``refresh`` is set. """ + if not refresh and _cache["data"] is not None: + if time.monotonic() - _cache["ts"] < CACHE_TTL: + return _cache["data"] + + with _lock: + # Somebody may have refreshed it while we waited for the lock. + if not refresh and _cache["data"] is not None: + if time.monotonic() - _cache["ts"] < CACHE_TTL: + return _cache["data"] + data = _sample() + _cache["data"] = data + _cache["ts"] = time.monotonic() + return data + + +def _sample() -> dict: + """One full sweep across every running container.""" try: client = get_client() containers = safe_call(client.containers.list) # running only diff --git a/backend/services/update_service.py b/backend/services/update_service.py index 3eda132..736c214 100644 --- a/backend/services/update_service.py +++ b/backend/services/update_service.py @@ -10,6 +10,7 @@ import asyncio import logging import time from dataclasses import asdict, dataclass +from collections.abc import Callable from typing import Optional import httpx @@ -50,6 +51,49 @@ _CACHE: dict[str, UpdateStatus] = {} # background loop doesn't re-notify on every cycle. _NOTIFIED: set[str] = set() +#: Optional sink for cache writes. +#: +#: This module is shared with the agent, which has no database, so persistence +#: cannot live here. The central app registers a callback that mirrors each +#: entry into SQLite (see ``services/image_status_store.py``) and seeds the +#: cache from it at startup; the agent registers nothing and behaves exactly as +#: before. Without it a restart blanked every update badge until the next +#: background sweep — up to an hour — and re-announced updates it had already +#: notified about. +_persist_cb: Optional[Callable[[UpdateStatus, bool], None]] = None + + +#: Optional sink for "these images are still in use", same opt-in shape as +#: _persist_cb. Keeps both the dict and the table from growing one entry per +#: image tag that was ever running, for the life of the install. +_prune_cb: Optional[Callable[[set], int]] = None + + +def set_persist_callback( + callback: Optional[Callable[[UpdateStatus, bool], None]], + prune: Optional[Callable[[set], int]] = None, +) -> None: + global _persist_cb, _prune_cb + _persist_cb = callback + _prune_cb = prune + + +def restore_cache(entries: list[tuple[UpdateStatus, bool]]) -> None: + """Seed the in-memory cache from persisted rows at startup.""" + for status, notified in entries: + _CACHE[status.image] = status + if notified: + _NOTIFIED.add(status.image) + + +def _store(status: UpdateStatus, notified: bool) -> None: + _CACHE[status.image] = status + if _persist_cb is not None: + try: + _persist_cb(status, notified) + except Exception as exc: # noqa: BLE001 - persistence is best-effort + logger.debug("Could not persist update status for %s: %s", status.image, exc) + # --------------------------------------------------------------------------- # # Image reference parsing @@ -168,9 +212,11 @@ async def check_image(image: str) -> UpdateStatus: checked_at=time.time(), error=error, ) - _CACHE[image] = status if update_available and image not in _NOTIFIED: + # Marked before the attempt, not after: a notifier that is down should + # not make every cycle re-announce the same update. _NOTIFIED.add(image) + _store(status, True) try: await notify_service.notify( EVENT_UPDATE_AVAILABLE, @@ -179,8 +225,10 @@ async def check_image(image: str) -> UpdateStatus: ) except Exception as exc: # noqa: BLE001 - notifications are best-effort logger.debug("update notify failed for %s: %s", image, exc) - elif not update_available: - _NOTIFIED.discard(image) + else: + if not update_available: + _NOTIFIED.discard(image) + _store(status, image in _NOTIFIED) return status @@ -291,9 +339,22 @@ async def check_all() -> dict[str, dict]: images = _all_running_images() for image in images: await check_image(image) + _forget_unused(set(images)) return {k: v.to_dict() for k, v in _CACHE.items()} +def _forget_unused(keep: set) -> None: + """Drop images no running container references any more.""" + for image in [i for i in _CACHE if i not in keep]: + del _CACHE[image] + _NOTIFIED.discard(image) + if _prune_cb is not None: + try: + _prune_cb(keep) + except Exception as exc: # noqa: BLE001 - housekeeping is best-effort + logger.debug("Could not prune persisted update statuses: %s", exc) + + def get_cache() -> dict[str, dict]: return {k: v.to_dict() for k, v in _CACHE.items()} diff --git a/backend/tests/test_runtime_state.py b/backend/tests/test_runtime_state.py new file mode 100644 index 0000000..be5887f --- /dev/null +++ b/backend/tests/test_runtime_state.py @@ -0,0 +1,284 @@ +"""State that used to live in module dicts (F10, F11). + +Three things were kept in process memory, all of them assuming exactly one +uvicorn worker without saying so, and all of them lost on restart: + +* the live stats sweep, which was not cached at all and got polled every five + seconds by two pages at once; +* the registry digests behind the update badges, so a restart blanked every + badge for up to an hour and re-announced updates already notified about; +* the login rate limiter, which an attacker could reset by getting the process + to restart, and which multiplied by the worker count. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlmodel import Session, delete, select + + +# --------------------------------------------------------------------------- # +# F10 — the stats cache +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def stats(monkeypatch): + from services import stats_service + + stats_service._cache["data"] = None + stats_service._cache["ts"] = 0.0 + return stats_service + + +def test_repeat_calls_inside_the_window_reuse_one_sweep(stats, monkeypatch): + """The dashboard and the stacks list both poll this every 5s. + + Each sweep is one blocking daemon call per running container, so serving + them from one sample is the whole point. + """ + calls = [] + monkeypatch.setattr(stats, "_sample", lambda: calls.append(1) or {"a": {}}) + + assert stats.stack_stats() == {"a": {}} + assert stats.stack_stats() == {"a": {}} + assert stats.stack_stats() == {"a": {}} + assert len(calls) == 1 + + +def test_the_cache_expires(stats, monkeypatch): + calls = [] + monkeypatch.setattr(stats, "_sample", lambda: calls.append(1) or {}) + + stats.stack_stats() + # Pretend the window has passed rather than sleeping through it. + stats._cache["ts"] -= stats.CACHE_TTL + 1 + stats.stack_stats() + assert len(calls) == 2 + + +def test_refresh_bypasses_the_cache(stats, monkeypatch): + calls = [] + monkeypatch.setattr(stats, "_sample", lambda: calls.append(1) or {}) + + stats.stack_stats() + stats.stack_stats(refresh=True) + assert len(calls) == 2 + + +def test_a_failing_sample_is_not_cached_as_a_success(stats, monkeypatch): + """A Docker outage returns {} — it must not stick for the whole window + once the daemon is back.""" + monkeypatch.setattr(stats, "_sample", dict) + assert stats.stack_stats() == {} + + monkeypatch.setattr(stats, "_sample", lambda: {"a": {"containers": 1}}) + assert stats.stack_stats(refresh=True) == {"a": {"containers": 1}} + + +# --------------------------------------------------------------------------- # +# F11 — the image update cache survives a restart +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def clean_image_status(db): + from database import engine + from models.runtime_state import ImageStatus + from services import update_service + + with Session(engine) as session: + session.exec(delete(ImageStatus)) + session.commit() + update_service._CACHE.clear() + update_service._NOTIFIED.clear() + update_service.set_persist_callback(None) + yield + update_service.set_persist_callback(None) + update_service._CACHE.clear() + update_service._NOTIFIED.clear() + + +def test_a_saved_status_comes_back_after_a_restart(clean_image_status): + """The badge is there immediately instead of blank until the next sweep.""" + from services import image_status_store, update_service + + status = update_service.UpdateStatus( + image="nginx:latest", + update_available=True, + current_digest="sha256:old", + remote_digest="sha256:new", + checked_at=123.0, + ) + image_status_store.save(status, notified=True) + + # Simulate a restart: memory is empty, then install() seeds it. + update_service._CACHE.clear() + update_service._NOTIFIED.clear() + assert image_status_store.install() == 1 + + cached = update_service.get_cache()["nginx:latest"] + assert cached["update_available"] is True + assert cached["remote_digest"] == "sha256:new" + # And the "already told the user" mark comes back with it, so the restart + # does not re-announce the same update. + assert "nginx:latest" in update_service._NOTIFIED + + +def test_installing_starts_mirroring_writes(clean_image_status): + from database import engine + from models.runtime_state import ImageStatus + from services import image_status_store, update_service + + image_status_store.install() + update_service._store( + update_service.UpdateStatus( + image="redis:7", update_available=False, current_digest="sha256:a", + remote_digest="sha256:a", checked_at=1.0, + ), + notified=False, + ) + with Session(engine) as session: + row = session.get(ImageStatus, "redis:7") + assert row is not None and row.update_available is False + + +def test_saving_the_same_image_twice_updates_it(clean_image_status): + from database import engine + from models.runtime_state import ImageStatus + from services import image_status_store, update_service + + def status(remote): + return update_service.UpdateStatus( + image="app:1", update_available=True, current_digest="sha256:x", + remote_digest=remote, checked_at=1.0, + ) + + image_status_store.save(status("sha256:one"), notified=False) + image_status_store.save(status("sha256:two"), notified=True) + + with Session(engine) as session: + rows = session.exec(select(ImageStatus).where(ImageStatus.image == "app:1")).all() + assert len(rows) == 1 + assert rows[0].remote_digest == "sha256:two" + assert rows[0].notified is True + + +def test_pruning_drops_images_no_stack_uses_any_more(clean_image_status): + from database import engine + from models.runtime_state import ImageStatus + from services import image_status_store, update_service + + for image in ("kept:1", "gone:1"): + image_status_store.save( + update_service.UpdateStatus( + image=image, update_available=False, current_digest=None, + remote_digest=None, checked_at=0.0, + ), + notified=False, + ) + + assert image_status_store.prune(keep={"kept:1"}) == 1 + with Session(engine) as session: + assert [r.image for r in session.exec(select(ImageStatus)).all()] == ["kept:1"] + + +def test_the_agent_has_no_persistence_wired_up(clean_image_status): + """``update_service`` is shared with the agent, which has no database. + + Persistence therefore has to be opt-in, registered by the central app — + if this module ever imports the database directly, the agent breaks. + """ + import inspect + + from services import update_service + + source = inspect.getsource(update_service) + assert "from database import" not in source + assert "import database" not in source + assert update_service._persist_cb is None + + +# --------------------------------------------------------------------------- # +# F11 — the login rate limiter +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def clean_attempts(db): + from database import engine + from models.runtime_state import LoginAttempt + + with Session(engine) as session: + session.exec(delete(LoginAttempt)) + session.commit() + yield + + +def test_the_eleventh_attempt_in_a_minute_is_refused(client, clean_attempts): + body = {"username": "nobody", "password": "wrong"} + for _ in range(10): + assert client.post("/api/auth/login", json=body).status_code == 401 + assert client.post("/api/auth/login", json=body).status_code == 429 + + +def test_the_limit_survives_a_restart(db, clean_attempts): + """In memory this reset on every restart, so an attacker could clear their + own budget by getting the process to fall over.""" + from database import engine + from fastapi import HTTPException + from routers import auth as auth_router + + with Session(engine) as session: + for _ in range(10): + auth_router._check_rate_limit(session, "10.0.0.9") + + # A fresh session stands in for a fresh process — the counter is a table. + with Session(engine) as session: + with pytest.raises(HTTPException) as excinfo: + auth_router._check_rate_limit(session, "10.0.0.9") + assert excinfo.value.status_code == 429 + + +def test_different_clients_have_separate_budgets(db, clean_attempts): + """Only meaningful because uvicorn runs with --proxy-headers; without it + every request carries the frontend container's IP and one client would + throttle everyone.""" + from database import engine + from fastapi import HTTPException + from routers import auth as auth_router + + with Session(engine) as session: + for _ in range(10): + auth_router._check_rate_limit(session, "10.0.0.1") + auth_router._check_rate_limit(session, "10.0.0.2") # must not raise + with pytest.raises(HTTPException): + auth_router._check_rate_limit(session, "10.0.0.1") + + +def test_attempts_age_out_of_the_window(db, clean_attempts): + from database import engine + from models.runtime_state import LoginAttempt + from routers import auth as auth_router + + old = datetime.now(timezone.utc) - timedelta(minutes=5) + with Session(engine) as session: + for _ in range(10): + session.add(LoginAttempt(ip="10.0.0.3", at=old)) + session.commit() + auth_router._check_rate_limit(session, "10.0.0.3") # window has moved on + + +def test_stale_rows_are_pruned(db, clean_attempts): + from database import engine + from models.runtime_state import LoginAttempt + from routers import auth as auth_router + + ancient = datetime.now(timezone.utc) - timedelta(days=2) + with Session(engine) as session: + session.add(LoginAttempt(ip="10.0.0.4", at=ancient)) + session.commit() + auth_router._check_rate_limit(session, "10.0.0.5") + remaining = session.exec(select(LoginAttempt)).all() + assert all(r.ip != "10.0.0.4" for r in remaining), "the table must not grow forever" diff --git a/backend/tests/test_stack_locking.py b/backend/tests/test_stack_locking.py new file mode 100644 index 0000000..35a1456 --- /dev/null +++ b/backend/tests/test_stack_locking.py @@ -0,0 +1,206 @@ +"""One compose operation per stack (F7), and the lock surviving a restart (F11). + +``docker compose`` does no locking of its own. Before 0.47.0 nothing stopped two +``update`` calls landing on the same project at once — two browser tabs, or the +auto-update pass picking up a stack somebody had just clicked — and both would +run ``pull`` then ``up -d`` and fight over recreating the same containers. + +There was a busy flag, but it only fed the status column: no handler consulted +it before acting. The lock is a table now, so it also holds across workers and +across a restart. +""" +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlmodel import Session + + +@pytest.fixture +def lock_session(db): + from database import engine + + with Session(engine) as session: + yield session + + +@pytest.fixture(autouse=True) +def clean_locks(db): + """Locks are global state; don't let one test leak into the next.""" + from database import engine + from models.runtime_state import StackLock + from sqlmodel import delete + + with Session(engine) as session: + session.exec(delete(StackLock)) + session.commit() + yield + + +@pytest.fixture +def svc(): + from services import stack_lock_service + + return stack_lock_service + + +# --------------------------------------------------------------------------- # +# The lock itself +# --------------------------------------------------------------------------- # + + +def test_a_second_caller_is_refused(svc, lock_session): + svc.acquire(lock_session, "demo", "update", "alice") + with pytest.raises(svc.StackBusy) as excinfo: + svc.acquire(lock_session, "demo", "start", "bob") + assert excinfo.value.action == "update", "the error names the operation in flight" + + +def test_releasing_frees_it(svc, lock_session): + svc.acquire(lock_session, "demo", "update") + svc.release(lock_session, "demo") + svc.acquire(lock_session, "demo", "start") # must not raise + + +def test_releasing_a_lock_nobody_holds_is_harmless(svc, lock_session): + svc.release(lock_session, "never-locked") + + +def test_different_stacks_do_not_block_each_other(svc, lock_session): + svc.acquire(lock_session, "one", "update") + svc.acquire(lock_session, "two", "update") + + +def test_hold_releases_even_when_the_operation_fails(svc, lock_session): + """A failed deploy must not leave the stack locked.""" + with pytest.raises(RuntimeError): + with svc.hold(lock_session, "demo", "update"): + raise RuntimeError("compose blew up") + assert not svc.is_busy(lock_session, "demo") + + +def test_an_expired_lock_is_taken_over(svc, lock_session): + """The recovery path for a worker killed mid-deploy. + + Without it the stack would stay locked forever and the only fix would be + editing the database by hand. + """ + from models.runtime_state import StackLock + + past = datetime.now(timezone.utc) - timedelta(minutes=5) + lock_session.add( + StackLock( + stack_id="orphaned", + action="update", + owner="a worker that died", + acquired_at=past - timedelta(minutes=30), + expires_at=past, + ) + ) + lock_session.commit() + + assert not svc.is_busy(lock_session, "orphaned") + svc.acquire(lock_session, "orphaned", "start") # must not raise + + +def test_active_lists_current_locks_and_skips_expired(svc, lock_session): + from models.runtime_state import StackLock + + svc.acquire(lock_session, "live", "update") + lock_session.add( + StackLock( + stack_id="dead", + action="start", + expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), + ) + ) + lock_session.commit() + + assert svc.active(lock_session) == {"live": "update"} + + +def test_prune_removes_only_expired_locks(svc, lock_session): + from models.runtime_state import StackLock + + svc.acquire(lock_session, "live", "update") + lock_session.add( + StackLock( + stack_id="dead", + action="start", + expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), + ) + ) + lock_session.commit() + + assert svc.prune_expired(lock_session) == 1 + assert svc.active(lock_session) == {"live": "update"} + + +def test_the_lock_outlives_the_session_that_took_it(svc, db): + """It is a row, not a set in one worker's memory — that is the point.""" + from database import engine + + with Session(engine) as first: + svc.acquire(first, "persisted", "update") + + with Session(engine) as second: + assert svc.is_busy(second, "persisted") + with pytest.raises(svc.StackBusy): + svc.acquire(second, "persisted", "start") + + +# --------------------------------------------------------------------------- # +# Enforcement at the HTTP entry point +# --------------------------------------------------------------------------- # + + +@pytest.fixture +def locked_stack(db): + """A registered stack that is already mid-operation.""" + from database import engine + from models.stack import Stack + from services import compose_service, stack_lock_service + + stack_id = "lock-fixture" + compose_service.write_compose(stack_id, "services:\n a:\n image: alpine\n") + with Session(engine) as session: + if not session.get(Stack, stack_id): + session.add(Stack(id=stack_id, name="lock fixture")) + session.commit() + stack_lock_service.acquire(session, stack_id, "update", "someone-else") + return stack_id + + +@pytest.mark.parametrize("action", ["start", "stop", "restart", "pull", "update", "down"]) +def test_lifecycle_calls_are_refused_while_a_stack_is_busy(as_admin, locked_stack, action): + """409, and crucially *without* reaching Docker — the guard is in front.""" + response = as_admin.post(f"/api/stacks/{locked_stack}/{action}") + assert response.status_code == 409 + assert "busy" in response.json()["detail"].lower() + assert "update in progress" in response.json()["detail"] + + +def test_the_stacks_list_shows_a_busy_stack_as_updating(as_admin, locked_stack): + rows = {row["id"]: row for row in as_admin.get("/api/stacks").json()} + assert rows[locked_stack]["status"] == "updating" + + +def test_an_unlocked_stack_is_not_refused(as_admin, db): + """The guard must not block ordinary use. + + Without a Docker daemon the call fails further in — anything but 409 means + it got past the lock, which is what this asserts. + """ + from database import engine + from models.stack import Stack + from services import compose_service + + stack_id = "unlocked-fixture" + compose_service.write_compose(stack_id, "services:\n a:\n image: alpine\n") + with Session(engine) as session: + if not session.get(Stack, stack_id): + session.add(Stack(id=stack_id, name="unlocked fixture")) + session.commit() + + assert as_admin.post(f"/api/stacks/{stack_id}/start").status_code != 409 diff --git a/backend/version.py b/backend/version.py index 2ab99a9..7f31f5d 100644 --- a/backend/version.py +++ b/backend/version.py @@ -1,3 +1,3 @@ """Single source of truth for the StackPilot release version.""" -APP_VERSION = "0.46.0" +APP_VERSION = "0.47.0" diff --git a/frontend/package.json b/frontend/package.json index 9e5a17e..7a23481 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,7 +1,7 @@ { "name": "stackpilot-frontend", "private": true, - "version": "0.46.0", + "version": "0.47.0", "type": "module", "scripts": { "dev": "vite",