fix(worker): PR1 卡死/失效 Critical 修复 6 项（限流器+id查询+TokenBucket+daily_limit+last_run_at+补偿）

apps/worker/main.py

@@ -41,6 +41,19 @@ def _write_heartbeat() -> None:
         _HEALTH_FILE.write_text(str(time.time()))
 
 
+def _update_topic_run_status(topic_id: str, *, error: str | None) -> None:
+    """记录主题抓取的最近运行时间与错误（Critical #4：失败可查可补抓）。
+
+    抓取失败此前静默无痕，定位不到出问题的主题。这里在每次抓取后持久化
+    last_run_at / last_error，失败信息入库便于排查与补抓。
+    """
+    try:
+        with session_scope() as session:
+            TopicRepository(session).update_run_status(topic_id, error=error)
+    except Exception:
+        logger.exception("Failed to persist topic run status for %s", topic_id)
+
+
 def _retry_with_backoff(fn, *args, max_retries: int = 3, base_delay: float = 5.0, **kwargs):
     """带指数退避的重试执行"""
     for attempt in range(max_retries):
@@ -120,8 +133,10 @@ def topic_dispatch_job() -> None:
                 result.get("inserted", 0) if result else 0,
                 result.get("processed", 0) if result else 0,
             )
-        except Exception:
+            _update_topic_run_status(c["id"], error=None)
+        except Exception as e:
             logger.exception("topic_dispatch failed for %s", c["name"])
+            _update_topic_run_status(c["id"], error=str(e))
     _write_heartbeat()
 
 

infra/migrations/versions/e5f6a7b8c9d0_add_topic_run_status.py

@@ -0,0 +1,53 @@
+"""add topic_subscriptions.last_run_at / last_error for fetch failure tracking
+
+Revision ID: e5f6a7b8c9d0
+Revises: d4e5f6a7b8c9
+Create Date: 2026-07-17 16:00:00.000000
+
+目的：给 topic_subscriptions 加 last_run_at / last_error，记录主题抓取的最近运行
+时间与错误信息。此前抓取失败静默无痕，无法定位失败主题、无法补抓。
+PG 用 ADD COLUMN IF NOT EXISTS，SQLite 用 try/except 兜底，保持幂等（对齐 c3d4e5f6a7b8 写法）。
+"""
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "e5f6a7b8c9d0"
+down_revision = "d4e5f6a7b8c9"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    # PG 支持 ADD COLUMN IF NOT EXISTS；SQLite 不支持，用 try/except 兜底
+    if bind.dialect.name == "postgresql":
+        op.execute(
+            "ALTER TABLE topic_subscriptions ADD COLUMN IF NOT EXISTS last_run_at TIMESTAMP NULL"
+        )
+        op.execute(
+            "ALTER TABLE topic_subscriptions ADD COLUMN IF NOT EXISTS last_error VARCHAR(500) NULL"
+        )
+    else:
+        for ddl in (
+            "last_run_at TIMESTAMP NULL",
+            "last_error VARCHAR(500) NULL",
+        ):
+            try:
+                op.execute(f"ALTER TABLE topic_subscriptions ADD COLUMN {ddl}")
+            except Exception:
+                pass
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name == "postgresql":
+        op.execute("ALTER TABLE topic_subscriptions DROP COLUMN IF EXISTS last_error")
+        op.execute("ALTER TABLE topic_subscriptions DROP COLUMN IF EXISTS last_run_at")
+    else:
+        # SQLite DROP COLUMN 支持参差，try/except 兜底
+        for col in ("last_error", "last_run_at"):
+            try:
+                op.execute(f"ALTER TABLE topic_subscriptions DROP COLUMN {col}")
+            except Exception:
+                pass

packages/ai/cs_feed_orchestrator.py

@@ -29,13 +29,16 @@ def __init__(self, size: int, fill_rate: int):
         self.lock = threading.Lock()
 
     def acquire(self, timeout: float = 60) -> bool:
