fix(mongo): MongoBatchUploader.stop 防止 in-flight 資料 silent loss#136
Open
csp0924 wants to merge 2 commits into
Open
fix(mongo): MongoBatchUploader.stop 防止 in-flight 資料 silent loss#136csp0924 wants to merge 2 commits into
csp0924 wants to merge 2 commits into
Conversation
當 Mongo 在 stop() 時不可用，原本 _flush_loop 停止分支只執行單次 flush_all()， 失敗時 _handle_write_failure 把 documents restore 回 queue 並印 "重試 N/M" WARNING（誤導 operator，因為 loop 已 break 不會有後續 retry），最終 documents silent 留在記憶體中，process 結束後資料遺失，無任何 ERROR log。 修復：新增 _shutdown_drain() 取代 stop 分支對 flush_all 的單次呼叫： - bounded retry write_batch(documents) 最多 _SHUTDOWN_FLUSH_MAX_ATTEMPTS=3 次， 每次間隔 0.1s 給暫時性網路抖動恢復時間 - 全部失敗時 emit ERROR log，明確列出每個 collection 未落庫筆數 - 不走 _handle_write_failure 的 WARNING + restore 路徑（避免誤導 operator） H3 closed-loop probe scenario：300 筆 + 不可用 Mongo + immediate stop - 修復前：queue 殘留 300 筆，僅有 1 條 "重試 1/2" WARNING - 修復後：3 次 retry 後 emit ERROR "shutdown 後仍有 300 筆資料無法寫入 Mongo（per-collection: metrics=300）"，operator 可從 log 辨認遺失規模 Public API signature 不變（patch-level）；新增的常數為 module-private。
There was a problem hiding this comment.
Pull request overview
This PR hardens MongoBatchUploader.stop() shutdown behavior to avoid silent in-memory data loss when MongoDB is unavailable, by adding a shutdown-specific bounded retry drain path and validating it with new tests.
Changes:
- Add
_shutdown_drain()to drain queues and perform bounded retry writes during shutdown, emitting an ERROR with per-collection lost counts when retries are exhausted. - Update
_flush_loopstop andCancelledErrorbranches to use_shutdown_drain()instead of the normal failure/restore path. - Add shutdown-focused tests covering persistent failure (must ERROR with counts) and transient failure (eventual persistence).
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
csp_lib/mongo/uploader.py |
Adds shutdown drain + bounded retry + improved error logging; routes stop/cancel paths through it. |
tests/mongo/test_uploader_shutdown.py |
Adds regression tests to ensure stop() does not silently lose in-flight docs. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
313
to
317
| @@ -308,8 +316,9 @@ async def _flush_loop(self) -> None: | |||
| await self._wait_for_trigger() | |||
|
|
|||
Comment on lines
331
to
333
| # 結束前 bounded retry flush 所有資料 | ||
| await self._shutdown_drain() | ||
| break |
Comment on lines
+352
to
+356
| pending: dict[str, list[dict[str, Any]]] = {} | ||
| for collection_name in list(self._queues.keys()): | ||
| docs = await self._queues[collection_name].drain() | ||
| if docs: | ||
| pending[collection_name] = docs |
Comment on lines
+96
to
+100
| @pytest.mark.asyncio | ||
| async def test_stop_with_transient_failure_eventually_persists(self) -> None: | ||
| """ | ||
| 暫時性故障 — 前幾次 write_batch 失敗、第 N 次成功。 | ||
| bounded retry 應該讓資料最終落庫，queue 清空且不噴 ERROR。 |
回應 PR #136 Copilot review 四點： 1. `_flush_loop` 改 try/finally，drain 從 break 內 inline 移到 finally： - 若 stop_event 在進 loop 前就 set，原本 `while not stop_event.is_set()` 會直接跳過 drain → silent loss；現在 finally 必跑一次 drain。 - CancelledError 與正常 stop 兩條退出路徑都收斂到同一個 drain 呼叫， 避免實作分歧。 2. Python 3.11+ cancel 語意處理：drain 內含多個 await 點（drain queue + write_batch + sleep），cancel count 累積會在每個 await 重新 raise。 進 finally 前若是 cancelled，先 `current_task().uncancel()` 清掉 pending count 讓 drain 跑完，再於最末重新 raise CancelledError 保留 cancellation 語意。 3. `_shutdown_drain` 在每輪 retry 前、retry 全成功後、retry 配額耗盡前 都再 drain 一次 queue，把 stop 期間其他協程 enqueue 進來的 late docs 併入 pending；確保 lost-count ERROR 反映 shutdown 全程實際 未落庫筆數，不會被 race 進來的 docs silently 漏掉。 4. test_stop_with_transient_failure_eventually_persists 加 LogCapture 斷言：retry-success 路徑不應留下任何 ERROR log，防止未來改動把 transient failure 誤判成 silent loss。 完整回歸：tests/mongo/ 143 passed；tests/ 全體 4853 passed。
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.
問題
Closed-loop probe 發現：
MongoBatchUploader.stop()在 Mongo 不可用時靜默丟資料。Shutdown 路徑單次flush_all失敗後走_handle_write_failure→queue.restore()→ loop break，docs 滯留 in-memory queue 直到 process exit。WARNING log 寫 "重試 1/2" 還會誤導 operator 以為會繼續重試。Sandbox 證據（
sandbox/data_upload_backpressure_demo.pyH3 scenario）：300 docs enqueue + immediate stop with unavailable Mongo → queue size = 300、write_calls = 1、persisted = 0。改動
csp_lib/mongo/uploader.py新增_shutdown_drain()：drain queues → bounded retry（預設 3 次、_SHUTDOWN_FLUSH_RETRY_DELAY=0.1s），不再走_handle_write_failure的誤導 WARNING + restore 路徑_flush_loopstop branch +CancelledErrorbranch 都改呼叫_shutdown_drain()驗證
tests/mongo/test_uploader_shutdown.py（2 案：persistent failure 確認 ERROR log；transient failure 確認 eventual persist）