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

@@ -36,13 +36,24 @@ class QueueItem:
         """A queue item for the scheduled events."""
 
         time: datetime
+        stop_event: bool
+        """Whether this is a stop event.
+
+        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.stop_event = stop_event
             self.dispatch = dispatch
 
     # pylint: disable=too-many-arguments
@@ -365,7 +376,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 +398,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(

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()