Skip to content

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

Closed
Color2333 wants to merge 1 commit into
worker/pr1-criticalfrom
worker/pr2-concurrency
Closed

fix(worker): PR2 并发/竞态 High 修复 6 项(metadata保护+AnalysisReport唯一+retry退避+重复skim+heartbeat+arxiv500)#47
Color2333 wants to merge 1 commit into
worker/pr1-criticalfrom
worker/pr2-concurrency

Conversation

@Color2333

Copy link
Copy Markdown
Owner

背景

Worker 抓取逻辑审计的 6 个 High 并发/竞态问题。基于 PR1(#46)之上,文件集与 PR1/PR3 互斥。PR1 合入后可改 base 到 main 合入。

修复清单

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.py paper_idunique=True,并发 skim 同一论文不再产生重复行。packages/storage/repositories/analysis.py _get_or_createIntegrityError 捕获: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_dispatchingapps/worker/main.py topic_dispatch_job 期间置 True,IdleDetector.is_idle 读到即视为繁忙,避免 idle 与 topic_dispatch 抢同一批 unread 论文重复 embed/skim。仅 worker 进程内生效(idle_processor 与 topic_dispatch 同在 worker 容器)。

2e worker heartbeat 不掩盖故障(High)

apps/worker/main.py _write_heartbeat 改写 JSON {ts, error},job 全部失败时不写心跳让其自然过期。docker-compose.yml healthcheck 改判时效(20min 过期)而非仅 test -f。新增 scripts/worker_healthcheck.py(读 JSON 判 ts 时效,文件缺失/损坏/过期均判不健康)。

2f arxiv_client 500 重试复用限流(High)

packages/integrations/arxiv_client.py acquire_api 移入重试循环内(此前循环外只 acquire 一次,500 回退的二次请求绕过限流器)。500 回退改 continue 回循环顶部重新限流,二次失败由循环统一处理。

验证

  • AST 解析全部 9 个改动文件
  • ruff check / format 全通过
  • pytest:55 passed, 2 skipped(新增 5 测试:TestAnalysisRepository 4 + PaperRepository upsert 保留 skim metadata 1)
  • alembic upgrade head 迁移链完整跑到新 head f6a7b8c9d0e1(基于 PR1 的 e5f6a7b8c9d0
  • worker_healthcheck.py 三态验证(缺失→exit1,新鲜→exit0,过期→exit1)

部署影响

  • 需跑 alembic upgrade head(新迁移幂等,先删重复行再加唯一索引)
  • docker-compose healthcheck 命令变更(用 python -m scripts.worker_healthcheck
  • 无新增配置项

不做(诚实标注)

  • embed/skim/deep 各自 session 无统一事务:架构改动大,留后续
  • 跨进程 _active_tasks/_current_slot 共享:纯限流只共享 bucket 即可

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。
@github-actions

Copy link
Copy Markdown

🔍 OpenCode PR Review Required

这是一个受保护的分支,merge 前需要进行 code review。

请运行以下命令进行 OpenCode review:

/oc review https://github.com/Color2333/PaperMind/pull/$PR_NUM

或者在 PR 页面评论 /oc 来触发 OpenCode review。


This is an automated reminder from PR Review Gate.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant