Ensure that start events are executed before stop events

This allows us to filter the `stop` events when we desire to
merge consecutive events.

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

Author: Marenz

src/frequenz/dispatch/_bg_service.py

@@ -33,16 +33,32 @@ class DispatchScheduler(BackgroundService):
 
     @dataclass(order=True)
     class QueueItem:
-        """A queue item for the scheduled events."""
+        """A queue item for the scheduled events.
+
+        This class is used to define the order of the queue items based on the
+        scheduled time, whether it is a stop event and finally the dispatch ID
+        (for uniqueness).
+        """
 
         time: datetime
+        priority: int
+        """Sort priority when the time is the same.
+
+        Exists to make sure that when two events are scheduled at the same time,
+        the start event is executed first, allowing filters down the data pipeline
+        to consider the start event when deciding whether to execute the
+        stop event.
+        """
         dispatch_id: int
         dispatch: Dispatch = field(compare=False)
 
-        def __init__(self, time: datetime, dispatch: Dispatch) -> None:
+        def __init__(
+            self, time: datetime, dispatch: Dispatch, stop_event: bool
+        ) -> None:
             """Initialize the queue item."""
             self.time = time
             self.dispatch_id = dispatch.id
+            self.priority = int(stop_event)
             self.dispatch = dispatch
 
     # pylint: disable=too-many-arguments
@@ -103,7 +119,7 @@ def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         )
 
     async def new_running_state_event_receiver(
-        self, type: str, unify_running_intervals: bool = True
+        self, type: str, *, unify_running_intervals: bool = True
     ) -> Receiver[Dispatch]:
         """Create a new receiver for running state events of the specified type.
 
@@ -365,7 +381,10 @@ def _schedule_start(self, dispatch: Dispatch) -> None:
         # Schedule the next run
         try:
             if next_run := dispatch.next_run:
-                heappush(self._scheduled_events, self.QueueItem(next_run, dispatch))
+                heappush(
+                    self._scheduled_events,
+                    self.QueueItem(next_run, dispatch, stop_event=False),
+                )
                 _logger.debug(
                     "Scheduled dispatch %s to start at %s", dispatch.id, next_run
                 )
@@ -384,7 +403,9 @@ def _schedule_stop(self, dispatch: Dispatch) -> None:
         if dispatch.duration and dispatch.duration > timedelta(seconds=0):
             until = dispatch.until
             assert until is not None
-            heappush(self._scheduled_events, self.QueueItem(until, dispatch))
+            heappush(
+                self._scheduled_events, self.QueueItem(until, dispatch, stop_event=True)
+            )
             _logger.debug("Scheduled dispatch %s to stop at %s", dispatch, until)
 
     def _update_changed_running_state(

src/frequenz/dispatch/_dispatcher.py

@@ -200,7 +200,7 @@ def new_lifecycle_events_receiver(
         return self._bg_service.new_lifecycle_events_receiver(dispatch_type)
 
     async def new_running_state_event_receiver(
-        self, dispatch_type: str, unify_running_intervals: bool = True
+        self, dispatch_type: str, *, unify_running_intervals: bool = True
     ) -> Receiver[Dispatch]:
         """Return running state event receiver.
 
@@ -241,5 +241,5 @@ async def new_running_state_event_receiver(
             A new receiver for dispatches whose running status changed.
         """
         return await self._bg_service.new_running_state_event_receiver(
-            dispatch_type, unify_running_intervals
+            dispatch_type, unify_running_intervals=unify_running_intervals
         )

tests/test_mananging_actor.py

@@ -147,14 +147,40 @@ def test_heapq_dispatch_compare(test_env: TestEnv) -> None:
     # Push two events with the same 'until' time onto the heap
     heapq.heappush(
         scheduled_events,
-        DispatchScheduler.QueueItem(until_time, Dispatch(dispatch1)),
+        DispatchScheduler.QueueItem(until_time, Dispatch(dispatch1), True),
     )
     heapq.heappush(
         scheduled_events,
-        DispatchScheduler.QueueItem(until_time, Dispatch(dispatch2)),
+        DispatchScheduler.QueueItem(until_time, Dispatch(dispatch2), True),
     )
 
 
+def test_heapq_dispatch_start_stop_compare(test_env: TestEnv) -> None:
+    """Test that the heapq compare function works."""
+    dispatch1 = test_env.generator.generate_dispatch()
+    dispatch2 = test_env.generator.generate_dispatch()
+
+    # Simulate two dispatches with the same 'until' time
+    now = datetime.now(timezone.utc)
+    until_time = now + timedelta(minutes=5)
+
+    # Create the heap
+    scheduled_events: list[DispatchScheduler.QueueItem] = []
+
+    # Push two events with the same 'until' time onto the heap
+    heapq.heappush(
+        scheduled_events,
+        DispatchScheduler.QueueItem(until_time, Dispatch(dispatch1), stop_event=False),
+    )
+    heapq.heappush(
+        scheduled_events,
+        DispatchScheduler.QueueItem(until_time, Dispatch(dispatch2), stop_event=True),
+    )
+
+    assert scheduled_events[0].dispatch_id == dispatch1.id
+    assert scheduled_events[1].dispatch_id == dispatch2.id
+
+
 async def test_dry_run(test_env: TestEnv, fake_time: time_machine.Coordinates) -> None:
     """Test the dry run mode."""
     dispatch = test_env.generator.generate_dispatch()