diff --git a/app/scheduler.py b/app/scheduler.py index 77c67b945..01811e3e2 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -71,6 +71,7 @@ class _SchedulerHandle: loop: asyncio.AbstractEventLoop handle: asyncio.Future[Any] | concurrent.futures.Future[Any] completion: asyncio.Future[Any] | concurrent.futures.Future[Any] + kind: str # Agent 自主定时任务前缀下沉到 application 门面，此处保留兼容导出。 @@ -250,6 +251,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): loop: asyncio.AbstractEventLoop, handle: asyncio.Future[Any] | concurrent.futures.Future[Any], completion: asyncio.Future[Any] | concurrent.futures.Future[Any] | None = None, + kind: str = "job", ) -> bool: """登记调度器拥有的句柄；关闭竞态下拒绝并取消新句柄。""" if completion is None: @@ -269,6 +271,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): loop=loop, handle=handle, completion=completion, + kind=kind, ) completion.add_done_callback(self._remove_handle) return True @@ -321,6 +324,23 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): return_exceptions=True, ) + async def _await_progress_handles(self, job_id: str, generation: int) -> None: + """等待同一轮任务已提交的进度更新，保证最终状态最后写入缓存。""" + with self._lock: + handles = tuple( + handle + for handle in self._handles.values() + if handle.job_id == job_id + and handle.generation == generation + and handle.kind == "progress" + ) + if not handles: + return + await asyncio.gather( + *(self._wait_handle(handle) for handle in handles), + return_exceptions=True, + ) + @staticmethod def _track_cross_thread_completion( coro: Any, @@ -352,6 +372,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job_id: str, generation: int, on_unstarted_cancel: Optional[Callable[[], None]] = None, + kind: str = "job", ) -> bool: """向主循环提交协程，并以独立完成信号跟踪真实收尾。""" completion: concurrent.futures.Future[Any] = concurrent.futures.Future() @@ -415,6 +436,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): loop=target_loop, handle=handle, completion=completion, + kind=kind, ) handle.add_done_callback(cancel_target_task) return registered @@ -921,6 +943,9 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): """ 完成定时任务 """ + # 业务函数返回前提交的进度回调可能仍在等待 Redis I/O；先收敛它们， + # 避免迟到的 running 快照覆盖 success/failed 终态。 + await self._await_progress_handles(job_id, generation) finished_at = self._format_time() with self._lock: current_job = self._jobs.get(job_id) @@ -1126,6 +1151,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): _update(), job_id=job_id, generation=job.get("_generation", 0), + kind="progress", ) return update_progress @@ -1360,6 +1386,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job_id: str, generation: int = 0, on_unstarted_cancel: Optional[Callable[[], None]] = None, + kind: str = "job", ) -> bool: """ 把协程提交到事件循环执行，兼容以下调用环境： @@ -1390,6 +1417,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): generation=generation, loop=running_loop, handle=handle, + kind=kind, ) if on_unstarted_cancel: handle.add_done_callback( @@ -1407,6 +1435,7 @@ class Scheduler(ConfigReloadMixin, metaclass=SingletonClass): job_id=job_id, generation=generation, on_unstarted_cancel=on_unstarted_cancel, + kind=kind, ) elif self._lifecycle_state in {"stopping", "stopped"}: coro.close() diff --git a/app/schemas/subscribe.py b/app/schemas/subscribe.py index f1be43ca6..48911dc35 100644 --- a/app/schemas/subscribe.py +++ b/app/schemas/subscribe.py @@ -48,6 +48,12 @@ def compute_subscribe_completed_episode(subscribe: "Subscribe") -> Optional[int] class Subscribe(OptionalMediaIdentityMixin, BaseModel): """订阅输入与响应模型，媒体身份必须为空对或完整有效对。""" + # 表单用空字符串表达“全部”时必须保留显式清空语义，更新接口才能覆盖存量规则。 + CLEARABLE_FILTER_FIELDS: ClassVar[frozenset[str]] = frozenset({ + "filter", "include", "exclude", "quality", "resolution", "effect", + "audio_quality", "audio_format", + }) + # 公共创建和更新接口不得接收系统字段和运行事实；其余字段默认作为订阅输入透传。 PUBLIC_WRITE_EXCLUDED_FIELDS: ClassVar[frozenset[str]] = frozenset({ "id", "poster", "backdrop", "vote", "description", "lack_episode", "completed_episode", @@ -186,14 +192,14 @@ class Subscribe(OptionalMediaIdentityMixin, BaseModel): 音乐等媒体类型的 season、total_episode、episode_priority 等数值或容器字段 在表单中常以空字符串提交，而 Pydantic 不会把空字符串自动转为 None，会直接抛出 校验异常导致接口返回 422。这里把空字符串键移除，等价于该字段未提供，从而复用字段 - 默认值（如 ``total_episode`` 回退为 0、``sites`` 回退为空列表）。媒体身份键保留为 - None，以便更新接口区分“未提交”与“显式清空完整身份对”。 + 默认值（如 ``total_episode`` 回退为 0、``sites`` 回退为空列表）。媒体身份键以及可清空 + 的筛选字段保留为 None，以便更新接口区分“未提交”与“显式清空”。 """ if isinstance(data, dict): data = dict(data) for key, value in list(data.items()): if isinstance(value, str) and value == "": - if key in {"media_source", "media_id"}: + if key in {"media_source", "media_id"} or key in cls.CLEARABLE_FILTER_FIELDS: data[key] = None else: data.pop(key) diff --git a/tests/test_scheduler_lifecycle.py b/tests/test_scheduler_lifecycle.py index 09f8e2909..52500886b 100644 --- a/tests/test_scheduler_lifecycle.py +++ b/tests/test_scheduler_lifecycle.py @@ -340,16 +340,15 @@ async def test_sync_job_callback_and_finish_handles_are_owned(monkeypatch) -> No scheduler = _scheduler("callback-handles", job) await asyncio.to_thread(scheduler.start, "callback-handles") - await asyncio.wait_for( - asyncio.gather(update_started.wait(), finish_started.wait()), - timeout=1, - ) + await asyncio.wait_for(update_started.wait(), timeout=1) + await asyncio.sleep(0) assert len(scheduler._handles) == 2 + assert not finish_started.is_set() await scheduler.stop_async() - assert cancelled == 2 + assert cancelled == 1 assert scheduler._handles == {} @@ -387,6 +386,46 @@ async def test_stale_progress_cannot_update_replaced_job(monkeypatch) -> None: assert scheduler._handles == {} +@pytest.mark.anyio +async def test_final_progress_waits_for_pending_update(monkeypatch) -> None: + """任务终态必须等待已提交的进度回调，避免 running 快照迟到覆盖。""" + update_started = asyncio.Event() + allow_update = asyncio.Event() + finished = asyncio.Event() + writes = [] + + class BlockingProgress(_AsyncProgressStub): + """把中间进度写停在终态收尾之前。""" + + async def update(self, **_kwargs) -> None: + update_started.set() + await allow_update.wait() + writes.append("update") + + async def end(self, **_kwargs) -> None: + writes.append("end") + finished.set() + + async def job(progress_callback) -> None: + progress_callback(value=100, text="业务处理完成") + + monkeypatch.setattr(scheduler_module, "ProgressHelper", _ProgressStub) + monkeypatch.setattr(scheduler_module, "AsyncProgressHelper", BlockingProgress) + scheduler = _scheduler("progress-order", job) + + assert scheduler.start("progress-order") is True + await asyncio.wait_for(update_started.wait(), timeout=1) + await asyncio.sleep(0) + + assert writes == [] + assert not finished.is_set() + + allow_update.set() + await asyncio.wait_for(finished.wait(), timeout=1) + + assert writes == ["update", "end"] + + @pytest.mark.anyio async def test_replaced_job_keeps_active_state_without_stale_progress(monkeypatch) -> None: """同 ID 新 generation 显示真实运行态，但不继承旧任务进度详情。""" diff --git a/tests/test_scheduler_progress.py b/tests/test_scheduler_progress.py index ea5c9486f..0c50ce887 100644 --- a/tests/test_scheduler_progress.py +++ b/tests/test_scheduler_progress.py @@ -136,7 +136,13 @@ def test_scheduler_runs_async_job_from_current_event_loop(monkeypatch): async def run_task(): """从已运行的事件循环启动定时服务。""" scheduler.start(job_id) - await asyncio.sleep(0) + + async def wait_until_finished() -> None: + """等待任务及其异步进度句柄全部收敛。""" + while scheduler._handles or scheduler._active_job_generations: + await asyncio.sleep(0) + + await asyncio.wait_for(wait_until_finished(), timeout=1) scheduler = _build_scheduler(job_id, task) target_loop = asyncio.new_event_loop() diff --git a/tests/test_subscribe_filter_clear.py b/tests/test_subscribe_filter_clear.py new file mode 100644 index 000000000..781d1ce81 --- /dev/null +++ b/tests/test_subscribe_filter_clear.py @@ -0,0 +1,66 @@ +"""订阅更新显式清空筛选条件的回归测试。""" + +from types import SimpleNamespace + +import pytest + +from app.api.endpoints.subscribe import update_subscribe +from app.schemas.subscribe import Subscribe + + +class _SubscribeRow: + """提供更新端点所需字段的最小订阅替身。""" + + def __init__(self) -> None: + self.id = 1 + self.username = "alice" + self.type = "电影" + self.resolution = "4K" + self.total_episode = 0 + self.lack_episode = 0 + + def to_dict(self) -> dict: + """返回当前订阅快照。""" + return dict(self.__dict__) + + +class _MutationService: + """记录端点交给订阅写服务的更新 payload。""" + + def __init__(self, subscribe: _SubscribeRow) -> None: + self.subscribe = subscribe + self.payload = None + + async def get_accessible(self, _subscribe_id: int, _actor) -> _SubscribeRow: + """返回当前用户可访问的订阅。""" + return self.subscribe + + async def update(self, _subscribe_id: int, payload: dict, _actor, **_kwargs): + """应用更新并返回已发布事件的变更结果。""" + old = self.subscribe.to_dict() + self.payload = dict(payload) + self.subscribe.__dict__.update(payload) + return SimpleNamespace( + old=old, + new=self.subscribe.to_dict(), + event_published=True, + ) + + +@pytest.mark.anyio +async def test_update_subscribe_clears_explicit_empty_resolution() -> None: + """从 4K 切换到全部时，空字符串必须作为显式 None 写入而非被忽略。""" + subscribe = _SubscribeRow() + mutation = _MutationService(subscribe) + subscribe_in = Subscribe(id=1, resolution="") + + response = await update_subscribe( + subscribe_in=subscribe_in, + mutation=mutation, + current_user=SimpleNamespace(name="alice", is_superuser=False), + ) + + assert response.success is True + assert "resolution" in subscribe_in.model_fields_set + assert mutation.payload["resolution"] is None + assert subscribe.resolution is None