fix(waveform): release the exact async stream endpoint when the last curve is removed

bec_widgets/widgets/plots/waveform/waveform.py

@@ -1508,15 +1508,13 @@ class Waveform(PlotBase):
 
     def _curve_clean_up(self, curve: Curve):
         """
-        Clean up the curve by disconnecting the async update signal (even for sync curves).
+        Clean up the curve, releasing its async stream subscription when it was the last
+        curve fed by that stream (a safe no-op for sync and DAP curves).
 
         Args:
             curve(Curve): The curve to clean up.
         """
-        self.bec_dispatcher.disconnect_slot(
-            self.on_async_readback,
-            MessageEndpoints.device_async_readback(self.scan_id, curve.name()),
-        )
+        self._release_async_stream(curve)
         curve.rpc_register.remove_rpc(curve)
 
         # Remove itself from the DAP summary only for side panels
@@ -1535,6 +1533,29 @@ class Waveform(PlotBase):
                 self.plot_item.removeItem(c)
                 self._curve_clean_up(c)
 
+    def _release_async_stream(self, curve: Curve) -> None:
+        """
+        Drop the curve's signal from its shared async stream; unsubscribe the exact
+        endpoint recorded at setup time once no curve uses the stream anymore.
+
+        Disconnecting with a re-derived endpoint instead would silently miss (the
+        subscription is keyed by device name and endpoint family), leaving the stream
+        subscribed until widget teardown.
+        """
+        signal_config = getattr(curve.config, "signal", None)
+        device = getattr(signal_config, "device", None)
+        if device is None:
+            return
+        signal = signal_config.signal
+        for stream_key, stream in list(self._async_streams_setup.items()):
+            if stream_key[0] != device or signal not in stream["signals"]:
+                continue
+            stream["signals"].remove(signal)
+            if not stream["signals"]:
+                self.bec_dispatcher.disconnect_slot(self.on_async_readback, stream["endpoint"])
+                self._async_streams_setup.pop(stream_key, None)
+            return
+
     def _check_curve_id(self, curve_id: str) -> bool:
         """
         Check if a curve ID exists in the plot widget.
@@ -1794,9 +1815,9 @@ class Waveform(PlotBase):
         shared = self._async_streams_setup.get(stream_key)
         if shared is not None:
             # Another curve already subscribed to this exact stream this scan.
-            shared.append(signal)
+            shared["signals"].append(signal)
             logger.info(
-                f"Async signals {shared} share a single subscription on endpoint "
+                f"Async signals {shared['signals']} share a single subscription on endpoint "
                 f"'{endpoint_str}'; reusing it instead of subscribing again."
             )
             return
@@ -1811,7 +1832,9 @@ class Waveform(PlotBase):
         #         "waveform widget will be removed in a future release."
         #     )
 
-        self._async_streams_setup[stream_key] = [signal]
+        # Record the exact endpoint next to the subscribers so _curve_clean_up can
+        # release precisely what was registered here.
+        self._async_streams_setup[stream_key] = {"signals": [signal], "endpoint": new_endpoint}
         self.bec_dispatcher.disconnect_slot(self.on_async_readback, old_endpoint)
         self.bec_dispatcher.connect_slot(
             self.on_async_readback, new_endpoint, from_start=True, cb_info={"scan_id": self.scan_id}

tests/unit_tests/test_waveform.py

@@ -913,6 +913,45 @@ def test_setup_async_curve(qtbot, mocked_client, monkeypatch):
     connect_spy.assert_called_once()
 
 
+def test_curve_clean_up_releases_exact_async_stream(qtbot, mocked_client, monkeypatch):
+    """
+    Removing the last curve of an async stream must unsubscribe the exact endpoint
+    recorded at setup time; while other curves still share the stream, the
+    subscription stays untouched.
+    """
+    wf = create_widget(qtbot, Waveform, client=mocked_client)
+    wf.old_scan_id = "111"
+    wf.scan_id = "222"
+
+    c = wf.plot(arg1="async_device", label="async_device-async_device")
+    wf._async_curves = [c]
+
+    disconnect_spy = MagicMock()
+    monkeypatch.setattr(wf.bec_dispatcher, "disconnect_slot", disconnect_spy)
+
+    wf._async_streams_setup = {}
+    wf._setup_async_curve(c)
+    setup_disconnects = disconnect_spy.call_count  # old-scan endpoint disconnect
+
+    stream_key, stream = next(iter(wf._async_streams_setup.items()))
+    recorded_endpoint = stream["endpoint"]
+    # simulate a second curve sharing the stream
+    stream["signals"].append("second_signal")
+
+    # stream still shared -> nothing released
+    wf._curve_clean_up(c)
+    assert disconnect_spy.call_count == setup_disconnects
+    assert stream_key in wf._async_streams_setup
+
+    # last user gone -> the EXACT recorded endpoint is released
+    stream["signals"].clear()
+    stream["signals"].append(c.config.signal.signal)
+    wf._curve_clean_up(c)
+    assert disconnect_spy.call_count == setup_disconnects + 1
+    assert disconnect_spy.call_args[0][1] is recorded_endpoint
+    assert stream_key not in wf._async_streams_setup
+
+
 def test_on_async_readback_add_update(qtbot, mocked_client):
     """
     Test that on_async_readback extends or replaces async data depending on metadata instruction.