fix(notification): NotificationBatcher dedup 吞 TRIGGERED + shutdown silent-loss#147
Open
csp0924 wants to merge 2 commits into
Open
fix(notification): NotificationBatcher dedup 吞 TRIGGERED + shutdown silent-loss#147csp0924 wants to merge 2 commits into
csp0924 wants to merge 2 commits into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR fixes two independent bugs in NotificationBatcher that could (1) drop TRIGGERED notifications during the dedup window and (2) hide shutdown-time delivery failures by swallowing per-channel exceptions without any aggregate signal.
Changes:
- Update dedup key to include
(alarm_key, event)so TRIGGERED/RESOLVED don’t overwrite each other in the same flush window. - Make
_send_to_channels()report failures back toflush(), emit an aggregate WARNING for flush-time failures, and emit an aggregate ERROR during_final_flush()on shutdown. - Add tests covering the dedup bug and the shutdown/visibility behavior, plus new observability properties (
last_flush_failure_count,last_flush_failures).
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
csp_lib/notification/batcher.py |
Fixes dedup key semantics and adds aggregate failure reporting/logging + exposes last-flush failure metrics. |
tests/notification/test_batcher_silent_loss.py |
Adds regression tests for dedup preserving TRIGGERED/RESOLVED and for shutdown-time aggregate failure visibility. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+190
to
+210
| aggregated_failures: list[tuple[str, str]] = [] | ||
| self._last_flush_failures = aggregated_failures | ||
|
|
||
| if not items: | ||
| return | ||
|
|
||
| # 去重 | ||
| if self._config.deduplicate_by_key: | ||
| items = self._deduplicate(items) | ||
|
|
||
| # 分組 | ||
| groups = self._group(items) | ||
|
|
||
| # 逐組發送 | ||
| for group_items in groups.values(): | ||
| await self._send_to_channels(group_items) | ||
| aggregated_failures.extend(await self._send_to_channels(group_items)) | ||
|
|
||
| if aggregated_failures: | ||
| failure_summary = ", ".join(f"{name}: {err}" for name, err in aggregated_failures) | ||
| logger.warning( | ||
| "NotificationBatcher: flush 期間 {failure_count} 個 channel 發送失敗 ({failure_summary})", | ||
| failure_count=len(aggregated_failures), | ||
| failure_summary=failure_summary, | ||
| ) |
Comment on lines
+118
to
+130
| pending = len(self._queue) | ||
| if pending == 0: | ||
| return | ||
| try: | ||
| await self.flush() | ||
| except Exception: | ||
| logger.opt(exception=True).warning(f"NotificationBatcher: 停止時 flush 失敗，{pending} 則通知待重試") | ||
| await asyncio.sleep(1) | ||
| remaining = len(self._queue) | ||
| if remaining == 0: | ||
| return | ||
| try: | ||
| await self.flush() | ||
| except Exception: | ||
| dropped = len(self._queue) | ||
| self._queue.clear() | ||
| logger.error(f"NotificationBatcher: 重試仍失敗，{dropped} notifications dropped") | ||
| await self.flush() | ||
| if self._last_flush_failures: | ||
| failure_summary = ", ".join(f"{name}: {err}" for name, err in self._last_flush_failures) | ||
| logger.error( | ||
| "NotificationBatcher: 停止時 flush 完成但 {failure_count} 個 channel 失敗，" | ||
| "{pending} 則 in-flight 通知未送達（failures={failure_summary}）", | ||
| failure_count=len(self._last_flush_failures), | ||
| pending=pending, | ||
| failure_summary=failure_summary, | ||
| ) |
Comment on lines
102
to
108
| await self._flush_task | ||
| except asyncio.CancelledError: | ||
| pass | ||
| self._flush_task = None | ||
| # 確保所有殘留通知都已發送（含重試） | ||
| await self._final_flush() | ||
| logger.info("NotificationBatcher: 已停止") |
Comment on lines
+167
to
+170
| error_msgs = [m for lvl, m in records if lvl in ("ERROR", "WARNING")] | ||
| aggregate = [m for m in error_msgs if "channel" in m.lower() and "fail" in m.lower()] | ||
| assert aggregate, ( | ||
| f"Expected aggregate ERROR/WARNING log about channel failures during shutdown, got records: {records}" |
csp0924
added a commit
that referenced
this pull request
May 13, 2026
PR #147 review fixes: (a) flush() 改成按 channel name 聚合失敗（dict 去重）， 避免同一 channel 在多 group 失敗時 over-count； 新增 last_flush_failure_attempts 屬性記錄 attempt 次數， aggregate WARNING 同時呈現「獨特失敗 channel」和「總失敗次數」。 (b) _final_flush() 在 _lock 內取 pending 計數， 避免與 shutdown 期間殘留 dispatch() race； log wording 從「in-flight 通知未送達」改為 「通知未送達到失敗 channel」，反映 partial-fail 語義。 (c) _on_stop 殘留 comment 已不再 retry，更新為 「對殘留通知做一次最終 flush；channel 失敗會走 aggregate ERROR log」。 (d) test_final_flush_logs_aggregate_error_when_channels_fail 改用穩定信號： - 先 assert batcher.last_flush_failure_count >= 1（不依賴 log 文字） - log 比對改 match "NotificationBatcher" + "停止時"， 避開依賴 exception 文字（"simulated ... failure"）。
…RIGGERED
兩個獨立 bug 在同一檔，bundle 一個 PR 修。
H3 dedup 吞 TRIGGERED:
- _dedup_key 原本只用 alarm_key 作 key；_deduplicate keep-latest 邏輯下，
同 alarm_key 但不同 event (TRIGGERED + RESOLVED) 在同一 flush window 內
會互蓋。Alarm chatter 場景 (5s flush window 內 TRIGGERED→RESOLVED)
operator 從未收到 TRIGGERED，只看到 RESOLVED。
- Fix: dedup key 改為 f"{alarm_key}:{event.value}"，保留 same-event keep-latest 行為。
- Sandbox: 200 對 TRIGGERED/RESOLVED → 修前 0/200 TRIGGERED、修後 200/200。
H2 _final_flush retry path 不可達:
- _send_to_channels 對每個 channel 包 per-channel try/except、swallow exception，
所以 flush() 對 channel 失敗永遠不 raise → _final_flush 的 try/except + sleep +
retry + "重試仍失敗 N notifications dropped" 整段是 dead code。
Shutdown 時 channel 全掛 (如 MongoDB 連線收掉) → flush 看似成功、in-flight 整批
silently discard、無 aggregate ERROR log。
- Fix (方案 B aggregate-in-batcher，保 dispatch happy path 不 raise):
* _send_to_channels 改回傳 list[(channel_name, error_repr)] 失敗清單
* flush() 累計 per-channel 失敗、若 >0 emit aggregate WARNING
* _final_flush 移除 dead retry/sleep code，改成「flush 後檢查失敗清單，>0 emit ERROR」
* 新增 last_flush_failure_count / last_flush_failures property 提供可觀測 metric
- Sandbox: 500 events all-failing channel shutdown → 修前無 aggregate ERROR、
修後同時有 flush WARNING + final_flush ERROR (含 failure summary)。
驗證:
- 新增 tests/notification/test_batcher_silent_loss.py (4 case，未修前 3 FAIL/1 PASS)
- tests/notification/ 全 64 PASS
- 完整回歸 4855 PASS / 7 skipped
- ruff + format + mypy 全綠
PR #147 review fixes: (a) flush() 改成按 channel name 聚合失敗（dict 去重）， 避免同一 channel 在多 group 失敗時 over-count； 新增 last_flush_failure_attempts 屬性記錄 attempt 次數， aggregate WARNING 同時呈現「獨特失敗 channel」和「總失敗次數」。 (b) _final_flush() 在 _lock 內取 pending 計數， 避免與 shutdown 期間殘留 dispatch() race； log wording 從「in-flight 通知未送達」改為 「通知未送達到失敗 channel」，反映 partial-fail 語義。 (c) _on_stop 殘留 comment 已不再 retry，更新為 「對殘留通知做一次最終 flush；channel 失敗會走 aggregate ERROR log」。 (d) test_final_flush_logs_aggregate_error_when_channels_fail 改用穩定信號： - 先 assert batcher.last_flush_failure_count >= 1（不依賴 log 文字） - log 比對改 match "NotificationBatcher" + "停止時"， 避開依賴 exception 文字（"simulated ... failure"）。
csp0924
force-pushed
the
fix/notification-batcher-silent-loss
branch
from
00e1963 to
a440fc6
Compare
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
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.
Bug
兩個獨立 bug，都在
csp_lib/notification/batcher.py，bundle 一起修。H2 —
_final_flushretry 整段死碼_send_to_channelsper-channel try/except 自己 swallow 所有 exception →flush()在 channel 失敗時不會 raise →_final_flush的try/except + sleep + retry + log "重試仍失敗 N notifications dropped"結構不可達。Shutdown 時 channel 全掛 = in-flight notification silent discard，無 aggregate ERROR log。Sandbox：500 events、所有 channel raise → 0 events received, 1 channel attempt, 無 retry-warning log。
H3 — dedup 吞掉 TRIGGERED
_dedup_key只用item.alarm_key，沒含item.event(NotificationEvent: TRIGGERED / RESOLVED / ACK / ...)。_deduplicate「keep latest」邏輯下，同 alarm_key 不同 event 在 flush window (default 5s) 內 → 後到覆蓋先到。Operator 看到「alarm resolved」但從未收到 alarm 觸發。短促 alarm（< flush_interval）trigger 訊號完全消失。Sandbox：200 個 TRIGGERED+RESOLVED pair → 0/200 TRIGGERED, 200/200 RESOLVED delivered。
Fix
H3（one-line）
_dedup_key:return f"{item.alarm_key}:{item.event.value}"同 alarm_key + 同 event 仍 dedup（保護原意圖避免轟炸）；不同 event 不再互殺。
H2（方案 B：aggregate-in-batcher，保守）
_send_to_channels維持 per-channel swallow（保dispatch()happy path 不 raise）但改回傳list[tuple[channel_name, error_repr]]失敗清單flush()累計失敗、emit aggregate WARNING log_final_flush刪掉 dead retry/sleep，改成「flush 後若失敗清單非空 → ERROR log（含 pending 數 + failure summary）」last_flush_failure_count/last_flush_failures提供 caller 可觀測 metricSandbox 對比
Tests
tests/notification/test_batcher_silent_loss.py（新檔，4 case）：test_dedup_preserves_triggered_and_resolved（H3 核心）test_dedup_still_collapses_repeated_same_event（H3 保護 case）test_final_flush_logs_aggregate_error_when_channels_fail（H2 loguru capture）test_send_to_channels_reports_failure_count（H2 metric property）Pre-fix 3 fail / 1 pass → Post-fix 4/4 pass。
Public API
Quality gates