some cleanup

Author: CamDavidsonPilon

CHANGELOG.md

@@ -3,6 +3,7 @@
 #### Bug fixes
 
 - Added an app-update database migration to repair stale open assignment-history rows in existing leader databases. This was caused by the new "re-assignment" dialog box in the Pioreactor's page. The biggest impact was duplicate logs showing up on the Logs UI page.
+- Fixed repeated in-process actions such as dosing/pump runs to clean up MQTT listeners, signal handlers, and short-lived MQTT clients more reliably, preventing runaway localhost socket/file-descriptor growth in long-lived jobs.
 
 ### 26.3.3
 

core/pioreactor/pubsub.py

@@ -54,6 +54,7 @@ def shutdown(self) -> None:
                 self._reset_sockets(sockpair_only=True)
 
     def loop_stop(self) -> MQTTErrorCode:
+        # fast exits
         thread = self._thread
         if thread is None:
             return MQTTErrorCode.MQTT_ERR_INVAL
@@ -285,6 +286,18 @@ def subscribe_and_callback(
     """
     assert callable(callback), "callback should be callable - do you need to change the order of arguments?"
 
+    def remove_callback_subscription(client: Client, topic: str) -> None:
+        # This cleanup assumes effective ownership of this topic filter on the provided client.
+        # Paho stores one callback per topic filter, and unsubscribe() removes the filter for the
+        # whole client. On a genuinely shared long-lived client, this can clobber another caller's
+        # callback/subscription if they reused the same topic. Today we accept that exclusivity risk
+        # to avoid unbounded listener growth, but it is a known failure mode of this teardown path.
+        with suppress(KeyError, ValueError):
+            client.message_callback_remove(topic)
+
+        with suppress(ValueError):
+            client.unsubscribe(topic)
+
     def wrap_callback(actual_callback: Callable[[pt.MQTTMessage], Any]) -> Callable[..., Any]:
         def _callback(client: Client, userdata: dict[str, Any], message: pt.MQTTMessage) -> Any:
             try:
@@ -346,26 +359,13 @@ def on_subscribe(
             if on_cleanup is not None:
 
                 def cleanup_subscription(topic: str = topic) -> None:
-                    _remove_callback_subscription(client, topic)
+                    remove_callback_subscription(client, topic)
 
                 on_cleanup.append(cleanup_subscription)
 
     return client
 
 
-def _remove_callback_subscription(client: Client, topic: str) -> None:
-    # This cleanup assumes effective ownership of this topic filter on the provided client.
-    # Paho stores one callback per topic filter, and unsubscribe() removes the filter for the
-    # whole client. On a genuinely shared long-lived client, this can clobber another caller's
-    # callback/subscription if they reused the same topic. Today we accept that exclusivity risk
-    # to avoid unbounded listener growth, but it is a known failure mode of this teardown path.
-    with suppress(KeyError, ValueError):
-        client.message_callback_remove(topic)
-
-    with suppress(ValueError):
-        client.unsubscribe(topic)
-
-
 def prune_retained_messages(topics_to_prune: str = "#") -> None:
     topics = []
 

core/pioreactor/utils/__init__.py

@@ -210,7 +210,7 @@ def __init__(
         self.unit = unit
         self.experiment = experiment
         self.name = name
-        self.state = st("init")
+        self.state = st.INIT
         self.exit_event = Event()
         self._source = source
         self.is_long_running_job = is_long_running_job
@@ -269,7 +269,7 @@ def __init__(
             )
         assert self.mqtt_client is not None
 
-        self.state = st("init")
+        self.state = st.INIT
         self.publish_setting("$state", self.state)
 
         # Teardown for signal handlers and passive MQTT listeners is centralized in __exit__.
@@ -290,13 +290,13 @@ def _on_disconnect(self, *args: object) -> None:
         self._exit()
 
     def __enter__(self) -> Self:
-        self.state = st("ready")
+        self.state = st.READY
         self.publish_setting("$state", self.state)
 
         return self
 
     def __exit__(self, *args: object) -> None:
-        self.state = st("disconnected")
+        self.state = st.DISCONNECTED
         self._exit()
         try:
             self.publish_setting("$state", self.state)
@@ -335,8 +335,8 @@ def start_passive_listeners(self) -> None:
 
     def _remove_passive_listeners(self) -> None:
         while self._mqtt_cleanup_callables:
-            cleanup = self._mqtt_cleanup_callables.pop()
-            cleanup()
+            _cleanup = self._mqtt_cleanup_callables.pop()
+            _cleanup()
 
     def _remove_signal_handlers(self) -> None:
         if not self._registered_signal_handlers:

core/tests/test_automation_yamls.py

@@ -3,6 +3,7 @@
 from functools import cache
 from typing import Any
 
+import pytest
 from pioreactor.automations import *  # noqa: F403, F401
 from pioreactor.background_jobs.dosing_automation import available_dosing_automations
 from pioreactor.background_jobs.dosing_automation import DosingAutomationJobContrib
@@ -46,6 +47,7 @@ def get_automation_yaml_filename(type_: str, automation_name: str) -> str:
     raise FileNotFoundError(f"Unable to locate YAML for automation '{automation_name}' in '{type_}'.")
 
 
+@pytest.mark.slow
 def test_automations_and_their_yamls_have_the_same_data() -> None:
     try:
         for type_, available_automations in [

core/tests/test_dosing_automation.py

@@ -1032,10 +1032,7 @@ def publish(self, topic: str, payload=None, qos: int = 0, **kwargs):
                 stop_messages.append((topic, payload, qos))
             return None
 
-        def loop_stop(self) -> None:
-            return None
-
-        def disconnect(self) -> None:
+        def shutdown(self) -> None:
             return None
 
     class StubAutomation(DosingAutomationJob):