Archived
snowy-yang
65a81af94a
为什么：_task_monitor_loop 是一个 170 行 / 圈复杂度 58 / 嵌套 8 层的单函数，把
7 组互不相关的周期性作业塞在一起——终态回收、超时取消、缓存清扫、限流桶清理、
账本对账、孤儿目录补扫、心跳离线——整轮共用一个 try/except。任何一处改动都要
通读全部 7 组逻辑；上一轮恢复它的启动点时，也正是这个体量让判读很吃力。
改法：一组作业一个具名函数，另抽 _run_monitor_tick 只负责编排顺序（循环本身
只剩「sleep + try + 限频记录」三件事）。
复杂度（同口径 AST 实测，重构前后）：
_task_monitor_loop 170 行 / 圈 58 / 嵌套 8 → 17 行 / 圈 3 / 嵌套 2
文件内最大函数 170 行 → 66 行（_reap_finished_tasks）
文件内 >80 行函数 1 → 0 · 嵌套 ≥5 层 1 → 0
代价：总行数 231 → 291（+60，全部是新的函数签名与 docstring）
等价性证据（重构的分内之事，不是附赠）：
1. AST **数值字面量多重集**比对：旧 16 种 / 新 16 种，计数零差异——确认搬运
中没有改任何阈值（4h 硬超时 / 8 次重试 / 60s / 24h / 5h / 90 天 / 600s …）。
2. 剔除 docstring 后的**字符串字面量**比对：旧版有的一处不少——确认用户可见
状态文案与日志格式没被抄错。
3. 新增 20 例行为契约测试（tests/test_monitor_jobs.py），按「调用后世界里发生
了什么」断言，不引用实现细节，故可跨后续重构复用。
刻意保留的语义（是决定，不是疏漏）：
- 单个 tick 仍共用一个 try/except：前段作业抛异常会跳过本轮剩余作业。改为
逐作业隔离属于**行为变更**，本提交不夹带；已用 test_tick_exception_skips_
rest_of_tick 如实钉住现状，让那次变更有据可依、不会悄悄发生。
- _run_periodic_ledger_jobs / _sweep_orphan_task_dirs 的局部 try/except 原样
保留（原注释即写明「下轮重试」意图），其兜底粒度各有独立用例。
- 代结算的前提仍是「worker_thread 字段存在且线程已死」：没有线程句柄的任务
不退款——拿不到句柄就无法判断 worker 是否已在别处结算过，贸然退会双退。
该约束由新增用例固化（原实现如此，此前无测试记录）。
验证：ruff 全过 · pyright 0/0 · pytest 209 passed（189→209）·
生产 data/ 未被触碰（全部用例走 tmp_path 沙箱）。
391 lines
16 KiB
Python
391 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""任务监控循环与设备心跳。
|
||
|
||
监控循环原是一个 170 行 / 圈复杂度 58 / 嵌套 8 层的单函数，内含 7 组互不相关
|
||
的周期性作业。此处按「一组作业一个具名函数」拆开：改超时取消时不必通读账本
|
||
对账与缓存清扫，每项作业的异常兜底粒度也一目了然（见各自 docstring）。
|
||
"""
|
||
|
||
import os
|
||
import shutil
|
||
import threading
|
||
import time
|
||
import traceback
|
||
|
||
import auth_system
|
||
from core.config import (
|
||
CANCEL_AFTER_SECONDS,
|
||
HEARTBEAT_INTERVAL,
|
||
PAUSE_AFTER_SECONDS,
|
||
RESULT_RETENTION_SECONDS,
|
||
)
|
||
from core.state import (
|
||
DEVICE_LAST_SEEN,
|
||
DEVICE_LAST_SEEN_LOCK,
|
||
DOWNLOAD_TASKS,
|
||
DOWNLOAD_TASKS_LOCK,
|
||
)
|
||
|
||
from ._shared import (
|
||
_LEDGER_LAST_CLEANUP,
|
||
_LEDGER_LAST_RECOVER,
|
||
_MONITOR_ERR_TS,
|
||
_TASKS_SWEEP_LAST,
|
||
_sweep_rate_buckets,
|
||
)
|
||
from ._taskcore import (
|
||
_task_files,
|
||
)
|
||
from .api_service import (
|
||
api_service,
|
||
)
|
||
|
||
# ---- 监控循环各项作业的周期与阈值（集中一处，避免魔数散落在循环体里）----
|
||
|
||
# 任务硬超时：运行/排队/暂停超过该时长仍未结束 → 置取消
|
||
_TASK_HARD_TIMEOUT = 4 * 3600
|
||
|
||
# 终态任务结果文件清理的最大重试轮数（文件被占用时下轮再试）
|
||
_CLEANUP_MAX_TRIES = 8
|
||
|
||
# 账本周期兜底：孤儿对账每 60s、settled 行清理每 24h
|
||
_LEDGER_RECOVER_INTERVAL = 60
|
||
_LEDGER_CLEANUP_INTERVAL = 24 * 3600
|
||
|
||
# 孤儿账退回阈值：严格大于任务 4h 硬超时 + 监控 tick 抖动——账本行 created_at
|
||
# 整秒截断使账龄恒 ≥ 任务龄，旧的「4h 相等」口径存在抢退合法 4h+ 长任务的窗口
|
||
_LEDGER_ORPHAN_MAX_AGE = 5 * 3600
|
||
|
||
# settled 账本行保留 90 天后清理（防长运行进程下表无界增长）
|
||
_LEDGER_SETTLED_RETENTION = 90 * 86400
|
||
|
||
# tasks/ 孤儿目录补扫：每小时一次，仅回收 >24h 且无任务记录的目录
|
||
_TASKS_SWEEP_INTERVAL = 3600
|
||
_TASKS_ORPHAN_MAX_AGE = 24 * 3600
|
||
|
||
# 监控异常限频窗口（秒）：持续异常不静默，也不每 tick 刷屏
|
||
_MONITOR_ERR_WINDOW = 600
|
||
|
||
|
||
def _quiesce_tasks_for_restore() -> None:
|
||
"""恢复数据库前取消全部在途任务（worker 不再引用旧库用户结算）"""
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
for task in DOWNLOAD_TASKS.values():
|
||
if task.get('status') in ('queued', 'running', 'paused'):
|
||
task['status'] = 'canceled'
|
||
task['msg'] = '数据库恢复中，任务已取消（费用按实际交付结算）'
|
||
task['completed_at'] = time.time()
|
||
try:
|
||
task['cancel'].set()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
# 数据库恢复前静默在途任务
|
||
auth_system.register_before_restore_hook(_quiesce_tasks_for_restore)
|
||
|
||
|
||
def _sweep_download_cache(max_age_seconds: int = RESULT_RETENTION_SECONDS) -> None:
|
||
"""下载缓存清扫（配套：批量任务不再即时删除共享缓存，改由这里按年龄回收）。
|
||
只清扫 downloads 根目录的普通文件，不碰 tasks/ 子目录（那是任务专属交付物）。"""
|
||
try:
|
||
root = api_service.downloads_path
|
||
now = time.time()
|
||
for p in root.iterdir():
|
||
try:
|
||
if p.is_file() and now - p.stat().st_mtime > max_age_seconds:
|
||
p.unlink()
|
||
except OSError:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _reap_finished_tasks(now: float) -> None:
|
||
"""终态任务回收：按 completed_at 保留 RESULT_RETENTION_SECONDS 后删记录与文件。
|
||
|
||
锁内只收集过期清单，文件删除与 rmtree 移到锁外——GB 级任务目录的删除曾
|
||
阻塞任务锁数十秒，期间全站进度上报/提交/取消/领取全部停摆。
|
||
|
||
单条删除失败保留记录下轮重试，最多 _CLEANUP_MAX_TRIES 轮；超限后放弃记录
|
||
但大声记日志，文件留给 _sweep_download_cache 按年龄兜底。
|
||
"""
|
||
expired: list[tuple] = []
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
for tid, task in DOWNLOAD_TASKS.items():
|
||
if task.get('status') not in ('done', 'error', 'canceled'):
|
||
continue
|
||
done_at = task.get('completed_at') or 0
|
||
# 兼容无 completed_at 的旧记录：回退到 start
|
||
if not done_at:
|
||
done_at = task.get('start', 0)
|
||
if now - done_at <= RESULT_RETENTION_SECONDS:
|
||
continue
|
||
expired.append((tid, _task_files(task),
|
||
api_service.downloads_path / 'tasks' / tid))
|
||
if not expired:
|
||
return
|
||
|
||
cleanup_results: list[tuple] = []
|
||
for tid, files, task_dir in expired:
|
||
all_removed = True
|
||
for f in files:
|
||
try:
|
||
os.remove(f)
|
||
except FileNotFoundError:
|
||
pass
|
||
except OSError:
|
||
all_removed = False # 文件暂时不可删（被占用等）
|
||
if task_dir.exists():
|
||
shutil.rmtree(str(task_dir), ignore_errors=True)
|
||
if task_dir.exists():
|
||
all_removed = False
|
||
cleanup_results.append((tid, all_removed))
|
||
|
||
finished_done: list[str] = []
|
||
finished_retry: list[str] = []
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
for tid, all_removed in cleanup_results:
|
||
task = DOWNLOAD_TASKS.get(tid)
|
||
if task is None:
|
||
finished_done.append(tid) # 锁外期间已被其他路径移除
|
||
continue
|
||
if all_removed:
|
||
finished_done.append(tid)
|
||
else:
|
||
task['_cleanup_tries'] = int(task.get('_cleanup_tries') or 0) + 1
|
||
if task['_cleanup_tries'] >= _CLEANUP_MAX_TRIES:
|
||
# 重试多次仍失败：放弃记录但大声记录日志
|
||
api_service.logger.warning(
|
||
f"任务 {tid} 结果文件多次清理失败，已放弃（文件可能被占用）")
|
||
finished_done.append(tid)
|
||
else:
|
||
finished_retry.append(tid)
|
||
for tid in finished_done:
|
||
DOWNLOAD_TASKS.pop(tid, None)
|
||
if finished_done:
|
||
api_service.logger.info(f"监控线程清理过期终态任务: {finished_done}")
|
||
if finished_retry:
|
||
api_service.logger.info(f"监控线程保留清理失败任务待重试: {finished_retry}")
|
||
|
||
|
||
def _cancel_overtime_tasks(now: float) -> None:
|
||
"""运行/排队/暂停超过硬上限仍未结束的任务 → 置取消并唤醒 cancel Event。
|
||
|
||
worker 线程已死则无人再走终态结算，监控代为按账本退款——ledger_settle 幂等，
|
||
worker 若已结算过此处返回 0，无双退。结算移到任务锁外（同 _mark_task_terminal
|
||
口径）：库忙 15s 不再随任务锁波及全站任务端点。
|
||
"""
|
||
timeout_settle: list[str] = []
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
for tid, task in DOWNLOAD_TASKS.items():
|
||
if (task.get('status') in ('queued', 'running', 'paused')
|
||
and now - task.get('start', 0) > _TASK_HARD_TIMEOUT):
|
||
task['status'] = 'canceled'
|
||
task['msg'] = '任务超时未完成，已自动取消（费用按实际交付结算）'
|
||
task['completed_at'] = time.time()
|
||
try:
|
||
task['cancel'].set()
|
||
except Exception:
|
||
pass
|
||
# worker 线程已死则无人再走终态结算——监控代为按账本结算退款
|
||
th = task.get('worker_thread')
|
||
if th is not None and not th.is_alive():
|
||
timeout_settle.append(tid)
|
||
api_service.logger.warning(f"任务超时自动取消: {tid}")
|
||
for tid in timeout_settle:
|
||
try:
|
||
auth_system.ledger_settle(tid, 0, 'timeout')
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _run_periodic_ledger_jobs(now: float) -> None:
|
||
"""账本周期兜底（原实现仅启动执行一次）。
|
||
|
||
- 孤儿对账：open 且账龄 > _LEDGER_ORPHAN_MAX_AGE 的行每分钟检查并退回
|
||
- settled 行清理：每小时低频执行，防长运行进程下表无界增长
|
||
|
||
用「时间戳门槛」而非「tick 取模」——单轮工作段超过周期时取模落点会被整段
|
||
跳过，时间戳门槛只延迟不漏跑。
|
||
|
||
自带 try/except：账本库忙不该让本轮后续作业（孤儿目录补扫、心跳离线处理）
|
||
一起落空。
|
||
"""
|
||
try:
|
||
if now - _LEDGER_LAST_RECOVER['ts'] > _LEDGER_RECOVER_INTERVAL:
|
||
_LEDGER_LAST_RECOVER['ts'] = now
|
||
auth_system.ledger_recover_orphans(max_age_seconds=_LEDGER_ORPHAN_MAX_AGE)
|
||
if now - _LEDGER_LAST_CLEANUP['ts'] > _LEDGER_CLEANUP_INTERVAL:
|
||
_LEDGER_LAST_CLEANUP['ts'] = now
|
||
auth_system.ledger_cleanup_settled(_LEDGER_SETTLED_RETENTION)
|
||
except Exception as e:
|
||
api_service.logger.warning(f"账本周期兜底异常（下轮重试）: {e}")
|
||
|
||
|
||
def _sweep_orphan_task_dirs(now: float) -> None:
|
||
"""tasks/ 孤儿目录每小时低频补扫：>24h 且无任务记录才移除。
|
||
|
||
与 _reap_finished_tasks 互补——那条路径由任务记录驱动，记录已丢的目录
|
||
（进程曾在删除前崩溃等）只能靠这里按年龄兜底。
|
||
|
||
自带 try/except：同 _run_periodic_ledger_jobs，单条竞态与整体异常都不该
|
||
打断本轮后续作业。
|
||
"""
|
||
try:
|
||
if now - _TASKS_SWEEP_LAST['ts'] <= _TASKS_SWEEP_INTERVAL:
|
||
return
|
||
_TASKS_SWEEP_LAST['ts'] = now
|
||
tasks_root = api_service.downloads_path / 'tasks'
|
||
if not tasks_root.exists():
|
||
return
|
||
for d in tasks_root.iterdir():
|
||
try:
|
||
if (d.is_dir() and d.name not in DOWNLOAD_TASKS
|
||
and now - d.stat().st_mtime > _TASKS_ORPHAN_MAX_AGE):
|
||
shutil.rmtree(str(d), ignore_errors=True)
|
||
api_service.logger.info(f"tasks 孤儿目录补扫移除: {d.name}")
|
||
except FileNotFoundError:
|
||
pass # 单条竞态不中止整轮扫描
|
||
except OSError:
|
||
pass
|
||
except Exception as e:
|
||
api_service.logger.warning(f"tasks 孤儿目录补扫异常（下轮重试）: {e}")
|
||
|
||
|
||
def _handle_offline_devices(now: float) -> None:
|
||
"""心跳离线处理（两档，阈值见 core.config）。
|
||
|
||
- 离线 > CANCEL_AFTER_SECONDS（30 分钟）：取消该设备所有未完成任务。文件删除
|
||
与退款统一交给 worker 的取消收尾/终态清理；worker 已不在时由启动对账按账本兜底。
|
||
- 离线 > PAUSE_AFTER_SECONDS：批量任务先置暂停，让出排队名额给下一位
|
||
（「断线不占坑」）。
|
||
"""
|
||
with DEVICE_LAST_SEEN_LOCK:
|
||
offline = {d: now - ts for d, ts in DEVICE_LAST_SEEN.items()
|
||
if now - ts > PAUSE_AFTER_SECONDS}
|
||
if not offline:
|
||
return
|
||
with DOWNLOAD_TASKS_LOCK:
|
||
for tid, task in DOWNLOAD_TASKS.items():
|
||
dev = task.get('device_id')
|
||
gap = offline.get(dev) if dev else None
|
||
if not gap:
|
||
continue
|
||
st = task.get('status')
|
||
if gap > CANCEL_AFTER_SECONDS and st in ('queued', 'running', 'paused'):
|
||
# 离线超 30 分钟：请求取消（文件删除与退款统一交给 worker 的
|
||
# 取消收尾/终态清理；worker 已不在时由启动对账按账本兜底）
|
||
task['status'] = 'canceled'
|
||
task['msg'] = '设备离线超过 30 分钟，任务已自动取消'
|
||
task['completed_at'] = time.time()
|
||
try:
|
||
task['cancel'].set()
|
||
except Exception:
|
||
pass
|
||
api_service.logger.info(f"设备离线超 30 分钟自动请求取消: {tid}")
|
||
elif task.get('type') == 'batch' and st in ('queued', 'running'):
|
||
# 心跳超时：先暂停，让出排队名额给下一位
|
||
task['status'] = 'paused'
|
||
task['msg'] = '设备离线，任务暂停（重新打开页面自动恢复）'
|
||
task['pause'].set()
|
||
|
||
|
||
def _log_monitor_error() -> None:
|
||
"""监控循环异常限频记录。
|
||
|
||
持续异常不能完全静默（否则监控失能无人知晓），也不能每 tick 刷屏：每
|
||
_MONITOR_ERR_WINDOW 秒最多一条，附堆栈尾部。
|
||
|
||
必须在 except 块内**直接**调用——traceback.format_exc() 取的是当前活跃异常，
|
||
转手排队/异步回调后再调用会拿到 NoneType 堆栈。
|
||
"""
|
||
now = time.time()
|
||
_MONITOR_ERR_TS['n'] += 1
|
||
if now - _MONITOR_ERR_TS['ts'] > _MONITOR_ERR_WINDOW:
|
||
api_service.logger.warning(
|
||
f"任务监控循环异常（近 10 分钟累计 {_MONITOR_ERR_TS['n']} 次，"
|
||
f"最后异常见堆栈）: {traceback.format_exc()[-600:]}")
|
||
_MONITOR_ERR_TS['ts'] = now
|
||
_MONITOR_ERR_TS['n'] = 0
|
||
|
||
|
||
def _run_monitor_tick(now: float) -> None:
|
||
"""单轮监控作业（7 步，顺序即优先级）。
|
||
|
||
先回收已终态任务腾出名额，再处理超时，最后处理离线设备——顺序与拆分前一致。
|
||
|
||
⚠️ 参数 ``now`` 由调用方传入（其余作业内部各自取 time.time()）：
|
||
_reap_finished_tasks 与 _cancel_overtime_tasks 共用同一轮的时间基准，避免
|
||
同一轮里两次取时导致边界任务在两处判断不一致。
|
||
"""
|
||
_reap_finished_tasks(now)
|
||
_cancel_overtime_tasks(now)
|
||
_sweep_download_cache(RESULT_RETENTION_SECONDS)
|
||
_sweep_rate_buckets()
|
||
_run_periodic_ledger_jobs(now)
|
||
_sweep_orphan_task_dirs(now)
|
||
_handle_offline_devices(now)
|
||
|
||
|
||
def _task_monitor_loop() -> None:
|
||
"""任务监控循环：每 HEARTBEAT_INTERVAL 跑一轮 _run_monitor_tick。
|
||
|
||
单轮里任何一步抛异常都由 _log_monitor_error 限频记录，循环本身不退出
|
||
（保持拆分前的兜底粒度：拆分前整个函数体共用一个 try/except）。
|
||
注意 _run_periodic_ledger_jobs 与 _sweep_orphan_task_dirs 自带局部
|
||
try/except，它们失败不会中断本轮其余作业。
|
||
|
||
本函数曾因迁移丢失启动入口（全项目零引用、一次都没跑过），现由
|
||
start_task_monitor() 在应用 lifespan 中拉起——见该函数 docstring。
|
||
"""
|
||
while True:
|
||
time.sleep(HEARTBEAT_INTERVAL)
|
||
try:
|
||
_run_monitor_tick(time.time())
|
||
except Exception:
|
||
_log_monitor_error()
|
||
|
||
|
||
_MONITOR_STARTED = False
|
||
_MONITOR_START_LOCK = threading.Lock()
|
||
|
||
|
||
def start_task_monitor() -> bool:
|
||
"""启动任务监控线程（幂等）。返回本次是否真正启动了线程。
|
||
|
||
为什么需要显式启动：Flask 单体时代由模块级 `threading.Thread(target=
|
||
_task_monitor_loop, daemon=True).start()` 拉起（基线 main.py:4594），
|
||
8446166「移除 Flask」把实现搬进 services 包时**启动点丢失**，此后
|
||
`_task_monitor_loop` 全项目零引用，监控循环一次都没跑过。后果：
|
||
心跳超时不再暂停离线设备的批量任务；离线 >30 分钟 / 运行 >4 小时的任务
|
||
不再自动取消；终态任务记录与 tasks/ 目录、downloads 缓存不再回收；
|
||
`ledger_recover_orphans` 不再执行——预扣未退款的账永久悬挂在 open 状态。
|
||
现改由应用 lifespan 调用（见 web.app._make_lifespan），恢复设计意图。
|
||
|
||
幂等：多 worker 或测试多次构建应用时只启动一个线程。
|
||
"""
|
||
global _MONITOR_STARTED
|
||
with _MONITOR_START_LOCK:
|
||
if _MONITOR_STARTED:
|
||
return False
|
||
threading.Thread(target=_task_monitor_loop, name='task-monitor', daemon=True).start()
|
||
_MONITOR_STARTED = True
|
||
api_service.logger.info('任务监控线程已启动')
|
||
return True
|
||
|
||
|
||
def _touch_device(device_id: str) -> None:
|
||
"""记录设备心跳时间（设备表容量上限——device_id 是客户端任意串，
|
||
登录用户可提交海量随机 ID 使字典无界增长，超限清最旧；
|
||
入口已统一钳 64，此处兜底再钳一次防新调用点绕过）"""
|
||
device_id = str(device_id or '')[:64]
|
||
if not device_id:
|
||
return
|
||
with DEVICE_LAST_SEEN_LOCK:
|
||
if len(DEVICE_LAST_SEEN) >= 5000:
|
||
# 保留最近心跳的设备：按最后心跳时间排序，砍掉最旧的一半
|
||
for d in sorted(DEVICE_LAST_SEEN, key=lambda k: DEVICE_LAST_SEEN.get(k, 0.0))[:2500]:
|
||
DEVICE_LAST_SEEN.pop(d, None)
|
||
DEVICE_LAST_SEEN[device_id] = time.time()
|