fix(pipeline): notify without write lock + real success counts [修复组①] - #20
Conversation
…,C12,C18,C19,C23]
Restructure notifier_dispatch.dispatch_notifications into three phases so
notification sends never hold the SQLite write lock, and callers get real
outcome counts instead of unconditional success:
- C1/C23: dispatch_notifications now runs Phase A (caller session: query
rules, create pending NotificationLog rows, flush+commit), Phase B (no
session open: sequential sends, one resolved notifier instance per rule),
Phase C (fresh short session: bulk-persist outcomes, single commit).
Returns {"sent": n, "failed": m}.
- C12: pipeline.py step 5 consumes that aggregate, assigns
PipelineResult.notifications_sent (previously dead), and logs/emits real
sent/failed numbers - warning level when every attempted send failed.
- C3: ai_processor.process_with_ai now returns the number of records
actually enriched (0 on empty input or unregistered processor_type, with
a warning log) instead of None. pipeline.py step 4 uses that real count
for ai_processed and only emits "AI 处理完成" when something was actually
enriched; a config error now emits a warning-level "失败/跳过" event.
- C18: unknown notifier_type on a rule now logs a warning with the rule id
and notifier_type instead of skipping silently.
- C19: runner.py's terminal-failure handler now prefixes the persisted
error_message with the same retry-classification pipeline.py already
computes (effective_error_type), so ops can distinguish retryable from
permanent failures from the stored message alone.
Tests extended in tests/unit/pipeline/{test_notifier_dispatch,
test_notifier_dispatch_errors,test_ai_processor,test_pipeline}.py and
tests/unit/test_runner.py: phase-commit-before-send ordering, partial/total
send failure aggregates, unknown notifier_type/processor_type warnings, and
the exact error_message prefix format.
|
✅ Health: 7.1 📋 At a glance Files & modules (2)
🚨 Change risk: 9.3/10 (high)
🔎 More signals (1)🔥 Hotspots touched (4)
1 more
👀 Suggested reviewers @xujinghua 📊 Full report · ⭐ Star Repowise · 📥 Install bot · Last updated 2026-07-18 13:35 UTC |
📝 WalkthroughSummary by CodeRabbit
WalkthroughAI processing and notification dispatch now return structured counts that the pipeline records in events and results. Notification persistence is separated from network sends, and runner failures now persist taxonomy-prefixed error messages. ChangesPipeline outcome reporting
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Pipeline
participant AIProcessor
participant NotifierDispatch
participant NotificationLog
Pipeline->>AIProcessor: process records
AIProcessor-->>Pipeline: enriched count
Pipeline->>NotifierDispatch: dispatch notifications
NotifierDispatch->>NotificationLog: commit pending rows
NotifierDispatch->>NotifierDispatch: send and aggregate outcomes
NotifierDispatch->>NotificationLog: persist final statuses
NotifierDispatch-->>Pipeline: sent and failed counts
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
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. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request restructures the notification dispatching pipeline into three distinct phases (Phase A: queue pending logs, Phase B: execute network sends without an open session, Phase C: persist outcomes in a short-lived session) to prevent slow network requests from holding the SQLite write lock. It also updates the AI processor to return the count of enriched records and logs warnings for misconfigured processors. The pipeline runner is enhanced to include taxonomy classifications in persisted error messages. Review feedback suggests handling potential null values in rule.notifier_config to avoid AttributeError and batching database queries in Phase C of the notification dispatch to resolve an N+1 query issue.
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.
| ack_required = bool(_ack_secret(rule.notifier_config)) | ||
| for record in records: | ||
| log = NotificationLog( | ||
| log_id = str(uuid.uuid4()) | ||
| session.add(NotificationLog( | ||
| id=log_id, | ||
| rule_id=rule.id, | ||
| record_id=record.id, | ||
| status="pending", | ||
| ack_status="not_required", | ||
| ) | ||
| session.add(log) | ||
| await session.flush() | ||
| )) | ||
| send_plan.append(_PendingSend( | ||
| log_id=log_id, | ||
| notifier=notifier, | ||
| notifier_config=rule.notifier_config, |
There was a problem hiding this comment.
如果 rule.notifier_config 在数据库中为 None（例如未配置的 JSON 列），直接调用 _ack_secret(rule.notifier_config) 会导致 AttributeError: 'NoneType' object has no attribute 'get' 异常。建议在循环前对其进行空值处理，使用 rule.notifier_config or {}，以提高代码的健壮性。
notifier_config = rule.notifier_config or {}
ack_required = bool(_ack_secret(notifier_config))
for record in records:
log_id = str(uuid.uuid4())
session.add(NotificationLog(
id=log_id,
rule_id=rule.id,
record_id=record.id,
status="pending",
ack_status="not_required",
))
send_plan.append(_PendingSend(
log_id=log_id,
notifier=notifier,
notifier_config=notifier_config,| for outcome in outcomes: | ||
| log = await write_session.get(NotificationLog, outcome.log_id) | ||
| if log is None: | ||
| continue | ||
| log.status = outcome.status | ||
| log.response_data = outcome.response_data | ||
| log.error_message = outcome.error_message | ||
| log.ack_status = outcome.ack_status |
There was a problem hiding this comment.
在 Phase C 中，在循环内对每个 outcome 依次调用 write_session.get(NotificationLog, outcome.log_id) 会导致 IN 子句进行批量查询，将数据库交互次数减少到
| for outcome in outcomes: | |
| log = await write_session.get(NotificationLog, outcome.log_id) | |
| if log is None: | |
| continue | |
| log.status = outcome.status | |
| log.response_data = outcome.response_data | |
| log.error_message = outcome.error_message | |
| log.ack_status = outcome.ack_status | |
| log_ids = [outcome.log_id for outcome in outcomes] | |
| result = await write_session.execute( | |
| select(NotificationLog).where(NotificationLog.id.in_(log_ids)) | |
| ) | |
| logs = {log.id: log for log in result.scalars().all()} | |
| for outcome in outcomes: | |
| log = logs.get(outcome.log_id) | |
| if log is None: | |
| continue | |
| log.status = outcome.status | |
| log.response_data = outcome.response_data | |
| log.error_message = outcome.error_message | |
| log.ack_status = outcome.ack_status |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@backend/pipeline/ai_processor.py`:
- Line 142: Update the zip call in the record/enrichment loop to pass
strict=False explicitly, preserving the current truncating behavior while
satisfying Ruff B905.
In `@backend/pipeline/notifier_dispatch.py`:
- Around line 164-178: Update the Phase C persistence flow in run_pipeline so a
commit failure does not discard the accumulated sent and failed delivery counts.
Catch the persistence error, retain and return those counts alongside an
explicit persistence error/status, and ensure callers do not interpret the
result as a generic zero-send notification failure or safe retry.
- Around line 152-159: Close the phase-A database session scope before running
notification I/O: restructure the flow around session flush/commit so the
session context exits before the _send_one loop executes. Update
tests/unit/pipeline/test_notifier_dispatch_errors.py lines 157-217 to assert the
session __aexit__ event occurs before notifier.send(), replacing the
commit-order-only assertion.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3ee723d5-6fcd-401c-a977-2938b87a8150
📒 Files selected for processing (9)
backend/pipeline/ai_processor.pybackend/pipeline/notifier_dispatch.pybackend/pipeline/pipeline.pybackend/pipeline/runner.pytests/unit/pipeline/test_ai_processor.pytests/unit/pipeline/test_notifier_dispatch.pytests/unit/pipeline/test_notifier_dispatch_errors.pytests/unit/pipeline/test_pipeline.pytests/unit/test_runner.py
| ) | ||
|
|
||
| enriched = 0 | ||
| for record, enrichment in zip(records, result.enrichments): |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target file around the flagged lines.
sed -n '1,240p' backend/pipeline/ai_processor.py
printf '\n---\n'
# Find related uses of records/result.enrichments to infer shape/length assumptions.
rg -n "result\.enrichments|zip\(records" backend/pipeline -S
printf '\n---\n'
# Look for tests or docs that mention enrichment batching or truncation behavior.
rg -n "enrichment|enrichments|truncat|zip\\(" tests backend -SRepository: 2233admin/opencli-admin
Length of output: 23346
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- backend/processors/base.py ---\n'
sed -n '1,120p' backend/processors/base.py
printf '\n--- tests/unit/pipeline/test_ai_processor.py ---\n'
sed -n '1,220p' tests/unit/pipeline/test_ai_processor.py
printf '\n--- related processor length handling ---\n'
rg -n "len\\(records\\)|len\\(.*enrichments\\)|zip\\(records, result\\.enrichments|strict=False" backend/processors tests/unit/pipeline -SRepository: 2233admin/opencli-admin
Length of output: 252
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- backend/processors/base.py ---'
sed -n '1,120p' backend/processors/base.py
echo
echo '--- tests/unit/pipeline/test_ai_processor.py ---'
sed -n '1,260p' tests/unit/pipeline/test_ai_processor.py
echo
echo '--- related processor length handling ---'
rg -n "len\\(records\\)|len\\(.*enrichments\\)|zip\\(records, result\\.enrichments|strict=False" backend/processors tests/unit/pipeline -SRepository: 2233admin/opencli-admin
Length of output: 11622
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- backend/processors/claude_processor.py ---'
sed -n '1,180p' backend/processors/claude_processor.py
echo
echo '--- backend/processors/openai_processor.py ---'
sed -n '1,180p' backend/processors/openai_processor.py
echo
echo '--- backend/processors/local_processor.py ---'
sed -n '1,180p' backend/processors/local_processor.py
echo
echo '--- backend/processors/external_http_processor.py ---'
sed -n '1,220p' backend/processors/external_http_processor.pyRepository: 2233admin/opencli-admin
Length of output: 17871
Make the zip truncation explicit. zip() still silently drops extra items here; add strict=False so the current 1:1 pairing contract is explicit and Ruff B905 is satisfied.
🧰 Tools
🪛 Ruff (0.15.21)
[warning] 142-142: zip() without an explicit strict= parameter
Add explicit value for parameter strict=
(B905)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/pipeline/ai_processor.py` at line 142, Update the zip call in the
record/enrichment loop to pass strict=False explicitly, preserving the current
truncating behavior while satisfying Ruff B905.
Source: Linters/SAST tools
| if not send_plan: | ||
| return {"sent": 0, "failed": 0} | ||
|
|
||
| await session.flush() | ||
| await session.commit() | ||
|
|
||
| # ── Phase B: sends, sequential, no session open ────────────────────────── | ||
| outcomes = [await _send_one(task) for task in send_plan] |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Ensure the database session is closed before notification I/O.
Committing releases the transaction, but the caller’s session context remains entered during every send.
backend/pipeline/notifier_dispatch.py#L152-L159: exit the phase-A session scope before executing_send_one.tests/unit/pipeline/test_notifier_dispatch_errors.py#L157-L217: assert session__aexit__occurs beforenotifier.send(), rather than only checking commit order.
📍 Affects 2 files
backend/pipeline/notifier_dispatch.py#L152-L159(this comment)tests/unit/pipeline/test_notifier_dispatch_errors.py#L157-L217
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/pipeline/notifier_dispatch.py` around lines 152 - 159, Close the
phase-A database session scope before running notification I/O: restructure the
flow around session flush/commit so the session context exits before the
_send_one loop executes. Update
tests/unit/pipeline/test_notifier_dispatch_errors.py lines 157-217 to assert the
session __aexit__ event occurs before notifier.send(), replacing the
commit-order-only assertion.
| # ── Phase C: persist outcomes in a fresh short-lived session ───────────── | ||
| from backend.database import AsyncSessionLocal | ||
|
|
||
| async with AsyncSessionLocal() as write_session: | ||
| for outcome in outcomes: | ||
| log = await write_session.get(NotificationLog, outcome.log_id) | ||
| if log is None: | ||
| continue | ||
| log.status = outcome.status | ||
| log.response_data = outcome.response_data | ||
| log.error_message = outcome.error_message | ||
| log.ack_status = outcome.ack_status | ||
| await write_session.commit() | ||
|
|
||
| return {"sent": sent, "failed": failed} |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Preserve delivery counts when outcome persistence fails.
If the phase-C commit raises after successful external sends, the aggregate is discarded and run_pipeline reports a generic notification failure with notifications_sent == 0. Return the delivery counts alongside a persistence error/status so already-completed sends are reported accurately and are not mistaken for safe retries.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@backend/pipeline/notifier_dispatch.py` around lines 164 - 178, Update the
Phase C persistence flow in run_pipeline so a commit failure does not discard
the accumulated sent and failed delivery counts. Catch the persistence error,
retain and return those counts alongside an explicit persistence error/status,
and ensure callers do not interpret the result as a generic zero-send
notification failure or safe retry.
1 participant
修复组① — 锁与假信号 (账本 C1/C3/C12/C18/C19/C23)
采集链路专项审计后的第一组修复。Sonnet 实施, Fable 审计通过 (三段式结构 + 作用域/签名疑点逐一排除)。
notifier_dispatch三段式重构: Phase A 短会话建 pending 日志行并 commit; Phase B 零 session 顺序发送 (每 rule 复用一个 notifier 实例, 单条失败不炸批); Phase C 新短会话批量回写结果单次 commit。SQLite 写锁不再横跨网络 I/Odispatch_notifications返回{sent, failed}聚合; step5 报真实数字, 全败升 warning 级;PipelineResult.notifications_sent从死字段变真值process_with_ai返回真实富化条数; 未注册 processor_type 落 warning 日志; step4 由真实计数驱动, 0 条时事件改报 "AI 处理失败/跳过" (warning) 而非全绿 "完成"[{error_type}] {message}(复用effective_error_type, 零 schema 改动)Test
边界说明
Phase C 崩溃窗口: 发送已发生但结果行留在 pending — 可观测性仍优于旧行为 (旧: 整批日志行直接丢失)。