Fix filter not seeing the latest dispatches after periodic fetch()'s

By using a workflow that avoids reassigning the dispatches dict.

When `fetch()`, which runs periodically, would assign
a fresh map to `self._dispatches` the filter would
not see the latest dispatches, as it would still
reference the old map.

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

Author: Marenz

RELEASE_NOTES.md

@@ -16,3 +16,4 @@
 
 * The merge by type class now uses the correct logger path.
 * The merge by type was made more robust under heavy load, making sure to use the same `now` for all dispatches that are checked.
+* Fix that the merge filter was using an outdated dispatches dict once fetch() ran.

src/frequenz/dispatch/_bg_service.py

@@ -366,8 +366,8 @@ async def _fetch(self, timer: Timer) -> None:
         """
         self._initial_fetch_event.clear()
 
-        old_dispatches = self._dispatches
-        self._dispatches = {}
+        old_dispatches = set(self._dispatches.keys())
+        new_dispatches = {}
 
         try:
             _logger.debug("Fetching dispatches for microgrid %s", self._microgrid_id)
@@ -381,9 +381,9 @@ async def _fetch(self, timer: Timer) -> None:
                         continue
                     dispatch = Dispatch(client_dispatch)
 
-                    self._dispatches[dispatch.id] = dispatch
-                    old_dispatch = old_dispatches.pop(dispatch.id, None)
-                    if not old_dispatch:
+                    new_dispatches[dispatch.id] = dispatch
+                    old_dispatch = self._dispatches.get(dispatch.id, None)
+                    if old_dispatch is None:
                         _logger.debug("New dispatch: %s", dispatch)
                         await self._update_dispatch_schedule_and_notify(
                             dispatch, None, timer
@@ -396,23 +396,33 @@ async def _fetch(self, timer: Timer) -> None:
                         )
                         await self._lifecycle_events_tx.send(Updated(dispatch=dispatch))
 
-            _logger.debug("Received %s dispatches", len(self._dispatches))
+            _logger.debug("Received %s dispatches", len(new_dispatches))
 
         except grpc.aio.AioRpcError as error:
             _logger.error("Error fetching dispatches: %s", error)
-            self._dispatches = old_dispatches
             return
 
-        for dispatch in old_dispatches.values():
-            _logger.debug("Deleted dispatch: %s", dispatch)
-            await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
-            await self._update_dispatch_schedule_and_notify(None, dispatch, timer)
+        # Delete old dispatches
+        for dispatch_id in old_dispatches - new_dispatches.keys():
+            try:
+                dispatch = self._dispatches.pop(dispatch_id)
+                _logger.debug("Deleted dispatch: %s", dispatch)
+                await self._update_dispatch_schedule_and_notify(None, dispatch, timer)
+
+                # Set deleted only here as it influences the result of dispatch.started
+                # which is used in the func call above ^
+                dispatch._set_deleted()  # pylint: disable=protected-access
+                await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
+            except KeyError:
+                _logger.warning(
+                    "Inconsistency in cache detected. Tried to delete non-existing dispatch %s",
+                    dispatch_id,
+                )
 
-            # Set deleted only here as it influences the result of dispatch.started
-            # which is used in above in _running_state_change
-            dispatch._set_deleted()  # pylint: disable=protected-access
-            await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
+        # Update the dispatch list with the dispatches
+        self._dispatches.update(new_dispatches)
 
+        # Set event to indicate fetch ran at least once
         self._initial_fetch_event.set()
 
     async def _update_dispatch_schedule_and_notify(