diff --git a/services/tasks.py b/services/tasks.py index ce88bb6..913e422 100644 --- a/services/tasks.py +++ b/services/tasks.py @@ -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 diff --git a/tests/test_monitor_jobs.py b/tests/test_monitor_jobs.py new file mode 100644 index 0000000..d1a3b17 --- /dev/null +++ b/tests/test_monitor_jobs.py @@ -0,0 +1,464 @@ +# -*- coding: utf-8 -*- +"""任务监控循环 7 组周期性作业的行为契约（沙箱，不触碰生产 data/）。 + +为什么：``_task_monitor_loop`` 拆分前是 170 行 / 圈复杂度 57 的单函数，7 组互不 +相关的作业共用一个 try/except。这种「按作业拆函数」的改动最容易引入两类损伤： + + 1. **异常兜底粒度被改变**——某项作业失败后，本轮其余作业还跑不跑； + 2. **锁内/锁外边界被挪动**——文件删除搬回锁内会重新阻塞全站任务端点 + （原注释记载「GB 级任务目录删除曾阻塞锁数十秒」）。 + +本文件按「调用后世界里发生了什么」逐条钉住可观察行为，断言不引用实现细节， +故可跨重构复用。 + +账本口径同 tests/test_zip_workers.py：一律走 ``reserve_and_open`` 真预扣。 +""" +import os +import threading +import time +from typing import Any + +import pytest + +import auth_system as auth +import core.db as core_db +from core.config import ( + CANCEL_AFTER_SECONDS, + PAUSE_AFTER_SECONDS, + RESULT_RETENTION_SECONDS, +) +from core.state import ( + DEVICE_LAST_SEEN, + DEVICE_LAST_SEEN_LOCK, + DOWNLOAD_TASKS, + DOWNLOAD_TASKS_LOCK, +) +from services import tasks as services_tasks +from services._shared import ( + _LEDGER_LAST_CLEANUP, + _LEDGER_LAST_RECOVER, + _MONITOR_ERR_TS, + _TASKS_SWEEP_LAST, +) +from services.api_service import api_service + +PW = 'Passw0rd12' +CREDITS = 50 + +# 与服务的硬上限/阈值一致（测试内不引用私有常量，避免「测试跟着实现改」） +_HARD_TIMEOUT = 4 * 3600 +_TASKS_ORPHAN_AGE = 25 * 3600 +_LEDGER_ORPHAN_MAX_AGE = 5 * 3600 +_LEDGER_SETTLED_RETENTION = 90 * 86400 + + +class _NullLogger: + """吞掉 api_service.logger 输出，避免测试往项目根 music_api.log 追加。""" + + def debug(self, *a: Any, **k: Any) -> None: ... + def info(self, *a: Any, **k: Any) -> None: ... + def warning(self, *a: Any, **k: Any) -> None: ... + def error(self, *a: Any, **k: Any) -> None: ... + + +@pytest.fixture() +def sandbox(tmp_path, monkeypatch): + """临时站点库 + 临时下载目录 + 空任务表 + 复位监控节流全局。""" + db_path = tmp_path / 'site.db' + monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(auth.db, 'DB_PATH', db_path) + monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups') + monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key') + monkeypatch.setattr(core_db, 'DATA_DIR', tmp_path) + monkeypatch.setattr(core_db, 'DB_PATH', db_path) + auth.init_db() + auth.create_user('payer', PW, email='p@example.com', credits=CREDITS) + + downloads = tmp_path / 'downloads' + downloads.mkdir() + monkeypatch.setattr(api_service, 'downloads_path', downloads) + monkeypatch.setattr(api_service, 'logger', _NullLogger()) + + # 监控节流全局是模块级 dict，逐用例复位——否则用例间互相影响 + for state in (_LEDGER_LAST_RECOVER, _LEDGER_LAST_CLEANUP, _TASKS_SWEEP_LAST): + state['ts'] = 0.0 + _MONITOR_ERR_TS['ts'] = 0.0 + _MONITOR_ERR_TS['n'] = 0 + + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS.clear() + with DEVICE_LAST_SEEN_LOCK: + DEVICE_LAST_SEEN.clear() + yield {'root': tmp_path, 'downloads': downloads} + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS.clear() + with DEVICE_LAST_SEEN_LOCK: + DEVICE_LAST_SEEN.clear() + + +def _add_task(tid: str, **over: Any) -> dict[str, Any]: + """登记一条任务记录（不预扣——监控作业只读单条任务的内部字段）。""" + task: dict[str, Any] = { + 'type': 'batch', 'task_id': tid, 'status': 'queued', + 'start': time.time(), 'msg': '排队中...', + 'cancel': threading.Event(), 'pause': threading.Event(), + 'device_id': 'dev-test', 'result': None, + } + task.update(over) + with DOWNLOAD_TASKS_LOCK: + DOWNLOAD_TASKS[tid] = task + return task + + +def _has(tid: str) -> bool: + with DOWNLOAD_TASKS_LOCK: + return tid in DOWNLOAD_TASKS + + +# ============ ① 终态任务回收 ============ + +def test_reap_removes_expired_task_and_files(sandbox): + """超过保留期：记录、交付文件与任务目录一并回收。""" + downloads = sandbox['downloads'] + delivered = downloads / 'out.zip' + delivered.write_bytes(b'z') + task_dir = downloads / 'tasks' / 't-done' + task_dir.mkdir(parents=True) + (task_dir / 'inner.mp3').write_bytes(b'a') + + _add_task('t-done', status='done', + completed_at=time.time() - RESULT_RETENTION_SECONDS - 60, + result={'zip_path': str(delivered), 'file_path': str(task_dir / 'inner.mp3')}) + + services_tasks._reap_finished_tasks(time.time()) + + assert not _has('t-done'), '过期终态任务记录应被移除' + assert not delivered.exists(), '交付文件应被删除' + assert not task_dir.exists(), '任务目录应被递归删除' + + +def test_reap_keeps_task_within_retention(sandbox): + """未到保留期不动（保留期从 completed_at 起算）。""" + _add_task('t-fresh', status='done', completed_at=time.time()) + services_tasks._reap_finished_tasks(time.time()) + assert _has('t-fresh') + + +def test_reap_keeps_legacy_record_without_completed_at(sandbox): + """无 completed_at 的旧记录回退到 start 计算（兼容既有在途记录）。""" + _add_task('t-legacy', status='done', start=time.time(), completed_at=None) + services_tasks._reap_finished_tasks(time.time()) + assert _has('t-legacy'), 'start 很新 → 不应被回收' + + +def test_reap_ignores_non_terminal_tasks(sandbox): + """未终态任务不归它管（哪怕 start 很久——那是超时取消的职责）。""" + _add_task('t-run', status='running', start=time.time() - 10 * _HARD_TIMEOUT) + services_tasks._reap_finished_tasks(time.time()) + assert _has('t-run') + + +def test_reap_retries_then_gives_up(sandbox): + """交付文件删不掉：记 _cleanup_tries 保留记录下轮重试，达上限后放弃记录。""" + # 用**目录**冒充交付文件：os.remove 必然抛 OSError + stuck = sandbox['downloads'] / 'stuck-dir' + stuck.mkdir() + _add_task('t-stuck', status='done', + completed_at=time.time() - RESULT_RETENTION_SECONDS - 60, + result={'file_path': str(stuck)}) + + now = time.time() + services_tasks._reap_finished_tasks(now) + assert _has('t-stuck'), '删除失败应保留记录待下轮重试' + with DOWNLOAD_TASKS_LOCK: + assert DOWNLOAD_TASKS['t-stuck']['_cleanup_tries'] == 1 + + for _ in range(services_tasks._CLEANUP_MAX_TRIES): + services_tasks._reap_finished_tasks(now) + assert not _has('t-stuck'), '连续失败达上限后应放弃记录（文件留给缓存清扫兜底）' + + +# ============ ② 超时任务取消 ============ + +def _finished_thread() -> threading.Thread: + """已结束的线程替身——对应生产里 worker 崩溃/退出的情形（is_alive() 为 False）。""" + th = threading.Thread(target=lambda: None) + th.start() + th.join() + return th + + +def test_cancel_overtime_settles_dead_worker(sandbox, monkeypatch): + """超硬上限且 worker 线程已死：置取消、唤醒 cancel Event、代为退款。""" + settled: list[tuple] = [] + monkeypatch.setattr(auth, 'ledger_settle', + lambda tid, ok, st: settled.append((tid, ok, st))) + + task = _add_task('t-slow', status='running', + start=time.time() - _HARD_TIMEOUT - 60, + worker_thread=_finished_thread()) + services_tasks._cancel_overtime_tasks(time.time()) + + assert task['status'] == 'canceled' + assert task['cancel'].is_set(), '须唤醒 cancel Event 让阻塞中的 worker 退出' + assert settled == [('t-slow', 0, 'timeout')] + + +def test_cancel_overtime_without_worker_thread_does_not_settle(sandbox, monkeypatch): + """无 worker_thread 字段的任务不代结算——代结算的前提是「线程曾存在且已死」， + 没有线程句柄就无法判断 worker 是否已在别处结算过，贸然退款会双退。""" + settled: list[tuple] = [] + monkeypatch.setattr(auth, 'ledger_settle', + lambda tid, ok, st: settled.append((tid, ok, st))) + + task = _add_task('t-noth', status='queued', + start=time.time() - _HARD_TIMEOUT - 60) + services_tasks._cancel_overtime_tasks(time.time()) + + assert task['status'] == 'canceled' + assert settled == [] + + +def test_cancel_overtime_skips_live_worker(sandbox, monkeypatch): + """worker 仍在跑：只置取消，不代结算（交 worker 自己收敛，避免双退）。""" + settled: list[tuple] = [] + monkeypatch.setattr(auth, 'ledger_settle', + lambda tid, ok, st: settled.append((tid, ok, st))) + + stop = threading.Event() + alive = threading.Thread(target=stop.wait, daemon=True) + alive.start() + try: + task = _add_task('t-live', status='running', + start=time.time() - _HARD_TIMEOUT - 60, worker_thread=alive) + services_tasks._cancel_overtime_tasks(time.time()) + assert task['status'] == 'canceled' + assert settled == [], 'worker 活着 → 不应代结算' + finally: + stop.set() + alive.join(timeout=5) + + +def test_cancel_overtime_ignores_recent(sandbox): + """未超硬上限的任务不动。""" + task = _add_task('t-new', status='running', start=time.time() - 60) + services_tasks._cancel_overtime_tasks(time.time()) + assert task['status'] == 'running' + + +# ============ ⑤ 账本周期兜底 ============ + +def test_ledger_jobs_respect_intervals(sandbox, monkeypatch): + """时间戳门槛：到期执行一次并推进门槛，未到期不重复执行。""" + calls: list[str] = [] + monkeypatch.setattr(auth, 'ledger_recover_orphans', + lambda max_age_seconds=0: calls.append(f'recover:{max_age_seconds}')) + monkeypatch.setattr(auth, 'ledger_cleanup_settled', + lambda days: calls.append(f'cleanup:{days}')) + + now = time.time() + services_tasks._run_periodic_ledger_jobs(now) + assert calls == [f'recover:{_LEDGER_ORPHAN_MAX_AGE}', + f'cleanup:{_LEDGER_SETTLED_RETENTION}'] + + calls.clear() + services_tasks._run_periodic_ledger_jobs(now + 1) + assert calls == [], '门槛已推进，同一周期内不应重复执行' + + calls.clear() + services_tasks._run_periodic_ledger_jobs(now + 61) + assert calls == [f'recover:{_LEDGER_ORPHAN_MAX_AGE}'], '仅到期的那个触发' + + +def test_ledger_jobs_swallow_exception(sandbox, monkeypatch): + """账本作业抛错由自身吞掉（不能让本轮其余作业一起落空）。""" + def boom(max_age_seconds=0): + raise RuntimeError('库忙') + + monkeypatch.setattr(auth, 'ledger_recover_orphans', boom) + services_tasks._run_periodic_ledger_jobs(time.time()) # 不抛即通过 + + +# ============ ⑥ tasks/ 孤儿目录补扫 ============ + +def test_sweep_orphans_removes_stale_only(sandbox): + """>24h 且无任务记录的目录被回收；在册任务目录与新鲜目录保留。""" + tasks_root = sandbox['downloads'] / 'tasks' + tasks_root.mkdir(parents=True) + orphan = tasks_root / 'orphan' + orphan.mkdir() + in_registry = tasks_root / 'in-registry' + in_registry.mkdir() + fresh = tasks_root / 'fresh' + fresh.mkdir() + + old = time.time() - _TASKS_ORPHAN_AGE + os.utime(orphan, (old, old)) + os.utime(in_registry, (old, old)) + + _add_task('in-registry', status='running') + + services_tasks._sweep_orphan_task_dirs(time.time()) + + assert not orphan.exists(), '无记录的陈旧目录应被回收' + assert in_registry.exists(), '在册任务目录不得被补扫误删' + assert fresh.exists(), '未到年龄的目录保留' + + +def test_sweep_orphans_honours_interval(sandbox): + """未到补扫周期的调用直接跳过。""" + tasks_root = sandbox['downloads'] / 'tasks' + tasks_root.mkdir(parents=True) + first = tasks_root / 'orphan-a' + first.mkdir() + old = time.time() - _TASKS_ORPHAN_AGE + os.utime(first, (old, old)) + + now = time.time() + services_tasks._sweep_orphan_task_dirs(now) + assert not first.exists(), '首次到期应执行' + + second = tasks_root / 'orphan-b' + second.mkdir() + os.utime(second, (old, old)) + services_tasks._sweep_orphan_task_dirs(now + 1) + assert second.exists(), '门槛未到 → 本轮不应补扫' + + +# ============ ⑦ 心跳离线处理 ============ + +def test_offline_device_cancels_after_threshold(sandbox): + """离线超 30 分钟：其未完成任务置取消并唤醒 cancel Event。""" + now = time.time() + with DEVICE_LAST_SEEN_LOCK: + DEVICE_LAST_SEEN['dev-off'] = now - CANCEL_AFTER_SECONDS - 60 + task = _add_task('t-off', status='running', device_id='dev-off') + + services_tasks._handle_offline_devices(now) + + assert task['status'] == 'canceled' + assert task['cancel'].is_set() + assert '30 分钟' in task['msg'] + + +def test_offline_device_pauses_batch_before_threshold(sandbox): + """离线刚过暂停阈值：批量任务先暂停，让出排队名额（「断线不占坑」）。""" + now = time.time() + with DEVICE_LAST_SEEN_LOCK: + DEVICE_LAST_SEEN['dev-soft'] = now - PAUSE_AFTER_SECONDS - 1 + task = _add_task('t-soft', status='running', type='batch', device_id='dev-soft') + + services_tasks._handle_offline_devices(now) + + assert task['status'] == 'paused' + assert task['pause'].is_set() + assert not task['cancel'].is_set(), '未到取消阈值不应发取消' + + +def test_offline_handler_ignores_online_device(sandbox): + """心跳新鲜的设备：其任务不受影响。""" + now = time.time() + with DEVICE_LAST_SEEN_LOCK: + DEVICE_LAST_SEEN['dev-on'] = now + task = _add_task('t-on', status='running', device_id='dev-on') + + services_tasks._handle_offline_devices(now) + + assert task['status'] == 'running' + + +# ============ 异常限频 ============ + +def test_log_monitor_error_rate_limits(sandbox, monkeypatch): + """限频：窗口内多次异常只记一条（不静默也不刷屏）。""" + events: list[str] = [] + + class _Recorder: + def warning(self, msg: str, *a: Any, **k: Any) -> None: + events.append(msg) + + monkeypatch.setattr(api_service, 'logger', _Recorder()) + + for _ in range(3): + services_tasks._log_monitor_error() + assert len(events) == 1, '同一窗口内只应记录一条' + + _MONITOR_ERR_TS['ts'] = time.time() - services_tasks._MONITOR_ERR_WINDOW - 1 + services_tasks._log_monitor_error() + assert len(events) == 2, '窗口过期后应重新放行' + + +# ============ 单轮编排与兜底粒度 ============ + +_JOBS = ( + '_reap_finished_tasks', '_cancel_overtime_tasks', '_sweep_download_cache', + '_sweep_rate_buckets', '_run_periodic_ledger_jobs', '_sweep_orphan_task_dirs', + '_handle_offline_devices', +) + + +def test_tick_runs_all_jobs_in_order(monkeypatch): + """单轮按固定顺序执行全部 7 步——顺序即优先级，重构不得打乱。""" + order: list[str] = [] + for name in _JOBS: + monkeypatch.setattr( + services_tasks, name, (lambda n: lambda *a, **k: order.append(n))(name)) + + services_tasks._run_monitor_tick(time.time()) + + assert order == list(_JOBS) + + +def test_tick_exception_skips_rest_of_tick(monkeypatch): + """前段作业抛异常 → 本轮剩余作业跳过。 + + 如实钉住拆分前的粒度：整个 tick 共用一个 try/except，而非逐作业隔离。 + （改为逐作业隔离属行为变更，需单独评估——本测试存在的意义正是让那次 + 变更有据可依、不会悄悄发生。） + """ + called: list[str] = [] + + def boom(_now: float) -> None: + raise RuntimeError('reap 炸了') + + monkeypatch.setattr(services_tasks, '_reap_finished_tasks', boom) + monkeypatch.setattr(services_tasks, '_handle_offline_devices', + lambda _now: called.append('offline')) + + with pytest.raises(RuntimeError): + services_tasks._run_monitor_tick(time.time()) + assert called == [], '本轮后续作业随异常一起中断（保持原语义）' + + +class _StopLoop(Exception): + """从假 sleep 里抛出，把无限循环的单轮执行切断用于断言。""" + + +def test_loop_survives_tick_exception(monkeypatch): + """循环本身不退出：单轮异常被限频记录后继续跑下一轮。""" + state = {'sleeps': 0, 'logged': 0} + + class _FakeTime: + @staticmethod + def sleep(_seconds: float) -> None: + state['sleeps'] += 1 + if state['sleeps'] >= 2: + raise _StopLoop + + @staticmethod + def time() -> float: + return time.time() + + def boom(_now: float) -> None: + raise RuntimeError('tick 炸了') + + monkeypatch.setattr(services_tasks, 'time', _FakeTime) + monkeypatch.setattr(services_tasks, '_run_monitor_tick', boom) + monkeypatch.setattr(services_tasks, '_log_monitor_error', + lambda: state.__setitem__('logged', state['logged'] + 1)) + + with pytest.raises(_StopLoop): + services_tasks._task_monitor_loop() + + assert state['sleeps'] == 2, '第 1 轮异常后应继续到第 2 轮' + assert state['logged'] == 1, '每轮异常记一次'