fix: interrupt monitoring buffer wait on shutdown - #3457
Conversation
Monitoring shutdown could block for a fraction of the configured flush interval because the worker used uninterruptible sleeps. Use an event-backed wait so stop wakes the worker immediately, with a focused regression test.
There was a problem hiding this comment.
Pull request overview
This PR updates the SDK monitoring buffer’s background flush worker so shutdown can interrupt the worker’s wait immediately (instead of waiting for a sleep slice), and adds a regression test to ensure stop() completes quickly even with a long flush_interval.
Changes:
- Replace the flush loop’s sliced
time.sleep()waiting with an interruptiblethreading.Event.wait(timeout=...). - Ensure the stop signal wakes the flush thread immediately on shutdown.
- Add a regression test validating
stop()returns quickly and the flush thread terminates.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| sdk/nexent/monitor/monitoring.py | Adds a stop event and uses it to make the flush loop’s wait interruptible during shutdown. |
| test/sdk/monitor/test_monitoring_buffer.py | Adds a regression test asserting stop() is not delayed by flush_interval-based waits. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| except Exception as e: | ||
| logger.error(f"Error in monitoring flush loop: {e}") | ||
|
|
||
| for _ in range(10): | ||
| if not self._running: | ||
| return | ||
| time.sleep(self._flush_interval / 10) | ||
| if self._stop_event.wait(timeout=self._flush_interval): | ||
| return |
There was a problem hiding this comment.
All three review points were fair - pushed f1481ba.
Retry latency: right, the except fell straight through to a full flush_interval wait, so a transient DB error delayed the next attempt by up to 30s by default. The loop now tracks a retry_delay that starts at _error_retry_delay (1s, or the flush interval if that is shorter) and doubles up to the interval, resetting to 0 on a clean pass. The wait is still _stop_event.wait(...), so shutdown stays immediate on the backoff path too.
Test import: dropped the spec_from_file_location loading. The suite's conftest.py already registers sdk.nexent.monitor.monitoring in sys.modules before collection, so a plain from sdk.nexent.monitor.monitoring import MonitoringRecordBuffer gets the same module the rest of test_monitoring.py uses, with no second copy of the module-level singletons.
Timing bound: raised 0.5s to 2.0s. The legacy code slept in ten flush_interval / 10 chunks, so with the 20s interval the pre-fix worst case is ~2s - the bound still fails the old behaviour while leaving headroom on a loaded runner.
Added test_flush_error_does_not_delay_retry_by_a_full_interval, which stubs _flush_to_db to raise and asserts the wait sequence is [1.0, 2.0, 4.0] rather than three 20s waits. RED-verified: with monitoring.py reverted it fails at that assertion (the loop waits the full interval each time). test/sdk/monitor/ is 94 passed on the fixed tree; the one failure there, test_agent_context_survives_delayed_async_stream_iteration, fails identically on the stashed tree (missing async plugin locally) and is unrelated.
| import importlib.util | ||
| from pathlib import Path | ||
| import time | ||
|
|
||
|
|
||
| MODULE_PATH = Path(__file__).parents[3] / "sdk" / "nexent" / "monitor" / "monitoring.py" | ||
|
|
||
|
|
||
| def _load_monitoring_module(): | ||
| spec = importlib.util.spec_from_file_location("monitoring_under_test", MODULE_PATH) | ||
| assert spec is not None | ||
| assert spec.loader is not None | ||
| module = importlib.util.module_from_spec(spec) | ||
| spec.loader.exec_module(module) | ||
| return module |
| started = time.monotonic() | ||
| buffer.stop() | ||
|
|
||
| assert time.monotonic() - started < 0.5 |
Addresses review feedback: an exception in the flush loop still fell through to a full flush_interval wait, so a transient failure delayed the next retry by up to the whole interval. Retries now start from a short interruptible delay and double up to the flush interval, resetting on success. Also import the monitoring module normally in the buffer test instead of loading a second copy via spec_from_file_location, and relax the shutdown timing bound to 2s, which still fails the pre-fix worst case.
Monitoring shutdown could block for up to one-tenth of the configured flush interval because the worker slept in uninterruptible slices. The worker now waits on a stop event, so shutdown wakes it immediately while retaining periodic flushing.
The regression test takes about 2 seconds and fails on the original code, then passes in under 0.5 seconds with this change. Focused pytest, Ruff, compilation, and both diff gates pass; the source file's three existing import-order findings are unchanged from
develop.Closes #3403