Add retry delay and auto restart to ActorService

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

RELEASE_NOTES.md

@@ -23,6 +23,7 @@ This release introduces a more flexible and powerful mechanism for managing disp
 * A new feature "merge strategy" (`MergeByType`, `MergeByTypeTarget`) has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
 * The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1600.
 * Actor management with dispatches has been simplified:
-    * `Dispatcher.start_dispatching(dispatch_type, actor_factory, merge_strategy)` to manage your actor for the given type and merge strategy. All you need provide is an actor factory.
+    * `Dispatcher.start_dispatching(dispatch_type, actor_factory, merge_strategy, retry_interval)` to manage your actor for the given type and merge strategy. All you need provide is an actor factory.
     * `Dispatcher.stop_dispatching(dispatch_type)` to stop dispatching for the given type.
     * `Dispatcher.is_dispatching(dispatch_type)` to check if dispatching is active for the given type.
+    * Dispatches that failed to start will now be retried after a delay.

src/frequenz/dispatch/_actor_dispatcher.py

@@ -7,9 +7,11 @@
 import logging
 from collections.abc import Callable
 from dataclasses import dataclass
+from datetime import timedelta
 from typing import Any, Awaitable
 
-from frequenz.channels import Broadcast, Receiver
+from frequenz.channels import Broadcast, Receiver, select
+from frequenz.channels.timer import SkipMissedAndResync, Timer
 from frequenz.client.dispatch.types import TargetComponents
 from frequenz.sdk.actor import Actor, BackgroundService
 
@@ -32,6 +34,7 @@ class DispatchInfo:
     """Additional options."""
 
 
+# pylint: disable=too-many-instance-attributes
 class ActorDispatcher(BackgroundService):
     """Helper class to manage actors based on dispatches.
 
@@ -141,6 +144,7 @@ def __init__(
         ],
         running_status_receiver: Receiver[Dispatch],
         dispatch_identity: Callable[[Dispatch], int] | None = None,
+        retry_interval: timedelta = timedelta(seconds=60),
     ) -> None:
         """Initialize the dispatch handler.
 
@@ -150,6 +154,7 @@ def __init__(
             running_status_receiver: The receiver for dispatch running status changes.
             dispatch_identity: A function to identify to which actor a dispatch refers.
                 By default, it uses the dispatch ID.
+            retry_interval: The interval between retries.
         """
         super().__init__()
         self._dispatch_identity: Callable[[Dispatch], int] = (
@@ -159,6 +164,9 @@ def __init__(
         self._dispatch_rx = running_status_receiver
         self._actor_factory = actor_factory
         self._actors: dict[int, Actor] = {}
+        self._start_failed_dispatches: dict[int, Dispatch] = {}
+        self._retry_timer = Timer(retry_interval, SkipMissedAndResync())
+
         self._updates_channel = Broadcast[DispatchInfo](
             name="dispatch_updates_channel", resend_latest=True
         )
@@ -176,7 +184,8 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
             options=dispatch.payload,
         )
 
-        actor: Actor | None = self._actors.get(self._dispatch_identity(dispatch))
+        identity = self._dispatch_identity(dispatch)
+        actor: Actor | None = self._actors.get(identity)
 
         if actor:
             sent_str = ""
@@ -189,23 +198,28 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
                 sent_str,
             )
         else:
+            # Remove pending restarts for this identity
+            self._start_failed_dispatches.pop(identity, None)
+
             try:
                 _logger.info("Starting actor for dispatch type %r", dispatch.type)
                 actor = await self._actor_factory(
                     dispatch_update,
                     self._updates_channel.new_receiver(limit=1, warn_on_overflow=False),
                 )
-                self._actors[self._dispatch_identity(dispatch)] = actor
 
                 actor.start()
 
             except Exception as e:  # pylint: disable=broad-except
                 _logger.error(
-                    "Failed to start actor for dispatch type %r: %s",
+                    "Failed to start actor for dispatch type %r",
                     dispatch.type,
-                    e,
-                    exc_info=True,
+                    exc_info=e,
                 )
+                self._start_failed_dispatches[identity] = dispatch
+            else:
+                # No exception occurred, so we can add the actor to the list
+                self._actors[identity] = actor
 
     async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None:
         """Stop all actors.
@@ -222,9 +236,16 @@ async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None:
             )
 
     async def _run(self) -> None:
