diff --git a/backend/package/yuxi/channel/runtime/__init__.py b/backend/package/yuxi/channel/runtime/__init__.py new file mode 100644 index 00000000..8b137891 --- /dev/null +++ b/backend/package/yuxi/channel/runtime/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/package/yuxi/channel/runtime/backoff.py b/backend/package/yuxi/channel/runtime/backoff.py new file mode 100644 index 00000000..781e3b2f --- /dev/null +++ b/backend/package/yuxi/channel/runtime/backoff.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +import asyncio +import logging +import random +import time +from dataclasses import dataclass +from typing import Any +from collections.abc import Awaitable, Callable + +logger = logging.getLogger(__name__) + +BackoffFn = Callable[[int], float] + + +@dataclass +class BackoffConfig: + base_delay: float = 5.0 + max_delay: float = 300.0 + exponent: float = 2.0 + jitter: bool = True + jitter_factor: float = 0.1 + max_retries: int = 10 + + def compute_delay(self, attempt: int) -> float: + delay = self.base_delay * (self.exponent**attempt) + delay = min(delay, self.max_delay) + if self.jitter: + jitter_amount = delay * self.jitter_factor + delay += random.uniform(-jitter_amount, jitter_amount) + delay = max(delay, 0.01) + return delay + + +class ErrorBackoff: + """错误退避控制器 — 管理重试逻辑与指数退避。 + + 使用按 key 分锁策略，不同 key 之间的退避操作互不阻塞， + 避免粗粒度单锁成为高并发场景下的全局瓶颈。 + """ + + def __init__(self, config: BackoffConfig | None = None): + self.config = config or BackoffConfig() + self._attempts: dict[str, int] = {} + self._last_error: dict[str, float] = {} + self._locks: dict[str, asyncio.Lock] = {} + self._global_lock = asyncio.Lock() + + def _get_lock(self, key: str) -> asyncio.Lock: + """获取指定 key 的独立锁，必要时惰性创建。""" + lock = self._locks.get(key) + if lock is None: + lock = asyncio.Lock() + self._locks[key] = lock + return lock + + async def execute( + self, + key: str, + operation: Callable[[], Awaitable[Any]], + *, + on_retry: Callable[[int, Exception], Awaitable[None]] | None = None, + ) -> Any: + from yuxi.channel.errors import _infer_severity + from yuxi.channel.protocols import ErrorSeverity + + lock = self._get_lock(key) + async with lock: + attempt = self._attempts.get(key, 0) + + for i in range(self.config.max_retries + 1): + try: + result = await operation() + self._attempts[key] = 0 + self._last_error.pop(key, None) + return result + except Exception as e: + self._attempts[key] = attempt + i + 1 + self._last_error[key] = time.time() + + if _infer_severity(e) == ErrorSeverity.FATAL: + logger.error("Operation '%s' failed with fatal error: %s", key, e) + raise + + if i >= self.config.max_retries: + logger.error("Operation '%s' failed after %d retries: %s", key, self.config.max_retries, e) + raise + + delay = self.config.compute_delay(attempt + i) + logger.warning( + "Operation '%s' failed (attempt %d/%d), retrying in %.2fs: %s", + key, + i + 1, + self.config.max_retries + 1, + delay, + e, + ) + + if on_retry: + await on_retry(i, e) + + await asyncio.sleep(delay) + + def get_attempts(self, key: str) -> int: + return self._attempts.get(key, 0) + + def reset(self, key: str) -> None: + self._attempts.pop(key, None) + self._last_error.pop(key, None) + self._locks.pop(key, None) + + def is_backing_off(self, key: str, cooldown_seconds: float = 300.0) -> bool: + last_error = self._last_error.get(key) + if last_error is None: + return False + return time.time() - last_error < cooldown_seconds + + def get_backoff_remaining(self, key: str, cooldown_seconds: float = 300.0) -> float: + last_error = self._last_error.get(key) + if last_error is None: + return 0.0 + remaining = cooldown_seconds - (time.time() - last_error) + return max(remaining, 0.0) diff --git a/backend/package/yuxi/channel/runtime/boot.py b/backend/package/yuxi/channel/runtime/boot.py new file mode 100644 index 00000000..7a9f8b4d --- /dev/null +++ b/backend/package/yuxi/channel/runtime/boot.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Awaitable, Callable + +from .startup import StartupTaskRunner + +logger = logging.getLogger(__name__) + +StartupTask = Callable[[], Awaitable[Any]] + + +class BootManager: + """启动管理器 — 协调系统启动阶段的任务执行顺序。""" + + def __init__(self): + self._phases: dict[str, list[tuple[int, StartupTask]]] = { + "early": [], + "config": [], + "plugins": [], + "channels": [], + "services": [], + "late": [], + } + self._runner = StartupTaskRunner() + + def register(self, phase: str, task: StartupTask, *, priority: int = 0) -> None: + """注册一个启动任务到指定阶段。 + + priority 值越大优先级越高，同阶段内优先执行。 + """ + if phase not in self._phases: + raise ValueError(f"Unknown boot phase: {phase}") + self._phases[phase].append((priority, task)) + + async def run_startup_tasks(self) -> None: + """按阶段顺序执行所有启动任务，阶段内按 priority 降序执行。""" + for phase, tasks in self._phases.items(): + if not tasks: + continue + ordered = sorted(tasks, key=lambda x: x[0], reverse=True) + logger.info("Boot phase: %s (%d tasks)", phase, len(ordered)) + for _, task in ordered: + try: + await self._runner.run(task) + except Exception: + logger.exception("Boot task failed in phase '%s'", phase) + raise + + def get_phases(self) -> dict[str, list[StartupTask]]: + return {k: [t for _, t in v] for k, v in self._phases.items()} + + def to_dict(self) -> dict[str, Any]: + return { + "phases": { + phase: [getattr(t, "__name__", str(t)) for t in tasks] + for phase, tasks in self.get_phases().items() + }, + "total_tasks": sum(len(v) for v in self._phases.values()), + } diff --git a/backend/package/yuxi/channel/runtime/manager.py b/backend/package/yuxi/channel/runtime/manager.py new file mode 100644 index 00000000..ef835704 --- /dev/null +++ b/backend/package/yuxi/channel/runtime/manager.py @@ -0,0 +1,1048 @@ +from __future__ import annotations + +import asyncio +import contextvars +import logging +import time +from collections import defaultdict +from contextlib import suppress +from dataclasses import dataclass, field +from enum import Enum, StrEnum +from typing import Any, TYPE_CHECKING +from collections.abc import Callable + +from pathlib import Path + +from yuxi.channel.context import ChannelContext +from yuxi.channel.protocols import ( + ConfigProtocol, + GatewayProtocol, + HealthProtocol, + LifecycleProtocol, + MessageProtocol, + StartupProtocol, + TransportProtocol, +) + +from .boot import BootManager +from .trace import GatewayReadyState, StartupTrace + +if TYPE_CHECKING: + from yuxi.channel.config.diff import ConfigDiff + from yuxi.channel.config.reloader import ConfigReloader + from yuxi.channel.events.bus import ChannelEventBus + +logger = logging.getLogger(__name__) + +_gateway_ctx: contextvars.ContextVar[ChannelGateway | None] = contextvars.ContextVar("channel_gateway", default=None) + + +class ChannelState(Enum): + STOPPED = "stopped" + STARTING = "starting" + RUNNING = "running" + STOPPING = "stopping" + ERROR = "error" + RESTARTING = "restarting" + + +class HealthState(StrEnum): + HEALTHY = "healthy" + DEGRADED = "degraded" + UNHEALTHY = "unhealthy" + UNKNOWN = "unknown" + MAINTENANCE = "maintenance" + + +@dataclass +class ChannelSnapshot: + channel_type: str + account_id: str + state: ChannelState + connected: bool = False + health: str = "unknown" + last_event_at: float = 0.0 + last_error: str | None = None + active_runs: int = 0 + uptime_seconds: float = 0.0 + restart_count: int = 0 + last_start_at: float = 0.0 + last_stop_at: float = 0.0 + name: str | None = None + enabled: bool | None = None + configured: bool | None = None + last_message_at: float = 0.0 + dm_policy: str | None = None + allow_from: list[str] | None = None + mode: str | None = None + last_disconnect: DisconnectInfo | None = None + + def to_dict(self) -> dict[str, Any]: + result: dict[str, Any] = { + "channel_type": self.channel_type, + "account_id": self.account_id, + "name": self.name, + "state": self.state.value, + "connected": self.connected, + "health": self.health, + "health_state": self.health, + "last_event_at": self.last_event_at, + "last_error": self.last_error, + "active_runs": self.active_runs, + "busy": self.active_runs > 0, + "uptime_seconds": self.uptime_seconds, + "restart_count": self.restart_count, + "last_start_at": self.last_start_at, + "last_stop_at": self.last_stop_at, + "last_connected_at": self.last_start_at, + "last_message_at": self.last_message_at if self.last_message_at > 0 else None, + "last_disconnect": self.last_disconnect.to_dict() if self.last_disconnect else None, + "configured": self.configured, + "enabled": self.enabled, + "dm_policy": self.dm_policy, + "allow_from": self.allow_from, + "mode": self.mode, + } + return result + + +@dataclass +class DisconnectInfo: + at: float + status: int | None = None + error: str | None = None + logged_out: bool = False + + def to_dict(self) -> dict[str, Any]: + return { + "at": self.at, + "status": self.status, + "error": self.error, + "logged_out": self.logged_out, + } + + +@dataclass +class HealthPolicy: + stale_transport_timeout_ms: int = 300_000 + startup_grace_ms: int = 10_000 + busy_activity_ms: int = 60_000 + max_reconnect_attempts: int = 10 + + def apply(self, snap: ChannelSnapshot, gateway_started_at: float) -> HealthState: + now = time.time() + + if snap.last_start_at > 0 and (now - snap.last_start_at) * 1000 < self.startup_grace_ms: + return HealthState.UNKNOWN + + if snap.state == ChannelState.STOPPED and snap.last_disconnect and snap.last_disconnect.logged_out: + return HealthState.MAINTENANCE + + if snap.state in (ChannelState.STOPPED, ChannelState.ERROR) and snap.active_runs == 0: + return HealthState.UNHEALTHY + + if snap.state in (ChannelState.STOPPED, ChannelState.ERROR) and snap.active_runs > 0: + return HealthState.DEGRADED + + if snap.state == ChannelState.RUNNING: + if snap.connected and snap.last_event_at > 0: + stale_ms = (now - snap.last_event_at) * 1000 + if stale_ms > self.stale_transport_timeout_ms: + return HealthState.DEGRADED + + if snap.last_event_at > 0 and snap.active_runs > 0: + busy_ms = (now - snap.last_event_at) * 1000 + if busy_ms > self.busy_activity_ms: + return HealthState.DEGRADED + + return HealthState.HEALTHY + + if snap.state == ChannelState.STARTING: + return HealthState.UNKNOWN + + return HealthState.UNKNOWN + + +@dataclass +class HealthSummary: + total: int = 0 + running: int = 0 + stopped: int = 0 + degraded: int = 0 + unhealthy: int = 0 + unknown: int = 0 + connected: int = 0 + total_active_runs: int = 0 + + +@dataclass +class GatewayHealthReport: + status: str = "unknown" + summary: HealthSummary = field(default_factory=HealthSummary) + channels: dict[str, dict[str, Any]] = field(default_factory=dict) + + +@dataclass +class GatewayState: + channels: dict[str, ChannelSnapshot] = field(default_factory=dict) + global_config: dict[str, Any] = field(default_factory=dict) + route_bindings: dict[str, str] = field(default_factory=dict) + running: bool = False + started_at: float = 0.0 + last_health_check: float = 0.0 + + +class ChannelGateway(GatewayProtocol): + """渠道网关 — 统一管理所有渠道实例的生命周期、健康检查与消息路由。""" + + def __init__( + self, + global_config: dict[str, Any] | None = None, + event_bus: ChannelEventBus | None = None, + state_dir: str | Path | None = None, + config_path: str | Path | None = None, + ): + self.global_config = global_config or {} + self._event_bus = event_bus + self.state_dir = Path(state_dir) if state_dir else None + self.config_path = Path(config_path) if config_path else None + self._channels: dict[str, Any] = {} + self._states: dict[str, ChannelState] = {} + self._snapshots: dict[str, ChannelSnapshot] = {} + self._tasks: dict[str, asyncio.Task] = {} + self._health_checks: dict[str, asyncio.Task] = {} + self._message_handlers: list[Callable] = [] + self._event_handlers: list[Callable] = [] + self._running = False + self._started_at = 0.0 + self._lock = asyncio.Lock() + self._boot_manager = BootManager() + self._startup_trace = StartupTrace() + self._restart_counts: dict[str, int] = defaultdict(int) + self._max_restarts_per_hour = self.global_config.get("gateway", {}).get("channel_max_restarts_per_hour", 10) + self._restart_windows: dict[str, list[float]] = defaultdict(list) + self._manually_stopped: set[str] = set() + + from .backoff import BackoffConfig, ErrorBackoff + + self._error_backoff = ErrorBackoff(BackoffConfig(base_delay=5.0, max_delay=300.0, max_retries=10)) + + self._reloader: ConfigReloader | None = None + + self._register_boot_phases() + + # ── GatewayProtocol ───────────────────────────────────── + + async def start(self) -> None: + async with self._lock: + if self._running: + logger.warning("Gateway already running") + return + self._running = True + self._started_at = time.time() + logger.info("ChannelGateway starting...") + + if self.state_dir or self.config_path: + from yuxi.channel.security.fix import fix_security + + results = await fix_security( + config=self.global_config, + state_dir=self.state_dir, + config_path=self.config_path, + ) + for r in results: + if r.status == "error": + logger.warning("Security fix error during startup: %s", r.errors) + + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + max_concurrent = self.global_config.get("gateway", {}).get("channel_start_concurrency", 10) + semaphore = asyncio.Semaphore(max_concurrent) + + async def _start_one(plugin: Any, account_id: str) -> None: + async with semaphore: + async def _do_start(): + await self.start_channel(plugin.id, account_id, self.global_config) + + result = await _runner.run( + _do_start, + name=f"{plugin.id}:{account_id}", + timeout=30.0, + ) + if not result.success: + logger.warning("Channel %s:%s startup failed: %s", plugin.id, account_id, result.error) + self._startup_trace.add_phase( + f"channel:{plugin.id}:{account_id}", + result.duration_ms, + success=False, + error=result.error, + ) + + from .startup import StartupTaskRunner + _runner = StartupTaskRunner(default_timeout=30.0, max_retries=1) + + plugins = ChannelPluginRegistry.all() + tasks: list[asyncio.Task] = [] + + for plugin in plugins: + if not isinstance(plugin, ConfigProtocol): + continue + try: + account_ids = await plugin.list_account_ids() + except Exception: + logger.exception("Failed to list accounts for %s", getattr(plugin, "id", "unknown")) + continue + for account_id in account_ids: + try: + account = await plugin.resolve_account(account_id) + if not plugin.is_configured(account): + continue + if hasattr(plugin, "is_enabled") and not plugin.is_enabled(account, self.global_config): + logger.debug("Channel %s:%s is disabled, skipping", plugin.id, account_id) + continue + tasks.append(asyncio.create_task(_start_one(plugin, account_id))) + except Exception: + logger.exception("Failed to prepare channel %s:%s", plugin.id, account_id) + + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + for plugin in plugins: + if not isinstance(plugin, ConfigProtocol): + continue + try: + account_ids = await plugin.list_account_ids() + except Exception: + continue + for account_id in account_ids: + key = f"{plugin.id}:{account_id}" + snap = self._snapshots.get(key) + if snap is None: + continue + try: + account = await plugin.resolve_account(account_id) + snap.configured = plugin.is_configured(account) + snap.enabled = ( + plugin.is_enabled(account, self.global_config) + if hasattr(plugin, 'is_enabled') + else True + ) + snap.name = ( + plugin.describe_account(account, self.global_config).get('name') + if hasattr(plugin, 'describe_account') + else account.get('name') + ) + snap.dm_policy = account.get('dm_policy', 'pairing') + snap.allow_from = account.get('allow_from', []) + snap.mode = account.get('mode', 'socket') + except Exception: + logger.debug("Failed to backfill snapshot metadata for %s", key) + + logger.info("ChannelGateway started with %d channels", len(self._channels)) + await self._publish_event("gateway.start", channel_count=len(self._channels)) + + from yuxi.channel.config.reloader import ConfigReloader + from yuxi.channel.config.reload_plan import ReloadMode + + self._reloader = ConfigReloader( + poll_interval_seconds=self.global_config.get("gateway", {}) + .get("reload", {}) + .get("poll_interval_seconds", 5.0), + debounce_seconds=self.global_config.get("gateway", {}).get("reload", {}).get("debounce_seconds", 0.3), + reload_mode=ReloadMode(self.global_config.get("gateway", {}).get("reload", {}).get("mode", "hybrid")), + event_bus=self._event_bus, + ) + + async def _get_snapshot(): + from yuxi.channel.config.snapshot import AccountConfigEntry, ConfigSnapshot, hash_config_value + + accounts: list[AccountConfigEntry] = [] + for plugin_item in ChannelPluginRegistry.all(): + if not isinstance(plugin_item, ConfigProtocol): + continue + try: + account_ids = await plugin_item.list_account_ids() + except Exception: + logger.exception( + "ConfigReloader: list_account_ids failed for %s", getattr(plugin_item, "id", "unknown") + ) + continue + for account_id in account_ids: + try: + account = await plugin_item.resolve_account(account_id) + configured = plugin_item.is_configured(account) + enabled = False + if hasattr(plugin_item, "is_enabled"): + enabled = plugin_item.is_enabled(account, self.global_config) + config_hash = hash_config_value(account) if isinstance(account, dict) else "" + accounts.append( + AccountConfigEntry( + channel_type=plugin_item.id, + account_id=account_id, + enabled=enabled, + configured=configured, + config_hash=config_hash, + ) + ) + except Exception: + logger.exception("ConfigReloader: resolve_account failed for %s:%s", plugin_item.id, account_id) + + return ConfigSnapshot( + version=f"{int(time.time())}-{len(accounts)}", + accounts=accounts, + route_bindings_hash="", + ) + + def _on_diff(diff: ConfigDiff): + if not diff.has_changes: + return + logger.info("ChannelGateway: applying config diff %s", diff.summary) + + for entry in diff.added: + if entry.enabled and entry.configured: + logger.info("ChannelGateway: hot-starting new channel %s:%s", entry.channel_type, entry.account_id) + asyncio.create_task(self.start_channel(entry.channel_type, entry.account_id, self.global_config)) + + for entry in diff.removed: + logger.info("ChannelGateway: stopping removed channel %s:%s", entry.channel_type, entry.account_id) + asyncio.create_task(self.stop_channel(entry.channel_type, entry.account_id)) + + for entry in diff.changed: + if not entry.enabled: + logger.info("ChannelGateway: stopping disabled channel %s:%s", entry.channel_type, entry.account_id) + asyncio.create_task(self.stop_channel(entry.channel_type, entry.account_id)) + else: + logger.info( + "ChannelGateway: restarting changed channel %s:%s", entry.channel_type, entry.account_id + ) + asyncio.create_task(self.restart_channel(entry.channel_type, entry.account_id)) + + if diff.routes_changed: + logger.info("ChannelGateway: route bindings changed, rebuilding index") + if hasattr(self, "_reload_bindings"): + asyncio.create_task(self._reload_bindings()) + + self._reloader.on_diff(_on_diff) + await self._reloader.start(_get_snapshot) + logger.info("ChannelGateway: ConfigReloader started") + + async def stop(self) -> None: + async with self._lock: + if not self._running: + return + self._running = False + logger.info("ChannelGateway stopping...") + + if self._reloader: + await self._reloader.stop() + self._reloader = None + + for key in list(self._channels.keys()): + await self.stop_channel(*key.split(":", 1)) + + for task in list(self._tasks.values()): + if not task.done(): + task.cancel() + with suppress(asyncio.CancelledError): + await task + for task in list(self._health_checks.values()): + if not task.done(): + task.cancel() + with suppress(asyncio.CancelledError): + await task + + logger.info("ChannelGateway stopped") + await self._publish_event("gateway.stop") + + async def restart(self) -> None: + await self.stop() + await asyncio.sleep(0.5) + await self.start() + + def is_running(self) -> bool: + return self._running + + def get_state(self) -> GatewayState: + return GatewayState( + channels=dict(self._snapshots), + global_config=dict(self.global_config), + running=self._running, + started_at=self._started_at, + last_health_check=time.time(), + ) + + def get_snapshot(self, channel_type: str, account_id: str) -> ChannelSnapshot | None: + return self._snapshots.get(f"{channel_type}:{account_id}") + + def list_channels(self) -> list[str]: + return list(self._channels.keys()) + + def inc_active_runs(self, channel_type: str, account_id: str) -> None: + key = f"{channel_type}:{account_id}" + snap = self._snapshots.get(key) + if snap is not None: + snap.active_runs += 1 + + def dec_active_runs(self, channel_type: str, account_id: str) -> None: + key = f"{channel_type}:{account_id}" + snap = self._snapshots.get(key) + if snap is not None: + snap.active_runs = max(0, snap.active_runs - 1) + + def update_last_message_at(self, channel_type: str, account_id: str) -> None: + key = f"{channel_type}:{account_id}" + snap = self._snapshots.get(key) + if snap is not None: + snap.last_message_at = time.time() + + def get_health(self) -> GatewayHealthReport: + policy = HealthPolicy( + stale_transport_timeout_ms=self.global_config.get("gateway", {}).get( + "stale_transport_timeout_ms", 300_000 + ), + startup_grace_ms=self.global_config.get("gateway", {}).get( + "startup_grace_ms", 10_000 + ), + ) + + channels_data: dict[str, dict[str, Any]] = {} + running_count = 0 + stopped_count = 0 + connected_count = 0 + total_active_runs = 0 + healthy_count = 0 + degraded_count = 0 + unhealthy_count = 0 + unknown_count = 0 + + for key, snap in self._snapshots.items(): + parts = key.split(":", 1) + if len(parts) != 2: + continue + + state_val = snap.state.value if isinstance(snap.state, ChannelState) else str(snap.state) + is_running = snap.state == ChannelState.RUNNING + health_state = policy.apply(snap, self._started_at) + + data: dict[str, Any] = { + "name": snap.name, + "enabled": snap.enabled, + "configured": snap.configured, + "running": is_running, + "connected": snap.connected, + "busy": snap.active_runs > 0, + "active_runs": snap.active_runs, + "last_run_activity_at": snap.last_event_at if snap.last_event_at > 0 else None, + "last_transport_activity_at": snap.last_event_at if snap.last_event_at > 0 else None, + "last_message_at": snap.last_message_at if snap.last_message_at > 0 else None, + "last_start_at": snap.last_start_at if snap.last_start_at > 0 else None, + "reconnect_attempts": 0, + "restart_pending": self._states.get(key) == ChannelState.RESTARTING, + "state": state_val, + "health_state": health_state.value, + "last_error": snap.last_error, + "dm_policy": snap.dm_policy, + "allow_from": snap.allow_from, + "mode": snap.mode, + } + channels_data[key] = data + + if health_state == HealthState.HEALTHY: + healthy_count += 1 + elif health_state == HealthState.DEGRADED: + degraded_count += 1 + elif health_state == HealthState.UNHEALTHY: + unhealthy_count += 1 + else: + unknown_count += 1 + + if is_running: + running_count += 1 + else: + stopped_count += 1 + if snap.connected: + connected_count += 1 + total_active_runs += snap.active_runs + + total = len(channels_data) + + summary = HealthSummary( + total=total, + running=running_count, + stopped=stopped_count, + degraded=degraded_count, + unhealthy=unhealthy_count, + unknown=unknown_count, + connected=connected_count, + total_active_runs=total_active_runs, + ) + + if unhealthy_count == 0 and total > 0: + status = HealthState.HEALTHY.value + elif degraded_count > 0: + status = HealthState.DEGRADED.value + elif running_count > 0: + status = HealthState.DEGRADED.value + else: + status = HealthState.UNHEALTHY.value + + return GatewayHealthReport(status=status, summary=summary, channels=channels_data) + + def get_all_snapshots(self) -> dict[str, ChannelSnapshot]: + return dict(self._snapshots) + + async def start_all(self) -> None: + await self.start() + + async def stop_all(self) -> None: + await self.stop() + + # ── Channel Lifecycle ─────────────────────────────────── + + async def start_channel( + self, + channel_type: str, + account_id: str, + config: dict[str, Any] | None = None, + ) -> bool: + if config is None: + config = self.global_config + key = f"{channel_type}:{account_id}" + async with self._lock: + if key in self._channels: + logger.warning("Channel %s already running", key) + return False + + current_state = self._states.get(key) + if current_state in (ChannelState.STARTING, ChannelState.RESTARTING): + logger.warning("Channel %s is already %s", key, current_state.value) + return False + + if not self._check_restart_rate(key): + logger.warning("Channel %s restart rate limit exceeded", key) + return False + + self._states[key] = ChannelState.STARTING + self._snapshots[key] = ChannelSnapshot( + channel_type=channel_type, + account_id=account_id, + state=ChannelState.STARTING, + last_start_at=time.time(), + ) + + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + plugin = ChannelPluginRegistry.get(channel_type) + if plugin is None: + logger.error("Plugin not found for channel type: %s", channel_type) + await self._apply_start_error(key, "Plugin not found") + return False + + instance_error: str | None = None + + try: + instance = plugin() + except Exception as e: + await self._apply_start_error(key, str(e)) + return False + + try: + if isinstance(instance, StartupProtocol): + await instance.on_startup(config) + + if isinstance(instance, TransportProtocol): + await instance.connect() + + async with self._lock: + if self._states.get(key) != ChannelState.STARTING: + raise RuntimeError(f"Channel {key} state changed to {self._states.get(key)} during startup") + + self._channels[key] = instance + self._states[key] = ChannelState.RUNNING + self._manually_stopped.discard(key) + if key in self._snapshots: + self._snapshots[key].state = ChannelState.RUNNING + self._snapshots[key].connected = isinstance(instance, TransportProtocol) + self._snapshots[key].health = "healthy" + + if isinstance(instance, LifecycleProtocol): + self._tasks[key] = asyncio.create_task(self._channel_loop(key, instance)) + + if isinstance(instance, HealthProtocol): + self._health_checks[key] = asyncio.create_task(self._health_check_loop(key, instance)) + + logger.info("Channel %s started", key) + await self._publish_event("channel.running", channel_type=channel_type, account_id=account_id) + return True + + except Exception as e: + logger.exception("Failed to start channel %s", key) + instance_error = str(e) + + await self._apply_start_error(key, instance_error) + await self._publish_event( + "channel.error", channel_type=channel_type, account_id=account_id, error=instance_error + ) + return False + + async def stop_channel( + self, + channel_type: str, + account_id: str, + *, + cleanup_restart_data: bool = True, + manual: bool = False, + ) -> bool: + key = f"{channel_type}:{account_id}" + if manual: + self._manually_stopped.add(key) + async with self._lock: + instance = self._channels.pop(key, None) + if instance is None: + return False + + self._states[key] = ChannelState.STOPPING + self._snapshots[key].state = ChannelState.STOPPING + + await self._publish_event("channel.stopping", channel_type=channel_type, account_id=account_id) + + try: + if isinstance(instance, TransportProtocol): + await instance.disconnect() + + if isinstance(instance, LifecycleProtocol): + await instance.on_shutdown() + + task = self._tasks.pop(key, None) + if task and not task.done(): + task.cancel() + with suppress(asyncio.CancelledError): + await task + health_task = self._health_checks.pop(key, None) + if health_task and not health_task.done(): + health_task.cancel() + with suppress(asyncio.CancelledError): + await health_task + + async with self._lock: + self._states[key] = ChannelState.STOPPED + if key in self._snapshots: + snap = self._snapshots[key] + snap.state = ChannelState.STOPPED + snap.connected = False + snap.last_stop_at = time.time() + snap.last_disconnect = DisconnectInfo( + at=snap.last_stop_at, + error=snap.last_error, + logged_out=key in self._manually_stopped, + ) + + logger.info("Channel %s stopped", key) + return True + + except Exception as e: + logger.exception("Failed to stop channel %s", key) + async with self._lock: + self._states[key] = ChannelState.ERROR + if key in self._snapshots: + snap = self._snapshots[key] + snap.state = ChannelState.ERROR + snap.last_error = str(e) + snap.last_stop_at = time.time() + snap.last_disconnect = DisconnectInfo( + at=snap.last_stop_at, + error=str(e), + logged_out=False, + ) + return False + + finally: + if cleanup_restart_data: + self._restart_counts.pop(key, None) + self._restart_windows.pop(key, None) + + def is_manually_stopped(self, channel_type: str, account_id: str) -> bool: + return f"{channel_type}:{account_id}" in self._manually_stopped + + async def restart_channel(self, channel_type: str, account_id: str) -> bool: + key = f"{channel_type}:{account_id}" + self._restart_counts[key] += 1 + self._restart_windows[key].append(time.time()) + + await self.stop_channel(channel_type, account_id, cleanup_restart_data=False) + + attempt = self._error_backoff.get_attempts(key) + delay = self._error_backoff.config.compute_delay(attempt) + logger.info("Channel %s restart attempt %d, waiting %.2fs", key, attempt + 1, delay) + await asyncio.sleep(max(delay, 0.1)) + + success = await self.start_channel(channel_type, account_id, self.global_config) + if success: + self._error_backoff.reset(key) + return success + + def _check_restart_rate(self, key: str) -> bool: + window = self._restart_windows.get(key, []) + now = time.time() + hour_ago = now - 3600 + recent = [t for t in window if t > hour_ago] + self._restart_windows[key] = recent + return len(recent) < self._max_restarts_per_hour + + async def _publish_event(self, topic: str, **kwargs) -> None: + if self._event_bus is None: + return + try: + await self._event_bus.publish(topic, **kwargs) + except Exception: + logger.exception("EventBus publish failed: topic=%s", topic) + + async def _apply_start_error(self, key: str, error: str) -> None: + async with self._lock: + if self._states.get(key) != ChannelState.STARTING: + return + self._states[key] = ChannelState.ERROR + if key in self._snapshots: + self._snapshots[key].state = ChannelState.ERROR + self._snapshots[key].last_error = error + + # ── Message & Event Routing ───────────────────────────── + + def on_message(self, handler: Callable) -> Callable: + self._message_handlers.append(handler) + return handler + + def on_event(self, handler: Callable) -> Callable: + self._event_handlers.append(handler) + return handler + + async def send_message( + self, + channel_type: str, + account_id: str, + message: dict[str, Any], + ) -> bool: + key = f"{channel_type}:{account_id}" + instance = self._channels.get(key) + if instance is None: + logger.warning("Cannot send message: channel %s not found", key) + return False + + if not isinstance(instance, MessageProtocol): + logger.warning("Channel %s does not support MessageProtocol", key) + return False + + try: + await instance.send_message(message) + return True + except Exception: + logger.exception("Failed to send message to %s", key) + return False + + async def route_message(self, context: ChannelContext) -> bool: + for handler in self._message_handlers: + try: + result = await handler(context) + if result is True: + return True + except Exception: + logger.exception("Message handler failed") + return False + + # ── Internal Loops ────────────────────────────────────── + + async def _channel_loop(self, key: str, instance: LifecycleProtocol) -> None: + try: + while self._running and self._states.get(key) == ChannelState.RUNNING: + try: + await instance.on_tick() + except Exception as e: + logger.exception("Channel loop error for %s", key) + self._snapshots[key].last_error = str(e) + await asyncio.sleep(1.0) + except asyncio.CancelledError: + pass + except Exception as e: + logger.exception("Channel loop crashed for %s", key) + async with self._lock: + self._snapshots[key].last_error = str(e) + self._states[key] = ChannelState.ERROR + self._snapshots[key].state = ChannelState.ERROR + parts = key.split(":", 1) + if len(parts) == 2: + asyncio.create_task(self.restart_channel(parts[0], parts[1])) + + async def _health_check_loop(self, key: str, instance: HealthProtocol) -> None: + interval = self.global_config.get("gateway", {}).get("channel_health_check_minutes", 5) + interval_seconds = max(interval * 60, 30) + + try: + while self._running and self._states.get(key) in ( + ChannelState.RUNNING, + ChannelState.STARTING, + ): + try: + healthy = await instance.health_check() + self._snapshots[key].health = "healthy" if healthy else "unhealthy" + if not healthy: + logger.warning("Health check failed for %s", key) + except Exception as e: + logger.exception("Health check error for %s", key) + self._snapshots[key].health = "error" + self._snapshots[key].last_error = str(e) + await asyncio.sleep(interval_seconds) + except asyncio.CancelledError: + pass + + # ── Boot & Startup ────────────────────────────────────── + + def _register_boot_phases(self) -> None: + self._boot_manager.register("early", self._preflight_check, priority=100) + self._boot_manager.register("plugins", self._registry_check, priority=100) + self._boot_manager.register("channels", self._run_bootstrap_hooks, priority=50) + self._boot_manager.register("channels", self._warmup_channels_wrapper, priority=100) + self._boot_manager.register("late", self._readiness_check, priority=100) + self._boot_manager.register("late", self._mark_running, priority=50) + + async def _preflight_check(self) -> None: + self._startup_trace.start_phase("preflight") + try: + from yuxi.storage.postgres.manager import pg_manager + async with pg_manager.get_async_session_context() as session: + from sqlalchemy import text + await session.execute(text("SELECT 1")) + self._startup_trace.complete_phase("preflight", ok=True) + except Exception as e: + self._startup_trace.complete_phase("preflight", ok=False, error=str(e)) + self._startup_trace.finalize(GatewayReadyState.FAILED) + raise + + async def _registry_check(self) -> None: + self._startup_trace.start_phase("registry") + from yuxi.channel.plugins.registry import ChannelPluginRegistry + plugins = ChannelPluginRegistry.all() + if not plugins: + self._startup_trace.complete_phase("registry", ok=False, error="No registered plugins") + self._startup_trace.finalize(GatewayReadyState.FAILED) + raise RuntimeError("No plugins registered in ChannelPluginRegistry") + self._startup_trace.complete_phase("registry", ok=True) + + async def _run_bootstrap_hooks(self) -> None: + self._startup_trace.start_phase("bootstrap") + from yuxi.channel.plugins.registry import ChannelPluginRegistry + + errors: list[str] = [] + for plugin in ChannelPluginRegistry.all(): + if not isinstance(plugin, LifecycleProtocol): + continue + try: + await plugin.run_startup_maintenance(self.global_config) + except Exception as e: + logger.exception("Bootstrap hook failed for plugin: %s", getattr(plugin, "id", "unknown")) + errors.append(str(e)) + + if errors: + self._startup_trace.complete_phase("bootstrap", ok=False, error="; ".join(errors)) + else: + self._startup_trace.complete_phase("bootstrap", ok=True) + + async def _warmup_channels_wrapper(self) -> None: + self._startup_trace.start_phase("channel-warmup") + + async def _readiness_check(self) -> None: + self._startup_trace.start_phase("readiness") + running = sum(1 for s in self._snapshots.values() if s.state == ChannelState.RUNNING) + total = len(self._snapshots) + + if total == 0: + ready_state = GatewayReadyState.NOT_READY + error = "No channels configured" + elif running == total: + ready_state = GatewayReadyState.READY + error = None + elif running > 0: + ready_state = GatewayReadyState.DEGRADED + error = None + else: + ready_state = GatewayReadyState.FAILED + error = "All channels failed to start" + + self._startup_trace.complete_phase("readiness", ok=ready_state != GatewayReadyState.FAILED, error=error) + + warmup_ok = running > 0 + if warmup_ok: + self._startup_trace.complete_phase("channel-warmup", ok=True) + else: + self._startup_trace.complete_phase("channel-warmup", ok=False, error="No channels running") + + self._startup_trace.finalize(ready_state) + + async def _mark_running(self) -> None: + self._startup_trace.start_phase("running") + await self._publish_event("gateway.ready", ready_state=self._startup_trace.ready_state.value) + self._startup_trace.complete_phase("running", ok=True) + + async def boot(self, overall_timeout: float = 120.0) -> None: + logger.info("ChannelGateway boot sequence starting...") + self._startup_trace.start() + + try: + await self._boot_manager.run_startup_tasks() + + try: + await asyncio.wait_for(self.start(), timeout=overall_timeout) + except TimeoutError: + logger.error("ChannelGateway start timed out after %.0fs", overall_timeout) + self._startup_trace.add_phase("channel-warmup", overall_timeout * 1000, success=False, + error=f"Overall startup timeout ({overall_timeout}s)") + + self._startup_trace.complete() + logger.info("ChannelGateway boot sequence completed in %.2fs", self._startup_trace.duration) + except Exception as e: + self._startup_trace.fail(str(e)) + logger.exception("ChannelGateway boot sequence failed") + raise + + def get_startup_trace(self) -> dict[str, Any]: + return self._startup_trace.to_dict() + + def get_boot_status(self) -> dict[str, Any]: + return self._boot_manager.to_dict() + + # ── Plugin Dependency Check ───────────────────────────── + + def _check_plugin_dependencies(self) -> None: + from yuxi.channel.plugins.dependency import DependencyGraph + from yuxi.channel.plugins.discovery import ChannelPluginDiscovery + + graph = DependencyGraph() + manifests = ChannelPluginDiscovery.discover_manifests("manifests/channels") + for manifest in manifests: + graph.add(manifest) + + has_cycle, cycle = graph.has_cycle() + if has_cycle: + logger.error("Plugin dependency cycle detected: %s", " -> ".join(cycle)) + raise RuntimeError(f"Plugin dependency cycle: {' -> '.join(cycle)}") + + +def get_gateway() -> ChannelGateway: + gw = _gateway_ctx.get() + if gw is None: + gw = ChannelGateway() + _gateway_ctx.set(gw) + return gw + + +def set_gateway(gw: ChannelGateway) -> None: + _gateway_ctx.set(gw) + + +def reset_gateway() -> None: + _gateway_ctx.set(None) + + +def __getattr__(name: str): + if name == "gateway": + return get_gateway() + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/backend/package/yuxi/channel/runtime/startup.py b/backend/package/yuxi/channel/runtime/startup.py new file mode 100644 index 00000000..87c3b5ac --- /dev/null +++ b/backend/package/yuxi/channel/runtime/startup.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import asyncio +import logging +import time +from dataclasses import dataclass, field +from typing import Any, Awaitable, Callable + +from yuxi.channel.runtime.backoff import BackoffConfig + +logger = logging.getLogger(__name__) + +StartupTask = Callable[[], Awaitable[Any]] + + +@dataclass +class StartupTaskResult: + name: str + success: bool + duration_ms: float + error: str | None = None + skipped: bool = False + + +_StartupBackoffConfig = BackoffConfig( + base_delay=0.5, + max_delay=5.0, + exponent=1.5, + jitter=True, + jitter_factor=0.1, + max_retries=3, +) + + +class StartupTaskRunner: + """启动任务执行器 — 支持超时、重试、并行执行与结果追踪。""" + + def __init__(self, default_timeout: float = 30.0, max_retries: int = 2, backoff_config: BackoffConfig | None = None): + self.default_timeout = default_timeout + self.max_retries = max_retries + self._backoff_config = backoff_config or _StartupBackoffConfig + self._results: list[StartupTaskResult] = [] + self._results_lock = asyncio.Lock() + + async def run( + self, + task: StartupTask, + *, + name: str | None = None, + timeout: float | None = None, + retries: int | None = None, + ) -> StartupTaskResult: + task_name = name or getattr(task, "__name__", "unknown") + timeout_val = timeout if timeout is not None else self.default_timeout + retry_count = retries if retries is not None else self.max_retries + + start = time.time() + last_error: Exception | None = None + + for attempt in range(retry_count + 1): + try: + await asyncio.wait_for(task(), timeout=timeout_val) + duration_ms = (time.time() - start) * 1000 + result = StartupTaskResult( + name=task_name, + success=True, + duration_ms=duration_ms, + ) + async with self._results_lock: + self._results.append(result) + logger.info("Startup task '%s' completed in %.2fms", task_name, duration_ms) + return result + except asyncio.TimeoutError: + last_error = asyncio.TimeoutError(f"Task '{task_name}' timed out after {timeout_val}s") + logger.warning("Startup task '%s' timed out (attempt %d/%d)", task_name, attempt + 1, retry_count + 1) + except Exception as e: + last_error = e + logger.exception("Startup task '%s' failed (attempt %d/%d)", task_name, attempt + 1, retry_count + 1) + if attempt < retry_count: + await asyncio.sleep(self._backoff_config.compute_delay(attempt)) + + duration_ms = (time.time() - start) * 1000 + result = StartupTaskResult( + name=task_name, + success=False, + duration_ms=duration_ms, + error=str(last_error) if last_error else "Unknown error", + ) + async with self._results_lock: + self._results.append(result) + return result + + async def run_parallel( + self, + tasks: list[tuple[StartupTask, str]], + *, + timeout: float | None = None, + ) -> list[StartupTaskResult]: + """并行执行多个启动任务。""" + coros = [self.run(task, name=name, timeout=timeout) for task, name in tasks] + return await asyncio.gather(*coros, return_exceptions=True) + + async def get_results(self) -> list[StartupTaskResult]: + async with self._results_lock: + return list(self._results) + + async def clear_results(self) -> None: + async with self._results_lock: + self._results.clear() diff --git a/backend/package/yuxi/channel/runtime/trace.py b/backend/package/yuxi/channel/runtime/trace.py new file mode 100644 index 00000000..5404f79e --- /dev/null +++ b/backend/package/yuxi/channel/runtime/trace.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import time as _time +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any + + +class PhaseStatus(StrEnum): + PENDING = "pending" + RUNNING = "running" + OK = "ok" + FAILED = "failed" + SKIPPED = "skipped" + + +class GatewayReadyState(StrEnum): + NOT_READY = "not_ready" + READY = "ready" + DEGRADED = "degraded" + FAILED = "failed" + + +@dataclass +class PhaseMark: + name: str + status: PhaseStatus = PhaseStatus.PENDING + started_at: float | None = None + finished_at: float | None = None + + @property + def elapsed_ms(self) -> float | None: + if self.started_at and self.finished_at: + return (self.finished_at - self.started_at) * 1000 + return None + + +@dataclass +class StartupTrace: + """启动追踪器 — 记录系统启动各阶段耗时。""" + + started_at: float = 0.0 + completed_at: float = 0.0 + failed_at: float = 0.0 + error: str | None = None + phases: list[dict[str, Any]] = field(default_factory=list) + + _phase_marks: list[PhaseMark] = field(default_factory=list, repr=False) + _active_phase: str | None = field(default=None, repr=False) + ready_state: GatewayReadyState = GatewayReadyState.NOT_READY + trace_id: str = "" + + def start(self) -> None: + self.started_at = _time.time() + self.phases.clear() + + def add_phase(self, name: str, duration_ms: float, *, success: bool = True, error: str | None = None) -> None: + self.phases.append( + { + "name": name, + "duration_ms": round(duration_ms, 2), + "success": success, + "error": error, + } + ) + + def complete(self) -> None: + self.completed_at = _time.time() + + def fail(self, error: str) -> None: + self.failed_at = _time.time() + self.error = error + + @property + def duration(self) -> float: + if self.completed_at: + return self.completed_at - self.started_at + if self.failed_at: + return self.failed_at - self.started_at + return _time.time() - self.started_at + + @property + def success(self) -> bool: + return self.completed_at > 0 and self.failed_at == 0 + + def to_dict(self) -> dict[str, Any]: + return { + "trace_id": self.trace_id, + "started_at": self.started_at, + "completed_at": self.completed_at, + "failed_at": self.failed_at, + "duration_seconds": round(self.duration, 3), + "success": self.success, + "ready_state": self.ready_state.value, + "error": self.error, + "phases": list(self.phases), + } + + def start_phase(self, name: str) -> None: + mark = PhaseMark(name=name, status=PhaseStatus.RUNNING, started_at=_time.time()) + self._upsert_mark(mark) + self._active_phase = name + + def complete_phase(self, name: str, ok: bool = True, error: str | None = None) -> None: + mark = self._find_mark(name) + if mark is None: + return + mark.status = PhaseStatus.OK if ok else PhaseStatus.FAILED + mark.finished_at = _time.time() + self._active_phase = None + duration_ms = mark.elapsed_ms or 0 + self.add_phase(name, duration_ms, success=ok, error=error) + + def skip_phase(self, name: str) -> None: + mark = self._find_mark(name) + if mark is not None: + mark.status = PhaseStatus.SKIPPED + else: + mark = PhaseMark(name=name, status=PhaseStatus.SKIPPED) + self._phase_marks.append(mark) + self._active_phase = None + + def finalize(self, ready_state: GatewayReadyState) -> None: + self.ready_state = ready_state + if not self.completed_at and not self.failed_at: + self.completed_at = _time.time() + + def summary(self) -> dict[str, Any]: + return { + "trace_id": self.trace_id, + "ready_state": self.ready_state.value, + "total_elapsed_ms": round(self.duration * 1000, 2) if self.duration else None, + "phases": [ + { + "name": p.name, + "status": p.status.value, + "elapsed_ms": round(p.elapsed_ms, 2) if p.elapsed_ms else None, + } + for p in self._phase_marks + ], + "errors": [ + e["error"] for e in self.phases if not e.get("success") and e.get("error") + ], + } + + def _upsert_mark(self, mark: PhaseMark) -> None: + existing = self._find_mark(mark.name) + if existing: + idx = self._phase_marks.index(existing) + self._phase_marks[idx] = mark + else: + self._phase_marks.append(mark) + + def _find_mark(self, name: str) -> PhaseMark | None: + for p in self._phase_marks: + if p.name == name: + return p + return None diff --git a/backend/package/yuxi/channel/runtime/trace_ctx.py b/backend/package/yuxi/channel/runtime/trace_ctx.py new file mode 100644 index 00000000..76ff8d31 --- /dev/null +++ b/backend/package/yuxi/channel/runtime/trace_ctx.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import contextvars +import logging +import uuid +from collections.abc import Generator +from contextlib import contextmanager +from typing import Any + + +_trace_id_var: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "channel_trace_id", default=None +) + + +class TraceIdFilter(logging.Filter): + """日志过滤器：自动将当前 trace_id 注入 LogRecord 的 trace_id 属性。 + + 配合 logging.Formatter 使用，在日志格式中包含 %(trace_id)s 即可。 + """ + + def filter(self, record: logging.LogRecord) -> bool: + trace_id = get_trace_id() + record.trace_id = trace_id or "-" + return True + + +def generate_trace_id() -> str: + return uuid.uuid4().hex[:16] + + +def get_trace_id() -> str | None: + return _trace_id_var.get() + + +def set_trace_id(trace_id: str | None) -> None: + _trace_id_var.set(trace_id) + + +@contextmanager +def trace_context(trace_id: str | None = None, **metadata: Any) -> Generator[str, None, None]: + token = _trace_id_var.set(trace_id or generate_trace_id()) + try: + yield _trace_id_var.get() + finally: + _trace_id_var.reset(token) + + +def install_trace_filter(logger_name: str | None = None) -> None: + root_logger = logging.getLogger(logger_name) + if not any(isinstance(f, TraceIdFilter) for f in root_logger.filters): + root_logger.addFilter(TraceIdFilter()) \ No newline at end of file