Fix dispatch dict passed to MergeStrategy not getting updated for future fetch's (#174)

* We are also switching to use the same `now` while evaluating other
dispatches in the MergeStrategy, but this is just an enhancement and not
a fix
* We still used logging without "path" in some places, that was fixed

Author: Marenz

RELEASE_NOTES.md

@@ -6,15 +6,14 @@
 
 ## Upgrading
 
-The `frequenz.dispatch.TargetComponents` type was removed, use `frequenz.client.dispatch.TargetComponents` instead.
+<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
 
 ## New Features
 
-* The dispatcher offers two new parameters to control the client's call and stream timeout:
-  - `call_timeout`: The maximum time to wait for a response from the client.
-  - `stream_timeout`: The maximum time to wait before restarting a stream.
-* While the dispatch stream restarts we refresh our dispatch cache as well, to ensure we didn't miss any updates.
+<!-- Here goes the main new features and examples or instructions on how to use them -->
 
 ## Bug Fixes
 
-* Fixed that dispatches are never retried on failure, but instead an infinite loop of retry logs is triggered.
+* 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.

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
   # plugins.mkdocstrings.handlers.python.import)
   "frequenz-sdk >= 1.0.0-rc2100, < 1.0.0-rc2200",
   "frequenz-channels >= 1.6.1, < 2.0.0",
-  "frequenz-client-dispatch >= 0.11.0, < 0.12.0",
+  "frequenz-client-dispatch >= 0.11.1, < 0.12.0",
   "frequenz-client-common >= 0.3.2, < 0.4.0",
   "frequenz-client-base >= 0.11.0, < 0.12.0",
 ]

src/frequenz/dispatch/_actor_dispatcher.py

@@ -253,11 +253,13 @@ async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None:
         identity = self._dispatch_identity(stopping_dispatch)
 
         if actor_and_channel := self._actors.pop(identity, None):
+            _logger.info("Stopping actor for dispatch type %r", stopping_dispatch.type)
             await actor_and_channel.actor.stop(msg)
             await actor_and_channel.channel.close()
         else:
             _logger.warning(
-                "Actor for dispatch type %r is not running", stopping_dispatch.type
+                "Actor for dispatch type %r is not running, ignoring stop request",
+                stopping_dispatch.type,
             )
 
     async def _run(self) -> None:

src/frequenz/dispatch/_bg_service.py

@@ -337,7 +337,7 @@ async def _run(self) -> None:
                             pass
 
     async def _execute_scheduled_event(self, dispatch: Dispatch, timer: Timer) -> None:
-        """Execute a scheduled event.
+        """Execute a scheduled event and schedules the next one if any.
 
         Args:
             dispatch: The dispatch to execute.
@@ -366,8 +366,7 @@ async def _fetch(self, timer: Timer) -> None:
         """
         self._initial_fetch_event.clear()
 
-        old_dispatches = self._dispatches
-        self._dispatches = {}
+        new_dispatches = {}
 
         try:
             _logger.debug("Fetching dispatches for microgrid %s", self._microgrid_id)
@@ -381,9 +380,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 +395,40 @@ 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():
+        # We make a copy because we mutate self._dispatches.keys() inside the loop
+        for dispatch_id in frozenset(self._dispatches.keys() - new_dispatches.keys()):
+            # Use try/except as the `self._dispatches` cache can be mutated by
+            # stream delete events while we're iterating
+            try:
+                dispatch = self._dispatches.pop(dispatch_id)
+            except KeyError as error:
+                _logger.warning(
+                    "Inconsistency in cache detected. "
+                    + "Tried to delete non-existing dispatch %s (%s)",
+                    dispatch_id,
+                    error,
+                )
+                continue
+
             _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)
 
-            # Set deleted only here as it influences the result of dispatch.started
-            # which is used in above in _running_state_change
+            # Set deleted only here as it influences the result of
+            # dispatch.started, which is used in
+            # _update_dispatch_schedule_and_notify above.
             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(

src/frequenz/dispatch/_dispatch.py

@@ -43,10 +43,29 @@ def started(self) -> bool:
         Returns:
             True if the dispatch is started, False otherwise.
         """
+        now = datetime.now(tz=timezone.utc)
+        return self.started_at(now)
+
+    def started_at(self, now: datetime) -> bool:
+        """Check if the dispatch has started.
+
+        A dispatch is considered started if the current time is after the start
+        time but before the end time.
+
+        Recurring dispatches are considered started if the current time is after
+        the start time of the last occurrence but before the end time of the
+        last occurrence.
+
+        Args:
+            now: time to use as now
+
+        Returns:
+            True if the dispatch is started
+        """
         if self.deleted:
             return False
 
-        return super().started
+        return super().started_at(now)
 
     # noqa is needed because of a bug in pydoclint that makes it think a `return` without a return
     # value needs documenting

src/frequenz/dispatch/_merge_strategies.py

@@ -5,6 +5,7 @@
 
 import logging
 from collections.abc import Mapping
+from datetime import datetime, timezone
 from sys import maxsize
 from typing import Any
 
@@ -15,6 +16,8 @@
 from ._bg_service import MergeStrategy
 from ._dispatch import Dispatch
 
+_logger = logging.getLogger(__name__)
+
 
 def _hash_positive(args: Any) -> int:
     """Make a positive hash."""
@@ -39,24 +42,43 @@ def filter(
         Keeps stop events only if no other dispatches matching the
         strategy's criteria are running.
         """
-        if dispatch.started:
-            logging.debug("Keeping start event %s", dispatch.id)
+        now = datetime.now(tz=timezone.utc)
+
+        if dispatch.started_at(now):
+            _logger.debug("Keeping start event %s", dispatch.id)
             return True
 
-        other_dispatches_running = any(
-            existing_dispatch.started
+        running_dispatch_list = [
+            existing_dispatch
             for existing_dispatch in dispatches.values()
             if (
                 self.identity(existing_dispatch) == self.identity(dispatch)
                 and existing_dispatch.id != dispatch.id
             )
+        ]
+
+        other_dispatches_running = any(
+            running_dispatch.started_at(now)
+            for running_dispatch in running_dispatch_list
         )
 
-        logging.debug(
-            "stop event %s because other_dispatches_running=%s",
+        _logger.debug(
+            "%s stop event %s because other_dispatches_running=%s",
+            "Ignoring" if other_dispatches_running else "Allowing",
             dispatch.id,
             other_dispatches_running,
         )
+
+        if other_dispatches_running:
+            if _logger.isEnabledFor(logging.DEBUG):
+                _logger.debug(
+                    "Active other dispatches: %s",
+                    list(
+                        running_dispatch.id
+                        for running_dispatch in running_dispatch_list
+                    ),
+                )
+
         return not other_dispatches_running