From bb2a5d38c8a8cd7697f36e4ab0d96156c28725c3 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Sun, 23 Aug 2026 11:10:16 +0800 Subject: [PATCH] fix(workflow): route async cache deletion through database worker --- app/api/dependencies/workflow.py | 17 ++++++++++------- app/application/configuration.py | 6 ++++++ app/application/workflow.py | 18 +++++++++++++++--- app/startup/context.py | 5 +++-- app/startup/modules_initializer.py | 2 +- .../architecture/dependency-baseline.json | 5 +++-- tests/test_agent_lifecycle.py | 5 ++++- tests/test_configuration_ports.py | 7 ++++++- tests/test_workflow_mutation_command.py | 10 +++++++--- 9 files changed, 55 insertions(+), 20 deletions(-) diff --git a/app/api/dependencies/workflow.py b/app/api/dependencies/workflow.py index ca156a55f..ff3acbda8 100644 --- a/app/api/dependencies/workflow.py +++ b/app/api/dependencies/workflow.py @@ -1,6 +1,6 @@ """工作流领域的请求级 command/query 依赖。""" -from typing import Any, cast +from typing import cast from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession @@ -10,6 +10,7 @@ from app.adapters.external.server import MoviePilotServerHelper from app.api.context import get_async_session, get_host_runtime, get_sync_session from app.application.scheduling import Scheduler from app.application.workflow import ( + WorkflowCachePort, WorkflowDefinitionCommand, WorkflowMutationCommand, WorkflowQueryService, @@ -26,6 +27,7 @@ def get_workflow_mutation_command( """组装请求级工作流写用例和提交后的调度副作用。""" scheduler = Scheduler() workflow_manager = WorkFlowManager() + system_config = cast(WorkflowCachePort, runtime.workflow.system_config()) return WorkflowMutationCommand( repository=runtime.workflow.repository(db), unit_of_work=runtime.persistence.sync_transaction(db), @@ -35,9 +37,9 @@ def get_workflow_mutation_command( remove_event=workflow_manager.remove_workflow_event, refresh_event=workflow_manager.update_workflow_event, stop_running=global_vars.stop_workflow, - delete_cache=lambda workflow_id: cast( - Any, runtime.workflow.system_config() - ).delete(f"WorkflowCache-{workflow_id}"), + delete_cache=lambda workflow_id: system_config.delete( + f"WorkflowCache-{workflow_id}" + ), ) @@ -46,13 +48,14 @@ def get_workflow_definition_command( runtime: HostRuntime = Depends(get_host_runtime), ) -> WorkflowDefinitionCommand: """组装工作流创建、复用和重置的异步写用例。""" + system_config = cast(WorkflowCachePort, runtime.workflow.system_config()) return WorkflowDefinitionCommand( repository=runtime.workflow.repository(db), unit_of_work=runtime.persistence.async_transaction(db), stop_running=global_vars.stop_workflow, - delete_cache=lambda workflow_id: cast( - Any, runtime.workflow.system_config() - ).delete(f"WorkflowCache-{workflow_id}"), + async_delete_cache=lambda workflow_id: system_config.async_delete( + f"WorkflowCache-{workflow_id}" + ), report_fork=MoviePilotServerHelper.async_workflow_fork_by_id, ) diff --git a/app/application/configuration.py b/app/application/configuration.py index 83c3eabcc..7db7c5861 100644 --- a/app/application/configuration.py +++ b/app/application/configuration.py @@ -318,6 +318,12 @@ class SystemConfigService: result = await self._async_executor.run(partial(self._writer.set, key, value)) return cast(bool | None, result) + async def async_delete(self, key: Any) -> Any: + """异步删除配置，并等待数据库提交或回滚完成。""" + if self._async_executor is None: + raise RuntimeError("系统配置异步数据库执行端口尚未配置") + return await self._async_executor.run(partial(self._writer.delete, key)) + def delete(self, key: Any) -> Any: """删除配置。""" return self._writer.delete(key) diff --git a/app/application/workflow.py b/app/application/workflow.py index 580902b73..a1a0843bd 100644 --- a/app/application/workflow.py +++ b/app/application/workflow.py @@ -29,6 +29,18 @@ class AsyncWorkflowQueryRepository(Protocol): ... +class WorkflowCachePort(Protocol): + """工作流重置所需的同步与异步配置缓存端口。""" + + def delete(self, key: Any) -> Any: + """删除配置缓存。""" + ... + + async def async_delete(self, key: Any) -> Any: + """通过异步数据库执行端口删除配置缓存。""" + ... + + class WorkflowQueryService: """提供工作流列表和详情查询，隔离 API 与数据库会话。""" @@ -364,14 +376,14 @@ class WorkflowDefinitionCommand: repository: AsyncWorkflowDefinitionRepository, unit_of_work: AsyncUnitOfWork, stop_running: Callable[[int], None], - delete_cache: Callable[[int], None], + async_delete_cache: Callable[[int], Awaitable[Any]], report_fork: Optional[Callable[[int], Awaitable[object]]] = None, ) -> None: """保存异步事务和提交后运行时副作用端口。""" self._repository = repository self._unit_of_work = unit_of_work self._stop_running = stop_running - self._delete_cache = delete_cache + self._async_delete_cache = async_delete_cache self._report_fork = report_fork async def create(self, payload: Mapping[str, Any]) -> WorkflowMutationResult: @@ -448,7 +460,7 @@ class WorkflowDefinitionCommand: await self._repository.stage_reset(workflow_id, reset_count=True) await self._commit() self._stop_running(workflow_id) - self._delete_cache(workflow_id) + await self._async_delete_cache(workflow_id) return WorkflowMutationResult(True) async def _commit(self) -> None: diff --git a/app/startup/context.py b/app/startup/context.py index a51d7ded0..2ba54b23c 100644 --- a/app/startup/context.py +++ b/app/startup/context.py @@ -1,6 +1,6 @@ """宿主启动阶段构建的类型化运行时上下文。""" -from collections.abc import AsyncGenerator, Generator +from collections.abc import AsyncGenerator, Callable, Generator from dataclasses import dataclass from typing import Protocol @@ -16,6 +16,7 @@ from app.application.subscription.mutation import ( SubscriptionHistoryMutationRepository, SubscriptionMutationRepository, ) +from app.application.workflow import WorkflowCachePort class AgentChatRepositoryFactory(Protocol): @@ -162,7 +163,7 @@ class WorkflowRuntime: """工作流定义、状态与缓存操作所需的数据工厂。""" repository: RepositoryFactory - system_config: StandaloneRepositoryFactory + system_config: Callable[[], WorkflowCachePort] @dataclass(frozen=True, slots=True) diff --git a/app/startup/modules_initializer.py b/app/startup/modules_initializer.py index a3af35fc2..4cd1f2286 100644 --- a/app/startup/modules_initializer.py +++ b/app/startup/modules_initializer.py @@ -671,7 +671,7 @@ async def init_modules() -> HostRuntime: ), workflow=WorkflowRuntime( repository=WorkflowOper, - system_config=SystemConfigOper, + system_config=get_configured_system_config, ), configuration=runtime_configuration, settings=runtime_settings, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 8b33ab2f0..0d992facf 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -13,8 +13,8 @@ "runtime_to_db": [], "workflow_to_db": [] }, - "edge_count": 6434, - "edge_sha256": "5c47cae41f5d90e1030757d0cb47db8eed373424dda5db17c900271e1d03a9c8", + "edge_count": 6435, + "edge_sha256": "ed079837faf943d744ba7c6dee9172449bc204a2068a89db75e030e0be12fbb2", "edges": [ "app -> app.runtime", "app -> app.runtime.compat", @@ -5979,6 +5979,7 @@ "app.startup.context -> app.application.subscription.delete", "app.startup.context -> app.application.subscription.identity", "app.startup.context -> app.application.subscription.mutation", + "app.startup.context -> app.application.workflow", "app.startup.database -> app.adapters", "app.startup.database -> app.adapters.system", "app.startup.database -> app.adapters.system.backup", diff --git a/tests/test_agent_lifecycle.py b/tests/test_agent_lifecycle.py index 8c6209fa7..b610bb032 100644 --- a/tests/test_agent_lifecycle.py +++ b/tests/test_agent_lifecycle.py @@ -139,7 +139,10 @@ async def test_agent_initialization_failure_does_not_stop_module_startup( monkeypatch.setattr(modules_initializer, "check_auth", check_auth) try: - await modules_initializer.init_modules() + runtime = await modules_initializer.init_modules() + assert runtime.workflow.system_config() is ( + modules_initializer.get_configured_system_config() + ) finally: await modules_initializer.stop_database_worker() diff --git a/tests/test_configuration_ports.py b/tests/test_configuration_ports.py index 6ada1e34e..c79fa355e 100644 --- a/tests/test_configuration_ports.py +++ b/tests/test_configuration_ports.py @@ -88,6 +88,7 @@ def test_system_config_service_supports_separate_reader_and_writer() -> None: reader.get.return_value = "old" writer = MagicMock() writer.set.return_value = True + writer.delete.return_value = True service = SystemConfigService( reader=reader, writer=writer, @@ -98,13 +99,17 @@ def test_system_config_service_supports_separate_reader_and_writer() -> None: assert service.set("key", "new") is True assert asyncio.run(service.async_set("key", "new")) is True service.delete("key") + assert asyncio.run(service.async_delete("key")) is True reader.get.assert_called_once_with("key") assert writer.set.call_args_list == [ (("key", "new"), {}), (("key", "new"), {}), ] - writer.delete.assert_called_once_with("key") + assert writer.delete.call_args_list == [ + (("key",), {}), + (("key",), {}), + ] def test_user_configuration_service_supports_sync_and_async_writes() -> None: diff --git a/tests/test_workflow_mutation_command.py b/tests/test_workflow_mutation_command.py index 03f5589c7..0f587cb9a 100644 --- a/tests/test_workflow_mutation_command.py +++ b/tests/test_workflow_mutation_command.py @@ -192,7 +192,7 @@ def _definition_command(*, existing=None, commit_error=None, report_fork=None): "repository": repository, "unit_of_work": unit_of_work, "stop_running": Mock(), - "delete_cache": Mock(), + "async_delete_cache": AsyncMock(), "report_fork": report_fork or AsyncMock(), } return WorkflowDefinitionCommand(**dependencies), dependencies @@ -279,7 +279,7 @@ async def test_reset_commit_failure_does_not_stop_runtime_or_delete_cache(): dependencies["unit_of_work"].rollback.assert_awaited_once_with() dependencies["stop_running"].assert_not_called() - dependencies["delete_cache"].assert_not_called() + dependencies["async_delete_cache"].assert_not_awaited() @pytest.mark.asyncio @@ -289,9 +289,13 @@ async def test_reset_commits_before_runtime_cleanup(): command, dependencies = _definition_command(existing=_workflow()) dependencies["unit_of_work"].commit.side_effect = lambda: calls.append("commit") dependencies["stop_running"].side_effect = lambda _id: calls.append("stop") - dependencies["delete_cache"].side_effect = lambda _id: calls.append("cache") + async def delete_cache(_id): + calls.append("cache") + + dependencies["async_delete_cache"].side_effect = delete_cache result = await command.reset(7) assert result.success is True assert calls == ["commit", "stop", "cache"] + dependencies["async_delete_cache"].assert_awaited_once_with(7)