From ed2ee086de0f3baf8ed0cf95071de944c0a730e2 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Sat, 18 Jul 2026 03:43:28 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=20=E9=87=8D=E6=9E=84UI=E6=93=8D?= =?UTF-8?q?=E4=BD=9C=E8=B0=83=E5=BA=A6=EF=BC=8C=E5=BC=95=E5=85=A5=E4=BC=98?= =?UTF-8?q?=E5=85=88=E7=BA=A7=E8=B0=83=E5=BA=A6=E4=B8=8E=E7=9B=91=E6=8E=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 主要变更： 1. 新增UIActionScheduler实现优先级UI任务调度，支持CRITICAL/HIGH/NORMAL/LOW四级优先级 2. 替换原有SendQueue为UIActionScheduler，统一所有UI操作的调度逻辑 3. 为截图、登录、重启等API添加调度器封装，支持超时、限流与fast-fail机制 4. 新增Prometheus监控指标，统计UI任务执行、等待耗时、超时、限流与快速失败情况 5. 为QrCapture添加命令执行超时保护，避免X11操作卡死 6. 扩展StatusResponse与状态接口，暴露调度器指标 7. 新增完整的单元测试覆盖调度器功能 8. 为看门狗注入调度器，实现kill前队列清空与快速失败窗口 --- bridge/tests/test_ui_action_scheduler.py | 251 +++++++++++++ bridge/woc_bridge/app.py | 9 +- .../messaging/ui_action_scheduler.py | 332 ++++++++++++++++++ bridge/woc_bridge/models/base.py | 1 + bridge/woc_bridge/models/status.py | 6 + bridge/woc_bridge/routes/diagnostic.py | 71 +++- bridge/woc_bridge/routes/login.py | 52 ++- bridge/woc_bridge/routes/screenshot.py | 14 +- bridge/woc_bridge/routes/status.py | 19 +- bridge/woc_bridge/ui/flow.py | 25 +- bridge/woc_bridge/ui/metrics.py | 98 +++++- bridge/woc_bridge/ui/qr_capture.py | 19 +- bridge/woc_bridge/ui/watchdog.py | 26 +- 13 files changed, 884 insertions(+), 39 deletions(-) create mode 100644 bridge/tests/test_ui_action_scheduler.py create mode 100644 bridge/woc_bridge/messaging/ui_action_scheduler.py diff --git a/bridge/tests/test_ui_action_scheduler.py b/bridge/tests/test_ui_action_scheduler.py new file mode 100644 index 0000000..f9da070 --- /dev/null +++ b/bridge/tests/test_ui_action_scheduler.py @@ -0,0 +1,251 @@ +"""UIActionScheduler 单元测试。 + +覆盖： +- 向后兼容 enqueue +- 优先级排序（CRITICAL 优先于 NORMAL） +- drain（空队列 / 有 pending / 超时） +- 队列满抛 RATE_LIMITED +- 等待超时抛 TIMEOUT +- metrics 准确性 +- fast-fail 机制（mark_wechat_dead 后非 CRITICAL 抛 WECHAT_NOT_READY） +- CRITICAL 任务在 fast-fail 窗口内正常执行 +- CRITICAL 任务自动 delay_ms=0 +""" + +from __future__ import annotations + +import asyncio +import os +import sys +import time + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from woc_bridge.messaging.ui_action_scheduler import UIActionScheduler, Priority +from woc_bridge.models import BridgeError + +pytestmark = pytest.mark.asyncio + + +async def test_backward_compat_enqueue(): + """向后兼容 enqueue：以 NORMAL 优先级入队并返回结果。""" + scheduler = UIActionScheduler(send_delay_ms=10, max_calls_per_sec=100) + await scheduler.start() + try: + result = await scheduler.enqueue(lambda: asyncio.sleep(0.01, result="ok")) + assert result == "ok" + + metrics = scheduler.get_metrics() + assert metrics["executed_total"] == 1 + assert metrics["executed_by_priority"]["NORMAL"] == 1 + finally: + await scheduler.stop() + + +async def test_priority_ordering(): + """CRITICAL 任务应先于后入队的 NORMAL 任务开始执行。""" + scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) + await scheduler.start() + try: + start_times: list[tuple[str, str]] = [] + + async def _track(result: str, delay: float = 0.01) -> str: + start_times.append(("start", result)) + await asyncio.sleep(delay) + return result + + # 第一个 NORMAL 立即开始执行 + normal_task = asyncio.create_task( + scheduler.enqueue(lambda: _track("normal", 0.1)) + ) + await asyncio.sleep(0.02) # 等第一个开始 + + critical_task = asyncio.create_task( + scheduler.execute( + lambda: _track("critical", 0.01), priority=Priority.CRITICAL + ) + ) + normal2_task = asyncio.create_task( + scheduler.execute( + lambda: _track("normal2", 0.01), priority=Priority.NORMAL + ) + ) + + critical_result = await critical_task + normal2_result = await normal2_task + normal_result = await normal_task + + assert critical_result == "critical" + assert normal2_result == "normal2" + assert normal_result == "normal" + + # CRITICAL 应在 normal2 之前开始 + critical_idx = next(i for i, (_, r) in enumerate(start_times) if r == "critical") + normal2_idx = next(i for i, (_, r) in enumerate(start_times) if r == "normal2") + assert critical_idx < normal2_idx, ( + f"CRITICAL 应在 NORMAL2 之前执行，实际顺序: {start_times}" + ) + finally: + await scheduler.stop() + + +async def test_drain_empty_queue(): + """空队列 drain 应立即返回 True。""" + scheduler = UIActionScheduler() + result = await scheduler.drain(timeout=1.0) + assert result is True + + +async def test_drain_with_pending(): + """有 pending 任务的队列 drain 应等待任务完成。""" + scheduler = UIActionScheduler(send_delay_ms=10) + await scheduler.start() + try: + await scheduler.enqueue(lambda: asyncio.sleep(0.05, result="ok")) + result = await scheduler.drain(timeout=2.0) + assert result is True + await asyncio.sleep(0.01) # 让 worker 完全进入 idle + finally: + await scheduler.stop() + + +async def test_drain_timeout(): + """长任务未完成时 drain 应超时返回 False。""" + scheduler = UIActionScheduler(send_delay_ms=0) + await scheduler.start() + try: + # 入队长任务但不 await（避免阻塞测试） + asyncio.create_task( + scheduler.enqueue(lambda: asyncio.sleep(1.0, result="slow")) + ) + await asyncio.sleep(0.05) # 让任务开始执行 + result = await scheduler.drain(timeout=0.1) + assert result is False + finally: + await scheduler.stop() + + +async def test_queue_full_rate_limited(): + """队列满时入队应抛 RATE_LIMITED。""" + scheduler = UIActionScheduler(send_delay_ms=1000, max_queue_size=2) + await scheduler.start() + try: + # 第一个任务开始执行（worker 取走后队列腾出空位） + asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.5))) + await asyncio.sleep(0.05) # 让 worker 取走第一个任务并开始执行 + + # 再入队 2 个排队任务（队列满：2 个 pending） + asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.01))) + asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.01))) + await asyncio.sleep(0.02) # 确保两个任务都入队 + + # 第 3 个排队任务应被拒绝（队列已满） + with pytest.raises(BridgeError) as exc_info: + await scheduler.enqueue(lambda: asyncio.sleep(0.01)) + assert exc_info.value.code == "RATE_LIMITED" + finally: + await scheduler.stop() + + +async def test_wait_timeout(): + """等待结果超时应抛 TIMEOUT。""" + scheduler = UIActionScheduler(send_delay_ms=1000, max_calls_per_sec=100) + await scheduler.start() + try: + # 第一个任务慢（执行 + 默认延时会让第二个等待很久） + asyncio.create_task(scheduler.enqueue(lambda: asyncio.sleep(0.5))) + await asyncio.sleep(0.02) # 让第一个开始 + + with pytest.raises(BridgeError) as exc_info: + await scheduler.enqueue( + lambda: asyncio.sleep(0.01), wait_timeout_ms=100 + ) + assert exc_info.value.code == "TIMEOUT" + finally: + await scheduler.stop() + + +async def test_metrics_accuracy(): + """metrics 应准确统计各优先级执行次数。""" + scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) + await scheduler.start() + try: + async def _ret(v): + return v + + await scheduler.execute(lambda: _ret("a"), priority=Priority.CRITICAL) + await scheduler.execute(lambda: _ret("b"), priority=Priority.HIGH) + await scheduler.execute(lambda: _ret("c"), priority=Priority.NORMAL) + await scheduler.execute(lambda: _ret("d"), priority=Priority.LOW) + + metrics = scheduler.get_metrics() + assert metrics["executed_total"] == 4 + assert metrics["executed_by_priority"]["CRITICAL"] == 1 + assert metrics["executed_by_priority"]["HIGH"] == 1 + assert metrics["executed_by_priority"]["NORMAL"] == 1 + assert metrics["executed_by_priority"]["LOW"] == 1 + assert metrics["pending_count"] == 0 + finally: + await scheduler.stop() + + +async def test_fast_fail_wechat_dead(): + """mark_wechat_dead 后非 CRITICAL 任务应 fast-fail 抛 WECHAT_NOT_READY。""" + scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) + await scheduler.start() + try: + scheduler.mark_wechat_dead(2.0) # 启动 2 秒 fast-fail 窗口 + + with pytest.raises(BridgeError) as exc_info: + await scheduler.execute( + lambda: asyncio.sleep(0.01, result="ok"), + priority=Priority.NORMAL, + ) + assert exc_info.value.code == "WECHAT_NOT_READY" + + metrics = scheduler.get_metrics() + assert metrics["fast_fail_total"] == 1 + finally: + await scheduler.stop() + + +async def test_critical_bypasses_fast_fail(): + """CRITICAL 任务在 fast-fail 窗口内应正常执行。""" + scheduler = UIActionScheduler(send_delay_ms=0, max_calls_per_sec=100) + await scheduler.start() + try: + scheduler.mark_wechat_dead(2.0) # 启动 fast-fail 窗口 + + result = await scheduler.execute( + lambda: asyncio.sleep(0.01, result="critical_ok"), + priority=Priority.CRITICAL, + ) + assert result == "critical_ok" + + metrics = scheduler.get_metrics() + assert metrics["executed_total"] == 1 + assert metrics["executed_by_priority"]["CRITICAL"] == 1 + assert metrics["fast_fail_total"] == 0 + finally: + await scheduler.stop() + + +async def test_critical_auto_delay_zero(): + """CRITICAL 任务应自动 delay_ms=0，不受默认 send_delay_ms 影响。""" + scheduler = UIActionScheduler(send_delay_ms=5000, max_calls_per_sec=100) + await scheduler.start() + try: + async def _ret(v): + return v + + t0 = time.monotonic() + await scheduler.execute(lambda: _ret("first"), priority=Priority.CRITICAL) + await scheduler.execute(lambda: _ret("second"), priority=Priority.CRITICAL) + elapsed = time.monotonic() - t0 + + # 若 delay_ms=5000 生效，第二个任务会等 5s。自动 delay=0 后应在 1s 内完成 + assert elapsed < 1.5, f"CRITICAL 自动 delay=0 失效，耗时 {elapsed:.2f}s" + finally: + await scheduler.stop() diff --git a/bridge/woc_bridge/app.py b/bridge/woc_bridge/app.py index 3d802db..e7e64e2 100644 --- a/bridge/woc_bridge/app.py +++ b/bridge/woc_bridge/app.py @@ -24,6 +24,7 @@ from woc_bridge.models import BridgeError, ErrorResponse from woc_bridge.version import BRIDGE_VERSION from woc_bridge.db import DbReader from woc_bridge.messaging import SendQueue +from woc_bridge.messaging.ui_action_scheduler import UIActionScheduler from woc_bridge.ui import QrCapture, XdotoolDriver # 路由 @@ -380,7 +381,7 @@ def _init_state(cfg: BridgeConfig) -> None: _state.db_reader = DbReader(db_root=cfg.wechat_db, key_cache=_state.key_cache) _state.qr_capture = QrCapture(display=cfg.display) - _state.send_queue = SendQueue( + _state.send_queue = UIActionScheduler( send_delay_ms=cfg.send_delay_ms, max_calls_per_sec=cfg.max_calls_per_sec, max_queue_size=cfg.max_queue_size, @@ -496,7 +497,11 @@ def _init_state(cfg: BridgeConfig) -> None: ) # 构造 HA 模块 - _state.watchdog = WeChatWatchdog(backend=_state.xdotool_backend, interval=10.0) + _state.watchdog = WeChatWatchdog( + backend=_state.xdotool_backend, + interval=10.0, + scheduler=_state.send_queue, + ) _state.resource_reaper = ResourceReaper( debug_screenshot_dir="/tmp/woc_debug", max_age_hours=24, diff --git a/bridge/woc_bridge/messaging/ui_action_scheduler.py b/bridge/woc_bridge/messaging/ui_action_scheduler.py new file mode 100644 index 0000000..a6f01a1 --- /dev/null +++ b/bridge/woc_bridge/messaging/ui_action_scheduler.py @@ -0,0 +1,332 @@ +"""UIActionScheduler —— 基于 SendQueue 的优先级调度器。 + +在父类 SendQueue 串行化队列之上扩展： +1. 优先级调度（PriorityQueue + IntEnum 优先级），CRITICAL 任务优先出队且不受 + 默认延时约束； +2. fast-fail 机制：微信进程死亡期间，非 CRITICAL 任务立即失败，避免堆积； +3. Prometheus metrics 暴露：执行计数 / 等待与执行耗时 / 超时 / 限流 / fast-fail。 + +元素类型为 5 元组 `(priority_int, seq, coro_factory, future, delay_ms)`， +其中 seq 为单调递增序列号，保证同优先级 FIFO。 +""" + +from __future__ import annotations + +import asyncio +import enum +import logging +import time +from typing import Any, Awaitable, Callable, Optional + +from woc_bridge.messaging.send_queue import SendQueue, CoroFactory +from woc_bridge.models import BridgeError +from woc_bridge.ui.metrics import ( + ui_action_executed, # Counter, label=priority + ui_action_wait_duration, # Histogram + ui_action_exec_duration, # Histogram, label=priority + ui_action_timeout, # Counter + ui_action_rate_limited, # Counter + ui_action_fast_fail, # Counter +) + + +logger = logging.getLogger("woc-bridge") + + +def _priority_name(priority: int) -> str: + """将 priority int 转为名称，非法值返回 "?"。""" + try: + return Priority(priority).name + except ValueError: + return "?" + + +class Priority(enum.IntEnum): + """UI 操作优先级。 + + 数值越小优先级越高；CRITICAL 不受 fast-fail 与默认延时影响。 + """ + + CRITICAL = 0 # wechat_restart / diagnostic_autofix + HIGH = 1 # logout / friend_accept / login_qr_capture + NORMAL = 2 # send_text / send_file / set_remark / add_friend + LOW = 3 # screenshot / moments_like / moments_comment + + +class UIActionScheduler(SendQueue): + """优先级 UI 调度器。 + + 覆盖父类的 ``_queue`` 为 :class:`asyncio.PriorityQueue`，元素为 5 元组 + ``(priority, seq, coro_factory, future, delay_ms)``。提供 :meth:`execute` + 入口携带优先级与 trace_id，并暴露 metrics / drain 等运维接口。 + """ + + def __init__( + self, + send_delay_ms: int = 3000, + max_calls_per_sec: int = 10, + max_queue_size: int = 100, + ) -> None: + super().__init__( + send_delay_ms=send_delay_ms, + max_calls_per_sec=max_calls_per_sec, + max_queue_size=max_queue_size, + ) + # 覆盖父类的 Queue 为 PriorityQueue；元素 5 元组 + self._queue: asyncio.PriorityQueue[ + tuple[int, int, CoroFactory, asyncio.Future, Optional[int]] + ] = asyncio.PriorityQueue(maxsize=max_queue_size) + # 单调递增序列号，保证同优先级 FIFO + self._seq: int = 0 + # fast-fail 截止时间戳（monotonic）；>0 表示微信死亡期间 + self._wechat_dead_until: float = 0.0 + # metrics 计数器 + self._metrics: dict[str, Any] = { + "executed_total": 0, + "executed_by_priority": {p.name: 0 for p in Priority}, + "wait_duration_ms_sum": 0.0, + "exec_duration_ms_sum": 0.0, + "timeout_total": 0, + "rate_limited_total": 0, + "fast_fail_total": 0, + } + + def mark_wechat_dead(self, duration_sec: float = 10.0) -> None: + """标记微信进程死亡，期间非 CRITICAL 任务 fast-fail。 + + Args: + duration_sec: 死亡窗口秒数，期间入队/出队的非 CRITICAL 任务 + 会被立即以 WECHAT_NOT_READY 失败 + """ + self._wechat_dead_until = time.monotonic() + duration_sec + logger.warning( + "[scheduler] mark_wechat_dead %.1fs, %d pending tasks will fast-fail", + duration_sec, self._queue.qsize(), + ) + + async def execute( + self, + action: CoroFactory, + priority: Priority = Priority.NORMAL, + delay_ms: Optional[int] = None, + wait_timeout_ms: Optional[int] = None, + trace_id: str = "", + ) -> Any: + """按优先级入队并等待执行结果。 + + CRITICAL 任务若未指定 delay_ms，自动设为 0（立即执行）。 + + Args: + action: 返回 coroutine 的工厂函数 + priority: 优先级，默认 NORMAL + delay_ms: 自定义执行后延时毫秒，None 用默认 send_delay_ms + wait_timeout_ms: 等待结果超时毫秒，None 表示无限等待 + trace_id: 追踪 ID，用于日志关联 + + Returns: + coroutine 的执行结果 + + Raises: + BridgeError: 队列满 / 等待超时 / 任务执行异常透传 + """ + if priority == Priority.CRITICAL and delay_ms is None: + delay_ms = 0 + return await self._enqueue_with_priority( + action, priority, delay_ms, wait_timeout_ms, trace_id, + ) + + async def _enqueue_with_priority( + self, + coro_factory: CoroFactory, + priority: Priority, + delay_ms: Optional[int], + wait_timeout_ms: Optional[int], + trace_id: str, + ) -> Any: + """实际入队逻辑：构造 5 元组入 PriorityQueue，等待 future 结果。""" + loop = asyncio.get_running_loop() + future: asyncio.Future = loop.create_future() + self._seq += 1 + item = (int(priority), self._seq, coro_factory, future, delay_ms) + try: + self._queue.put_nowait(item) + except asyncio.QueueFull: + self._metrics["rate_limited_total"] += 1 + ui_action_rate_limited.inc() + logger.warning( + "[scheduler] 队列已满，拒绝入队 (maxsize=%d, priority=%s, trace_id=%s)", + self._queue.maxsize, priority.name, trace_id, + ) + raise BridgeError( + code="RATE_LIMITED", + message=f"UI 调度器队列已满（{self._queue.maxsize}），请稍后重试", + details={"retry_after": 3}, + ) + pending = self._queue.qsize() + logger.info( + "[scheduler] 入队 priority=%s pending=%d trace_id=%s", + priority.name, pending, trace_id, + ) + wait_start = time.monotonic() + if wait_timeout_ms is not None and wait_timeout_ms > 0: + try: + result = await asyncio.wait_for( + future, timeout=wait_timeout_ms / 1000.0, + ) + except asyncio.TimeoutError: + # 竞争窗口：worker 可能刚好在此时完成并 set_result。 + if future.done() and not future.cancelled(): + logger.info( + "[scheduler] 等待超时但任务刚好完成，取结果 (pending=%d)", + pending, + ) + return future.result() + # 取消 future，worker 出队时发现 cancelled 会跳过执行 + future.cancel() + wait_sec = wait_timeout_ms / 1000.0 + self._metrics["timeout_total"] += 1 + ui_action_timeout.inc() + logger.warning( + "[scheduler] 等待超时 (pending=%d, wait_timeout=%.1fs, trace_id=%s)", + pending, wait_sec, trace_id, + ) + raise BridgeError( + code="TIMEOUT", + message=f"UI 调度器等待超时（{pending} 个待处理，已等 {wait_sec:.1f}s）", + details={"retry_after": max(1, int(self.send_delay_ms / 1000))}, + ) + else: + result = await future + wait_ms = (time.monotonic() - wait_start) * 1000 + self._metrics["wait_duration_ms_sum"] += wait_ms + ui_action_wait_duration.observe(wait_ms / 1000.0) + return result + + async def enqueue( + self, + coro_factory: CoroFactory, + delay_ms: Optional[int] = None, + wait_timeout_ms: Optional[int] = None, + ) -> Any: + """向后兼容包装：以 NORMAL 优先级入队。""" + return await self._enqueue_with_priority( + coro_factory, Priority.NORMAL, delay_ms, wait_timeout_ms, "", + ) + + async def _run(self) -> None: + """worker 主循环（覆盖父类）。 + + 解包 5 元组 ``(priority, seq, coro_factory, future, delay_ms)``， + 按优先级出队执行；含 fast-fail 与 metrics 埋点。 + """ + while True: + try: + priority, seq, coro_factory, future, custom_delay_ms = await self._queue.get() + except asyncio.CancelledError: + raise + + # 调用方已超时取消：跳过执行与延时 + if future.cancelled(): + self._queue.task_done() + logger.info( + "[scheduler] 出队任务已取消，跳过 (pending=%d, priority=%s)", + self._queue.qsize(), + _priority_name(priority), + ) + continue + + # fast-fail 检查：微信已死期间，非 CRITICAL 任务直接失败 + if (self._wechat_dead_until > time.monotonic() + and priority > Priority.CRITICAL): # CRITICAL (=0) 不受影响 + logger.info( + "[scheduler] fast-fail (wechat dead, %.1fs remaining, priority=%s)", + self._wechat_dead_until - time.monotonic(), + _priority_name(priority), + ) + if not future.done(): + future.set_exception(BridgeError( + code="WECHAT_NOT_READY", + message="微信进程未就绪（重启中），请稍后重试", + details={"retry_after": 5}, + )) + self._metrics["fast_fail_total"] += 1 + ui_action_fast_fail.inc() # Prometheus + self._queue.task_done() + continue # 不延时，立即处理下一个 + + logger.info( + "[scheduler] 出队，开始处理 (pending=%d, priority=%s)", + self._queue.qsize(), + _priority_name(priority), + ) + executed = False + t_exec = time.perf_counter() + try: + self._check_rate_limit() # 继承父类 + self._recent_call_times.append(time.monotonic()) + executed = True + pname = _priority_name(priority) + logger.info("[scheduler] 开始执行任务 (priority=%s)", pname) + result = await coro_factory() + exec_ms = (time.perf_counter() - t_exec) * 1000 + logger.info( + "[scheduler] 任务执行完成 (%.0fms, priority=%s)", + exec_ms, pname, + ) + if not future.done(): + future.set_result(result) + except asyncio.CancelledError: + if not future.done(): + future.cancel() + raise + except Exception as e: + exec_ms = (time.perf_counter() - t_exec) * 1000 + logger.warning( + "[scheduler] 任务执行抛异常 %s: %s (%.0fms, priority=%s)", + type(e).__name__, e, exec_ms, _priority_name(priority), + ) + if not future.done(): + future.set_exception(e) + finally: + self._queue.task_done() + if executed: + exec_ms = (time.perf_counter() - t_exec) * 1000 + self._metrics["executed_total"] += 1 + self._metrics["exec_duration_ms_sum"] += exec_ms + try: + p = Priority(priority) + self._metrics["executed_by_priority"][p.name] += 1 + ui_action_executed.labels(priority=p.name).inc() # Prometheus + ui_action_exec_duration.labels(priority=p.name).observe(exec_ms / 1000.0) # Prometheus + except ValueError: + pass + + delay = custom_delay_ms if custom_delay_ms is not None else self.send_delay_ms + logger.info( + "[scheduler] 延时 %dms 后处理下一个 (priority=%s)", + delay, _priority_name(priority), + ) + await asyncio.sleep(delay / 1000.0) + + def get_metrics(self) -> dict: + """返回调度器 metrics 快照（含 pending_count）。""" + return {**self._metrics, "pending_count": self._queue.qsize()} + + async def drain(self, timeout: float = 5.0) -> bool: + """等待队列排空。 + + Args: + timeout: 最长等待秒数 + + Returns: + True 表示队列已排空；False 表示超时仍有任务待处理 + """ + try: + await asyncio.wait_for(self._queue.join(), timeout=timeout) + return True + except asyncio.TimeoutError: + logger.warning( + "[scheduler] drain timeout %.1fs, %d tasks pending", + timeout, self._queue.qsize(), + ) + return False diff --git a/bridge/woc_bridge/models/base.py b/bridge/woc_bridge/models/base.py index d0cd8e2..b449cba 100644 --- a/bridge/woc_bridge/models/base.py +++ b/bridge/woc_bridge/models/base.py @@ -40,6 +40,7 @@ ERROR_CODES: dict[str, int] = { "DB_VERIFY_FAILED": 500, "AMBIGUOUS_CONTACT": 400, "BRIDGE_CIRCUITED": 503, + "WECHAT_NOT_READY": 503, } diff --git a/bridge/woc_bridge/models/status.py b/bridge/woc_bridge/models/status.py index bfb3123..9718f42 100644 --- a/bridge/woc_bridge/models/status.py +++ b/bridge/woc_bridge/models/status.py @@ -22,6 +22,8 @@ class StatusResponse(BaseModel): - media_supported：图片/文件发送是否已实现真实传输（与 BRIDGE_CAPABILITIES 中 image_send/file_send 同步） - bridge_capabilities：当前 bridge 支持的能力集合，客户端据此协商 + - ui_scheduler：UI 调度器指标（仅 UIActionScheduler 实例才有，含 + executed_total/executed_by_priority/pending_count 等） """ bridge_version: str = Field(description="bridge 版本号") @@ -48,3 +50,7 @@ class StatusResponse(BaseModel): default_factory=list, description="bridge 支持的能力集合，如 ['text_send','image_send','long_poll']", ) + ui_scheduler: Optional[dict] = Field( + default=None, + description="UI 调度器指标（仅 UIActionScheduler 实例才有，含 executed_total/executed_by_priority/pending_count 等）", + ) diff --git a/bridge/woc_bridge/routes/diagnostic.py b/bridge/woc_bridge/routes/diagnostic.py index c5554ed..7ddbc63 100644 --- a/bridge/woc_bridge/routes/diagnostic.py +++ b/bridge/woc_bridge/routes/diagnostic.py @@ -11,6 +11,7 @@ from woc_bridge.config import ( _DIAGNOSTIC_ITEM_BY_ID, _require_db_reader, _require_xdotool, + _require_send_queue, _state, ) from woc_bridge.db.coordinator import ( @@ -18,6 +19,7 @@ from woc_bridge.db.coordinator import ( _resolve_db_state, with_db_retry, ) +from woc_bridge.messaging.ui_action_scheduler import Priority from woc_bridge.models import ( BridgeError, ConnectivityResponse, @@ -226,27 +228,60 @@ async def diagnostic_autofix(check_id: str) -> DiagnosticRunResult: if check_id == "wechat_running": xdotool = _require_xdotool() + scheduler = _require_send_queue() + + # 启动 fast-fail 窗口：覆盖 autostart 拉起新进程的 ~5-10s 期间， + # 避免队列内积压任务对新窗口执行 xdotool 失败形成风暴 + if hasattr(scheduler, "mark_wechat_dead"): + try: + scheduler.mark_wechat_dead(15.0) + except Exception as exc: + logger.warning("diagnostic/autofix: mark_wechat_dead failed: %s", exc) + + async def _autofix_flow(): + # 1. pkill 旧进程（不强制 SIGKILL，给微信优雅退出机会） + try: + proc = await asyncio.create_subprocess_exec( + "pkill", "-x", "wechat", + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + ) + await proc.wait() + except Exception: + pass + + # 2. 等 2s 让 autostart watchdog 拉起 + await asyncio.sleep(2) + + # 3. 检查 autostart 是否已拉起 + new_pid = await xdotool.check_wechat_pid() + + # 4. autostart 未拉起，bridge 显式启动兜底 + if new_pid is None: + new_pid = await xdotool.start_wechat(timeout_sec=10) + + return new_pid - # 1. pkill 旧进程（不强制 SIGKILL，给微信优雅退出机会） try: - proc = await asyncio.create_subprocess_exec( - "pkill", "-x", "wechat", - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.DEVNULL, + if hasattr(scheduler, "execute"): + new_pid = await scheduler.execute( + _autofix_flow, + priority=Priority.CRITICAL, + wait_timeout_ms=30000, + trace_id="autofix_wechat_running", + ) + else: + new_pid = await _autofix_flow() + except Exception as exc: + logger.warning("diagnostic/autofix: wechat_running autofix failed: %s", exc) + return DiagnosticRunResult( + check_id=check_id, + passed=False, + severity=item["severity"], + message=f"autofix 执行失败: {exc}", + auto_repairable=True, + repair_plan="检查 X 会话是否正常（Xvnc/openbox），或通过 VNC 桌面手动启动微信", ) - await proc.wait() - except Exception: - pass - - # 2. 等 2s 让 autostart watchdog 拉起 - await asyncio.sleep(2) - - # 3. 检查 autostart 是否已拉起 - new_pid = await xdotool.check_wechat_pid() - - # 4. autostart 未拉起，bridge 显式启动兜底 - if new_pid is None: - new_pid = await xdotool.start_wechat(timeout_sec=10) # 5. 返回结果 if new_pid is not None: diff --git a/bridge/woc_bridge/routes/login.py b/bridge/woc_bridge/routes/login.py index 530e569..4c47615 100644 --- a/bridge/woc_bridge/routes/login.py +++ b/bridge/woc_bridge/routes/login.py @@ -10,8 +10,10 @@ from woc_bridge.config import ( _require_xdotool, _require_qr_capture, _require_db_reader, + _require_send_queue, ) from woc_bridge.db.coordinator import with_db_retry +from woc_bridge.messaging.ui_action_scheduler import Priority from woc_bridge.models import ( QrLoginStartResult, QrLoginWaitResult, @@ -50,13 +52,28 @@ async def login_qr_start() -> QrLoginStartResult: logger.info("login/qr/start: 收到请求") xdotool = _require_xdotool() qr_capture = _require_qr_capture() + scheduler = _require_send_queue() # 激活微信窗口（非阻塞，避免 VNC 无人操作时 --sync 死等） t0 = time.perf_counter() logger.info("login/qr/start: 激活窗口 + 截图开始") - try: + + async def _qr_capture_flow(): await xdotool._activate_window_fast() - qr_data_url = await qr_capture.capture_qr_code() + return await qr_capture.capture_qr_code() + + try: + if hasattr(scheduler, "execute"): + qr_data_url = await scheduler.execute( + _qr_capture_flow, + priority=Priority.HIGH, + wait_timeout_ms=15000, + trace_id="qr_start", + ) + else: + # legacy 回退：直接调用 + await xdotool._activate_window_fast() + qr_data_url = await qr_capture.capture_qr_code() except BridgeError as e: logger.warning( "login/qr/start: 失败 BridgeError code=%s msg=%s (%.0fms)", @@ -197,8 +214,17 @@ async def login_logout() -> LogoutResponse: if state_before == LoginState.LOGGED_IN.value: t1 = time.perf_counter() logger.info("login/logout: logout 调用开始") + scheduler = _require_send_queue() try: - await xdotool.logout() + if hasattr(scheduler, "execute"): + await scheduler.execute( + lambda: xdotool.logout(), + priority=Priority.HIGH, + wait_timeout_ms=30000, + trace_id="logout", + ) + else: + await xdotool.logout() except BridgeError as e: logger.warning( "login/logout: 失败 BridgeError code=%s msg=%s (%.0fms)", @@ -277,8 +303,26 @@ async def wechat_restart() -> RestartResponse: t1 = time.perf_counter() logger.info("wechat/restart: restart 调用开始") + scheduler = _require_send_queue() + + # 启动 fast-fail 窗口：覆盖新进程启动期（~5-10s），避免 restart 期间 + # 队列内积压的 NORMAL 任务对新窗口执行 xdotool 失败形成风暴 + if hasattr(scheduler, "mark_wechat_dead"): + try: + scheduler.mark_wechat_dead(8.0) + except Exception as exc: + logger.warning("wechat/restart: mark_wechat_dead failed: %s", exc) + try: - new_pid = await xdotool.restart_wechat(timeout_sec=timeout_sec) + if hasattr(scheduler, "execute"): + new_pid = await scheduler.execute( + lambda: xdotool.restart_wechat(timeout_sec=timeout_sec), + priority=Priority.CRITICAL, + wait_timeout_ms=60000, + trace_id="wechat_restart", + ) + else: + new_pid = await xdotool.restart_wechat(timeout_sec=timeout_sec) except BridgeError as e: logger.warning( "wechat/restart: 失败 BridgeError code=%s msg=%s (%.0fms)", diff --git a/bridge/woc_bridge/routes/screenshot.py b/bridge/woc_bridge/routes/screenshot.py index ad4bacc..e558bae 100644 --- a/bridge/woc_bridge/routes/screenshot.py +++ b/bridge/woc_bridge/routes/screenshot.py @@ -5,7 +5,8 @@ import logging from fastapi import APIRouter from fastapi.responses import Response -from woc_bridge.config import _require_qr_capture +from woc_bridge.config import _require_qr_capture, _require_send_queue +from woc_bridge.messaging.ui_action_scheduler import Priority from woc_bridge.models import BridgeError logger = logging.getLogger("woc-bridge") @@ -37,9 +38,18 @@ async def screenshot() -> Response: """ logger.info("screenshot: 收到请求") qr_capture = _require_qr_capture() + scheduler = _require_send_queue() try: - png_bytes = await qr_capture.capture_full_screenshot() + if hasattr(scheduler, "execute"): + png_bytes = await scheduler.execute( + lambda: qr_capture.capture_full_screenshot(), + priority=Priority.LOW, + wait_timeout_ms=10000, + trace_id="screenshot", + ) + else: + png_bytes = await qr_capture.capture_full_screenshot() except BridgeError as e: logger.warning("screenshot: 失败 %s: %s", e.code, e.message) raise diff --git a/bridge/woc_bridge/routes/status.py b/bridge/woc_bridge/routes/status.py index 95fc99e..77bb195 100644 --- a/bridge/woc_bridge/routes/status.py +++ b/bridge/woc_bridge/routes/status.py @@ -9,6 +9,7 @@ from fastapi import APIRouter from woc_bridge.config import _state, _require_xdotool, _require_db_reader from woc_bridge.db.coordinator import with_db_retry, _resolve_db_state from woc_bridge.models import StatusResponse +from woc_bridge.ui.metrics import send_queue_pending as send_queue_pending_gauge from woc_bridge.version import BRIDGE_VERSION, BRIDGE_CAPABILITIES, MEDIA_SUPPORTED logger = logging.getLogger("woc-bridge") @@ -77,6 +78,21 @@ async def get_status() -> StatusResponse: uptime_seconds = time.monotonic() - _state.start_time + send_queue = _state.send_queue + send_queue_pending = send_queue.pending_count() if send_queue else 0 + ui_scheduler_metrics = None + if send_queue is not None and hasattr(send_queue, "get_metrics"): + try: + ui_scheduler_metrics = send_queue.get_metrics() + except Exception as exc: + logger.warning("[status] get_metrics failed: %s", exc) + + # 修复既有 Prometheus Gauge bug：每次 status 请求时 set 当前 pending 数 + try: + send_queue_pending_gauge.set(send_queue_pending) + except Exception as exc: + logger.debug("[status] send_queue_pending_gauge.set failed: %s", exc) + return StatusResponse( bridge_version=BRIDGE_VERSION, wechat_running=wechat_running, @@ -93,7 +109,8 @@ async def get_status() -> StatusResponse: display=_state.config.display, max_batch_size=_state.config.max_batch_size, poll_interval_ms=_state.config.poll_interval_ms, - send_queue_pending=_state.send_queue.pending_count() if _state.send_queue else 0, + send_queue_pending=send_queue_pending, media_supported=MEDIA_SUPPORTED, bridge_capabilities=list(BRIDGE_CAPABILITIES), + ui_scheduler=ui_scheduler_metrics, ) diff --git a/bridge/woc_bridge/ui/flow.py b/bridge/woc_bridge/ui/flow.py index bac88d6..4adf8b4 100644 --- a/bridge/woc_bridge/ui/flow.py +++ b/bridge/woc_bridge/ui/flow.py @@ -23,7 +23,7 @@ from dataclasses import dataclass, field from typing import Optional from woc_bridge.ui.backends.base import WindowGeometry -from woc_bridge.ui.errors import FlowError +from woc_bridge.ui.errors import FlowError, PermanentError logger = logging.getLogger("woc-bridge") @@ -227,6 +227,29 @@ class Flow: # ------------------------------------------------------------------ # 子类实现 # ------------------------------------------------------------------ + def _ensure_valid_geometry(self, geom: WindowGeometry) -> WindowGeometry: + """校验窗口几何有效（尺寸 >= 200x200），无效则抛 PermanentError。 + + 微信启动过程中可能出现隐藏/加载小窗（如 1x1），虽然 backend 的 + _find_main_window_id 已做 min_area 过滤，这里再做一次防御性校验， + 避免后续 click_element 用错误几何计算坐标导致点到窗口外。 + + Args: + geom: 待校验的窗口几何 + + Returns: + 原始 geom（校验通过） + + Raises: + PermanentError: geom 无效（width/height < 200） + """ + if geom.width < 200 or geom.height < 200: + raise PermanentError( + code="WINDOW_NOT_FOUND", + message=f"窗口几何无效: {geom.width}x{geom.height}（需 >=200x200）", + ) + return geom + async def run(self, ctx: FlowContext) -> FlowResult: """子类实现：编排各 transition，返回 FlowResult。""" raise NotImplementedError("subclass must implement run()") diff --git a/bridge/woc_bridge/ui/metrics.py b/bridge/woc_bridge/ui/metrics.py index faa0dbc..732468c 100644 --- a/bridge/woc_bridge/ui/metrics.py +++ b/bridge/woc_bridge/ui/metrics.py @@ -1,6 +1,10 @@ """Prometheus 指标定义：发送计数/耗时/失败/图像匹配/DB校验/队列/运行态/熔断/缓存/自适应等待。 所有指标用 prometheus_client 定义，通过 /metrics 端点暴露（app.py 挂载）。 + +prometheus_client 为可选依赖：缺失时本模块降级为 NoOp 桩件，所有指标调用 +（.labels()/.inc()/.set()/.observe() 等）变为 no-op，bridge 核心业务不受影响。 +app.py 的 /metrics 端点会同步降级（make_asgi_app 仍 ImportError → 禁用并 warning）。 """ from __future__ import annotations @@ -8,15 +12,59 @@ from __future__ import annotations import logging import time -from prometheus_client import ( - Counter, - Gauge, - Histogram, - Info, -) - logger = logging.getLogger("woc-bridge") + +class _NoOpMetric: + """prometheus_client 缺失时的 NoOp 桩件。 + + 覆盖上层所有调用模式：实例化（接收任意参数）+ + .labels(*args, **kwargs) → 返回 self，支持链式 .inc()/.dec()/.set()/.observe()。 + 所有操作 no-op，不持有任何状态。 + """ + + __slots__ = () + + def __init__(self, *args, **kwargs) -> None: + pass + + def labels(self, *args, **kwargs) -> "_NoOpMetric": + return self + + def inc(self, amount: float = 1.0) -> None: + pass + + def dec(self, amount: float = 1.0) -> None: + pass + + def set(self, value: float) -> None: + pass + + def observe(self, amount: float) -> None: + pass + + def info(self, val) -> None: + pass + + +try: + from prometheus_client import ( + Counter, + Gauge, + Histogram, + Info, + ) + _PROMETHEUS_AVAILABLE = True +except ImportError: + logger.warning( + "[metrics] prometheus_client 不可用，指标降级为 NoOp；" + "/metrics 端点将自动禁用，核心业务不受影响" + ) + _PROMETHEUS_AVAILABLE = False + # 类型忽略：故意将多个指标类名重绑定为同一个 NoOp 类， + # 使所有 Counter/Gauge/Histogram/Info(...) 实例化返回 _NoOpMetric 实例 + Counter = Gauge = Histogram = Info = _NoOpMetric # type: ignore[assignment,misc] + # 发送总数（按 flow_name / result 标签） send_total = Counter( "woc_send_total", @@ -102,6 +150,42 @@ send_duration_cached = Histogram( ) +# UIActionScheduler 调度器指标 +ui_action_executed = Counter( + "woc_ui_action_executed_total", + "UI actions executed total", + ["priority"], # CRITICAL / HIGH / NORMAL / LOW +) + +ui_action_wait_duration = Histogram( + "woc_ui_action_wait_duration_seconds", + "UI action wait duration (queue wait time)", + buckets=(0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0), +) + +ui_action_exec_duration = Histogram( + "woc_ui_action_exec_duration_seconds", + "UI action execution duration", + ["priority"], + buckets=(0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0), +) + +ui_action_timeout = Counter( + "woc_ui_action_timeout_total", + "UI action wait timeout count", +) + +ui_action_rate_limited = Counter( + "woc_ui_action_rate_limited_total", + "UI action rate limited (queue full) count", +) + +ui_action_fast_fail = Counter( + "woc_ui_action_fast_fail_total", + "UI action fast-fail count (wechat dead window)", +) + + def set_circuit_state(name: str, state_int: int) -> None: """更新熔断器状态指标。 diff --git a/bridge/woc_bridge/ui/qr_capture.py b/bridge/woc_bridge/ui/qr_capture.py index 586545c..e704009 100644 --- a/bridge/woc_bridge/ui/qr_capture.py +++ b/bridge/woc_bridge/ui/qr_capture.py @@ -17,6 +17,8 @@ import tempfile from woc_bridge.models import BridgeError +_CMD_TIMEOUT_SEC = 5.0 + class QrCapture: """屏幕截图与二维码裁剪器。 @@ -46,15 +48,26 @@ class QrCapture: return env async def _run(self, args: list[str]) -> tuple[int, bytes, bytes]: - """执行一条命令并返回 (returncode, stdout, stderr)。""" + """执行一条命令并返回 (returncode, stdout, stderr)。 + + 所有子进程执行均受 _CMD_TIMEOUT_SEC 超时保护，避免 scrot 等命令卡死 + 导致无限阻塞（与并发 UI 操作竞争 X11 时可能发生）。 + """ proc = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=self._env(), ) - stdout, stderr = await proc.communicate() - return proc.returncode, stdout, stderr + try: + stdout, stderr = await asyncio.wait_for( + proc.communicate(), timeout=_CMD_TIMEOUT_SEC + ) + return proc.returncode, stdout, stderr + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + raise async def _capture_full_png(self, dest_path: str) -> None: """用 `scrot ` 截取整个屏幕到 PNG 文件。 diff --git a/bridge/woc_bridge/ui/watchdog.py b/bridge/woc_bridge/ui/watchdog.py index ec8f0e8..d5a5095 100644 --- a/bridge/woc_bridge/ui/watchdog.py +++ b/bridge/woc_bridge/ui/watchdog.py @@ -9,7 +9,7 @@ from __future__ import annotations import asyncio import logging import time -from typing import Optional +from typing import Any, Optional from woc_bridge.ui.backends.base import BackendProtocol @@ -30,10 +30,12 @@ class WeChatWatchdog: backend: BackendProtocol, interval: float = 10.0, fail_threshold: int = 2, + scheduler: Optional[Any] = None, ) -> None: self.backend = backend self.interval = interval self.fail_threshold = fail_threshold + self._scheduler = scheduler # UIActionScheduler 实例（用于 kill 前 drain + mark_wechat_dead） self._fail_count = 0 self._task: Optional[asyncio.Task] = None self._stopped = False @@ -139,8 +141,30 @@ class WeChatWatchdog: 不直接启动 WeChat（避免与 autostart 竞态），仅 kill 旧进程。 所有子进程超时均执行 proc.kill() 收尸，避免僵尸进程泄漏。 + + kill 前若注入了 scheduler，先 drain 等待队列清空，再 mark_wechat_dead + 启动 fast-fail 窗口，避免 kill 后队列内任务形成失败风暴。 """ logger.warning("[watchdog] autofix: killing wechat for restart") + + # kill 前等待 UI 操作队列清空（避免 kill 正在执行的 UI 操作） + # 并启动 fast-fail 窗口拦截 kill 后新入队的非 CRITICAL 任务 + if self._scheduler is not None: + drained = await self._scheduler.drain(timeout=5.0) + if not drained: + try: + pending = self._scheduler.pending_count() + except Exception: + pending = -1 + logger.warning( + "[watchdog] drain timeout, force kill with %d UI ops pending", + pending, + ) + try: + self._scheduler.mark_wechat_dead(15.0) + except Exception as exc: + logger.warning("[watchdog] mark_wechat_dead failed: %s", exc) + try: # 获取所有 wechat PID proc = await asyncio.create_subprocess_exec(