diff --git a/bec_widgets/utils/bec_connector.py b/bec_widgets/utils/bec_connector.py index eb85b23d..e4f71bc1 100644 --- a/bec_widgets/utils/bec_connector.py +++ b/bec_widgets/utils/bec_connector.py @@ -51,11 +51,16 @@ class ConnectionConfig(BaseModel): class WorkerSignals(QObject): progress = Signal(dict) completed = Signal() + failed = Signal(str) class Worker(QRunnable): """ Worker class to run a function in a separate thread. + + On success, ``signals.completed`` is emitted. If the function raises, + the exception is logged and ``signals.failed`` is emitted with the + formatted traceback instead; it never propagates into the thread pool. """ def __init__(self, func, *args, **kwargs): @@ -69,8 +74,29 @@ class Worker(QRunnable): """ Run the specified function in the thread. """ - self.func(*self.args, **self.kwargs) - self.signals.completed.emit() + try: + self.func(*self.args, **self.kwargs) + except Exception: + error_msg = traceback.format_exc() + logger.error( + f"Worker task {getattr(self.func, '__qualname__', self.func)} failed:\n{error_msg}" + ) + self._emit_outcome(self.signals.failed, error_msg) + else: + self._emit_outcome(self.signals.completed) + + @staticmethod + def _emit_outcome(signal, *args): + """Emit the outcome signal, tolerating a signal source deleted mid-task. + + During application shutdown the Qt side of ``WorkerSignals`` can be destroyed + while the task is still running; the late emit must not escape ``run`` as an + unhandled exception in the thread pool. + """ + try: + signal.emit(*args) + except RuntimeError: + logger.debug("Worker outcome signal emitted after its source was deleted") class BECConnector: @@ -224,8 +250,9 @@ class BECConnector: return None connector_parent = self._get_rpc_parent_ancestor() return connector_parent.gui_id if connector_parent else None - except: - logger.error(f"Error getting parent_id for {self.__class__.__name__}") + except Exception as e: + logger.error(f"Error getting parent_id for {self.__class__.__name__}: {e}") + return None def _get_rpc_parent_ancestor(self) -> BECConnector | None: """ @@ -375,8 +402,16 @@ class BECConnector: worker.signals.completed.connect(on_complete) # Keep a reference to the worker so it is not garbage collected. self._workers.append(worker) - # When the worker is done, remove it from our list. - worker.signals.completed.connect(lambda: self._workers.remove(worker)) + + # When the worker is done (success or failure), remove it from our list. + def _discard_worker(*_): + try: + self._workers.remove(worker) + except ValueError: + pass + + worker.signals.completed.connect(_discard_worker) + worker.signals.failed.connect(_discard_worker) self._thread_pool.start(worker) return worker diff --git a/bec_widgets/widgets/services/bec_queue/bec_queue.py b/bec_widgets/widgets/services/bec_queue/bec_queue.py index 0e96aee8..8936d23a 100644 --- a/bec_widgets/widgets/services/bec_queue/bec_queue.py +++ b/bec_widgets/widgets/services/bec_queue/bec_queue.py @@ -4,6 +4,7 @@ import json from bec_lib import messages from bec_lib.endpoints import MessageEndpoints +from bec_lib.logger import bec_logger from bec_qthemes import material_icon from qtpy.QtCore import Property, Qt, Signal, Slot from qtpy.QtGui import QColor @@ -20,6 +21,8 @@ from bec_widgets.widgets.control.buttons.button_reset.button_reset import ResetB from bec_widgets.widgets.control.buttons.button_resume.button_resume import ResumeButton from bec_widgets.widgets.control.buttons.stop_button.stop_button import StopButton +logger = bec_logger.logger + class BECQueue(BECWidget, CompactPopupWidget): """ @@ -243,8 +246,8 @@ class BECQueue(BECWidget, CompactPopupWidget): try: color = self.status_colors.get(content, "black") # Default to black if not found item.setForeground(QColor(color)) - except: - return item + except Exception as e: + logger.warning(f"Could not apply status color for queue item '{content}': {e}") return item def set_row( diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index e97ff5bd..e53063ab 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -390,3 +390,33 @@ def scan_history_factory(tmpdir): return create_history_file(file_path, data, metadata) return _factory + + +@pytest.fixture(scope="session", autouse=True) +def _register_worker_thread_dummies(): + """Qt thread-pool threads register a permanent ``threading._DummyThread`` the first + time Python inspects them (e.g. a loguru call inside a background Worker). Saturate + the global pool once up front - with thread expiry disabled so the threads persist - + so bec_lib's threads_check fixture never sees these registrations appear mid-test.""" + import threading + + from qtpy.QtCore import QRunnable, QThreadPool + + pool = QThreadPool.globalInstance() + pool.setExpiryTimeout(-1) + barrier = threading.Barrier(pool.maxThreadCount() + 1, timeout=10) + + class _Warmup(QRunnable): + def run(self): + threading.current_thread() + try: + barrier.wait() + except threading.BrokenBarrierError: + pass + + for _ in range(pool.maxThreadCount()): + pool.start(_Warmup()) + try: + barrier.wait() + except threading.BrokenBarrierError: + pass diff --git a/tests/unit_tests/test_bec_connector.py b/tests/unit_tests/test_bec_connector.py index b21a726c..8dd77952 100644 --- a/tests/unit_tests/test_bec_connector.py +++ b/tests/unit_tests/test_bec_connector.py @@ -243,3 +243,37 @@ def test_bec_connector_terminate_registered_once_qapp_exists(qtbot): handler = BECConnector.EXIT_HANDLERS.pop(fresh_client, None) if handler is not None: QApplication.instance().aboutToQuit.disconnect(handler) + + +def test_bec_connector_submit_task_failure_removes_worker(bec_connector, qtbot): + """A task that raises must not leak the worker + reference, must emit failed, and must not call on_complete.""" + import threading + + failures = [] + completed = [] + # the worker starts before submit_task returns, so the raise is gated until the + # failed connection below exists - otherwise this test races its own setup + connected = threading.Event() + + def boom(): + connected.wait(timeout=5) + raise RuntimeError("task failed on purpose") + + worker = bec_connector.submit_task(boom, on_complete=lambda: completed.append(True)) + worker.signals.failed.connect(lambda msg: failures.append(msg)) + connected.set() + + qtbot.waitUntil(lambda: worker not in bec_connector._workers, timeout=5000) + qtbot.waitUntil(lambda: len(failures) == 1, timeout=5000) + assert completed == [] + assert "task failed on purpose" in failures[0] + + +def test_bec_connector_parent_id_returns_none_on_error(bec_connector): + """parent_id must swallow only Exception and + return None explicitly.""" + with mock.patch.object( + bec_connector, "_get_rpc_parent_ancestor", side_effect=ValueError("broken hierarchy") + ): + assert bec_connector.parent_id is None