fix(ops): #516 内存可观测 + 两道上限

- /health 追加 memory 块: 进程 RSS / virtual_memory total/available/percent / cgroup v2 current/high/max/events
- worker 轮询 (2s 节流) 监控子进程 RSS, 超 TICK_WORKER_RSS_LIMIT_BYTES (默认 6GiB, 0 关闭) terminate + 明确中文错误; 不用 RLIMIT_AS (polars/numpy 虚拟内存远大于 RSS 会误杀)
- 重任务准入: 系统可用内存低于 TICK_HEAVY_MIN_AVAILABLE_BYTES (默认 1GiB, 0 关闭) 时 FIFO 继续等待
- README 补两个环境变量说明

README.md

@@ -555,6 +555,13 @@ POLARS_MAX_THREADS=2        # 主要收益项: 用约 40% 的构建耗时换 500
 
 - **确认预热规模**:`data/logs/backend.log` 中的 `enriched 历史缓存: N rows` 表示本次预热实际载入的行数,N 越大占用越高。历史数据较少时(例如首次只同步了几十天)内存会明显低于上表。
 
+**内存保护环境变量**(`/health` 的 `memory` 字段可实时观察主进程 RSS / 系统可用 / cgroup 节流计数):
+
+| 变量 | 默认 | 作用 |
+| :-- | :-- | :-- |
+| `TICK_WORKER_RSS_LIMIT_BYTES` | `6442450944` (6GiB) | 回测/优化独立 worker 的 RSS 上限, 超限终止并返回明确错误(而非拖垮整机); `0` = 不限制 |
+| `TICK_HEAVY_MIN_AVAILABLE_BYTES` | `1073741824` (1GiB) | 系统可用内存低于该值时, 新的重任务(回测/因子批量/挖掘)挂起等待内存恢复; `0` = 不检查 |
+
 ### 跑起来后的第一次使用
 
 1. **设置 → 凭据与能力** → 点 **重新检测**,确认档位标签与能力路由矩阵

backend/app/api/routes.py

@@ -1,6 +1,8 @@
 """API 路由 — Phase 0 仅 /health 与 /api/capabilities。"""
 from __future__ import annotations
 
+from pathlib import Path
+
 from fastapi import APIRouter, Request
 
 from app import __version__
@@ -10,6 +12,55 @@ from app.tickflow.policy import detect_capabilities, tier_label
 router = APIRouter()
 
 
+def _read_cgroup_int(path: Path) -> int | None:
+    try:
+        text = path.read_text(encoding="utf-8").strip()
+    except OSError:
+        return None
+    if not text or text == "max":
+        return None
+    try:
+        return int(text)
+    except ValueError:
+        return None
+
+
+def _memory_snapshot() -> dict:
+    """内存可观测性 (issue #516): 主进程 RSS + 系统可用 + cgroup v2 节流态。
+
+    让外部看门狗在「服务活着但被 cgroup MemoryHigh 节流到假死」时有数可判,
+    而不是只有 status=ok。
+    """
+    import psutil
+
+    vm = psutil.virtual_memory()
+    snap: dict[str, object] = {
+        "rss_bytes": psutil.Process().memory_info().rss,
+        "system_total_bytes": vm.total,
+        "system_available_bytes": vm.available,
+        "system_used_pct": vm.percent,
+    }
+    cg = Path("/sys/fs/cgroup")
+    if _read_cgroup_int(cg / "memory.current") is not None:
+        snap["cgroup_current_bytes"] = _read_cgroup_int(cg / "memory.current")
+        snap["cgroup_high_bytes"] = _read_cgroup_int(cg / "memory.high")
+        snap["cgroup_max_bytes"] = _read_cgroup_int(cg / "memory.max")
+        events: dict[str, int] = {}
+        try:
+            for line in (cg / "memory.events").read_text(encoding="utf-8").splitlines():
+                parts = line.split()
+                if len(parts) == 2:
+                    try:
+                        events[parts[0]] = int(parts[1])
+                    except ValueError:
+                        continue
+        except OSError:
+            pass
+        if events:
+            snap["cgroup_events"] = events
+    return snap
+
+
 @router.get("/health")
 def health() -> dict:
     return {
@@ -17,6 +68,7 @@ def health() -> dict:
         "version": __version__,
         # 三态: none(无key/无效) / free(免费key) / api_key(付费档)
         "mode": tf_client.current_mode(),
+        "memory": _memory_snapshot(),
     }
 
 

backend/app/backtest/worker.py

@@ -28,6 +28,32 @@ class BacktestWorkerError(RuntimeError):
 _CANCEL_GRACE_SECONDS = 5.0
 
 
