fix(dispatcher): add dispatcher relay

bec_widgets/utils/bec_dispatcher.py

@@ -19,6 +19,7 @@ from bec_lib.service_config import ServiceConfig
 from qtpy.QtCore import QObject
 from qtpy.QtCore import Signal as pyqtSignal
 
+from bec_widgets.utils.error_popups import SafeSlot
 from bec_widgets.utils.rpc_logging import elapsed_seconds, format_elapsed
 from bec_widgets.utils.serialization import register_serializer_extension
 
@@ -72,6 +73,8 @@ class QtThreadSafeCallback(QObject):
     the owning widget alive. Non-method callables (lambdas, partials, module functions)
     have no owning object whose lifetime could be tracked, so they are kept alive by the
     wrapper itself; pass ``owner=`` to ``connect_slot`` to bind their lifetime to a widget.
+    Create the wrapper on the GUI thread: queued messages are delivered to its Qt slot,
+    which resolves the callback only when delivery occurs.
     """
 
     cb_signal = pyqtSignal(dict, dict)
@@ -101,7 +104,8 @@ class QtThreadSafeCallback(QObject):
             self.cb_owner = louie.saferef.safe_ref(owner) if owner is not None else None
             self._strong_cb = cb
         self.cb_ref = louie.saferef.safe_ref(cb)
-        self.cb_signal.connect(cb)
+        self._active = True
+        self.cb_signal.connect(self._deliver)
         self.topics = set()
 
     @property
@@ -120,9 +124,33 @@ class QtThreadSafeCallback(QObject):
             return False
         return self.cb_ref == other.cb_ref and self.cb_info == other.cb_info
 
+    def invalidate(self) -> None:
+        """Suppress pending deliveries and release any strongly held callback."""
+        self._active = False
+        self._strong_cb = None
+
+    @SafeSlot(dict, dict)
+    def _deliver(self, msg_content: dict, metadata: dict) -> None:
+        """Resolve and retain a live callback for the duration of this delivery."""
+        if not self._active:
+            return
+        owner = self.cb_owner() if self.cb_owner is not None else None
+        if self.cb_owner is not None and owner is None:
+            return
+        if isinstance(owner, QObject) and not shiboken6.isValid(owner):
+            return
+        if owner is not None and getattr(owner, "_destroyed", False):
+            return
+
+        # Never connect Qt directly to a plain Python receiver's bound method: a
+        # queued invocation can outlive that receiver. Resolving here also holds
+        # the receiver alive while its callback runs.
+        callback = self.cb_ref()
+        if callback is not None:
+            callback(msg_content, metadata)
+
     def __call__(self, msg_content, metadata):
-        if self.cb_ref() is None:
-            # callback has been deleted
+        if not self._active:
             return
         self.cb_signal.emit(msg_content, metadata)
 
@@ -284,11 +312,12 @@ class BECDispatcher:
             break
         else:
             return
-        self.client.connector.unregister(topics, cb=connected_slot)
         topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics)
-        self._registered_slots[connected_slot].topics.difference_update(set(topics_str))
-        if not self._registered_slots[connected_slot].topics:
-            del self._registered_slots[connected_slot]
+        if not connected_slot.topics.difference(topics_str):
+            self._release_slot(connected_slot)
+        else:
+            self.client.connector.unregister(topics, cb=connected_slot)
+            connected_slot.topics.difference_update(topics_str)
 
     def disconnect_topics(self, topics: str | list):
         """
