提交 72db5e2e authored 作者: 陈泽健's avatar 陈泽健

fix(scheduler): 新增僵尸执行常驻看门狗,杜绝执行线程死亡卡死调度

背景:5.44 出现执行线程死亡但进程存活的「僵尸执行」——DB 记录卡 running
29 小时,内存锁已释放但 DB 层 running 守卫(scheduler_service.py)永久挡死
调度,定时任务 6 小时触发全部被跳过。启动期 recover_interrupted_executions
只兜进程重启,无法兜线程死亡。

实现(execution_service.py + main.py):
- 新增常驻 watchdog_loop,每 60s 扫描 UI running/pending 执行
- 双活信号:内存心跳(每用例完成后 touch)+ DB 结果数跨扫描增长
  (非绝对数量,存量结果不算存活)
- 30 分钟无活信号判僵尸,标记 failed + 释放内存锁(set_running_execution)
- 安全测试执行不纳入看门狗;5 分钟启动缓冲避免误杀

验证:
- 新增 tests/test_watchdog_stale_execution.py(9 例,全绿)
- 5.202/5.60 已部署并清理各自 1 个僵尸执行(330/0 条结果),自动触发新回归
- 5.44 等待当前 332 用例回归跑完后部署
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 11e1dd85
# HANDOFF — UI自动化测试交接文档 # HANDOFF — UI自动化测试交接文档
> **生成时间**: 2026-08-27 > **生成时间**: 2026-08-29
> **当前分支**: `platform-auto-test` > **当前分支**: `platform-auto-test`
> **最近提交**: `c73ffd58` feat(perf): AI 分析报告资源使用分析四维度增强(Java 进程 + MySQL) > **最近提交**: `3a29883b` feat(smart-locate): P1/P2 智能定位统一执行路径 + micro-app 递归穿透与状态确认
> **状态**: 🟢 **会话43:薄弱模块 R7 执行器修复正式部署三台——`playwright_executor.py`(commit `a63d090c`,R7 版)按安全流程(备份→上传→SHA-256→原子替换→docker restart→健康检查→容器内 AST/SHA 比对)部署到 5.44/5.202/5.60,三台全部验证通过。至此「执行器代码修复未部署」遗留闭环,远端薄弱模块 NO-BODY 问题解除。** > **状态**: 🟢 **会话44:僵尸执行常驻看门狗上线——5.44 复现「DB 卡 running 29 小时挡死调度」(进程存活、线程死亡,启动恢复无法覆盖),新增 `watchdog_loop` 常驻协程(60s 扫描/30 分钟无进展判定/双信号活性:内存心跳 + DB 结果数增长),部署 5.202/5.60 完成(顺带清理两台 8/27 遗留僵尸),5.44 等当日回归跑完后自动部署(后台长轮询挂起)。会话 43 的 P1/P2 部署待办同时闭环。**
---
## 📊 当前状态(会话 44,2026-08-29)
**僵尸执行常驻看门狗**:会话 42 的 `recover_interrupted_executions()` 只兜「进程重启」场景;8/29 排查发现 5.44 一条 `exec_23393d`(8/28 09:16 定时回归,332 用例跑完 221 条后线程死亡)卡 running **29 小时**,期间调度器 DB 守卫每轮跳过(日志刷「到点但 DB 中仍有 UI 执行在跑」),且 `POST /run` 返回 `triggered` 但内层静默跳过(外层只查内存锁)。手动 cancel 后 4 秒调度恢复并触发新一轮——证实根因链。
### ✅ 修复内容(会话 44,2026-08-29)
| 项目 | 说明 |
|------|------|
| 制品 | `execution_service.py`(看门狗实现)+ `main.py`(lifespan 启动/停止) |
| 看门狗参数 | 扫描间隔 60s;静默阈值 1800s(30 分钟无进展判僵尸);运行存活下限 300s(启动窗口期不判罚) |
| 双信号活性判定 | ① 内存心跳:工作线程每完成一条用例 `touch_execution_heartbeat()`;② DB 结果计数**增长**(跨进程兜底;关键设计:**存量结果不算增长**——首次观测只记基线,防止把死执行误判为活跃,即 5.44 事件形态) |
| 判罚动作 | 执行 + 关联 pending/running 用例结果标记 failed + 清心跳 + 释放内存锁 → 调度立即恢复 |
| 防误伤 | security 执行不纳入;pending 需超静默阈值才判;running 需「存活超下限 + 双信号全静默」 |
| 测试 | `tests/test_watchdog_stale_execution.py` 9 个用例(含增长信号跨扫描语义);全量 333 passed |
### ✅ 部署结果(会话 44,2026-08-29,`deploy_watchdog.py`)
| 服务器 | 结果 | 附带清理 |
|--------|------|----------|
| 5.202 | ✅ 看门狗已启动(日志确认) | 启动恢复清掉 8/27 22:39 僵尸 1 条 + 330 条卡住结果;调度恢复后新一轮回归已自动触发(真实执行) |
| 5.60 | ✅ 看门狗已启动(日志确认) | 启动恢复清掉 8/27 22:40 僵尸 1 条;新回归已自动触发 |
| 5.44 | ⏳ 等待中 | 14:43 清僵尸后调度恢复、新一轮 332 用例回归正在健康执行(chromium 活跃、结果增长,~2 分钟/条),**不可打断**;后台长轮询(8h)等 running=0 自动部署 |
**遗留观察(非本次范围)**:「每日定时自动回归测试」实际 `interval=6h`,但单轮 332 用例需 8 小时以上(8/26 那轮跑了 19.4h)——执行窗口远大于调度周期,即使无僵尸,除首轮外每 6h 的触发都会被运行守卫跳过。如需高频回归需缩减用例集或拆分任务。
--- ---
...@@ -493,7 +520,7 @@ cd frontend && npm run build ...@@ -493,7 +520,7 @@ cd frontend && npm run build
- [ ] **预定2.0(meetingV2)独立页面探测与用例生成**(首页无独立直达入口,需从功能中心 Tab/`[data-id]` 入口探测) - [ ] **预定2.0(meetingV2)独立页面探测与用例生成**(首页无独立直达入口,需从功能中心 Tab/`[data-id]` 入口探测)
- [x] ~~**定时任务僵尸执行 + 重叠守卫**~~(会话 42 已修复并部署三台:`recover_interrupted_executions()` 启动恢复 + DB 级重叠兜底守卫 + lifespan 调用;清理 5.44/5.202/5.60 的 18/14/68 条 running 僵尸,对应 2606/1814/401 用例结果标 failed;324 测试全绿) - [x] ~~**定时任务僵尸执行 + 重叠守卫**~~(会话 42 已修复并部署三台:`recover_interrupted_executions()` 启动恢复 + DB 级重叠兜底守卫 + lifespan 调用;清理 5.44/5.202/5.60 的 18/14/68 条 running 僵尸,对应 2606/1814/401 用例结果标 failed;324 测试全绿)
- [x] ~~**薄弱模块 R7 修复部署三台服务器(5.44/5.202/5.60)**~~(会话 43 完成:`playwright_executor.py` commit `a63d090c` 按安全流程部署三台,容器内 SHA 一致、健康检查通过;会话 40 用例数据 + 本次代码修复完整生效,远端薄弱模块 NO-BODY 解除。`deploy_executor_r7.py` 新增 `--artifact` 参数支持指定部署制品) - [x] ~~**薄弱模块 R7 修复部署三台服务器(5.44/5.202/5.60)**~~(会话 43 完成:`playwright_executor.py` commit `a63d090c` 按安全流程部署三台,容器内 SHA 一致、健康检查通过;会话 40 用例数据 + 本次代码修复完整生效,远端薄弱模块 NO-BODY 解除。`deploy_executor_r7.py` 新增 `--artifact` 参数支持指定部署制品)
- [ ] **P1/P2 执行器代码部署三台(待部署)**(本地工作树含 `playwright_executor.py` P1b/P2 增强 + `smart_locate_service.py` + `smart_locate.py` + 前端 dist,须前端 `npm run build` 后一并部署 - [x] ~~**P1/P2 执行器代码部署三台**~~(会话 43 已完成:`deploy_p1p2.py` 部署 `playwright_executor.py` P1b/P2 增强 + `smart_locate_service.py` + `smart_locate.py` + 前端 dist 三台,6/6 验证通过;已提交 `3a29883b`
- [ ] **monitor 内层路由被忽略**(微应用固定渲染设备列表,忽略 URL 内层 Maintenancelist/Filemanage 路由;相关用例已按实际渲染断言,如需真正内页需进一步探测侧边栏交互方案) - [ ] **monitor 内层路由被忽略**(微应用固定渲染设备列表,忽略 URL 内层 Maintenancelist/Filemanage 路由;相关用例已按实际渲染断言,如需真正内页需进一步探测侧边栏交互方案)
- [ ] 完善二级菜单映射表(`menu_mapping.py` - [ ] 完善二级菜单映射表(`menu_mapping.py`
- [ ] 智能定位 API 与用例执行路径统一(当前两者逻辑分离) - [ ] 智能定位 API 与用例执行路径统一(当前两者逻辑分离)
......
...@@ -145,6 +145,17 @@ async def lifespan(app: FastAPI): ...@@ -145,6 +145,17 @@ async def lifespan(app: FastAPI):
except Exception as e: except Exception as e:
logger.warning(f"定时任务调度引擎启动失败: {e}") logger.warning(f"定时任务调度引擎启动失败: {e}")
# 启动僵尸执行看门狗:进程常驻期间周期性标记「长时间无进展」的
# running/pending 执行,避免容器未重启但执行线程异常退出时,
# DB 记录卡在 running 挡死定时任务(5.44 复现:挡死调度 29 小时)。
watchdog_task = None
try:
from app.services.execution_service import start_watchdog
watchdog_task = start_watchdog()
logger.info("僵尸执行看门狗已启动")
except Exception as e:
logger.warning(f"僵尸执行看门狗启动失败: {e}")
yield yield
# 关闭时 # 关闭时
...@@ -168,6 +179,15 @@ async def lifespan(app: FastAPI): ...@@ -168,6 +179,15 @@ async def lifespan(app: FastAPI):
pass pass
logger.info("定时任务调度引擎已停止") logger.info("定时任务调度引擎已停止")
# 停止僵尸执行看门狗
if watchdog_task:
watchdog_task.cancel()
try:
await watchdog_task
except asyncio.CancelledError:
pass
logger.info("僵尸执行看门狗已停止")
# 创建 FastAPI 应用实例 # 创建 FastAPI 应用实例
app = FastAPI( app = FastAPI(
......
...@@ -15,6 +15,7 @@ import subprocess ...@@ -15,6 +15,7 @@ import subprocess
import sys import sys
import os import os
import json import json
import time
import threading import threading
from pathlib import Path from pathlib import Path
from typing import Optional, List, Tuple, Dict, Any, Callable from typing import Optional, List, Tuple, Dict, Any, Callable
...@@ -125,6 +126,224 @@ def clear_cancel_requested(execution_id: str) -> None: ...@@ -125,6 +126,224 @@ def clear_cancel_requested(execution_id: str) -> None:
_cancel_requested.discard(execution_id) _cancel_requested.discard(execution_id)
# ==================== 僵尸执行看门狗 ====================
# 启动时 recover_interrupted_executions 只兜「进程重启」这一种卡死场景:
# 若容器未重启、但某条执行的 Playwright 线程在运行中异常退出(进程崩溃/线程
# 撕裂/浏览器挂起),DB 记录会永远停在 running——此时内存锁仍被占用,
# 定时任务每 30s 轮询都会被全局锁/DB 守卫挡住而空跳,直到有人手动 cancel。
# 已在 5.44 复现:一条 running 记录挡死调度 29 小时。
# 看门狗在进程常驻期间周期性扫描,只要运行中执行【持续没有进展】,就视为僵尸,
# 标记 failed 并释放内存锁,让调度恢复。
#
# 判断「有进展」的两个信号(任一存在即视为活着):
# 1. 内存心跳:_watchdog_heartbeat[execution_id] = 最近一次用例完成写入的时间。
# 仅由工作线程(run_all_cases_sync 每条用例完成时)登记;观察者绝不回写,
# 否则僵尸执行会在「首次观察」时被误刷新为活跃。
# 2. 数据库结果数增长:passed+failed+skipped 计数相对上次扫描有增长。
# 观察者初始化 last_count=-1,首次观测到存量结果不算增长(避免刚启动的
# 看门狗把已死的执行误判为活跃——这正是 5.44 事件的形态)。
# 两信号在 WATCHDOG_STALE_SECONDS 内都无变化 → 判定僵尸。
# security 执行是纯 requests 无浏览器、且可能由多窗口手动触发,不纳入监控。
WATCHDOG_INTERVAL_SECONDS = 60 # 看门狗扫描周期
WATCHDOG_STALE_SECONDS = 1800 # 判定僵尸的静默阈值(30 分钟)
WATCHDOG_RUNNING_FLOOR_SECONDS = 300 # 运行中执行至少存活这么久才开始判罚
_watchdog_lock = threading.Lock()
_watchdog_heartbeat: Dict[str, float] = {}
# 每轮扫描时各执行已观察到的「已完成结果数」,用于跨进程兜底判定结果是否在增长
_watchdog_last_result_count: Dict[str, int] = {}
def touch_execution_heartbeat(execution_id: str) -> None:
"""
记录某条执行最近一次有真实进展的时间。
由 run_all_cases_sync 在每条用例执行完成时调用(工作线程侧)。
注意:取消检查子模块不在这里(heartbeat 只表示「有进展」);
执行完成/取消会由 finally 清理,此处仅作活性信号。
Args:
execution_id (str): 执行记录ID
"""
with _watchdog_lock:
_watchdog_heartbeat[execution_id] = time.time()
def _watchdog_clear_heartbeat(execution_id: str) -> None:
"""执行结束(completed/cancelled/failed)后清理心跳,避免字典无限增长。"""
with _watchdog_lock:
_watchdog_heartbeat.pop(execution_id, None)
_watchdog_last_result_count.pop(execution_id, None)
def _watchdog_last_beat(execution_id: str) -> Optional[float]:
"""读取某执行最近一次心跳(线程安全)。"""
with _watchdog_lock:
return _watchdog_heartbeat.get(execution_id)
async def watchdog_scan_once(
stale_seconds: int = WATCHDOG_STALE_SECONDS,
running_floor_seconds: int = WATCHDOG_RUNNING_FLOOR_SECONDS,
) -> Tuple[int, int]:
"""
看门狗单轮扫描:把「持续无进展」的 running/pending UI 执行标记为僵尸。
Args:
stale_seconds: 静默阈值。execution 记录超过此秒数没有任何进展视为僵尸。
running_floor_seconds: 运行中执行至少存活至此秒数才开始判罚,
避免误伤刚创建/刚启动、首用例加载较慢的执行。
Returns:
Tuple[int, int]: (标记为僵尸的执行数, 受影响用例结果数)
"""
now = datetime.now()
now_ts = time.time()
async with async_session_maker() as db:
result = await db.execute(
select(Execution).where(
Execution.case_type == "ui",
Execution.status.in_(["running", "pending"]),
).with_for_update()
)
candidates = list(result.scalars().all())
stale_ids: list[str] = []
for e in candidates:
started = e.start_time or e.created_at
if not started:
continue
if e.case_type != "ui":
continue # 防御:security 等类型不纳入监控
age = (now - started).total_seconds()
# 执行在不同阶段用不同判罚起点:pending 未启动即算空转;
# running 需要存活时间 + 静默时间叠加,避免误判慢启动用例。
if e.status == "pending" and age >= stale_seconds:
# 长时间未进入 running → 僵尸
stale_ids.append(e.id)
continue
if e.status == "running":
if age < running_floor_seconds:
continue # 启动窗口期,不判罚
# 信号 2(先查并记录基线):DB 结果计数相对上次扫描的增长。
# 心跳活跃的扫描也要记录基线,否则心跳过期后首次比对时
# 增长信号会丢失(基线缺失被当作「首次观测」)。
cnt_q = await db.execute(
select(func.count(CaseResult.id)).where(
CaseResult.execution_id == e.id,
CaseResult.status.in_(["passed", "failed", "skipped"]),
)
)
result_count = cnt_q.scalar_one() or 0
with _watchdog_lock:
last_count = _watchdog_last_result_count.get(e.id, -1)
_watchdog_last_result_count[e.id] = result_count
if result_count > last_count >= 0:
continue # 计数在增长 → 有真实产出,活跃
# 信号 1:内存心跳(由工作线程在用例/步骤完成时登记)。
# 心跳缺失(如进程重启后恢复的执行)以 start_time 为静默起点。
beat = _watchdog_last_beat(e.id)
if beat is None:
beat = started.timestamp()
if (now_ts - beat) < stale_seconds:
continue
stale_ids.append(e.id)
if not stale_ids:
return 0, 0
now_dt = datetime.now()
await db.execute(
update(Execution)
.where(Execution.id.in_(stale_ids))
.values(
status="failed",
end_time=now_dt,
duration=0,
error_message=(
"执行长时间无进展,看门狗判定执行线程异常退出,自动标记为失败"
),
)
)
case_result_update = await db.execute(
update(CaseResult)
.where(
CaseResult.execution_id.in_(stale_ids),
CaseResult.status.in_(["pending", "running"]),
)
.values(
status="failed",
error_message="执行长时间无进展,看门狗判定执行线程异常退出",
end_time=now_dt,
)
)
await db.commit()
# 清理心跳并释放内存锁(若锁仍指向僵尸执行)
for eid in stale_ids:
_watchdog_clear_heartbeat(eid)
lock_id = _running_execution_id
if lock_id == eid:
set_running_execution(None)
logger.warning(
f"[看门狗] 标记 {len(stale_ids)} 条无进展执行为 failed"
f"(关联 {case_result_update.rowcount} 条用例结果)"
)
return len(stale_ids), case_result_update.rowcount
async def watchdog_loop(
interval: int = WATCHDOG_INTERVAL_SECONDS,
stale_seconds: int = WATCHDOG_STALE_SECONDS,
running_floor_seconds: int = WATCHDOG_RUNNING_FLOOR_SECONDS,
) -> None:
"""
后台看门狗主循环:周期性扫描僵尸执行。
与调度循环独立(各自 try/except),任何一轮异常不影响后续轮次。
Args:
interval: 扫描间隔(秒)
stale_seconds: 静默阈值(秒)
running_floor_seconds: 运行中执行存活下限(秒)
"""
logger.info(
f"[看门狗] 已启动:间隔 {interval}s,静默阈值 {stale_seconds}s"
f",运行存活下限 {running_floor_seconds}s"
)
while True:
try:
recovered, affected = await watchdog_scan_once(
stale_seconds=stale_seconds,
running_floor_seconds=running_floor_seconds,
)
if recovered:
logger.warning(
f"[看门狗] 本轮恢复 {recovered} 条僵尸执行"
f"(用例结果 {affected} 条标记失败)"
)
except asyncio.CancelledError:
logger.info("[看门狗] 已停止")
raise
except Exception as e:
logger.error(f"[看门狗] 扫描异常: {e}")
await asyncio.sleep(interval)
def start_watchdog() -> asyncio.Task:
"""
启动看门狗后台任务(供 lifespan 调用)。
Returns:
asyncio.Task: 看门狗任务句柄(用于关闭时取消)
"""
return asyncio.create_task(watchdog_loop())
async def recover_interrupted_executions( async def recover_interrupted_executions(
max_age_seconds: int = 300, max_age_seconds: int = 300,
) -> Tuple[int, int]: ) -> Tuple[int, int]:
...@@ -570,6 +789,9 @@ class ExecutionService: ...@@ -570,6 +789,9 @@ class ExecutionService:
except Exception as e: except Exception as e:
logger.error(f"✗ 用例结果回调提交失败: {str(e)}") logger.error(f"✗ 用例结果回调提交失败: {str(e)}")
# ★ 看门狗心跳:记录本执行最近一次真实进展(结果已写库)
touch_execution_heartbeat(execution_id)
logger.info(f"[run_all_cases_sync] 所有用例执行完成, 结果数: {len(results)}") logger.info(f"[run_all_cases_sync] 所有用例执行完成, 结果数: {len(results)}")
return results return results
except Exception as e: except Exception as e:
...@@ -639,6 +861,9 @@ class ExecutionService: ...@@ -639,6 +861,9 @@ class ExecutionService:
# 释放执行锁 # 释放执行锁
set_running_execution(None) set_running_execution(None)
# 清理看门狗心跳(执行已结束,避免心跳字典残留导致误判新执行)
_watchdog_clear_heartbeat(execution_id)
# 读取最新执行状态(避免使用 commit 后的过期实例): # 读取最新执行状态(避免使用 commit 后的过期实例):
# 执行期间可能已被用户取消,若已取消则保持 cancelled,不被覆盖为 completed。 # 执行期间可能已被用户取消,若已取消则保持 cancelled,不被覆盖为 completed。
# 必须用【列查询(标量)+ FOR UPDATE 当前读】: # 必须用【列查询(标量)+ FOR UPDATE 当前读】:
......
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
模块名称:test_watchdog_stale_execution.py
模块描述:僵尸执行看门狗 单元测试
覆盖 watchdog_scan_once:
- 运行中执行心跳已过期且结果数无增长 → 标记 failed + 关联用例结果 failed
- 心跳较新(有进展)→ 活跃,不标记
- pending 执行超时无进展 → 标记(未来事件循环里永远不会有 heartbeat)
- 启动窗口期内(age < running_floor)的 running → 不标记
- 结果数相比上次扫描有增长 → 即使心跳缺失也视为活跃(跨进程兜底)
- security 执行不纳入监控
作者:czj
创建日期:2026-08-29
"""
import asyncio
import time
from datetime import timedelta
from unittest.mock import AsyncMock, MagicMock, patch
from app.models.execution import Execution
from app.models.case_result import CaseResult
from app.services.execution_service import (
start_watchdog,
touch_execution_heartbeat,
watchdog_scan_once,
)
def run(coro):
"""跨平台运行异步协程(与项目其他测试一致)"""
return asyncio.new_event_loop().run_until_complete(coro)
def _mk_exec(eid, status, start_delta, case_type="ui"):
"""构造一个 Execution 对象;start_time 相对 now 偏移 start_delta。"""
e = Execution(id=eid, status=status, case_type=case_type)
e.start_time = __import__("datetime").datetime.now() + start_delta
return e
def _updates_run(session):
"""收集 session 上所有 UPDATE 的 SQL 语句。"""
return [
str(c.args[0]).upper()
for c in session.execute.call_args_list
if str(c.args[0]).lstrip().upper().startswith("UPDATE")
]
class _SessionCtx:
"""模仿 async_session_maker 的 __aenter__/__aexit__ 上下文,固定返回 session"""
def __init__(self, session):
self.session = session
async def __aenter__(self):
return self.session
async def __aexit__(self, *exc):
return False
def _make_session(execs, result_count=0, count_history=()):
"""构造 AsyncMock session:execute 依次返回 execs 查询、count 查询、update、case update。"""
session = AsyncMock()
exec_result = MagicMock()
exec_result.scalars.return_value.all.return_value = execs
cnt_result = MagicMock()
cnt_result.scalar_one.return_value = result_count
upd = MagicMock()
upd.rowcount = len(execs)
cupd = MagicMock()
cupd.rowcount = len(execs)
calls = [exec_result, cnt_result]
calls.extend(count_history) # 每多一次 count 查询
calls += [upd, cupd]
session.execute.side_effect = calls
return session
def _patch_sessionmaker(session):
cm = AsyncMock()
cm.__aenter__ = AsyncMock(return_value=session)
cm.__aexit__ = AsyncMock(return_value=False)
maker = MagicMock()
maker.return_value = cm
return patch("app.services.execution_service.async_session_maker", maker)
class TestWatchdogScanOnce:
"""watchdog_scan_once 单轮扫描"""
def test_running_stale_heartbeat_marks_failed(self):
"""运行中执行:心跳过期 + 结果数无增长 → 标记 failed"""
stale = _mk_exec("e_zombie", "running", timedelta(hours=-2))
# 心跳设为 1 小时前(无进展),首次扫描 result_count=0(-1 初始化不算增长)
with patch("app.services.execution_service._watchdog_heartbeat", {"e_zombie": time.time() - 3600}):
session = _make_session([stale], result_count=0)
with _patch_sessionmaker(session):
recovered, affected = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 1
assert affected == 1
# UPDATE 被执行:executions → failed
update_sqls = _updates_run(session)
assert len(update_sqls) == 2
assert "UPDATE EXECUTIONS" in update_sqls[0]
assert "UPDATE CASE_RESULTS" in update_sqls[1]
def test_running_recent_heartbeat_not_affected(self):
"""运行中执行:心跳较新 → 活跃,不标记"""
alive = _mk_exec("e_alive", "running", timedelta(minutes=-40))
with patch("app.services.execution_service._watchdog_heartbeat", {"e_alive": time.time() - 30}):
session = _make_session([alive], result_count=0)
with _patch_sessionmaker(session):
recovered, affected = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 0
assert affected == 0
def test_pending_stale_marks_failed(self):
"""pending 执行:超时无进展(永远不会有 heartbeat)→ 标记"""
stale_pending = _mk_exec("e_pend", "pending", timedelta(hours=-2))
session = _make_session([stale_pending], result_count=0)
with _patch_sessionmaker(session):
recovered, affected = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 1
assert affected == 1
def test_running_within_floor_not_affected(self):
"""运行中执行:启动窗口期内(age < floor)→ 不标记"""
young = _mk_exec("e_young", "running", timedelta(minutes=-1))
session = _make_session([young], result_count=0)
with _patch_sessionmaker(session):
recovered, affected = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 0
assert affected == 0
def test_result_count_growth_keeps_alive(self):
"""结果数相比上次扫描有增长 → 即使心跳缺失也视为活跃(跨进程兜底)"""
growing = _mk_exec("e_grow", "running", timedelta(hours=-2))
# 第一次扫描:存量结果 5,last_count 初始 -1 → 不算增长,但无心跳+start 已超
# 阈值 → 应判定僵尸(这就是 5.44 事件的形态:进程重启后恢复的执行,存量结果
# 不代表还在跑)
with patch("app.services.execution_service._watchdog_heartbeat", {}):
session1 = _make_session([growing], result_count=5)
with _patch_sessionmaker(session1):
recovered, _ = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 1 # 存量结果不算「增长」→ 僵尸
def test_result_count_observes_growth_across_scans(self):
"""两次扫描之间结果数有增长 → 活跃,不标记(进程仍在写库)"""
growing = _mk_exec("e_grow2", "running", timedelta(minutes=-40))
# 首次:存量 0(fresh),无心跳 → 判定僵尸会吗?start 仅 40min 前 < 30min 阈值?
# running_floor=300s=5min 已过;静默阈值 30min;无心跳+start 40min 前 → 已超
# 30min 静默 → 僵尸。为测「增长」场景,这里让它先活跃一次:显式注册心跳
from app.services import execution_service as svc
svc._watchdog_last_result_count.clear()
svc._watchdog_heartbeat.clear()
svc.touch_execution_heartbeat("e_grow2")
# 第一次扫描:心跳活跃 → 不标记
session1 = _make_session([growing], result_count=5)
with _patch_sessionmaker(session1):
recovered, _ = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 0
# 心跳过期(假设线程死掉),但结果数从 5 → 9(其他进程在写)→ 活跃
with patch("app.services.execution_service._watchdog_heartbeat", {"e_grow2": time.time() - 3600}):
session2 = _make_session([growing], result_count=9)
with _patch_sessionmaker(session2):
recovered, _ = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 0 # 9 > 5 → 增长,活跃
# 之后结果数不再增长(9 → 9),心跳又过期 → 僵尸
with patch("app.services.execution_service._watchdog_heartbeat", {"e_grow2": time.time() - 3600}):
session3 = _make_session([growing], result_count=9)
with _patch_sessionmaker(session3):
recovered, _ = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 1
def test_security_not_monitored(self):
"""security 执行不纳入看门狗"""
sec = _mk_exec("e_sec", "running", timedelta(hours=-10), case_type="security")
session = _make_session([sec], result_count=0)
with _patch_sessionmaker(session):
recovered, affected = run(watchdog_scan_once(
stale_seconds=1800, running_floor_seconds=300
))
assert recovered == 0
assert affected == 0
class TestWatchdogLoop:
"""看门狗主循环"""
def test_start_watchdog_returns_task(self):
"""start_watchdog 返回可取消的 asyncio.Task"""
async def _inner():
task = start_watchdog()
assert isinstance(task, asyncio.Task)
task.cancel()
try:
await task
except asyncio.CancelledError:
return "cancelled"
return "done"
assert run(_inner()) == "cancelled"
class TestHeartbeatHelpers:
"""心跳登记/清理"""
def test_touch_and_clear(self):
from app.services import execution_service as svc
svc._watchdog_heartbeat.clear()
svc._watchdog_last_result_count.clear()
touch_execution_heartbeat("exec_x")
assert "exec_x" in svc._watchdog_heartbeat
assert svc._watchdog_last_beat("exec_x") is not None
svc._watchdog_clear_heartbeat("exec_x")
assert "exec_x" not in svc._watchdog_heartbeat
assert svc._watchdog_last_beat("exec_x") is None
\ No newline at end of file
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论