Dispatches that failed to start will now be retried after a delay.

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

Author: Marenz

RELEASE_NOTES.md

@@ -62,8 +62,9 @@ Further changes:
 * 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-rc1800.
 * Actor management with dispatches has been simplified:
-    * `Dispatcher.start_managing(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_managing(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_managing(dispatch_type)` to stop dispatching for the given type.
     * `Dispatcher.is_managed(dispatch_type)` to check if dispatching is active for the given type.
+    * Dispatches that failed to start will now be retried after a delay.
 * A new method `Dispatcher.wait_for_initialization()` has been added to wait for all actors to be initialized.
 * When using `async with Dispatcher(..) as dispatcher`, the dispatcher will first wait for the dispatch service to be initialized before entering the block.

src/frequenz/dispatch/_actor_dispatcher.py

@@ -7,9 +7,10 @@
 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.client.dispatch.types import TargetComponents
 from frequenz.sdk.actor import Actor, BackgroundService
 
@@ -132,13 +133,61 @@ async def main():
     ```
     """
 
-    def __init__(
+    class RetryFailedDispatches:
+        """Manages the retry of failed dispatches."""
+
+        def __init__(self, retry_interval: timedelta) -> None:
+            """Initialize the retry manager.
+
+            Args:
+                retry_interval: The interval between retries.
+            """
+            self._retry_interval = retry_interval
+            self._channel = Broadcast[Dispatch](name="retry_channel")
+            self._sender = self._channel.new_sender()
+            self._tasks: set[asyncio.Task[None]] = set()
+
+        def new_receiver(self) -> Receiver[Dispatch]:
+            """Create a new receiver for dispatches to retry.
+
+            Returns:
+                The receiver.
+            """
+            return self._channel.new_receiver()
+
+        def retry(self, dispatch: Dispatch) -> None:
+            """Retry a dispatch.
+
+            Args:
+                dispatch: The dispatch information to retry.
+            """
+            task = asyncio.create_task(self._retry_after_delay(dispatch))
+            self._tasks.add(task)
+            task.add_done_callback(self._tasks.remove)
+
+        async def _retry_after_delay(self, dispatch: Dispatch) -> None:
+            """Retry a dispatch after a delay.
+
+            Args:
+                dispatch: The dispatch information to retry.
+            """
+            _logger.info(
+                "Will retry dispatch %s after %s",
+                dispatch.id,
+                self._retry_interval,
+            )
+            await asyncio.sleep(self._retry_interval.total_seconds())
+            _logger.info("Retrying dispatch %s now", dispatch.id)
+            await self._sender.send(dispatch)
+
+    def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         actor_factory: Callable[
             [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
         ],
         running_status_receiver: Receiver[Dispatch],
         dispatch_identity: Callable[[Dispatch], int] | None = None,
+        retry_interval: timedelta | None = timedelta(seconds=60),
     ) -> None:
         """Initialize the dispatch handler.
 
@@ -148,6 +197,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. If `None`, retries are disabled.
         """
         super().__init__()
         self._dispatch_identity: Callable[[Dispatch], int] = (
@@ -161,6 +211,11 @@ def __init__(
             name="dispatch_updates_channel", resend_latest=True
         )
         self._updates_sender = self._updates_channel.new_sender()
+        self._retrier = (
+            ActorDispatcher.RetryFailedDispatches(retry_interval)
+            if retry_interval
+            else None
+        )
 
     def start(self) -> None:
         """Start the background service."""
@@ -174,7 +229,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 = ""
@@ -193,17 +249,24 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
                     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,
                 )
+                if self._retrier:
+                    self._retrier.retry(dispatch)
+                else:
+                    _logger.error(
+                        "No retry mechanism enabled, dispatch %r failed", 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.
@@ -212,17 +275,31 @@ async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None:
             stopping_dispatch: The dispatch that is stopping the actor.
             msg: The message to be passed to the actors being stopped.
         """
-        if actor := self._actors.pop(self._dispatch_identity(stopping_dispatch), None):
+        actor: Actor | None = None
+        identity = self._dispatch_identity(stopping_dispatch)
+
+        actor = self._actors.get(identity)
+
+        if actor:
             await actor.stop(msg)
         else:
             _logger.warning(
                 "Actor for dispatch type %r is not running", stopping_dispatch.type
             )
 
     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."""
+        if not self._retrier:
+            async for dispatch in self._dispatch_rx:
+                await self._handle_dispatch(dispatch)
+        else:
+            retry_recv = self._retrier.new_receiver()
+
+            async for selected in select(retry_recv, self._dispatch_rx):
+                if retry_recv.triggered(selected):
+                    self._retrier.retry(selected.message)
+                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, Self
 
 from frequenz.channels import Receiver
@@ -270,6 +271,7 @@ async def start_managing(
             [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
         ],
         merge_strategy: MergeStrategy | None = None,
+        retry_interval: timedelta = timedelta(seconds=60),
     ) -> None:
         """Manage actors for a given dispatch type.
 
@@ -298,6 +300,7 @@ async def start_managing(
             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)
 
@@ -321,6 +324,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

@@ -84,6 +84,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:
@@ -170,6 +177,47 @@ 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))
+
+    # Replace failing mock actor factory with a working one
+    test_env.actors_service._actor_factory = MockActor.create
+
+    # Give retry task time to start
+    await asyncio.sleep(1)
+
+    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()