fix(actor_hli): cleanup of the redis connections on shutdown

Author: wyzula-jan

bec_lib/bec_lib/builtin_actor_hli.py

@@ -12,14 +12,10 @@
 if TYPE_CHECKING:
     from bec_lib.client import BECClient
 
-VAR_PREFIX = "_BuiltinActors"
-
 
 class ScanInterlockHli:
     def __init__(self, client: "BECClient", parent: "BuiltinActorHli") -> None:
         self._client = client
-        self._parent = parent
-        self._actor_name = "ScanInterlockActor"
         self._enabled = RedisConfigValue(
             connector=self._client.connector, endpoint=MessageEndpoints.scan_interlock_enabled()
         )
@@ -102,8 +98,17 @@ def clear_all(self):
             {"data": ScanInterlockModifyStateTableMessage(action="remove_all")},
         )
 
+    def shutdown(self):
+        """Unregister the config-value stream subscriptions from the connector."""
+        self._enabled.unregister_all()
+        self._trigger_setting.unregister_all()
+
 
 class BuiltinActorHli:
     def __init__(self, client: "BECClient") -> None:
         self._client = client
         self.scan_interlock = ScanInterlockHli(self._client, self)
+
+    def shutdown(self):
+        """Tear down builtin-actor client subscriptions (called on client shutdown)."""
+        self.scan_interlock.shutdown()

bec_lib/bec_lib/client.py

@@ -135,8 +135,8 @@ def __init__(
         if self._initialized:
             return
         self.__init_params: _InitParams = {
-            "config": config if config is not None else ServiceConfig(config_name="client"),
-            "connector_cls": connector_cls if connector_cls is not None else RedisConnector,
+            "config": (config if config is not None else ServiceConfig(config_name="client")),
+            "connector_cls": (connector_cls if connector_cls is not None else RedisConnector),
             "wait_for_server": wait_for_server,
             "prompt_for_acl": prompt_for_acl,
         }
@@ -164,6 +164,7 @@ def __init__(
         self._system_user = ""
         self.beamline_states = None
         self.messaging: MessagingContainer = None  # type: ignore
+        self.builtin_actors: BuiltinActorHli = None  # type: ignore
 
     def __new__(cls, *args, forced=False, **kwargs):
         if forced or BECClient._client is None:
@@ -336,6 +337,9 @@ def shutdown(self, per_thread_timeout_s: float | None = None):
         if self.history is not None:
             # pylint: disable=protected-access
             self.history._shutdown()
+        if self.builtin_actors is not None:
+            self.builtin_actors.shutdown()
+            self.builtin_actors = None  # type: ignore
         bec_logger.logger.remove()
         self.started = False
 

bec_server/bec_server/actors/builtin_actor_manager.py

@@ -21,6 +21,7 @@
 
 
 class ActorDict(dict):
+
     def __setitem__(self, key: type[ActorType], value: tuple[ActorType, Thread, Event], /) -> None:
         return super().__setitem__(key, value)
 
@@ -66,7 +67,7 @@ def _ping_clients(self, actor_name: str):
         )
 
     def _interlock_enabled_changed(self, enabled: bool):
-        self._start_actor(ScanInterlockActor) if enabled else self._stop_actor(ScanInterlockActor)
+        (self._start_actor(ScanInterlockActor) if enabled else self._stop_actor(ScanInterlockActor))
 
     def _start_all(self):
         if self._scan_interlock_enabled.value:
@@ -95,6 +96,7 @@ def _stop_actor(self, actor_class: type[ActorBase]):
         del actor
 
     def shutdown(self):
+        self._scan_interlock_enabled.unregister_all()
         for actor in list(self._actors_threads_and_stops):
             self._stop_actor(actor)
         self._client.shutdown()

bec_server/bec_server/actors/scan_interlock.py

@@ -126,5 +126,6 @@ def run(self):
         self._unlock()
 
     def stop(self, *_):
+        self._restart_scan_on_lock.unregister_all()
         self._unlock()
         super().stop()