diff --git a/app/application/outbox.py b/app/application/outbox.py index cab7484f3..2103d5545 100644 --- a/app/application/outbox.py +++ b/app/application/outbox.py @@ -3,11 +3,14 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Mapping +from concurrent.futures import Future from dataclasses import dataclass from datetime import datetime, timedelta, timezone +from threading import Condition from types import MappingProxyType from typing import Any, Generic, Optional, Protocol, TypeVar, Union +from app.runtime.log import logger from app.schemas.types import EventType T = TypeVar("T") @@ -28,6 +31,7 @@ SUBTITLE_TRANSFER_FAILED_TOPIC = "transfer.subtitle.failed" AUDIO_TRANSFER_COMPLETED_TOPIC = "transfer.audio.completed" AUDIO_TRANSFER_FAILED_TOPIC = "transfer.audio.failed" OUTBOX_LEASE_SECONDS = 60 +OUTBOX_MAX_INFLIGHT_TOPICS = 20 DURABLE_EVENT_TOPICS: Mapping[EventType, str] = MappingProxyType({ EventType.SubscribeAdded: SUBSCRIBE_ADDED_TOPIC, @@ -113,8 +117,14 @@ class OutboxStager(Protocol): class OutboxDispatchStore(Protocol): """使用独立短事务认领和结算 durable intent 的最小端口。""" - def claim(self, now: datetime, lease_until: datetime) -> Optional[ClaimedOutboxMessage]: - """原子认领一条到期消息。""" + def claim( + self, now: datetime, lease_until: datetime, *, + topic: Optional[str] = None, exclude_topics: tuple[str, ...] = (), + ) -> Optional[ClaimedOutboxMessage]: + """原子认领到期消息，可限定 topic 或排除当前正在执行的 topic。""" + + def renew(self, message_id: int, attempt: int, lease_until: datetime) -> bool: + """仅为仍属于当前 attempt 的 processing 消息续租。""" def claim_by_event_key( self, @@ -384,65 +394,201 @@ class OutboxDispatcher: self._close = close or (lambda: None) self._failure_observer = failure_observer or (lambda _dead: None) - def dispatch_one(self) -> bool: - """处理一条到期消息；无消息返回 False，handler 失败留待重试。""" + def claim_one( + self, *, topic: Optional[str] = None, exclude_topics: tuple[str, ...] = (), + ) -> Optional[ClaimedOutboxMessage]: + """按 topic 准入约束认领消息；短事务结束后才允许执行外部效果。""" now = self._clock() - message = self._repository.claim( - now, - now + timedelta(seconds=self._lease_seconds), + return self._repository.claim( + now, now + timedelta(seconds=self._lease_seconds), + topic=topic, exclude_topics=exclude_topics, ) + + def renew(self, message: ClaimedOutboxMessage) -> None: + """恢复扫描期间续租在途效果；已完成或被接管的 attempt 不会被改写。""" + self._repository.renew( + message.message_id, message.attempt, + self._clock() + timedelta(seconds=self._lease_seconds), + ) + + def dispatch_one(self) -> bool: + """同步处理一条到期消息；无消息返回 False，失败留待持久化重试。""" + message = self.claim_one() if message is None: return False + self.dispatch_claimed(message) + return True + + def dispatch_claimed(self, message: ClaimedOutboxMessage) -> None: + """执行已认领效果，完成与重试时间均取真实结算时刻。""" try: - handler = self._handlers[message.topic] - handler(message) - if not self._repository.complete( - message.message_id, - message.attempt, - now, - ): - raise OutboxLeaseLostError("Outbox 完成凭证已失效") + self._handlers[message.topic](message) except OutboxLeaseLostError: raise except Exception as error: - dead = message.attempt >= self._max_attempts - delay = min(3600, 2 ** max(0, message.attempt - 1)) - settled = self._repository.retry( - message.message_id, - message.attempt, - next_retry_at=now + timedelta(seconds=delay), - last_error=str(error)[:4000], - dead=dead, + self.retry_claimed(message, error) + return + if not self._repository.complete(message.message_id, message.attempt, self._clock()): + raise OutboxLeaseLostError("Outbox 完成凭证已失效") + + def retry_claimed(self, message: ClaimedOutboxMessage, error: Exception) -> None: + """仅当前 owner 可写入有限退避或死信，并记录可定位的消息身份。""" + dead = message.attempt >= self._max_attempts + delay = min(3600, 2 ** min(12, max(0, message.attempt - 1))) + settled = self._repository.retry( + message.message_id, message.attempt, + next_retry_at=self._clock() + timedelta(seconds=delay), + last_error=str(error)[:4000], dead=dead, + ) + if settled: + logger.warning( + f"Outbox {message.message_id} ({message.topic}) 第 {message.attempt} 次派发失败，" + f"{'进入死信' if dead else '等待重试'}：{error}" ) - if settled: - self._failure_observer(dead) - return True - return True + self._failure_observer(dead) def close(self) -> None: """释放 dispatcher 工厂创建的短生命周期持久化资源。""" self._close() +@dataclass(slots=True) +class _OutboxFlight: + """持有一个 topic 的 dispatcher、当前 lease 和真实 worker 完成句柄。""" + + dispatcher: OutboxDispatcher + message: ClaimedOutboxMessage + future: Optional[Future[None]] = None + + +class OutboxRecovery: + """按 topic 串行、跨 topic 有界并行恢复，慢事件不占据通知的准入位置。""" + + def __init__( + self, provider: Callable[[], OutboxDispatcher], + submit: Callable[[Callable[[], None]], Future[None]], + *, max_inflight: int = OUTBOX_MAX_INFLIGHT_TOPICS, + ) -> None: + """注入短生命周期 dispatcher 工厂和宿主管理的线程提交端口。""" + self._provider = provider + self._submit = submit + self._max_inflight = max_inflight + self._lock = Condition() + self._active: dict[str, _OutboxFlight] = {} + self._stopping = False + + def dispatch(self, limit: int) -> int: + """续租在途效果并启动空闲 topic；返回接收的 topic 数，每个最多处理 limit 条。""" + with self._lock: + if self._stopping: + return 0 + for flight in self._active.values(): + flight.dispatcher.renew(flight.message) + accepted = 0 + while accepted < limit and len(self._active) < self._max_inflight: + dispatcher = self._provider() + try: + message = dispatcher.claim_one(exclude_topics=tuple(self._active)) + except BaseException: + dispatcher.close() + raise + if message is None: + dispatcher.close() + break + flight = _OutboxFlight(dispatcher, message) + self._active[message.topic] = flight + self._start(flight, limit) + accepted += 1 + return accepted + + def _start(self, flight: _OutboxFlight, limit: int) -> None: + """先登记 owner 再提交，提交失败回退 lease；所有回调完成后才释放 topic。""" + try: + flight.future = self._submit(lambda: self._run(flight, limit)) + except Exception as error: + try: + flight.dispatcher.retry_claimed(flight.message, error) + finally: + self._finish(flight) + raise + flight.future.add_done_callback(lambda future: self._done(flight, future)) + + def _run(self, flight: _OutboxFlight, limit: int) -> None: + """同一 worker 顺序消化该 topic；租约即使到期也不启动第二个恢复 owner。""" + for index in range(limit): + flight.dispatcher.dispatch_claimed(flight.message) + with self._lock: + if self._stopping or index + 1 >= limit: + break + message = flight.dispatcher.claim_one(topic=flight.message.topic) + if message is None: + break + flight.message = message + + def _done(self, flight: _OutboxFlight, future: Future[None]) -> None: + """消费后台异常，未执行即取消的任务回到持久重试，资源收尾后再释放 owner。""" + try: + if future.cancelled(): + flight.dispatcher.retry_claimed(flight.message, RuntimeError("Outbox worker 执行前已取消")) + else: + future.result() + except Exception as error: + logger.error(f"Outbox {flight.message.message_id} 后台派发未完成：{error}") + finally: + self._finish(flight) + + def _finish(self, flight: _OutboxFlight) -> None: + """关闭当前 worker 的短生命周期资源，并在锁内撤销同一 topic owner。""" + try: + flight.dispatcher.close() + finally: + with self._lock: + self._active.pop(flight.message.topic, None) + self._lock.notify_all() + + def stop(self, timeout: float = 10.0) -> bool: + """封口并有限等待真实终态；超时保留 owner，禁止重配或释放下游资源。""" + with self._lock: + self._stopping = True + return self._lock.wait_for(lambda: not self._active, timeout=max(0.0, timeout)) + + _configured_dispatcher: Optional[Callable[[], OutboxDispatcher]] = None +_configured_recovery: Optional[OutboxRecovery] = None -def configure_outbox_dispatcher(provider: Callable[[], OutboxDispatcher]) -> None: - """由组合根登记短生命周期 dispatcher 工厂。""" - global _configured_dispatcher +def configure_outbox_dispatcher( + provider: Callable[[], OutboxDispatcher], + *, submit: Optional[Callable[[Callable[[], None]], Future[None]]] = None, +) -> None: + """登记 dispatcher 工厂；生产环境注入托管 worker，未注入时保留同步入口。""" + global _configured_dispatcher, _configured_recovery + if not stop_outbox_recovery(timeout=0): + raise RuntimeError("上一 Outbox 恢复 owner 尚未停止") _configured_dispatcher = provider + _configured_recovery = OutboxRecovery(provider, submit) if submit else None + + +def stop_outbox_recovery(timeout: float = 10.0) -> bool: + """调度器停止后等待 Outbox worker，必须先于插件、模块和数据库释放。""" + return _configured_recovery is None or _configured_recovery.stop(timeout) def reset_outbox_dispatcher() -> None: - """清除当前 lifespan 的 Outbox dispatcher 工厂。""" - global _configured_dispatcher + """仅在全部在途效果收敛后撤销当前 lifespan 的工厂与恢复 owner。""" + global _configured_dispatcher, _configured_recovery + if not stop_outbox_recovery(timeout=0): + raise RuntimeError("Outbox 恢复 owner 尚未停止") _configured_dispatcher = None + _configured_recovery = None def dispatch_pending_outbox(limit: int = 20) -> int: - """恢复有限数量到期 intent，供 Scheduler 与启动补偿复用。""" + """生产恢复按 topic 提交有界批次；同步兼容入口返回实际处理的消息数。""" if _configured_dispatcher is None: raise RuntimeError("Outbox dispatcher 尚未配置") + if _configured_recovery is not None: + return _configured_recovery.dispatch(limit) dispatcher = _configured_dispatcher() try: processed = 0 diff --git a/app/db/adapters/outbox.py b/app/db/adapters/outbox.py index 609d93cec..c9542b1dd 100644 --- a/app/db/adapters/outbox.py +++ b/app/db/adapters/outbox.py @@ -36,6 +36,9 @@ def _message(model: OutboxMessage, attempt: int) -> ClaimedOutboxMessage: def _claim_query( now_text: str, event_key: Optional[str] = None, + *, + topic: Optional[str] = None, + exclude_topics: tuple[str, ...] = (), ) -> Select[tuple[OutboxMessage]]: """构造到期且 lease 可取得的候选查询。""" statement = select(OutboxMessage).where( @@ -48,6 +51,10 @@ def _claim_query( ) if event_key is not None: statement = statement.where(OutboxMessage.event_key == event_key) + if topic is not None: + statement = statement.where(OutboxMessage.topic == topic) + if exclude_topics: + statement = statement.where(OutboxMessage.topic.not_in(exclude_topics)) return statement.order_by(OutboxMessage.id).limit(1) @@ -114,9 +121,12 @@ class SqlAlchemyOutboxDispatchStore: self, now: datetime, lease_until: datetime, + *, + topic: Optional[str] = None, + exclude_topics: tuple[str, ...] = (), ) -> Optional[ClaimedOutboxMessage]: """原子认领最早一条到期消息。""" - return self._claim(now, lease_until) + return self._claim(now, lease_until, topic=topic, exclude_topics=exclude_topics) def claim_by_event_key( self, @@ -132,11 +142,16 @@ class SqlAlchemyOutboxDispatchStore: now: datetime, lease_until: datetime, event_key: Optional[str] = None, + *, + topic: Optional[str] = None, + exclude_topics: tuple[str, ...] = (), ) -> Optional[ClaimedOutboxMessage]: """在独立事务中以 compare-and-swap 取得 lease。""" now_text = _iso(now) with self._session_factory() as session: - candidate = session.execute(_claim_query(now_text, event_key)).scalars().first() + candidate = session.execute(_claim_query( + now_text, event_key, topic=topic, exclude_topics=exclude_topics, + )).scalars().first() if candidate is None: return None next_attempt = candidate.attempt + 1 @@ -147,6 +162,20 @@ class SqlAlchemyOutboxDispatchStore: session.commit() return _message(candidate, next_attempt) if claimed else None + def renew(self, message_id: int, attempt: int, lease_until: datetime) -> bool: + """只延长当前 processing owner 的租约，绝不恢复已完成或被接管的记录。""" + with self._session_factory() as session: + changed = execute_dml( + session, + update(OutboxMessage).where( + OutboxMessage.id == message_id, + OutboxMessage.status == "processing", + OutboxMessage.attempt == attempt, + ).values(lease_until=_iso(lease_until)), + ) + session.commit() + return bool(changed) + def complete( self, message_id: int, diff --git a/app/modules/telegram/module.py b/app/modules/telegram/module.py index 0fe48b7a3..04a8a0495 100644 --- a/app/modules/telegram/module.py +++ b/app/modules/telegram/module.py @@ -472,8 +472,9 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): """ 发送消息 :param message: 消息体 - :return: 成功或失败 + 未确认发送时抛错，由普通模块派发记录错误，持久派发保留重试意图。 """ + failures = [] for conf in self.get_configs().values(): if not self.check_message(message, conf.name): continue @@ -485,9 +486,10 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): logger.warn("用户没有指定 Telegram用户ID，消息无法发送") return client: Telegram = self.get_instance(conf.name) + result = None if client: if message.file_path: - client.send_file( + result = client.send_file( file_path=message.file_path, file_name=message.file_name, title=message.title, @@ -497,7 +499,7 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): parse_mode=message.parse_mode, ) elif message.voice_path: - client.send_voice( + result = client.send_voice( voice_path=message.voice_path, userid=userid, caption=message.voice_caption, @@ -511,7 +513,7 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): original_message_id = ( message.original_message_id if has_interaction_context else None ) - client.send_msg( + result = client.send_msg( title=message.title, text=message.text, image=message.image, @@ -525,6 +527,10 @@ class TelegramModule(_MessageChannelModuleBase[Telegram]): parse_mode=message.parse_mode, rich_message=self._notification_rich_message(message), ) + if not result or not result.get("success"): + failures.append(conf.name or "Telegram") + if failures: + raise RuntimeError(f"Telegram 通知发送未确认：{', '.join(failures)}") def post_medias_message( self, message: Message, medias: List[MediaInfo] diff --git a/app/modules/telegram/telegram.py b/app/modules/telegram/telegram.py index 891ee19fc..273824e16 100644 --- a/app/modules/telegram/telegram.py +++ b/app/modules/telegram/telegram.py @@ -1927,9 +1927,10 @@ class Telegram: parse_mode = kwargs.get("parse_mode") try: if image: - return self._api.send_photo( + return self.__send_photo_or_text( photo=image, caption=self._prepare_text(caption, parse_mode), + disable_web_page_preview=disable_web_page_preview, **kwargs, ) else: @@ -1945,7 +1946,22 @@ class Telegram: raise NonRetryableTelegramError( f"发送 Telegram {'图片' if image else '文本'}消息被限流：{err}" ) from err - raise RetryException(f"发送{'图片' if image else '文本'}消息失败") + raise RetryException(f"发送{'图片' if image else '文本'}消息失败：{err}") from err + + def __send_photo_or_text( + self, *, photo: Any, caption: Optional[str], + disable_web_page_preview: Optional[bool] = None, **kwargs: Any, + ) -> Any: + """海报上传失败立即降级为同一正文，保留会话、话题和按钮；429 不追加请求。""" + try: + return self._api.send_photo(photo=photo, caption=caption, **kwargs) + except Exception as error: + if self._is_rate_limit_error(error) or not caption: + raise + logger.warning(f"Telegram 海报发送失败，改为发送文本：{error}") + return self._api.send_message( + text=caption, disable_web_page_preview=disable_web_page_preview, **kwargs, + ) @retry(RetryException, logger=logger) def __send_long_plain_message( @@ -1966,11 +1982,12 @@ class Telegram: for index, chunk in enumerate(chunks): current_reply_markup = reply_markup if index == 0 else None if image and index == 0: - ret = self._api.send_photo( + ret = self.__send_photo_or_text( **kwargs, photo=image, caption=chunk, reply_markup=current_reply_markup, + disable_web_page_preview=disable_web_page_preview, ) continue msg_kwargs = dict(**kwargs) @@ -2028,7 +2045,7 @@ class Telegram: ) elif item.content_type == ContentTypes.PHOTO or (image and i == 0): - ret = self._api.send_photo( + ret = self.__send_photo_or_text( **kwargs, photo=( getattr(item, "file_name", ""), @@ -2036,6 +2053,7 @@ class Telegram: ), caption=self._telegramify_item_caption(item), reply_markup=current_reply_markup, + disable_web_page_preview=disable_web_page_preview, ) elif item.content_type == ContentTypes.FILE: diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 508c4f4f8..33acd5f85 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -706,7 +706,7 @@ async def _initialize_modules() -> HostRuntime: configure_runtime_data_providers() configure_server_services(workflow_query, runtime_dependencies.subscription) configure_workflow_execution_composition() - configure_outbox_dispatcher(build_outbox_dispatcher) + configure_outbox_dispatcher(build_outbox_dispatcher, submit=ThreadHelper().submit) configure_transfer_retry_config( lambda: TransferRetryConfig( max_failed_retries=get_runtime_setting("TRANSFER_MAX_FAILED_RETRIES"), diff --git a/app/startup/initializers/scheduler.py b/app/startup/initializers/scheduler.py index 8de96012f..be153c4ce 100644 --- a/app/startup/initializers/scheduler.py +++ b/app/startup/initializers/scheduler.py @@ -5,6 +5,7 @@ from typing import Awaitable, Callable, Optional from app.application.agenttask import AgentTaskRepository from app.application.image import WallpaperHelper +from app.application.outbox import stop_outbox_recovery from app.application.scheduling import ( get_scheduler, register_scheduler_class, @@ -16,6 +17,7 @@ from app.chain.site import SiteChain from app.chain.subscribe.facade import SubscribeChain from app.chain.transfer.facade import TransferChain from app.chain.workflow import WorkflowChain +from app.runtime.execution import run_in_threadpool_to_completion from app.scheduler.chain import SchedulerChain from app.scheduler.facade import Scheduler from app.scheduler.services import SchedulerServices @@ -103,12 +105,16 @@ def stop_scheduler() -> Optional[Awaitable[None]]: asyncio.get_running_loop() except RuntimeError: scheduler.stop() + if not stop_outbox_recovery(): + raise RuntimeError("Outbox 后台处理尚未停止") reset_scheduler_bindings() return None async def stop_and_reset() -> None: """等待异步调度器收口后清除 provider。""" await scheduler.stop_async() + if not await run_in_threadpool_to_completion(stop_outbox_recovery): + raise RuntimeError("Outbox 后台处理尚未停止") reset_scheduler_bindings() return stop_and_reset() diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 20facebc8..392b19202 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -763,7 +763,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1123 | -| 内部导入边 | 9,581 | +| 内部导入边 | 9,586 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 49（债务已清零，49 条精确 containment） | @@ -773,9 +773,12 @@ flowchart LR | Model/Oper 自动事务与自建 Session | 0 | | 组合根外 `SystemConfigOper()` | 0 | +Outbox 按 topic 有界并行恢复复用 `ThreadHelper`，生命周期关闭通过受保护的线程等待让出事件循环； +新增日志与关闭屏障共 5 条合法依赖边（含父包展开），未新增模块或 SCC。 + 外部 MCP schema 紧凑投影由纯函数模块 `app.agent.tools.projection` 拥有，仅在请求头显式 opt-in 后应用；新增 1 个模块和 1 条合法端点依赖边，未新增 SCC。`list_exposed_tools` 返回类型注解使端点类型债务减少 5 项。 -目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,581；站点 API 依赖显式导入 SiteQueryPort 与 SiteStagingPort，新增 1 条合法类型边；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 +目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。整理失败反馈由 `app.application.transfer.feedback` 集中投影；已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列；网络连通性检测按 `app.application.nettest` 归档，并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理；音乐标签准入与批次共识由 `app.chain.transfer.music` 负责，预览证据由 `app.application.transfer.projection` 只读投影；宿主依赖边现为 9,586；站点 API 依赖显式导入 SiteQueryPort 与 SiteStagingPort，新增 1 条合法类型边；订阅独立日志复用运行时上下文、系统日志端口与资源鉴权，新增 19 条合法依赖边（含父包展开），未新增模块或 SCC；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条 Schema 到 Domain 的合法依赖边（含父包展开）；种子下载限流窗口由 `app.domain.torrent` 定义，失败冷却 owner 复用同一语义，新增 1 条合法 Chain 到 Domain 的导入边；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条模块导入边；资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链，新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验，音乐交互包未新增 SCC，插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器，依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`，文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端，经 `RequestUtils` 出站，宿主不再导入 pyTelegramBotAPI，启用 Telegram 的实例也不会因该 SDK 加载 redis；该依赖暂时保留给插件使用。Web Push 协议（aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递）由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现，消息端点与 `WebPushModule` 共用，宿主不再导入 pywebpush 及其连带的 aiohttp；该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端（经 `RequestUtils` 出站）与网络适配器 `app.adapters.network.feishu` 事件长连接（websocket-client 同步收发 pbbp2 帧），宿主不再导入 lark-oapi 及其上万个生成模块，长连接也不再依赖 SDK 的模块级事件循环；插件经 `app.sdk.feishu` 复用同一长连接，lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放，默认关闭，未开放时不生成文档。当前 `app.startup.lifecycle` 为 587、`app.factory` 为 600、`app.main` 为 602。专辑订阅入库对账新增 `app.chain.subscribe.music` 一个职责 owner；按注入的下载/整理历史 Port 读取事实，复用音频标签/CUE 读取和领域覆盖规则，依赖图增加 18 条合法边，不引入跨层数据库或 Adapter 直连。订阅业务活动新增 `app.application.subscription.activity` 及对应 DB Model、Oper、Adapter 四个模块，三入口与功能前提交对照均只增加这四个模块；性能基线仅同步结构计数，保留原有历史耗时采样和预算。音乐调用观察 `app.application.music.observation` 新增一个轻量模块，记录来源结果及请求预算；AcoustID及路径识别复用此上下文传递有界指纹候选，保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`，各启动入口增加一个模块；性能基线只同步模块数量，原有耗时预算、历史采样和生命周期资源约束保持有效。 架构专项验证分为两个 CI 投影：`Check event semantic policy` 先运行依赖、Adapter、出口和 Event 语义门禁，`Check host architecture snapshot` 再执行快照测试及一次 diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 051195390..8555ec9c8 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -100,7 +100,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1123 / 9,581 | `dependency-baseline.json` 当前快照；豆瓣短评与影评新增 `app.schemas.comment` 一个模块和 3 条合法依赖边（含父包展开），模块方法契约同步登记同步与异步入口，未新增 SCC 或跨层直连。 `dependency-baseline.json` 当前快照；站点 API 依赖显式导入 `SiteQueryPort` 与 `SiteStagingPort`，新增 1 条合法类型边；集数定位直接导入 FileItem owner，替换 1 条间接导出依赖边，模块数、边数、SCC 与边界不变；监控分发器直接导入 FileItem owner，以 1 条 Schema 依赖边替换间接导出依赖，模块数、边数、SCC 与边界不变；工作流扫描、整理与下载动作直接复用已有 Schema owner，新增 3 条合法依赖边，模块数、SCC 与边界不变；工作流媒体过滤与下载查询复用已有 Schema 元素类型，新增 2 条合法依赖边，模块数、SCC 与边界不变；TheAudioDB 相关艺术家结果复用既有领域类型，新增 2 条合法 Chain 到 Domain 类型依赖边（含父包展开），模块数、SCC 与边界不变；群晖 Chat 渠道类型改为直接导入已有 Schema owner，移除 1 条间接导出依赖边，模块数、SCC 与边界不变；系统链、转移规划、订阅查询和规则服务改为直接导入已有 Schema owner，新增 2 条 owner 依赖边并移除 4 条间接导出依赖边，模块数、SCC 与层间边界不变；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份；本次安全与可靠性修复新增 4 条合法依赖边：WebAgent 复用渠道类型、宿主历史清理复用文件系统代理、两个下载器复用纯领域种子哈希规则，未新增模块或 SCC；IP 归属地适配器复用 `app.schemas.common.JsonData` 声明原始 JSON 返回，新增 2 条合法类型依赖边（含父包展开），未新增模块、循环或越层依赖；TMDb 认证与片单读取复用模块既有异常，新增 2 条同模块依赖边；Session 类型使扫描器识别原有两处 close，发送与会话关闭路径未变；Audiences 构造参数复用既有 RequestUtils 类型，新增 3 条合法依赖边（含父包展开）；插件实例历史脏行清理复用既有 `app.schemas.plugin.validate_plugin_id` 进行 fail-closed 身份校验，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC | +| 宿主 Python 模块 / 内部依赖边 | 1123 / 9,586 | Outbox 恢复的日志与关闭屏障新增 5 条合法依赖边（含父包展开），未新增模块或 SCC。 `dependency-baseline.json` 当前快照；豆瓣短评与影评新增 `app.schemas.comment` 一个模块和 3 条合法依赖边（含父包展开），模块方法契约同步登记同步与异步入口，未新增 SCC 或跨层直连。 `dependency-baseline.json` 当前快照；站点 API 依赖显式导入 `SiteQueryPort` 与 `SiteStagingPort`，新增 1 条合法类型边；集数定位直接导入 FileItem owner，替换 1 条间接导出依赖边，模块数、边数、SCC 与边界不变；监控分发器直接导入 FileItem owner，以 1 条 Schema 依赖边替换间接导出依赖，模块数、边数、SCC 与边界不变；工作流扫描、整理与下载动作直接复用已有 Schema owner，新增 3 条合法依赖边，模块数、SCC 与边界不变；工作流媒体过滤与下载查询复用已有 Schema 元素类型，新增 2 条合法依赖边，模块数、SCC 与边界不变；TheAudioDB 相关艺术家结果复用既有领域类型，新增 2 条合法 Chain 到 Domain 类型依赖边（含父包展开），模块数、SCC 与边界不变；群晖 Chat 渠道类型改为直接导入已有 Schema owner，移除 1 条间接导出依赖边，模块数、SCC 与边界不变；系统链、转移规划、订阅查询和规则服务改为直接导入已有 Schema owner，新增 2 条 owner 依赖边并移除 4 条间接导出依赖边，模块数、SCC 与层间边界不变；Telegram Rich Message 失败降级复用 `app.schemas.exception` 的 `ImmediateException`，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。目录监控成功通知在批次完成后复用主事件循环开启静默窗口，新增 2 条合法 Chain 到 Runtime 依赖边（含父包展开），未新增模块或 SCC。成功通知聚合复用既有 owner 和领域模型，统一交由通知模板渲染；移除汇总发送端的模板上下文直连后，原聚合实现的 14 条新增依赖边减为 12 条，未新增模块或 SCC。社区勋章名单读取复用 `app.foundation.url` 标准化 GitHub 镜像地址，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC。专辑订阅入库对账新增 1 个 Chain owner 和 18 条合法依赖边（含父包展开），仍经注入 Port 读取历史并通过既有 mutation 写入；订阅业务活动新增 4 个模块和 50 条合法内部依赖边（含父包展开），Application 仅定义事实与持久化 Port，DB Adapter/Oper 实现存取，组合根注入，未增加 Application/Chain 到具体 Adapter 或数据库的直连；插件实例校验复用 `app.domain.plugin` 的物理身份规则，兼容下划线 ID，新增 2 条合法依赖边（含父包展开）；种子失败冷却复用 `app.domain.torrent` 的限流语义，新增 1 条合法导入边；系统设置合同按 `app.application.settings` 包组织，系统设置 API 端点独立按 `app.api.endpoints.settings` 归档；消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理；已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档；网络连通性检测按 `app.application.nettest` 归档；工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文，新增 5 条合法模块导入边，资源过滤动作复用同一还原再新增 6 条；LLM 独占连接关闭复用宿主线程池；歌词字形转换复用 `app.foundation.text`；订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口，将非 TMDB 影视身份投影为 TMDB 身份；本次安全与可靠性修复新增 4 条合法依赖边：WebAgent 复用渠道类型、宿主历史清理复用文件系统代理、两个下载器复用纯领域种子哈希规则，未新增模块或 SCC；IP 归属地适配器复用 `app.schemas.common.JsonData` 声明原始 JSON 返回，新增 2 条合法类型依赖边（含父包展开），未新增模块、循环或越层依赖；TMDb 认证与片单读取复用模块既有异常，新增 2 条同模块依赖边；Session 类型使扫描器识别原有两处 close，发送与会话关闭路径未变；Audiences 构造参数复用既有 RequestUtils 类型，新增 3 条合法依赖边（含父包展开）；插件实例历史脏行清理复用既有 `app.schemas.plugin.validate_plugin_id` 进行 fail-closed 身份校验，新增 2 条合法依赖边（含父包展开），未新增模块或 SCC | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 986ac56d9..bd9c1a651 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -952,6 +952,15 @@ Durable post-commit side effects have a separate boundary: - Immediate request-thread delivery and the dispatcher both claim before calling a handler. Lease acquisition is atomic, while complete/retry is fenced by the claimed attempt so an expired owner cannot settle a newer claim. +- Scheduled recovery owns at most 20 active topic workers, submitted through + the injected `ThreadHelper` port. Each worker processes a bounded batch in + sequence; a blocked transfer event cannot prevent a download notification + topic from running. Each scan renews active leases and excludes active topics + from new claims. An expired lease never releases a live in-process owner. + Scheduler shutdown seals recovery and waits for actual worker completion + before plugin, module or database teardown; a timeout preserves ownership and + fails shutdown rather than resetting bindings. Retry backoff starts when the + failed handler finishes, not when the message was claimed. - `PostCommitResult` distinguishes the committed business value from completed and pending effects. This does not provide exactly-once delivery: a process may stop after an external sink succeeds but before complete is persisted. Outbox diff --git a/docs/testing.md b/docs/testing.md index c3c5c089e..90c9a4fac 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -51,6 +51,11 @@ focused 测试验证，不要求启动产品全套测试。 目标 base 复现或其他充分证据说明，并按 `docs/rules/12-collaboration-and-distribution.md` 的已记录维护者决定处置。已预授权的同一范围无需逐次重问，缺少证据时保留未知，不伪报通过。 +持久通知恢复由 `tests/test_outbox_recovery.py` 使用临时 SQLite、真实线程及屏障验证： +整理事件阻塞时下载通知仍能完成、同类事件不会并行重复认领、在途租约续期、崩溃租约接管、 +提交失败与取消后的重试，以及关闭超时保留资源。`tests/test_telegram.py` 覆盖海报失败后的文本 +兜底、会话/话题/按钮保留、429 不追加请求，以及发送失败不能被 outbox 当作成功确认。 + ## 隔离模型（`tests/conftest.py`） 收集任何测试模块、`import app.*` **之前**，conftest 完成两件事： diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index a7d81c615..0d12cd92b 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/tests/fixtures/architecture/concurrency-baseline.json @@ -680,6 +680,12 @@ }, "target": "app.runtime.execution.run_in_threadpool_to_completion" }, + "app/startup/initializers/scheduler.py:app.runtime.execution.run_in_threadpool_to_completion": { + "owners": { + "stop_scheduler.stop_and_reset": 1 + }, + "target": "app.runtime.execution.run_in_threadpool_to_completion" + }, "app/startup/lifecycle/__init__.py:app.runtime.execution.run_in_threadpool_to_completion": { "owners": { "offload_shutdown_callback.invoke": 1, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 04e6e1ca8..6270bec0c 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -943,8 +943,8 @@ "runtime_only": true } }, - "edge_count": 9581, - "edge_sha256": "0b5b37eaea05045caac6bf672acc899c84c38ec1a60441d991d159421406fe95", + "edge_count": 9586, + "edge_sha256": "1c1cd133e201b6e67da7a81b5872875401bc7b8e61d2f51a459e6e0925dca679", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3880,6 +3880,8 @@ "app.application.notification -> app.schemas", "app.application.notification -> app.schemas.system", "app.application.notification -> app.schemas.types", + "app.application.outbox -> app.runtime", + "app.application.outbox -> app.runtime.log", "app.application.outbox -> app.schemas", "app.application.outbox -> app.schemas.types", "app.application.plugin.admission -> app.application", @@ -10241,6 +10243,7 @@ "app.startup.initializers.scheduler -> app.application", "app.startup.initializers.scheduler -> app.application.agenttask", "app.startup.initializers.scheduler -> app.application.image", + "app.startup.initializers.scheduler -> app.application.outbox", "app.startup.initializers.scheduler -> app.application.scheduling", "app.startup.initializers.scheduler -> app.chain", "app.startup.initializers.scheduler -> app.chain.mediaserver", @@ -10251,6 +10254,8 @@ "app.startup.initializers.scheduler -> app.chain.transfer", "app.startup.initializers.scheduler -> app.chain.transfer.facade", "app.startup.initializers.scheduler -> app.chain.workflow", + "app.startup.initializers.scheduler -> app.runtime", + "app.startup.initializers.scheduler -> app.runtime.execution", "app.startup.initializers.scheduler -> app.scheduler", "app.startup.initializers.scheduler -> app.scheduler.chain", "app.startup.initializers.scheduler -> app.scheduler.facade", diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index 18dfed5a6..2a994b955 100644 --- a/tests/test_architecture_dependencies.py +++ b/tests/test_architecture_dependencies.py @@ -996,7 +996,7 @@ def test_modules_initializer_delegates_outbox_composition(): composition_source = (APP_ROOT / "startup" / "composition" / "outbox.py").read_text(encoding="utf-8") package_source = (APP_ROOT / "startup" / "composition" / "__init__.py").read_text(encoding="utf-8") - assert "configure_outbox_dispatcher(build_outbox_dispatcher)" in initializer_source + assert "configure_outbox_dispatcher(build_outbox_dispatcher, submit=ThreadHelper().submit)" in initializer_source for retired_name in ("_build_outbox_handlers", "_build_outbox_dispatcher"): assert retired_name not in initializer_source for detail in ( diff --git a/tests/test_outbox_recovery.py b/tests/test_outbox_recovery.py new file mode 100644 index 000000000..94322766d --- /dev/null +++ b/tests/test_outbox_recovery.py @@ -0,0 +1,312 @@ +"""持久恢复的阻塞隔离、租约所有权、关闭屏障与重试时间回归。""" + +import asyncio +from concurrent.futures import Future, ThreadPoolExecutor +from datetime import datetime, timedelta, timezone +from threading import Event +from unittest.mock import MagicMock + +import pytest +from sqlalchemy import create_engine, select +from sqlalchemy.orm import sessionmaker + +from app.application import outbox +from app.application.outbox import OutboxDispatcher, OutboxIntent, OutboxRecovery +from app.db.adapters.outbox import SqlAlchemyOutboxDispatchStore, SqlAlchemyOutboxStager +from app.db.models.outbox import OutboxMessage + + +@pytest.fixture +def database(tmp_path): + """创建支持独立线程短事务的临时文件数据库与可推进时钟。""" + engine = create_engine(f"sqlite:///{tmp_path / 'outbox.db'}") + OutboxMessage.__table__.create(engine) + factory = sessionmaker(engine, expire_on_commit=False) + now = [datetime(2026, 10, 17, tzinfo=timezone.utc)] + yield factory, SqlAlchemyOutboxDispatchStore(factory), now + engine.dispose() + + +def stage(factory, now, *topics): + """按给定顺序暂存具备独立稳定键的消息。""" + with factory() as session: + for index, topic in enumerate(topics): + SqlAlchemyOutboxStager(session).stage(OutboxIntent(f"{topic}:{index}", topic, {}), now) + session.commit() + + +def rows(factory): + """读取脱离 Session 的快照，避免测试跨线程共享会话。""" + with factory() as session: + return session.scalars(select(OutboxMessage).order_by(OutboxMessage.id)).all() + + +def test_blocked_transfer_does_not_block_notification_or_reclaim_live_owner(database): + """整理积压只占一个位置，下载通知可完成；超出租期仍不重复认领活跃效果。""" + factory, store, now = database + stage(factory, now[0], *(["transfer.completed"] * 16), "download.added.notification") + started, release, notified, closed = Event(), Event(), Event(), Event() + received = [] + + def transfer(message): + """首条整理事件停在屏障，其余消息记录实际执行顺序。""" + received.append(message.message_id) + if len(received) == 1: + started.set() + assert release.wait(5) + + def notify(_message): + """记录通知已运行，不依赖真实 Telegram。""" + notified.set() + + def provider(): + """每个恢复 worker 获取独立 dispatcher 生命周期。""" + return OutboxDispatcher(store, { + "transfer.completed": transfer, "download.added.notification": notify, + }, clock=lambda: now[0], close=closed.set) + + with ThreadPoolExecutor(max_workers=2) as pool: + recovery = OutboxRecovery(provider, pool.submit, max_inflight=2) + try: + assert recovery.dispatch(20) == 2 + assert started.wait(2) and notified.wait(2) and closed.wait(2) + for seconds in (61, 122): + now[0] += timedelta(seconds=seconds) + recovery.dispatch(20) + snapshot = rows(factory) + assert snapshot[0].attempt == 1 + assert snapshot[0].lease_until == (now[0] + timedelta(seconds=60)).isoformat() + assert all(row.status == "pending" for row in snapshot[1:16]) + assert snapshot[-1].status == "completed" + assert received == [1] + release.set() + finally: + release.set() + assert recovery.stop(5) + # stop 封口后不再领取下一条，重启后的新恢复器可继续消化剩余批次。 + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [] + + def submit(operation): + """保留真实 worker 句柄，等待批次执行完再关闭。""" + future = pool.submit(operation) + futures.append(future) + return future + + recovery = OutboxRecovery(provider, submit) + try: + recovery.dispatch(20) + for future in futures: + future.result(timeout=5) + finally: + assert recovery.stop(5) + assert received == list(range(1, 17)) + assert all(row.status == "completed" for row in rows(factory)) + + +def test_stop_retains_owner_and_rejects_reset_until_worker_finishes(database, monkeypatch): + """关闭超时既不撤销 owner 也不释放资源，禁止重新装配覆盖活跃批次。""" + factory, store, now = database + stage(factory, now[0], "blocked", "later") + started, release = Event(), Event() + close = MagicMock() + + def block(_message): + """模拟等待外部系统的已开始效果。""" + started.set() + assert release.wait(5) + + def provider(): + """绑定测试时钟及资源关闭观测。""" + return OutboxDispatcher(store, {"blocked": block}, clock=lambda: now[0], close=close) + + monkeypatch.setattr(outbox, "_configured_recovery", None) + monkeypatch.setattr(outbox, "_configured_dispatcher", None) + with ThreadPoolExecutor(max_workers=1) as pool: + outbox.configure_outbox_dispatcher(provider, submit=pool.submit) + try: + assert outbox.dispatch_pending_outbox(1) == 1 and started.wait(2) + assert not outbox.stop_outbox_recovery(0) + close.assert_not_called() + assert outbox.dispatch_pending_outbox(20) == 0 + with pytest.raises(RuntimeError, match="尚未停止"): + outbox.reset_outbox_dispatcher() + with pytest.raises(RuntimeError, match="尚未停止"): + outbox.configure_outbox_dispatcher(provider, submit=pool.submit) + finally: + release.set() + assert outbox.stop_outbox_recovery(5) + outbox.reset_outbox_dispatcher() + close.assert_called_once() + assert [row.status for row in rows(factory)] == ["completed", "pending"] + + +@pytest.mark.parametrize("cancel", [False, True]) +def test_unstarted_worker_releases_lease_and_capacity(database, cancel): + """提交拒绝或排队取消都保留重试记录、释放资源与准入容量。""" + factory, store, now = database + stage(factory, now[0], "notification") + close, handler = MagicMock(), MagicMock() + future = Future() + + def submit(_operation): + """模拟已封口线程池与尚未开始的可取消任务。""" + if not cancel: + raise RuntimeError("executor closed") + return future + + recovery = OutboxRecovery(lambda: OutboxDispatcher( + store, {"notification": handler}, clock=lambda: now[0], close=close, + ), submit, max_inflight=1) + if cancel: + assert recovery.dispatch(1) == 1 + assert recovery.dispatch(1) == 0 + assert future.cancel() + else: + with pytest.raises(RuntimeError, match="executor closed"): + recovery.dispatch(1) + assert recovery.stop(0) + close.assert_called_once() + handler.assert_not_called() + row = rows(factory)[0] + assert row.status == "pending" and row.lease_until is None + assert row.next_retry_at == (now[0] + timedelta(seconds=1)).isoformat() + + +def test_expired_owner_recovers_with_fenced_renewal(database): + """崩溃遗留 processing 能重新认领，旧 attempt 无法续租或覆盖新结果。""" + factory, store, now = database + stage(factory, now[0], "notification") + old = store.claim(now[0], now[0] + timedelta(seconds=60)) + now[0] += timedelta(seconds=61) + new = store.claim(now[0], now[0] + timedelta(seconds=60)) + assert old.event_key == new.event_key and new.attempt == old.attempt + 1 + assert not store.renew(old.message_id, old.attempt, now[0] + timedelta(hours=1)) + assert not store.complete(old.message_id, old.attempt, now[0]) + assert store.complete(new.message_id, new.attempt, now[0]) + assert not store.renew(new.message_id, new.attempt, now[0] + timedelta(hours=1)) + + +@pytest.mark.parametrize("fail", [False, True]) +def test_settlement_uses_finish_time(database, fail): + """慢处理的完成时间与退避起点均取结束时刻，避免失败立即再次重试。""" + factory, store, now = database + stage(factory, now[0], "notification") + + def handler(_message): + """模拟发送消耗九十秒，覆盖超过 lease 的同步结算时间。""" + now[0] += timedelta(seconds=90) + if fail: + raise RuntimeError("offline") + + dispatcher = OutboxDispatcher(store, {"notification": handler}, clock=lambda: now[0]) + assert dispatcher.dispatch_one() + row = rows(factory)[0] + if fail: + assert row.status == "pending" and row.last_error == "offline" + assert row.next_retry_at == (now[0] + timedelta(seconds=1)).isoformat() + assert not dispatcher.dispatch_one() + else: + assert row.completed_at == now[0].isoformat() + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_scheduler_stop_preserves_dependencies_on_recovery_timeout(monkeypatch, asynchronous): + """同步和异步关闭都先等恢复 worker，超时不能清除下游绑定。""" + from app.startup.initializers import scheduler as initializer + + scheduler, reset = MagicMock(), MagicMock() + + async def stop_async(): + """模拟调度器异步关闭完成。""" + return None + + scheduler.stop_async = stop_async + monkeypatch.setattr(initializer, "Scheduler", lambda: scheduler) + monkeypatch.setattr(initializer, "reset_scheduler_bindings", reset) + monkeypatch.setattr(initializer, "stop_outbox_recovery", lambda: False) + + async def shutdown(): + """在真实事件循环中调用关闭入口。""" + await initializer.stop_scheduler() + + with pytest.raises(RuntimeError, match="Outbox 后台处理尚未停止"): + if asynchronous: + asyncio.run(shutdown()) + else: + initializer.stop_scheduler() + reset.assert_not_called() + + +def test_scheduler_shutdown_keeps_event_loop_available(monkeypatch): + """关闭屏障等待线程期间，事件循环仍能完成该线程等待的异步 handler。""" + from app.startup.initializers import scheduler as initializer + + started, release = Event(), Event() + reset = MagicMock() + + def wait_for_worker(): + """模拟 worker 正等待事件循环上的处理返回。""" + started.set() + return release.wait(3) + + async def no_op(): + """模拟已停止的定时器。""" + + scheduler = MagicMock() + scheduler.stop_async = no_op + monkeypatch.setattr(initializer, "Scheduler", lambda: scheduler) + monkeypatch.setattr(initializer, "reset_scheduler_bindings", reset) + monkeypatch.setattr(initializer, "stop_outbox_recovery", wait_for_worker) + + async def shutdown(): + """并发运行生命周期等待和异步 handler，确认等待不会占住事件循环。""" + async def finish_handler(): + """屏障启动后由事件循环放行 worker。""" + while not started.is_set(): + await asyncio.sleep(0) + reset.assert_not_called() + release.set() + + await asyncio.gather(initializer.stop_scheduler(), finish_handler()) + + try: + asyncio.run(shutdown()) + finally: + release.set() + reset.assert_called_once() + + +def test_claim_error_closes_dispatcher(database): + """认领失败不能泄漏工厂资源或登记不存在的 topic owner。""" + _, store, now = database + store = MagicMock(wraps=store) + store.claim.side_effect = RuntimeError("database unavailable") + close, submit = MagicMock(), MagicMock() + recovery = OutboxRecovery(lambda: OutboxDispatcher( + store, {}, clock=lambda: now[0], close=close, + ), submit) + with pytest.raises(RuntimeError, match="database unavailable"): + recovery.dispatch(20) + assert recovery.stop(0) + close.assert_called_once() + submit.assert_not_called() + + +def test_worker_settlement_failure_is_observed_and_releases_owner(database, monkeypatch): + """结算数据库故障保留 processing 租约用于恢复，同时回收已退出 worker 的容量。""" + factory, store, now = database + stage(factory, now[0], "notification") + monkeypatch.setattr(store, "complete", MagicMock(side_effect=RuntimeError("database unavailable"))) + logged, closed = MagicMock(), Event() + monkeypatch.setattr(outbox.logger, "error", logged) + with ThreadPoolExecutor(max_workers=1) as pool: + recovery = OutboxRecovery(lambda: OutboxDispatcher( + store, {"notification": lambda _message: None}, clock=lambda: now[0], close=closed.set, + ), pool.submit, max_inflight=1) + try: + assert recovery.dispatch(1) == 1 and closed.wait(2) + finally: + assert recovery.stop(5) + logged.assert_called_once() + assert rows(factory)[0].status == "processing" diff --git a/tests/test_telegram.py b/tests/test_telegram.py index 04c8d6796..149634622 100644 --- a/tests/test_telegram.py +++ b/tests/test_telegram.py @@ -11,6 +11,7 @@ import pytest from app.domain.context import Context, MediaInfo, TorrentInfo from app.domain.metainfo import MetaInfo +from app.modules.telegram.botapi import TelegramApiError, TelegramNetworkError from app.modules.telegram.module import TelegramModule from app.modules.telegram.telegram import Telegram from app.schemas.message import Message @@ -353,6 +354,96 @@ def test_send_msg_markdown_escaping(telegram): assert send_kwargs["text"].startswith("*测试标题*\n") +@pytest.mark.parametrize("error", [ + TelegramApiError("sendPhoto", "invalid photo", 400), + TelegramNetworkError("sendPhoto", "timeout"), +]) +def test_photo_failure_falls_back_without_losing_notification_context(telegram, error): + """图片被拒绝或网络失败时立即发正文，保留格式、链接、话题与按钮。""" + telegram.bot.send_photo.side_effect = error + telegram.bot.send_message.return_value = {"message_id": 18, "chat": {"id": "fake_chat_id"}} + telegram._telegram_topic_id = 101 + result = telegram.send_msg( + title="下载已添加", text="正文", image="https://example.com/poster.jpg", + link="https://example.com/detail", parse_mode="HTML", disable_web_page_preview=True, + buttons=[[{"text": "查看", "url": "https://example.com"}]], + ) + assert result == {"success": True, "message_id": 18, "chat_id": "fake_chat_id"} + telegram.bot.send_photo.assert_called_once() + telegram.bot.send_message.assert_called_once() + photo = telegram.bot.send_photo.call_args.kwargs + text = telegram.bot.send_message.call_args.kwargs + assert text["text"] == photo["caption"] + assert "正文" in text["text"] and "https://example.com/detail" in text["text"] + for key in ("chat_id", "message_thread_id", "reply_markup", "parse_mode"): + assert text[key] == photo[key] + assert text["message_thread_id"] == 101 and text["disable_web_page_preview"] is True + + +def test_photo_rate_limit_does_not_send_fallback_or_retry(telegram): + """429 必须尊重冷却期，不再用文本请求加重限流。""" + telegram.bot.send_photo.side_effect = TelegramApiError( + "sendPhoto", "Too Many Requests", 429, {"retry_after": 17}, + ) + result = telegram.send_msg(title="通知", image="https://example.com/poster.jpg") + assert result["success"] is False + telegram.bot.send_photo.assert_called_once() + telegram.bot.send_message.assert_not_called() + + +@pytest.mark.parametrize("parse_mode", ["", "MarkdownV2"]) +def test_long_photo_notification_preserves_all_text_after_fallback(telegram, parse_mode): + """长正文首段图片失败时改发文本，其余分段与首段按钮保持完整。""" + telegram.bot.send_photo.side_effect = TelegramNetworkError("sendPhoto", "timeout") + body = "完整通知内容" * 800 + result = telegram.send_msg( + title="", text=body, parse_mode=parse_mode, image="https://example.com/poster.jpg", + buttons=[[{"text": "查看", "url": "https://example.com"}]], + ) + assert result["success"] is True + telegram.bot.send_photo.assert_called_once() + chunks = [call.kwargs for call in telegram.bot.send_message.call_args_list] + assert len(chunks) > 1 + assert "".join(chunk["text"] for chunk in chunks).replace("\n", "") == body + assert chunks[0]["reply_markup"] is not None + assert all(chunk["reply_markup"] is None for chunk in chunks[1:]) + + +def test_notification_failure_is_not_acknowledged_by_outbox(telegram, monkeypatch): + """图片和正文最终都失败时，模块抛错使持久消息进入重试而非完成。""" + from datetime import datetime, timezone + + from app.application.outbox import ClaimedOutboxMessage, OutboxDispatcher + + telegram.bot.send_photo.side_effect = TelegramNetworkError("sendPhoto", "timeout") + telegram.bot.send_message.side_effect = TelegramApiError("sendMessage", "forbidden", 403) + # 只移除测试中的真实退避等待，保留完整重试与模块派发路径。 + monkeypatch.setattr("app.runtime.execution.time.sleep", lambda _seconds: None) + module = TelegramModule() + monkeypatch.setattr(module, "get_configs", lambda: {"tg": SimpleNamespace(name="tg")}) + monkeypatch.setattr(module, "check_message", lambda *_args: True) + monkeypatch.setattr(module, "get_instance", lambda _name: telegram) + store = MagicMock() + store.claim.return_value = ClaimedOutboxMessage(1, "download:1", "notification", {}, 1, 1) + message = Message(title="下载完成", image="https://example.com/poster.jpg") + dispatcher = OutboxDispatcher( + store, {"notification": lambda _claimed: module.post_message(message)}, + clock=lambda: datetime(2026, 10, 17, tzinfo=timezone.utc), + ) + assert dispatcher.dispatch_one() + store.complete.assert_not_called() + store.retry.assert_called_once() + assert store.retry.call_args.kwargs["dead"] is False + assert "Telegram 通知发送未确认" in store.retry.call_args.kwargs["last_error"] + + +def test_photo_success_does_not_send_duplicate_text(telegram): + """正常图片通知仍只发一次，文本兜底仅在失败时启用。""" + assert telegram.send_msg(title="通知", image="https://example.com/poster.jpg")["success"] + telegram.bot.send_photo.assert_called_once() + telegram.bot.send_message.assert_not_called() + + def test_telegramify_current_fields_are_used_directly(): """telegramify 对象直接使用当前 MarkdownV2 字段""" from telegramify_markdown.content import ContentTrace, File, Text