Make stop join timeout configurable and add test

Introduce STOP_JOIN_TIMEOUT (5s) and use it for the writer thread join timeout instead of a hardcoded value. Reorder lifecycle checks in VideoRecorder.start to raise if the recorder is marked _abandoned before short-circuiting on is_running, preventing accidental restarts of abandoned recorders. Add a BlockingWriteGear fake, fixture, and a test (test_stop_timeout_marks_abandoned_and_prevents_restart) that forces a stop timeout, verifies the recorder is marked abandoned, prevents restart, and performs cleanup of the blocked writer thread.

Author: C-Achard

dlclivegui/services/video_recorder.py

@@ -23,6 +23,8 @@
 
 logger = logging.getLogger(__name__)
 
+STOP_JOIN_TIMEOUT = 5.0  # seconds
+
 
 @dataclass
 class RecorderStats:
@@ -88,10 +90,10 @@ def start(self) -> None:
             raise RuntimeError("vidgear is required for video recording. Install it with 'pip install vidgear'.")
 
         with self._lifecycle_lock:
-            if self.is_running:
-                return
             if self._abandoned:
                 raise RuntimeError("Cannot restart VideoRecorder, as a leftover thread is still running.")
+            if self.is_running:
+                return
             if self._writer is not None:
                 # Best-effort cleanup of a stale writer to avoid leaking resources.
                 logger.warning(
@@ -222,7 +224,7 @@ def stop(self) -> None:
 
             t = self._writer_thread
             if t is not None:
-                t.join(timeout=5.0)
+                t.join(timeout=STOP_JOIN_TIMEOUT)
                 if t.is_alive():
                     with self._stats_lock:
                         self._encode_error = RuntimeError(

tests/services/test_video_recorder.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import threading
 import time
 from pathlib import Path
 
@@ -75,6 +76,38 @@ def gray_frame():
     return np.zeros((48, 64), dtype=np.uint8)
 
 
+class BlockingWriteGear(FakeWriteGear):
+    """
+    Fake WriteGear that blocks inside write() to simulate a hung encoder/IO stall.
+    """
+
+    instances = []
+
+    def __init__(self, output: str, **kwargs):
+        super().__init__(output, **kwargs)
+        self.entered_write = threading.Event()
+        self.release_write = threading.Event()
+        BlockingWriteGear.instances.append(self)
+
+    def write(self, frame):
+        # Signal that the writer thread is now stuck inside write()
+        self.entered_write.set()
+        # Block until the test releases us (long timeout as safety, but test releases explicitly)
+        self.release_write.wait(timeout=60.0)
+        # If released, behave like normal FakeWriteGear
+        return super().write(frame)
+
+
+@pytest.fixture
+def patch_blocking_writegear(monkeypatch):
+    """
+    Patch module-level WriteGear to BlockingWriteGear for the hung-thread test.
+    """
+    BlockingWriteGear.instances.clear()
+    monkeypatch.setattr(vr_mod, "WriteGear", BlockingWriteGear)
+    return BlockingWriteGear
+
+
 # ----------------------------
 # Tests
 # ----------------------------
@@ -295,3 +328,42 @@ def test_overlay_frame_size_mismatch_still_detected(patch_writegear, output_path
         rec.write(np.zeros((48, 64, 3), dtype=np.uint8), timestamp=2.0)
 
     rec.stop()
+
+
+def test_stop_timeout_marks_abandoned_and_prevents_restart(
+    patch_blocking_writegear, monkeypatch, output_path, rgb_frame
+):
+    # Make stop timeout tiny so the test is fast.
+    monkeypatch.setattr(vr_mod, "STOP_JOIN_TIMEOUT", 0.01)
+
+    rec = vr_mod.VideoRecorder(output_path, frame_size=(48, 64), buffer_size=10)
+    rec.start()
+    assert rec.is_running is True
+
+    # Enqueue one frame so writer thread enters BlockingWriteGear.write() and blocks.
+    assert rec.write(rgb_frame, timestamp=1.0) is True
+
+    wg = patch_blocking_writegear.instances[0]
+    wait_until(lambda: wg.entered_write.is_set(), timeout=1.0)
+
+    # Now stop: should time out and mark abandoned
+    rec.stop()
+
+    assert rec._abandoned is True
+    err = rec._current_error()
+    assert err is not None
+    assert "Failed to stop VideoRecorder within timeout" in str(err)
+
+    # Thread should still be alive (blocked in write)
+    assert rec.is_running is True
+
+    # Restart should be prevented while abandoned
+    with pytest.raises(RuntimeError, match="Cannot restart VideoRecorder"):
+        rec.start()
+
+    # ---- Cleanup to avoid leaking a blocked daemon thread into other tests ----
+    wg.release_write.set()  # let write() return
+    wait_until(lambda: not rec.is_running, timeout=2.0)
+
+    # Call stop again to clear resources / reset flags (now it can join cleanly)
+    rec.stop()