csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py

@@ -70,11 +70,14 @@ def suppress_mca_callbacks(mcs_card: MCSCard, restore_after_timeout: None | floa
     """
     with mcs_card._rlock:
         mcs_card._omit_mca_callbacks.set()  # pylint: disable=protected-access
-        try:
-            yield
-        finally:
-            if restore_after_timeout is not None:
-                time.sleep(restore_after_timeout)
+    try:
+        # MCA callbacks must be able to acquire the data lock while suppression
+        # is active so they are discarded instead of queued until afterward.
+        yield
+    finally:
+        if restore_after_timeout is not None:
+            time.sleep(restore_after_timeout)
+            with mcs_card._rlock:
                 mcs_card._omit_mca_callbacks.clear()  # pylint: disable=protected-access
 
 
@@ -467,7 +470,16 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard):
                         logger.info(
                             f"Software triggered scan: {self._current_data_index}/{self.scan_parameters.num_points} points received."
                         )
-                        if self._current_data_index == self.scan_parameters.num_points:
+                        if self._current_data_index > self.scan_parameters.num_points:
+                            exception = RuntimeError(
+                                f"MCS card {self.name} emitted {self._current_data_index} "
+                                f"data points, but the scan requested "
+                                f"{self.scan_parameters.num_points}."
+                            )
+                            logger.error(str(exception))
+                            for callback in self._scan_done_callbacks:
+                                callback(exception=exception)
+                        elif self._current_data_index == self.scan_parameters.num_points:
                             for callback in self._scan_done_callbacks:
                                 callback(exception=None)
                     else:

tests/tests_devices/test_mcs_card.py

@@ -24,7 +24,7 @@ from csaxs_bec.devices.epics.mcs_card.mcs_card import (
     READMODE,
     MCSCard,
 )
-from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS
+from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS, suppress_mca_callbacks
 from csaxs_bec.devices.utils.utils import fetch_scan_info
 
 
@@ -167,9 +167,10 @@ def test_mcs_card_csaxs_complete_and_stop(mock_mcs_csaxs: MCSCardCSAXS):
     """
     Test complete method of MCSCarcCSAXS.
 
-    Two use cases:
+    Three use cases:
     I. Acquisition is stopped externally
     II. Acquisition completes normally
+    III. Acquisition emits more points than requested
     """
     mcs = mock_mcs_csaxs
     mcs.scan_parameters = fetch_scan_info(mcs.scan_info)
@@ -219,6 +220,38 @@ def test_mcs_card_csaxs_complete_and_stop(mock_mcs_csaxs: MCSCardCSAXS):
     mcs._start_monitor_async_data_emission.wait(2)
     assert not mcs._start_monitor_async_data_emission.is_set()
 
+    ##############################
+    # III. Extra data fails fast #
+    ##############################
+    mcs._current_data_index = 11
+    st = mcs.complete()
+    with pytest.raises(RuntimeError, match="emitted 11 data points, but the scan requested 10"):
+        st.wait(timeout=1)
+
+
+def test_suppress_mca_callbacks_discards_callbacks_without_blocking(mock_mcs_csaxs: MCSCardCSAXS):
+    """A complete MCA callback batch is discarded while suppression is active."""
+    mcs = mock_mcs_csaxs
+    callback_finished = threading.Event()
+
+    def emit_callbacks():
+        for index in range(mcs.NUM_MCA_CHANNELS):
+            counter = getattr(mcs.counters, f"mca{index + 1}")
+            mcs._on_counter_update(index, obj=counter, timestamp=1.0)
+        callback_finished.set()
+
+    worker = threading.Thread(target=emit_callbacks)
+    with mock.patch.object(mcs.mca, "put") as mca_put:
+        with suppress_mca_callbacks(mcs, restore_after_timeout=0):
+            worker.start()
+            assert callback_finished.wait(timeout=1)
+
+        worker.join(timeout=1)
+        assert not worker.is_alive()
+        assert mcs._current_data == {}
+        assert mcs._current_data_index == 0
+        mca_put.assert_not_called()
+
 
 def test_mcs_on_stop(mock_mcs_csaxs: MCSCardCSAXS):
     """Test that on stop sets the omit_mca_callbacks flag. Also test that on stage clears the omit_mca_callbacks flag."""