+        # 修超时判定 bug：此前用 last_refill（每次 _refill 都更新为 now），time.time()-last_refill
+        # 永远≈0，永不超时 → arxiv 全局限流时 acquire 卡死 worker 线程。改用 start_time 判定
+        start_time = time.time()
         while True:
             with self.lock:
                 self._refill()
                 if self.tokens >= 1:
                     self.tokens -= 1
                     return True
-            if time.time() - self.last_refill > timeout:
+            if time.time() - start_time > timeout:
                 return False
             time.sleep(1)
 

packages/ai/idle_processor.py

@@ -170,6 +170,26 @@ def _get_unread_papers(self, limit: int = 10) -> list[tuple[str, str]]:
             ).all()
             return [(str(p.id), p.title) for p in papers]
 
+    def _get_stuck_skimmed_papers(self, limit: int = 3) -> list[tuple[str, str]]:
+        """获取已 skim 但卡住未精读的论文（Critical #6 补偿）
+
+        之前只挑 unread 无 AnalysisReport 的论文走 embed+skim，但 skim 之后
+        read_status 变 skimmed、deep_dive_md 仍为空——这些论文卡在 skimmed 永远
+        不会被闲时补偿精读。这里挑出 skimmed 且 AnalysisReport 有 summary_md 但
+        deep_dive_md 为空的论文，单独走 deep_dive 补偿，配额受限避免一次补偿太多。
+        """
+        with session_scope() as session:
+            papers = session.execute(
+                select(Paper.id, Paper.title)
+                .where(Paper.read_status == "skimmed")
+                .join(AnalysisReport, Paper.id == AnalysisReport.paper_id)
+                .where(AnalysisReport.summary_md.is_not(None))
+                .where(AnalysisReport.deep_dive_md.is_(None))
+                .order_by(Paper.created_at.asc())  # 优先处理旧的
+                .limit(limit)
+            ).all()
+            return [(str(p.id), p.title) for p in papers]
+
     def _process_batch(self) -> int:
         """
         处理一批论文（带任务追踪）
@@ -277,7 +297,56 @@ def _process_batch(self) -> int:
         self._papers_processed += processed
         self.detector.mark_task_executed()
 
-        return processed
+        # Critical #6 补偿：对已 skim 但卡住未精读的论文补一次精读（独立配额受限）
+        deep_compensated = self._compensate_stuck_skimmed()
+        return processed + deep_compensated
+
+    def _compensate_stuck_skimmed(self) -> int:
+        """补偿已 skim 但未精读的论文（Critical #6）。
+
+        skim 完成后 read_status 变 skimmed，但 deep_dive_md 仍空的论文此前无人再
+        触发精读，永久卡在 skimmed。这里在 skim 批次跑完后单独补一批精读，配额严格
+        受限（deep_read_compensation，默认 2），避免闲时一次精读太多拖垮 LLM 速率。
+        """
+        # 闲时精读配额：保守默认 2，可通过 settings 调整
+        quota = getattr(get_settings(), "deep_read_compensation", 2)
+        if quota <= 0:
+            return 0
+
+        stuck = self._get_stuck_skimmed_papers(limit=quota)
+        if not stuck:
+            return 0
+
+        logger.info("🔧 闲时补偿精读：%d 篇卡在 skimmed 的论文", len(stuck))
+        limiter = get_rate_limiter()
+        pipelines = PaperPipelines()
+        compensated = 0
+
+        for paper_id, title in stuck:
+            # 繁忙时放弃剩余补偿，避免与用户请求争抢 LLM
+            if not self.detector.is_idle():
+                logger.warning("系统不再空闲，中止 skimmed 补偿")
+                break
+            if not limiter.start_task():
+                logger.debug("并发数已达上限，中止 skimmed 补偿")
+                break
+            try:
+                # 精读走 LLM 限流；失败不抛出，下一篇继续
+                if not acquire_api("llm", timeout=30.0):
+                    logger.warning("LLM API 限流，跳过精读补偿：%s", title[:40])
+                    continue
+                try:
+                    pipelines.deep_dive(paper_id)
+                    compensated += 1
+                    logger.info("✅ 闲时补偿精读完成：%s", title[:40])
+                except Exception as e:
+                    logger.warning("闲时补偿精读失败：%s - %s", title[:40], e)
+            finally:
+                limiter.end_task()
+            time.sleep(1)
+
+        logger.info("📊 skimmed 补偿完成：精读=%d/%d", compensated, len(stuck))
+        return compensated
 
     def _run_loop(self):
         """主循环"""

packages/ai/rate_limiter.py

@@ -5,31 +5,106 @@
 
 from __future__ import annotations
 
+import contextlib
+import json
 import logging
+import os
 import time
 from datetime import UTC, datetime
 from threading import Lock
+from typing import Any
 
 from packages.config import get_settings
 
 logger = logging.getLogger(__name__)
 
 
+def _shared_state_dir() -> Any:
+    """获取跨进程共享状态目录（backend/api 与 worker 容器挂载同一卷）。
+
+    无写权限或目录不可用时返回 None，调用方降级到进程内内存桶。
+    """
+    try:
+        d = get_settings().rate_limiter_state_dir
+        d.mkdir(parents=True, exist_ok=True)
+        return d
+    except Exception:  # 配置缺失 / 无写权限
+        return None
+
+
 class TokenBucket:
-    """令牌桶算法实现"""
+    """令牌桶算法实现（跨进程共享版本）
 
