fix(worker): PR2 并发/竞态 High 修复 6 项（metadata保护+AnalysisReport唯一+retry退避+重复skim+heartbeat+arxiv500）

apps/worker/main.py

@@ -23,7 +23,11 @@
     run_topic_ingest,
     run_weekly_graph_maintenance,
 )
-from packages.ai.idle_processor import start_idle_processor, stop_idle_processor
+from packages.ai.idle_processor import (
+    set_dispatching,
+    start_idle_processor,
+    stop_idle_processor,
+)
 from packages.config import get_settings
 from packages.logging_setup import setup_logging
 from packages.storage.db import session_scope
@@ -33,12 +37,23 @@
 logger = logging.getLogger(__name__)
 
 _HEALTH_FILE = Path("/tmp/worker_heartbeat")
+# 心跳健康判定：最近一次心跳距现在超过此秒数视为不健康（捕获 worker 卡死/全部任务失败）
+_HEARTBEAT_STALE_SECONDS = 1200  # 20 分钟（cron job 最小间隔 30min，留足缓冲）
+
+
+def _write_heartbeat(error: str | None = None) -> None:
+    """写入心跳文件供外部健康检查（High 2e：记录最近一次错误，不再掩盖故障）。
 
+    此前无条件写时间戳，healthcheck 仅 test -f → 即使所有 job 失败 worker 仍判健康。
+    现写入 JSON {ts, error}：健康检查读 ts 判定时效，error 字段记录最近致命错误。
+    job 全部失败时不写心跳（让心跳自然过期 → healthcheck 反映故障）。
+    """
+    import json
 
-def _write_heartbeat() -> None:
-    """写入心跳文件供外部健康检查"""
     with contextlib.suppress(OSError):
-        _HEALTH_FILE.write_text(str(time.time()))
+        _HEALTH_FILE.write_text(
+            json.dumps({"ts": time.time(), "error": error[:200] if error else None})
+        )
 
 
 def _update_topic_run_status(topic_id: str, *, error: str | None) -> None:
@@ -122,22 +137,34 @@ def topic_dispatch_job() -> None:
         len(candidates),
         ", ".join(c["name"] for c in candidates),
     )
-    for c in candidates:
-        try:
-            result = _retry_with_backoff(
-                run_topic_ingest, c["id"], max_retries=_RETRY_MAX, base_delay=_RETRY_DELAY
-            )
-            logger.info(
-                "topic %s done: inserted=%s, processed=%s",
-                c["name"],
-                result.get("inserted", 0) if result else 0,
-                result.get("processed", 0) if result else 0,
-            )
-            _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()
+    # High 2d：置调度标志，idle_processor 检测到即视为繁忙，避免抢同一批论文重复处理
+    # High 2e：全部失败不写 heartbeat，让心跳自然过期 → healthcheck 反映故障
+    set_dispatching(True)
+    failures: list[str] = []
+    try:
+        for c in candidates:
+            try:
+                result = _retry_with_backoff(
+                    run_topic_ingest, c["id"], max_retries=_RETRY_MAX, base_delay=_RETRY_DELAY
+                )
+                logger.info(
+                    "topic %s done: inserted=%s, processed=%s",
+                    c["name"],
+                    result.get("inserted", 0) if result else 0,
+                    result.get("processed", 0) if result else 0,
+                )
+                _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))
+                failures.append(f"{c['name']}: {e}")
+    finally:
+        set_dispatching(False)
+    if not failures:
+        _write_heartbeat()
+    else:
+        # 全部失败时不写健康心跳，仅记录致命错误到日志（healthcheck 靠时效捕获）
+        logger.error("topic_dispatch 全部失败，跳过心跳写入：%s", failures)
 
 
 def brief_job() -> None:
@@ -160,8 +187,10 @@ def brief_job() -> None:
             result.get("saved_path", "N/A") if result else "N/A",
             result.get("email_sent", False) if result else False,
         )
