Eric WagonerandClaude Fable 5
08b741671d
The findings clustered exactly where prediction said: the unreviewed web layer. The big ones: decisions made while an extract/resolve job runs are now refused with a 409 (the job's end-of-run rewrite from a start-of-run snapshot would silently revert them); a cross-origin guard blocks preflight-free mutations from hostile webpages (bodyless run triggers, cross-site photo form posts); the JobRunner sets terminal status in a finally catching BaseException (a greenlet death could wedge every future run behind 409s) and writes tracebacks into the visible job log; and a boot token lets clients accept the revision reset after a server restart instead of freezing forever. Even the thrice-audited core yielded one HIGH: an unvetoed bare typo-read sibling of a confident row duplicated its add when the game wasn't in the collection — diff now treats it as satisfied. Second-copy adds carry a flag through to_add.csv and the upload log so verify honestly reports them unverifiable instead of OK. Also: merged_into chains collapse transitively; diff/enrich treat a BGG queue timeout like a missing token; enrich prunes orphaned games.json keys; the wizard shell-quotes .env values and creates the file 0600 from the first byte; fsio stats the tmp inode before replace and uses unique tmp names; an explicit missing --config errors; storage state is owner-only; extract re-extracts corrupt caches, aborts on 3 identical failures, and exits nonzero when nothing succeeded; torn JSON artifacts degrade with in-browser warnings instead of 500ing every page; photo uploads are atomic with cache-invalidation ordered first; the pipeline page computes `running` before the buttons that depend on it; the photo dropzone alerts on network failure; and lost-contact banners clear on recovery everywhere. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""One-at-a-time background execution of pipeline stages for the web UI.
|
|
|
|
The stages share the data/ artifacts, so running two concurrently is
|
|
unsupported everywhere in the pipeline — the runner enforces it with a
|
|
single slot. Stage output (typer.echo goes to stdout) is captured by
|
|
swapping sys.stdout for the job's duration; that swap is process-wide,
|
|
which is safe here only because the runner holds the single slot and the
|
|
web server itself never writes to stdout (uvicorn logs on stderr).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from collections.abc import Callable
|
|
from contextlib import redirect_stdout
|
|
|
|
import typer
|
|
|
|
MAX_LOG_LINES = 1000 # buffer cap; snapshots serve the last 200
|
|
|
|
|
|
class _LineBuffer(io.TextIOBase):
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._lines: list[str] = []
|
|
self._partial = ""
|
|
|
|
def write(self, text: str) -> int:
|
|
with self._lock:
|
|
self._partial += text
|
|
*complete, self._partial = self._partial.split("\n")
|
|
self._lines.extend(complete)
|
|
if len(self._lines) > MAX_LOG_LINES: # bound memory on long runs
|
|
del self._lines[: len(self._lines) - MAX_LOG_LINES]
|
|
return len(text)
|
|
|
|
def lines(self) -> list[str]:
|
|
with self._lock:
|
|
return self._lines + ([self._partial] if self._partial else [])
|
|
|
|
|
|
class JobRunner:
|
|
"""At most one running job; finished state persists until the next start."""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._thread: threading.Thread | None = None
|
|
self._stage = ""
|
|
self._status = "idle" # idle | running | done | failed
|
|
self._buffer = _LineBuffer()
|
|
self._error = ""
|
|
self._started = 0.0
|
|
self._finished = 0.0
|
|
|
|
def start(self, stage: str, fn: Callable[[], object]) -> bool:
|
|
"""Begin a job; False when one is already running."""
|
|
with self._lock:
|
|
if self._status == "running":
|
|
return False
|
|
self._stage = stage
|
|
self._status = "running"
|
|
self._buffer = _LineBuffer()
|
|
self._error = ""
|
|
self._started = time.time()
|
|
self._finished = 0.0
|
|
self._thread = threading.Thread(target=self._run, args=(fn,), daemon=True)
|
|
self._thread.start()
|
|
return True
|
|
|
|
def _run(self, fn: Callable[[], object]) -> None:
|
|
status, error = "done", ""
|
|
try:
|
|
try:
|
|
with redirect_stdout(self._buffer):
|
|
fn()
|
|
except typer.Exit as exc:
|
|
if exc.exit_code:
|
|
status, error = "failed", f"exited with code {exc.exit_code}"
|
|
except SystemExit as exc:
|
|
if exc.code:
|
|
status, error = "failed", f"exited with code {exc.code}"
|
|
except BaseException as exc: # incl. GreenletExit et al: the
|
|
# runner must NEVER stay "running" — that wedges every
|
|
# future stage behind a 409 until a server restart
|
|
status, error = "failed", f"{type(exc).__name__}: {exc}"
|
|
# the short error names the exception; the log carries the
|
|
# traceback, or a failure is undiagnosable from the UI
|
|
self._buffer.write("\n" + traceback.format_exc())
|
|
finally:
|
|
with self._lock:
|
|
self._status = status
|
|
self._error = error
|
|
self._finished = time.time()
|
|
|
|
def snapshot(self) -> dict:
|
|
with self._lock:
|
|
return {
|
|
"stage": self._stage,
|
|
"status": self._status,
|
|
"log": self._buffer.lines()[-200:],
|
|
"error": self._error,
|
|
"started": self._started,
|
|
"finished": self._finished,
|
|
}
|
|
|
|
def wait(self, timeout: float = 10.0) -> None:
|
|
"""Test hook: block until the current job's thread finishes."""
|
|
thread = self._thread
|
|
if thread is not None:
|
|
thread.join(timeout)
|