-    def __init__(self, rate: float = 10.0, capacity: int = 20):
+    状态持久化到共享状态文件，通过 fcntl.flock(LOCK_EX) 互斥，使 api/worker
+    两个进程共享同一令牌桶。此前每个进程各持一份内存桶，实际速率是配置值的 2 倍，
+    arxiv 全局限流时容易触发 429。无共享文件或写失败时降级到进程内内存桶（行为
+    与旧版一致），不会因共享层故障而阻塞业务。
+    """
+
+    def __init__(self, rate: float = 10.0, capacity: int = 20, api_type: str = ""):
         """
         Args:
             rate: 令牌生成速率 (个/秒)
             capacity: 桶容量 (最大令牌数)
+            api_type: API 类型，用于命名共享状态文件（如 arxiv → pm_rate_arxiv.state）
         """
         self.rate = rate
         self.capacity = capacity
+        self.api_type = api_type
+        # 进程内内存桶（共享文件不可用时降级使用，也作 _refill 兜底初值）
         self.tokens = float(capacity)
         self.last_update = time.time()
         self._lock = Lock()
 
+        # 跨进程共享状态文件 + flock 互斥
+        self._state_path: Any = None
+        self._state_fp = None
+        self._shared = False
+        state_dir = _shared_state_dir()
+        if state_dir is not None and api_type:
+            path = state_dir / f"pm_rate_{api_type}.state"
+            try:
+                # 句柄保持打开以复用 flock；用 a+b：不存在则创建，flock 对同一
+                # inode 生效（跨容器共享卷同主机 inode）。close() 释放。
+                self._state_fp = open(path, "a+b")  # noqa: SIM115
+                self._state_path = path
+                self._shared = True
+            except OSError:
+                logger.warning("TokenBucket(%s): 共享状态文件不可写，降级进程内内存桶", api_type)
+                self._shared = False
+
+    def close(self) -> None:
+        """释放共享状态文件句柄"""
+        if self._state_fp is not None:
+            with contextlib.suppress(OSError):
+                self._state_fp.close()
+            self._state_fp = None
+            self._shared = False
+
+    def __del__(self):
+        self.close()
+
+    def _read_shared(self) -> tuple[float, float]:
+        """从共享文件读 (tokens, last_update)。文件空/损坏时用当前内存值初始化。"""
+        self._state_fp.seek(0)
+        raw = self._state_fp.read()
+        if not raw:
+            return float(self.capacity), time.time()
+        try:
+            data = json.loads(raw)
+            return float(data.get("tokens", self.capacity)), float(
+                data.get("last_update", time.time())
+            )
+        except (ValueError, TypeError):
+            return float(self.capacity), time.time()
+
+    def _write_shared(self, tokens: float, last_update: float) -> None:
+        self._state_fp.seek(0)
+        self._state_fp.truncate()
+        self._state_fp.write(json.dumps({"tokens": tokens, "last_update": last_update}).encode())
+        self._state_fp.flush()
+        os.fsync(self._state_fp.fileno())
+
     def acquire(self, tokens: int = 1, timeout: float | None = None) -> bool:
         """获取令牌
 
