Add a restart delay to actors to avoid spam-restarting of buggy actors

Signed-off-by: Christian Parpart <[email protected]>

Author: christianparpart

RELEASE_NOTES.md

@@ -25,6 +25,8 @@ This version ships an experimental version of the **Power Manager**, adds prelim
 
 - Move `microgrid.ComponentGraph` class to `microgrid.component_graph.ComponentGraph`, exposing only the high level interface functions through the `microgrid` package.
 
+- An actor that is crashing will no longer instantly restart but induce an artificial delay to avoid potential spam-restarting.
+
 ## New Features
 
 - New and improved documentation.

src/frequenz/sdk/actor/_actor.py

@@ -54,6 +54,20 @@ def start(self) -> None:
     async def _run(self) -> None:
         """Run this actor's logic."""
 
+    async def _delay_if_restart(self, iteration: int) -> None:
+        """Delay the restart of this actor's n'th iteration.
+
+        Args:
+            iteration: The current iteration of the restart.
+        """
+        # NB: I think it makes sense (in the future) to think about deminishing returns
+        # the longer the actor has been running.
+        # Not just for the restart-delay but actually for the n_restarts counter as well.
+        if iteration > 0:
+            delay: int = 1 << iteration # delay for 1, 2, 4, 8, ... seconds
+            _logger.info("Actor %s: Waiting %s seconds...", self, delay)
+            await asyncio.sleep(delay)
+
     async def _run_loop(self) -> None:
         """Run this actor's task in a loop until `_restart_limit` is reached.
 
@@ -67,6 +81,7 @@ async def _run_loop(self) -> None:
         n_restarts = 0
         while True:
             try:
+                await self._delay_if_restart(n_restarts)
                 await self._run()
                 _logger.info("Actor %s: _run() returned without error.", self)
             except asyncio.CancelledError:

tests/actor/test_actor.py

@@ -194,7 +194,21 @@ async def test_basic_actor(caplog: pytest.LogCaptureFixture) -> None:
     ]
 
 
-@pytest.mark.parametrize("restart_limit", [0, 1, 2, 10])
+def expected_wait_time(iterations: int) -> int:
+    """Calculate the expected wait time for a given iteration.
+
+    Args:
+        iterations: The iteration to calculate the wait time for.
+
+    Returns:
+        The expected wait time in seconds.
+    """
+    if iterations == 0:
+        return 0
+    return expected_wait_time(iterations - 1) + 1 << iterations
+
+
+@pytest.mark.parametrize("restart_limit", [0, 1, 2, 3])
 async def test_restart_on_unhandled_exception(
     restart_limit: int, caplog: pytest.LogCaptureFixture
 ) -> None:
@@ -211,7 +225,8 @@ async def test_restart_on_unhandled_exception(
 
     channel: Broadcast[int] = Broadcast("channel")
 
-    async with asyncio.timeout(2.0):
+    async with asyncio.timeout(1 + expected_wait_time(restart_limit)):
+        print(f"Expecting a wait time of {expected_wait_time(restart_limit)} seconds")
         with actor_restart_limit(restart_limit):
             actor = RaiseExceptionActor(
                 channel.new_receiver(),
@@ -240,6 +255,10 @@ async def test_restart_on_unhandled_exception(
                     *ACTOR_INFO,
                     f"Actor test: Restarting ({i}/{restart_limit})...",
                 ),
+                (
+                    *ACTOR_INFO,
+                    f"Actor RaiseExceptionActor[test]: Waiting {1 << (1 + i)} seconds...",
+                ),
             ]
         )
     expected_log.extend(
@@ -260,6 +279,8 @@ async def test_restart_on_unhandled_exception(
             (*RUN_INFO, "All 1 actor(s) finished."),
         ]
     )
+    print("expected_log:", expected_log)
+    print("caplog.record_tuples:", caplog.record_tuples)
     assert caplog.record_tuples == expected_log