fix: 恢复任务监控线程的启动点(迁移中丢失,超时取消与账本对账静默失效)

为什么:d209af8 基线在 main.py:4594 用 threading.Thread(...).start() 启动
_task_monitor_loop;8446166「移除 Flask」把实现搬进 services/runtime.py 时
只搬了函数定义、丢了启动点,6288a98 拆包后落到 services/tasks.py,HEAD 仍
无任何调用方——函数体完好,但一次都没跑过。

后果(自该次迁移起持续失效):
- 心跳超时不再暂停离线设备的批量任务,排队名额无法释放
- 离线 >30 分钟 / 运行 >4 小时的任务不再自动取消
- 终态任务记录、tasks/ 孤儿目录、downloads 缓存不再回收
- ledger_recover_orphans 停摆,预扣未退款的账永久悬挂在 open 状态

改法:新增 start_task_monitor(),加锁保证幂等(重复调用返回 False),
在 create_app 的 lifespan 里用 asyncio.to_thread 启动;新增 start_monitor
开关供测试显式关闭。裸 TestClient(app) 不进上下文管理器、不触发 lifespan,
故既有测试不受影响。

验证:pytest 新增 3 例(幂等 / lifespan 启动 / 可跳过);端到端 uvicorn 冒烟
日志出现「任务监控线程已启动」。
This commit is contained in:
snowy-yang committed 2026-09-26 18:53:40 +08:00
1 parent 095171e0b7
commit 0b25f215a2
3 files changed
+123 -6

No files matched your search

