From d2f6e10c94f3145341383f3637d33ebe8417e7ce Mon Sep 17 00:00:00 2001 From: Color2333 <1552429809@qq.com> Date: Sat, 18 Jul 2026 19:23:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(worker):=20PR1=20=E5=8D=A1=E6=AD=BB/?= =?UTF-8?q?=E5=A4=B1=E6=95=88=20Critical=20=E4=BF=AE=E5=A4=8D=206=20?= =?UTF-8?q?=E9=A1=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1a 跨进程共享限流器：rate_limiter.py TokenBucket.acquire 改 fcntl.flock + 共享状态文件（pm_rate_{api}.state），api/worker 两容器共享同一令牌桶， 此前各持内存桶实际速率翻倍易触发 429。无写权限时降级进程内内存桶（行为同旧版）。 新增 rate_limiter_state_dir 配置（默认 /app/data 共享卷）。 1b ingest.py id: 查询改 fetch_by_ids：arxiv 不支持 id: 作为 search_query 前缀，逐个 fetch_latest(query=f"id:{mid}") 永远拿不到目标论文。改一次批量 fetch_by_ids（走 id_list 参数，正确入口）。 1c cs_feed TokenBucket 超时 bug：acquire 超时判定用 last_refill（每次 _refill 更新为 now），time.time()-last_refill 永远≈0 永不超时 → arxiv 全局限流时 acquire 卡死 worker 线程。改用 start_time 判定。 1d daily_limit 累加 + 跨天重置：cs_feed update_run_status 此前 last_run_count = count（覆盖），当日多次抓取重置配额绕过 daily_limit。 改累加；跨天先清零，避免昨天余量带进今天。 1e TopicSubscription 加 last_run_at/last_error：抓取失败此前静默无痕无法 补抓。加字段 + alembic 迁移（幂等）+ TopicRepository.update_run_status + worker topic_dispatch_job 成功/失败后更新。 1f idle_processor skimmed 论文补偿：skim 后 read_status 变 skimmed、 deep_dive_md 仍空的论文此前无人再触发精读，永久卡在 skimmed。新增 _get_stuck_skimmed_papers + _compensate_stuck_skimmed（配额受限，默认2）， 在 skim 批次跑完后单独补一次精读。 测试：新增 TopicRepository.update_run_status（持久化+截断+清空）、 CSFeedRepository.update_run_status（当日累加+跨天重置+未知静默）共 5 测试， 全套 50 passed。 --- apps/worker/main.py | 17 ++- .../e5f6a7b8c9d0_add_topic_run_status.py | 53 +++++++ packages/ai/cs_feed_orchestrator.py | 5 +- packages/ai/idle_processor.py | 71 ++++++++- packages/ai/rate_limiter.py | 144 +++++++++++++++--- packages/ai/tools/handlers/ingest.py | 11 +- packages/config.py | 9 ++ packages/storage/models.py | 4 + packages/storage/repositories/cs_feed.py | 9 +- packages/storage/repositories/topic.py | 9 ++ tests/test_repositories.py | 77 +++++++++- 11 files changed, 379 insertions(+), 30 deletions(-) create mode 100644 infra/migrations/versions/e5f6a7b8c9d0_add_topic_run_status.py diff --git a/apps/worker/main.py b/apps/worker/main.py index d8f116b..e372698 100644 --- a/apps/worker/main.py +++ b/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() diff --git a/infra/migrations/versions/e5f6a7b8c9d0_add_topic_run_status.py b/infra/migrations/versions/e5f6a7b8c9d0_add_topic_run_status.py new file mode 100644 index 0000000..e56b527 --- /dev/null +++ b/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 diff --git a/packages/ai/cs_feed_orchestrator.py b/packages/ai/cs_feed_orchestrator.py index fc1f31c..9d42e4e 100644 --- a/packages/ai/cs_feed_orchestrator.py +++ b/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) diff --git a/packages/ai/idle_processor.py b/packages/ai/idle_processor.py index a1f727a..9cf1207 100644 --- a/packages/ai/idle_processor.py +++ b/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): """主循环""" diff --git a/packages/ai/rate_limiter.py b/packages/ai/rate_limiter.py index 28fe2fe..4db57f6 100644 --- a/packages/ai/rate_limiter.py +++ b/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"), } # 当前并发配置 diff --git a/packages/ai/tools/handlers/ingest.py b/packages/ai/tools/handlers/ingest.py index 3a0ef6d..f846415 100644 --- a/packages/ai/tools/handlers/ingest.py +++ b/packages/ai/tools/handlers/ingest.py @@ -129,15 +129,16 @@ def _ingest_arxiv( all_papers = arxiv_client.fetch_latest(query=query, max_results=50) selected_papers = [p for p in all_papers if p.arxiv_id in selected_set] - # 补充搜索结果中没有的（可能 ID 不在前50条中），逐个按 ID 拉取 + # 补充搜索结果中没有的（可能 ID 不在前50条中），批量按 ID 拉取 + # 此前用 fetch_latest(query=f"id:{mid}") —— arxiv 不支持 id: 作为 search_query 前缀， + # 永远拿不到目标论文。改用 fetch_by_ids（走 id_list 参数，正确入口），一次批量查 found_ids = {p.arxiv_id for p in selected_papers} missing_ids = selected_set - found_ids - for mid in missing_ids: + if missing_ids: try: - extra = arxiv_client.fetch_latest(query=f"id:{mid}", max_results=1) - selected_papers.extend(extra) + selected_papers.extend(arxiv_client.fetch_by_ids(list(missing_ids))) except Exception: - logger.warning("Failed to fetch arxiv paper %s", mid) + logger.warning("Failed to fetch arxiv papers by ids: %s", list(missing_ids)[:5]) failed_papers: list[dict] = [] ingested_papers: list[dict] = [] diff --git a/packages/config.py b/packages/config.py index 73948b5..c3a5bdb 100644 --- a/packages/config.py +++ b/packages/config.py @@ -4,6 +4,7 @@ @author Color2333 """ +import contextlib import os from functools import lru_cache from pathlib import Path @@ -32,6 +33,9 @@ class Settings(BaseSettings): database_url: str = "sqlite:////app/data/papermind.db" pdf_storage_root: Path = Path("./data/papers") brief_output_root: Path = Path("./data/briefs") + # 跨进程共享限流器状态目录：backend/api 与 worker 容器需挂载同一目录才能共享令牌桶。 + # 默认 /app/data 是 docker-compose 中两个服务共享的 pm_data 数据卷挂载点。 + rate_limiter_state_dir: Path = Path("/app/data") skim_score_threshold: float = 0.65 daily_cron: str = "0 21 * * *" weekly_cron: str = "0 22 * * 0" @@ -64,6 +68,8 @@ class Settings(BaseSettings): # Worker 调度 worker_retry_max: int = 2 worker_retry_base_delay: float = 5.0 + # 闲时补偿精读配额：闲时对已 skim 但卡住未精读的论文补一次精读的上限（0 禁用） + deep_read_compensation: int = 2 # 并发与缓存 paper_concurrency: int = 5 @@ -99,6 +105,9 @@ def get_settings() -> Settings: settings = Settings() settings.pdf_storage_root.mkdir(parents=True, exist_ok=True) settings.brief_output_root.mkdir(parents=True, exist_ok=True) + # 跨进程限流器状态目录（容器共享卷），无写权限时静默降级到进程内内存桶 + with contextlib.suppress(OSError): + settings.rate_limiter_state_dir.mkdir(parents=True, exist_ok=True) # SQLite 需要预先创建数据库文件所在目录；PostgreSQL 等远程库无需 if settings.database_url.startswith("sqlite:"): db_parent = Path(settings.database_url.replace("sqlite:///", "")).parent diff --git a/packages/storage/models.py b/packages/storage/models.py index 35607d8..81274e0 100644 --- a/packages/storage/models.py +++ b/packages/storage/models.py @@ -238,6 +238,10 @@ class TopicSubscription(Base): nullable=False, default=7 ) # 日期范围（最近 N 天） + # 抓取状态追踪（修 Critical #4：此前无 last_run_at，抓取失败静默无痕无法补抓） + last_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + last_error: Mapped[str | None] = mapped_column(String(500), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow, nullable=False) updated_at: Mapped[datetime] = mapped_column( DateTime, default=_utcnow, onupdate=_utcnow, nullable=False diff --git a/packages/storage/repositories/cs_feed.py b/packages/storage/repositories/cs_feed.py index 0463005..4dd2db3 100644 --- a/packages/storage/repositories/cs_feed.py +++ b/packages/storage/repositories/cs_feed.py @@ -74,8 +74,13 @@ def delete_subscription(self, category_code: str) -> bool: def update_run_status(self, category_code: str, count: int): sub = self.get_subscription(category_code) if sub: - sub.last_run_at = datetime.now(UTC) - sub.last_run_count = count + now = datetime.now(UTC) + # 修 daily_limit 失效 bug：此前 last_run_count = count（覆盖），当日多次抓取会重置配额， + # 绕过 daily_limit。改累加；跨天先清零，避免昨天余量带进今天 + if sub.last_run_at is not None and sub.last_run_at.date() != now.date(): + sub.last_run_count = 0 + sub.last_run_at = now + sub.last_run_count = (sub.last_run_count or 0) + count sub.status = "active" self.session.commit() diff --git a/packages/storage/repositories/topic.py b/packages/storage/repositories/topic.py index a2b4852..43189c9 100644 --- a/packages/storage/repositories/topic.py +++ b/packages/storage/repositories/topic.py @@ -110,6 +110,15 @@ def update_topic( self.session.flush() return topic + def update_run_status(self, topic_id: str, *, error: str | None = None) -> None: + """记录主题最近一次抓取的时间与错误（Critical #4：失败有持久化痕迹，可查可补抓）""" + topic = self.session.get(TopicSubscription, topic_id) + if topic is None: + return + topic.last_run_at = datetime.now(UTC) + topic.last_error = None if error is None else error[:500] + self.session.commit() + def delete_topic(self, topic_id: str) -> None: topic = self.session.get(TopicSubscription, topic_id) if topic is not None: diff --git a/tests/test_repositories.py b/tests/test_repositories.py index 8517f75..05f5b52 100644 --- a/tests/test_repositories.py +++ b/tests/test_repositories.py @@ -6,11 +6,12 @@ from __future__ import annotations -from datetime import date +from datetime import UTC, date, datetime, timedelta from packages.domain.enums import ReadStatus from packages.domain.schemas import PaperCreate from packages.storage.repositories import ( + CSFeedRepository, IeeeQuotaRepository, PaperRepository, TopicRepository, @@ -123,6 +124,42 @@ def test_get_by_name(self, db_session): assert found.query == "computer vision" assert repo.get_by_name("nonexistent") is None + def test_update_run_status_persists_last_run_at_and_error(self, db_session): + """update_run_status 持久化 last_run_at/last_error（Critical #4：抓取失败可查可补抓）""" + repo = TopicRepository(db_session) + topic = repo.upsert_topic(name="FailTopic", query="q") + tid = topic.id + db_session.flush() + + # 成功：error=None，清空 last_error，写入 last_run_at + repo.update_run_status(tid, error=None) + db_session.refresh(topic) + assert topic.last_run_at is not None + assert topic.last_error is None + + # 失败：error 写入并截断到 500 + long_err = "x" * 800 + repo.update_run_status(tid, error=long_err) + db_session.refresh(topic) + assert topic.last_error is not None + assert len(topic.last_error) == 500 + assert topic.last_run_at is not None + + # 未知 topic_id：静默跳过，不抛 + repo.update_run_status("nonexistent-id", error="boom") + + def test_update_run_status_allows_next_error_clears_previous(self, db_session): + """连续抓取：上次错误在下次成功时被清空""" + repo = TopicRepository(db_session) + topic = repo.upsert_topic(name="Recover", query="q") + db_session.flush() + repo.update_run_status(topic.id, error="first fail") + db_session.refresh(topic) + assert topic.last_error == "first fail" + repo.update_run_status(topic.id, error=None) + db_session.refresh(topic) + assert topic.last_error is None + class TestIeeeQuotaRepository: def test_quota_lifecycle_with_topic(self, db_session): @@ -145,3 +182,41 @@ def test_quota_lifecycle_with_topic(self, db_session): # 重置 quota_repo.reset_quota(tid, today, new_limit=5) assert quota_repo.get_remaining(tid, today) == 5 + + +class TestCSFeedRepository: + def test_update_run_status_accumulates_same_day(self, db_session): + """当日多次抓取累加（Critical #3：覆盖 bug 会绕过 daily_limit）""" + repo = CSFeedRepository(db_session) + repo.upsert_subscription(category_code="cs.AI", daily_limit=30) + db_session.flush() + + repo.update_run_status("cs.AI", count=10) + sub = repo.get_subscription("cs.AI") + assert sub.last_run_count == 10 + + # 同日第二次抓取：应累加而非覆盖 + repo.update_run_status("cs.AI", count=5) + db_session.refresh(sub) + assert sub.last_run_count == 15 + + def test_update_run_status_resets_across_day(self, db_session): + """跨天抓取先清零，避免昨天余量带进今天（Critical #3）""" + repo = CSFeedRepository(db_session) + repo.upsert_subscription(category_code="cs.LG", daily_limit=30) + sub = repo.get_subscription("cs.LG") + # 模拟昨天已抓 20，last_run_at 设为昨天 + sub.last_run_at = datetime.now(UTC) - timedelta(days=1) + sub.last_run_count = 20 + db_session.commit() + + # 今天再抓 8：跨天应先清零，再累加 → 8（而非 20+8=28） + repo.update_run_status("cs.LG", count=8) + db_session.refresh(sub) + assert sub.last_run_count == 8 + assert sub.status == "active" + + def test_update_run_status_unknown_category_silent(self, db_session): + """未知 category_code：静默跳过，不抛""" + repo = CSFeedRepository(db_session) + repo.update_run_status("cs.NONEXIST", count=5)