Send STOPPED message when duration is reached

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

Author: Marenz

RELEASE_NOTES.md

@@ -6,7 +6,7 @@
 
 ## Upgrading
 
-<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+* `Dispatcher.running_state_change` now also sends a message when the duration specified in the dispatch has passed. If no duration is specified, no STOPPED message will be sent.
 
 ## New Features
 

src/frequenz/dispatch/_dispatch.py

@@ -117,6 +117,10 @@ def running(self, type_: str) -> RunningState:
         if not self.active or self.deleted:
             return RunningState.STOPPED
 
+        # A dispatch without duration is always running
+        if self.duration is None:
+            return RunningState.RUNNING
+
         now = datetime.now(tz=timezone.utc)
         if until := self._until(now):
             return RunningState.RUNNING if now < until else RunningState.STOPPED

src/frequenz/dispatch/actor.py

@@ -183,10 +183,20 @@ def next_run_info() -> tuple[datetime, datetime] | None:
             _logger.info("Dispatch %s scheduled for %s", dispatch.id, next_time)
             await asyncio.sleep((next_time - now).total_seconds())
 
-            _logger.info("Dispatch ready: %s", dispatch)
+            _logger.info("Dispatch %s executing...", dispatch)
             await self._running_state_change_sender.send(dispatch)
 
-        _logger.info("Dispatch finished: %s", dispatch)
+            # Wait for the duration of the dispatch if set
+            if dispatch.duration:
+                _logger.info(
+                    "Dispatch %s running for %s", dispatch.id, dispatch.duration
+                )
+                await asyncio.sleep(dispatch.duration.total_seconds())
+
+                _logger.info("Dispatch %s runtime duration reached", dispatch.id)
+                await self._running_state_change_sender.send(dispatch)
+
+        _logger.info("Dispatch completed: %s", dispatch)
         self._scheduled.pop(dispatch.id)
 
     def _running_state_change(

tests/test_frequenz_dispatch.py

@@ -236,6 +236,15 @@ async def test_dispatch_schedule(
     fake_time.shift(next_run - _now() - timedelta(seconds=1))
     await asyncio.sleep(1)
 
+    # Expect notification of the dispatch being ready to run
     ready_dispatch = await actor_env.ready_dispatches.receive()
 
     assert ready_dispatch == dispatch
+
+    # Shift time to the end of the dispatch
+    fake_time.shift(dispatch.duration + timedelta(seconds=1))
+    await asyncio.sleep(1)
+
+    # Expect notification to stop the dispatch
+    done_dispatch = await actor_env.ready_dispatches.receive()
+    assert done_dispatch == dispatch