fix: interrupt monitoring buffer wait on shutdown

sdk/nexent/monitor/monitoring.py

@@ -2496,11 +2496,13 @@ def __init__(self):
             os.getenv("MODEL_MONITORING_BATCH_SIZE", "100"))
         self._flush_interval: int = int(
             os.getenv("MODEL_MONITORING_FLUSH_INTERVAL_SECONDS", "30"))
+        self._error_retry_delay: float = min(1.0, float(self._flush_interval))
         self._consecutive_failures: int = 0
         self._max_failures: int = 3
         self._degraded_until: float = 0.0
         self._last_flush_time: float = time.time()
         self._running: bool = False
+        self._stop_event = threading.Event()
         self._flush_thread: Optional[threading.Thread] = None
         self._lock = threading.Lock()
 
@@ -2511,6 +2513,7 @@ def _start_flush_thread(self) -> None:
         with self._lock:
             if self._running:
                 return
+            self._stop_event.clear()
             self._running = True
             self._flush_thread = threading.Thread(
                 target=self._flush_loop,
@@ -2526,6 +2529,7 @@ def add_record(self, record: dict) -> None:
         self._buffer.append(record)
 
     def _flush_loop(self) -> None:
+        retry_delay = 0.0
         while self._running:
             try:
                 now = time.time()
@@ -2537,13 +2541,20 @@ def _flush_loop(self) -> None:
                 if should_flush:
                     self._flush_to_db()
                     self._last_flush_time = now
+                retry_delay = 0.0
             except Exception as e:
                 logger.error(f"Error in monitoring flush loop: {e}")
+                # Back off from a short delay instead of always waiting a full
+                # flush interval, so a transient error does not stall retries.
+                retry_delay = min(
+                    self._flush_interval,
+                    self._error_retry_delay if retry_delay <= 0.0
+                    else retry_delay * 2,
+                )
 
-            for _ in range(10):
-                if not self._running:
-                    return
-                time.sleep(self._flush_interval / 10)
+            wait_time = retry_delay if retry_delay > 0.0 else self._flush_interval
+            if self._stop_event.wait(timeout=wait_time):
+                return
 
     def _flush_to_db(self) -> None:
         now = time.time()
@@ -2623,6 +2634,7 @@ def _write_batch(self, batch: List[dict]) -> None:
 
     def stop(self) -> None:
         self._running = False
+        self._stop_event.set()
         if self._flush_thread and self._flush_thread.is_alive():
             self._flush_thread.join(timeout=5)
         logger.info("Monitoring buffer flush thread stopped")

test/sdk/monitor/test_monitoring_buffer.py

@@ -0,0 +1,46 @@
+import time
+
+from sdk.nexent.monitor.monitoring import MonitoringRecordBuffer
+
+
+def test_stop_interrupts_flush_interval(monkeypatch):
+    monkeypatch.setenv("ENABLE_MODEL_MONITORING", "true")
+    monkeypatch.setenv("MODEL_MONITORING_FLUSH_INTERVAL_SECONDS", "20")
+    buffer = MonitoringRecordBuffer()
+
+    time.sleep(0.1)
+    started = time.monotonic()
+    buffer.stop()
+
+    # The legacy worst case was ~2s (one tenth of the 20s interval); a 2s bound
+    # still fails the old behaviour while leaving headroom on a loaded runner.
+    assert time.monotonic() - started < 2.0
+    assert not buffer._flush_thread.is_alive()
+
+
+def test_flush_error_does_not_delay_retry_by_a_full_interval(monkeypatch):
+    monkeypatch.setenv("ENABLE_MODEL_MONITORING", "false")
+    monkeypatch.setenv("MODEL_MONITORING_FLUSH_INTERVAL_SECONDS", "20")
+    buffer = MonitoringRecordBuffer()
+
+    waits = []
+
+    def fake_wait(timeout=None):
+        waits.append(timeout)
+        return len(waits) >= 3
+
+    monkeypatch.setattr(buffer._stop_event, "wait", fake_wait)
+
+    def boom():
+        raise RuntimeError("transient")
+
+    monkeypatch.setattr(buffer, "_flush_to_db", boom)
+    buffer._buffer.append({"record": 1})
+    buffer._last_flush_time = 0.0
+    buffer._running = True
+
+    buffer._flush_loop()
+
+    # Retries back off from a short delay instead of always waiting the full
+    # 20s flush interval.
+    assert waits == [1.0, 2.0, 4.0]