Fix issue with actor restarting instead of stopping

When an actor threw an exception during cancellation,
it restarted rather than stopping as expected.
Now it stops and raises an unhandled exception.

Signed-off-by: Elzbieta Kotulska <[email protected]>

Author: ela-kotulska-frequenz

RELEASE_NOTES.md

@@ -14,4 +14,4 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+- Fixed issue where actors would restart instead of stopping when exceptions occurred during cancellation. Actors now properly stop and surface the unhandled exception.

src/frequenz/sdk/actor/_actor.py

@@ -44,13 +44,31 @@ class Actor(BackgroundService, abc.ABC):
         This is mostly used for testing purposes and shouldn't be set in production.
     """
 
+    def __init__(self, *, name: str | None = None) -> None:
+        """Create actor instance.
+
+        Args:
+            name: The name of this background service.
+        """
+        super().__init__(name=name)
+        self._is_cancelled = False
+
     def start(self) -> None:
         """Start this actor.
 
         If this actor is already running, this method does nothing.
         """
         if self.is_running:
             return
+
+        if self._is_cancelled:
+            _logger.error(
+                (
+                    "Actor %s: was canceled and can't be re-run."
+                    "Please create another actor instance.",
+                ),
+                self,
+            )
         self._tasks.clear()
         self._tasks.add(asyncio.create_task(self._run_loop()))
 
@@ -94,6 +112,12 @@ async def _run_loop(self) -> None:
                 _logger.info("Actor %s: Cancelled.", self)
                 raise
             except Exception:  # pylint: disable=broad-except
+                if self._is_cancelled:
+                    _logger.exception(
+                        "Actor %s: Raised an unhandled exception during stop.", self
+                    )
+                    raise
+
                 _logger.exception("Actor %s: Raised an unhandled exception.", self)
                 limit_str = "∞" if self._restart_limit is None else self._restart_limit
                 limit_str = f"({n_restarts}/{limit_str})"
@@ -113,3 +137,14 @@ async def _run_loop(self) -> None:
             break
 
         _logger.info("Actor %s: Stopped.", self)
+
+    def cancel(self, msg: str | None = None) -> None:
+        """Cancel actor.
+
+        Cancelled actor can't be started again.
+
+        Args:
+            msg: The message to be passed to the tasks being cancelled.
+        """
+        self._is_cancelled = True
+        return super().cancel(msg)

tests/actor/test_actor.py

@@ -389,5 +389,17 @@ async def test_actor_stop_if_error_was_raised_during_cancel(
     assert msg == 5
     assert actor.is_running is True
 
-    await actor.stop()
+    # Actor raises unhandled exception
+    # this exception is propagated to user
+    # during stop.
+    with pytest.raises(ExceptionGroup):
+        await actor.stop()
     assert actor.restart_count == 0
+
+    assert caplog.record_tuples == [
+        (*ACTOR_INFO, "Actor RaiseExceptionOnCancelActor[test]: Started."),
+        (
+            *ACTOR_ERROR,
+            "Actor RaiseExceptionOnCancelActor[test]: Raised an unhandled exception during stop.",
+        ),
+    ]