+def _env_int(name: str, default: int) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        return int(float(raw))
+    except ValueError:
+        return default
+
+
+# worker RSS 上限 (issue #516): 超限 terminate 并返回明确错误, 而不是把整机
+# 拖进 swap/内存节流假死。默认 6GiB; 0 = 不限制。刻意不用 RLIMIT_AS ——
+# polars/numpy 的虚拟内存映射远大于实际 RSS, AS 上限会大量误杀。
+_WORKER_RSS_LIMIT_BYTES = _env_int("TICK_WORKER_RSS_LIMIT_BYTES", 6 * 1024**3)
+_RSS_CHECK_INTERVAL_S = 2.0
+
+
+def _process_rss_bytes(pid: int | None) -> int:
+    if pid is None:
+        return 0
+    try:
+        return psutil.Process(pid).memory_info().rss
+    except psutil.Error:
+        return 0
+
+
 class _PeakRssSampler:
     """Track whole-task and resettable phase RSS peaks with one sampling thread."""
 
@@ -316,19 +342,42 @@ def run_worker_task(
     failure: dict[str, Any] | None = None
     ipc_started = time.perf_counter()
     cancel_started: float | None = None
+    last_rss_check = 0.0
 
     try:
         while result is None and failure is None:
+            # 每轮只取一次时钟 (取消宽限与 RSS 检查共用), 测试用脚本时钟锁定调用量
+            now = time.monotonic()
             if cancel_event is not None and cancel_event.is_set():
                 process_cancel.set()
                 if cancel_started is None:
-                    cancel_started = time.monotonic()
-                elif time.monotonic() - cancel_started >= _CANCEL_GRACE_SECONDS:
+                    cancel_started = now
+                elif now - cancel_started >= _CANCEL_GRACE_SECONDS:
                     process.terminate()
                     process.join(timeout=5.0)
                     raise BacktestWorkerError(
                         "backtest worker did not stop within 5 seconds after cancellation"
                     )
+            if (
+                _WORKER_RSS_LIMIT_BYTES > 0
+                and now - last_rss_check >= _RSS_CHECK_INTERVAL_S
+            ):
+                last_rss_check = now
+                # getattr: 测试替身进程可能没有 pid 属性, 缺失时按 0 处理 (不触发)
+                rss = _process_rss_bytes(getattr(process, "pid", None))
+                if rss > _WORKER_RSS_LIMIT_BYTES:
+                    logger.error(
+                        "backtest worker rss %.2f GiB exceeds limit %.2f GiB, terminating",
+                        rss / 1024**3, _WORKER_RSS_LIMIT_BYTES / 1024**3,
+                    )
+                    process.terminate()
+                    process.join(timeout=5.0)
+                    raise BacktestWorkerError(
+                        f"回测任务内存超过上限 {_WORKER_RSS_LIMIT_BYTES / 1024**3:.1f}GiB"
+                        f" (当前 {rss / 1024**3:.1f}GiB), 已中止。"
+                        "可缩小回测时间范围/标的数量, 或用环境变量 "
+                        "TICK_WORKER_RSS_LIMIT_BYTES 调整上限 (0=不限制)。"
+                    )
             try:
                 message = events.get(timeout=0.1)
             except queue.Empty:

backend/app/services/heavy_job_limiter.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import threading
 import time
 from collections import deque
@@ -12,6 +13,31 @@ from typing import ClassVar, Literal
 HeavyJobKind = Literal["normal", "mining", "exclusive"]
 
 
+def _env_int(name: str, default: int) -> int:
+    raw = os.environ.get(name, "").strip()
+    if not raw:
+        return default
+    try:
+        return int(float(raw))
+    except ValueError:
+        return default
+
+
+# 内存准入 (issue #516): 可用内存低于阈值时重任务不进门, 挂起等待恢复,
+# 避免叠加任务把整机拖进 swap。0 = 关闭。等待者以 _MEMORY_POLL_S 周期重查
+# (内存恢复不会触发 condition notify, 必须轮询)。
+_HEAVY_MIN_AVAILABLE_BYTES = _env_int("TICK_HEAVY_MIN_AVAILABLE_BYTES", 1024**3)
+_MEMORY_POLL_S = 5.0
+
+
+def _memory_admission_ok() -> bool:
+    if _HEAVY_MIN_AVAILABLE_BYTES <= 0:
+        return True
+    import psutil
+
+    return psutil.virtual_memory().available >= _HEAVY_MIN_AVAILABLE_BYTES
+
+
 class HeavyJobLimitTimeoutError(TimeoutError):
     """Raised when a heavy-job slot cannot be acquired before its deadline."""
 
@@ -70,7 +96,11 @@ class HeavyJobLimiter:
                 while True:
                     if cancel_event is not None and cancel_event.is_set():
                         return False
-                    if self._waiters[0] is ticket and self._used + weight <= self.capacity:
+                    if (
+                        self._waiters[0] is ticket
+                        and self._used + weight <= self.capacity
+                        and _memory_admission_ok()
+                    ):
                         self._used += weight
                         self._acquired[kind] += 1
                         return True
@@ -83,6 +113,9 @@ class HeavyJobLimiter:
                         wait_for = self._cancel_poll_interval
                         if remaining is not None:
                             wait_for = min(wait_for, remaining)
+                    # 内存准入需要周期性重查 (内存恢复不触发 notify)
+                    if wait_for is None or wait_for > _MEMORY_POLL_S:
+                        wait_for = _MEMORY_POLL_S
                     self._condition.wait(wait_for)
             finally:
                 self._waiters.remove(ticket)

backend/tests/test_memory_guards.py

@@ -0,0 +1,88 @@
+"""#516 内存保护 — /health 可观测性、重任务内存准入、worker RSS 上限配置。"""
+from __future__ import annotations
+
+import threading
+
+import psutil
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from app.api import routes as routes_api
+from app.services import heavy_job_limiter as hjl
+
+
+class _VM:
+    def __init__(self, available: int):
+        self.available = available
+
+
+# ── /health 内存块 ────────────────────────────────────────────────────
+
+
+def test_health_reports_memory_snapshot():
+    app = FastAPI()
+    app.include_router(routes_api.router)
+    body = TestClient(app).get("/health").json()
+    assert body["status"] == "ok"
+    mem = body["memory"]
+    assert mem["rss_bytes"] > 0
+    assert mem["system_total_bytes"] > 0
+    assert mem["system_available_bytes"] >= 0
+    assert 0 <= mem["system_used_pct"] <= 100
+
+
+# ── 重任务内存准入 (psutil 在 _memory_admission_ok 内函数级 import,
+#    patch 真模块的 virtual_memory) ───────────────────────────────────
+
+
+def test_memory_admission_blocks_and_recovers(monkeypatch):
+    monkeypatch.setattr(hjl, "_HEAVY_MIN_AVAILABLE_BYTES", 1024**3)
+    monkeypatch.setattr(psutil, "virtual_memory", lambda: _VM(512 * 1024**2))
+    limiter = hjl.HeavyJobLimiter(capacity=2)
+    # 内存不足: 有空位也不放行, 超时返回 False
+    assert limiter.acquire("normal", timeout=0.2) is False
+    # 内存恢复 (轮询重查, 无需 notify): 正常获准
+    monkeypatch.setattr(psutil, "virtual_memory", lambda: _VM(4 * 1024**3))
+    assert limiter.acquire("normal", timeout=5.0) is True
+    limiter.release("normal")
+
+
+def test_memory_admission_disabled(monkeypatch):
+    monkeypatch.setattr(hjl, "_HEAVY_MIN_AVAILABLE_BYTES", 0)
+    limiter = hjl.HeavyJobLimiter(capacity=1)
+    assert limiter.acquire("exclusive", timeout=0.1) is True
+    limiter.release("exclusive")
+
+
+def test_memory_gate_keeps_cancel_semantics(monkeypatch):
+    monkeypatch.setattr(hjl, "_HEAVY_MIN_AVAILABLE_BYTES", 1024**3)
+    monkeypatch.setattr(psutil, "virtual_memory", lambda: _VM(0))
+    limiter = hjl.HeavyJobLimiter(capacity=1)
+    cancel = threading.Event()
+    t = threading.Timer(0.1, cancel.set)
+    t.start()
+    assert limiter.acquire("normal", timeout=5.0, cancel_event=cancel) is False
+    t.join()
+
+
+# ── worker RSS 上限配置解析 ──────────────────────────────────────────
+
+
+def test_worker_rss_limit_env_parsing(monkeypatch):
+    from app.backtest import worker
+
+    monkeypatch.setenv("TICK_WORKER_RSS_LIMIT_BYTES", "0")
+    assert worker._env_int("TICK_WORKER_RSS_LIMIT_BYTES", 123) == 0
+    monkeypatch.setenv("TICK_WORKER_RSS_LIMIT_BYTES", "2147483648")
+    assert worker._env_int("TICK_WORKER_RSS_LIMIT_BYTES", 123) == 2147483648
+    monkeypatch.delenv("TICK_WORKER_RSS_LIMIT_BYTES")
+    assert worker._env_int("TICK_WORKER_RSS_LIMIT_BYTES", 123) == 123
+    monkeypatch.setenv("TICK_WORKER_RSS_LIMIT_BYTES", "not-a-number")
+    assert worker._env_int("TICK_WORKER_RSS_LIMIT_BYTES", 123) == 123
+
+
+def test_process_rss_bytes_for_invalid_pid():
+    from app.backtest.worker import _process_rss_bytes
+
+    assert _process_rss_bytes(None) == 0
+    assert _process_rss_bytes(999999999) == 0