diff --git a/frigate/comms/dispatcher.py b/frigate/comms/dispatcher.py index f9373c30ee..f0988b5710 100644 --- a/frigate/comms/dispatcher.py +++ b/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: diff --git a/frigate/comms/inter_process.py b/frigate/comms/inter_process.py index c40e53ddb7..c45f94ffc6 100644 --- a/frigate/comms/inter_process.py +++ b/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)}" diff --git a/frigate/test/test_ipc_handler_errors.py b/frigate/test/test_ipc_handler_errors.py new file mode 100644 index 0000000000..d78d09cd4b --- /dev/null +++ b/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()