Add method to re-send dispatch states after startup

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

Author: Marenz

src/frequenz/dispatch/_dispatcher.py

@@ -210,6 +210,17 @@ async def start(self) -> None:
         """Start the actor."""
         self._actor.start()
 
+    async def resend_current_running_states(self, dispatch_type: str) -> None:
+        """Resend the current running states of all dispatches of a given type.
+
+        Warning: Usually you don't need to call this method. It is only useful
+        when you need to recover the current running state of your actor.
+
+        Args:
+            dispatch_type: The type of dispatches to resend.
+        """
+        await self._actor.resend_current_running_states(dispatch_type)
+
     @property
     def client(self) -> Client:
         """Return the client."""

src/frequenz/dispatch/actor.py

@@ -84,6 +84,23 @@ def __init__(
         always at index 0.
         """
 
+    async def resend_current_running_states(self, dispatch_type: str) -> None:
+        """Trigger a resend of all running states.
+
+        Causes the current state to be sent out again on the running state
+        change channel.
+
+        Args:
+            dispatch_type: The type of dispatches to resend.
+        """
+        _logger.info(
+            "Resending current running states for dispatch type %s", dispatch_type
+        )
+        for dispatch in self._dispatches.values():
+            if dispatch.type == dispatch_type:
+                _logger.debug("Resending dispatch %s", dispatch)
+                await self._send_running_state_change(dispatch)
+
     async def _run(self) -> None:
         """Run the actor."""
         _logger.info("Starting dispatch actor for microgrid %s", self._microgrid_id)

tests/test_frequenz_dispatch.py

@@ -482,3 +482,48 @@ async def test_notification_on_actor_start(
     # Expect notification of the running dispatch being ready to run
     ready_dispatch = await actor_env.running_state_change.receive()
     assert ready_dispatch.started
+
+
+async def test_notification_on_resend(
+    actor_env: ActorTestEnv,
+    generator: DispatchGenerator,
+) -> None:
+    """Test that the correct notifications are sent when resending dispatches."""
+    # Generate a dispatch that is already running
+    running_dispatch = generator.generate_dispatch()
+    running_dispatch = replace(
+        running_dispatch,
+        active=True,
+        duration=timedelta(seconds=10),
+        start_time=_now() - timedelta(seconds=5),
+        recurrence=RecurrenceRule(),
+        type="I_SHOULD_RUN",
+    )
+    # Generate a dispatch that is not running
+    stopped_dispatch = generator.generate_dispatch()
+    stopped_dispatch = replace(
+        stopped_dispatch,
+        active=False,
+        duration=timedelta(seconds=5),
+        start_time=_now() - timedelta(seconds=5),
+        recurrence=RecurrenceRule(),
+        type="I_SHOULD_NOT_RUN",
+    )
+    await actor_env.actor.stop()
+
+    # Create the dispatches
+    actor_env.client.set_dispatches(
+        actor_env.microgrid_id, [running_dispatch, stopped_dispatch]
+    )
+
+    # Start the actor
+    actor_env.actor.start()
+    # Resend the stopped dispatch, but expect no notification
+    await actor_env.actor.resend_current_running_states(stopped_dispatch.type)
+    # Resend the running dispatch
+    await actor_env.actor.resend_current_running_states(running_dispatch.type)
+
+    # Expect notification of the running dispatch being ready to run
+    ready_dispatch = await actor_env.running_state_change.receive()
+    assert ready_dispatch.started
+    assert ready_dispatch.type == running_dispatch.type