Kris
ed2ee086de
主要变更： 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前队列清空与快速失败窗口
216 lines
8.0 KiB
Python
216 lines
8.0 KiB
Python
"""WeChat 进程看门狗：10s 检查间隔，连续 2 次失败触发 autofix。
|
||
|
||
autofix 策略：尝试重启 WeChat 进程（通过 backend.activate_window 或 pgrep+kill）。
|
||
失败仅告警不抛异常，避免拖垮 bridge 主流程。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from typing import Any, Optional
|
||
|
||
from woc_bridge.ui.backends.base import BackendProtocol
|
||
|
||
logger = logging.getLogger("woc-bridge")
|
||
|
||
|
||
class WeChatWatchdog:
|
||
"""WeChat 进程看门狗。
|
||
|
||
Attributes:
|
||
backend: L1 Backend 实例（用于 pgrep/kill 等子进程操作）
|
||
interval: 检查间隔秒数，默认 10.0
|
||
fail_threshold: 连续失败触发 autofix 的阈值，默认 2
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
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
|
||
|
||
async def start(self) -> None:
|
||
"""启动看门狗后台任务。"""
|
||
if self._task is not None and not self._task.done():
|
||
return
|
||
self._stopped = False
|
||
self._task = asyncio.create_task(self._run())
|
||
logger.info("[watchdog] started (interval=%.1fs)", self.interval)
|
||
|
||
async def stop(self) -> None:
|
||
"""停止看门狗。"""
|
||
self._stopped = True
|
||
if self._task is not None and not self._task.done():
|
||
self._task.cancel()
|
||
try:
|
||
await asyncio.wait_for(self._task, timeout=2.0)
|
||
except (asyncio.TimeoutError, asyncio.CancelledError):
|
||
pass
|
||
self._task = None
|
||
logger.info("[watchdog] stopped")
|
||
|
||
async def _run(self) -> None:
|
||
"""主循环：每 interval 秒检查一次。"""
|
||
while not self._stopped:
|
||
try:
|
||
await asyncio.sleep(self.interval)
|
||
await self._check()
|
||
except asyncio.CancelledError:
|
||
break
|
||
except Exception as exc:
|
||
logger.warning("[watchdog] check exception: %s", exc)
|
||
|
||
async def _check(self) -> None:
|
||
"""执行一次检查：WeChat 运行态 + X11 可用性。"""
|
||
wechat_ok = await self._is_wechat_running()
|
||
x11_ok = await self._is_x11_available()
|
||
|
||
if wechat_ok and x11_ok:
|
||
if self._fail_count > 0:
|
||
logger.info(
|
||
"[watchdog] recovered after %d failures", self._fail_count
|
||
)
|
||
self._fail_count = 0
|
||
return
|
||
|
||
self._fail_count += 1
|
||
logger.warning(
|
||
"[watchdog] check failed (wechat=%s x11=%s) fail_count=%d/%d",
|
||
wechat_ok, x11_ok, self._fail_count, self.fail_threshold,
|
||
)
|
||
|
||
if self._fail_count >= self.fail_threshold:
|
||
logger.error(
|
||
"[watchdog] fail threshold reached, trigger autofix"
|
||
)
|
||
await self._autofix_wechat()
|
||
# autofix 后重置计数，等下一轮检查
|
||
self._fail_count = 0
|
||
|
||
async def _is_wechat_running(self) -> bool:
|
||
"""检查 WeChat 进程是否运行（用 pgrep -x wechat）。"""
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"pgrep", "-x", "wechat",
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
try:
|
||
stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=3.0)
|
||
except asyncio.TimeoutError:
|
||
proc.kill()
|
||
await proc.wait()
|
||
return False
|
||
return proc.returncode == 0 and bool(stdout.decode(errors="ignore").strip())
|
||
except Exception as exc:
|
||
logger.debug("[watchdog] pgrep wechat exception: %s", exc)
|
||
return False
|
||
|
||
async def _is_x11_available(self) -> bool:
|
||
"""检查 X11 是否可用（用 xdpyinfo）。"""
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"xdpyinfo",
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
try:
|
||
await asyncio.wait_for(proc.communicate(), timeout=3.0)
|
||
except asyncio.TimeoutError:
|
||
proc.kill()
|
||
await proc.wait()
|
||
return False
|
||
return proc.returncode == 0
|
||
except Exception as exc:
|
||
logger.debug("[watchdog] xdpyinfo exception: %s", exc)
|
||
return False
|
||
|
||
async def _autofix_wechat(self) -> None:
|
||
"""自动修复：尝试 kill WeChat 让 autostart 拉起。
|
||
|
||
不直接启动 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(
|
||
"pgrep", "-x", "wechat",
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
try:
|
||
stdout, _ = await asyncio.wait_for(
|
||
proc.communicate(), timeout=3.0
|
||
)
|
||
except asyncio.TimeoutError:
|
||
proc.kill()
|
||
await proc.wait()
|
||
logger.warning("[watchdog] autofix pgrep timeout")
|
||
return
|
||
pids = []
|
||
for line in stdout.decode(errors="ignore").splitlines():
|
||
line = line.strip()
|
||
if line:
|
||
try:
|
||
pids.append(int(line))
|
||
except ValueError:
|
||
pass
|
||
# kill 所有 wechat 进程
|
||
for pid in pids:
|
||
try:
|
||
kill_proc = await asyncio.create_subprocess_exec(
|
||
"kill", "-TERM", str(pid),
|
||
stdout=asyncio.subprocess.PIPE,
|
||
stderr=asyncio.subprocess.PIPE,
|
||
)
|
||
try:
|
||
await asyncio.wait_for(
|
||
kill_proc.communicate(), timeout=3.0
|
||
)
|
||
except asyncio.TimeoutError:
|
||
kill_proc.kill()
|
||
await kill_proc.wait()
|
||
logger.info(
|
||
"[watchdog] autofix killed wechat pid=%d", pid
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"[watchdog] autofix kill pid=%d failed: %s", pid, exc
|
||
)
|
||
except Exception as exc:
|
||
logger.error("[watchdog] autofix exception: %s", exc)
|