From 111b458ddf4a157ee711e5d410d113db06ec14e4 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 26 Jul 2026 22:59:15 +0000 Subject: [PATCH] fix: reserve and persist background jobs --- dashboard/jobs.py | 111 ++++++++++++++++++++++++ dashboard/server.py | 140 +++++++++++++++++++++---------- tests/test_job_registry.py | 53 ++++++++++++ tests/test_server_reliability.py | 31 +++++++ 4 files changed, 292 insertions(+), 43 deletions(-) create mode 100644 dashboard/jobs.py create mode 100644 tests/test_job_registry.py diff --git a/dashboard/jobs.py b/dashboard/jobs.py new file mode 100644 index 0000000..d7a4893 --- /dev/null +++ b/dashboard/jobs.py @@ -0,0 +1,111 @@ +"""Persistent, thread-safe background job state.""" + +from __future__ import annotations + +import threading +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable + +from dashboard.persistence import atomic_write_json, load_json + +_ACTIVE = {"queued", "running"} + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +class JobRegistry: + """Reserve jobs before spawning and persist their lifecycle.""" + + def __init__(self, path: str | Path, *, history_limit: int = 100): + self.path = Path(path) + self.history_limit = history_limit + self._lock = threading.RLock() + loaded = load_json(self.path, {"jobs": []}) or {"jobs": []} + self._jobs = { + job["id"]: dict(job) + for job in loaded.get("jobs", []) + if isinstance(job, dict) and job.get("id") + } + changed = False + for job in self._jobs.values(): + if job.get("status") in _ACTIVE: + job.update( + status="interrupted", + finished_at=_now(), + error="process restarted before job completed", + ) + changed = True + if changed: + self._save_locked() + + def _save_locked(self) -> None: + jobs = sorted(self._jobs.values(), key=lambda job: job.get("created_at", "")) + if len(jobs) > self.history_limit: + keep = jobs[-self.history_limit :] + self._jobs = {job["id"]: job for job in keep} + jobs = keep + atomic_write_json(self.path, {"jobs": jobs}) + + def reserve(self, kind: str, *, details: dict[str, Any] | None = None) -> dict[str, Any] | None: + with self._lock: + if any( + job.get("kind") == kind and job.get("status") in _ACTIVE + for job in self._jobs.values() + ): + return None + job = { + "id": uuid.uuid4().hex, + "kind": kind, + "status": "queued", + "created_at": _now(), + "started_at": None, + "finished_at": None, + "progress": {}, + "details": details or {}, + "result": None, + "error": None, + } + self._jobs[job["id"]] = job + self._save_locked() + return dict(job) + + def get(self, job_id: str) -> dict[str, Any] | None: + with self._lock: + job = self._jobs.get(job_id) + return dict(job) if job else None + + def active(self, kind: str) -> dict[str, Any] | None: + with self._lock: + for job in self._jobs.values(): + if job.get("kind") == kind and job.get("status") in _ACTIVE: + return dict(job) + return None + + def update_progress(self, job_id: str, progress: dict[str, Any]) -> None: + with self._lock: + job = self._jobs[job_id] + job["progress"] = dict(progress) + self._save_locked() + + def run(self, job_id: str, operation: Callable[[], Any]) -> Any: + with self._lock: + job = self._jobs[job_id] + if job["status"] != "queued": + raise RuntimeError(f"job {job_id} is not queued") + job.update(status="running", started_at=_now()) + self._save_locked() + try: + result = operation() + except Exception as exc: + with self._lock: + job.update(status="error", error=str(exc), finished_at=_now()) + self._save_locked() + raise + with self._lock: + job.update(status="complete", result=result, finished_at=_now()) + self._save_locked() + return result diff --git a/dashboard/server.py b/dashboard/server.py index 84c5ad5..e6109ae 100644 --- a/dashboard/server.py +++ b/dashboard/server.py @@ -37,9 +37,11 @@ from dashboard.persistence import ( merge_observation, onchain_refresh_due, ) +from dashboard.jobs import JobRegistry _shutdown_event = threading.Event() _background_threads = [] +_threads_lock = threading.Lock() @asynccontextmanager @@ -47,15 +49,19 @@ async def lifespan(_app): """Own background worker startup and graceful shutdown.""" _shutdown_event.clear() scraper_thread = threading.Thread(target=scraper_loop, name="scraper-scheduler") - _background_threads.append(scraper_thread) + with _threads_lock: + _background_threads.append(scraper_thread) scraper_thread.start() try: yield finally: _shutdown_event.set() - for thread in list(_background_threads): + with _threads_lock: + threads = list(_background_threads) + for thread in threads: thread.join(timeout=30) - _background_threads.clear() + with _threads_lock: + _background_threads.clear() app = FastAPI(title="Bitcoin Accumulation Zone Monitor", lifespan=lifespan) @@ -65,8 +71,10 @@ DATA_DIR = os.path.join(BASE_DIR, "data") CACHE_PATH = os.path.join(DATA_DIR, "cache.json") HISTORY_PATH = os.path.join(DATA_DIR, "score_history.jsonl") LLM_SETTINGS_PATH = os.path.join(CONFIG_DIR, "llm_settings.json") +JOBS_PATH = os.path.join(DATA_DIR, "jobs.json") os.makedirs(DATA_DIR, exist_ok=True) +_jobs = JobRegistry(JOBS_PATH) # Background scraper state _scraper_lock = threading.Lock() @@ -74,6 +82,32 @@ _scraper_running = False _last_update = None _last_error = None + +def _job_worker(job_id, operation): + try: + _jobs.run(job_id, operation) + except Exception: + log.error("Background job %s failed:\n%s", job_id, traceback.format_exc()) + finally: + current = threading.current_thread() + with _threads_lock: + if current in _background_threads: + _background_threads.remove(current) + + +def _spawn_job(job, operation): + """Start an already-reserved job in a tracked, non-daemon thread.""" + thread = threading.Thread( + target=_job_worker, + args=(job["id"], operation), + name=f"{job['kind']}-{job['id'][:8]}", + ) + with _threads_lock: + _background_threads.append(thread) + thread.start() + return thread + + # ── Cache management ────────────────────────────────────────────────────── def load_cache(): @@ -268,14 +302,20 @@ def run_scrape(force_full=False): _scraper_running = False +def _run_scheduled_refresh(force_full=False): + job = _jobs.reserve("refresh", details={"full": force_full, "scheduled": True}) + if job is not None: + _jobs.run(job["id"], lambda: run_scrape(force_full=force_full)) + + def scraper_loop(): - """Background loop: quick refresh every 15min. Full scrape only on first boot with no data.""" + """Background loop: refresh quickly every 15 minutes, with on-chain TTL handling.""" cache = load_cache() - has_data = any(cache.get(k, {}).get("value") is not None + has_data = any(cache.get(k, {}).get("value") is not None for k in ["puell_multiple", "mvrv_zscore", "nupl"]) - run_scrape(force_full=not has_data) # Full only if no cached on-chain data + _run_scheduled_refresh(force_full=not has_data) while not _shutdown_event.wait(900): - run_scrape() # Quick refresh only + _run_scheduled_refresh() # ── LLM Settings (preserved from original) ─────────────────────────────── @@ -449,16 +489,31 @@ def api_history(): return load_history()[-90:] # Last 90 entries -@app.post("/api/refresh") +@app.post("/api/refresh", status_code=202) def api_refresh(full: bool = False): - """Trigger a scrape. Quick refresh (default) updates price + F&G only (~2s). - Full refresh (?full=true) also re-scrapes on-chain data via Playwright (~2-3min).""" - if _scraper_running: - return JSONResponse({"error": "Scrape already in progress"}, status_code=409) - t = threading.Thread(target=run_scrape, kwargs={"force_full": full}, daemon=True) - t.start() + """Atomically reserve and start a quick or full metric refresh.""" + job = _jobs.reserve("refresh", details={"full": full, "scheduled": False}) + if job is None: + active = _jobs.active("refresh") + return JSONResponse( + {"error": "Scrape already in progress", "job": active}, status_code=409 + ) + _spawn_job(job, lambda: run_scrape(force_full=full)) mode = "full (on-chain + price + F&G)" if full else "quick (price + F&G only)" - return {"ok": True, "message": f"Scrape started — {mode}"} + return { + "ok": True, + "job_id": job["id"], + "status": job["status"], + "message": f"Scrape started — {mode}", + } + + +@app.get("/api/jobs/{job_id}") +def api_job_status(job_id: str): + job = _jobs.get(job_id) + if job is None: + return JSONResponse({"error": "Job not found"}, status_code=404) + return job # Settings routes (preserved) @@ -1500,9 +1555,6 @@ loadSettings(); # ── Backtest API ─────────────────────────────────────────────────────── -_history_collector_running = False -_history_collector_progress = {} - @app.get("/api/backtest") def api_backtest(mode: str = "classic"): @@ -1528,43 +1580,45 @@ def api_backtest_history(): return JSONResponse({"error": str(e)}, status_code=500) -@app.post("/api/backtest/collect") +@app.post("/api/backtest/collect", status_code=202) def api_backtest_collect(): - """Trigger historical data collection.""" - global _history_collector_running, _history_collector_progress - if _history_collector_running: - return JSONResponse({"error": "Collection already in progress", "progress": _history_collector_progress}, status_code=409) + """Atomically reserve and start historical data collection.""" + initial_progress = {"status": "starting", "current": "", "step": 0, "total": 0} + job = _jobs.reserve("history") + if job is None: + active = _jobs.active("history") + return JSONResponse( + {"error": "Collection already in progress", "job": active}, status_code=409 + ) + _jobs.update_progress(job["id"], initial_progress) def _run_collector(): - global _history_collector_running, _history_collector_progress - _history_collector_running = True - _history_collector_progress = {"status": "starting", "current": "", "step": 0, "total": 0} - try: - from scrapers.history_collector import collect_all_history + from scrapers.history_collector import collect_all_history - def progress_cb(metric, step, total): - _history_collector_progress = {"status": "scraping", "current": metric, "step": step + 1, "total": total} + def progress_cb(metric, step, total): + _jobs.update_progress(job["id"], { + "status": "scraping", "current": metric, + "step": step + 1, "total": total, + }) - collect_all_history(progress_cb=progress_cb) - _history_collector_progress = {"status": "complete"} - except Exception as e: - log.error("History collection error: %s", traceback.format_exc()) - _history_collector_progress = {"status": "error", "error": str(e)} - finally: - _history_collector_running = False + collect_all_history(progress_cb=progress_cb) + _jobs.update_progress(job["id"], {"status": "complete"}) + return {"collected": True} - t = threading.Thread(target=_run_collector, daemon=True) - t.start() - return {"ok": True, "message": "Collection started"} + _spawn_job(job, _run_collector) + return {"ok": True, "job_id": job["id"], "status": job["status"], + "message": "Collection started"} @app.get("/api/backtest/status") def api_backtest_status(): - """Check if historical data exists and collection status.""" + """Check historical data and expose only the active collection job's progress.""" from scrapers.history_collector import history_status status = history_status() - status["collecting"] = _history_collector_running - status["progress"] = _history_collector_progress + active = _jobs.active("history") + status["collecting"] = active is not None + status["job_id"] = active.get("id") if active else None + status["progress"] = active.get("progress", {}) if active else {} return status diff --git a/tests/test_job_registry.py b/tests/test_job_registry.py new file mode 100644 index 0000000..1d86df5 --- /dev/null +++ b/tests/test_job_registry.py @@ -0,0 +1,53 @@ +import threading + +from dashboard.jobs import JobRegistry + + +def test_job_registry_atomically_reserves_only_one_job_per_kind(tmp_path): + registry = JobRegistry(tmp_path / "jobs.json") + barrier = threading.Barrier(10) + results = [] + + def reserve(): + barrier.wait() + results.append(registry.reserve("refresh", details={"full": False})) + + threads = [threading.Thread(target=reserve) for _ in range(10)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + reserved = [job for job in results if job is not None] + assert len(reserved) == 1 + assert reserved[0]["id"] + assert reserved[0]["status"] == "queued" + assert registry.active("refresh")["id"] == reserved[0]["id"] + + +def test_job_registry_tracks_completion_and_result(tmp_path): + registry = JobRegistry(tmp_path / "jobs.json") + job = registry.reserve("history") + + result = registry.run(job["id"], lambda: {"records": 42}) + + assert result == {"records": 42} + saved = registry.get(job["id"]) + assert saved["status"] == "complete" + assert saved["result"] == {"records": 42} + assert saved["started_at"] + assert saved["finished_at"] + assert registry.active("history") is None + + +def test_job_registry_marks_abandoned_active_jobs_interrupted_on_restart(tmp_path): + path = tmp_path / "jobs.json" + first = JobRegistry(path) + job = first.reserve("refresh") + + restarted = JobRegistry(path) + + recovered = restarted.get(job["id"]) + assert recovered["status"] == "interrupted" + assert recovered["finished_at"] + assert restarted.active("refresh") is None diff --git a/tests/test_server_reliability.py b/tests/test_server_reliability.py index 2d910a9..93c4065 100644 --- a/tests/test_server_reliability.py +++ b/tests/test_server_reliability.py @@ -117,3 +117,34 @@ def test_expired_onchain_timestamp_triggers_real_refresh(server, monkeypatch, tm assert saved["puell_multiple"]["source"] == "lookintobitcoin" assert saved["sopr"]["source"] == "checkonchain" assert saved["_onchain_timestamp"] != old.isoformat() + + +def test_refresh_job_is_reserved_before_thread_start(server, monkeypatch, tmp_path): + from dashboard.jobs import JobRegistry + + registry = JobRegistry(tmp_path / "jobs.json") + monkeypatch.setattr(server, "_jobs", registry, raising=False) + monkeypatch.setattr(server, "_scraper_running", False) + + started = server.api_refresh(full=False) + duplicate = server.api_refresh(full=False) + + assert started["job_id"] + assert started["status"] == "queued" + assert registry.get(started["job_id"])["status"] == "queued" + assert duplicate.status_code == 409 + + +def test_history_collection_has_job_id_and_job_scoped_progress(server, monkeypatch, tmp_path): + from dashboard.jobs import JobRegistry + + registry = JobRegistry(tmp_path / "jobs.json") + monkeypatch.setattr(server, "_jobs", registry, raising=False) + + started = server.api_backtest_collect() + job = registry.get(started["job_id"]) + + assert started["status"] == "queued" + assert job["kind"] == "history" + assert job["progress"] == {"status": "starting", "current": "", "step": 0, "total": 0} + assert server.api_job_status(started["job_id"])["id"] == started["job_id"]