@@ -43,17 +118,36 @@ def acquire(self, tokens: int = 1, timeout: float | None = None) -> bool:
         start_time = time.time()
 
         while True:
-            with self._lock:
-                now = time.time()
-                # 补充令牌
-                elapsed = now - self.last_update
-                self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
-                self.last_update = now
-
-                # 尝试获取
-                if self.tokens >= tokens:
-                    self.tokens -= tokens
-                    return True
+            # 共享路径：flock 互斥，跨进程读写同一桶状态
+            if self._shared:
+                import fcntl
+
+                with self._lock:  # 进程内串行化（避免同进程多线程争抢 flock）
+                    fcntl.flock(self._state_fp.fileno(), fcntl.LOCK_EX)
+                    try:
+                        now = time.time()
+                        cur_tokens, last_update = self._read_shared()
+                        elapsed = now - last_update
+                        cur_tokens = min(self.capacity, cur_tokens + elapsed * self.rate)
+                        if cur_tokens >= tokens:
+                            cur_tokens -= tokens
+                            self._write_shared(cur_tokens, now)
+                            # 同步进程内缓存（get_available_tokens 读它）
+                            self.tokens = cur_tokens
+                            self.last_update = now
+                            return True
+                    finally:
+                        fcntl.flock(self._state_fp.fileno(), fcntl.LOCK_UN)
+            else:
+                # 降级路径：进程内内存桶（与旧版行为一致）
+                with self._lock:
+                    now = time.time()
+                    elapsed = now - self.last_update
+                    self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
+                    self.last_update = now
+                    if self.tokens >= tokens:
+                        self.tokens -= tokens
+                        return True
 
             # 检查超时
             if timeout is not None and (time.time() - start_time) >= timeout:
@@ -63,7 +157,19 @@ def acquire(self, tokens: int = 1, timeout: float | None = None) -> bool:
             time.sleep(0.1)
 
     def get_available_tokens(self) -> float:
-        """获取当前可用令牌数"""
+        """获取当前可用令牌数（跨进程读，非强一致）"""
+        if self._shared:
+            import fcntl
+
+            with self._lock:
+                fcntl.flock(self._state_fp.fileno(), fcntl.LOCK_SH)
+                try:
+                    now = time.time()
+                    cur_tokens, last_update = self._read_shared()
+                    elapsed = now - last_update
+                    return min(self.capacity, cur_tokens + elapsed * self.rate)
+                finally:
+                    fcntl.flock(self._state_fp.fileno(), fcntl.LOCK_UN)
         with self._lock:
             now = time.time()
             elapsed = now - self.last_update
@@ -95,12 +201,12 @@ class APIRateLimiter:
     def __init__(self):
         self.settings = get_settings()
 
-        # 初始化令牌桶（多个 API 独立限流）
+        # 初始化令牌桶（多个 API 独立限流）；api_type 用于命名跨进程共享状态文件
         self._buckets = {
-            "llm": TokenBucket(rate=5.0, capacity=10),
-            "arxiv": TokenBucket(rate=2.0, capacity=5),
-            "embedding": TokenBucket(rate=3.0, capacity=8),
-            "vision": TokenBucket(rate=1.0, capacity=3),
+            "llm": TokenBucket(rate=5.0, capacity=10, api_type="llm"),
+            "arxiv": TokenBucket(rate=2.0, capacity=5, api_type="arxiv"),
+            "embedding": TokenBucket(rate=3.0, capacity=8, api_type="embedding"),
+            "vision": TokenBucket(rate=1.0, capacity=3, api_type="vision"),
         }
 
         # 当前并发配置