-    except Exception:
-        logger.exception("Daily brief job failed after retries")
+    except Exception as e:
+        # High 2e：失败不写心跳，让健康检查靠时效捕获故障
+        logger.exception("Daily brief job failed after retries: %s", e)
+        return
     _write_heartbeat()
 
 
@@ -171,15 +200,22 @@ def weekly_graph_job() -> None:
         _retry_with_backoff(
             run_weekly_graph_maintenance, max_retries=_RETRY_MAX, base_delay=_RETRY_DELAY
         )
-    except Exception:
-        logger.exception("Weekly graph job failed after retries")
+    except Exception as e:
+        # High 2e：失败不写心跳
+        logger.exception("Weekly graph job failed after retries: %s", e)
+        return
     _write_heartbeat()
 
 
 def cs_feed_dispatch_job():
-    """每小时同步分类 + 执行订阅抓取"""
-    cs_orchestrator.sync_categories()
-    cs_orchestrator.run()
+    """每小时同步分类 + 执行订阅抓取（High 2e：失败不写心跳）"""
+    try:
+        cs_orchestrator.sync_categories()
+        cs_orchestrator.run()
+    except Exception as e:
+        logger.exception("cs_feed_dispatch failed: %s", e)
+        return
+    _write_heartbeat()
 
 
 def run_worker() -> None:

docker-compose.yml

@@ -80,7 +80,9 @@ services:
       postgres:
         condition: service_healthy
     healthcheck:
-      test: ["CMD", "test", "-f", "/tmp/worker_heartbeat"]
+      # High 2e：改判时效而非仅文件存在——所有 job 失败时心跳不再写，文件过期即不健康。
+      # 心跳超过 20 分钟（1200s）视为不健康（worker 卡死或全部任务失败）
+      test: ["CMD", "python", "-m", "scripts.worker_healthcheck"]
       interval: 30s
       timeout: 5s
       start_period: 40s

infra/migrations/versions/f6a7b8c9d0e1_analysis_report_unique.py

@@ -0,0 +1,70 @@
+"""add unique constraint on analysis_reports.paper_id
+
+Revision ID: f6a7b8c9d0e1
+Revises: e5f6a7b8c9d0
+Create Date: 2026-07-17 17:00:00.000000
+
+目的：给 analysis_reports.paper_id 加 unique 约束，防止重复 skim/deep 产生
+重复行。此前无约束，并发处理同一论文会插入多行 AnalysisReport，下游读 summary
+时取到任意一行（行为不确定）。
+
+迁移顺序：
+1. 先删除重复行（同一 paper_id 多行的，保留 created_at 最早的一行）
+2. 再加 unique 约束
+
+PG 用窗口函数删重复 + CREATE UNIQUE INDEX IF NOT EXISTS；
+SQLite 用 rowid 删重复（保留 MIN(rowid)）+ CREATE UNIQUE INDEX IF NOT EXISTS。
+"""
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "f6a7b8c9d0e1"
+down_revision = "e5f6a7b8c9d0"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if bind.dialect.name == "postgresql":
+        # 删除重复行：同一 paper_id 保留 created_at 最早的一行
+        op.execute(
+            """
+            DELETE FROM analysis_reports a USING analysis_reports b
+            WHERE a.paper_id = b.paper_id
+              AND a.id <> b.id
+              AND a.created_at > b.created_at
+            """
+        )
+        # 兜底：若仍有同 paper_id 同 created_at 的重复（极端竞态），保留 id 最小者
+        op.execute(
+            """
+            DELETE FROM analysis_reports a USING analysis_reports b
+            WHERE a.paper_id = b.paper_id
+              AND a.id > b.id
+              AND a.created_at = b.created_at
+            """
+        )
+        op.execute(
+            "CREATE UNIQUE INDEX IF NOT EXISTS uq_analysis_reports_paper_id "
+            "ON analysis_reports (paper_id)"
+        )
+    else:
+        # SQLite：用 rowid 删重复，保留 MIN(rowid)
+        op.execute(
+            """
+            DELETE FROM analysis_reports
+            WHERE rowid NOT IN (
+                SELECT MIN(rowid) FROM analysis_reports GROUP BY paper_id
+            )
+            """
+        )
+        op.execute(
+            "CREATE UNIQUE INDEX IF NOT EXISTS uq_analysis_reports_paper_id "
+            "ON analysis_reports (paper_id)"
+        )
+
+
+def downgrade() -> None:
+    op.execute("DROP INDEX IF EXISTS uq_analysis_reports_paper_id")

