fix: close messagebus client before main() returns on shutdown

ovos_core/__main__.py

@@ -44,7 +44,7 @@ def main(alive_hook=on_alive, started_hook=on_started, ready_hook=on_ready,
 
     # Connect this process to the OpenVoiceOS message bus
     bus = MessageBusClient()
-    bus.run_in_thread()
+    bus_thread = bus.run_in_thread()
     bus.connected_event.wait()
 
     skill_manager = SkillManager(bus, watchdog,
@@ -65,6 +65,28 @@ def main(alive_hook=on_alive, started_hook=on_started, ready_hook=on_ready,
 
     skill_manager.shutdown()
 
+    # Stop the messagebus websocket thread and its event dispatcher before
+    # the interpreter starts tearing down. `bus.run_in_thread()` spawns a
+    # daemon thread that keeps receiving messages and dispatching them onto
+    # `bus.emitter`'s internal ThreadPoolExecutor for as long as the socket
+    # is open. If that thread is still alive when Python begins interpreter
+    # shutdown, an incoming message can call `emitter.emit()` -> `executor
+    # .submit()` after the executor has already been torn down, raising
+    # `RuntimeError: cannot schedule new futures after shutdown` from a
+    # background thread and leaving the process unable to exit cleanly
+    # (systemd then waits out the stop timeout and SIGKILLs it).
+    #
+    # `bus.close()` itself is fire-and-forget: it signals the run_forever
+    # loop to stop and closes the socket, but does not wait for the
+    # receiver thread to actually exit. We join that thread here, with a
+    # bounded timeout, to narrow the window further. This is not a
+    # guarantee the thread has fully stopped by the time we return (the
+    # join can time out), but it removes most of the residual race where a
+    # buffered inbound frame is still being dispatched during teardown.
+    bus.close()
+    if bus_thread is not None:
+        bus_thread.join(timeout=3)
+
     LOG.info('Skills service shutdown complete!')
 
 

test/unittests/test_main.py

@@ -0,0 +1,61 @@
+"""Regression test for the shutdown-hang bug where the SkillManager's
+messagebus client is never closed, leaving the websocket dispatch thread
+alive to race against interpreter teardown and raise
+`RuntimeError: cannot schedule new futures after shutdown`
+(https://github.com/OpenVoiceOS/ovos-core - shutdown hang / SIGKILL after
+60s on `systemctl restart`).
+"""
+import unittest
+from unittest.mock import MagicMock, patch
+
+
+class TestMainShutdown(unittest.TestCase):
+
+    @patch('ovos_core.__main__.setup_locale')
+    @patch('ovos_core.__main__.init_service_logger')
+    @patch('ovos_core.__main__.wait_for_exit_signal')
+    @patch('ovos_core.__main__.SkillManager')
+    @patch('ovos_core.__main__.MessageBusClient')
+    def test_bus_is_closed_before_main_returns(self, mock_bus_cls, mock_manager_cls,
+                                               mock_wait, mock_init_logger,
+                                               mock_setup_locale):
+        """After the exit signal arrives and the skill manager has been
+        shut down, `main()` must close the bus's websocket connection and
+        join the receiver thread so the background dispatch thread stops
+        before the interpreter starts tearing down. Without this, the
+        daemon thread spawned by `bus.run_in_thread()` keeps calling
+        `emitter.emit()` -> `executor.submit()` and can lose the race with
+        the executor/interpreter shutdown, raising `RuntimeError: cannot
+        schedule new futures after shutdown` and hanging the process until
+        SIGKILL. `bus.close()` alone is fire-and-forget and does not wait
+        for the thread to exit, so `main()` must also join it with a
+        bounded timeout.
+        """
+        from ovos_core.__main__ import main
+
+        mock_bus = MagicMock()
+        mock_bus_thread = MagicMock()
+        mock_bus.run_in_thread.return_value = mock_bus_thread
+        mock_bus_cls.return_value = mock_bus
+        mock_manager = MagicMock()
+        mock_manager_cls.return_value = mock_manager
+
+        main()
+
+        # shutdown must happen before the bus is closed, and both must
+        # happen before main() returns
+        mock_manager.shutdown.assert_called_once()
+        mock_bus.close.assert_called_once()
+
+        # the receiver thread returned by `bus.run_in_thread()` must be
+        # joined with a bounded timeout after `bus.close()`, so `main()`
+        # doesn't block forever if the thread never stops
+        mock_bus_thread.join.assert_called_once()
+        join_args, join_kwargs = mock_bus_thread.join.call_args
+        timeout = join_kwargs.get('timeout', join_args[0] if join_args else None)
+        self.assertIsNotNone(timeout)
+        self.assertGreater(timeout, 0)
+
+
+if __name__ == '__main__':
+    unittest.main()