hhhge/NeteaseUrl
Archived
1
0 Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: 拆分任务监控循环的 7 组周期作业（170 行 / 圈 58 → 17 行 / 圈 3）

Browse Source 
为什么：_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 沙箱）。
This commit is contained in:
snowy-yang committed 2026-09-26 19:04:36 +08:00
1 parent 3de75f7479
commit 65a81af94a
2 files changed
+732 -164
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

services/tasks.py
+268 -164
View File
@@ -1,5 +1,9 @@
# -*- coding: utf-8 -*-
"""任务监控循环与设备心跳。
监控循环原是一个 170 行 / 圈复杂度 58 / 嵌套 8 层的单函数，内含 7 组互不相关
的周期性作业。此处按「一组作业一个具名函数」拆开：改超时取消时不必通读账本
对账与缓存清扫，每项作业的异常兜底粒度也一目了然（见各自 docstring）。
"""
import os
@@ -36,6 +40,32 @@ 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 不再引用旧库用户结算）"""
@@ -71,176 +101,250 @@ def _sweep_download_cache(max_age_seconds: int = RESULT_RETENTION_SECONDS) -> No
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:
"""后台监控：
- 心跳超时（40s）→ 暂停离线设备的批量任务，释放排队名额给下一位
- 心跳超时 > 30 分钟 → 自动取消该设备的所有未完成任务并清理文件
- 终态任务按 completed_at 保留 35 分钟（与前端「领取」入口对齐）
- 运行中/排队/暂停任务超过 4 小时仍未结束 → 置取消（cancel Event 一并置位）
- 下载缓存按年龄清扫（配套，同为 35 分钟口径）
- 顺带清理限流桶中的过期项（不再只依赖"新建桶时"的清理时机）
"""任务监控循环：每 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:
now = time.time()
# 终态任务清理（按完成时刻；删除失败保留记录下轮重试，最多 8 次）----
# 锁内只收集过期清单，文件删除移到锁外——GB 级任务目录
# 的删除曾阻塞锁数十秒，期间全站进度上报/提交/取消/领取全部停摆
finished_done: list[str] = []
finished_retry: list[str] = []
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))
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))
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'] >= 8:
# 重试多次仍失败：放弃记录但大声记录日志（文件留给缓存清扫兜底）
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}")
# ---- 长时间未结束的任务：置取消（含 cancel Event，唤醒可能的阻塞等待）----
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) > 4 * 3600):
task['status'] = 'canceled'
task['msg'] = '任务超时未完成，已自动取消（费用按实际交付结算）'
task['completed_at'] = time.time()
try:
task['cancel'].set()
except Exception:
pass
# worker 线程已死则无人再走终态结算——监控代为按账本
# 结算退款（settle 幂等：worker 若已结算过，此处返回 0）
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}")
# 结算移到任务锁外（同 _mark_task_terminal 口径）——
# 库忙 15s 不再随任务锁波及全站任务端点
for tid in timeout_settle:
try:
auth_system.ledger_settle(tid, 0, 'timeout')
except Exception:
pass
# ---- 下载缓存清扫（保留期同交付口径 35 分钟）+ 限流桶过期项顺手清理 ----
_sweep_download_cache(RESULT_RETENTION_SECONDS)
_sweep_rate_buckets()
# 账本周期兜底（原仅启动执行一次）----
# 孤儿对账：长运行进程内 open 且账龄 > 5h（严格大于任务 4h 硬超时 + 监控
# tick 抖动——账本行 created_at 整秒截断使账龄恒 ≥ 任务龄，旧 4h 相等口径
# 存在抢退合法 4h+ 长任务的窗口）的行每分钟周期检查退回；
# settled 行 90 天清理改为每小时低频执行，防长运行进程下表无界增长
try:
# 时间戳门槛替代 tick 取模——单轮工作段超过周期时
# 取模落点会被整段跳过，时间戳门槛只延迟不漏跑
if now - _LEDGER_LAST_RECOVER['ts'] > 60:
_LEDGER_LAST_RECOVER['ts'] = now
auth_system.ledger_recover_orphans(max_age_seconds=5 * 3600)
if now - _LEDGER_LAST_CLEANUP['ts'] > 24 * 3600:
_LEDGER_LAST_CLEANUP['ts'] = now
auth_system.ledger_cleanup_settled(90 * 86400)
except Exception as e:
api_service.logger.warning(f"账本周期兜底异常（下轮重试）: {e}")
# tasks/ 孤儿目录每小时低频补扫（>24h 且无记录）----
try:
if now - _TASKS_SWEEP_LAST['ts'] > 3600:
_TASKS_SWEEP_LAST['ts'] = now
tasks_root = api_service.downloads_path / 'tasks'
if tasks_root.exists():
for d in tasks_root.iterdir():
try:
if (d.is_dir() and d.name not in DOWNLOAD_TASKS
and now - d.stat().st_mtime > 24 * 3600):
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}")
# ---- 设备心跳离线处理 ----
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:
continue
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()
_run_monitor_tick(time.time())
except Exception:
# 持续异常不能完全静默（否则监控失能无人知晓）：限频记录，每 10 分钟最多一条
_now = time.time()
_MONITOR_ERR_TS['n'] += 1
if _now - _MONITOR_ERR_TS['ts'] > 600:
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
_log_monitor_error()
_MONITOR_STARTED = False