From ec64ed9a44693ada9d3356e95a09eaf7852f424f Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Mon, 3 Aug 2026 11:26:41 +0200 Subject: [PATCH] fix(scan_control): restore last scan parameters without blocking the GUI --- .../control/scan_control/scan_control.py | 233 ++++++++-- tests/unit_tests/test_scan_control.py | 413 +++++++++++++++++- 2 files changed, 604 insertions(+), 42 deletions(-) diff --git a/bec_widgets/widgets/control/scan_control/scan_control.py b/bec_widgets/widgets/control/scan_control/scan_control.py index db92e19c..9cb37984 100644 --- a/bec_widgets/widgets/control/scan_control/scan_control.py +++ b/bec_widgets/widgets/control/scan_control/scan_control.py @@ -1,12 +1,15 @@ +import threading from collections import defaultdict -from types import NoneType, SimpleNamespace +from functools import partial +from types import NoneType from typing import Optional from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger +from bec_lib.scan_history import ScanHistory from bec_qthemes import material_icon from pydantic import BaseModel, Field -from qtpy.QtCore import QSignalBlocker, Qt, Signal -from qtpy.QtGui import QColor +from qtpy.QtCore import QSignalBlocker, Qt, QTimer, Signal from qtpy.QtWidgets import ( QApplication, QComboBox, @@ -23,7 +26,7 @@ from qtpy.QtWidgets import ( from bec_widgets.utils.bec_connector import ConnectionConfig from bec_widgets.utils.bec_widget import BECWidget -from bec_widgets.utils.colors import apply_theme, get_accent_colors +from bec_widgets.utils.colors import apply_theme from bec_widgets.utils.error_popups import SafeProperty, SafeSlot from bec_widgets.widgets.control.buttons.stop_button.stop_button import StopButton from bec_widgets.widgets.control.scan_control.scan_docstring import render_scan_tooltip_html @@ -33,6 +36,8 @@ from bec_widgets.widgets.control.scan_control.scan_info_dialog import ScanInfoDi from bec_widgets.widgets.control.scan_control.scan_selection_dialog import ScanSelectionDialog from bec_widgets.widgets.editors.scan_metadata.scan_metadata import ScanMetadata +logger = bec_logger.logger + class ScanParameterConfig(BaseModel): name: str @@ -56,11 +61,15 @@ class ScanControl(BECWidget, QWidget): ICON_NAME = "tune" ARG_BOX_POSITION: int = 2 SUPPORTED_SCAN_BASE_CLASSES = {"ScanBase", "SyncFlyScanBase", "AsyncFlyScanBase", "ScanBaseV4"} + RECENT_SCAN_HISTORY_COUNT = 50 + MAX_HISTORY_LOOKBACK = 500 + LAST_SCAN_FETCH_TIMEOUT_MS = 30_000 scan_started = Signal() scan_selected = Signal(str) device_selected = Signal(str) scan_args = Signal(list) + _last_scan_parameters_received = Signal(int, str, object) def __init__( self, @@ -110,6 +119,18 @@ class ScanControl(BECWidget, QWidget): self._hide_scan_selector_settings_button = False self._scan_info_adapter = ScanInfoAdapter() self._scan_info_dialog: ScanInfoDialog | None = None + self._last_scan_parameters_received.connect(self._apply_last_scan_parameters) + # Window-lookup results memoized per scan name, valid while the stream tip is + # unchanged. Fetch workers can overlap - the watchdog re-enables the button while a + # hung worker is still running - so the memo is guarded by its own lock. + self._last_scan_lookup_memo: dict[str, tuple[str | None, tuple[list, dict] | None]] = {} + self._last_scan_lookup_lock = threading.Lock() + # Generation counter to discard results of superseded or timed-out fetches + self._last_scan_fetch_generation = 0 + self._last_scan_fetch_watchdog = QTimer(self) + self._last_scan_fetch_watchdog.setSingleShot(True) + self._last_scan_fetch_watchdog.setInterval(self.LAST_SCAN_FETCH_TIMEOUT_MS) + self._last_scan_fetch_watchdog.timeout.connect(self._on_last_scan_parameters_timeout) # Create and set main layout self._init_UI() @@ -118,14 +139,6 @@ class ScanControl(BECWidget, QWidget): """ Initializes the UI of the scan control widget. Create the top box for scan selection and populate scans to main combobox. """ - palette = get_accent_colors() - if palette is None: - palette = SimpleNamespace( - default=QColor("blue"), - success=QColor("green"), - warning=QColor("orange"), - emergency=QColor("red"), - ) # Scan selection box self.scan_selection_group = QWidget(self) QVBoxLayout(self.scan_selection_group) @@ -369,34 +382,192 @@ class ScanControl(BECWidget, QWidget): """ Requests the last executed scan parameters from BEC and restores them to the scan control widget. """ + if not self.last_scan_button.isEnabled(): + return current_scan = self.comboBox_scan_selection.currentText() - history = ( - self.client.connector.xread( - MessageEndpoints.scan_history(), from_start=True, user_id=self.object_name + if not current_scan: + # e.g. an empty selector after a filter change - nothing to restore + return + self.last_scan_button.setEnabled(False) + self._last_scan_fetch_generation += 1 + generation = self._last_scan_fetch_generation + self._last_scan_fetch_watchdog.start() + try: + # ``completed`` is success-only; ``failed`` carries a traceback string, which + # _on_last_scan_parameters_finished does not take, so bind the generation instead. + self.submit_task( + self._fetch_last_executed_scan_parameters, + generation, + current_scan, + on_complete=partial(self._on_last_scan_parameters_finished, generation), + on_failed=lambda _msg, gen=generation: self._on_last_scan_parameters_finished(gen), ) - or [] - ) + except Exception: + self._on_last_scan_parameters_finished(generation) + raise + def _fetch_last_executed_scan_parameters(self, generation: int, scan_name: str): + """Fetch the latest parameters for ``scan_name`` without touching the Qt UI.""" + try: + parameters = self._lookup_last_scan_parameters(scan_name) + except Exception: + logger.exception(f"Failed to fetch parameters for scan {scan_name}") + return + if parameters is None: + return + try: + self._last_scan_parameters_received.emit(generation, scan_name, parameters) + except RuntimeError: + # the widget was deleted while the fetch was in flight - nothing to deliver to + logger.debug(f"ScanControl deleted before parameters for {scan_name} arrived") + + def _lookup_last_scan_parameters(self, scan_name: str) -> tuple[list, dict] | None: + """ + Find the newest parameters for ``scan_name`` in the client-side scan-history cache or + in a strictly bounded window of the scan-history stream. + + There is deliberately no full-stream path: decoding an entry allocates hundreds of + GC-tracked objects, and the resulting stop-the-world collections stall the Qt event + loop from any thread. + """ + endpoint = MessageEndpoints.scan_history() + # Single one-entry probe of the stream tip. The stream is append-only, so the tip + # both proves the in-memory cache is caught up (it lags when the newest scan's file + # is not readable yet, e.g. over NFS) and keys the memo: repeat lookups with an + # unchanged tip are answered for the cost of this one decode. + tip = self._read_history_window(endpoint, 1) + tip_id = getattr(tip[0].get("data"), "scan_id", None) if tip else None + + parameters = self._find_parameters_in_history_cache(scan_name, tip_id) + if parameters is not None: + return parameters + + # The lock is held only around the dict access, never across a redis read: a hung + # fetch must not block the worker that the watchdog let the user start. + with self._last_scan_lookup_lock: + memo = self._last_scan_lookup_memo.get(scan_name) + if memo is not None and memo[0] == tip_id: + logger.debug(f"Scan history unchanged; reusing memoized lookup for {scan_name}") + return memo[1] + + parameters = self._search_history_windows(endpoint, scan_name) + with self._last_scan_lookup_lock: + # Overlapping workers race to write here; the loser stores a result for an + # older tip, which the tip comparison above discards on the next lookup. + self._last_scan_lookup_memo[scan_name] = (tip_id, parameters) + return parameters + + def _search_history_windows(self, endpoint, scan_name: str) -> tuple[list, dict] | None: + """Search the recent, then the deep, stream window for ``scan_name``.""" + history = self._read_history_window(endpoint, self.RECENT_SCAN_HISTORY_COUNT) + parameters = self._find_last_scan_parameters(scan_name, history) + if parameters is not None or len(history) < self.RECENT_SCAN_HISTORY_COUNT: + # fewer entries than requested means the whole stream was already searched + return parameters + if self.MAX_HISTORY_LOOKBACK <= self.RECENT_SCAN_HISTORY_COUNT: + return None + + history = self._read_history_window(endpoint, self.MAX_HISTORY_LOOKBACK) + parameters = self._find_last_scan_parameters(scan_name, history) + if parameters is None and len(history) >= self.MAX_HISTORY_LOOKBACK: + logger.warning( + f"No execution of {scan_name} found within the last " + f"{self.MAX_HISTORY_LOOKBACK} scans; older history is not searched." + ) + return parameters + + def _read_history_window(self, endpoint, count: int) -> list[dict]: + """Read at most ``count`` newest scan-history entries, oldest-first.""" + history = self.client.connector.get_last(endpoint, count=count) + if isinstance(history, dict): + # get_last returns a single message dict instead of a list for count == 1 + history = [history] + return history or [] + + def _find_parameters_in_history_cache( + self, scan_name: str, tip_id: str | None + ) -> tuple[list, dict] | None: + """Search the in-memory scan history (newest first) for ``scan_name``. + + The cache is only trusted when its newest entry matches the stream tip: it skips + scans whose file is not readable yet, so a lagging cache could otherwise return an + older execution than the stream holds. + """ + history = getattr(self.client, "history", None) + if not isinstance(history, ScanHistory): + # not available before the client services are started (e.g. in Qt Designer) + return None + length = len(history) + if length == 0: + return None + newest = getattr(history[length - 1], "_msg", None) + if newest is None or newest.scan_id != tip_id: + return None + # iterate by index from the end instead of slicing to avoid copying the whole cache + for index in range(length - 1, -1, -1): + msg = getattr(history[index], "_msg", None) + if msg is not None and msg.scan_name == scan_name: + return self._parameters_from_request_inputs(msg.request_inputs) + return None + + @staticmethod + def _parameters_from_request_inputs(request_inputs: dict | None) -> tuple[list, dict]: + """Split ``request_inputs`` into the argument bundle and merged keyword arguments.""" + ri = request_inputs or {} + return ri.get("arg_bundle", []), {**ri.get("inputs", {}), **ri.get("kwargs", {})} + + @staticmethod + def _find_last_scan_parameters(scan_name: str, history: list[dict]) -> tuple[list, dict] | None: + """Return the newest matching argument and keyword bundles from ``history``.""" for scan in reversed(history): scan_data = scan.get("data") if not scan_data: continue - if scan_data.scan_name != current_scan: + if scan_data.scan_name != scan_name: continue - ri = getattr(scan_data, "request_inputs", {}) or {} - args_list = ri.get("arg_bundle", []) - if args_list and self.arg_box: - self.arg_box.set_parameters(args_list) + return ScanControl._parameters_from_request_inputs( + getattr(scan_data, "request_inputs", None) + ) + return None - inputs = ri.get("inputs", {}) - kwargs = ri.get("kwargs", {}) - merged = {**inputs, **kwargs} - if merged and self.kwarg_boxes: - for box in self.kwarg_boxes: - box.set_parameters(merged) - break + @SafeSlot(int, str, object) + def _apply_last_scan_parameters( + self, generation: int, scan_name: str, parameters: tuple[list, dict] + ): + """Apply asynchronously fetched parameters on the GUI thread.""" + if generation != self._last_scan_fetch_generation: + # result of a timed-out or superseded fetch + return + if self.comboBox_scan_selection.currentText() != scan_name: + logger.debug(f"Discarding fetched parameters for {scan_name}: scan selection changed") + return + + args_list, kwargs = parameters + if args_list and self.arg_box: + self.arg_box.set_parameters(args_list) + + if kwargs and self.kwarg_boxes: + for box in self.kwarg_boxes: + box.set_parameters(kwargs) + + @SafeSlot() + def _on_last_scan_parameters_finished(self, generation: int | None = None): + """Re-enable parameter restoration after the background request finishes.""" + if generation is not None and generation != self._last_scan_fetch_generation: + # a stale fetch finished after a timeout or a newer request; leave the UI state alone + return + self._last_scan_fetch_watchdog.stop() + self.last_scan_button.setEnabled(True) + + @SafeSlot() + def _on_last_scan_parameters_timeout(self): + """Recover the UI if the background fetch hangs (e.g. unreachable Redis).""" + logger.warning("Timed out while fetching the last executed scan parameters") + # invalidate the in-flight fetch so a late result is not applied + self._last_scan_fetch_generation += 1 + self.last_scan_button.setEnabled(True) @SafeProperty(str) def current_scan(self): @@ -713,6 +884,10 @@ class ScanControl(BECWidget, QWidget): def cleanup(self): """Cleanup the scan control widget.""" + # Invalidate any in-flight fetch so a late result is dropped instead of being + # applied to (or logged against) a widget that is going away. + self._last_scan_fetch_generation += 1 + self._last_scan_fetch_watchdog.stop() if self._scan_info_dialog is not None: self._scan_info_dialog.close() self._scan_info_dialog.deleteLater() diff --git a/tests/unit_tests/test_scan_control.py b/tests/unit_tests/test_scan_control.py index c3c64a6f..707b4968 100644 --- a/tests/unit_tests/test_scan_control.py +++ b/tests/unit_tests/test_scan_control.py @@ -1,10 +1,12 @@ # pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring +from threading import Event, get_ident from types import SimpleNamespace -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, call, patch import pytest from bec_lib.endpoints import MessageEndpoints from bec_lib.messages import AvailableResourceMessage, ScanHistoryMessage +from bec_lib.scan_history import ScanHistory from qtpy.QtCore import QModelIndex, QPoint, Qt from qtpy.QtWidgets import QCheckBox, QDialog, QStyle @@ -12,6 +14,7 @@ from bec_widgets.utils.forms_from_types.items import StrFormItem from bec_widgets.utils.widget_io import WidgetIO from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox from bec_widgets.widgets.control.scan_control import ScanControl +from bec_widgets.widgets.control.scan_control import scan_control as scan_control_module from bec_widgets.widgets.control.scan_control.scan_control import ScanControlConfig from bec_widgets.widgets.control.scan_control.scan_info_adapter import ScanInfoAdapter from bec_widgets.widgets.control.scan_control.scan_selection_dialog import ScanSelectionDialog @@ -1184,32 +1187,43 @@ def test_changing_scans_remember_parameters(scan_control, mocked_client): def test_scan_selection_does_not_fetch_last_scan_parameters( scan_control, mocked_client, monkeypatch ): - xread = MagicMock(wraps=mocked_client.connector.xread) - monkeypatch.setattr(mocked_client.connector, "xread", xread) + get_last = MagicMock(wraps=mocked_client.connector.get_last) + xrange = MagicMock(wraps=mocked_client.connector.xrange) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + monkeypatch.setattr(mocked_client.connector, "xrange", xrange) scan_control.comboBox_scan_selection.setCurrentText("line_scan") assert scan_control.comboBox_scan_selection.currentText() == "line_scan" scan_control.comboBox_scan_selection.setCurrentText("grid_scan") - xread.assert_not_called() + get_last.assert_not_called() + xrange.assert_not_called() def test_restore_last_scan_parameters_button_fetches_on_demand( - scan_control, mocked_client, monkeypatch + scan_control, mocked_client, monkeypatch, qtbot ): - xread = MagicMock(wraps=mocked_client.connector.xread) - monkeypatch.setattr(mocked_client.connector, "xread", xread) + get_last = MagicMock(wraps=mocked_client.connector.get_last) + xrange = MagicMock(wraps=mocked_client.connector.xrange) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + monkeypatch.setattr(mocked_client.connector, "xrange", xrange) scan_control.comboBox_scan_selection.setCurrentText("grid_scan") scan_control.comboBox_scan_selection.setCurrentText("line_scan") - xread.assert_not_called() + get_last.assert_not_called() + xrange.assert_not_called() scan_control.last_scan_button.click() - xread.assert_called_once_with( - MessageEndpoints.scan_history(), from_start=True, user_id=scan_control.object_name - ) + qtbot.waitUntil(lambda: get_last.call_count == 2) + # a one-entry tip probe (memo validity check), then the recent window + assert get_last.call_args_list == [ + call(MessageEndpoints.scan_history(), count=1), + call(MessageEndpoints.scan_history(), count=scan_control.RECENT_SCAN_HISTORY_COUNT), + ] + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + xrange.assert_not_called() args, kwargs = scan_control.get_scan_parameters(bec_object=False) assert args == ["samx", 0.0, 2.0] assert kwargs["steps"] == 10 @@ -1217,11 +1231,345 @@ def test_restore_last_scan_parameters_button_fetches_on_demand( assert kwargs["exp_time"] == 2 -def test_get_scan_parameters_from_redis(scan_control): +def test_restore_last_scan_parameters_does_not_block_gui( + scan_control, mocked_client, monkeypatch, qtbot +): + original_get_last = mocked_client.connector.get_last + fetch_started = Event() + allow_fetch_to_finish = Event() + fetch_thread_ids = [] + apply_thread_ids = [] + + original_set_parameters = scan_control.arg_box.set_parameters + + def tracking_set_parameters(*args, **kwargs): + apply_thread_ids.append(get_ident()) + return original_set_parameters(*args, **kwargs) + + def blocking_get_last(*args, **kwargs): + fetch_thread_ids.append(get_ident()) + fetch_started.set() + allow_fetch_to_finish.wait(timeout=5) + return original_get_last(*args, **kwargs) + + monkeypatch.setattr(mocked_client.connector, "get_last", blocking_get_last) + monkeypatch.setattr(scan_control.arg_box, "set_parameters", tracking_set_parameters) + gui_thread_id = get_ident() + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + try: + qtbot.waitUntil(fetch_started.is_set) + assert len(fetch_thread_ids) == 1 + assert fetch_thread_ids[0] != gui_thread_id + assert not scan_control.last_scan_button.isEnabled() + finally: + allow_fetch_to_finish.set() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + assert apply_thread_ids == [gui_thread_id] + args, kwargs = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert kwargs["steps"] == 10 + + +def test_restore_last_scan_parameters_reenables_button_after_fetch_failure( + scan_control, mocked_client, monkeypatch, qtbot +): + get_last = MagicMock(side_effect=RuntimeError("history unavailable")) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + + scan_control.last_scan_button.click() + + qtbot.waitUntil(lambda: get_last.call_count == 1) + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + + +def test_restore_last_scan_parameters_reenables_button_when_worker_task_raises( + scan_control, monkeypatch, qtbot +): + # bypass the fetch's own error handling: the worker emits ``failed`` instead of + # ``completed``, so the button must recover from that signal, not from the watchdog + monkeypatch.setattr( + scan_control._last_scan_fetch_watchdog, "start", MagicMock(name="watchdog_disabled") + ) + monkeypatch.setattr( + scan_control, + "_fetch_last_executed_scan_parameters", + MagicMock(side_effect=RuntimeError("worker exploded")), + ) + + scan_control.last_scan_button.click() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + + +def test_restore_last_scan_parameters_escalates_to_bounded_history_window( + scan_control, mocked_client, monkeypatch, qtbot +): + recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"}) + # a full window means older entries may exist, so the deeper - but still capped - read runs + full_window = [{"data": recent_scan}] * scan_control.RECENT_SCAN_HISTORY_COUNT + # responses for: tip probe, recent window, deep window + get_last = MagicMock( + side_effect=[[{"data": recent_scan}], full_window, [{"data": scan_history}]] + ) + xrange = MagicMock(wraps=mocked_client.connector.xrange) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + monkeypatch.setattr(mocked_client.connector, "xrange", xrange) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + qtbot.waitUntil(lambda: get_last.call_count == 3) + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + assert get_last.call_args_list == [ + call(MessageEndpoints.scan_history(), count=1), + call(MessageEndpoints.scan_history(), count=scan_control.RECENT_SCAN_HISTORY_COUNT), + call(MessageEndpoints.scan_history(), count=scan_control.MAX_HISTORY_LOOKBACK), + ] + # the unbounded full-stream read is gone: it stalled the event loop from any thread + xrange.assert_not_called() + args, kwargs = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert kwargs["steps"] == 10 + + +def test_restore_last_scan_parameters_never_reads_unbounded_stream( + scan_control, mocked_client, monkeypatch, qtbot +): + recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"}) + # every window comes back full, so the lookup can never conclude the stream is exhausted + get_last = MagicMock( + side_effect=lambda _endpoint, count: [{"data": recent_scan}] * count # noqa: ARG005 + ) + xrange = MagicMock(wraps=mocked_client.connector.xrange) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + monkeypatch.setattr(mocked_client.connector, "xrange", xrange) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + xrange.assert_not_called() + assert ( + max(call_args.kwargs["count"] for call_args in get_last.call_args_list) + == scan_control.MAX_HISTORY_LOOKBACK + ) + + +def test_restore_last_scan_parameters_logs_when_lookback_exhausted( + scan_control, mocked_client, monkeypatch, qtbot +): + recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"}) + get_last = MagicMock(side_effect=lambda _endpoint, count: [{"data": recent_scan}] * count) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + # bec_logger is loguru-based, so caplog does not see it + warning = MagicMock() + monkeypatch.setattr(scan_control_module.logger, "warning", warning) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + + assert any("older history is not searched" in c.args[0] for c in warning.call_args_list) + + +def test_restore_last_scan_parameters_skips_deeper_read_when_stream_exhausted( + scan_control, mocked_client, monkeypatch, qtbot +): + recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"}) + # a partial window proves the whole stream was searched; no deeper read is allowed + get_last = MagicMock(return_value=[{"data": recent_scan}]) + xrange = MagicMock(wraps=mocked_client.connector.xrange) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + monkeypatch.setattr(mocked_client.connector, "xrange", xrange) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + qtbot.waitUntil(lambda: get_last.call_count == 2) + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + assert get_last.call_args_list == [ + call(MessageEndpoints.scan_history(), count=1), + call(MessageEndpoints.scan_history(), count=scan_control.RECENT_SCAN_HISTORY_COUNT), + ] + xrange.assert_not_called() + + +def test_restore_last_scan_parameters_memoizes_missing_scan_lookups( + scan_control, mocked_client, monkeypatch, qtbot +): + recent_scan = scan_history.model_copy(update={"scan_name": "grid_scan"}) + get_last = MagicMock(side_effect=lambda _endpoint, count: [{"data": recent_scan}] * count) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + calls_after_first = get_last.call_count # tip probe + recent window + deep window + + scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + + # the repeat click for a never-measured scan costs one tip probe, no window reads + assert get_last.call_count == calls_after_first + 1 + assert get_last.call_args_list[-1] == call(MessageEndpoints.scan_history(), count=1) + + +def test_restore_last_scan_parameters_memo_reapplies_and_invalidates( + scan_control, mocked_client, monkeypatch, qtbot +): + get_last = MagicMock(wraps=mocked_client.connector.get_last) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + + scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + calls_after_first = get_last.call_count # tip probe + recent window (hit) + + # user edits the form; restoring again is answered from the memo with one probe read + scan_control.arg_box.set_parameters(["samx", 1.0, 5.0]) + scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + args, _ = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert get_last.call_count == calls_after_first + 1 + assert get_last.call_args_list[-1] == call(MessageEndpoints.scan_history(), count=1) + + # a new history entry moves the stream tip and invalidates the memo + new_msg = scan_history.model_copy(update={"scan_name": "grid_scan", "scan_id": "memo_tip"}) + mocked_client.connector.xadd(topic=MessageEndpoints.scan_history(), msg_dict={"data": new_msg}) + scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + # tip probe plus a fresh recent-window read + assert get_last.call_count == calls_after_first + 3 + + +def test_restore_last_scan_parameters_handles_single_entry_window( + scan_control, mocked_client, monkeypatch, qtbot +): + # get_last returns a bare dict (not a list) for count == 1 + monkeypatch.setattr(scan_control, "RECENT_SCAN_HISTORY_COUNT", 1) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + args, kwargs = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert kwargs["steps"] == 10 + + +def test_restore_last_scan_parameters_uses_history_cache( + scan_control, mocked_client, monkeypatch, qtbot +): + history = ScanHistory(mocked_client, load_threaded=False) + history._scan_data[scan_history.scan_id] = scan_history + history._scan_ids.append(scan_history.scan_id) + monkeypatch.setattr(mocked_client, "history", history, raising=False) + get_last = MagicMock(wraps=mocked_client.connector.get_last) + monkeypatch.setattr(mocked_client.connector, "get_last", get_last) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + # only the one-entry tip probe validating the cache; no window reads + get_last.assert_called_once_with(MessageEndpoints.scan_history(), count=1) + args, kwargs = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert kwargs["steps"] == 10 + + +def test_restore_last_scan_parameters_skips_stale_history_cache( + scan_control, mocked_client, monkeypatch, qtbot +): + """A cache whose newest entry lags the stream tip (e.g. the newest scan's file is not + readable yet) must not answer the lookup with an older execution.""" + stale = scan_history.model_copy( + update={"scan_id": "stale_entry", "request_inputs": {"arg_bundle": ["samx", 5.0, 9.0]}} + ) + history = ScanHistory(mocked_client, load_threaded=False) + history._scan_data.clear() + history._scan_ids.clear() + history._scan_data[stale.scan_id] = stale + history._scan_ids.append(stale.scan_id) + monkeypatch.setattr(mocked_client, "history", history, raising=False) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + # the stream entry (steps=10), not the stale cache entry, wins + args, kwargs = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert kwargs["steps"] == 10 + + +def test_restore_last_scan_parameters_discards_result_on_scan_switch( + scan_control, mocked_client, monkeypatch, qtbot +): + original_get_last = mocked_client.connector.get_last + fetch_started = Event() + allow_fetch_to_finish = Event() + + def blocking_get_last(*args, **kwargs): + fetch_started.set() + allow_fetch_to_finish.wait(timeout=5) + return original_get_last(*args, **kwargs) + + monkeypatch.setattr(mocked_client.connector, "get_last", blocking_get_last) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + qtbot.waitUntil(fetch_started.is_set) + scan_control.comboBox_scan_selection.setCurrentText("grid_scan") + allow_fetch_to_finish.set() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + # the fetched line_scan parameters must not leak into the grid_scan form + args, _ = scan_control.get_scan_parameters(bec_object=False) + assert args != ["samx", 0.0, 2.0] + + +def test_restore_last_scan_parameters_watchdog_recovers_from_hanging_fetch( + scan_control, mocked_client, monkeypatch, qtbot +): + original_get_last = mocked_client.connector.get_last + fetch_started = Event() + allow_fetch_to_finish = Event() + + def blocking_get_last(*args, **kwargs): + fetch_started.set() + allow_fetch_to_finish.wait(timeout=5) + return original_get_last(*args, **kwargs) + + monkeypatch.setattr(mocked_client.connector, "get_last", blocking_get_last) + scan_control._last_scan_fetch_watchdog.setInterval(50) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + original_args, _ = scan_control.get_scan_parameters(bec_object=False) + scan_control.last_scan_button.click() + qtbot.waitUntil(fetch_started.is_set) + + # the watchdog re-enables the button while the fetch is still hanging + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + allow_fetch_to_finish.set() + + # the late result must be discarded once the watchdog has given up on the fetch + qtbot.wait(200) + args, _ = scan_control.get_scan_parameters(bec_object=False) + assert args == original_args + + +def test_get_scan_parameters_from_redis(scan_control, qtbot): scan_name = "line_scan" scan_control.comboBox_scan_selection.setCurrentText(scan_name) scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) args, kwargs = scan_control.get_scan_parameters(bec_object=False) @@ -1301,7 +1649,7 @@ def test_scan_metadata_is_passed_to_scan_function(scan_control: ScanControl): scans.grid_scan.assert_called_once_with(metadata=TEST_MD) -def test_restore_parameters_with_fewer_arg_bundles(scan_control): +def test_restore_parameters_with_fewer_arg_bundles(scan_control, qtbot): """ Ensure that when more argument bundles are present than exist in the stored history, restoring parameters regenerates the arg box to the @@ -1318,6 +1666,7 @@ def test_restore_parameters_with_fewer_arg_bundles(scan_control): # Trigger restore of parameters from history scan_control.last_scan_button.click() + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) # After restore, arg_box should have only one bundle (the history size) assert scan_control.arg_box.count_arg_rows() == 1 @@ -1420,3 +1769,41 @@ def test_allowed_scans_reordered_full_list_keeps_order(scan_control): # the same list in the supported order clears the filter scan_control.allowed_scans = supported assert scan_control.config.allowed_scans is None + + +def test_restore_last_scan_parameters_memo_survives_overlapping_workers( + scan_control, mocked_client, monkeypatch, qtbot +): + """The watchdog re-enables the button while a slow worker is still running, so two + fetch workers can touch the memo at once. Both must finish and leave a usable memo.""" + original_get_last = mocked_client.connector.get_last + first_fetch_running = Event() + release_first_fetch = Event() + worker_threads = set() + + def slow_get_last(*args, **kwargs): + worker_threads.add(get_ident()) + if not first_fetch_running.is_set(): + first_fetch_running.set() + release_first_fetch.wait(timeout=5) + return original_get_last(*args, **kwargs) + + monkeypatch.setattr(mocked_client.connector, "get_last", slow_get_last) + scan_control._last_scan_fetch_watchdog.setInterval(50) + + scan_control.comboBox_scan_selection.setCurrentText("line_scan") + scan_control.last_scan_button.click() + qtbot.waitUntil(first_fetch_running.is_set) + + # the watchdog gives the button back while the first worker is still blocked + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + scan_control.last_scan_button.click() + qtbot.waitUntil(lambda: len(worker_threads) == 2) + release_first_fetch.set() + + qtbot.waitUntil(scan_control.last_scan_button.isEnabled) + qtbot.waitUntil(lambda: "line_scan" in scan_control._last_scan_lookup_memo) + assert get_ident() not in worker_threads + args, kwargs = scan_control.get_scan_parameters(bec_object=False) + assert args == ["samx", 0.0, 2.0] + assert kwargs["steps"] == 10