bvandeusenandClaude Opus 5.5
fb99f8b551
#5816 "Restoring mode": while a backup goes back in, the scheduler holds, ingest and HTMX answer 503 with Retry-After (agents keep their samples), and every page says what is happening. A safety backup of Steward as it stands is made first, to /data/backups and the bucket. The backup's key is put in place, the restore audited, and Steward exits with code 3 so Swarm, compose or systemd starts it fresh. A failed in-place restore changes nothing and says why on Backups; a full replace that fails after the wipe restarts into first-run setup. #5817 "Settings → Backups: Restore": a Restore button on each sent backup and on the file card. The chosen backup is checked (the same comparison as Check, now a shared partial) and held server-side for an hour under a token; it starts only once the admin types "restore", and an older backup also needs "Replace everything" ticked. A newer backup, or one whose key can't open its secrets, gets no button. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from typing import Callable, Coroutine
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ScheduledTask:
|
|
name: str
|
|
coro_factory: Callable[[], Coroutine]
|
|
interval_seconds: int
|
|
run_on_startup: bool = False
|
|
|
|
|
|
@dataclass
|
|
class _DueTracker:
|
|
"""Decides which scheduled tasks are due, with a self-overlap guard.
|
|
|
|
Pure (no asyncio, no clock of its own — `now` is passed in) so the
|
|
scheduling policy is unit-testable without timing races. A task whose prior
|
|
run is still in flight is NOT re-fired: overlapping poll runs stack up open
|
|
connections/subprocesses and amplify any per-poll resource use — the same
|
|
failure mode behind the fd-leak lockups. The skipped task is retried on the
|
|
next tick once it completes (its last_run isn't advanced while skipped).
|
|
"""
|
|
last_run: dict[str, float] = field(default_factory=dict)
|
|
in_flight: set[str] = field(default_factory=set)
|
|
|
|
def due(self, tasks: list[ScheduledTask], now: float) -> list[ScheduledTask]:
|
|
ready: list[ScheduledTask] = []
|
|
for task in tasks:
|
|
if now - self.last_run.get(task.name, 0) < task.interval_seconds:
|
|
continue
|
|
if task.name in self.in_flight:
|
|
logger.warning(
|
|
"Scheduled task %r still running — skipping this tick",
|
|
task.name)
|
|
continue
|
|
ready.append(task)
|
|
return ready
|
|
|
|
def begin(self, tasks: list[ScheduledTask], now: float) -> None:
|
|
"""Start the clock: a task that doesn't run on startup is first due
|
|
one interval from now. Without this a task that had never run counted
|
|
from 0 on the loop's clock (host uptime), so every one of them fired
|
|
on the first tick after a deploy, the hourly cleanup with its Docker
|
|
log prune among them, holding up agent ingest for minutes (#5431)."""
|
|
for task in tasks:
|
|
if not task.run_on_startup:
|
|
self.last_run.setdefault(task.name, now)
|
|
|
|
def mark_started(self, task: ScheduledTask, now: float) -> None:
|
|
self.in_flight.add(task.name)
|
|
self.last_run[task.name] = now
|
|
|
|
def mark_done(self, name: str) -> None:
|
|
self.in_flight.discard(name)
|
|
|
|
|
|
async def start_scheduler(tasks: list[ScheduledTask], paused=None) -> None:
|
|
"""Run scheduled tasks in a loop. Call with asyncio.create_task().
|
|
`paused`, when given, is asked each tick: while it says so nothing new
|
|
starts (a restore is under way, steward/backup/restoring.py)."""
|
|
tracker = _DueTracker()
|
|
|
|
def _spawn(task: ScheduledTask, now: float) -> None:
|
|
tracker.mark_started(task, now)
|
|
asyncio.create_task(_run_task(task, tracker))
|
|
|
|
started = asyncio.get_event_loop().time()
|
|
tracker.begin(tasks, started)
|
|
for task in tasks:
|
|
if task.run_on_startup:
|
|
logger.info(f"Startup task: {task.name}")
|
|
_spawn(task, started)
|
|
|
|
while True:
|
|
now = asyncio.get_event_loop().time()
|
|
if paused is None or not paused():
|
|
for task in tracker.due(tasks, now):
|
|
_spawn(task, now)
|
|
await asyncio.sleep(1)
|
|
|
|
|
|
async def _run_task(task: ScheduledTask, tracker: _DueTracker) -> None:
|
|
try:
|
|
await task.coro_factory()
|
|
except Exception:
|
|
logger.exception(f"Scheduled task {task.name!r} raised an exception")
|
|
finally:
|
|
tracker.mark_done(task.name)
|