mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-04 06:15:14 +08:00
refactor(startup): enforce symmetric lifecycle cleanup
This commit is contained in:
jxxghp
@@ -229,6 +229,11 @@ class ModuleManager(metaclass=Singleton):
|
||||
converged = self._runtime.shutdown(reason="application_shutdown")
|
||||
self._refresh_running_projection()
|
||||
if converged:
|
||||
eventmanager.remove_event_listener(
|
||||
EventType.ConfigChanged,
|
||||
self.handle_config_changed,
|
||||
)
|
||||
eventmanager.unregister_handler_instance_resolver("modules")
|
||||
logger.info("模块运行时关闭完成")
|
||||
else:
|
||||
logger.error("模块运行时关闭后仍有资源 owner 未收敛")
|
||||
|
||||
@@ -14,6 +14,7 @@ from app.application.database import (
|
||||
DatabaseGovernance,
|
||||
DatabaseHealthService,
|
||||
configure_database_governance,
|
||||
reset_database_governance,
|
||||
)
|
||||
from app.application.maintenance import (
|
||||
DataCleanupService,
|
||||
@@ -165,6 +166,7 @@ def publish_database_services(composition: DatabaseComposition) -> None:
|
||||
|
||||
def reset_database_services() -> None:
|
||||
"""撤销当前 lifespan 发布的全部数据库派生服务。"""
|
||||
reset_database_governance()
|
||||
reset_transaction_runners()
|
||||
reset_workflow_query()
|
||||
reset_plugin_persistence()
|
||||
|
||||
@@ -1,8 +1,7 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import sys
|
||||
from collections.abc import Awaitable
|
||||
from typing import Callable, cast
|
||||
from typing import Callable
|
||||
|
||||
from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper
|
||||
|
||||
@@ -21,11 +20,13 @@ from app.adapters.system.host import SystemUtils
|
||||
from app.adapters.system.resource import (
|
||||
ResourceHelper,
|
||||
configure_resource_version_provider,
|
||||
reset_resource_version_provider,
|
||||
)
|
||||
from app.application.configuration import (
|
||||
TransferRetryConfig,
|
||||
configure_transfer_retry_config,
|
||||
get_configured_system_config,
|
||||
reset_transfer_retry_config,
|
||||
)
|
||||
from app.application.messaging.agent import (
|
||||
dispatch_web_agent_message_event,
|
||||
@@ -39,11 +40,11 @@ from app.application.messaging.message import (
|
||||
MessageHelper,
|
||||
stop_message,
|
||||
)
|
||||
from app.application.module import configure_module_runtime
|
||||
from app.application.module import configure_module_runtime, reset_module_runtime
|
||||
from app.application.outbox import configure_outbox_dispatcher
|
||||
from app.application.security.url import close_image_proxy_block_log_coalescer
|
||||
from app.application.service import configure_service_directory
|
||||
from app.application.workflow import configure_workflow_execution
|
||||
from app.application.service import configure_service_directory, reset_service_directory
|
||||
from app.application.workflow import configure_workflow_execution, reset_workflow_execution
|
||||
from app.command import CommandChain
|
||||
from app.db.adapters.workflow import (
|
||||
TransactionalWorkflowExecutionService,
|
||||
@@ -59,6 +60,7 @@ from app.runtime.extensions.module_manager import ModuleManager
|
||||
from app.runtime.extensions.service_config import (
|
||||
ServiceConfigHelper,
|
||||
configure_service_config_reader,
|
||||
reset_service_config_reader,
|
||||
)
|
||||
from app.runtime.log import logger
|
||||
from app.runtime.settings import (
|
||||
@@ -72,10 +74,12 @@ from app.schemas.types import EventType, SystemConfigKey
|
||||
from app.startup.composition.agent import (
|
||||
compose_agent,
|
||||
publish_agent_services,
|
||||
reset_agent_services,
|
||||
)
|
||||
from app.startup.composition.chain import (
|
||||
configure_chain_runtime_context,
|
||||
configure_wallpaper_services,
|
||||
reset_chain_services,
|
||||
)
|
||||
from app.startup.composition.configuration import (
|
||||
compose_configuration,
|
||||
@@ -92,25 +96,39 @@ from app.startup.composition.database import (
|
||||
start_database_runtime,
|
||||
stop_database_runtime,
|
||||
)
|
||||
from app.startup.composition.network import configure_application_network_ports
|
||||
from app.startup.composition.outbox import build_outbox_dispatcher
|
||||
from app.startup.composition.network import (
|
||||
configure_application_network_ports,
|
||||
reset_application_network_ports,
|
||||
)
|
||||
from app.startup.composition.outbox import build_outbox_dispatcher, reset_outbox_services
|
||||
from app.startup.composition.runtime import (
|
||||
RuntimeInputs,
|
||||
compose_runtime,
|
||||
compose_runtime_dependencies,
|
||||
publish_runtime,
|
||||
reset_runtime,
|
||||
)
|
||||
from app.startup.composition.security import (
|
||||
configure_security_access,
|
||||
configure_security_services,
|
||||
reset_security_access,
|
||||
reset_security_services,
|
||||
)
|
||||
from app.startup.composition.server import configure_server_services, reset_server_services
|
||||
from app.startup.initializers.agent import (
|
||||
configure_agent_data_context,
|
||||
init_agent,
|
||||
reset_agent_data_context,
|
||||
)
|
||||
from app.startup.composition.server import configure_server_services
|
||||
from app.startup.initializers.agent import configure_agent_data_context, init_agent
|
||||
from app.startup.initializers.resources import (
|
||||
init_managed_resources,
|
||||
reset_managed_resources,
|
||||
stop_managed_resources,
|
||||
)
|
||||
from app.startup.initializers.scheduler import configure_scheduler_agent_tasks
|
||||
from app.startup.initializers.scheduler import (
|
||||
configure_scheduler_agent_tasks,
|
||||
reset_scheduler_bindings,
|
||||
)
|
||||
|
||||
|
||||
def configure_runtime_data_providers() -> None:
|
||||
@@ -123,6 +141,88 @@ def configure_runtime_data_providers() -> None:
|
||||
)
|
||||
|
||||
|
||||
def reset_runtime_data_providers() -> None:
|
||||
"""按发布逆序撤销模块、服务目录和配置读取 Provider。"""
|
||||
reset_service_directory()
|
||||
reset_module_runtime()
|
||||
reset_service_config_reader()
|
||||
|
||||
|
||||
def reset_event_services() -> None:
|
||||
"""撤销启动组合根登记的事件通知、监听器和宿主 resolver。"""
|
||||
event_manager = EventManager.get_existing_instance()
|
||||
if event_manager is None:
|
||||
return
|
||||
event_manager.remove_event_listener(
|
||||
EventType.NoticeMessage,
|
||||
dispatch_web_agent_message_event,
|
||||
)
|
||||
event_manager.unregister_handler_instance_resolver("host")
|
||||
event_manager.reset_error_notifier()
|
||||
|
||||
|
||||
def reset_tool_services() -> None:
|
||||
"""撤销工具管理器持有的 Agent 数据上下文与目录快照。"""
|
||||
from app.agent.tools.manager import moviepilot_tool_manager
|
||||
|
||||
moviepilot_tool_manager.reset_data_context()
|
||||
|
||||
|
||||
def _module_provider_reset_steps() -> tuple[tuple[str, Callable[[], object]], ...]:
|
||||
"""返回当前模块 lifespan 的完整 Provider 逆序撤销清单。"""
|
||||
return (
|
||||
("事件服务", reset_event_services),
|
||||
("资源版本 Provider", reset_resource_version_provider),
|
||||
("认证访问服务", reset_security_access),
|
||||
("Chain 服务", reset_chain_services),
|
||||
("托管资源引用", reset_managed_resources),
|
||||
("调度器绑定", reset_scheduler_bindings),
|
||||
("Agent 工具服务", reset_tool_services),
|
||||
(
|
||||
"Agent 服务",
|
||||
lambda: reset_agent_services(
|
||||
data_context_resetter=reset_agent_data_context,
|
||||
),
|
||||
),
|
||||
("传输重试配置", reset_transfer_retry_config),
|
||||
("Outbox 服务", reset_outbox_services),
|
||||
("工作流执行服务", reset_workflow_execution),
|
||||
("中心服务", reset_server_services),
|
||||
("运行时数据 Provider", reset_runtime_data_providers),
|
||||
("HostRuntime 投影", reset_runtime),
|
||||
("数据库服务", reset_database_services),
|
||||
("认证服务", reset_security_services),
|
||||
("应用网络端口", reset_application_network_ports),
|
||||
("配置服务", reset_configuration),
|
||||
)
|
||||
|
||||
|
||||
def reset_module_providers() -> bool:
|
||||
"""逐项撤销模块 Provider；单项失败不跳过后续 owner，并返回整体结果。"""
|
||||
converged = True
|
||||
for name, callback in _module_provider_reset_steps():
|
||||
try:
|
||||
callback()
|
||||
except Exception as error:
|
||||
logger.error("撤销%s失败：%s", name, error)
|
||||
converged = False
|
||||
return converged
|
||||
|
||||
|
||||
def _existing_singleton(instance_type: type) -> object | None:
|
||||
"""读取已经物化的 Singleton，启动失败回滚时禁止反向创建新 owner。"""
|
||||
getter = getattr(instance_type, "get_existing_instance", None)
|
||||
return getter() if callable(getter) else None
|
||||
|
||||
|
||||
def _call_existing_singleton(instance_type: type, method_name: str) -> object:
|
||||
"""调用已存在 Singleton 的关闭方法；owner 未创建时视为已经收敛。"""
|
||||
instance = _existing_singleton(instance_type)
|
||||
if instance is None:
|
||||
return True
|
||||
return getattr(instance, method_name)()
|
||||
|
||||
|
||||
def notify_event_error(title: str, message: str) -> None:
|
||||
"""将事件总线错误转发到系统消息通道。"""
|
||||
MessageHelper().put(
|
||||
@@ -345,14 +445,36 @@ async def stop_modules() -> bool:
|
||||
return converged
|
||||
|
||||
await run_step("图片代理安全日志合并器", close_image_proxy_block_log_coalescer)
|
||||
await run_step("模块", lambda: ModuleManager().shutdown(), offload=True)
|
||||
await run_step("事件消费", lambda: EventManager().stop_async())
|
||||
await run_step(
|
||||
"模块",
|
||||
lambda: _call_existing_singleton(ModuleManager, "shutdown"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"事件消费",
|
||||
lambda: _call_existing_singleton(EventManager, "stop_async"),
|
||||
)
|
||||
await run_step("浏览器会话", close_browser_sessions, offload=True)
|
||||
await run_step("托管资源", stop_managed_resources)
|
||||
await run_step("DoH服务", lambda: DohHelper().shutdown(), offload=True)
|
||||
await run_step("线程池", lambda: ThreadHelper().shutdown(), offload=True)
|
||||
await run_step("Redis缓存连接", lambda: RedisHelper().close(), offload=True)
|
||||
await run_step("异步Redis缓存连接", lambda: AsyncRedisHelper().close())
|
||||
await run_step(
|
||||
"DoH服务",
|
||||
lambda: _call_existing_singleton(DohHelper, "shutdown"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"线程池",
|
||||
lambda: _call_existing_singleton(ThreadHelper, "shutdown"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"Redis缓存连接",
|
||||
lambda: _call_existing_singleton(RedisHelper, "close"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"异步Redis缓存连接",
|
||||
lambda: _call_existing_singleton(AsyncRedisHelper, "close"),
|
||||
)
|
||||
# Web Agent 的取消 finally 可能还要写入最终展示快照，必须先完成任务收尾，再关闭写入准入。
|
||||
web_agent_drained = await run_step(
|
||||
"Web Agent后台任务",
|
||||
@@ -362,29 +484,38 @@ async def stop_modules() -> bool:
|
||||
if not web_agent_drained:
|
||||
web_agent_drained = await run_step("Web Agent后台任务收尾", wait_web_agent_background_tasks)
|
||||
if web_agent_drained:
|
||||
await run_step(
|
||||
"Agent会话持久化准入",
|
||||
lambda: get_configured_agent_chat_persistence().begin_shutdown(),
|
||||
)
|
||||
persistence_drained = await run_step(
|
||||
"Agent会话持久化",
|
||||
lambda: get_configured_agent_chat_persistence().shutdown(),
|
||||
)
|
||||
try:
|
||||
persistence = get_configured_agent_chat_persistence()
|
||||
except RuntimeError:
|
||||
persistence = None
|
||||
if persistence is None:
|
||||
persistence_drained = True
|
||||
else:
|
||||
await run_step(
|
||||
"Agent会话持久化准入",
|
||||
persistence.begin_shutdown,
|
||||
)
|
||||
persistence_drained = await run_step(
|
||||
"Agent会话持久化",
|
||||
persistence.shutdown,
|
||||
)
|
||||
else:
|
||||
persistence_drained = False
|
||||
all_converged = False
|
||||
logger.error("Web Agent任务未完成收尾，跳过持久化和数据库关闭以保护活动事务")
|
||||
if persistence_drained:
|
||||
await run_step("数据库任务", stop_database_runtime)
|
||||
await run_step("前端服务", stop_frontend, offload=True)
|
||||
await run_step("临时文件", clear_temp, offload=True)
|
||||
if persistence_drained:
|
||||
if not database_runtime_active():
|
||||
await run_step("数据库服务", reset_database_services)
|
||||
await run_step("配置服务", reset_configuration)
|
||||
await run_step("模块 Provider", reset_module_providers)
|
||||
await run_step("数据库连接", close_database)
|
||||
else:
|
||||
all_converged = False
|
||||
logger.error("数据库任务未收敛，跳过数据库连接关闭以避免运行中事务使用已释放连接")
|
||||
await run_step("前端服务", stop_frontend, offload=True)
|
||||
await run_step("临时文件", clear_temp, offload=True)
|
||||
logger.error(
|
||||
"数据库任务未收敛，保留全部 Provider 和数据库连接以供诊断与重试"
|
||||
)
|
||||
return all_converged
|
||||
|
||||
|
||||
@@ -509,14 +640,8 @@ async def init_modules() -> HostRuntime:
|
||||
except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常
|
||||
logger.error(f"模块启动失败后的消息资源清理失败：{cleanup_error}")
|
||||
try:
|
||||
await stop_database_runtime()
|
||||
if await stop_modules() is False:
|
||||
logger.error("模块启动失败后的资源 owner 未完全收敛")
|
||||
except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常
|
||||
logger.error(f"模块启动失败后的数据库任务清理失败：{cleanup_error}")
|
||||
if not database_runtime_active():
|
||||
reset_database_services()
|
||||
reset_configuration()
|
||||
try:
|
||||
await cast(Callable[[], Awaitable[None]], close_database)()
|
||||
except Exception as cleanup_error: # noqa: BLE001 保留原始启动异常
|
||||
logger.error(f"模块启动失败后的数据库连接清理失败：{cleanup_error}")
|
||||
logger.error(f"模块启动失败后的统一资源清理失败：{cleanup_error}")
|
||||
raise
|
||||
|
||||
@@ -31,6 +31,15 @@ def reset_scheduler_runtime() -> None:
|
||||
reset_scheduler_class()
|
||||
|
||||
|
||||
def reset_scheduler_bindings() -> None:
|
||||
"""在 Scheduler 已停止后撤销当前 lifespan 的仓储、业务能力和类登记。"""
|
||||
# Singleton 元类保留旧动态接口，当前 owner 在此精确收窄为 Scheduler。
|
||||
scheduler = Scheduler.get_existing_instance() # type: ignore[no-untyped-call]
|
||||
if scheduler is not None:
|
||||
scheduler.reset_runtime_bindings()
|
||||
reset_scheduler_runtime()
|
||||
|
||||
|
||||
def configure_scheduler_agent_tasks(repository: AgentTaskRepository) -> None:
|
||||
"""在调度器启动前注入自主任务仓储。"""
|
||||
configure_scheduler_runtime()
|
||||
@@ -91,18 +100,14 @@ def stop_scheduler() -> Optional[Awaitable[None]]:
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
try:
|
||||
scheduler.stop()
|
||||
finally:
|
||||
reset_scheduler_runtime()
|
||||
scheduler.stop()
|
||||
reset_scheduler_bindings()
|
||||
return None
|
||||
|
||||
async def stop_and_reset() -> None:
|
||||
"""等待异步调度器收口后清除 provider。"""
|
||||
try:
|
||||
await scheduler.stop_async()
|
||||
finally:
|
||||
reset_scheduler_runtime()
|
||||
await scheduler.stop_async()
|
||||
reset_scheduler_bindings()
|
||||
|
||||
return stop_and_reset()
|
||||
|
||||
|
||||
@@ -78,7 +78,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
|
||||
| Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
|
||||
| Python 源码量 | 约 298,700 行 | 60 个文件超过 1,000 行，10 个超过 2,000 行 |
|
||||
| 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 |
|
||||
| 全量 mypy 历史债务 | 10,320 / 592 文件 | Runtime 唯一组合 owner 零新增类型债，`modules.py` 低水位下降 4；strict frontier 当前覆盖 41 个文件 |
|
||||
| 全量 mypy 历史债务 | 10,317 / 592 文件 | 生命周期统一撤销接线继续清除 `modules.py` 3 项调用类型债；strict frontier 当前覆盖 41 个文件 |
|
||||
| Ruff 历史诊断 | 644 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
|
||||
| 覆盖率低水位 | Application 81.00%，Domain 80.67% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
|
||||
|
||||
@@ -678,7 +678,8 @@ Pylint 10/10、Ruff、mypy/复杂度 ratchet、宿主与最新官方插件基线
|
||||
- [x] L4.3 已将 HostRuntime、命名领域 Runtime 和旧 `ApiDataPorts` 投影统一移到单词型
|
||||
`startup/composition/runtime.py`；`RuntimeDependencies` 由 Agent、Chain、HostRuntime 复用，
|
||||
`HostRuntime.tasks` 显式绑定 lifecycle 的 TaskRegistry，Chain 不再二次构造 transfer execution。
|
||||
- [ ] 保留生命周期 manifest 和顺序快照；高扇出在组合根是允许的，但业务实现不能继续沉积其中。
|
||||
- [x] 生命周期按显式逆序 reset manifest 撤销全部 Provider；数据库 worker 未收敛时保留
|
||||
HostRuntime、Provider 与连接供诊断重试，启动失败复用同一正式清理路径且不反向物化 owner。
|
||||
|
||||
### ARCH-204 收敛 sync/async 重复与测试存量
|
||||
|
||||
|
||||
@@ -168,7 +168,7 @@ canonical 主程序；兼容只经统一 Compat/SDK 门面提供。
|
||||
| S5-L1 PluginHelper/PluginManager | `DELIVERED` | S2,S3 | `fa40f29df`：市场/包/依赖/备份/健康服务各归 owner；Facade 只转发稳定 ABI，构造归 typed PluginRuntime |
|
||||
| S5-L2 Agent/LLM provider | `DELIVERED` | S3-L7 | `c000c4fff`、`3a33da0b3`、`dbdf7d058`：catalog、发现、认证、session、runtime 分离；Manager/Facade 只保留稳定 API，旧包根符号仅由精确 Compat 承接 |
|
||||
| S5-L3 Domain projection | `VERIFIED` | S3-L6 | `MediaInfo` canonical 路径和 setter ABI 保留；四来源纯投影归入单词 owner，输入不可变、重复业务语义清零 |
|
||||
| S5-L4 Startup composition | `PLANNED` | S2,S3,S5-L1,S5-L2 | `initializers/modules.py` 仅负责顺序/注册/重启决策，构造按领域进入 composition |
|
||||
| S5-L4 Startup composition | `VERIFIED` | S2,S3,S5-L1,S5-L2 | 领域构造全部归单词型 composition owner；HostRuntime/ApiData 同源；显式逆序 reset manifest 在 worker 收敛后撤销 Provider，失败保留完整诊断边界 |
|
||||
| S5-L5 Sync/async 重复清零 | `PLANNED` | S3,S5 | 双 ABI 外壳共享纯逻辑，重复业务实现清零，Session/客户端不跨并发边界复用 |
|
||||
| S5-L6 最终兼容与交付 | `PLANNED` | S3,S5-L1,S5-L2,S5-L3,S5-L4,S5-L5 | canonical 旧实现/重复导出清零；同步官方插件仓验证；锁定全量、Pylint、架构、现有类型/覆盖率门禁全部通过并推送；完成后结束本轮战略任务 |
|
||||
|
||||
|
||||
@@ -172,13 +172,14 @@ autonomous-task repository/execution service set. The runtime owner projects tho
|
||||
objects into `AgentChatRuntime`. Agent composition may accept
|
||||
the Agent initializer's data-context registrar as a callback, but must not import
|
||||
`startup.initializers`; tool-manager, scheduler and Agent activation order remains in
|
||||
the initializer. Database runtime ownership rejects overlapping lifespans; after a
|
||||
successful worker shutdown, the
|
||||
configuration/database composition owners must revoke every provider they published
|
||||
before releasing database connections. Other domain providers remain owned by their
|
||||
matching composition batch and must satisfy the same symmetric lifecycle contract
|
||||
before S5-L4 closes. A failed shutdown retains both the runtime owner and its providers
|
||||
for diagnosis and retry instead of publishing a second owner.
|
||||
the initializer. Database runtime ownership rejects overlapping lifespans. After a
|
||||
successful worker shutdown, `initializers/modules.py` invokes the explicit reverse-order
|
||||
reset manifest: event/resource/security/Chain/managed-resource/Scheduler/tool/Agent,
|
||||
then transfer/outbox/workflow/server/runtime/database/network/configuration providers,
|
||||
and only afterward releases database connections. Startup rollback uses the same reset
|
||||
manifest and never materializes an owner merely to stop it. A failed database shutdown
|
||||
retains the runtime owner, every provider and the connection boundary for diagnosis and
|
||||
retry instead of publishing a second owner or silently entering a partial lifespan.
|
||||
Startup publishes its frozen, slotted `HostRuntime` through FastAPI `app.state`.
|
||||
API dependencies must narrow that object to a domain runtime (for example,
|
||||
`AgentChatRuntime`) instead of adding a string key to a global service map.
|
||||
|
||||
@@ -3281,7 +3281,7 @@
|
||||
"attr-defined": 2,
|
||||
"misc": 1,
|
||||
"no-any-return": 2,
|
||||
"no-untyped-call": 14,
|
||||
"no-untyped-call": 11,
|
||||
"no-untyped-def": 6,
|
||||
"return-value": 2
|
||||
},
|
||||
|
||||
@@ -357,11 +357,11 @@ async def test_init_modules_cleans_partial_message_owner_on_failure(monkeypatch)
|
||||
"_initialize_modules",
|
||||
failing_initialize_modules,
|
||||
)
|
||||
stop_database_runtime = AsyncMock()
|
||||
stop_modules = AsyncMock(return_value=True)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"stop_database_runtime",
|
||||
stop_database_runtime,
|
||||
"stop_modules",
|
||||
stop_modules,
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
@@ -370,4 +370,4 @@ async def test_init_modules_cleans_partial_message_owner_on_failure(monkeypatch)
|
||||
assert raised.value is startup_error
|
||||
close.assert_called_once_with()
|
||||
assert MessageHelper.get_existing_instance() is None
|
||||
stop_database_runtime.assert_awaited_once_with()
|
||||
stop_modules.assert_awaited_once_with()
|
||||
|
||||
@@ -1257,16 +1257,9 @@ def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch):
|
||||
"database_runtime_active",
|
||||
lambda: database_state["active"],
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"reset_database_services",
|
||||
MagicMock(side_effect=lambda: order.append("database-services")),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"reset_configuration",
|
||||
MagicMock(side_effect=lambda: order.append("configuration")),
|
||||
)
|
||||
dependencies["reset_module_providers"].side_effect = lambda: order.append(
|
||||
"providers"
|
||||
) or True
|
||||
dependencies["close_database"].side_effect = lambda: order.append("connection")
|
||||
|
||||
converged = asyncio.run(modules_initializer.stop_modules())
|
||||
@@ -1277,12 +1270,65 @@ def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch):
|
||||
"persistence-admission",
|
||||
"persistence",
|
||||
"database",
|
||||
"database-services",
|
||||
"configuration",
|
||||
"providers",
|
||||
"connection",
|
||||
]
|
||||
|
||||
|
||||
def test_stop_modules_retains_providers_when_database_worker_remains_active(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
"""数据库 worker 未收敛时不得撤销 Provider 或关闭仍被使用的连接。"""
|
||||
dependencies = _patch_module_shutdown_dependencies(monkeypatch)
|
||||
persistence = MagicMock()
|
||||
persistence.begin_shutdown = MagicMock()
|
||||
persistence.shutdown = AsyncMock()
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"get_configured_agent_chat_persistence",
|
||||
MagicMock(return_value=persistence),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"stop_database_runtime",
|
||||
AsyncMock(side_effect=RuntimeError("database busy")),
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"database_runtime_active",
|
||||
lambda: True,
|
||||
)
|
||||
|
||||
assert asyncio.run(modules_initializer.stop_modules()) is False
|
||||
dependencies["reset_module_providers"].assert_not_called()
|
||||
dependencies["close_database"].assert_not_awaited()
|
||||
|
||||
|
||||
def test_reset_module_providers_preserves_reverse_order_after_failure(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
"""单个 Provider reset 失败时仍按既定逆序执行全部后续 owner。"""
|
||||
calls: list[str] = []
|
||||
|
||||
def fail() -> None:
|
||||
"""记录失败 owner 后抛出异常。"""
|
||||
calls.append("second")
|
||||
raise RuntimeError("reset failed")
|
||||
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"_module_provider_reset_steps",
|
||||
lambda: (
|
||||
("first", lambda: calls.append("first")),
|
||||
("second", fail),
|
||||
("third", lambda: calls.append("third")),
|
||||
),
|
||||
)
|
||||
|
||||
assert modules_initializer.reset_module_providers() is False
|
||||
assert calls == ["first", "second", "third"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_modules_cancellation_does_not_skip_database_runtime_cleanup(
|
||||
monkeypatch,
|
||||
@@ -1459,11 +1505,9 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict:
|
||||
):
|
||||
instance = MagicMock()
|
||||
setattr(instance, method_name, MagicMock())
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
name,
|
||||
MagicMock(return_value=instance),
|
||||
)
|
||||
instance_type = MagicMock(return_value=instance)
|
||||
instance_type.get_existing_instance.return_value = instance
|
||||
monkeypatch.setattr(modules_initializer, name, instance_type)
|
||||
key = name.removesuffix("Helper").removesuffix("Manager").lower()
|
||||
dependencies[key] = getattr(instance, method_name)
|
||||
|
||||
@@ -1496,29 +1540,20 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict:
|
||||
|
||||
async_redis = MagicMock()
|
||||
async_redis.close = AsyncMock()
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"AsyncRedisHelper",
|
||||
MagicMock(return_value=async_redis),
|
||||
)
|
||||
async_redis_type = MagicMock(return_value=async_redis)
|
||||
async_redis_type.get_existing_instance.return_value = async_redis
|
||||
monkeypatch.setattr(modules_initializer, "AsyncRedisHelper", async_redis_type)
|
||||
dependencies["async_redis"] = async_redis.close
|
||||
close_database = AsyncMock()
|
||||
monkeypatch.setattr(modules_initializer, "close_database", close_database)
|
||||
dependencies["close_database"] = close_database
|
||||
reset_database_services = MagicMock()
|
||||
reset_module_providers = MagicMock(return_value=True)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"reset_database_services",
|
||||
reset_database_services,
|
||||
"reset_module_providers",
|
||||
reset_module_providers,
|
||||
)
|
||||
dependencies["reset_database_services"] = reset_database_services
|
||||
reset_configuration = MagicMock()
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"reset_configuration",
|
||||
reset_configuration,
|
||||
)
|
||||
dependencies["reset_configuration"] = reset_configuration
|
||||
dependencies["reset_module_providers"] = reset_module_providers
|
||||
return dependencies
|
||||
|
||||
|
||||
|
||||
@@ -499,13 +499,21 @@ def test_config_event_reloads_same_instance_and_tracks_selector_changes(
|
||||
assert running.events == ["create", "start", "stop", "start", "stop"]
|
||||
|
||||
|
||||
def test_shutdown_is_irreversible(module_manager_harness) -> None:
|
||||
def test_shutdown_is_irreversible(module_manager_harness, monkeypatch) -> None:
|
||||
"""shutdown 撤销全部可见实例，并拒绝通过 load_modules 再次启动。"""
|
||||
manager = module_manager_harness.manager
|
||||
_enable_sample(module_manager_harness.config_values)
|
||||
manager.load_modules()
|
||||
running = manager.get_running_module("SampleModule")
|
||||
assert running is not None
|
||||
remove_listener = Mock()
|
||||
unregister_resolver = Mock()
|
||||
monkeypatch.setattr(eventmanager, "remove_event_listener", remove_listener)
|
||||
monkeypatch.setattr(
|
||||
eventmanager,
|
||||
"unregister_handler_instance_resolver",
|
||||
unregister_resolver,
|
||||
)
|
||||
|
||||
manager.shutdown()
|
||||
|
||||
@@ -514,6 +522,11 @@ def test_shutdown_is_irreversible(module_manager_harness) -> None:
|
||||
manager.load_modules()
|
||||
assert manager.get_running_module("SampleModule") is None
|
||||
assert type(running).instances == [running]
|
||||
remove_listener.assert_called_once_with(
|
||||
EventType.ConfigChanged,
|
||||
manager.handle_config_changed,
|
||||
)
|
||||
unregister_resolver.assert_called_once_with("modules")
|
||||
|
||||
|
||||
def test_shutdown_reports_unreleased_module_owner(module_manager_harness) -> None:
|
||||
|
||||
@@ -78,6 +78,29 @@ def test_scheduler_initializer_stop_awaits_in_running_loop(monkeypatch):
|
||||
scheduler.stop.assert_not_called()
|
||||
|
||||
|
||||
def test_scheduler_initializer_retains_bindings_when_async_stop_fails(monkeypatch):
|
||||
"""Scheduler 未停止时不得撤销仓储、业务能力或 concrete class 登记。"""
|
||||
scheduler = Mock()
|
||||
scheduler.stop_async = AsyncMock(side_effect=RuntimeError("scheduler busy"))
|
||||
reset_bindings = Mock()
|
||||
monkeypatch.setattr(scheduler_initializer, "Scheduler", Mock(return_value=scheduler))
|
||||
monkeypatch.setattr(
|
||||
scheduler_initializer,
|
||||
"reset_scheduler_bindings",
|
||||
reset_bindings,
|
||||
)
|
||||
|
||||
async def scenario() -> None:
|
||||
"""等待异步关闭失败并验证 reset 未被调用。"""
|
||||
result = scheduler_initializer.stop_scheduler()
|
||||
assert result is not None
|
||||
with pytest.raises(RuntimeError, match="scheduler busy"):
|
||||
await result
|
||||
|
||||
asyncio.run(scenario())
|
||||
reset_bindings.assert_not_called()
|
||||
|
||||
|
||||
def test_clear_cache_is_manual_only(monkeypatch):
|
||||
"""缓存清理任务应仅手动执行，不注册到调度器自动运行。"""
|
||||
background_scheduler = _BackgroundSchedulerStub()
|
||||
|
||||
Reference in New Issue
Block a user