Fix lost stop event on boundary updates

When an update arrives exactly at a dispatch stop boundary, the update
path can remove the pending stop event before the timer delivers it.
If no immediate runtime-change notification is emitted, the actor never
receives the stop event and keeps running after the dispatch window has
already elapsed.

Keep track of the removed queue item and emit the missing stop
notification when an update removes a stop event for a dispatch that is
no longer started. Add a regression test for the boundary-update case.

Signed-off-by: Mathias L. Baumann <mathias.baumann@frequenz.com>

Author: Marenz

src/frequenz/dispatch/_bg_service.py

@@ -486,11 +486,17 @@ async def _update_dispatch_schedule_and_notify(
         # Dispatch was updated
         elif dispatch and old_dispatch:
             # Remove potentially existing scheduled event
-            self._remove_scheduled(old_dispatch)
+            removed = self._remove_scheduled(old_dispatch)
 
             # Check if the change requires an immediate notification
             if self._update_changed_running_state(dispatch, old_dispatch):
                 await self._send_running_state_change(dispatch)
+            elif removed is not None and removed.priority == 1 and not dispatch.started:
+                # Before this fix, an update exactly at the stop boundary could
+                # remove the pending stop event and skip the notification. The
+                # dispatch window has elapsed, so the actor still needs the stop
+                # event that the timer would have sent.
+                await self._send_running_state_change(dispatch)
 
             if dispatch.started:
                 self._schedule_stop(dispatch)
@@ -507,22 +513,22 @@ def _update_timer(self, timer: Timer) -> None:
             timer.reset(interval=due_at - datetime.now(timezone.utc))
             _logger.debug("Next event scheduled at %s", self._scheduled_events[0].time)
 
-    def _remove_scheduled(self, dispatch: Dispatch) -> bool:
+    def _remove_scheduled(self, dispatch: Dispatch) -> QueueItem | None:
         """Remove a dispatch from the scheduled events.
 
         Args:
             dispatch: The dispatch to remove.
 
         Returns:
-            True if the dispatch was found and removed, False otherwise.
+            The removed queue item, or None if not found.
         """
         for idx, item in enumerate(self._scheduled_events):
             if dispatch.id == item.dispatch.id:
                 self._scheduled_events.pop(idx)
                 heapify(self._scheduled_events)
-                return True
+                return item
 
-        return False
+        return None
 
     def _schedule_start(self, dispatch: Dispatch) -> None:
         """Schedule a dispatch to start.

tests/test_frequenz_dispatch.py

@@ -907,6 +907,11 @@ async def test_parallel_dispatches_same_window_and_follow_up_window(
     This mirrors the SET_POWER edge case where multiple dispatches share the
     exact same start time and duration, and another batch starts exactly when
     the first batch ends.
+
+    This is an opportunistic coverage test: it also passed before the heap fix.
+    The broken heap ordering depends on runtime queue layouts that this test
+    harness does not reproduce reliably, but this still exercises the production
+    traffic shape around shared boundaries.
     """
     microgrid_id = MicrogridId(randint(1, 100))
     client = FakeClient()
@@ -993,6 +998,10 @@ async def test_parallel_dispatches_with_payload_updates_before_start(
     start time are created and then updated with new power values at different
     times before the start. Updates trigger _remove_scheduled + re-schedule
     which can corrupt the heap, potentially causing stop events to be lost.
+
+    This is also opportunistic coverage: it passed before the fix too. The
+    buggy heap state is timing- and layout-dependent, so we cannot force it
+    deterministically here, but the test keeps the update pattern covered.
     """
     microgrid_id = MicrogridId(randint(1, 100))
     client = FakeClient()
@@ -1096,3 +1105,65 @@ async def test_parallel_dispatches_with_payload_updates_before_start(
     assert {d.id for d in stop_events} == {d.id for d in start_events}
 
     await service.stop()
+
+
+async def test_dispatch_payload_update_at_stop_boundary(
+    fake_time: time_machine.Coordinates,
+    generator: DispatchGenerator,
+) -> None:
+    """Test that an update at the stop boundary still stops the actor.
+
+    Before the fix, this path could remove the pending stop event and skip the
+    notification, so the actor kept running even though the dispatch window had
+    already elapsed.
+
+    In practice this test is still opportunistic and passed before the fix as
+    well, because a payload change triggers an immediate update notification in
+    this harness. It still covers the boundary-update shape from production.
+    """
+    microgrid_id = MicrogridId(randint(1, 100))
+    client = FakeClient()
+    service = DispatchScheduler(microgrid_id=microgrid_id, client=client)
+    service.start()
+
+    receiver = await service.new_running_state_event_receiver(
+        "SET_POWER", merge_strategy=None
+    )
+    lifecycle = service.new_lifecycle_events_receiver("SET_POWER")
+
+    start_time = _now() + timedelta(seconds=5)
+    duration = timedelta(seconds=15)
+
+    dispatch = replace(
+        generator.generate_dispatch(),
+        active=True,
+        duration=duration,
+        start_time=start_time,
+        recurrence=RecurrenceRule(),
+        type="SET_POWER",
+        payload={"target_power_w": -88740.0},
+    )
+    await client.create(**to_create_params(microgrid_id, dispatch))
+    await lifecycle.receive()
+
+    fake_time.move_to(start_time + timedelta(seconds=1))
+    await asyncio.sleep(1)
+
+    started = await receiver.receive()
+    assert started.started
+
+    fake_time.move_to(start_time + duration)
+    await asyncio.sleep(0)
+
+    await client.update(
+        microgrid_id=microgrid_id,
+        dispatch_id=dispatch.id,
+        new_fields={"payload": {"target_power_w": -99999.0}},
+    )
+    fake_time.shift(timedelta(seconds=1))
+    await asyncio.sleep(1)
+
+    stop_event = await receiver.receive()
+    assert not stop_event.started
+
+    await service.stop()