-        """Wait for dispatches and handle them."""
-        async for dispatch in self._dispatch_rx:
-            await self._handle_dispatch(dispatch=dispatch)
+        """Run the background service."""
+        async for selected in select(self._retry_timer, self._dispatch_rx):
+            if self._retry_timer.triggered(selected) and self._start_failed_dispatches:
+                _logger.info("Retrying failed dispatches")
+                failed_dispatches = self._start_failed_dispatches.values()
+                self._start_failed_dispatches = {}
+                for dispatch in failed_dispatches:
+                    await self._start_actor(dispatch)
+            elif self._dispatch_rx.triggered(selected):
+                await self._handle_dispatch(selected.message)
 
     async def _handle_dispatch(self, dispatch: Dispatch) -> None:
         """Handle a dispatch.

src/frequenz/dispatch/_dispatcher.py

@@ -8,6 +8,7 @@
 import asyncio
 import logging
 from asyncio import Event
+from datetime import timedelta
 from typing import Awaitable, Callable
 
 from frequenz.channels import Receiver
@@ -245,6 +246,7 @@ async def start_dispatching(
             [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
         ],
         merge_strategy: MergeStrategy | None = None,
+        retry_interval: timedelta = timedelta(seconds=60),
     ) -> None:
         """Manage actors for a given dispatch type.
 
@@ -259,6 +261,7 @@ async def start_dispatching(
             dispatch_type: The type of the dispatch to manage.
             actor_factory: The factory to create actors.
             merge_strategy: The strategy to merge running intervals.
+            retry_interval: Retry interval for when actor creation fails.
         """
         dispatcher = self._actor_dispatchers.get(dispatch_type)
 
@@ -281,6 +284,7 @@ def id_identity(dispatch: Dispatch) -> int:
             dispatch_identity=(
                 id_identity if merge_strategy is None else merge_strategy.identity
             ),
+            retry_interval=retry_interval,
         )
 
         self._actor_dispatchers[dispatch_type] = dispatcher

tests/test_mananging_actor.py

@@ -86,6 +86,13 @@ async def create(
         actor = cls(initial_dispatch, receiver)
         return actor
 
+    @classmethod
+    async def create_fail(
+        cls, __: DispatchInfo, _: Receiver[DispatchInfo]
+    ) -> "MockActor":
+        """Create a new actor."""
+        raise ValueError("Failed to create actor")
+
 
 @dataclass
 class TestEnv:
@@ -175,6 +182,45 @@ async def test_simple_start_stop(
     # pylint: enable=protected-access
 
 
+async def test_start_failed(
+    test_env: TestEnv, fake_time: time_machine.Coordinates
+) -> None:
+    """Test auto-retry after 60 seconds."""
+    # pylint: disable=protected-access
+    test_env.actors_service._actor_factory = MockActor.create_fail
+
+    now = _now()
+    duration = timedelta(minutes=10)
+    dispatch = test_env.generator.generate_dispatch()
+    dispatch = replace(
+        dispatch,
+        id=1,
+        active=True,
+        dry_run=False,
+        duration=duration,
+        start_time=now,
+        payload={"test": True},
+        type="UNIT_TEST",
+        recurrence=replace(
+            dispatch.recurrence,
+            frequency=Frequency.UNSPECIFIED,
+        ),
+    )
+
+    # Send status update to start actor, expect no DispatchInfo for the start
+    await test_env.running_status_sender.send(Dispatch(dispatch))
+    fake_time.shift(timedelta(seconds=1))
+    await asyncio.sleep(1)
+
+    assert test_env.actors_service._start_failed_dispatches[1].id == dispatch.id
+    test_env.actors_service._actor_factory = MockActor.create
+
+    fake_time.shift(timedelta(seconds=65))
+    await asyncio.sleep(65)
+
+    assert test_env.actor(1).is_running is True
+
+
 def test_heapq_dispatch_compare(test_env: TestEnv) -> None:
     """Test that the heapq compare function works."""
     dispatch1 = test_env.generator.generate_dispatch()
@@ -311,6 +357,7 @@ async def new_mock_receiver(
         # pylint: disable=protected-access
         assert "MANAGE_TEST" in dispatcher._actor_dispatchers
         actor_manager = dispatcher._actor_dispatchers["MANAGE_TEST"]
+        # pylint: disable=comparison-with-callable
         assert actor_manager._actor_factory == MockActor.create
 
         dispatch = Dispatch(