Make camera streaming non-blocking to keep MQTT alive

inorbit_edge/robot.py

@@ -1093,6 +1093,12 @@ def publish(image, width, height, ts):
                 camera_id, image, int(width), int(height), int(ts)
             )
 
+        # Tear down any streamer previously registered under this id so its
+        # worker thread is not leaked.
+        existing_streamer = self.camera_streamers.get(camera_id)
+        if existing_streamer is not None:
+            existing_streamer.shutdown()
+
         self.camera_streamers[camera_id] = CameraStreamer(camera, publish)
         with self.camera_streaming_mutex:
             if self.camera_streaming_on:
@@ -1202,6 +1208,23 @@ def disconnect(self):
         self.logger.info("Ending robot session")
         self._stop_cameras_streaming()
 
+        # Permanently terminate the camera worker threads. Signal all first,
+        # then join, so the total wait is the slowest single teardown rather
+        # than the sum. The join is capped because a worker can be mid-open on
+        # an unreachable stream; daemon threads guarantee process exit even if
+        # a worker overruns the cap.
+        camera_teardown_timeout = 12
+        streamers = list(self.camera_streamers.values())
+        for streamer in streamers:
+            streamer.shutdown()
+        for streamer in streamers:
+            streamer.join(timeout=camera_teardown_timeout)
+            if streamer.is_alive():
+                self.logger.warning(
+                    "Camera streamer worker did not exit within "
+                    f"{camera_teardown_timeout}s"
+                )
+
         # Send offline status (best effort, non-blocking)
         # InOrbit will detect offline via data absence if this fails
         self._send_robot_status(online=False)

inorbit_edge/tests/test_video.py

@@ -1,11 +1,46 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
+import threading
+import time
+
 from inorbit_edge.robot import RobotSession
-from inorbit_edge.video import OpenCVCamera
+from inorbit_edge.video import CameraStreamer, OpenCVCamera
 from inorbit_edge.robot import INORBIT_MODULE_CAMERAS
 
 
