From cd97b1300a27a5da689567be687d790b8c364c10 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 23 Aug 2026 15:41:07 +0800 Subject: [PATCH] fix(scheduler): close async task lifecycle --- app/scheduler.py | 107 ++++++++++++++++-- app/startup/scheduler_initializer.py | 12 +- .../adr/0007-background-action-reliability.md | 2 + .../backend-architecture-next-stage.md | 3 + tests/test_scheduler_cache_expiry.py | 29 ++++- tests/test_scheduler_progress.py | 34 ++++++ 6 files changed, 175 insertions(+), 12 deletions(-) diff --git a/app/scheduler.py b/app/scheduler.py index 5882034ed..8fd963e6d 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1,4 +1,6 @@ import asyncio +from concurrent.futures import CancelledError as ConcurrentCancelledError +from concurrent.futures import Future as ConcurrentFuture import gc import hashlib import inspect @@ -129,6 +131,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): self._auth_count = 0 # 用户认证失败消息发送 self._auth_message = False + # 记录由 Scheduler 提交到事件循环的协程，避免 stop() 后继续悬挂。 + self._async_tasks: set[asyncio.Task[Any] | ConcurrentFuture[Any]] = set() + self._accepting_async_tasks = True def on_config_changed(self) -> None: """ @@ -250,6 +255,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): config = get_scheduler_runtime_config() # 停止定时服务 self.stop() + self._accepting_async_tasks = True # 调试模式不启动定时服务 if config.dev: @@ -558,6 +564,8 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ 准备定时任务 """ + if not getattr(self, "_accepting_async_tasks", True): + return None started_at = self._format_time() with self._lock: job = self._jobs.get(job_id) @@ -822,6 +830,38 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): # 协程收尾在事件循环上完成，同步路径（线程池/调用线程）提交到事件循环执行 await self.__finish_job(job_id=job_id, success=success, error=error) + def _track_async_task( + self, + task: asyncio.Task[Any] | ConcurrentFuture[Any], + ) -> asyncio.Task[Any] | ConcurrentFuture[Any]: + """登记 Scheduler 自有协程任务，并在完成后移除和消费异常。""" + tasks = getattr(self, "_async_tasks", None) + if tasks is None: + tasks = set() + self._async_tasks = tasks + + def _discard(done: asyncio.Task[Any] | ConcurrentFuture[Any]) -> None: + """释放已完成任务，并避免跨线程 Future 产生未取回异常。""" + with self._lock: + tasks.discard(done) + try: + done.exception() + except (asyncio.CancelledError, ConcurrentCancelledError): + pass + + with self._lock: + tasks.add(task) + task.add_done_callback(_discard) + return task + + def _create_async_task(self, coro: Any) -> bool: + """在当前事件循环创建并登记任务，返回是否已异步接管。""" + if not getattr(self, "_accepting_async_tasks", True): + coro.close() + return False + self._track_async_task(asyncio.create_task(coro)) + return True + def start(self, job_id: str, *args, **kwargs) -> None: """ 启动定时服务 @@ -837,12 +877,18 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): running_loop = None target_loop = global_vars.loop if running_loop: - asyncio.create_task(self.__run_coro_job(coro=coro, job_id=job_id, job=job)) - return True + return self._create_async_task( + self.__run_coro_job(coro=coro, job_id=job_id, job=job) + ) if target_loop and target_loop.is_running(): - asyncio.run_coroutine_threadsafe( - self.__run_coro_job(coro=coro, job_id=job_id, job=job), - target_loop, + if not getattr(self, "_accepting_async_tasks", True): + coro.close() + return False + self._track_async_task( + asyncio.run_coroutine_threadsafe( + self.__run_coro_job(coro=coro, job_id=job_id, job=job), + target_loop, + ) ) return True asyncio.run(coro) @@ -895,8 +941,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job_id=job_id, success=success, error=error )) - @staticmethod - def _submit_to_loop(coro: Any) -> None: + def _submit_to_loop(self, coro: Any) -> None: """ 把协程提交到事件循环执行，兼容以下调用环境： - 已在事件循环内（async 任务内部）：排队为独立任务，避免阻塞 @@ -908,12 +953,55 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): except RuntimeError: running_loop = None if running_loop: - asyncio.create_task(coro) + self._create_async_task(coro) elif global_vars.loop and global_vars.loop.is_running(): - asyncio.run_coroutine_threadsafe(coro, global_vars.loop) + if not getattr(self, "_accepting_async_tasks", True): + coro.close() + return + self._track_async_task( + asyncio.run_coroutine_threadsafe(coro, global_vars.loop) + ) else: asyncio.run(coro) + def _cancel_async_tasks(self) -> tuple[asyncio.Task[Any] | ConcurrentFuture[Any], ...]: + """停止接收新协程并请求取消现有 Scheduler 任务。""" + with self._lock: + self._accepting_async_tasks = False + tasks = tuple(getattr(self, "_async_tasks", ())) + for task in tasks: + if isinstance(task, asyncio.Task): + loop = task.get_loop() + if loop.is_running(): + loop.call_soon_threadsafe(task.cancel) + else: + task.cancel() + else: + task.cancel() + return tasks + + async def async_stop(self, *, timeout_seconds: float = 30.0) -> None: + """异步关闭 Scheduler，并在有限预算内等待其协程任务收口。""" + self.stop() + tasks = tuple(getattr(self, "_async_tasks", ())) + if not tasks: + return + awaitables = [] + for task in tasks: + if isinstance(task, asyncio.Future): + awaitables.append(task) + else: + awaitables.append(asyncio.wrap_future(task)) + try: + await asyncio.wait_for( + asyncio.gather(*awaitables, return_exceptions=True), + timeout=timeout_seconds, + ) + except asyncio.TimeoutError: + logger.error("等待定时器协程任务收口超时") + for task in tasks: + task.cancel() + @staticmethod def _get_agent_task_job_id(task_id: int) -> str: """生成 Agent 自主定时任务的调度器 Job ID。""" @@ -1396,6 +1484,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ 关闭定时服务 """ + self._cancel_async_tasks() with lock: try: if self._scheduler: diff --git a/app/startup/scheduler_initializer.py b/app/startup/scheduler_initializer.py index 038e45578..5d9688085 100644 --- a/app/startup/scheduler_initializer.py +++ b/app/startup/scheduler_initializer.py @@ -1,3 +1,5 @@ +import asyncio + from app.application.scheduling import register_scheduler_class from app.scheduler import Scheduler @@ -14,9 +16,15 @@ def init_scheduler(): def stop_scheduler(): """ - 停止定时器 + 停止定时器；生命周期事件循环中返回有限等待的兼容协程。 """ - Scheduler().stop() + scheduler = Scheduler() + try: + asyncio.get_running_loop() + except RuntimeError: + scheduler.stop() + return None + return scheduler.async_stop() def restart_scheduler(): diff --git a/docs/adr/0007-background-action-reliability.md b/docs/adr/0007-background-action-reliability.md index f17713380..a91c848d6 100644 --- a/docs/adr/0007-background-action-reliability.md +++ b/docs/adr/0007-background-action-reliability.md @@ -86,6 +86,8 @@ Event Contract Registry 是 53 个事件的逐项机器清单。下表按相同 - IMDb 同步 `clear_cache()` ABI 在事件循环内触发的异步缓存清理登记为 `module.imdb.cache_clear`；同步调用方式和无运行事件循环时的立即清理行为保持不变，宿主关停后不再 接受新的清理任务。 +- Scheduler 的协程作业与异步进度收尾由 Scheduler 自有任务集合持有；同步 `start()` / `stop()` ABI 保持， + 生命周期关闭入口额外等待有限预算，跨线程提交的 Future 也会在停止时收到取消请求。 ### Transfer pending / 文件整理 diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md index 3d3b44926..e36b4a244 100644 --- a/docs/refactor/backend-architecture-next-stage.md +++ b/docs/refactor/backend-architecture-next-stage.md @@ -838,6 +838,9 @@ ADR 必须逐个映射当前 Event、BackgroundTasks、Scheduler job、Agent tas 原语义；未启动完整 lifespan 的协议校验和旧直接调用通过兼容依赖回退到默认登记器。 - IMDb 同步清缓存兼容入口在运行事件循环时改由 TaskRegistry 登记异步缓存清理任务，owner 为 `module.imdb.cache_clear`；同步签名、模块调用方式和无事件循环时的立即清理语义保持不变。 +- Scheduler 的协程作业和异步进度收尾不再使用无主 `create_task` 或丢弃跨线程 Future；由 Scheduler 自有 + 任务集合登记、停止时取消，生命周期入口通过异步兼容包装器在有限预算内等待收口，保留旧同步 + `Scheduler.start()` / `Scheduler.stop()` 与插件调度 ABI。 #### ARCH-251：用现有数据库做首个 durable side-effect pilot diff --git a/tests/test_scheduler_cache_expiry.py b/tests/test_scheduler_cache_expiry.py index 7220b1621..5c19e0b64 100644 --- a/tests/test_scheduler_cache_expiry.py +++ b/tests/test_scheduler_cache_expiry.py @@ -1,5 +1,6 @@ +import asyncio import threading -from unittest.mock import Mock +from unittest.mock import AsyncMock, Mock from app import scheduler as scheduler_module from app.scheduler import Scheduler @@ -46,6 +47,32 @@ def test_scheduler_initializer_starts_background_jobs(monkeypatch): scheduler.init.assert_called_once_with() +def test_scheduler_initializer_stop_preserves_sync_abi(monkeypatch): + """同步调用停止入口仍应立即关闭 Scheduler，不返回 coroutine。""" + scheduler = Mock() + monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler)) + + assert scheduler_initializer.stop_scheduler() is None + scheduler.stop.assert_called_once_with() + scheduler.async_stop.assert_not_called() + + +def test_scheduler_initializer_stop_awaits_in_running_loop(monkeypatch): + """生命周期事件循环中的停止入口应返回可等待的异步收口。""" + scheduler = Mock() + scheduler.async_stop = AsyncMock() + monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler)) + + async def scenario(): + result = scheduler_initializer.stop_scheduler() + assert result is not None + await result + + asyncio.run(scenario()) + scheduler.async_stop.assert_awaited_once_with() + scheduler.stop.assert_not_called() + + def test_clear_cache_is_manual_only(monkeypatch): """缓存清理任务应仅手动执行，不注册到调度器自动运行。""" background_scheduler = _BackgroundSchedulerStub() diff --git a/tests/test_scheduler_progress.py b/tests/test_scheduler_progress.py index 6f6d10f32..465ecf7ae 100644 --- a/tests/test_scheduler_progress.py +++ b/tests/test_scheduler_progress.py @@ -11,7 +11,11 @@ from app.scheduler import Scheduler def _build_scheduler(job_id, func): """构造不启动 APScheduler 的定时服务测试对象。""" scheduler = object.__new__(Scheduler) + scheduler._scheduler = None + scheduler._event = threading.Event() scheduler._lock = threading.RLock() + scheduler._async_tasks = set() + scheduler._accepting_async_tasks = True scheduler._jobs = { job_id: { "name": "测试定时服务", @@ -169,6 +173,36 @@ def test_scheduler_records_cancelled_async_job_as_failed(): assert progress.error == "任务已取消" +def test_scheduler_async_stop_cancels_owned_async_jobs(): + """Scheduler 关停应取消并等待自身登记的异步作业。""" + job_id = f"test-owned-task-{uuid4()}" + started = asyncio.Event() + cancelled = asyncio.Event() + + async def task(): + """等待关停信号，验证任务确实由 Scheduler 持有。""" + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled.set() + raise + + scheduler = _build_scheduler(job_id, task) + + async def run_task(): + """在当前事件循环启动并收口异步作业。""" + scheduler.start(job_id) + await started.wait() + assert len(scheduler._async_tasks) == 1 + await scheduler.async_stop(timeout_seconds=1) + + asyncio.run(run_task()) + + assert cancelled.is_set() + assert scheduler._async_tasks == set() + + def test_scheduler_returns_none_for_unknown_job(): """未注册且无历史进度的定时服务应返回空。""" job_id = f"test-unknown-{uuid4()}"