Use the same now for all in merge check

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

Author: Marenz

RELEASE_NOTES.md

@@ -15,3 +15,4 @@
 ## Bug Fixes
 
 * 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.

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/_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.
+
+        Params:
+            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
 
@@ -41,12 +42,14 @@ def filter(
         Keeps stop events only if no other dispatches matching the
         strategy's criteria are running.
         """
-        if dispatch.started:
+        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
+            existing_dispatch.started_at(now)
             for existing_dispatch in dispatches.values()
             if (
                 self.identity(existing_dispatch) == self.identity(dispatch)