keep the ipc thread alive when a handler raises

An exception from any dispatcher handler escaped InterProcessCommunicator.read, which only caught ZMQError. The REP thread died owing a reply, so every InterProcessRequestor blocked forever. read now logs the error and replies with an empty response, and Dispatcher.publish and publish_local catch per communicator so one failing communicator can't take down the rest.

frigate/comms/dispatcher.py

@@ -490,7 +490,12 @@ class Dispatcher:
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """Handle publishing to communicators."""
         for comm in self.comms:
-            comm.publish(topic, payload, retain)
+            try:
+                comm.publish(topic, payload, retain)
+            except Exception:
+                logger.exception(
+                    "Error publishing %s to %s", topic, type(comm).__name__
+                )
 
     def publish_local(self, topic: str, payload: Any) -> None:
         """Publish to every communicator except MQTT.
@@ -501,7 +506,12 @@ class Dispatcher:
             if isinstance(comm, MqttClient):
                 continue
 
-            comm.publish(topic, payload, False)
+            try:
+                comm.publish(topic, payload, False)
+            except Exception:
+                logger.exception(
+                    "Error publishing %s to %s", topic, type(comm).__name__
+                )
 
     def _publish_notices(self) -> None:
         if self.notice_registry is None:

frigate/comms/inter_process.py

@@ -50,7 +50,13 @@ class InterProcessCommunicator(Communicator):
 
                     if isinstance(raw, list):
                         (topic, value) = raw
-                        response = self._dispatcher(topic, value)
+
+                        # REP owes a reply, so a handler error must not end the thread
+                        try:
+                            response = self._dispatcher(topic, value)
+                        except Exception:
+                            logger.exception("Error handling IPC topic %s", topic)
+                            response = None
                     else:
                         logging.warning(
                             f"Received unexpected data type in ZMQ recv_json: {type(raw)}"

frigate/test/test_ipc_handler_errors.py

@@ -0,0 +1,76 @@
+"""Tests that IPC handler errors cannot stall inter-process communication."""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import zmq
+
+from frigate.comms.dispatcher import Dispatcher
+from frigate.comms.inter_process import InterProcessCommunicator
+
+
+class TestInterProcessHandlerErrors(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.address = f"ipc://{os.path.join(self.tmp.name, 'comms')}"
+
+        with patch("frigate.comms.inter_process.SOCKET_REP_REQ", self.address):
+            self.communicator = InterProcessCommunicator()
+
+        self.context = zmq.Context()
+        self.requestor = self.context.socket(zmq.REQ)
+        self.requestor.setsockopt(zmq.RCVTIMEO, 5000)
+        self.requestor.connect(self.address)
+
+    def tearDown(self):
+        self.requestor.close(linger=0)
+        self.context.destroy(linger=0)
+        self.communicator.stop()
+        self.tmp.cleanup()
+
+    def test_handler_exception_still_replies_and_keeps_reading(self):
+        def dispatcher(topic, value):
+            if topic == "boom":
+                raise ValueError("handler failed")
+            return value
+
+        self.communicator.subscribe(dispatcher)
+        self.communicator.start()
+
+        with self.assertLogs("frigate.comms.inter_process", level="ERROR"):
+            self.requestor.send_json(("boom", None))
+            self.assertEqual(self.requestor.recv_json(), [])
+
+        self.requestor.send_json(("echo", "ok"))
+        self.assertEqual(self.requestor.recv_json(), "ok")
+        self.assertTrue(self.communicator.reader_thread.is_alive())
+
+
+class TestDispatcherPublishErrors(unittest.TestCase):
+    def test_failing_communicator_does_not_block_the_others(self):
+        failing = MagicMock()
+        failing.publish.side_effect = ValueError("publish failed")
+        healthy = MagicMock()
+
+        with (
+            patch("frigate.comms.dispatcher.CameraActivityManager"),
+            patch("frigate.comms.dispatcher.AudioActivityManager"),
+        ):
+            dispatcher = Dispatcher(
+                MagicMock(), MagicMock(), MagicMock(), {}, [failing, healthy]
+            )
+
+        healthy.publish.reset_mock()
+
+        with self.assertLogs("frigate.comms.dispatcher", level="ERROR"):
+            dispatcher.publish("reviews", "{}")
+            dispatcher.publish_local("notices", "[]")
+
+        healthy.publish.assert_any_call("reviews", "{}", False)
+        healthy.publish.assert_any_call("notices", "[]", False)
+
+
+if __name__ == "__main__":
+    unittest.main()