+def _wait_until(predicate, timeout=2.0):
+    """Poll ``predicate`` until it is truthy or ``timeout`` elapses."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return bool(predicate())
+
+
+class FakeCamera:
+    """Minimal Camera double that records open/close calls without touching cv2."""
+
+    rate = 50
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self.opens = 0
+        self.closes = 0
+
+    def open(self):
+        with self._lock:
+            self.opens += 1
+
+    def close(self):
+        with self._lock:
+            self.closes += 1
+
+    def get_frame_jpg(self):
+        return None, 0, 0, 0
+
+
 def test_robot_session_register_camera(
     mock_mqtt_client, mock_inorbit_api, mocker, mock_sleep
 ):
@@ -33,6 +68,9 @@ def test_robot_session_register_camera(
     robot_session._handle_in_cmd(
         f"load_module|{INORBIT_MODULE_CAMERAS}|{runlevel}".encode()
     )
+    # The worker opens the camera asynchronously; wait until it has actually
+    # started before tearing down, so teardown has a live session to close.
+    assert _wait_until(lambda: opencv_camera.capture_thread is not None, timeout=5)
     # Override _is_disconnected method to simulate successful MQTT client disconnection
     robot_session._is_disconnected = lambda: True
     robot_session.disconnect()
@@ -41,6 +79,71 @@ def test_robot_session_register_camera(
     camera_stream_stop_spy.assert_called_once()
     opencv_camera_close_spy.assert_called_once()
 
-    # Verify threads are stopped
-    assert not camera_stream.thread.is_alive()
-    assert not opencv_camera.capture_thread.is_alive()
+    # disconnect() shuts down and joins the worker, so it must be gone. Join again
+    # with a timeout to keep the assertion non-flaky regardless of scheduling.
+    camera_stream.join(timeout=15)
+    assert not camera_stream.is_alive()
+    if opencv_camera.capture_thread is not None:
+        opencv_camera.capture_thread.join(timeout=10)
+        assert not opencv_camera.capture_thread.is_alive()
+
+
+def test_camera_streamer_start_stop_start_keeps_single_worker():
+    """A stop->start cycle resumes streaming on the same single worker thread."""
+    camera = FakeCamera()
+    streamer = CameraStreamer(camera, lambda *args: None)
+    worker = streamer._worker
+    try:
+        streamer.start()
+        assert _wait_until(lambda: camera.opens >= 1)
+
+        streamer.stop()
+        assert _wait_until(lambda: camera.closes >= 1)
+
+        streamer.start()
+        assert _wait_until(lambda: camera.opens >= 2)
+
+        # Same worker thread throughout -- no respawn, no ghost/duplicate.
+        assert streamer._worker is worker
+        assert streamer.is_alive()
+    finally:
+        streamer.shutdown()
+        streamer.join(timeout=5)
+        assert not streamer.is_alive()
+
+
+def test_camera_streamer_shutdown_terminates_worker():
+    camera = FakeCamera()
+    streamer = CameraStreamer(camera, lambda *args: None)
+    streamer.start()
+    assert _wait_until(lambda: camera.opens >= 1)
+
+    streamer.shutdown()
+    streamer.join(timeout=5)
+    assert not streamer.is_alive()
+    # The camera is released on the way out.
+    assert _wait_until(lambda: camera.closes >= 1)
+
+
+def test_register_camera_twice_replaces_and_shuts_down_old(
+    mock_mqtt_client, mock_inorbit_api, mocker, mock_sleep
+):
+    robot_session = RobotSession(
+        robot_id="id_123", robot_name="name_123", api_key="apikey_123"
+    )
+    camera_id = "cam0"
+
+    # Streaming is off (no load_module), so workers stay idle -- no cv2 open.
+    robot_session.register_camera(camera_id, OpenCVCamera(None))
+    first = robot_session.camera_streamers[camera_id]
+
+    robot_session.register_camera(camera_id, OpenCVCamera(None))
+    second = robot_session.camera_streamers[camera_id]
+
+    assert second is not first
+    first.join(timeout=5)
+    assert not first.is_alive()  # old streamer's worker was shut down
+    assert second.is_alive()
+
+    second.shutdown()
+    second.join(timeout=5)

inorbit_edge/video.py

@@ -76,14 +76,17 @@ def open(self):
                 self.capture = cv2.VideoCapture(self.video_url)
             if not self.running:
                 self.running = True
-                self.capture_thread = threading.Thread(target=self._run)
+                self.capture_thread = threading.Thread(target=self._run, daemon=True)
                 self.capture_thread.start()
 
     def close(self):
         """Closes the capturing device / stream"""
         self.running = False
-        self.logger.info("Waiting for the capture thread to finish")
-        self.capture_thread.join()
+
+        if self.capture_thread is not None:
+            self.logger.info("Waiting for the capture thread to finish")
+            self.capture_thread.join()
+            self.capture_thread = None
         with self.capture_mutex:
             if self.capture is not None:
                 self.capture.release()
@@ -114,43 +117,91 @@ def _run(self):
 
 
 class CameraStreamer:
-    """Streams video from a camera to InOrbit"""
+    """Streams video from a camera to InOrbit.
+
+    A single long-lived worker thread owns the camera lifecycle. ``start``,
+    ``stop`` and ``shutdown`` only toggle threading Events, so they never block
+    the caller -- in particular the MQTT callback thread that dispatches module
+    load/unload, where a blocking call would starve the keepalive and drop the
+    connection. Opening/closing the camera (which can block while a stream is
+    unreachable) happens entirely on the worker thread.
+
+    Because there is exactly one worker, ``camera.open()`` and ``camera.close()``
+    are always serialized -- the device is never opened and closed concurrently,
+    which removes the start/stop races of a spawn-per-start model.
+    """
+
+    # Backoff after a failed streaming session so a permanently-broken URL does
+    # not hot-loop open->fail->open. A pending stop/shutdown short-circuits it.
+    BACKOFF_SECONDS = 1.0
 
     def __init__(self, camera, publish_frame_callback):
         self.logger = logging.getLogger(__class__.__name__)
         self.camera = camera
-        self.running = False
-        self.mutex = threading.Lock()
-        self.thread = None
         self.publish_frame = publish_frame_callback
-        self.must_stop = False
+        # Set => the worker should be streaming. Cleared => paused.
+        self._streaming = threading.Event()
+        # Set => the worker should exit permanently.
+        self._shutdown = threading.Event()
+        self._worker = threading.Thread(target=self._run, daemon=True)
+        self._worker.start()
 
     def start(self):
-        """Streams video to the platform"""
-        with self.mutex:
-            self.must_stop = False
-            if not self.running:
-                self.running = True
-                self.thread = threading.Thread(target=self._run)
-                self.thread.start()
+        """Request streaming. Non-blocking; safe to call repeatedly."""
+        self._streaming.set()
 
     def stop(self):
-        """Stops streaming video to the platform"""
-        self.must_stop = True
-        self.logger.info("Waiting for the streaming thread to finish")
-        self.thread.join()
+        """Pause streaming. Non-blocking; safe to call repeatedly.
+
+        The worker thread stays alive and idle, ready for a later ``start()``.
+        Use ``shutdown()`` to terminate the worker permanently.
+        """
+        self._streaming.clear()
+
+    def shutdown(self):
+        """Permanently stop the worker. Non-blocking.
+
+        ``_shutdown`` is set before ``_streaming`` so a worker waking from the
+        idle wait observes the shutdown rather than starting a doomed session.
+        """
+        self._shutdown.set()
+        self._streaming.set()
+
+    def join(self, timeout=None):
+        """Wait for the worker thread to exit (after ``shutdown()``)."""
+        self._worker.join(timeout)
+
+    def is_alive(self):
+        """Return True while the worker thread is running."""
+        return self._worker.is_alive()
 
     def _run(self):
-        """This thread takes care of getting video from a camera at the desired rate,
-        converting it to the right format and publishing the video frames"""
-        self.camera.open()
-        while True:
-            jpg, width, height, ts = self.camera.get_frame_jpg()
-            if jpg is not None:
-                self.publish_frame(jpg, width, height, ts)
-            time.sleep(1.0 / self.camera.rate)
-            with self.mutex:
-                if self.must_stop:
-                    break
-        self.camera.close()
-        self.running = False
+        """Worker loop: idle until streaming is requested, then grab frames from
+        the camera at the desired rate and publish them until paused or shut down.
+
+        The per-session body is wrapped so a failed ``open()``/``get_frame_jpg()``
+        cannot kill the long-lived worker -- it logs, backs off, and waits for the
+        next ``start()``.
+        """
+        while not self._shutdown.is_set():
+            # Idle until streaming or shutdown is requested.
+            self._streaming.wait()
+            if self._shutdown.is_set():
+                break
+            try:
+                self.camera.open()
+                while self._streaming.is_set() and not self._shutdown.is_set():
+                    jpg, width, height, ts = self.camera.get_frame_jpg()
+                    if jpg is not None:
+                        self.publish_frame(jpg, width, height, ts)
+                    time.sleep(1.0 / self.camera.rate)
+            except Exception:
+                self.logger.exception("Camera streaming session failed")
+                # Back off so a permanently-failing open does not hot-loop while
+                # streaming is still requested.
+                self._shutdown.wait(timeout=self.BACKOFF_SECONDS)
+            finally:
+                try:
+                    self.camera.close()
+                except Exception:
+                    self.logger.exception("Error closing camera")