Fix exception when two dispatches have equal start time

The heapq will first look at the given time to compare,
but when it is identical it will look at the attached dispatch objects
for comparison.
But those just aren't ready for so much responsibility.
UNTIL NOW!
Cudos to Merlin for finding this bug!

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/_dispatch.py

@@ -8,7 +8,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from enum import Enum
-from typing import Iterator, cast
+from typing import Any, Iterator, Self, cast
 
 from dateutil import rrule
 from frequenz.client.dispatch.types import Dispatch as BaseDispatch

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,18 @@ class DispatchingActor(Actor):
     dispatches as necessary.
     """
 
+    @dataclass(order=True)
+    class QueueItem:
+        """A queue item for the scheduled events."""
+
+        priority: tuple[datetime, int]
+        dispatch: Dispatch = field(compare=False)
+
+        def __init__(self, time: datetime, dispatch: Dispatch) -> None:
+            """Initialize the queue item."""
+            self.priority = (time, dispatch.id)
+            self.dispatch = dispatch
+
     # pylint: disable=too-many-arguments
     def __init__(
         self,
@@ -61,7 +74,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 +97,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 +258,11 @@ 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].priority[0]
             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].priority[0]
+            )
 
     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()