Fix exception when comparing dispatches with equal start times

Previously, when two dispatches had the same start time, `heapq` would attempt to compare the dispatch objects themselves, leading to exceptions due to unsupported comparisons. This commit ensures dispatch objects are now properly handled when start times are identical, preventing such errors.

Thanks to Merlin for identifying this issue!

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

Author: Marenz

RELEASE_NOTES.md

@@ -2,4 +2,16 @@
 
 ## Summary
 
-This is a hot fix for recurrence not working
+<!-- Here goes a general summary of what this release is about -->
+
+## Upgrading
+
+<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+
+## New Features
+
+<!-- Here goes the main new features and examples or instructions on how to use them -->
+
+## Bug Fixes
+
+* Fixed a crash in the `DispatchManagingActor` when dispatches shared an equal start time.

src/frequenz/dispatch/actor.py

@@ -4,6 +4,7 @@
 """The dispatch actor."""
 
 import logging
+from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from heapq import heappop, heappush
 
@@ -30,6 +31,20 @@ class DispatchingActor(Actor):
     dispatches as necessary.
     """
 
+    @dataclass(order=True)
+    class QueueItem:
+        """A queue item for the scheduled events."""
+
+        time: datetime
+        dispatch_id: int
+        dispatch: Dispatch = field(compare=False)
+
+        def __init__(self, time: datetime, dispatch: Dispatch) -> None:
+            """Initialize the queue item."""
+            self.time = time
+            self.dispatch_id = dispatch.id
+            self.dispatch = dispatch
+
     # pylint: disable=too-many-arguments
     def __init__(
         self,
@@ -61,7 +76,7 @@ def __init__(
         Interval is chosen arbitrarily, as it will be reset on the first event.
         """
 
-        self._scheduled_events: list[tuple[datetime, Dispatch]] = []
+        self._scheduled_events: list["DispatchingActor.QueueItem"] = []
         """The scheduled events, sorted by time.
 
         Each event is a tuple of the scheduled time and the dispatch.
@@ -84,9 +99,11 @@ async def _run(self) -> None:
                 if not self._scheduled_events:
                     continue
                 _logger.debug(
-                    "Executing scheduled event: %s", self._scheduled_events[0][1]
+                    "Executing scheduled event: %s", self._scheduled_events[0].dispatch
+                )
+                await self._execute_scheduled_event(
+                    heappop(self._scheduled_events).dispatch
                 )
-                await self._execute_scheduled_event(heappop(self._scheduled_events)[1])
             elif selected_from(selected, stream):
                 _logger.debug("Received dispatch event: %s", selected.message)
                 dispatch = Dispatch(selected.message.dispatch)
@@ -243,9 +260,9 @@ async def _update_dispatch_schedule_and_notify(
     def _update_timer(self) -> None:
         """Update the timer to the next event."""
         if self._scheduled_events:
-            due_at: datetime = self._scheduled_events[0][0]
+            due_at: datetime = self._scheduled_events[0].time
             self._next_event_timer.reset(interval=due_at - datetime.now(timezone.utc))
-            _logger.debug("Next event scheduled at %s", self._scheduled_events[0][0])
+            _logger.debug("Next event scheduled at %s", self._scheduled_events[0].time)
 
     def _remove_scheduled(self, dispatch: Dispatch) -> bool:
         """Remove a dispatch from the scheduled events.
@@ -256,8 +273,8 @@ def _remove_scheduled(self, dispatch: Dispatch) -> bool:
         Returns:
             True if the dispatch was found and removed, False otherwise.
         """
-        for idx, (_, sched_dispatch) in enumerate(self._scheduled_events):
-            if dispatch.id == sched_dispatch.id:
+        for idx, item in enumerate(self._scheduled_events):
+            if dispatch.id == item.dispatch.id:
                 self._scheduled_events.pop(idx)
                 return True
 
@@ -276,7 +293,7 @@ def _schedule_start(self, dispatch: Dispatch) -> None:
         # Schedule the next run
         try:
             if next_run := dispatch.next_run:
-                heappush(self._scheduled_events, (next_run, dispatch))
+                heappush(self._scheduled_events, self.QueueItem(next_run, dispatch))
                 _logger.debug(
                     "Scheduled dispatch %s to start at %s", dispatch.id, next_run
                 )
@@ -295,7 +312,7 @@ 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, (until, dispatch))
+            heappush(self._scheduled_events, self.QueueItem(until, dispatch))
             _logger.debug("Scheduled dispatch %s to stop at %s", dispatch, until)
 
     def _update_changed_running_state(

tests/test_mananging_actor.py

@@ -4,6 +4,7 @@
 """Test the dispatch runner."""
 
 import asyncio
+import heapq
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from typing import AsyncIterator, Iterator
@@ -17,6 +18,7 @@
 from pytest import fixture
 
 from frequenz.dispatch import Dispatch, DispatchManagingActor, DispatchUpdate
+from frequenz.dispatch.actor import DispatchingActor
 
 
 @fixture
@@ -128,6 +130,27 @@ async def test_simple_start_stop(
     assert test_env.actor.is_running is False
 
 
+def test_heapq_dispatch_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[DispatchingActor.QueueItem] = []
+
+    # Push two events with the same 'until' time onto the heap
+    heapq.heappush(
+        scheduled_events, DispatchingActor.QueueItem(until_time, Dispatch(dispatch1))
+    )
+    heapq.heappush(
+        scheduled_events, DispatchingActor.QueueItem(until_time, Dispatch(dispatch2))
+    )
+
+
 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()