fix(pipeline): resilient cursor save + SQLite cursor lock + gateway retry + scheduled autoretry [修复组④]

[2233admin]

修复组④ — 游标与重试 (账本 C10/C11/C13/C14)

Sonnet 实施, Fable 审计通过 (代码 diff + migration 单 head 确认 + 主 repo 真跑测试)。

• C11 cursor save 无 try/except → 包错: 保存失败 log ERROR (source+cursor 值) + run 保持 success (记录已落库) + cursor_advanced=False, 下轮补推。不再假失败触发重采
• C10 SQLite 游标丢失更新 (SELECT FOR UPDATE 在 SQLite 是静默 no-op) → 乐观锁 version 列: UPDATE ... WHERE id=? AND version=? 检 rowcount, 0 行=他人已推进则重读重试 (3 次上限)。SQLite/Postgres 同语义, 不依赖后端锁。带 alembic migration t9y0z1a2b3c4 (batch_alter + server_default=0, 现有行填 0)
• C13 504/520/522/524 误判永久 → 加进 RETRY_STATUS ({429,500,502,503,504,520,522,524}); api/rss channel 网关状态映射 retryable (照 web_scraper 先例)
• C14 重试无 backoff + scheduled 无 autoretry → run_collection 加 retry_backoff+retry_backoff_max=600+jitter+max_retries=3; run_scheduled_collection 补 autoretry policy

审计确认

• alembic heads = 单一 head t9y0z1a2b3c4 (migration 正确挂链尾, down_revision=s8x9y0z1a2b3; 之前担心的 z6..add_plans 只是文件名字母序靠后非 chain head)
• migration upgrade/downgrade 用 batch_alter (SQLite 友好), applied+reverted 双向验证过

Deviation (审计已收, 均合理)

1. env: worktree 默认 pin cp314 (lxml 无 wheel), agent uv sync --python 3.13 重建 (.venv gitignored, 无害) — 也暴露了 worktree env 坑, 后续统一在主 repo 验测
2. C13 只修 fetch() 路径 (base.py 注释确认 runner 只调 fetch, collect() 是 prod-dead 的重复处理器), 不动 collect()
3. C14 run_scheduled 不绑 _AlertOnRetriesExhaustedTask — 该类 on_failure 把 args[0] 当 CollectionTask id 解析, 但 scheduled 首参是 schedule_id (不同 id 空间), 复用会 silently 查错行。只加 retry config 不绑 alert task。好判断, 避免了一个静默 bug

Test

• 主 repo cp313 env 真跑: 1272 passed, 2 failed, 1 skipped。2 failed = 账本 P3-6 已知存量 flake (nodes_install netbird/ssh GBK), 与本改动无关。1 skip = Postgres-gated cursor 测试 (无 TEST_DATABASE_URL_PG)

合并顺序

④⑤⑥ 共同热点 pipeline.py。本 PR 改 C11 区 (cursor save ~372)。与 ⑤ 的 C21 区 (~404) 不同区域, 但同文件, 后合者 rebase。

[coderabbitai]

Warning

Review limit reached

@2233admin, you've reached your PR review limit, so we couldn't start this review.

Next review available in: 57 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 4a8522f9-8f8d-4fb5-bc2e-e889625f08c6

📥 Commits

Reviewing files that changed from the base of the PR and between cb6c1ab and 158c38f.

📒 Files selected for processing (16)

• backend/channels/api_channel.py
• backend/channels/rss_channel.py
• backend/migrations/versions/t9y0z1a2b3c4_add_source_cursor_version.py
• backend/models/source_cursor.py
• backend/pipeline/cursor_store.py
• backend/pipeline/error_taxonomy.py
• backend/pipeline/http_client.py
• backend/pipeline/pipeline.py
• backend/worker/tasks.py
• tests/unit/channels/test_api_channel.py
• tests/unit/channels/test_rss_fetch.py
• tests/unit/pipeline/test_db_cursor_store.py
• tests/unit/pipeline/test_error_taxonomy.py
• tests/unit/pipeline/test_http_client.py
• tests/unit/pipeline/test_pipeline_cursor.py
• tests/unit/worker/test_tasks.py

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[repowise-bot]

✅ Health: 7.7

📋 At a glance
4 hotspots touched · 9 new findings introduced · 1 dead-code finding.

Files & modules (2)

• backend (3 files)
  • backend/worker/tasks.py
  • backend/pipeline/pipeline.py
  • backend/channels/api_channel.py
• tests (1 file)
  • .../channels/test_api_channel.py

🚨 Change risk: 9.3/10 (high)
This change's risk is driven by:

• more lines added than baseline
• more scattered than baseline

🔎 More signals (2)

🔥 Hotspots touched (4)

• backend/worker/tasks.py — 3 commits/90d, 1 dependents · primary owner: xujinghua (100%)
• backend/pipeline/pipeline.py — 10 commits/90d, 6 dependents · primary owner: xujinghua (100%)
• .../channels/test_api_channel.py — 2 commits/90d, 0 dependents · primary owner: xujinghua (100%)

1 more

• backend/channels/api_channel.py — 1 commits/90d, 1 dependents · primary owner: xujinghua (100%)

💀 Dead code (1 finding)

• 💀 backend/worker/tasks.py send_notification (confidence 1.00)

👀 Suggested reviewers @xujinghua

---

📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-18 18:17 UTC
_{Silence on a single PR with [skip repowise] in the title · Per-repo toggle on repowise.dev/settings?tab=bot}

[gemini-code-assist]

Code Review

This pull request introduces several robust improvements to the pipeline's reliability and error handling. Key changes include migrating from SELECT ... FOR UPDATE to optimistic concurrency control using a new version column on source_cursors (ensuring SQLite compatibility), making cursor-saving failures non-fatal to prevent false run failures, and enhancing retry strategies with exponential backoff and jitter for Celery tasks. Additionally, gateway timeouts and Cloudflare-specific error codes are now classified as retryable. The reviewer's feedback correctly points out that 408 (Request Timeout) should be added to RETRY_STATUS in http_client.py to align with its classification as a retryable status in error_taxonomy.py.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

在 backend/pipeline/error_taxonomy.py 中，您已将 408 (Request Timeout) 归类为可重试的 HTTP 状态码（is_retryable_http_status）。然而，在 backend/pipeline/http_client.py 的 RETRY_STATUS 集合中，并没有包含 408。这会导致 RateLimitedClient 在遇到 408 时不会在 HTTP 客户端级别进行轻量级重试，而是直接抛出异常并触发重量级的 Celery 任务级重试。建议将 408 补充到 RETRY_STATUS 中，以保持一致性并提高重试效率。

Suggested change

	RETRY_STATUS = frozenset({429, 500, 502, 503, 504, 520, 522, 524})
	RETRY_STATUS = frozenset({408, 429, 500, 502, 503, 504, 520, 522, 524})