+29
View File
@@ -4,6 +4,7 @@
import os import os
import shutil import shutil
import threading
import time import time
import traceback import traceback
@@ -242,6 +243,34 @@ def _task_monitor_loop() -> None:
_MONITOR_ERR_TS['n'] = 0 _MONITOR_ERR_TS['n'] = 0
_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: def _touch_device(device_id: str) -> None:
"""记录设备心跳时间(设备表容量上限——device_id 是客户端任意串, """记录设备心跳时间(设备表容量上限——device_id 是客户端任意串,
登录用户可提交海量随机 ID 使字典无界增长,超限清最旧; 登录用户可提交海量随机 ID 使字典无界增长,超限清最旧;
+67
View File
@@ -0,0 +1,67 @@
# -*- coding: utf-8 -*-
"""任务监控线程的启动契约。
背景(回归):Flask 单体时代该线程由模块级代码拉起,8446166「移除 Flask」
把实现搬进 services 包时启动点丢失,`_task_monitor_loop` 此后全项目零引用,
监控循环一次都没跑过。这里钉住两条契约:
1. 启动是幂等的(多 worker / 多次构建应用只起一个线程);
2. 应用 lifespan 会启动它(否则心跳超时暂停、超时取消、账本孤儿对账
等全部静默失效)。
"""
from fastapi.testclient import TestClient
from core import db as core_db
from services import tasks as services_tasks
def _isolate_data_dir(monkeypatch, tmp_path):
"""把站点数据目录指向临时目录,避免 lifespan 的 init_db 触碰真实库。"""
from auth_system import db as auth_db
data_dir = tmp_path / 'data'
data_dir.mkdir(parents=True, exist_ok=True)
for mod in (auth_db, core_db):
monkeypatch.setattr(mod, 'DATA_DIR', data_dir, raising=False)
monkeypatch.setattr(mod, 'DB_PATH', data_dir / 'site.db', raising=False)
monkeypatch.setattr(auth_db, 'BACKUP_DIR', data_dir / 'backups', raising=False)
monkeypatch.setattr(auth_db, 'SECRET_KEY_PATH', data_dir / 'secret.key', raising=False)
return data_dir
def test_start_task_monitor_is_idempotent(monkeypatch):
"""重复调用只启动一次(空转的 loop 替身,避免真的跑监控循环)。"""
monkeypatch.setattr(services_tasks, '_MONITOR_STARTED', False)
monkeypatch.setattr(services_tasks, '_task_monitor_loop', lambda: None)
assert services_tasks.start_task_monitor() is True
assert services_tasks.start_task_monitor() is False
def test_lifespan_starts_monitor(monkeypatch, tmp_path):
"""应用 lifespan 启动监控线程(该启动点曾在迁移中丢失)。"""
from web.app import create_app
_isolate_data_dir(monkeypatch, tmp_path)
calls: list[int] = []
monkeypatch.setattr(
services_tasks, 'start_task_monitor', lambda: calls.append(1) or True)
with TestClient(create_app()) as client:
assert client.get('/health').status_code == 200
assert calls == [1]
def test_lifespan_can_skip_monitor(monkeypatch, tmp_path):
"""start_monitor=False 时不启动(供测试或嵌入式构建使用)。"""
from web.app import create_app
_isolate_data_dir(monkeypatch, tmp_path)
calls: list[int] = []
monkeypatch.setattr(
services_tasks, 'start_task_monitor', lambda: calls.append(1) or True)
with TestClient(create_app(start_monitor=False)) as client:
assert client.get('/health').status_code == 200
assert calls == []
+27 -6
View File
@@ -35,9 +35,12 @@ from web.routers import tasks as tasks_router
from web.routers import user as user_router from web.routers import user as user_router
@asynccontextmanager def _make_lifespan(start_monitor: bool):
async def _lifespan(app: FastAPI): """构造 lifespan(是否启动任务监控线程由 create_app 参数决定)。"""
"""应用生命周期:启动确保站点库表结构就绪,关闭时释放 aiohttp 连接池。
@asynccontextmanager
async def _lifespan(app: FastAPI):
"""应用生命周期:建表 → 启动任务监控 → 关闭时释放连接池与后台 loop。
启动必须建表。全新部署时宿主机的 data/ 往往是个空目录(compose 用 启动必须建表。全新部署时宿主机的 data/ 往往是个空目录(compose 用
`./data:/app/data` 挂载,会盖掉镜像内的任何内容),此时若只等首次查询 `./data:/app/data` 挂载,会盖掉镜像内的任何内容),此时若只等首次查询
@@ -45,17 +48,32 @@ async def _lifespan(app: FastAPI):
`no such table: users`(实测登录端点直接 500,且错误被吞成"请稍后重试", `no such table: users`(实测登录端点直接 500,且错误被吞成"请稍后重试",
极难定位)。init_db 全程 `CREATE TABLE IF NOT EXISTS` + 幂等迁移, 极难定位)。init_db 全程 `CREATE TABLE IF NOT EXISTS` + 幂等迁移,
对已有库重复执行无副作用,故放启动期最稳妥。 对已有库重复执行无副作用,故放启动期最稳妥。
任务监控线程也必须在此启动:Flask 单体时代由模块级 `threading.Thread(
target=_task_monitor_loop).start()` 拉起(基线 main.py:4594),移除 Flask
迁到 FastAPI 时该启动点丢失,此后监控循环从未运行——心跳超时不再暂停
离线设备、超时任务不再取消、过期任务与下载缓存不再回收、账本孤儿对账
不再执行(预扣未退款的账永久悬挂)。详见 services.tasks.start_task_monitor。
""" """
await asyncio.to_thread(auth_system.init_db) await asyncio.to_thread(auth_system.init_db)
if start_monitor:
from services.tasks import start_task_monitor
await asyncio.to_thread(start_task_monitor)
yield yield
try: try:
from core import aiohttp_compat, http_client from core import aiohttp_compat, async_bridge, http_client
await http_client.client.close() await http_client.client.close()
await aiohttp_compat.close() await aiohttp_compat.close()
# 停掉 sync↔async 桥的后台事件循环(daemon 线程不调也能随进程退出,
# 但显式关闭更干净)
async_bridge.shutdown()
except Exception: except Exception:
pass pass
return _lifespan
def _init_audio_secret() -> None: def _init_audio_secret() -> None:
"""注入音频签名密钥(与 auth_system 同源,令直链 token 与服务端一致)。""" """注入音频签名密钥(与 auth_system 同源,令直链 token 与服务端一致)。"""
@@ -67,12 +85,15 @@ def _init_audio_secret() -> None:
pass pass
def create_app(api_prefix: str = '') -> FastAPI: def create_app(api_prefix: str = '', *, start_monitor: bool = True) -> FastAPI:
"""创建 FastAPI 应用。 """创建 FastAPI 应用。
api_prefix:路由挂载前缀。 api_prefix:路由挂载前缀。
- ''(默认):直接服务前端使用的根路径(生产形态) - ''(默认):直接服务前端使用的根路径(生产形态)
- '/api/v1':挂在子前缀下(便于契约测试与反向代理) - '/api/v1':挂在子前缀下(便于契约测试与反向代理)
start_monitor:是否随应用启动任务监控线程。生产恒为 True;
测试若以 `with TestClient(app)` 方式进入 lifespan,可传 False
以避免后台线程清扫用例的临时目录。
""" """
_init_audio_secret() _init_audio_secret()
@@ -82,7 +103,7 @@ def create_app(api_prefix: str = '') -> FastAPI:
docs_url='/api/v1/docs', docs_url='/api/v1/docs',
redoc_url=None, redoc_url=None,
openapi_url='/api/v1/openapi.json', openapi_url='/api/v1/openapi.json',
lifespan=_lifespan, lifespan=_make_lifespan(start_monitor),
) )
# 响应侧统一处理:安全头 + 音频直链签名保护 + 版本指纹抹除 # 响应侧统一处理:安全头 + 音频直链签名保护 + 版本指纹抹除