diff --git a/backend/application.py b/backend/application.py index 95dea4e..4ed9570 100644 --- a/backend/application.py +++ b/backend/application.py @@ -128,14 +128,19 @@ class DashboardService( profile_supplier=self._resolved_llm_profile, ) self.screener.ensure_builtin_strategies() - self._background_stop = threading.Event() - self._background_thread = self.jobs.start_scheduler( + + def start_background_jobs(self) -> threading.Thread: + return self.jobs.start_scheduler( self._background_refresh_tick, - self._background_stop, interval_seconds=5, initial_delay_seconds=3, ) + def stop_background_jobs(self, timeout_seconds: float = 5) -> bool: + scheduler_stopped = self.jobs.stop_scheduler(timeout_seconds) + workers_stopped = self.jobs.wait_for_idle(timeout_seconds) + return scheduler_stopped and workers_stopped + def _load_system_credentials(self, environment: dict[str, str]) -> dict[str, Any]: encrypted = self.database.get_system_setting("credentials") diff --git a/backend/bootstrap/runtime.py b/backend/bootstrap/runtime.py index f45d2b3..3a1ea68 100644 --- a/backend/bootstrap/runtime.py +++ b/backend/bootstrap/runtime.py @@ -16,12 +16,13 @@ def main(handler_class: type[Any] | None = None, service: Any | None = None) -> parser.add_argument("--port", type=int, default=8765) args = parser.parse_args() server = ThreadingHTTPServer((args.host, args.port), handler_class) - print(f"Xiaobai Review Web is running at http://{args.host}:{args.port}") - print("Press Ctrl+C to stop.") try: + service.start_background_jobs() + print(f"Xiaobai Review Web is running at http://{args.host}:{args.port}") + print("Press Ctrl+C to stop.") server.serve_forever() except KeyboardInterrupt: pass finally: - service._background_stop.set() + service.stop_background_jobs() server.server_close() diff --git a/backend/jobs/runner.py b/backend/jobs/runner.py index 9d46708..417ebfb 100644 --- a/backend/jobs/runner.py +++ b/backend/jobs/runner.py @@ -18,6 +18,9 @@ class InProcessJobRunner: self.repository = repository self._locks: dict[str, threading.Lock] = {} self._locks_guard = threading.Lock() + self._scheduler_guard = threading.Lock() + self._scheduler_stop = threading.Event() + self._scheduler_thread: threading.Thread | None = None def submit( self, job_id: str, idempotency_key: str, action: JobAction, @@ -52,27 +55,42 @@ class InProcessJobRunner: return True def start_scheduler( - self, callback: Callable[[], None], stop_event: threading.Event, - interval_seconds: float, initial_delay_seconds: float = 0, + self, callback: Callable[[], None], 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) + with self._scheduler_guard: + current = self._scheduler_thread + if current is not None and current.is_alive(): + return current + self._scheduler_stop.clear() - thread = threading.Thread( - target=schedule_loop, - name="background-job-scheduler", - daemon=True, - ) - thread.start() - return thread + def schedule_loop() -> None: + if self._scheduler_stop.wait(initial_delay_seconds): + return + while not self._scheduler_stop.is_set(): + try: + callback() + except Exception: + # Submitted jobs persist failures; the scheduler must stay alive. + pass + self._scheduler_stop.wait(interval_seconds) + + thread = threading.Thread( + target=schedule_loop, + name="background-job-scheduler", + daemon=True, + ) + self._scheduler_thread = thread + thread.start() + return thread + + def stop_scheduler(self, timeout_seconds: float = 5) -> bool: + with self._scheduler_guard: + thread = self._scheduler_thread + self._scheduler_stop.set() + if thread is not None and thread is not threading.current_thread(): + thread.join(max(0, timeout_seconds)) + return thread is None or not thread.is_alive() def wait_for_idle(self, timeout_seconds: float = 5) -> bool: deadline = time.monotonic() + max(0, timeout_seconds) diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 027a0e6..383f44e 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -374,8 +374,8 @@ }, { "path": "backend/application.py", - "bytes": 47552, - "lines": 1087 + "bytes": 47769, + "lines": 1092 }, { "path": "frontend/styles/theme.css", diff --git a/tests/test_background_lifecycle.py b/tests/test_background_lifecycle.py new file mode 100644 index 0000000..c1fe9ba --- /dev/null +++ b/tests/test_background_lifecycle.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import ast +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +from backend.bootstrap import runtime + + +class BackgroundLifecycleTests(unittest.TestCase): + def test_service_construction_does_not_start_the_scheduler(self) -> None: + source = (Path(__file__).parents[1] / "backend" / "application.py").read_text( + encoding="utf-8" + ) + tree = ast.parse(source) + service_class = next( + node for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == "DashboardService" + ) + constructor = next( + node for node in service_class.body + if isinstance(node, ast.FunctionDef) and node.name == "__init__" + ) + called_methods = { + node.func.attr for node in ast.walk(constructor) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + } + self.assertNotIn("start_scheduler", called_methods) + + def test_runtime_starts_jobs_after_bind_and_stops_before_close(self) -> None: + events: list[str] = [] + + class Service: + def start_background_jobs(self) -> None: + events.append("start-jobs") + + def stop_background_jobs(self) -> None: + events.append("stop-jobs") + + class Server: + def __init__(self, address: tuple[str, int], handler: object) -> None: + events.append("bind") + + def serve_forever(self) -> None: + events.append("serve") + raise KeyboardInterrupt + + def server_close(self) -> None: + events.append("close") + + with ( + patch.object(runtime, "ThreadingHTTPServer", Server), + patch.object(sys, "argv", ["server.py", "--port", "8797"]), + ): + runtime.main(object, Service()) + + self.assertEqual( + events, ["bind", "start-jobs", "serve", "stop-jobs", "close"] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_job_runner.py b/tests/test_job_runner.py index 5a5c5bf..a2939eb 100644 --- a/tests/test_job_runner.py +++ b/tests/test_job_runner.py @@ -68,6 +68,29 @@ class JobRunnerTests(unittest.TestCase): ) self.assertEqual(self.repository.recent(1)[0]["status"], "failed") + def test_scheduler_start_is_idempotent_and_stop_waits_for_exit(self) -> None: + first = self.runner.start_scheduler( + lambda: None, interval_seconds=60, initial_delay_seconds=60 + ) + second = self.runner.start_scheduler( + lambda: None, interval_seconds=60, initial_delay_seconds=60 + ) + self.assertIs(first, second) + self.assertTrue(first.is_alive()) + self.assertTrue(self.runner.stop_scheduler()) + self.assertFalse(first.is_alive()) + + def test_scheduler_can_restart_after_an_orderly_stop(self) -> None: + first = self.runner.start_scheduler( + lambda: None, interval_seconds=60, initial_delay_seconds=60 + ) + self.assertTrue(self.runner.stop_scheduler()) + second = self.runner.start_scheduler( + lambda: None, interval_seconds=60, initial_delay_seconds=60 + ) + self.assertIsNot(first, second) + self.assertTrue(self.runner.stop_scheduler()) + if __name__ == "__main__": unittest.main() diff --git a/tools/run_preservation_runtime.py b/tools/run_preservation_runtime.py index 5c90f28..27e5682 100644 --- a/tools/run_preservation_runtime.py +++ b/tools/run_preservation_runtime.py @@ -32,17 +32,22 @@ def main() -> None: from server import RequestHandler, SERVICE server = ThreadingHTTPServer(("127.0.0.1", args.port), RequestHandler) - print( - f"Preservation runtime is running at http://127.0.0.1:{args.port} " - f"with database {SERVICE.database.path}", - flush=True, - ) try: + if hasattr(SERVICE, "start_background_jobs"): + SERVICE.start_background_jobs() + print( + f"Preservation runtime is running at http://127.0.0.1:{args.port} " + f"with database {SERVICE.database.path}", + flush=True, + ) server.serve_forever() except KeyboardInterrupt: pass finally: - SERVICE._background_stop.set() + if hasattr(SERVICE, "stop_background_jobs"): + SERVICE.stop_background_jobs() + else: + SERVICE._background_stop.set() server.server_close()