diff --git a/app/backend/application.py b/app/backend/application.py index d9b6121..235743a 100644 --- a/app/backend/application.py +++ b/app/backend/application.py @@ -43,6 +43,7 @@ from backend.features.screener.service import ( from backend.features.sentiment import SentimentServiceMixin from backend.features.sentiment.routes import SentimentRoutesMixin from backend.features.system import SystemHttpMixin +from backend.features.system.health import HealthServiceMixin from backend.features.system.routes import SystemRoutesMixin from backend.features.system.service import SystemServiceMixin from backend.features.themes import ThemeServiceMixin @@ -77,6 +78,7 @@ LEGACY_SECRET_KEYS = { class DashboardService( + HealthServiceMixin, SystemServiceMixin, AccountApplicationMixin, JobServiceMixin, diff --git a/app/backend/features/system/health.py b/app/backend/features/system/health.py new file mode 100644 index 0000000..bf576ff --- /dev/null +++ b/app/backend/features/system/health.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + + +class HealthServiceMixin: + def health_status(self) -> dict[str, Any]: + """Return a cheap public readiness summary without provider secrets.""" + database_ready = False + database_status: dict[str, Any] = {} + try: + with self.database.connect() as connection: + database_ready = connection.execute("SELECT 1").fetchone() is not None + database_status = self.database.status() + except Exception: + database_ready = False + + try: + recent_jobs = self.jobs.repository.recent(12) + except Exception: + recent_jobs = [] + + last_sync = database_status.get("last_sync") or {} + snapshot_dates = int(database_status.get("snapshot_dates") or 0) + data_configured = bool(self.configured) + last_sync_success = last_sync.get("status") == "success" + if data_configured and snapshot_dates and last_sync_success: + data_status = "ready" + elif snapshot_dates: + data_status = "degraded" + else: + data_status = "unavailable" + + latest_job = recent_jobs[0] if recent_jobs else {} + failed_jobs = sum(1 for job in recent_jobs if job.get("status") == "failed") + if not database_ready: + jobs_status = "unavailable" + elif latest_job.get("status") == "running": + jobs_status = "running" + elif latest_job.get("status") == "failed": + jobs_status = "degraded" + else: + jobs_status = "ready" + + platform = self._platform_llm_profile() + primary_ready = self._profile_configured(platform["primary"]) + fallback_ready = self._profile_configured(platform["fallback"]) + models_status = "ready" if primary_ready else "unavailable" + + component_states = ( + "ready", + "ready" if database_ready else "unavailable", + data_status, + jobs_status, + models_status, + ) + return { + "ok": database_ready, + "status": ( + "ready" + if all(state in {"ready", "running"} for state in component_states) + else "degraded" + ), + "account_required": True, + "time": datetime.now().astimezone().isoformat(timespec="seconds"), + "components": { + "process": {"status": "ready"}, + "database": { + "status": "ready" if database_ready else "unavailable", + "snapshot_dates": snapshot_dates, + "last_snapshot_at": database_status.get("updated_at") or "", + }, + "data_sources": { + "status": data_status, + "configured": data_configured, + "snapshot_available": snapshot_dates > 0, + "latest_trade_date": str(last_sync.get("trade_date") or ""), + "last_success_at": ( + str(last_sync.get("finished_at") or "") if last_sync_success else "" + ), + }, + "jobs": { + "status": jobs_status, + "recent_failures": failed_jobs, + "latest_finished_at": str(latest_job.get("finished_at") or ""), + }, + "models": { + "status": models_status, + "primary": "ready" if primary_ready else "unavailable", + "fallback": "ready" if fallback_ready else "disabled", + }, + }, + } diff --git a/app/backend/features/system/routes.py b/app/backend/features/system/routes.py index 5956cc9..b3b475e 100644 --- a/app/backend/features/system/routes.py +++ b/app/backend/features/system/routes.py @@ -1,20 +1,12 @@ from __future__ import annotations -from datetime import datetime from http import HTTPStatus class SystemRoutesMixin: def _handle_system_public_get(self, parsed) -> bool: if parsed.path == "/api/health": - self.send_json( - { - "ok": True, - "storage": "sqlite", - "account_required": True, - "time": datetime.now().astimezone().isoformat(timespec="seconds"), - } - ) + self.send_json(self.application_service.health_status()) return True return False diff --git a/app/backend/jobs/repository.py b/app/backend/jobs/repository.py index 88a2c30..4065303 100644 --- a/app/backend/jobs/repository.py +++ b/app/backend/jobs/repository.py @@ -2,7 +2,7 @@ from __future__ import annotations import json from dataclasses import dataclass -from datetime import datetime +from datetime import datetime, timedelta from typing import Any from database import ReviewDatabase @@ -27,9 +27,37 @@ class SQLiteJobRunRepository: def start( self, job_id: str, idempotency_key: str, output_version: str, metadata: dict[str, Any] | None = None, - ) -> int: - now = datetime.now().astimezone().isoformat(timespec="seconds") + stale_after_seconds: int = 0, + ) -> int | None: + current = datetime.now().astimezone() + now = current.isoformat(timespec="seconds") + stale_before = ( + current - timedelta(seconds=max(1, int(stale_after_seconds or 1))) + ).isoformat(timespec="seconds") with self.database.connect() as connection: + # The process-local runner lock cannot protect two server processes. + connection.execute("BEGIN IMMEDIATE") + connection.execute( + """ + UPDATE job_runs + SET status = 'failed', finished_at = ?, error_code = 'StaleRun', + message = 'Previous run exceeded its execution window.' + WHERE job_id = ? AND idempotency_key = ? AND status = 'running' + AND started_at < ? + """, + (now, job_id, idempotency_key, stale_before), + ) + claimed = connection.execute( + """ + SELECT 1 FROM job_runs + WHERE job_id = ? AND idempotency_key = ? + AND status IN ('running', 'success') + LIMIT 1 + """, + (job_id, idempotency_key), + ).fetchone() + if claimed: + return None row = connection.execute( """ SELECT COALESCE(MAX(attempt), 0) + 1 AS attempt FROM job_runs diff --git a/app/backend/jobs/runner.py b/app/backend/jobs/runner.py index 417ebfb..c579e2d 100644 --- a/app/backend/jobs/runner.py +++ b/app/backend/jobs/runner.py @@ -51,8 +51,7 @@ class InProcessJobRunner: lock = self._lock(definition.lock_key) if not lock.acquire(blocking=False): return False - self._execute_locked(job_id, idempotency_key, action, metadata, lock) - return True + return self._execute_locked(job_id, idempotency_key, action, metadata, lock) def start_scheduler( self, callback: Callable[[], None], interval_seconds: float, @@ -105,13 +104,19 @@ class InProcessJobRunner: def _execute_locked( self, job_id: str, idempotency_key: str, action: JobAction, metadata: dict[str, Any] | None, lock: threading.Lock, - ) -> None: + ) -> bool: definition = self.registry.get(job_id) try: for attempt in range(1, definition.max_attempts + 1): run_id = self.repository.start( - job_id, idempotency_key, definition.output_version, metadata + job_id, + idempotency_key, + definition.output_version, + metadata, + definition.timeout_seconds, ) + if run_id is None: + return False started = time.perf_counter() try: result = action() @@ -119,7 +124,7 @@ class InProcessJobRunner: raise RuntimeError(str(result.get("error") or "Job reported failure")) elapsed_ms = round((time.perf_counter() - started) * 1000) self.repository.finish(run_id, "success", elapsed_ms) - return + return True except Exception as exc: elapsed_ms = round((time.perf_counter() - started) * 1000) self.repository.finish( @@ -127,7 +132,8 @@ class InProcessJobRunner: type(exc).__name__, str(exc), ) if attempt >= definition.max_attempts: - return + return True + return True finally: lock.release() diff --git a/app/config/architecture-inventory.json b/app/config/architecture-inventory.json index 134e937..c1ee4f3 100644 --- a/app/config/architecture-inventory.json +++ b/app/config/architecture-inventory.json @@ -609,16 +609,16 @@ "bytes": 6983, "lines": 146 }, + { + "path": "backend/application.py", + "bytes": 6837, + "lines": 180 + }, { "path": "backend/data/providers/tushare_daily.py", "bytes": 6837, "lines": 160 }, - { - "path": "backend/application.py", - "bytes": 6751, - "lines": 178 - }, { "path": "backend/features/market/insights_popularity.py", "bytes": 6739, @@ -829,11 +829,6 @@ "bytes": 1455, "lines": 48 }, - { - "path": "backend/features/system/routes.py", - "bytes": 1423, - "lines": 40 - }, { "path": "backend/features/themes/routes.py", "bytes": 1337, @@ -844,6 +839,11 @@ "bytes": 1195, "lines": 30 }, + { + "path": "backend/features/system/routes.py", + "bytes": 1178, + "lines": 32 + }, { "path": "frontend/pages/ladder/page.html", "bytes": 1143, diff --git a/app/tests/test_job_runner.py b/app/tests/test_job_runner.py index a2939eb..843af2a 100644 --- a/app/tests/test_job_runner.py +++ b/app/tests/test_job_runner.py @@ -59,6 +59,28 @@ class JobRunnerTests(unittest.TestCase): release.set() self.assertTrue(self.runner.wait_for_idle()) + def test_database_claim_rejects_the_same_job_from_another_runner(self) -> None: + other_runner = InProcessJobRunner(JobRegistry.load(), self.repository) + entered = threading.Event() + release = threading.Event() + calls = [] + + def wait() -> None: + calls.append("first") + entered.set() + release.wait(2) + + self.assertTrue(self.runner.submit("screener.automatic", "shared-key", wait)) + self.assertTrue(entered.wait(1)) + self.assertFalse( + other_runner.run_inline( + "screener.automatic", "shared-key", lambda: calls.append("second") + ) + ) + release.set() + self.assertTrue(self.runner.wait_for_idle()) + self.assertEqual(calls, ["first"]) + def test_failed_status_payload_is_recorded_as_failure(self) -> None: self.assertTrue( self.runner.run_inline( diff --git a/app/tests/test_system_health.py b/app/tests/test_system_health.py new file mode 100644 index 0000000..252e508 --- /dev/null +++ b/app/tests/test_system_health.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import json +import unittest +from types import SimpleNamespace + +from backend.features.system.health import HealthServiceMixin +from backend.features.system.routes import SystemRoutesMixin + + +class _Connection: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def execute(self, _query: str): + return self + + def fetchone(self): + return (1,) + + +class _Database: + def connect(self): + return _Connection() + + def status(self): + return { + "database": "secret-review.db", + "snapshot_dates": 31, + "updated_at": "2026-08-07T11:34:55+08:00", + "last_sync": { + "id": 991, + "trade_date": "20260807", + "source": "tushare", + "status": "success", + "finished_at": "2026-08-07T11:34:55+08:00", + "message": "https://provider.invalid?token=secret", + }, + } + + +class _Service(HealthServiceMixin): + def __init__(self): + self.database = _Database() + self.jobs = SimpleNamespace( + repository=SimpleNamespace( + recent=lambda _limit: [ + { + "id": 77, + "status": "success", + "finished_at": "2026-08-07T11:34:55+08:00", + "message": "secret upstream response", + } + ] + ) + ) + self._system_credentials = { + "tushare_token": "secret-token", + "primary_model_id": "primary-id", + "fallback_model_id": "fallback-id", + } + + @property + def configured(self): + return bool(self._system_credentials.get("tushare_token")) + + def _platform_llm_profile(self): + return { + "primary": { + "api_key": "secret-primary-key", + "base_url": "https://models.invalid/v1", + "model": "secret-primary-model", + }, + "fallback": { + "api_key": "secret-fallback-key", + "base_url": "https://models.invalid/v1", + "model": "secret-fallback-model", + }, + } + + @staticmethod + def _profile_configured(profile): + return all(profile.get(key) for key in ("api_key", "base_url", "model")) + + +class SystemHealthTests(unittest.TestCase): + def test_public_route_uses_the_service_health_contract(self): + handler = SystemRoutesMixin() + expected = {"ok": True, "components": {"process": {"status": "ready"}}} + handler.application_service = SimpleNamespace(health_status=lambda: expected) + responses = [] + handler.send_json = responses.append + + handled = handler._handle_system_public_get(SimpleNamespace(path="/api/health")) + + self.assertTrue(handled) + self.assertEqual(responses, [expected]) + + def test_health_distinguishes_runtime_components(self): + result = _Service().health_status() + + self.assertTrue(result["ok"]) + self.assertEqual(result["status"], "ready") + self.assertEqual( + set(result["components"]), + {"process", "database", "data_sources", "jobs", "models"}, + ) + self.assertEqual(result["components"]["data_sources"]["latest_trade_date"], "20260807") + self.assertEqual(result["components"]["models"]["primary"], "ready") + self.assertEqual(result["components"]["models"]["fallback"], "ready") + + def test_public_health_never_exposes_engineering_or_secret_fields(self): + payload = json.dumps(_Service().health_status(), ensure_ascii=False).lower() + + for forbidden in ( + "tushare", + "ifind", + "secret", + "token", + "https://", + ".db", + "primary-id", + "fallback-id", + "model", + ): + if forbidden == "model": + self.assertNotIn("secret-primary-model", payload) + else: + self.assertNotIn(forbidden, payload) + + +if __name__ == "__main__": + unittest.main()