From 1ac298736f2a92aad16ebad326f0d477342eb714 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Mon, 3 Aug 2026 11:26:41 +0200 Subject: [PATCH] fix(connector): connect on_failed callbacks before the worker starts --- bec_widgets/utils/bec_connector.py | 12 ++++++-- tests/unit_tests/test_bec_connector.py | 38 ++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/bec_widgets/utils/bec_connector.py b/bec_widgets/utils/bec_connector.py index 511a5edc..a39fe52d 100644 --- a/bec_widgets/utils/bec_connector.py +++ b/bec_widgets/utils/bec_connector.py @@ -369,7 +369,9 @@ class BECConnector: self.rpc_register.mark_broadcast_pending() self.rpc_register.broadcast() - def submit_task(self, fn, *args, on_complete: SafeSlot = None, **kwargs) -> Worker: + def submit_task( + self, fn, *args, on_complete: SafeSlot = None, on_failed: SafeSlot = None, **kwargs + ) -> Worker: """ Submit a task to run in a separate thread. The task will run the specified function with the provided arguments and emit the completed signal when done. @@ -380,7 +382,11 @@ class BECConnector: Args: fn: Function to run in a separate thread. *args: Arguments for the function. - on_complete: Slot to run when the task is complete. + on_complete: Slot to run when the task completes successfully. + on_failed: Slot to run when the task raises; receives the formatted traceback + string. Pass it here rather than connecting to ``worker.signals.failed`` + after this method returns: the worker starts before this method returns, + so a late connection can miss the emission of a fast-failing task. **kwargs: Keyword arguments for the function. Returns: @@ -400,6 +406,8 @@ class BECConnector: worker = Worker(fn, *args, **kwargs) if on_complete: worker.signals.completed.connect(on_complete) + if on_failed: + worker.signals.failed.connect(on_failed) # Keep a reference to the worker so it is not garbage collected. self._workers.append(worker) diff --git a/tests/unit_tests/test_bec_connector.py b/tests/unit_tests/test_bec_connector.py index 5b4fe749..534210be 100644 --- a/tests/unit_tests/test_bec_connector.py +++ b/tests/unit_tests/test_bec_connector.py @@ -300,3 +300,41 @@ def test_bec_connector_worker_completion_does_not_retain_owner(qtbot, mocked_cli gc.collect() assert ref() is None, "connector kept alive by worker completion closure" + + +def test_bec_connector_on_failed_never_misses_fast_failures(bec_connector, qtbot): + """``on_failed`` passed to submit_task is connected before the worker starts, so + even a task that raises immediately cannot emit ``failed`` before the connection + exists. Connecting to ``worker.signals.failed`` after submit_task returns cannot + give this guarantee.""" + failures = [] + + def boom(): + raise RuntimeError("instant failure") + + for _ in range(20): + bec_connector.submit_task(boom, on_failed=lambda msg: failures.append(msg)) + + qtbot.waitUntil(lambda: len(failures) == 20, timeout=5000) + qtbot.waitUntil(lambda: not bec_connector._workers, timeout=5000) + assert all("instant failure" in msg for msg in failures) + + +def test_bec_connector_worker_outcome_survives_deleted_signal_source(bec_connector, qtbot, capfd): + """At application shutdown the WorkerSignals C++ object can die before a late worker + finishes; the outcome emit must not escape Worker.run into the thread pool.""" + import threading + + import shiboken6 + + gate = threading.Event() + worker = bec_connector.submit_task(lambda: gate.wait(timeout=5)) + shiboken6.delete(worker.signals) + gate.set() + + qtbot.wait(300) # let the worker finish and attempt both emits + stderr = capfd.readouterr().err + assert "Error calling Python override of QRunnable::run()" not in stderr + # the discard connection died with the signal source; drop the worker manually + if worker in bec_connector._workers: + bec_connector._workers.remove(worker)