diff --git a/backend/bootstrap/container.py b/backend/bootstrap/container.py index b2ea3f7..6131c40 100644 --- a/backend/bootstrap/container.py +++ b/backend/bootstrap/container.py @@ -9,6 +9,7 @@ from backend.database.repositories import RepositoryBundle, build_repository_bun from backend.features.alerts import AlertService from backend.features.review import TradeJournalService from backend.features.screener import StrategyTrackingService +from backend.jobs import InProcessJobRunner, JobRegistry, SQLiteJobRunRepository from chart_data_provider import MarketChartClient from database import ReviewDatabase from ifind_client import IfindHttpClient @@ -30,6 +31,7 @@ class ApplicationContainer: mentor_skills: MentorSkillRegistry realtime_aggregator: WebRealtimeAggregator chart_data: MarketChartClient + jobs: InProcessJobRunner def build_application_container( @@ -41,6 +43,7 @@ def build_application_container( ) -> ApplicationContainer: data_gateway = build_data_gateway(credentials, tushare_token_supplier) repositories = build_repository_bundle(database) + jobs = InProcessJobRunner(JobRegistry.load(), SQLiteJobRunRepository(database)) return ApplicationContainer( database=database, repositories=repositories, @@ -53,4 +56,5 @@ def build_application_container( mentor_skills=MentorSkillRegistry(mentor_skills_dir, private_mentor_skills_dir), realtime_aggregator=data_gateway.realtime_observer, chart_data=data_gateway.chart_data, + jobs=jobs, ) diff --git a/backend/database/migrations/__init__.py b/backend/database/migrations/__init__.py index 3b15ae5..19e24bd 100644 --- a/backend/database/migrations/__init__.py +++ b/backend/database/migrations/__init__.py @@ -1,6 +1,7 @@ from .m0001_adopt_legacy import MIGRATION as M0001_ADOPT_LEGACY +from .m0002_job_runs import MIGRATION as M0002_JOB_RUNS from .runner import Migration, MigrationError, MigrationRunner -MIGRATIONS = (M0001_ADOPT_LEGACY,) +MIGRATIONS = (M0001_ADOPT_LEGACY, M0002_JOB_RUNS) __all__ = ["MIGRATIONS", "Migration", "MigrationError", "MigrationRunner"] diff --git a/backend/database/migrations/m0002_job_runs.py b/backend/database/migrations/m0002_job_runs.py new file mode 100644 index 0000000..e0a1f12 --- /dev/null +++ b/backend/database/migrations/m0002_job_runs.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import sqlite3 + +from backend.database.migrations.runner import Migration + + +def create_job_runs(connection: sqlite3.Connection) -> None: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS job_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_id TEXT NOT NULL, + idempotency_key TEXT NOT NULL, + status TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 1, + started_at TEXT NOT NULL, + finished_at TEXT, + elapsed_ms INTEGER NOT NULL DEFAULT 0, + error_code TEXT NOT NULL DEFAULT '', + message TEXT NOT NULL DEFAULT '', + output_version TEXT NOT NULL DEFAULT '', + metadata TEXT NOT NULL DEFAULT '{}', + UNIQUE(job_id, idempotency_key, attempt) + ) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS idx_job_runs_job_started + ON job_runs(job_id, started_at DESC, id DESC) + """ + ) + connection.execute( + """ + CREATE INDEX IF NOT EXISTS idx_job_runs_status + ON job_runs(status, started_at DESC, id DESC) + """ + ) + + +MIGRATION = Migration( + version="0002", + name="create_job_runs", + action=create_job_runs, + signature="job-runs:v1:id,job,key,status,attempt,times,elapsed,error,output,metadata", +) diff --git a/backend/jobs/__init__.py b/backend/jobs/__init__.py new file mode 100644 index 0000000..81dcff8 --- /dev/null +++ b/backend/jobs/__init__.py @@ -0,0 +1,5 @@ +from .registry import JobDefinition, JobRegistry +from .repository import SQLiteJobRunRepository +from .runner import InProcessJobRunner + +__all__ = ["InProcessJobRunner", "JobDefinition", "JobRegistry", "SQLiteJobRunRepository"] diff --git a/backend/jobs/registry.py b/backend/jobs/registry.py new file mode 100644 index 0000000..8a5f662 --- /dev/null +++ b/backend/jobs/registry.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from pathlib import Path + +from app_config import APP_DIR + + +@dataclass(frozen=True) +class JobDefinition: + job_id: str + schedule: str + input_date_policy: str + dependencies: tuple[str, ...] + lock_key: str + timeout_seconds: int + max_attempts: int + output_version: str + + +class JobRegistry: + def __init__(self, definitions: tuple[JobDefinition, ...]) -> None: + self.definitions = definitions + self._by_id = {item.job_id: item for item in definitions} + if len(self._by_id) != len(definitions): + raise ValueError("Background job IDs must be unique") + + @classmethod + def load(cls, path: Path | None = None) -> "JobRegistry": + config_path = path or APP_DIR / "config" / "jobs.config.json" + payload = json.loads(config_path.read_text(encoding="utf-8")) + definitions = tuple( + JobDefinition( + job_id=str(item["id"]), + schedule=str(item["schedule"]), + input_date_policy=str(item["input_date_policy"]), + dependencies=tuple(str(value) for value in item.get("dependencies") or []), + lock_key=str(item["lock_key"]), + timeout_seconds=max(1, int(item["timeout_seconds"])), + max_attempts=max(1, int(item["max_attempts"])), + output_version=str(item["output_version"]), + ) + for item in payload.get("jobs") or [] + ) + if not definitions: + raise ValueError("Background job registry is empty") + return cls(definitions) + + def get(self, job_id: str) -> JobDefinition: + try: + return self._by_id[job_id] + except KeyError as exc: + raise ValueError(f"Background job is not registered: {job_id}") from exc diff --git a/backend/jobs/repository.py b/backend/jobs/repository.py new file mode 100644 index 0000000..88a2c30 --- /dev/null +++ b/backend/jobs/repository.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from database import ReviewDatabase + + +@dataclass(frozen=True) +class SQLiteJobRunRepository: + database: ReviewDatabase + + def completed(self, job_id: str, idempotency_key: str) -> bool: + with self.database.connect() as connection: + row = connection.execute( + """ + SELECT 1 FROM job_runs + WHERE job_id = ? AND idempotency_key = ? AND status = 'success' + LIMIT 1 + """, + (job_id, idempotency_key), + ).fetchone() + return row is not None + + 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") + with self.database.connect() as connection: + row = connection.execute( + """ + SELECT COALESCE(MAX(attempt), 0) + 1 AS attempt FROM job_runs + WHERE job_id = ? AND idempotency_key = ? + """, + (job_id, idempotency_key), + ).fetchone() + cursor = connection.execute( + """ + INSERT INTO job_runs + (job_id, idempotency_key, status, attempt, started_at, + output_version, metadata) + VALUES (?, ?, 'running', ?, ?, ?, ?) + """, + ( + job_id, idempotency_key, int(row["attempt"]), now, + output_version, + json.dumps(metadata or {}, ensure_ascii=False, separators=(",", ":")), + ), + ) + connection.execute( + """ + DELETE FROM job_runs + WHERE id < (SELECT COALESCE(MAX(id), 0) - 20000 FROM job_runs) + AND status != 'running' + """ + ) + return int(cursor.lastrowid) + + def finish( + self, run_id: int, status: str, elapsed_ms: int, + error_code: str = "", message: str = "", + ) -> None: + now = datetime.now().astimezone().isoformat(timespec="seconds") + with self.database.connect() as connection: + connection.execute( + """ + UPDATE job_runs + SET status = ?, finished_at = ?, elapsed_ms = ?, + error_code = ?, message = ? + WHERE id = ? + """, + (status, now, elapsed_ms, error_code, message[:1000], int(run_id)), + ) + + def recent(self, limit: int = 20) -> list[dict[str, Any]]: + with self.database.connect() as connection: + rows = connection.execute( + """ + SELECT id, job_id, idempotency_key, status, attempt, started_at, + finished_at, elapsed_ms, error_code, message, output_version + FROM job_runs ORDER BY id DESC LIMIT ? + """, + (max(1, min(100, int(limit))),), + ).fetchall() + return [dict(row) for row in rows] diff --git a/backend/jobs/runner.py b/backend/jobs/runner.py new file mode 100644 index 0000000..9d46708 --- /dev/null +++ b/backend/jobs/runner.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import threading +import time +from collections.abc import Callable +from typing import Any + +from backend.jobs.registry import JobRegistry +from backend.jobs.repository import SQLiteJobRunRepository + + +JobAction = Callable[[], Any] + + +class InProcessJobRunner: + def __init__(self, registry: JobRegistry, repository: SQLiteJobRunRepository) -> None: + self.registry = registry + self.repository = repository + self._locks: dict[str, threading.Lock] = {} + self._locks_guard = threading.Lock() + + def submit( + self, job_id: str, idempotency_key: str, action: JobAction, + metadata: dict[str, Any] | None = None, + ) -> bool: + definition = self.registry.get(job_id) + if self.repository.completed(job_id, idempotency_key): + return False + lock = self._lock(definition.lock_key) + if not lock.acquire(blocking=False): + return False + thread = threading.Thread( + target=self._execute_locked, + args=(job_id, idempotency_key, action, metadata, lock), + name=f"job-{job_id}-{idempotency_key}"[:80], + daemon=True, + ) + thread.start() + return True + + def run_inline( + self, job_id: str, idempotency_key: str, action: JobAction, + metadata: dict[str, Any] | None = None, + ) -> bool: + definition = self.registry.get(job_id) + if self.repository.completed(job_id, idempotency_key): + return False + 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 + + def start_scheduler( + self, callback: Callable[[], None], stop_event: threading.Event, + interval_seconds: float, initial_delay_seconds: float = 0, + ) -> threading.Thread: + def schedule_loop() -> None: + if stop_event.wait(initial_delay_seconds): + return + while not stop_event.is_set(): + try: + callback() + except Exception: + # Submitted jobs persist their own failures; the scheduler must stay alive. + pass + stop_event.wait(interval_seconds) + + thread = threading.Thread( + target=schedule_loop, + name="background-job-scheduler", + daemon=True, + ) + thread.start() + return thread + + def wait_for_idle(self, timeout_seconds: float = 5) -> bool: + deadline = time.monotonic() + max(0, timeout_seconds) + while time.monotonic() <= deadline: + with self._locks_guard: + busy = any(lock.locked() for lock in self._locks.values()) + if not busy: + return True + time.sleep(0.01) + return False + + def _execute_locked( + self, job_id: str, idempotency_key: str, action: JobAction, + metadata: dict[str, Any] | None, lock: threading.Lock, + ) -> None: + 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 + ) + started = time.perf_counter() + try: + result = action() + if isinstance(result, dict) and result.get("status") == "failed": + 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 + except Exception as exc: + elapsed_ms = round((time.perf_counter() - started) * 1000) + self.repository.finish( + run_id, "failed", elapsed_ms, + type(exc).__name__, str(exc), + ) + if attempt >= definition.max_attempts: + return + finally: + lock.release() + + def _lock(self, lock_key: str) -> threading.Lock: + with self._locks_guard: + return self._locks.setdefault(lock_key, threading.Lock()) diff --git a/config/README.md b/config/README.md index 9296574..cc512e0 100644 --- a/config/README.md +++ b/config/README.md @@ -11,6 +11,8 @@ These registries describe the approved product surface during architecture migra known blocked datasets. - `data-quality.config.json`: freshness, coverage, units, adjustment, point-in-time, and fail-closed rules for every canonical data product. +- `jobs.config.json`: background schedules, dependencies, lock keys, retry policy, timeouts, + and output versions. During Stage 04 these files are contract inputs, not runtime replacements. Backend access in `api_access.py` remains authoritative until the HTTP governance phase switches it atomically. diff --git a/config/jobs.config.json b/config/jobs.config.json new file mode 100644 index 0000000..9d1cbf5 --- /dev/null +++ b/config/jobs.config.json @@ -0,0 +1,35 @@ +{ + "schema_version": 1, + "jobs": [ + { + "id": "market.refresh", + "schedule": "realtime polling or administrator request", + "input_date_policy": "requested trade date", + "dependencies": ["market provider", "database"], + "lock_key": "market-refresh", + "timeout_seconds": 120, + "max_attempts": 1, + "output_version": "dashboard-v1" + }, + { + "id": "screener.automatic", + "schedule": "trading day after 15:10 Asia/Shanghai", + "input_date_policy": "current completed trade date", + "dependencies": ["market.refresh", "factor data", "database"], + "lock_key": "automatic-screener", + "timeout_seconds": 900, + "max_attempts": 1, + "output_version": "screener-library-v8" + }, + { + "id": "market.ifind-event-enrichment", + "schedule": "on demand after market close", + "input_date_policy": "completed trade date", + "dependencies": ["ifind", "database"], + "lock_key": "ifind-event-enrichment", + "timeout_seconds": 180, + "max_attempts": 1, + "output_version": "ifind-event-v1" + } + ] +} diff --git a/docs/governance/architecture-inventory.json b/docs/governance/architecture-inventory.json index baef7f9..f73bee7 100644 --- a/docs/governance/architecture-inventory.json +++ b/docs/governance/architecture-inventory.json @@ -189,8 +189,7 @@ "heaven_readings" ], "background_job_methods": [ - "_background_refresh_loop", - "_run_background_sync", + "_background_refresh_tick", "_schedule_automatic_screeners", "_schedule_ifind_event_enrichment", "run_automatic_screeners" @@ -265,8 +264,8 @@ }, { "path": "server.py", - "bytes": 268422, - "lines": 5960 + "bytes": 268447, + "lines": 5956 }, { "path": "static/redesign-v2.css", diff --git a/docs/governance/stage-12-background-jobs.md b/docs/governance/stage-12-background-jobs.md new file mode 100644 index 0000000..91f7801 --- /dev/null +++ b/docs/governance/stage-12-background-jobs.md @@ -0,0 +1,30 @@ +# Stage 12: Governed Background Jobs + +Date: 2026-07-29 + +## Result + +- Registered market refresh, automatic screening, and iFinD event enrichment as versioned + jobs with schedules, date policy, dependencies, locks, timeouts, retry limits, and output + versions. +- Added the `job_runs` migration and persistent run ledger. +- Centralized worker and scheduler thread creation in `InProcessJobRunner`. +- Added process-level lock keys and persistent idempotency keys. +- Persisted successful, failed, and retried attempts with elapsed time and normalized error + type. +- Exposed recent job runs in administrator system status. +- Limited retained completed history while preserving running jobs. +- Kept existing feature services and the detailed market `sync_runs` audit unchanged. + +## Compatibility + +The runner remains in-process, matching the current single-container deployment. Jobs call the +same application methods as HTTP requests and do not depend on request-local account state. +The persistent contract permits a later worker process without changing job identities or +business calculations. + +## Timeout Boundary + +Timeouts are declared and elapsed time is recorded. Python threads cannot be terminated safely, +so hard cancellation remains cooperative until jobs move to a durable worker process. Locks and +idempotency prevent concurrent duplicate execution in the current single-process deployment. diff --git a/server.py b/server.py index f854460..0838c84 100644 --- a/server.py +++ b/server.py @@ -192,14 +192,15 @@ class DashboardService: self.mentor_skills = self.container.mentor_skills self.realtime_aggregator = self.container.realtime_aggregator self.chart_data = self.container.chart_data + self.jobs = self.container.jobs self.screener.ensure_builtin_strategies() self._background_stop = threading.Event() - self._background_thread = threading.Thread( - target=self._background_refresh_loop, - name="market-background-refresh", - daemon=True, + self._background_thread = self.jobs.start_scheduler( + self._background_refresh_tick, + self._background_stop, + interval_seconds=5, + initial_delay_seconds=3, ) - self._background_thread.start() def _tushare_client(self) -> TushareClient: gateway = getattr(self, "data_gateway", None) @@ -583,6 +584,7 @@ class DashboardService: self._system_credentials.get("background_refresh_enabled", True) ), **self.database.status(), + "jobs": self.jobs.repository.recent(12), }, "llm": { "primary_configured": self._profile_configured(platform["primary"]), @@ -772,36 +774,32 @@ class DashboardService: raise ValueError("用户不存在。") def request_background_sync(self, trade_date: str) -> bool: - if self.sync_lock.locked(): - return False normalized = normalize_date(trade_date) - threading.Thread( - target=self._run_background_sync, - args=(normalized,), - name=f"market-sync-{normalized}", - daemon=True, - ).start() - return True + key = f"manual:{normalized}:{time.time_ns()}" + return self.jobs.submit( + "market.refresh", + key, + lambda: self.sync_dashboard(normalized), + {"trade_date": normalized, "trigger": "administrator"}, + ) - def _run_background_sync(self, trade_date: str) -> None: - try: - self.sync_dashboard(trade_date) - except Exception: + def _background_refresh_tick(self) -> None: + if not ( + self.configured + and self._system_credentials.get("background_refresh_enabled", True) + ): return - - def _background_refresh_loop(self) -> None: - self._background_stop.wait(3) - while not self._background_stop.is_set(): - try: - if self.configured and self._system_credentials.get("background_refresh_enabled", True): - today = date.today().strftime("%Y%m%d") - snapshot = self.database.get_snapshot(today) or {} - if self._realtime_snapshot_due(today, snapshot): - self._run_background_sync(today) - self._schedule_automatic_screeners(today, snapshot) - except Exception: - pass - self._background_stop.wait(5) + today = date.today().strftime("%Y%m%d") + snapshot = self.database.get_snapshot(today) or {} + if self._realtime_snapshot_due(today, snapshot): + bucket = int(time.time() // 5) + self.jobs.submit( + "market.refresh", + f"realtime:{today}:{bucket}", + lambda: self.sync_dashboard(today), + {"trade_date": today, "trigger": "realtime-poll"}, + ) + self._schedule_automatic_screeners(today, snapshot) def register_account(self, username: str, password: str) -> dict[str, Any]: username = username.strip() @@ -1743,13 +1741,12 @@ class DashboardService: if last_attempt and (now - last_attempt).total_seconds() < 600: return False self._auto_screener_last_attempt[normalized_date] = now - threading.Thread( - target=self.run_automatic_screeners, - args=(normalized_date,), - name=f"automatic-screeners-{normalized_date}", - daemon=True, - ).start() - return True + return self.jobs.submit( + "screener.automatic", + f"{normalized_date}:v{SCREENER_LIBRARY_VERSION}", + lambda: self.run_automatic_screeners(normalized_date), + {"trade_date": normalized_date, "trigger": "post-close"}, + ) def run_automatic_screeners(self, trade_date: str) -> dict[str, Any]: normalized_date = normalize_date(trade_date) @@ -4586,13 +4583,12 @@ class DashboardService: now = datetime.now().astimezone() if trade_date == now.strftime("%Y%m%d") and now.time().replace(tzinfo=None) < dt_time(15, 0): return - thread = threading.Thread( - target=self._refresh_ifind_event_enrichment, - args=(trade_date,), - name=f"ifind-event-{trade_date}", - daemon=True, + self.jobs.submit( + "market.ifind-event-enrichment", + f"{trade_date}:v1", + lambda: self._refresh_ifind_event_enrichment(trade_date), + {"trade_date": trade_date, "trigger": "dashboard-enrichment"}, ) - thread.start() def _refresh_ifind_event_enrichment(self, trade_date: str) -> None: if not self._ifind_event_lock.acquire(blocking=False): diff --git a/tests/test_database_migrations.py b/tests/test_database_migrations.py index 9201e4f..3b85ab8 100644 --- a/tests/test_database_migrations.py +++ b/tests/test_database_migrations.py @@ -20,14 +20,17 @@ class DatabaseMigrationTests(unittest.TestCase): ).fetchall() self.assertEqual( [(row["version"], row["name"]) for row in rows], - [("0001", "adopt_legacy_schema")], + [ + ("0001", "adopt_legacy_schema"), + ("0002", "create_job_runs"), + ], ) ReviewDatabase(path) with database.connect() as connection: count = connection.execute( "SELECT COUNT(*) AS count FROM schema_migrations" ).fetchone()["count"] - self.assertEqual(count, 1) + self.assertEqual(count, 2) def test_connection_factory_enables_required_pragmas(self) -> None: with tempfile.TemporaryDirectory() as root: diff --git a/tests/test_governance_registries.py b/tests/test_governance_registries.py index e6e332b..f01c9cb 100644 --- a/tests/test_governance_registries.py +++ b/tests/test_governance_registries.py @@ -23,6 +23,7 @@ class GovernanceRegistryTests(unittest.TestCase): self.api = load("api.config.json") self.data = load("data-fields.config.json") self.quality = load("data-quality.config.json") + self.jobs = load("jobs.config.json") def test_features_are_unique_and_use_declared_roles(self) -> None: roles = set(self.features["roles"]) @@ -92,6 +93,18 @@ class GovernanceRegistryTests(unittest.TestCase): all(str(rule.get("unit_profile") or "none") in profiles for rule in rules.values()) ) + def test_background_jobs_have_complete_unique_runtime_contracts(self) -> None: + jobs = self.jobs["jobs"] + ids = [item["id"] for item in jobs] + self.assertEqual(len(ids), len(set(ids))) + for item in jobs: + self.assertTrue(item["schedule"]) + self.assertTrue(item["input_date_policy"]) + self.assertTrue(item["lock_key"]) + self.assertGreater(int(item["timeout_seconds"]), 0) + self.assertGreater(int(item["max_attempts"]), 0) + self.assertTrue(item["output_version"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_job_runner.py b/tests/test_job_runner.py new file mode 100644 index 0000000..5a5c5bf --- /dev/null +++ b/tests/test_job_runner.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import tempfile +import threading +import unittest +from pathlib import Path + +from backend.jobs import InProcessJobRunner, JobRegistry, SQLiteJobRunRepository +from database import ReviewDatabase + + +class JobRunnerTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.database = ReviewDatabase(Path(self.temporary.name) / "review.db") + self.repository = SQLiteJobRunRepository(self.database) + self.runner = InProcessJobRunner(JobRegistry.load(), self.repository) + + def test_successful_idempotent_job_runs_once(self) -> None: + calls = [] + self.assertTrue( + self.runner.run_inline( + "screener.automatic", "20260729:v8", lambda: calls.append("run") + ) + ) + self.assertFalse( + self.runner.run_inline( + "screener.automatic", "20260729:v8", lambda: calls.append("again") + ) + ) + self.assertEqual(calls, ["run"]) + self.assertEqual(self.repository.recent(1)[0]["status"], "success") + + def test_failure_is_persisted_and_can_be_retried_later(self) -> None: + def fail() -> None: + raise RuntimeError("provider unavailable") + + self.assertTrue(self.runner.run_inline("market.refresh", "failed-key", fail)) + failed = self.repository.recent(1)[0] + self.assertEqual(failed["status"], "failed") + self.assertEqual(failed["error_code"], "RuntimeError") + self.assertTrue( + self.runner.run_inline("market.refresh", "failed-key", lambda: None) + ) + self.assertEqual(self.repository.recent(1)[0]["attempt"], 2) + + def test_lock_key_rejects_concurrent_submission(self) -> None: + entered = threading.Event() + release = threading.Event() + + def wait() -> None: + entered.set() + release.wait(2) + + self.assertTrue(self.runner.submit("market.refresh", "first", wait)) + self.assertTrue(entered.wait(1)) + self.assertFalse(self.runner.submit("market.refresh", "second", lambda: None)) + release.set() + self.assertTrue(self.runner.wait_for_idle()) + + def test_failed_status_payload_is_recorded_as_failure(self) -> None: + self.assertTrue( + self.runner.run_inline( + "screener.automatic", "reported-failure", + lambda: {"status": "failed", "error": "missing factors"}, + ) + ) + self.assertEqual(self.repository.recent(1)[0]["status"], "failed") + + +if __name__ == "__main__": + unittest.main()