Add a restart delay to actors to avoid spam-restarting of buggy actors (#741)

Closes #732

I've decided to go for the hard coded 2.0 seconds delay for now. I very
much like the idea of (instead of hard-coding an exponential backoff
in), allow configuring the actor at construction time, and have it
passed a retry strategy (as suggested by @shms), so the dev of that
actor can decide what's best for this particular actor. This can then
default to a fixed-delay retry strategy.

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

@@ -6,6 +6,7 @@
 import abc
 import asyncio
 import logging
+from datetime import timedelta
 
 from ._background_service import BackgroundService
 
@@ -27,6 +28,9 @@ class Actor(BackgroundService, abc.ABC):
         comprehensive guide on how to use and implement actors properly.
     """
 
+    RESTART_DELAY: timedelta = timedelta(seconds=2)
+    """The delay to wait between restarts of this actor."""
+
     _restart_limit: int | None = None
     """The number of times actors can be restarted when they are stopped by unhandled exceptions.
 
@@ -54,6 +58,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 = self.RESTART_DELAY.total_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 +85,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

@@ -4,6 +4,7 @@
 """Simple test for the BaseActor."""
 
 import asyncio
+from datetime import timedelta
 
 import pytest
 from frequenz.channels import Broadcast, Receiver, Sender
@@ -211,7 +212,12 @@ async def test_restart_on_unhandled_exception(
 
     channel: Broadcast[int] = Broadcast("channel")
 
-    async with asyncio.timeout(2.0):
+    # NB: We're adding 1.0s to the timeout to account for the time it takes to
+    # run, crash, and restart the actor.
+    expected_wait_time = timedelta(
+        seconds=restart_limit * RaiseExceptionActor.RESTART_DELAY.total_seconds() + 1.0
+    )
+    async with asyncio.timeout(expected_wait_time.total_seconds()):
         with actor_restart_limit(restart_limit):
             actor = RaiseExceptionActor(
                 channel.new_receiver(),
@@ -229,6 +235,7 @@ async def test_restart_on_unhandled_exception(
         (*RUN_INFO, "Actor RaiseExceptionActor[test]: Starting..."),
         (*ACTOR_INFO, "Actor RaiseExceptionActor[test]: Started."),
     ]
+    restart_delay = Actor.RESTART_DELAY.total_seconds()
     for i in range(restart_limit):
         expected_log.extend(
             [
@@ -240,6 +247,10 @@ async def test_restart_on_unhandled_exception(
                     *ACTOR_INFO,
                     f"Actor test: Restarting ({i}/{restart_limit})...",
                 ),
+                (
+                    *ACTOR_INFO,
+                    f"Actor RaiseExceptionActor[test]: Waiting {restart_delay} seconds...",
+                ),
             ]
         )
     expected_log.extend(
@@ -260,6 +271,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