Fix wrong running behaviors with duration=None

* Dispatch.running() was always returning RUNNING for duration None
* Dispatch.next_run_after() was made aware that there is no next run for
  inf. Duration
* Dispatch._until() now raises when used with Dispatch.None
* Actor: delay marking a Dispatch as deleted, otherwise checks wrongly
  detect it as no longer running.

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

Author: Marenz

pyproject.toml

@@ -41,7 +41,8 @@ dependencies = [
   # plugins.mkdocstrings.handlers.python.import)
   "frequenz-sdk >= 1.0.0-rc900, < 1.0.0-rc1000",
   "frequenz-channels >= 1.1.0, < 2.0.0",
-  "frequenz-client-dispatch >= 0.6.0, < 0.7.0",
+#  "frequenz-client-dispatch >= 0.6.0, < 0.7.0",
+  "frequenz-client-dispatch @ git+https://github.com/frequenz-floss/[email protected]",
 ]
 dynamic = ["version"]
 

src/frequenz/dispatch/_dispatch.py

@@ -117,11 +117,14 @@ def running(self, type_: str) -> RunningState:
         if not self.active or self.deleted:
             return RunningState.STOPPED
 
-        # A dispatch without duration is always running
+        now = datetime.now(tz=timezone.utc)
+
+        # A dispatch without duration is always running once it started
         if self.duration is None:
-            return RunningState.RUNNING
+            if self.start_time <= now:
+                return RunningState.RUNNING
+            return RunningState.STOPPED
 
-        now = datetime.now(tz=timezone.utc)
         if until := self._until(now):
             return RunningState.RUNNING if now < until else RunningState.STOPPED
 
@@ -189,6 +192,7 @@ def next_run_after(self, after: datetime) -> datetime | None:
         if (
             not self.recurrence.frequency
             or self.recurrence.frequency == Frequency.UNSPECIFIED
+            or self.duration is None  # Infinite duration
         ):
             if after > self.start_time:
                 return None
@@ -240,7 +244,13 @@ def _until(self, now: datetime) -> datetime | None:
 
         Returns:
             The time when the dispatch should end or None if the dispatch is not running.
+
+        Raises:
+            ValueError: If the dispatch has no duration.
         """
+        if self.duration is None:
+            raise ValueError("_until: Dispatch has no duration")
+
         if (
             not self.recurrence.frequency
             or self.recurrence.frequency == Frequency.UNSPECIFIED

src/frequenz/dispatch/actor.py

@@ -118,14 +118,17 @@ async def _fetch(self) -> None:
 
         for dispatch in old_dispatches.values():
             _logger.info("Deleted dispatch: %s", dispatch)
-            dispatch._set_deleted()  # pylint: disable=protected-access
-            await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
             if task := self._scheduled.pop(dispatch.id, None):
                 task.cancel()
 
             if self._running_state_change(None, dispatch):
                 await self._send_running_state_change(dispatch)
 
+            # Set deleted only here as it influences the result of dispatch.running()
+            # which is used in above in _running_state_change
+            dispatch._set_deleted()  # pylint: disable=protected-access
+            await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
+
     def _update_dispatch_schedule(
         self, dispatch: Dispatch, old_dispatch: Dispatch | None
     ) -> None:

tests/test_frequenz_dispatch.py

@@ -18,7 +18,14 @@
 from frequenz.client.dispatch.types import Frequency
 from pytest import fixture
 
-from frequenz.dispatch import Created, Deleted, Dispatch, DispatchEvent, Updated
+from frequenz.dispatch import (
+    Created,
+    Deleted,
+    Dispatch,
+    DispatchEvent,
+    RunningState,
+    Updated,
+)
 from frequenz.dispatch.actor import DispatchingActor
 
 
@@ -220,6 +227,39 @@ async def test_existing_dispatch_deleted(
             assert dispatch == Dispatch(sample, deleted=True)
 
 
+async def test_dispatch_inf_duration_deleted(
+    actor_env: ActorTestEnv,
+    generator: DispatchGenerator,
+    fake_time: time_machine.Coordinates,
+) -> None:
+    """Test that a dispatch with infinite duration can be deleted while running."""
+    # Generate a dispatch with infinite duration (duration=None)
+    sample = generator.generate_dispatch()
+    sample = replace(
+        sample, active=True, duration=None, start_time=_now() + timedelta(seconds=5)
+    )
+    # Create the dispatch
+    sample = await _test_new_dispatch_created(actor_env, sample)
+    # Advance time to when the dispatch should start
+    fake_time.shift(timedelta(seconds=40))
+    await asyncio.sleep(40)
+    # Expect notification of the dispatch being ready to run
+    ready_dispatch = await actor_env.ready_dispatches.receive()
+    print("Received ready dispatch")
+    assert ready_dispatch.running(sample.type) == RunningState.RUNNING
+
+    # Now delete the dispatch
+    await actor_env.client.delete(
+        microgrid_id=actor_env.microgrid_id, dispatch_id=sample.id
+    )
+    fake_time.shift(timedelta(seconds=10))
+    print("Deleted dispatch")
+    await asyncio.sleep(1)
+    # Expect notification to stop the dispatch
+    done_dispatch = await actor_env.ready_dispatches.receive()
+    assert done_dispatch.running(sample.type) == RunningState.STOPPED
+
+
 async def test_dispatch_schedule(
     actor_env: ActorTestEnv,
     generator: DispatchGenerator,