diff --git a/app/backend/application.py b/app/backend/application.py index 95dea4e..4ed9570 100644 --- a/app/backend/application.py +++ b/app/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/app/backend/bootstrap/runtime.py b/app/backend/bootstrap/runtime.py index f45d2b3..3a1ea68 100644 --- a/app/backend/bootstrap/runtime.py +++ b/app/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/app/backend/jobs/runner.py b/app/backend/jobs/runner.py index 9d46708..417ebfb 100644 --- a/app/backend/jobs/runner.py +++ b/app/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/app/config/architecture-inventory.json b/app/config/architecture-inventory.json index 027a0e6..383f44e 100644 --- a/app/config/architecture-inventory.json +++ b/app/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/app/tests/test_background_lifecycle.py b/app/tests/test_background_lifecycle.py new file mode 100644 index 0000000..c1fe9ba --- /dev/null +++ b/app/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/app/tests/test_job_runner.py b/app/tests/test_job_runner.py index 5a5c5bf..a2939eb 100644 --- a/app/tests/test_job_runner.py +++ b/app/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/app/tools/run_preservation_runtime.py b/app/tools/run_preservation_runtime.py index 5c90f28..27e5682 100644 --- a/app/tools/run_preservation_runtime.py +++ b/app/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() diff --git a/docs/governance/code-reduction.md b/docs/governance/code-reduction.md index c044075..4c77101 100644 --- a/docs/governance/code-reduction.md +++ b/docs/governance/code-reduction.md @@ -28,6 +28,7 @@ | CR-08 | NDJSON流式传输 | 问师与复盘助手重复维护响应头、事件写入、完成、断线及关闭流程 | HTTP共享层拥有唯一流式连接生命周期 | 已完成 | | CR-09 | 应用服务门面 | 两个仅服务股池iFinD补全的方法仍错位在全局`DashboardService` | 原函数体机械归位到`PoolServiceMixin` | 已完成 | | CR-10 | Repository所有权 | 五行行业阶段覆盖的三项持久化方法仍错位在根级数据库门面 | 原函数体机械归位到问天Repository，兼容数据继续保留 | 已完成 | +| CR-11 | 后台任务生命周期 | 导入应用即启动调度线程，启动早于端口绑定，停止只置位但不等待 | 运行时显式启停，每个Runner只拥有一个可等待的调度线程 | 已完成 | ## CR-01验收口径 @@ -247,6 +248,32 @@ 本批基线为`xiaobai-reduction-09-service-facade-20260801`；检查点为 `xiaobai-reduction-10-repository-ownership-20260801`。 +## CR-11验收口径 + +- 导入`backend.application`或构造`DashboardService`不得启动后台调度；必须先成功绑定HTTP端口， + 再由运行时显式启动。 +- 同一`InProcessJobRunner`重复启动调度器必须返回同一活动线程，停止必须置位并在限定时间内等待退出， + 有序停止后允许重新启动。 +- 运行时关闭顺序固定为：停止调度、等待已提交任务、关闭HTTP服务器；迁移对比工具继续兼容原版入口。 +- 三个任务的注册定义、5秒刷新频率、3秒初始延迟、幂等键、锁、重试、业务函数和结果不得改变。 +- 声明的超时继续是目标与审计字段；Python线程不能安全强杀，本批不伪造硬取消能力。 + +## CR-11结果 + +- 删除`DashboardService`构造阶段的调度副作用，端口占用、模块导入和单元测试不再提前创建后台写线程； + `backend/bootstrap/runtime.py`成为正式启动与停止所有者。 +- `InProcessJobRunner`集中持有调度停止事件和线程引用；重复启动幂等，停止可等待，原任务锁、持久化运行 + 状态、成功幂等、失败记录和后续重试逻辑保持不变。 +- 新增语法树与运行顺序门禁，固定“构造不启动”“绑定后启动”“停止后关服”，并补齐重复启动与重启测试。 +- 候选328项、纯`app/`导出265项和45项Playwright通过；24个JavaScript文件、API/架构注册表、 + Git空白检查和SQLite完整性检查通过。 +- 本批没有可安全删除的重复任务实现；为补齐原先缺失的生命周期，生产代码净增加24行。增加内容仅为 + 调度状态、幂等启停和运行时委托，不新增业务层、任务或兼容包装。 +- 本批不修改页面、CSS、API、数据源、刷新计算、自动选股条件、数据库结构、权限或LLM。 + +本批基线为`xiaobai-reduction-10-repository-ownership-20260801`；检查点为 +`xiaobai-reduction-11-background-jobs-20260801`。 + ## 人工验收记录 - 2026-08-01：用户检查CR-02与CR-03运行结果，确认未发现明显异常。本记录仅表示本轮可见功能与