Fix crash when dispatches share equal start time (frequenz-floss#75)

- **Fix exception when two dispatches have equal start time**
- **Ensure we don't accidentally work on invalid rrules**

fixes
frequenz-floss#74

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

@@ -211,12 +211,18 @@ def _prepare_rrule(self) -> rrule.rrule:
 
         Returns:
             The rrule object.
+
+        Raises:
+            ValueError: If the interval is invalid.
         """
         count, until = (None, None)
         if end := self.recurrence.end_criteria:
             count = end.count
             until = end.until
 
+        if self.recurrence.interval is None or self.recurrence.interval < 1:
+            raise ValueError("Interval must be at least 1")
+
         rrule_obj = rrule.rrule(
             freq=_RRULE_FREQ_MAP[self.recurrence.frequency],
             dtstart=self.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()