@@ -297,16 +326,17 @@ class BECDispatcher:
         Args:
             topics(Union[str, list]): The topic(s) to disconnect from
         """
-        self.client.connector.unregister(topics)
         topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics)
 
         remove_slots = []
-        for connected_slot in self._registered_slots.values():
-            connected_slot.topics.difference_update(set(topics_str))
-
-            if not connected_slot.topics:
+        for connected_slot in list(self._registered_slots.values()):
+            if not connected_slot.topics.difference(topics_str):
+                connected_slot.invalidate()
                 remove_slots.append(connected_slot)
 
+        self.client.connector.unregister(topics)
+        for connected_slot in self._registered_slots.values():
+            connected_slot.topics.difference_update(topics_str)
         for connected_slot in remove_slots:
             self._registered_slots.pop(connected_slot, None)
 
@@ -354,6 +384,7 @@ class BECDispatcher:
 
     def _release_slot(self, qt_slot: QtThreadSafeCallback) -> None:
         """Unregister all topics of a slot wrapper and drop it from the registry."""
+        qt_slot.invalidate()
         topics = list(qt_slot.topics)
         if topics:
             self.client.connector.unregister(topics, cb=qt_slot)

tests/unit_tests/test_bec_dispatcher.py

@@ -1,12 +1,21 @@
 # pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring
+import gc
+import os
+import subprocess
+import sys
 import threading
 import time
+import weakref
+from pathlib import Path
 from unittest import mock
 
 import pytest
+import shiboken6
 from bec_lib import service_config
 from bec_lib.messages import GUIInstructionMessage, ScanMessage
 from bec_lib.serialization import MsgpackSerialization
+from qtpy.QtCore import QEvent, QObject, QThread
+from qtpy.QtWidgets import QApplication
 
 from bec_widgets.utils.bec_dispatcher import BECDispatcher, QtRedisConnector, QtThreadSafeCallback
 
@@ -272,3 +281,201 @@ def test_stop_cli_server_is_idempotent(bec_dispatcher):
         bec_dispatcher.stop_cli_server()
         bec_dispatcher.stop_cli_server()
     mock_logger.error.assert_not_called()
+
+
+def _queue_callback(wrapper):
+    """Queue a real cross-thread emission without allowing GUI delivery yet."""
+    thread = threading.Thread(target=wrapper, args=({"value": 1}, {"source": "worker"}))
+    thread.start()
+    thread.join(timeout=5)
+    assert not thread.is_alive()
+
+
+def _deliver_after_receiver_collection(cyclic):
+    """Run in a child process: the original queued bound method could segfault."""
+    app = QApplication([])
+    calls = []
+
+    class Receiver:
+        def __init__(self):
+            self.state = "alive"
+            if cyclic:
+                self.cycle = self
+
+        def receive(self, content, metadata):
+            calls.append("entered")
+            calls.append(self.state)
+
+    receiver = Receiver()
+    receiver_ref = weakref.ref(receiver)
+    wrapper = QtThreadSafeCallback(receiver.receive)
+    _queue_callback(wrapper)
+    del receiver
+    gc.collect()
+    assert receiver_ref() is None
+    assert wrapper.cb is None
+
+    # A second queued callback proves that Qt has drained the queued deliveries.
+    delivered = []
+    sentinel = QtThreadSafeCallback(lambda *_: delivered.append(True))
+    _queue_callback(sentinel)
+    app.processEvents()
+    assert delivered == [True]
+    assert calls == []
+    print("Collected receiver was not called", flush=True)
+
+
+@pytest.mark.parametrize("cyclic", [False, True], ids=["refcount", "cyclic-gc"])
+def test_qthreadsafe_callback_skips_collected_receiver_in_subprocess(cyclic):
+    """Keep native crash regressions from terminating the pytest process."""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-X",
+            "faulthandler",
+            "-c",
+            "from tests.unit_tests.test_bec_dispatcher import "
+            f"_deliver_after_receiver_collection; _deliver_after_receiver_collection({cyclic!r})",
+        ],
+        cwd=Path(__file__).resolve().parents[2],
+        env={**os.environ, "QT_QPA_PLATFORM": "offscreen", "OPHYD_CONTROL_LAYER": "dummy"},
+        capture_output=True,
+        text=True,
+        timeout=30,
+        check=False,
+    )
+    assert result.returncode == 0, (
+        f"Queued callback subprocess exited with {result.returncode}\n"
+        f"{result.stdout[-6000:]}\n{result.stderr[-6000:]}"
+    )
+    assert "Collected receiver was not called" in result.stdout
+
+
+def test_qthreadsafe_callback_delivers_live_receiver_on_gui_thread(qapp):
+    calls = []
+
+    class Receiver:
+        def receive(self, content, metadata):
+            calls.append((content, metadata, QThread.currentThread()))
+
+    receiver = Receiver()
+    wrapper = QtThreadSafeCallback(receiver.receive)
+    _queue_callback(wrapper)
+    assert calls == []
+
+    qapp.processEvents()
+
+    assert calls == [({"value": 1}, {"source": "worker"}, qapp.thread())]
+
+
+@pytest.mark.parametrize(
+    "disconnect", ["disconnect_owner", "disconnect_slot", "disconnect_topics", "disconnect_all"]
+)
+def test_dispatcher_disconnect_cancels_queued_delivery(bec_dispatcher, qapp, disconnect):
+    owner = QObject()
+    callback = mock.Mock(spec=[])
+    bec_dispatcher.connect_slot(callback, "lifetime-test", owner=owner)
+    wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback)
+    _queue_callback(wrapper)
+    callback.assert_not_called()
+
+    if disconnect == "disconnect_owner":
+        bec_dispatcher.disconnect_owner(owner)
+    elif disconnect == "disconnect_slot":
+        bec_dispatcher.disconnect_slot(callback, "lifetime-test")
+    elif disconnect == "disconnect_topics":
+        bec_dispatcher.disconnect_topics("lifetime-test")
+    else:
+        bec_dispatcher.disconnect_all()
+
+    # The connector may still hold an in-flight reference to the released wrapper.
+    _queue_callback(wrapper)
+    qapp.processEvents()
+    callback.assert_not_called()
+    assert wrapper not in bec_dispatcher._registered_slots
+
+
+@pytest.mark.parametrize("disconnect", ["disconnect_slot", "disconnect_topics"])
+def test_dispatcher_partial_disconnect_preserves_delivery(bec_dispatcher, qapp, disconnect):
+    owner = QObject()
+    callback = mock.Mock(spec=[])
+    bec_dispatcher.connect_slot(callback, ["removed-topic", "retained-topic"], owner=owner)
+    wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback)
+    _queue_callback(wrapper)
+
+    if disconnect == "disconnect_slot":
+        bec_dispatcher.disconnect_slot(callback, "removed-topic")
+    else:
+        bec_dispatcher.disconnect_topics("removed-topic")
+
+    qapp.processEvents()
+    callback.assert_called_once_with({"value": 1}, {"source": "worker"})
+    assert wrapper.topics == {"retained-topic"}
+    _queue_callback(wrapper)
+    qapp.processEvents()
+    assert callback.call_count == 2
+
+
+def test_dispatcher_reconnect_does_not_reactivate_old_delivery(bec_dispatcher, qapp):
+    owner = QObject()
+    callback = mock.Mock(spec=[])
+    bec_dispatcher.connect_slot(callback, "lifetime-test", owner=owner)
+    old_wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback)
+    _queue_callback(old_wrapper)
+    bec_dispatcher.disconnect_owner(owner)
+
+    bec_dispatcher.connect_slot(callback, "lifetime-test", owner=owner)
+    new_wrapper = next(slot for slot in bec_dispatcher._registered_slots if slot.cb is callback)
+    assert new_wrapper is not old_wrapper
+    _queue_callback(old_wrapper)
+    _queue_callback(new_wrapper)
+    qapp.processEvents()
+
+    callback.assert_called_once_with({"value": 1}, {"source": "worker"})
+
+
+def test_dispatcher_disconnect_releases_lambda_capture(bec_dispatcher):
+    owner = QObject()
+    owner_ref = weakref.ref(owner)
+    bec_dispatcher.connect_slot(
+        lambda _content, _metadata, target=owner: target.objectName(), "lifetime-test", owner=owner
+    )
+    wrapper = next(
+        slot
+        for slot in bec_dispatcher._registered_slots
+        if slot.cb_owner is not None and slot.cb_owner() is owner
+    )
+    callback_ref = weakref.ref(wrapper.cb)
+    del owner
+    gc.collect()
+    assert owner_ref() is not None
+
+    bec_dispatcher.disconnect_owner(owner_ref())
+    gc.collect()
+
+    # An in-flight wrapper must not keep the lambda or its captured owner alive.
+    assert callback_ref() is None
+    assert owner_ref() is None
+
+
+@pytest.mark.parametrize("teardown", ["native-delete", "cleanup", "garbage-collect"])
+def test_qthreadsafe_callback_checks_lambda_owner_at_delivery(qapp, teardown):
+    calls = []
+    owner = QObject()
+    owner_ref = weakref.ref(owner)
+    wrapper = QtThreadSafeCallback(lambda *_: calls.append(True), owner=owner)
+    _queue_callback(wrapper)
+
+    if teardown == "native-delete":
+        owner.deleteLater()
+        qapp.sendPostedEvents(owner, QEvent.Type.DeferredDelete)
+        assert not shiboken6.isValid(owner)
+    elif teardown == "cleanup":
+        owner._destroyed = True
+    else:
+        del owner
+        gc.collect()
+        assert owner_ref() is None
+
+    qapp.processEvents()
+    assert calls == []