diff --git a/backend/app/api/mining.py b/backend/app/api/mining.py index a72b80b6..eedcb67c 100644 --- a/backend/app/api/mining.py +++ b/backend/app/api/mining.py @@ -154,6 +154,44 @@ class MiningAutoStartRequest(BaseModel): return self +def _require_mining_lock(request: Request) -> None: + """降级模式 (撞锁启动) 下拒绝一切挖掘启动入口, 指引到挖掘页锁状态卡。""" + if not getattr(request.app.state, "mining_lock_acquired", True): + raise HTTPException( + status_code=409, + detail="挖掘锁被其他进程占用, 当前后端以降级模式运行 — " + "请在挖掘页查看锁状态并强制释放后重试", + ) + + +@router.get("/lock") +def get_lock_status(request: Request) -> dict[str, Any]: + """挖掘锁诊断: 持有者线索 + 是否允许强制释放 (fail-closed 判定)。""" + from app.services.mining_lock_admin import lock_status + + return lock_status(request.app.state.repo.store.data_dir) + + +@router.post("/lock/force-release") +def force_release_lock(request: Request) -> dict[str, Any]: + """终止陈旧锁持有者并由当前后端接管; 不满足陈旧判定时返回拒绝原因。""" + from app.services.mining_lock_admin import ( + MiningLockAdminError, + force_release, + ) + + lock = getattr(request.app.state, "mining_process_lock", None) + if lock is None: + raise HTTPException(status_code=409, detail="挖掘锁状态不可用") + try: + result = force_release(request.app.state.repo.store.data_dir, lock) + except MiningLockAdminError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + # 接管成功后同步降级标志, 挖掘入口立即恢复可用 + request.app.state.mining_lock_acquired = True + return result + + @router.get("/availability") def get_availability( request: Request, @@ -200,6 +238,7 @@ def list_runs( @router.post("/runs") def start_run(payload: MiningStartRequest, request: Request) -> dict[str, Any]: + _require_mining_lock(request) manager = _manager(request) worker_request = payload.model_dump(mode="json", exclude={"force"}) try: @@ -279,6 +318,7 @@ def start_auto_run(payload: MiningAutoStartRequest, request: Request) -> dict[st 筛选结果随请求持久化 (request.auto_screening), 供结果页展示达标因子清单与 失败原因分布; 无达标因子时返回 started=false 而不是报错。 """ + _require_mining_lock(request) from app.services.auto_mining import screen_all_factors manager = _manager(request) diff --git a/backend/app/desktop.py b/backend/app/desktop.py index b654f06e..2b3c4f59 100644 --- a/backend/app/desktop.py +++ b/backend/app/desktop.py @@ -79,13 +79,12 @@ def _ensure_data_dir_writable() -> None: def _precheck_mining_lock() -> str | None: - """启动前预检挖掘锁。返回 None = 可启动; 返回 str = 冲突原因 (给失败页)。 + """启动前预检挖掘锁。返回 None = 无冲突; 返回 str = 冲突原因 (仅日志)。 - app.main 的 lifespan 第一件事就是抢 data_dir/.mining_process_lock, - 被别的后端实例 (panel 模式、隧道后端或残留进程) 占住时 lifespan 直接 - 抛 MiningProcessLockError, uvicorn 吞掉后主线程只能等满超时报笼统的 - "启动失败"。这里用同一把锁试抢一下, 提前把冲突翻译成用户可操作的指引。 - 试抢成功立即释放, 真正的持有仍由 lifespan 负责。 + app.main 的 lifespan 会抢 data_dir/.mining_process_lock; 被别的后端 + 实例 (panel 模式、隧道后端或残留进程) 占住时, 现在 lifespan 只降级 + (挖掘入口 409, 面板可用, 挖掘页提供强制释放), 这里提前留一条日志 + 方便排查。试抢成功立即释放, 真正的持有仍由 lifespan 负责。 """ from app.config import settings from app.services.mining_process_lock import ( @@ -399,20 +398,12 @@ def main() -> int: if not _acquire_single_instance(): return 0 - # 挖掘锁预检: 被其他后端实例占住时 lifespan 必败, 提前秒显失败页并给出 - # 操作指引, 而不是让用户对着超时后的笼统"启动失败"猜 (见 _precheck_mining_lock) + # 挖掘锁预检: 撞锁只降级不再拦截启动 — lifespan 会以降级模式起面板 + # (挖掘入口 409), 用户可在挖掘页查看持有者并强制释放陈旧持有者; + # 这里只提前留一条日志方便排查。 conflict = _precheck_mining_lock() if conflict is not None: - logger.error("启动预检: %s", conflict.replace("\n", " ")) - try: - _open_static_window(_failed_html(conflict)) - except Exception: - # 失败页自身也打不开 (如缺 webview 依赖) 时不能让异常裸穿 main(): - # console=False 下进程闪退无反馈, 且本路径在 try/finally 之外, - # 单实例锁会跳过释放 - logger.exception("启动失败页窗口打开失败") - _release_single_instance() - return 1 + logger.warning("启动预检: %s", conflict.replace("\n", " ")) try: port = _find_free_port(_BASE_PORT) diff --git a/backend/app/main.py b/backend/app/main.py index 57730cfc..794a22d2 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -500,18 +500,27 @@ async def _application_lifespan(app: FastAPI): @asynccontextmanager async def lifespan(app: FastAPI): mining_process_lock = MiningProcessLock(settings.data_dir) + lock_acquired = True try: mining_process_lock.acquire() except MiningProcessLockError: - # uvicorn 吞掉 lifespan 异常只留一行 startup failed; 桌面版失败页指向 - # data/desktop.log, 先把真因 (含 traceback) 落盘, 不依赖 uvicorn 的日志行为 - logger.error("挖掘锁被其他后端实例占用, lifespan 启动失败", exc_info=True) - raise + # 撞锁不再让整个后端起不来 (典型: 旧实例挂死在 shutdown 持锁不放): + # 面板以降级模式启动 — 挖掘启动入口返回 409, 用户可在挖掘页查看 + # 持有者并 (满足陈旧判定时) 强制释放, 之后自动恢复。 + lock_acquired = False + logger.warning( + "挖掘锁被其他进程占用, 面板以降级模式启动 (挖掘功能暂不可用; " + "可在挖掘页查看锁状态并强制释放)" + ) + app.state.mining_process_lock = mining_process_lock + app.state.mining_lock_acquired = lock_acquired try: async with _application_lifespan(app): yield finally: - mining_process_lock.release() + # 用 lock.acquired 而非启动时的 flag: 强制释放接管后同样要在退出时释放 + if mining_process_lock.acquired: + mining_process_lock.release() app = FastAPI( diff --git a/backend/app/services/lab_schedule.py b/backend/app/services/lab_schedule.py index 8ff2ad4e..e3394644 100644 --- a/backend/app/services/lab_schedule.py +++ b/backend/app/services/lab_schedule.py @@ -18,7 +18,13 @@ def run_scheduled_lab(app_state: Any) -> dict[str, Any]: results: dict[str, Any] = {"mining": None, "guardian": None} try: - results["mining"] = run_weekly_mining(app_state) + if not getattr(app_state, "mining_lock_acquired", True): + # 降级模式 (挖掘锁被其他进程占用): 跳过周度挖掘而不是报错, + # 守护复验不受影响 + logger.warning("scheduled lab mining skipped: mining lock unavailable") + results["mining"] = {"skipped": "mining lock unavailable"} + else: + results["mining"] = run_weekly_mining(app_state) except Exception: logger.exception("scheduled lab mining enqueue failed") try: diff --git a/backend/app/services/mining_lock_admin.py b/backend/app/services/mining_lock_admin.py new file mode 100644 index 00000000..9d239bec --- /dev/null +++ b/backend/app/services/mining_lock_admin.py @@ -0,0 +1,263 @@ +"""挖掘锁状态诊断与 fail-closed 强制释放。 + +背景: 应用进程在 lifespan 抢 `.mining_process.lock`; 当持有者挂死 (典型: +热重载撞上挖掘运行, 旧进程卡在 shutdown 等待收尾) 或 worker 成为孤儿时, +锁会被无限期占用, 后续实例全部无法获得挖掘能力。 + +设计纪律 (设计稿讨论结论): +- 只提供「人工确认的强制停止按钮」, 不做启动时自动清理 — 终止进程是 + 破坏性操作, 必须人来确认。 +- 强制释放的判定全部 fail-closed: 持有者不明 (旧格式锁)、pid 可能被 + 复用、存在进行中的挖掘运行、挖掘存储近期仍有写入 — 任一命中即拒绝, + 宁可让用户手动处理也不误杀健康实例。 +""" + +from __future__ import annotations + +import json +import logging +import os +import signal +import subprocess +import time +from datetime import datetime +from pathlib import Path +from typing import Any + +from app.services.mining_jobs import ACTIVE_RUN_STATUSES +from app.services.mining_process_lock import ( + MiningProcessLock, + MiningProcessLockError, + probe_lock_held, + read_lock_holder, +) + +logger = logging.getLogger(__name__) + +# 挖掘存储静默多久才允许判定持有者陈旧 (秒)。健康实例空闲时也会持锁, +# 该阈值用于排除「正在收尾/写 artifact」的正常窗口。 +STALE_AFTER_SECONDS = 600.0 +_TERMINATE_GRACE_SECONDS = 6.0 +_RELEASE_WAIT_SECONDS = 5.0 + + +class MiningLockAdminError(ValueError): + """强制释放被拒绝 (fail-closed), message 即拒绝原因。""" + + +def lock_status( + data_dir: Path | str, + *, + self_pid: int | None = None, + now: float | None = None, + stale_after: float = STALE_AFTER_SECONDS, +) -> dict[str, Any]: + """汇总挖掘锁状态: 是否被持有、持有者线索、是否允许强制释放。""" + pid = os.getpid() if self_pid is None else self_pid + held = probe_lock_held(data_dir) + holder = read_lock_holder(data_dir) + mine = held and bool(holder) and holder.get("pid") == pid + holder_pid = holder.get("pid") if holder else None + holder_alive = _pid_alive(holder_pid) if holder_pid else False + pid_consistent = ( + _pid_matches_start_time(holder_pid, holder.get("started_at")) + if holder_alive + else True + ) + active_runs = _count_active_runs(data_dir) + store_idle = _store_idle_seconds(data_dir, now=time.time() if now is None else now) + releasable, reason = _classify( + held=held, + mine=mine, + holder_known=holder is not None, + holder_alive=holder_alive, + pid_consistent=pid_consistent, + active_runs=active_runs, + store_idle=store_idle, + stale_after=stale_after, + ) + return { + "held": held, + "mine": mine, + "holder": holder, + "holder_alive": holder_alive, + "pid_consistent": pid_consistent, + "active_runs": active_runs, + "store_idle_seconds": ( + round(store_idle, 1) if store_idle is not None else None + ), + "stale_after_seconds": stale_after, + "force_releasable": releasable, + "reason": reason, + } + + +def force_release( + data_dir: Path | str, + lock: MiningProcessLock, + *, + stale_after: float = STALE_AFTER_SECONDS, +) -> dict[str, Any]: + """终止陈旧持有者并让本进程接管锁; 任何不满足条件的情况直接拒绝。""" + status = lock_status(data_dir, stale_after=stale_after) + if not status["force_releasable"]: + raise MiningLockAdminError(status["reason"] or "不可强制释放") + if lock.acquired: + raise MiningLockAdminError("当前进程已持有挖掘锁") + target_pid = int(status["holder"]["pid"]) + + logger.warning( + "强制释放挖掘锁: 终止持有者 pid=%s (持有于 %s, 存储静默 %ss)", + target_pid, + status["holder"].get("started_at"), + status["store_idle_seconds"], + ) + _terminate_pid(target_pid) + if not _wait_condition( + lambda: (not _pid_alive(target_pid)) or (not probe_lock_held(data_dir)), + _TERMINATE_GRACE_SECONDS + _RELEASE_WAIT_SECONDS, + ): + raise MiningLockAdminError( + f"持有者 pid={target_pid} 未能终止或锁未释放, 请手动处理" + ) + try: + lock.acquire() + except MiningProcessLockError as exc: + raise MiningLockAdminError( + "持有者已终止, 但锁仍被其他进程抢先获取; 请重试" + ) from exc + return { + "released": True, + "terminated_pid": target_pid, + "reacquired": True, + } + + +def _classify( + *, + held: bool, + mine: bool, + holder_known: bool, + holder_alive: bool, + pid_consistent: bool, + active_runs: int, + store_idle: float | None, + stale_after: float, +) -> tuple[bool, str | None]: + if not held: + return False, "锁未被持有" + if mine: + return False, "锁由当前后端持有" + if not holder_known: + return False, "锁文件为旧格式或不可读, 无法定位持有进程 — 请手动查找并结束它" + if not holder_alive: + # 内核会在进程退出时释放 flock, 理论上到不了这里; 防御性兜底 + return False, "持有者进程已不存在, 锁应已释放 — 请刷新状态" + if not pid_consistent: + return False, "持有者 pid 已被其他进程复用, 拒绝自动终止 — 请手动确认" + if active_runs: + return False, f"仍有 {active_runs} 个进行中的挖掘运行, 不能终止持有者" + if store_idle is not None and store_idle < stale_after: + return False, ( + f"挖掘存储 {store_idle:.0f} 秒内仍有写入 (陈旧阈值 {stale_after:.0f}s), " + "持有者可能在正常收尾 — 请稍后再试" + ) + return True, None + + +def _pid_alive(pid: int | None) -> bool: + if not pid or pid <= 0: + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + except OSError: + return False + return True + + +def _pid_matches_start_time(pid: int, started_at: Any) -> bool: + """POSIX 下比对进程实际启动时间与锁内记录, 排除 pid 复用误杀。 + + 无法获取 (Windows / ps 不可用 / 记录缺失) 时返回 True — + 该信号只是附加保险, 主判定靠「无活跃 run + 存储静默」。 + """ + if os.name != "posix" or not isinstance(started_at, str): + return True + try: + recorded = datetime.fromisoformat(started_at) + except ValueError: + return True + try: + output = subprocess.run( + ["ps", "-p", str(pid), "-o", "lstart="], + capture_output=True, + text=True, + timeout=5, + check=True, + ).stdout.strip() + actual = datetime.strptime(output, "%a %b %d %H:%M:%S %Y") + except (OSError, subprocess.SubprocessError, ValueError): + return True + if recorded.tzinfo is not None: + # ps lstart 是本地时间墙钟; 统一成同口径再比 + recorded = recorded.astimezone().replace(tzinfo=None) + # ps 的 lstart 无亚秒精度, 放宽到 120s + return abs((actual - recorded).total_seconds()) <= 120 + + +def _runs_dir(data_dir: Path | str) -> Path: + return Path(data_dir) / "research" / "mining" / "runs" + + +def _count_active_runs(data_dir: Path | str) -> int: + count = 0 + for manifest in _runs_dir(data_dir).glob("*/manifest.json"): + try: + value = json.loads(manifest.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + continue + if isinstance(value, dict) and value.get("status") in ACTIVE_RUN_STATUSES: + count += 1 + return count + + +def _store_idle_seconds(data_dir: Path | str, *, now: float) -> float | None: + """挖掘 run 存储最近一次写入距今的秒数; 无任何文件返回 None (视为足够陈旧)。""" + runs_dir = _runs_dir(data_dir) + if not runs_dir.is_dir(): + return None + latest: float | None = None + for path in runs_dir.rglob("*"): + if not path.is_file(): + continue + try: + mtime = path.stat().st_mtime + except OSError: + continue + if latest is None or mtime > latest: + latest = mtime + if latest is None: + return None + return max(0.0, now - latest) + + +def _terminate_pid(pid: int) -> None: + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + return + except PermissionError as exc: + raise MiningLockAdminError(f"无权限终止持有者 pid={pid}") from exc + + +def _wait_condition(condition, timeout: float) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(0.2) + return condition() diff --git a/backend/app/services/mining_process_lock.py b/backend/app/services/mining_process_lock.py index f0323b07..4580a3c9 100644 --- a/backend/app/services/mining_process_lock.py +++ b/backend/app/services/mining_process_lock.py @@ -1,6 +1,9 @@ from __future__ import annotations +import json import os +import socket +from datetime import UTC, datetime from pathlib import Path from typing import BinaryIO @@ -9,12 +12,19 @@ class MiningProcessLockError(RuntimeError): """Another application process owns mining for this data directory.""" +_LOCK_SCHEMA_VERSION = 1 + + class MiningProcessLock: def __init__(self, data_dir: Path) -> None: self._path = Path(data_dir) / ".mining_process.lock" self._stream: BinaryIO | None = None - def acquire(self) -> None: + @property + def path(self) -> Path: + return self._path + + def acquire(self, *, role: str = "app") -> None: if self._stream is not None: return self._path.parent.mkdir(parents=True, exist_ok=True) @@ -30,6 +40,9 @@ class MiningProcessLock: stream.close() raise self._stream = stream + # flock 是唯一事实; 内容只是给诊断/强制释放用的持有者线索。 + # 持锁期间本进程是该文件唯一写入者, 无并发竞争。 + self._rewrite_locked(stream, holder_payload(role)) def release(self) -> None: stream = self._stream @@ -37,10 +50,93 @@ class MiningProcessLock: return self._stream = None try: + self._rewrite_locked( + stream, + { + "schema_version": _LOCK_SCHEMA_VERSION, + "pid": os.getpid(), + "released_at": datetime.now(UTC).isoformat(), + }, + ) _unlock_file(stream) finally: stream.close() + @property + def acquired(self) -> bool: + return self._stream is not None + + @staticmethod + def _rewrite_locked(stream: BinaryIO, payload: dict) -> None: + try: + stream.seek(0) + stream.truncate() + stream.write( + (json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8") + ) + stream.flush() + os.fsync(stream.fileno()) + except OSError: + # 诊断信息写失败不影响锁语义 (flock 已生效/将生效) + pass + + +def holder_payload(role: str) -> dict: + return { + "schema_version": _LOCK_SCHEMA_VERSION, + "pid": os.getpid(), + "started_at": datetime.now(UTC).isoformat(), + "role": role, + "host": socket.gethostname(), + } + + +def read_lock_holder(data_dir: Path | str) -> dict | None: + """读取锁文件中的持有者线索; 旧格式 ("0") / 损坏内容返回 None。 + + 返回值仅供参考: flock 才是持锁事实, 内容可能滞后 (例如持有者崩溃后 + 内核自动释放锁但内容未更新)。 + """ + path = Path(data_dir) / ".mining_process.lock" + try: + raw = path.read_bytes() + except OSError: + return None + try: + value = json.loads(raw.decode("utf-8", errors="replace")) + except (UnicodeError, json.JSONDecodeError): + return None + if not isinstance(value, dict): + return None + pid = value.get("pid") + if not isinstance(pid, int) or pid <= 0 or "released_at" in value: + return None + return value + + +def probe_lock_held(data_dir: Path | str) -> bool: + """非阻塞探测锁是否被任何进程持有 (含本进程自己)。""" + path = Path(data_dir) / ".mining_process.lock" + try: + stream = path.open("a+b") + except OSError: + return False + try: + os.set_inheritable(stream.fileno(), False) + _try_lock_file(stream) + except MiningProcessLockError: + return True + except OSError: + # 无法探测时按未持锁处理, 让真正的 acquire 路径给出准确错误 + return False + try: + _unlock_file(stream) + except OSError: + pass + finally: + stream.close() + return False + def _try_lock_file(stream: BinaryIO) -> None: if os.name == "nt": diff --git a/backend/tests/test_desktop_startup.py b/backend/tests/test_desktop_startup.py index 1bcb97da..3e297dee 100644 --- a/backend/tests/test_desktop_startup.py +++ b/backend/tests/test_desktop_startup.py @@ -99,28 +99,38 @@ def test_precheck_mining_lock_detects_conflict(data_dir): # ---------- main() 预检冲突路径 ---------- -def test_main_releases_lock_when_failure_window_crashes( +def test_main_proceeds_when_mining_lock_precheck_conflicts( tmp_path, monkeypatch, caplog ): - """预检冲突 + 失败页窗口自身崩溃 (如缺 webview 依赖): 异常不得裸穿 main()。 + """预检撞锁只降级: 留 WARNING 日志并继续启动, 不再提前退出失败页。 - 该路径在 main() 的 try/finally 之外, 不兜底会导致 console=False 闪退无反馈、 - .desktop.lock 跳过释放。断言: 记录异常日志、单实例锁仍释放、退出码 1。 + lifespan 会以降级模式起面板 (挖掘入口 409, 挖掘页提供强制释放), + 桌面版必须照常开窗, 否则用户没有入口处理锁冲突。 """ + import logging + monkeypatch.setattr(settings, "data_dir", tmp_path, raising=False) monkeypatch.setattr(desktop, "_guard_streams", lambda: None) monkeypatch.setattr(desktop, "_setup_logging", lambda: None) - monkeypatch.setattr(desktop, "_precheck_mining_lock", lambda: "冲突原因") + monkeypatch.setattr( + desktop, "_precheck_mining_lock", lambda: "另一个后端实例正在占用数据目录" + ) + monkeypatch.setattr(desktop, "_find_free_port", lambda base: 59998) + monkeypatch.setattr( + desktop, "_wait_for_server", + lambda port, timeout, done_event: True, + ) + opened = [] + monkeypatch.setattr(desktop, "_open_window", lambda url: opened.append(url)) - def _boom(_html_text): - raise RuntimeError("webview unavailable") + def _fake_uvmicorn(port, ready): + ready.set() - monkeypatch.setattr(desktop, "_open_static_window", _boom) + monkeypatch.setattr(desktop, "_run_uvmicorn", _fake_uvmicorn) - import logging + with caplog.at_level(logging.WARNING, logger="app.desktop"): + assert desktop.main() == 0 - with caplog.at_level(logging.ERROR, logger="app.desktop"): - assert desktop.main() == 1 - - assert "启动失败页窗口打开失败" in caplog.text + assert opened and opened[0].startswith("http://127.0.0.1:") + assert "启动预检" in caplog.text assert not (tmp_path / ".desktop.lock").exists() diff --git a/backend/tests/test_main_mining_lifespan.py b/backend/tests/test_main_mining_lifespan.py index 4181649c..cc7d932f 100644 --- a/backend/tests/test_main_mining_lifespan.py +++ b/backend/tests/test_main_mining_lifespan.py @@ -9,6 +9,10 @@ import pytest import app.main as main_module +def _app_stub() -> SimpleNamespace: + return SimpleNamespace(state=SimpleNamespace()) + + def test_lifespan_holds_mining_process_lock_around_application(monkeypatch) -> None: events: list[str] = [] @@ -16,11 +20,14 @@ def test_lifespan_holds_mining_process_lock_around_application(monkeypatch) -> N def __init__(self, data_dir) -> None: del data_dir events.append("lock_created") + self.acquired = False def acquire(self) -> None: + self.acquired = True events.append("lock_acquired") def release(self) -> None: + self.acquired = False events.append("lock_released") @asynccontextmanager @@ -34,8 +41,10 @@ def test_lifespan_holds_mining_process_lock_around_application(monkeypatch) -> N monkeypatch.setattr(main_module, "MiningProcessLock", LockStub) monkeypatch.setattr(main_module, "_application_lifespan", application_lifespan) + app = _app_stub() + async def exercise() -> None: - async with main_module.lifespan(SimpleNamespace()): + async with main_module.lifespan(app): events.append("request_serving") asyncio.run(exercise()) @@ -48,6 +57,8 @@ def test_lifespan_holds_mining_process_lock_around_application(monkeypatch) -> N "application_stopped", "lock_released", ] + assert app.state.mining_lock_acquired is True + assert app.state.mining_process_lock is not None def test_lifespan_releases_lock_when_application_shutdown_raises(monkeypatch) -> None: @@ -56,11 +67,14 @@ def test_lifespan_releases_lock_when_application_shutdown_raises(monkeypatch) -> class LockStub: def __init__(self, data_dir) -> None: del data_dir + self.acquired = False def acquire(self) -> None: + self.acquired = True events.append("acquired") def release(self) -> None: + self.acquired = False events.append("released") @asynccontextmanager @@ -74,7 +88,7 @@ def test_lifespan_releases_lock_when_application_shutdown_raises(monkeypatch) -> monkeypatch.setattr(main_module, "_application_lifespan", application_lifespan) async def exercise() -> None: - async with main_module.lifespan(SimpleNamespace()): + async with main_module.lifespan(_app_stub()): pass with pytest.raises(RuntimeError, match="shutdown failed"): @@ -83,17 +97,22 @@ def test_lifespan_releases_lock_when_application_shutdown_raises(monkeypatch) -> assert events == ["acquired", "released"] -def test_lifespan_logs_true_cause_when_lock_held(monkeypatch, caplog) -> None: - """抢锁失败: 先落 ERROR 日志 (桌面版失败页指向 desktop.log 的排查依据) 再上抛。 +def test_lifespan_degrades_instead_of_failing_when_lock_held( + monkeypatch, caplog +) -> None: + """抢锁失败: 不再让 lifespan 上抛 (整个后端起不来), 改为降级启动。 - uvicorn 会吞掉 lifespan 异常只留一行笼统的 startup failed, 真因必须由 - app 自己先写日志, 否则用户按失败页指引打开 desktop.log 查无对证。 + 面板照常可用, 挖掘启动入口返回 409; WARNING 日志留排查线索; + 未获锁不得在 shutdown 时误调 release。 """ from app.services.mining_process_lock import MiningProcessLockError + events: list[str] = [] + class LockStub: def __init__(self, data_dir) -> None: del data_dir + self.acquired = False def acquire(self) -> None: raise MiningProcessLockError("another process owns mining") @@ -101,16 +120,25 @@ def test_lifespan_logs_true_cause_when_lock_held(monkeypatch, caplog) -> None: def release(self) -> None: raise AssertionError("acquire 失败后不应调用 release") + @asynccontextmanager + async def application_lifespan(_app): + events.append("application_started") + yield + events.append("application_stopped") + monkeypatch.setattr(main_module, "MiningProcessLock", LockStub) + monkeypatch.setattr(main_module, "_application_lifespan", application_lifespan) + + app = _app_stub() async def exercise() -> None: - async with main_module.lifespan(SimpleNamespace()): - pass + async with main_module.lifespan(app): + events.append("request_serving") - with ( - caplog.at_level("ERROR", logger="app.main"), - pytest.raises(MiningProcessLockError), - ): - asyncio.run(exercise()) + with caplog.at_level("WARNING", logger="app.main"): + asyncio.run(exercise()) # 不再抛出 + assert events == ["application_started", "request_serving", "application_stopped"] assert "挖掘锁" in caplog.text + assert "降级" in caplog.text + assert app.state.mining_lock_acquired is False diff --git a/backend/tests/test_mining_lock_admin.py b/backend/tests/test_mining_lock_admin.py new file mode 100644 index 00000000..e4d58bd6 --- /dev/null +++ b/backend/tests/test_mining_lock_admin.py @@ -0,0 +1,304 @@ +"""挖掘锁持有者记录 / 陈旧判定 / 强制释放 + 撞锁降级启动测试。""" +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.api.mining import router as mining_router +from app.services.mining_lock_admin import ( + MiningLockAdminError, + _classify, + _count_active_runs, + _store_idle_seconds, + force_release, + lock_status, +) +from app.services.mining_process_lock import ( + MiningProcessLock, + MiningProcessLockError, + probe_lock_held, + read_lock_holder, +) + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] + +_HOLDER_SCRIPT = """ +import sys, time +from app.services.mining_process_lock import MiningProcessLock +lock = MiningProcessLock(sys.argv[1]) +lock.acquire(role="worker") +print("HELD", flush=True) +time.sleep(600) +""" + +_POSIX = pytest.mark.skipif(os.name == "nt", reason="外部持锁进程仅在 POSIX 验证") + + +def _spawn_holder(data_dir: Path) -> subprocess.Popen: + env = {**os.environ, "PYTHONPATH": str(_BACKEND_ROOT)} + proc = subprocess.Popen( + [sys.executable, "-c", _HOLDER_SCRIPT, str(data_dir)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=env, + cwd=str(_BACKEND_ROOT), + ) + line = proc.stdout.readline() + assert line.strip() == "HELD", f"外部持锁进程未就绪: {line!r} {proc.stderr.read()}" + return proc + + +def _kill(proc: subprocess.Popen) -> None: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + + +# ── 锁内容与探测 ─────────────────────────────────────────────── + + +def test_lock_records_holder_and_release_marker(tmp_path): + lock = MiningProcessLock(tmp_path) + lock.acquire() + holder = read_lock_holder(tmp_path) + assert holder is not None + assert holder["pid"] == os.getpid() + assert holder["role"] == "app" + assert probe_lock_held(tmp_path) is True + + lock.release() + assert probe_lock_held(tmp_path) is False + # 释放标记后不再把内容当作有效持有者 + assert read_lock_holder(tmp_path) is None + + +def test_second_acquire_conflicts_in_same_process(tmp_path): + first = MiningProcessLock(tmp_path) + first.acquire() + second = MiningProcessLock(tmp_path) + with pytest.raises(MiningProcessLockError): + second.acquire() + first.release() + + +def test_legacy_lock_content_is_unknown_holder(tmp_path): + path = tmp_path / ".mining_process.lock" + path.write_bytes(b"0") + assert read_lock_holder(tmp_path) is None + assert probe_lock_held(tmp_path) is False + + +def test_corrupt_lock_content_is_unknown_holder(tmp_path): + (tmp_path / ".mining_process.lock").write_text("{not json") + assert read_lock_holder(tmp_path) is None + + +def test_status_reports_mine(tmp_path): + lock = MiningProcessLock(tmp_path) + lock.acquire() + try: + status = lock_status(tmp_path) + assert status["held"] is True + assert status["mine"] is True + assert status["force_releasable"] is False + assert "当前后端" in status["reason"] + finally: + lock.release() + + +# ── 陈旧判定单元矩阵 ────────────────────────────────────────── + + +def test_classify_matrix(): + base = dict( + held=True, mine=False, holder_known=True, holder_alive=True, + pid_consistent=True, active_runs=0, store_idle=None, stale_after=600.0, + ) + ok, reason = _classify(**base) + assert ok and reason is None + + not_held = {**base, "held": False} + assert _classify(**not_held) == (False, "锁未被持有") + + legacy = {**base, "holder_known": False} + ok, reason = _classify(**legacy) + assert not ok and "旧格式" in reason + + reused = {**base, "pid_consistent": False} + ok, reason = _classify(**reused) + assert not ok and "复用" in reason + + active = {**base, "active_runs": 2} + ok, reason = _classify(**active) + assert not ok and "2 个进行中" in reason + + fresh = {**base, "store_idle": 12.0} + ok, reason = _classify(**fresh) + assert not ok and "陈旧阈值" in reason + + dead = {**base, "holder_alive": False} + ok, reason = _classify(**dead) + assert not ok and "已不存在" in reason + + +def test_count_active_runs_and_store_idle(tmp_path): + runs = tmp_path / "research" / "mining" / "runs" + assert _count_active_runs(tmp_path) == 0 + assert _store_idle_seconds(tmp_path, now=time.time()) is None + + done = runs / "run-done" + done.mkdir(parents=True) + (done / "manifest.json").write_text(json.dumps({"status": "succeeded"})) + active = runs / "run-live" + active.mkdir(parents=True) + (active / "manifest.json").write_text(json.dumps({"status": "running"})) + (active / "events.jsonl").write_text("{}\n") + + assert _count_active_runs(tmp_path) == 1 + idle = _store_idle_seconds(tmp_path, now=time.time()) + assert idle is not None and idle < 60 + + +def test_count_active_runs_tolerates_corrupt_manifest(tmp_path): + runs = tmp_path / "research" / "mining" / "runs" / "bad" + runs.mkdir(parents=True) + (runs / "manifest.json").write_text("{broken") + assert _count_active_runs(tmp_path) == 0 + + +# ── 强制释放 (真实外部持有进程, POSIX) ───────────────────────── + + +@_POSIX +def test_force_release_kills_external_holder_and_takes_over(tmp_path): + proc = _spawn_holder(tmp_path) + try: + status = lock_status(tmp_path) + assert status["held"] is True + assert status["mine"] is False + assert status["holder"]["role"] == "worker" + assert status["force_releasable"] is True, status["reason"] + + lock = MiningProcessLock(tmp_path) + report = force_release(tmp_path, lock) + assert report["released"] is True + assert report["terminated_pid"] == proc.pid + assert report["reacquired"] is True + assert lock.acquired is True + proc.wait(timeout=10) + assert lock_status(tmp_path)["mine"] is True + lock.release() + finally: + _kill(proc) + + +@_POSIX +def test_force_release_refuses_fresh_store(tmp_path): + proc = _spawn_holder(tmp_path) + try: + runs = tmp_path / "research" / "mining" / "runs" / "r1" + runs.mkdir(parents=True) + (runs / "manifest.json").write_text(json.dumps({"status": "succeeded"})) + status = lock_status(tmp_path) + assert status["force_releasable"] is False + assert "陈旧阈值" in status["reason"] + with pytest.raises(MiningLockAdminError, match="陈旧阈值"): + force_release(tmp_path, MiningProcessLock(tmp_path)) + assert proc.poll() is None, "被拒绝时不得误杀持有者" + finally: + _kill(proc) + + +@_POSIX +def test_force_release_refuses_active_runs(tmp_path): + proc = _spawn_holder(tmp_path) + try: + runs = tmp_path / "research" / "mining" / "runs" / "r2" + runs.mkdir(parents=True) + (runs / "manifest.json").write_text(json.dumps({"status": "running"})) + status = lock_status(tmp_path) + assert status["force_releasable"] is False + assert "进行中" in status["reason"] + assert proc.poll() is None + finally: + _kill(proc) + + +def test_force_release_refuses_when_lock_free(tmp_path): + with pytest.raises(MiningLockAdminError, match="未被持有"): + force_release(tmp_path, MiningProcessLock(tmp_path)) + + +# ── API 接线: 撞锁降级 + 锁端点 ──────────────────────────────── + + +def _api_client(tmp_path, *, acquired: bool): + app = FastAPI() + app.include_router(mining_router) + app.state.repo = SimpleNamespace(store=SimpleNamespace(data_dir=tmp_path)) + app.state.mining_lock_acquired = acquired + app.state.mining_process_lock = MiningProcessLock(tmp_path) + return TestClient(app) + + +def test_degraded_mode_rejects_mining_start(tmp_path): + client = _api_client(tmp_path, acquired=False) + resp = client.post( + "/api/backtest/mining/runs", + json={"factor_names": ["momentum_5d"], "budget_profile": "exploratory"}, + ) + assert resp.status_code == 409 + assert "降级" in resp.json()["detail"] + + +def test_lock_endpoints_free_lock(tmp_path): + client = _api_client(tmp_path, acquired=True) + body = client.get("/api/backtest/mining/lock").json() + assert body["held"] is False + assert body["force_releasable"] is False + resp = client.post("/api/backtest/mining/lock/force-release") + assert resp.status_code == 409 + assert "未被持有" in resp.json()["detail"] + + +@_POSIX +def test_lock_endpoints_full_recovery_loop(tmp_path): + proc = _spawn_holder(tmp_path) + try: + client = _api_client(tmp_path, acquired=False) + body = client.get("/api/backtest/mining/lock").json() + assert body["held"] is True + assert body["mine"] is False + assert body["force_releasable"] is True, body["reason"] + + resp = client.post("/api/backtest/mining/lock/force-release") + assert resp.status_code == 200 + assert resp.json() == { + "released": True, + "terminated_pid": proc.pid, + "reacquired": True, + } + proc.wait(timeout=10) + after = client.get("/api/backtest/mining/lock").json() + assert after["mine"] is True + + # 接管后挖掘入口必须立即恢复 — 不再是 409 降级 + # (stub app 无 manager, 返回 503; 真实环境此处应为 200/400) + resp = client.post( + "/api/backtest/mining/runs", + json={"factor_names": ["momentum_5d"], "budget_profile": "exploratory"}, + ) + assert resp.status_code != 409 + assert "降级" not in resp.json().get("detail", "") + finally: + _kill(proc) diff --git a/docs/features.md b/docs/features.md index 0d705a25..1da7673e 100644 --- a/docs/features.md +++ b/docs/features.md @@ -66,7 +66,7 @@ **策略守护(验证 tab)**:挖掘发布的策略在发布时自动建立守护基线(晋升门槛指标,天然 OOS);此后每天收盘数据更新后自动在新数据上切 K 段做真实回测复验,按稳健分对比基线判定三态——健康(退化≤阈值)/观察(20%-50% 或击穿软底线,连续 2 次告警)/退役(退化>50% 或交易数枯竭)。退役只标记与置灰,不自动删策略文件;已退役策略不再自动复验(重新守护=移除后重新入护)。任意策略也可手动入护(尾随参考基线,非 OOS,判定只针对入护后新数据)。数据纪律:守护层结果绝不反馈进取参加权(逐层 OOS);段落回测失败时只降级到观察不直接退役;新数据不足 30 个交易日跳过本轮。守护页还提供**实验室战报**(守护/挖掘/名额聚合总览,`/api/backtest/lab/report`)与**组合建议**:扫描健康守护冠军间近 126 个交易日的日收益 Pearson 相关性,≤0.7 的对列出,人工确认后生成等权 composite 策略并自动以尾随参考基线入护(幂等,不在最新建议中的对拒绝生成);**挖掘策略名额上限**(偏好项,默认 10,0=不限)在发布门槛通过后校验,名额满时新冠军停在 pending 候选等位。设计见 [自动策略实验室设计](./auto-strategy-lab-plan.md)。 -**因子与策略挖掘**:复用已有日频因子和 matrix-native 策略，通过 T-1 市场环境、相关去重和嵌套样本外验证生成研究候选。冠军排序按稳健分(0.6×中位+0.4×最差外层折),strict 档要求最差折 Sharpe>0;对冠军组合再做权重精调({0.5,1.0,1.5,2.0} 网格,消耗 run 剩余预算,退化>20% 放弃),精调结果作为独立候选行追加。任务在 spawn worker 中运行，支持持久 run ID、取消、刷新重连和显式发布；自动周度任务默认关闭且永不自动发布,周度挖掘与每日守护由调度单入口(`lab_schedule`)编排、失败互不阻断。完整口径见 [因子与策略挖掘](./mining.md)。 +**因子与策略挖掘**:复用已有日频因子和 matrix-native 策略，通过 T-1 市场环境、相关去重和嵌套样本外验证生成研究候选。冠军排序按稳健分(0.6×中位+0.4×最差外层折),strict 档要求最差折 Sharpe>0;对冠军组合再做权重精调({0.5,1.0,1.5,2.0} 网格,消耗 run 剩余预算,退化>20% 放弃),精调结果作为独立候选行追加。任务在 spawn worker 中运行，支持持久 run ID、取消、刷新重连和显式发布；自动周度任务默认关闭且永不自动发布,周度挖掘与每日守护由调度单入口(`lab_schedule`)编排、失败互不阻断。挖掘锁被其他进程占用时后端**降级启动**(挖掘入口 409、其余功能不受影响),挖掘页显示持有者横幅,满足陈旧判定(无进行中运行 + 存储静默 + pid 未复用,fail-closed)可一键强制停止并接管。完整口径见 [因子与策略挖掘](./mining.md)。 **策略设置统一**:回测页「高级策略设置」抽屉与策略页设置弹窗采用同一信息架构(策略条件 / 策略权重 / 风控触发 / 回测范围)。风控与持仓纪律(止损 · 止盈 · 移动止损 · 回撤止盈 · 最长持仓)以策略设置保存的值为默认,回测抽屉内可临时覆盖,也可「应用到策略」持久化;策略页保存的叠加条件随回测请求继承,与扫描 / 监控同一口径;回测范围(资产类型 / 股票池)为回测专属概念。 diff --git a/docs/mining.md b/docs/mining.md index b6aa2c52..24bcec15 100644 --- a/docs/mining.md +++ b/docs/mining.md @@ -108,6 +108,8 @@ run 级的“有效折”统计只描述因子赛道：它按每折被选中的 当前 run store 和重任务 limiter 使用进程内锁。生产环境必须只启动一个应用进程负责挖掘；多 worker Uvicorn 部署不能保证跨进程单飞、事件追加和容量限制。要支持多应用进程，必须先增加操作系统级文件锁或外部任务协调器。 +**挖掘锁冲突的降级与强制释放**：数据目录的 `.mining_process.lock`（flock）保证同一时刻只有一个应用进程负责挖掘；持锁者会把 `{pid, started_at, role}` 写进锁文件（flock 仍是唯一事实，内容只是诊断线索，旧格式 `"0"` 视为持有者未知）。撞锁不再导致后端启动失败——面板以降级模式运行：挖掘启动入口（手动/自动/周度调度）返回 409，守护复验等其他功能不受影响；挖掘页顶部显示持有者横幅（pid、角色、挖掘存储静默时长）。当且仅当满足陈旧判定（持有者信息可读、进程存活且 pid 未被复用、无进行中的挖掘运行、挖掘存储 10 分钟内无写入——全部 fail-closed，任一不满足即拒绝并给出原因）时，可一键「强制停止并接管」：终止持有者进程并由当前后端获取锁，挖掘入口立即恢复。不做启动时自动清理——终止进程是破坏性操作，必须人工确认。 + 每个运行目录包含 manifest、compact summary、最近 256 条事件和四个 Parquet artifact。大面板、完整矩阵对象和逐折宽结果不会通过 worker IPC 返回。 ## 自动运行 diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 95d141f0..dbabcf69 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1575,6 +1575,26 @@ export interface MiningAvailability { suggested_start: string | null } +/** 挖掘锁诊断 (撞锁降级模式的持有者线索 + 强制释放判定)。 */ +export interface MiningLockStatus { + held: boolean + mine: boolean + holder: { + pid: number + started_at?: string + role?: string + host?: string + schema_version?: number + } | null + holder_alive: boolean + pid_consistent: boolean + active_runs: number + store_idle_seconds: number | null + stale_after_seconds: number + force_releasable: boolean + reason: string | null +} + export interface MiningRequestV1 { factor_names: string[] strategy_ids?: string[] @@ -3462,6 +3482,14 @@ export const api = { ...(signal ? { signal } : {}), }), + miningLockStatus: () => + request('/api/backtest/mining/lock', { quiet: true }), + + miningForceReleaseLock: () => + request<{ released: boolean; terminated_pid: number; reacquired: boolean }>( + '/api/backtest/mining/lock/force-release', { method: 'POST' }, + ), + miningRuns: () => request<{ items: MiningRun[] }>('/api/backtest/mining/runs'), diff --git a/frontend/src/lib/queryKeys.ts b/frontend/src/lib/queryKeys.ts index 0a803d59..bc535b50 100644 --- a/frontend/src/lib/queryKeys.ts +++ b/frontend/src/lib/queryKeys.ts @@ -68,6 +68,7 @@ export const QK = { miningRun: (id: string) => ['backtest-mining-run', id] as const, miningResult: (id: string) => ['backtest-mining-result', id] as const, miningConfig: ['backtest-mining-config'] as const, + miningLock: ['backtest-mining-lock'] as const, guardianOverview: ['backtest-guardian-overview'] as const, guardianRuns: ['backtest-guardian-runs'] as const, labReport: ['backtest-lab-report'] as const, diff --git a/frontend/src/pages/backtest/MiningWorkbench.tsx b/frontend/src/pages/backtest/MiningWorkbench.tsx index a270f2ba..9e03a55b 100644 --- a/frontend/src/pages/backtest/MiningWorkbench.tsx +++ b/frontend/src/pages/backtest/MiningWorkbench.tsx @@ -343,6 +343,78 @@ function FactorTable({ result }: { result: MiningResult }) { ) } +/** 挖掘锁占用横幅: 撞锁降级时展示持有者线索, 满足陈旧判定可强制释放。 */ +function MiningLockBanner() { + const queryClient = useQueryClient() + const lockQuery = useQuery({ + queryKey: QK.miningLock, + queryFn: api.miningLockStatus, + staleTime: 10_000, + // 被他人持有时轮询等待恢复; 空闲/自己持有则不轮询 + refetchInterval: query => + query.state.data?.held && !query.state.data.mine ? 15_000 : false, + }) + const release = useMutation({ + mutationFn: api.miningForceReleaseLock, + onSuccess: () => { + toast('已终止持有者并由当前后端接管挖掘锁', 'success') + void queryClient.invalidateQueries({ queryKey: QK.miningLock }) + }, + onError: (e: Error) => toast(e.message, 'error'), + }) + const data = lockQuery.data + if (!data?.held || data.mine) return null + const holder = data.holder + const idleMin = data.store_idle_seconds != null + ? Math.max(1, Math.round(data.store_idle_seconds / 60)) + : null + return ( +
+
+
+ +
+
挖掘锁被其他进程占用 — 挖掘暂不可用
+
+ {holder ? ( + <> + 持有者 pid {holder.pid} + {holder.role ? ` · ${holder.role === 'worker' ? '挖掘 worker 进程' : '后端实例'}` : ''} + {idleMin != null ? ` · 挖掘存储已静默约 ${idleMin} 分钟` : ''} + {data.force_releasable + ? ' — 满足陈旧判定, 可强制释放' + : ` — 不可强制释放: ${data.reason}`} + + ) : ( + '锁文件为旧格式, 无法定位持有进程 — 请手动结束后端残留进程后重启面板' + )} +
+
+
+ {data.force_releasable && ( + + )} +
+
+ ) +} + export function MiningWorkbench() { const queryClient = useQueryClient() const [searchParams, setSearchParams] = useSearchParams() @@ -668,6 +740,8 @@ export function MiningWorkbench() { } return ( + <> +
+ ) }