fix(worker): PR2 并发/竞态 High 修复 6 项（metadata保护+AnalysisReport唯一+retry退避+重复skim+heartbeat+arxiv500） - #49
Merged
Merged
Conversation
2a upsert_paper 保护 skim metadata：更新分支此前
existing.metadata_json = data.metadata（整体覆盖）会把 skim 写入的
keywords/title_zh/abstract_zh 抹掉，重复抓取丢失已花钱算出的产物。
改合并：保留已有 skim 派生字段，其余由 arxiv 原始元数据更新覆盖。
2b AnalysisReport 加 unique + IntegrityError 处理：paper_id 加
unique=True，并发 skim 同一论文不再产生重复行。_get_or_create 加
IntegrityError 捕获：flush 命中 unique 约束时回滚并取已存在行。
alembic 迁移先删重复行（保留最早）再加唯一索引（幂等）。
2c daily_runner 内层 retry 加退避：此前 for 循环无 sleep 无退避，
失败后立即重发加速触发 429。加指数退避；429/限流类用更长退避
（10*2^n），其余用 3*2^n。
2d idle_processor 防重复处理：新增进程内 _dispatching 标志 +
set_dispatching/is_dispatching。worker topic_dispatch 期间置 True，
IdleDetector.is_idle 读到即视为繁忙，避免 idle 与 topic_dispatch
抢同一批 unread 论文重复 embed/skim。
2e worker heartbeat 不掩盖故障：_write_heartbeat 写 JSON {ts, error}，
job 全部失败时不写心跳让其自然过期。healthcheck 改判时效（20min）
而非仅 test -f。新增 scripts/worker_healthcheck.py。
2f arxiv_client 500 重试复用限流：acquire_api 移入重试循环内，
此前循环外只 acquire 一次，500 回退的二次请求绕过限流器。
500 回退改 continue 回循环顶部重新限流，二次失败由循环统一处理。
测试：新增 TestAnalysisRepository（_get_or_create 幂等+唯一无重复+
IntegrityError 恢复+skim 回写）、PaperRepository upsert 保留 skim
metadata，全套 50 passed。
🔍 OpenCode PR Review Required这是一个受保护的分支，merge 前需要进行 code review。 请运行以下命令进行 OpenCode review： 或者在 PR 页面评论 This is an automated reminder from PR Review Gate. |
5 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
背景
Worker 抓取逻辑审计的 6 个 High 并发/竞态问题。基于 PR1（#46，已合入 main）之上。本 PR 重建（原 #47 因 PR1 合并删 base 分支自动关闭）。
修复清单
2a upsert_paper 保护 skim metadata（High）
packages/storage/repositories/paper.py更新分支此前existing.metadata_json = data.metadata（整体覆盖）会把 skim 写入的keywords/title_zh/abstract_zh抹掉——重复抓取同一论文时丢失已花钱算出来的 skim 产物。改合并：保留已有 skim 派生字段，其余由 arxiv 原始元数据更新覆盖（categories/authors/source 等）。2b AnalysisReport 加 unique + IntegrityError 处理（High）
packages/storage/models.pypaper_id加unique=True，并发 skim 同一论文不再产生重复行。packages/storage/repositories/analysis.py_get_or_create加IntegrityError捕获：flush 命中 unique 约束时回滚并取已存在行。alembic 迁移f6a7b8c9d0e1先删重复行（保留 created_at 最早）再加唯一索引（幂等，PG 窗口函数 / SQLite rowid）。2c daily_runner 内层 retry 加退避（High）
packages/ai/daily_runner.py内层for _attempt in range(retry_limit+1)此前无 sleep 无退避，失败后立即重发加速触发 429。加指数退避；429/限流类用更长退避（10*2^n），其余用3*2^n。2d idle_processor 防重复处理（High）
packages/ai/idle_processor.py新增进程内_dispatching标志 +set_dispatching/is_dispatching。apps/worker/main.pytopic_dispatch_job期间置 True，IdleDetector.is_idle读到即视为繁忙，避免 idle 与 topic_dispatch 抢同一批 unread 论文重复 embed/skim。2e worker heartbeat 不掩盖故障（High）
apps/worker/main.py_write_heartbeat改写 JSON{ts, error}，job 全部失败时不写心跳让其自然过期。docker-compose.ymlhealthcheck 改判时效（20min 过期）而非仅test -f。新增scripts/worker_healthcheck.py。2f arxiv_client 500 重试复用限流（High）
packages/integrations/arxiv_client.pyacquire_api移入重试循环内（此前循环外只 acquire 一次，500 回退的二次请求绕过限流器）。500 回退改continue回循环顶部重新限流。验证
e5f6a7b8c9d0→ 本 PR 的f6a7b8c9d0e1）部署影响
alembic upgrade head（幂等，先删重复行再加唯一索引）python -m scripts.worker_healthcheck）