packages/ai/daily_runner.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import TYPE_CHECKING
 from uuid import UUID
@@ -167,6 +168,23 @@ def run_topic_ingest(topic_id: str, progress_callback: callable | None = None) -
                 break
             except Exception as exc:
                 last_error = str(exc)
+                # 修 High：内层 retry 此前无 sleep 无退避，失败后立即重发请求，限流场景
+                # 下加速触发 429。改指数退避；429/限流类错误用更长退避，其余快速失败重试。
+                if _attempt < topic.retry_limit:
+                    is_rate_limited = any(
+                        tok in str(exc).lower()
+                        for tok in ("429", "rate limit", "限流", "timeout", "timed out")
+                    )
+                    delay = 10.0 * (2**_attempt) if is_rate_limited else 3.0 * (2**_attempt)
+                    logger.warning(
+                        "topic %s 抓取失败 (attempt %d/%d): %s — %.0fs 后重试",
+                        topic_name,
+                        attempts,
+                        topic.retry_limit + 1,
+                        str(exc)[:120],
+                        delay,
+                    )
+                    time.sleep(delay)
 
         if last_error is not None:
             return {

packages/ai/idle_processor.py

@@ -19,6 +19,22 @@
 
 logger = logging.getLogger(__name__)
 
+# 进程内调度标志（High 2d）：topic_dispatch 抓取/处理期间置 True，
+# idle 检测读到即视为繁忙，避免 idle 与 topic_dispatch 抢同一批 unread 论文重复
+# embed/skim。仅 worker 进程内生效（idle_processor 与 topic_dispatch 同在 worker 容器）。
+_dispatching = False
+
+
+def set_dispatching(value: bool) -> None:
+    """设置 topic_dispatch 是否正在跑（供 worker main 调用）"""
+    global _dispatching
+    _dispatching = value
+
+
+def is_dispatching() -> bool:
+    """查询 topic_dispatch 是否正在跑"""
+    return _dispatching
+
 
 class IdleDetector:
     """
@@ -90,6 +106,11 @@ def is_idle(self) -> bool:
         Returns:
             bool: 是否空闲
         """
+        # High 2d：topic_dispatch 正在抓取/处理时不算空闲，避免与 idle 抢同一批论文
+        if is_dispatching():
+            logger.debug("topic_dispatch 正在跑，不满足空闲条件")
+            return False
+
         # 检查距离上次任务执行的时间
         if time.time() - self._last_task_time < self.idle_interval:
             return False

packages/integrations/arxiv_client.py

@@ -75,10 +75,6 @@ def fetch_latest(
         days_back 默认 0 = 不加日期过滤（否则经典老论文如 OpenShape/Uni3D 都会被筛掉）。
         订阅/定时任务需要最新增量时，由调用方显式传 days_back。
         """
-        # 获取速率限制许可（10 秒超时）
-        if not acquire_api("arxiv", timeout=10.0):
-            raise httpx.TimeoutException("ArXiv 速率限制等待超时，请稍后重试")
-
         structured_query = _build_arxiv_query(query, days_back)
         logger.info(
             "ArXiv search: %s → %s (sort=%s start=%d days_back=%d)",
@@ -96,9 +92,13 @@ def fetch_latest(
             "max_results": max_results,
         }
         # 自动重试（429 限流 + 网络抖动 + 500 回退）
+        # 修 High 2f：acquire_api 移入循环内，每次请求都重新获取限流许可
+        # （此前循环外只 acquire 一次，500 回退的二次请求绕过限流器）
         last_exc: Exception | None = None
         for attempt in range(3):
             try:
+                if not acquire_api("arxiv", timeout=10.0):
+                    raise httpx.TimeoutException("ArXiv 速率限制等待超时，请稍后重试")
                 response = self.client.get(ARXIV_API_URL, params=params)
                 response.raise_for_status()
                 return self._parse_atom(response.text)
@@ -112,13 +112,12 @@ def fetch_latest(
                     time.sleep(wait)
                     continue
                 elif status == 500 and "submittedDate:" in structured_query:
-                    # arXiv API 日期过滤可能有问题，尝试不带日期的查询
-                    logger.warning("ArXiv 500 错误（可能是日期过滤问题），尝试不带日期的查询")
-                    simple_query = _build_arxiv_query(query, days_back=0)  # 不添加日期
-                    params["search_query"] = simple_query
-                    response = self.client.get(ARXIV_API_URL, params=params)
-                    response.raise_for_status()
-                    return self._parse_atom(response.text)
+                    # arXiv API 日期过滤可能有问题，改不带日期的查询重试。
+                    # continue 回循环顶部重新 acquire_api（限流），二次失败由循环统一处理
+                    logger.warning("ArXiv 500 错误（可能是日期过滤问题），切无日期查询重试")
+                    structured_query = _build_arxiv_query(query, days_back=0)
+                    params["search_query"] = structured_query
+                    continue
                 raise
             except httpx.TimeoutException as exc:
                 last_exc = exc

packages/storage/models.py

@@ -85,6 +85,7 @@ class AnalysisReport(Base):
         ForeignKey("papers.id", ondelete="CASCADE"),
         nullable=False,
         index=True,
+        unique=True,
     )
     summary_md: Mapped[str | None] = mapped_column(Text, nullable=True)
     deep_dive_md: Mapped[str | None] = mapped_column(Text, nullable=True)

packages/storage/repositories/analysis.py

@@ -8,6 +8,7 @@
 from typing import TYPE_CHECKING
 
 from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
 
 if TYPE_CHECKING:
     from uuid import UUID
@@ -57,7 +58,14 @@ def _get_or_create(self, paper_id: UUID) -> AnalysisReport:
             return found
         report = AnalysisReport(paper_id=pid, key_insights={})
         self.session.add(report)
-        self.session.flush()
+        try:
+            self.session.flush()
+        except IntegrityError:
+            # 并发 skim 同一论文时，另一事务已插入行（paper_id 现为 unique）。
+            # 回滚本事务未提交改动并取已存在的行，避免重复插入并防止抛 IntegrityError
+            # 中断 skim 流程。此前无 unique 约束 → 重复 skim 产生重复行。
+            self.session.rollback()
+            return self.session.execute(q).scalar_one()
         return report
 
     def summaries_for_papers(self, paper_ids: list[str]) -> dict[str, str]:

packages/storage/repositories/paper.py

@@ -45,7 +45,16 @@ def upsert_paper(self, data: PaperCreate) -> Paper:
             existing.title = data.title
             existing.abstract = data.abstract
             existing.publication_date = data.publication_date
-            existing.metadata_json = data.metadata
+            # 修 High：此前 existing.metadata_json = data.metadata（整体覆盖）会把 skim
+            # 写入的 keywords/title_zh/abstract_zh 抹掉——重复抓取同一论文时丢失已花钱
+            # 算出来的 skim 产物。改为合并：保留已有的 skim 派生字段，其余由 arxiv 原始
+            # 元数据更新覆盖（categories/authors/source 等原始字段）
+            new_meta = dict(data.metadata or {})
+            old_meta = existing.metadata_json or {}
+            for skim_key in ("keywords", "title_zh", "abstract_zh"):
+                if skim_key in old_meta and old_meta[skim_key]:
+                    new_meta[skim_key] = old_meta[skim_key]
+            existing.metadata_json = new_meta
             existing.updated_at = datetime.now(UTC)
             self.session.flush()
             return existing