Kris
1c6bed1403
本次提交包含多项核心改进： 1. 新增微信公众号插件拉取传输模式配置，完善manifest与manifest加载逻辑 2. 新增运行态状态机与幂等冲突错误体系，补充错误映射与领域错误导出 3. 优化出站与入站上下文，新增幂等键、流式中断标记等字段 4. 完善发件箱仓储与模型，新增失败条目查询、投递原子语义字段 5. 修复签名验证阶段异常捕获逻辑，防御性处理内置NotImplementedError 6. 新增出站预算释放方法，完善机器人循环预算管控 7. 优化出站管道格式阶段，新增消息长度校验逻辑 8. 完善出站打字指示器阶段，新增重复启动防御与状态同步 9. 重构出站标记失败阶段，按源状态分支处理状态转换 10. 新增入站幂等过滤阶段，修复入站路由阶段空指针问题 11. 优化出站恢复扫描器，修复状态机调用与聚合根重建逻辑 12. 完善微信公众号适配器，新增类型校验与异常包装 13. 修复数据库事务回滚逻辑，简化不必要的显式回滚操作
592 lines
22 KiB
Python
592 lines
22 KiB
Python
"""入站传输管理器。
|
||
|
||
统一管理 PullerWorker 和 StreamWorker 的生命周期，订阅账号上下线事件，
|
||
动态启动/停止账号的传输任务。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import uuid
|
||
from collections.abc import Awaitable, Callable
|
||
from typing import Any
|
||
|
||
from yuxi.channels.application.transport.base_worker import (
|
||
TransportConfig,
|
||
)
|
||
from yuxi.channels.application.transport.puller_worker import PullerWorker
|
||
from yuxi.channels.application.transport.stream_worker import StreamWorker
|
||
from yuxi.channels.contract.dtos.channel import ChannelType
|
||
from yuxi.channels.contract.dtos.health import TransportHealthSnapshot
|
||
from yuxi.channels.contract.dtos.plugin import DomainEvent, EventHandler
|
||
from yuxi.channels.contract.plugin.extension_point import EventSubscription
|
||
from yuxi.channels.contract.plugin.manifest import FailurePolicy
|
||
from yuxi.channels.contract.ports.driven import (
|
||
ConfigPort,
|
||
LoggerPort,
|
||
PersistencePort,
|
||
)
|
||
from yuxi.channels.core.registry import PluginRegistry
|
||
|
||
__all__ = ["TransportManager"]
|
||
|
||
# 管理循环的轮询间隔（秒）：仅用于等待取消信号，不参与业务逻辑
|
||
_MANAGER_LOOP_INTERVAL_S: float = 1.0
|
||
|
||
|
||
class TransportManager:
|
||
"""入站传输全局管理器。
|
||
|
||
单例模式管理 PullerWorker 和 StreamWorker，订阅 ChannelAccountOnline/Offline
|
||
事件，根据 transport_mode 与适配器能力动态启动/停止对应账号的传输任务
|
||
（Task 11：pull 仅 Puller、stream 仅 Stream、both 优先 Stream 降级 Puller）。
|
||
|
||
职责：
|
||
- 从 PluginRegistry 动态构建 puller/stream 适配器注册表
|
||
- 创建并管理 PullerWorker、StreamWorker 单例
|
||
- 订阅账号上下线事件，驱动 Worker 启动/停止账号任务
|
||
- 提供全局健康状态查询
|
||
- 优雅关停所有传输任务
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
plugin_registry: PluginRegistry,
|
||
persistence_port: PersistencePort,
|
||
config_port: ConfigPort,
|
||
event_bus: Any,
|
||
circuit_breaker: Any,
|
||
logger: LoggerPort,
|
||
message_deliverer: Callable[[Any], Awaitable[Any]],
|
||
transport_config: TransportConfig | None = None,
|
||
) -> None:
|
||
"""初始化 TransportManager。
|
||
|
||
参数：
|
||
plugin_registry: 插件注册表，用于动态获取渠道适配器。
|
||
persistence_port: 持久化端口，用于游标持久化等。
|
||
config_port: 配置端口。
|
||
event_bus: 事件总线，用于订阅/发布事件。
|
||
circuit_breaker: 熔断器实例。
|
||
logger: 日志端口。
|
||
message_deliverer: 入站消息投递回调。
|
||
transport_config: 传输配置，为 None 时使用默认配置。
|
||
"""
|
||
self._plugin_registry = plugin_registry
|
||
self._persistence_port = persistence_port
|
||
self._config_port = config_port
|
||
self._event_bus = event_bus
|
||
self._circuit_breaker = circuit_breaker
|
||
self._logger = logger
|
||
self._message_deliverer = message_deliverer
|
||
self._config = transport_config or TransportConfig()
|
||
|
||
self._puller_registry: dict[ChannelType, Any] = {}
|
||
self._stream_connector_registry: dict[ChannelType, Any] = {}
|
||
self._puller_worker: PullerWorker | None = None
|
||
self._stream_worker: StreamWorker | None = None
|
||
self._manager_task: asyncio.Task[None] | None = None
|
||
self._running: bool = False
|
||
self._plugin_id = "transport-manager"
|
||
|
||
async def start(self) -> None:
|
||
"""启动传输管理器。
|
||
|
||
- 构建 puller_registry 和 stream_connector_registry
|
||
- 创建 PullerWorker 和 StreamWorker 实例
|
||
- 启动 Worker
|
||
- 订阅账号上下线事件
|
||
- 启动管理循环任务
|
||
"""
|
||
if self._running:
|
||
return
|
||
|
||
trace_id = str(uuid.uuid4())
|
||
|
||
puller_registry: dict[ChannelType, Any] = {}
|
||
stream_connector_registry: dict[ChannelType, Any] = {}
|
||
|
||
for channel_type, plugin_da in self._plugin_registry.listPluginAdapters():
|
||
if plugin_da.puller_adapters:
|
||
puller_registry[channel_type] = plugin_da.puller_adapters[0]
|
||
if plugin_da.stream_connector_adapters:
|
||
stream_connector_registry[channel_type] = plugin_da.stream_connector_adapters[0]
|
||
|
||
self._puller_registry = puller_registry
|
||
self._stream_connector_registry = stream_connector_registry
|
||
|
||
self._puller_worker = PullerWorker(
|
||
message_deliverer=self._message_deliverer,
|
||
logger=self._logger,
|
||
config_port=self._config_port,
|
||
event_publisher=self._event_bus,
|
||
circuit_breaker=self._circuit_breaker,
|
||
persistence_port=self._persistence_port,
|
||
config=self._config,
|
||
)
|
||
|
||
self._stream_worker = StreamWorker(
|
||
message_deliverer=self._message_deliverer,
|
||
logger=self._logger,
|
||
config_port=self._config_port,
|
||
event_publisher=self._event_bus,
|
||
circuit_breaker=self._circuit_breaker,
|
||
config=self._config,
|
||
)
|
||
|
||
await self._puller_worker.start()
|
||
await self._stream_worker.start()
|
||
|
||
self._registerEventHandlers()
|
||
|
||
self._running = True
|
||
self._manager_task = asyncio.create_task(
|
||
self._manager_loop(),
|
||
name="transport-manager-loop",
|
||
)
|
||
|
||
await self._logger.info(
|
||
"transport manager started",
|
||
trace_id=trace_id,
|
||
puller_channels=[ct for ct in puller_registry.keys()],
|
||
stream_channels=[ct for ct in stream_connector_registry.keys()],
|
||
)
|
||
|
||
async def stop(self, timeout: float = 5.0) -> None:
|
||
"""停止传输管理器。
|
||
|
||
- 取消事件订阅
|
||
- 取消管理循环任务
|
||
- 停止所有 Worker
|
||
- 等待所有任务完成
|
||
|
||
参数：
|
||
timeout: 等待 Worker 停止的超时时间（秒）。
|
||
"""
|
||
if not self._running:
|
||
return
|
||
|
||
trace_id = str(uuid.uuid4())
|
||
self._running = False
|
||
|
||
self._unregisterEventHandlers()
|
||
|
||
if self._manager_task is not None and not self._manager_task.done():
|
||
self._manager_task.cancel()
|
||
try:
|
||
await self._manager_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
self._manager_task = None
|
||
|
||
if self._puller_worker is not None:
|
||
await self._puller_worker.stop(timeout)
|
||
self._puller_worker = None
|
||
|
||
if self._stream_worker is not None:
|
||
await self._stream_worker.stop(timeout)
|
||
self._stream_worker = None
|
||
|
||
await self._logger.info(
|
||
"transport manager stopped",
|
||
trace_id=trace_id,
|
||
)
|
||
|
||
def getHealth(self) -> TransportHealthSnapshot:
|
||
"""返回传输引擎健康状态（同步，向后兼容）。
|
||
|
||
返回：
|
||
``TransportHealthSnapshot``，包含 running 状态和各 Worker 健康
|
||
子状态（``WorkerHealthSnapshot``）。
|
||
"""
|
||
return TransportHealthSnapshot(
|
||
running=self._running,
|
||
puller=self._puller_worker.getHealth() if self._puller_worker else None,
|
||
stream=self._stream_worker.getHealth() if self._stream_worker else None,
|
||
)
|
||
|
||
async def getTransportHealth(self) -> TransportHealthSnapshot:
|
||
"""返回传输引擎健康状态（实现 TransportHealthPort，FR-18）。
|
||
|
||
供 ``HealthAggregator`` 通过 ``TransportHealthPort`` 聚合，暴露每个
|
||
账号的运行状态（running/backoff/stopped/error）。实现幂等、无副作用，
|
||
不抛异常阻塞调用方。
|
||
|
||
返回：
|
||
``TransportHealthSnapshot``，结构见 ``TransportHealthPort`` 协议
|
||
文档。异常时降级返回 ``puller`` / ``stream`` 为 ``None`` 的快照。
|
||
"""
|
||
try:
|
||
return self.getHealth()
|
||
except Exception as exc:
|
||
# 健康检查不得抛异常阻塞调用方（与 HealthAggregator 超时降级策略一致）
|
||
await self._logger.warn(
|
||
"transport health query failed, returning empty state",
|
||
error=str(exc),
|
||
)
|
||
return TransportHealthSnapshot(
|
||
running=self._running,
|
||
puller=None,
|
||
stream=None,
|
||
)
|
||
|
||
async def _manager_loop(self) -> None:
|
||
"""管理循环，等待取消信号。"""
|
||
try:
|
||
while self._running:
|
||
await asyncio.sleep(_MANAGER_LOOP_INTERVAL_S)
|
||
except asyncio.CancelledError:
|
||
pass
|
||
|
||
def _registerEventHandlers(self) -> None:
|
||
"""注册事件处理器。"""
|
||
self._event_bus.register(
|
||
self._plugin_id,
|
||
EventSubscription(
|
||
event_type="ChannelAccountOnline",
|
||
handler=_AccountEventHandler(self._on_account_online),
|
||
priority=100,
|
||
failure_policy=FailurePolicy.DEGRADE,
|
||
),
|
||
)
|
||
self._event_bus.register(
|
||
self._plugin_id,
|
||
EventSubscription(
|
||
event_type="ChannelAccountOffline",
|
||
handler=_AccountEventHandler(self._on_account_offline),
|
||
priority=100,
|
||
failure_policy=FailurePolicy.DEGRADE,
|
||
),
|
||
)
|
||
self._event_bus.register(
|
||
self._plugin_id,
|
||
EventSubscription(
|
||
event_type="ChannelDegraded",
|
||
handler=_AccountEventHandler(self._on_channel_degraded),
|
||
priority=100,
|
||
failure_policy=FailurePolicy.DEGRADE,
|
||
),
|
||
)
|
||
self._event_bus.register(
|
||
self._plugin_id,
|
||
EventSubscription(
|
||
event_type="ChannelRecovered",
|
||
handler=_AccountEventHandler(self._on_channel_recovered),
|
||
priority=100,
|
||
failure_policy=FailurePolicy.DEGRADE,
|
||
),
|
||
)
|
||
self._event_bus.register(
|
||
self._plugin_id,
|
||
EventSubscription(
|
||
event_type="TransportErrorOccurred",
|
||
handler=_AccountEventHandler(self._on_transport_error),
|
||
priority=100,
|
||
failure_policy=FailurePolicy.DEGRADE,
|
||
),
|
||
)
|
||
# 配置热更新订阅（FR-04）：transport.* 配置变更时应用 hot/restart 策略
|
||
for config_event in ("ConfigChanged", "ConfigRollback"):
|
||
self._event_bus.register(
|
||
self._plugin_id,
|
||
EventSubscription(
|
||
event_type=config_event,
|
||
handler=_AccountEventHandler(self._on_config_changed),
|
||
priority=100,
|
||
failure_policy=FailurePolicy.DEGRADE,
|
||
),
|
||
)
|
||
|
||
def _unregisterEventHandlers(self) -> None:
|
||
"""取消事件订阅。"""
|
||
self._event_bus.unregister(self._plugin_id)
|
||
|
||
async def _on_account_online(self, event: DomainEvent) -> None:
|
||
"""处理账号上线事件。
|
||
|
||
从事件 payload 获取 channel_type 和 account_id，根据 transport_mode
|
||
与适配器能力选择启动 Puller 或 Stream 任务（Task 11.1 + 11.2）：
|
||
- ``pull``: 仅启动 Puller
|
||
- ``stream``: 仅启动 Stream
|
||
- ``both``: 优先 Stream，Puller 作为降级（Stream 适配器不可用时启动）
|
||
|
||
参数：
|
||
event: ChannelAccountOnline 事件。
|
||
"""
|
||
trace_id = event.trace_id or str(uuid.uuid4())
|
||
channel_type_raw = event.payload.get("channel_type")
|
||
account_id = event.payload.get("account_id")
|
||
if channel_type_raw is None or account_id is None:
|
||
await self._logger.warn(
|
||
"account online event missing required fields, skip",
|
||
trace_id=trace_id,
|
||
channel_type=channel_type_raw,
|
||
account_id=account_id,
|
||
)
|
||
return
|
||
channel_type = ChannelType(channel_type_raw)
|
||
|
||
transport_mode = await self._resolveTransportMode(channel_type, account_id)
|
||
puller_adapter = self._puller_registry.get(channel_type)
|
||
stream_adapter = self._stream_connector_registry.get(channel_type)
|
||
|
||
await self._logger.info(
|
||
"received account online event, starting transport",
|
||
trace_id=trace_id,
|
||
channel_type=channel_type,
|
||
account_id=account_id,
|
||
transport_mode=transport_mode,
|
||
has_puller_adapter=puller_adapter is not None,
|
||
has_stream_adapter=stream_adapter is not None,
|
||
)
|
||
|
||
if transport_mode == "pull":
|
||
if puller_adapter is not None and self._puller_worker is not None:
|
||
await self._puller_worker.start_account(channel_type, account_id, puller_adapter)
|
||
elif transport_mode == "stream":
|
||
if stream_adapter is not None and self._stream_worker is not None:
|
||
await self._stream_worker.start_account(channel_type, account_id, stream_adapter)
|
||
else:
|
||
# Task 11.2: both 模式优先 Stream，Puller 作为降级。
|
||
# Stream 适配器可用时仅启动 Stream（Stream 健康时不 poll）；
|
||
# Stream 适配器不可用时降级启动 Puller。
|
||
if stream_adapter is not None and self._stream_worker is not None:
|
||
await self._stream_worker.start_account(channel_type, account_id, stream_adapter)
|
||
elif puller_adapter is not None and self._puller_worker is not None:
|
||
await self._puller_worker.start_account(channel_type, account_id, puller_adapter)
|
||
|
||
async def _resolveTransportMode(
|
||
self,
|
||
channel_type: ChannelType,
|
||
account_id: str,
|
||
) -> str:
|
||
"""解析账号的传输模式（Task 11.1）。
|
||
|
||
优先使用账号级 ``transport_mode``（``ChannelAccount.transport_mode``），
|
||
账号不存在或读取失败时回退到 manifest 声明的渠道级 ``transport_mode``，
|
||
均不可用时回退到 ``both``。
|
||
|
||
参数：
|
||
channel_type: 渠道类型。
|
||
account_id: 渠道账号ID。
|
||
|
||
返回：
|
||
传输模式字符串（``pull`` / ``stream`` / ``both``）。
|
||
"""
|
||
try:
|
||
account = await self._persistence_port.getChannelAccount(channel_type, account_id)
|
||
except Exception as exc:
|
||
await self._logger.warn(
|
||
"failed to get channel account for transport_mode resolution, fallback to manifest default",
|
||
channel_type=channel_type,
|
||
account_id=account_id,
|
||
error=str(exc),
|
||
)
|
||
account = None
|
||
|
||
if account is not None:
|
||
return account.transport_mode
|
||
|
||
plugin = self._plugin_registry.getPluginByChannelType(channel_type)
|
||
if plugin is not None:
|
||
return plugin.manifest.transport_mode
|
||
return "both"
|
||
|
||
async def _on_account_offline(self, event: DomainEvent) -> None:
|
||
"""处理账号下线事件。
|
||
|
||
停止对应账号的所有传输任务。
|
||
|
||
参数：
|
||
event: ChannelAccountOffline 事件。
|
||
"""
|
||
trace_id = event.trace_id or str(uuid.uuid4())
|
||
channel_type_raw = event.payload.get("channel_type")
|
||
account_id = event.payload.get("account_id")
|
||
if channel_type_raw is None or account_id is None:
|
||
await self._logger.warn(
|
||
"account offline event missing required fields, skip",
|
||
trace_id=trace_id,
|
||
channel_type=channel_type_raw,
|
||
account_id=account_id,
|
||
)
|
||
return
|
||
channel_type = ChannelType(channel_type_raw)
|
||
reason = event.payload.get("reason", "offline")
|
||
|
||
await self._logger.info(
|
||
"received account offline event, stopping transport",
|
||
trace_id=trace_id,
|
||
channel_type=channel_type,
|
||
account_id=account_id,
|
||
reason=reason,
|
||
)
|
||
|
||
if self._puller_worker is not None:
|
||
await self._puller_worker.stop_account(channel_type, account_id, reason)
|
||
|
||
if self._stream_worker is not None:
|
||
await self._stream_worker.stop_account(channel_type, account_id, reason)
|
||
|
||
async def _on_channel_degraded(self, event: DomainEvent) -> None:
|
||
"""处理渠道降级事件。
|
||
|
||
P0 阶段留空，P1 阶段实现降级逻辑（如暂停非关键任务、调整轮询间隔等）。
|
||
|
||
参数：
|
||
event: ChannelDegraded 事件。
|
||
"""
|
||
pass
|
||
|
||
async def _on_channel_recovered(self, event: DomainEvent) -> None:
|
||
"""处理渠道恢复事件。
|
||
|
||
P0 阶段留空，P1 阶段实现恢复逻辑。
|
||
|
||
参数：
|
||
event: ChannelRecovered 事件。
|
||
"""
|
||
pass
|
||
|
||
async def _on_transport_error(self, event: DomainEvent) -> None:
|
||
"""处理传输错误事件。
|
||
|
||
P0 阶段仅记录日志，P1 阶段可用于监控告警。
|
||
|
||
参数：
|
||
event: TransportErrorOccurred 事件。
|
||
"""
|
||
await self._logger.warn(
|
||
"transport error occurred",
|
||
trace_id=event.trace_id,
|
||
channel_type=event.payload.get("channel_type"),
|
||
account_id=event.payload.get("account_id"),
|
||
error_category=event.payload.get("error_category"),
|
||
error_code=event.payload.get("error_code"),
|
||
)
|
||
|
||
async def _on_config_changed(self, event: DomainEvent) -> None:
|
||
"""处理配置变更事件（FR-04 配置热更新）。
|
||
|
||
监听 ``transport.*`` 前缀的全局配置变更，按 hot/restart 模式应用：
|
||
- hot 模式（``stall_timeout_ms`` / ``backoff_schedule`` /
|
||
``backoff_jitter``）：重新加载 TransportConfig 并更新 Worker 配置，
|
||
当前轮次结束后生效。
|
||
- restart 模式（``max_restart_attempts`` /
|
||
``graceful_shutdown_timeout_s``）：记录警告日志，需手动重启
|
||
TransportManager 生效（P1 阶段不实现自动重启，避免账号连接中断）。
|
||
|
||
参数：
|
||
event: ConfigChanged / ConfigRollback 事件，payload 包含 key。
|
||
"""
|
||
key = event.payload.get("key", "")
|
||
if not key.startswith("transport."):
|
||
return
|
||
# 仅处理全局配置（target 为 None 或空），忽略账户级配置
|
||
target = event.payload.get("target")
|
||
if target:
|
||
return
|
||
|
||
trace_id = event.trace_id or str(uuid.uuid4())
|
||
if key in _RESTART_REQUIRED_CONFIG_KEYS:
|
||
await self._logger.warn(
|
||
"transport config change requires restart to take effect",
|
||
trace_id=trace_id,
|
||
key=key,
|
||
event_type=event.event_type,
|
||
)
|
||
return
|
||
|
||
# 仅对 hot-reloadable 键触发重新加载，忽略未知 transport.* 键
|
||
if key not in _RELOADABLE_CONFIG_KEYS:
|
||
return
|
||
|
||
# hot 模式：重新加载配置并更新 Worker
|
||
await self._logger.info(
|
||
"transport config hot reload triggered",
|
||
trace_id=trace_id,
|
||
key=key,
|
||
event_type=event.event_type,
|
||
)
|
||
await self.reloadConfig()
|
||
|
||
async def reloadConfig(self) -> None:
|
||
"""重新加载传输配置并更新 Worker（FR-04 hot 模式）。
|
||
|
||
从 ``ConfigPort`` 读取最新 ``transport.*`` 配置，更新
|
||
``TransportManager._config`` 及两个 Worker 的 ``_config``。
|
||
Worker 在下一轮循环（退避等待、轮询间隔、心跳间隔）时自动应用新配置。
|
||
|
||
幂等：多次调用安全，仅更新配置不重启任务。
|
||
"""
|
||
trace_id = str(uuid.uuid4())
|
||
try:
|
||
new_config = await self._loadTransportConfig()
|
||
except Exception as exc:
|
||
await self._logger.error(
|
||
"transport config reload failed, keeping old config",
|
||
trace_id=trace_id,
|
||
error=str(exc),
|
||
)
|
||
return
|
||
|
||
self._config = new_config
|
||
if self._puller_worker is not None:
|
||
self._puller_worker._config = new_config
|
||
if self._stream_worker is not None:
|
||
self._stream_worker._config = new_config
|
||
|
||
await self._logger.info(
|
||
"transport config reloaded",
|
||
trace_id=trace_id,
|
||
stall_timeout_ms=new_config.stall_timeout_ms,
|
||
backoff_schedule=list(new_config.backoff_schedule),
|
||
backoff_jitter=new_config.backoff_jitter,
|
||
)
|
||
|
||
async def _loadTransportConfig(self) -> TransportConfig:
|
||
"""从 ConfigPort 加载 TransportConfig。
|
||
|
||
复用 BaseTransportWorker._loadConfig 的读取逻辑，集中配置加载
|
||
避免逻辑重复。
|
||
"""
|
||
# 委托任一 Worker 的 _loadConfig（两者配置相同），无 Worker 时
|
||
# 直接构造默认配置
|
||
if self._puller_worker is not None:
|
||
return await self._puller_worker._loadConfig()
|
||
if self._stream_worker is not None:
|
||
return await self._stream_worker._loadConfig()
|
||
return TransportConfig()
|
||
|
||
|
||
# transport.* 配置键的热更新模式分类（FR-04）
|
||
_RELOADABLE_CONFIG_KEYS: frozenset[str] = frozenset(
|
||
{
|
||
"transport.stall_timeout_ms",
|
||
"transport.backoff_schedule",
|
||
"transport.backoff_jitter",
|
||
"transport.stream_reconnect_backoff_ms",
|
||
"transport.stream_reconnect_max_backoff_ms",
|
||
}
|
||
)
|
||
_RESTART_REQUIRED_CONFIG_KEYS: frozenset[str] = frozenset(
|
||
{
|
||
"transport.max_restart_attempts",
|
||
"transport.graceful_shutdown_timeout_s",
|
||
}
|
||
)
|
||
|
||
|
||
class _AccountEventHandler(EventHandler):
|
||
"""账号事件处理器适配器。
|
||
|
||
将 EventHandler 接口适配到 TransportManager 的异步处理方法。
|
||
"""
|
||
|
||
def __init__(self, handler: Callable[[DomainEvent], Awaitable[None]]) -> None:
|
||
self._handler = handler
|
||
|
||
async def handle(self, event: DomainEvent) -> None:
|
||
await self._handler(event)
|