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,24 @@ 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
+
+        self._is_cancelled = False
         self._tasks.clear()
         self._tasks.add(asyncio.create_task(self._run_loop()))
 
@@ -94,6 +105,17 @@ async def _run_loop(self) -> None:
                 _logger.info("Actor %s: Cancelled.", self)
                 raise
             except Exception:  # pylint: disable=broad-except
+                if self._is_cancelled:
+                    # If actor was cancelled, but any tasks have failed with an exception
+                    # other than asyncio.CancelledError, those exceptions are combined
+                    # in an ExceptionGroup or BaseExceptionGroup.
+                    # We have to handle that case separately to stop actor instead
+                    # of restarting it.
+                    _logger.exception(
+                        "Actor %s: Raised an unhandled exception during stop.", self
+                    )
+                    break
+
                 _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 +135,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

@@ -391,3 +391,12 @@ async def test_actor_stop_if_error_was_raised_during_cancel(
 
     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.",
+        ),
+        (*ACTOR_INFO, "Actor RaiseExceptionOnCancelActor[test]: Stopped."),
+    ]