fix(dispatcher): select the wrapper subscribed to the requested topic on disconnect

bec_widgets/utils/bec_dispatcher.py

@@ -265,28 +265,42 @@ class BECDispatcher:
         """
         Disconnect a slot from a topic.
 
+        Among the wrappers matching the callback (and ``cb_info``, when given), the one
+        actually subscribed to the requested topics is released. If none of them is
+        subscribed to these topics, nothing is released and a warning names the actual
+        subscriptions — a blind unregister on the wrong wrapper would silently no-op.
+
         Args:
             slot(Callable): The slot to disconnect
             topics EndpointInfo | str | list[EndpointInfo] | list[str]: A topic or list of topics to unsub from.
             cb_info(dict | None): When the same slot was registered multiple times with
                 different cb_info payloads (e.g. per-signal async subscriptions), pass the
-                payload to select the exact registration; without it the first wrapper
-                matching the callback is used.
+                payload to narrow the candidate registrations before the topic match.
         """
         # find the right slot to disconnect from ;
         # slot callbacks are wrapped in QtThreadSafeCallback objects,
         # but the slot we receive here is the original callable
-        for connected_slot in self._registered_slots.values():
-            if connected_slot.cb != slot:
-                continue
-            if cb_info is not None and connected_slot.cb_info != cb_info:
-                continue
-            break
-        else:
+        topics_str, _ = self.client.connector.extract_raw_endpoints_from_info(topics)
+        requested = set(topics_str)
+        candidates = [
+            connected_slot
+            for connected_slot in self._registered_slots.values()
+            if connected_slot.cb == slot and (cb_info is None or connected_slot.cb_info == cb_info)
+        ]
+        # Among wrappers for this callback, pick the one actually subscribed to the
+        # requested topics; a bare unregister on the wrong wrapper would silently no-op
+        # and leave the real subscription alive.
+        connected_slot = next((s for s in candidates if s.topics & requested), None)
+        if connected_slot is None:
+            if candidates:
+                logger.warning(
+                    f"disconnect_slot({slot!r}): no registration matches topics "
+                    f"{sorted(requested)}; the slot is subscribed to "
+                    f"{sorted(set().union(*(s.topics for s in candidates)))}. Nothing released."
+                )
             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))
+        self._registered_slots[connected_slot].topics.difference_update(requested)
         if not self._registered_slots[connected_slot].topics:
             del self._registered_slots[connected_slot]
 

tests/unit_tests/test_bec_dispatcher.py

@@ -114,6 +114,54 @@ def test_dispatcher_disconnect_one(bec_dispatcher_w_connector, qtbot, send_msg_e
     cb2.assert_called_once()
 
 
+@pytest.mark.parametrize("topics_msg_list", [(("topic1", dummy_msg),)])
+def test_dispatcher_disconnect_wrong_topic_is_safe_noop(
+    bec_dispatcher_w_connector, qtbot, send_msg_event
+):
+    bec_dispatcher = bec_dispatcher_w_connector
+    cb1 = mock.Mock(spec=[])
+
+    bec_dispatcher.connect_slot(cb1, "topic1")
+    # disconnecting a topic the slot is NOT subscribed to must not release topic1
+    bec_dispatcher.disconnect_slot(cb1, "topic-wrong")
+    assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 1
+
+    send_msg_event.set()
+    qtbot.wait(10)
+    cb1.assert_called_once()
+
+    bec_dispatcher.disconnect_slot(cb1, "topic1")
+    assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 0
+
+
+@pytest.mark.parametrize("topics_msg_list", [(("topic1", dummy_msg), ("topic2", dummy_msg))])
+def test_dispatcher_disconnect_selects_wrapper_by_topic(
+    bec_dispatcher_w_connector, qtbot, send_msg_event
+):
+    # The same callback registered twice with different cb_info produces two wrappers;
+    # disconnecting a topic must release it from the wrapper that actually holds it,
+    # not silently no-op on the first wrapper that matches the callback.
+    bec_dispatcher = bec_dispatcher_w_connector
+    cb1 = mock.Mock(spec=[])
+
+    bec_dispatcher.connect_slot(cb1, "topic1", cb_info={"scan": "a"})
+    bec_dispatcher.connect_slot(cb1, "topic2", cb_info={"scan": "b"})
+    assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 2
+
+    bec_dispatcher.disconnect_slot(cb1, "topic2")
+    remaining = bec_dispatcher.client.connector._managed_connection._topics_cb
+    assert len(remaining) == 1
+    assert "topic1" in remaining
+
+    # release the remaining subscription explicitly instead of leaning on fixture teardown
+    bec_dispatcher.disconnect_slot(cb1, "topic1")
+    assert len(bec_dispatcher.client.connector._managed_connection._topics_cb) == 0
+
+    # unblock the fixture's message generator so connector.shutdown() can join
+    send_msg_event.set()
+    qtbot.wait(10)
+
+
 @pytest.mark.parametrize("topics_msg_list", [(("topic1", dummy_msg),)])
 def test_dispatcher_2_cb_same_topic(bec_dispatcher_w_connector, qtbot, send_msg_event):
     # test for BEC issue #276