Rewrite merging strategy design

Implement merge strategy based on TYPE+TARGET

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

Author: Marenz

Diff for: RELEASE_NOTES.md

@@ -13,7 +13,7 @@
 
 ## New Features
 
-* A new feature "unify running intervals" has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge & unify consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
+* A new feature "merger strategy" (`MergeByType`, `MergeByTypeTarget`) has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge & unify consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
 
 * The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1600.
 

Diff for: src/frequenz/dispatch/__init__.py

@@ -15,10 +15,12 @@
 
 """
 
+from ._bg_service import MergeStrategy
 from ._dispatch import Dispatch
 from ._dispatcher import Dispatcher
 from ._event import Created, Deleted, DispatchEvent, Updated
 from ._managing_actor import DispatchManagingActor, DispatchUpdate
+from ._merge_strategies import MergeByIdentity, MergeByType, MergeByTypeTarget, NoMerge
 
 __all__ = [
     "Created",
@@ -29,4 +31,9 @@
     "Dispatch",
     "DispatchManagingActor",
     "DispatchUpdate",
+    "MergeStrategy",
+    "MergeByIdentity",
+    "MergeByType",
+    "MergeByTypeTarget",
+    "NoMerge",
 ]

Diff for: src/frequenz/dispatch/_bg_service.py

@@ -3,8 +3,13 @@
 
 """The dispatch background service."""
 
+from __future__ import annotations
+
 import asyncio
+import functools
 import logging
+from abc import ABC, abstractmethod
+from collections.abc import ValuesView
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 from heapq import heappop, heappush
@@ -23,6 +28,22 @@
 """The logger for this module."""
 
 
+class MergeStrategy(ABC):
+    """Base class for strategies to merge running intervals."""
+
+    @abstractmethod
+    def filter(self, dispatches: dict[int, Dispatch], dispatch: Dispatch) -> bool:
+        """Filter dispatches based on the strategy.
+
+        Args:
+            dispatches: All dispatches, available as context.
+            dispatch: The dispatch to filter.
+
+        Returns:
+            True if the dispatch should be included, False otherwise.
+        """
+
+
 # pylint: disable=too-many-instance-attributes
 class DispatchScheduler(BackgroundService):
     """Dispatch background service.
@@ -104,6 +125,11 @@ def __init__(
         always at index 0.
         """
 
+    @property
+    def dispatches(self) -> ValuesView[Dispatch]:
+        """All currently cached dispatches."""
+        return self._dispatches.values()
+
     # pylint: disable=redefined-builtin
     def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         """Create a new receiver for lifecycle events.
@@ -119,54 +145,53 @@ def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         )
 
     async def new_running_state_event_receiver(
-        self, type: str, *, unify_running_intervals: bool = True
+        self, type: str, *, merge_strategy: MergeStrategy
     ) -> Receiver[Dispatch]:
         """Create a new receiver for running state events of the specified type.
 
-        If `unify_running_intervals` is True, running intervals from multiple
-        dispatches of the same type are considered as one continuous running
-        period. In this mode, any stop events are ignored as long as at least
-        one dispatch remains active.
+        `merge_strategy` is an instance of a class derived from
+        [`MergeStrategy`][frequenz.dispatch.MergeStrategy]. Available strategies
+        are:
+
+        * [`MergeByType`][frequenz.dispatch.MergeByType] — merges all dispatches
+          of the same type
+        * [`MergeByTypeTarget`][frequenz.dispatch.MergeByTypeTarget] — merges all
+          dispatches of the same type and target
+        * [`NoMerge`][frequenz.dispatch.NoMerge] — no merging, just send all
+          events
+
+        You can make your own strategy by subclassing
+
+        * [`MergeByIdentity`][frequenz.dispatch.MergeByIdentity] — Merges
+          dispatches based on a user defined identity function
+        * [`MergeStrategy`][frequenz.dispatch.MergeStrategy] — Merges based
+          on a user defined filter function
+
+        Running intervals from multiple dispatches will be merged, according to
+        the chosen strategy.
+
+        While merging, stop events are ignored as long as at least one
+        merge-criteria-matching dispatch remains active.
 
         Args:
             type: The type of events to receive.
-            unify_running_intervals: Whether to unify running intervals.
-
+            merge_strategy: The merge strategy to use.
         Returns:
             A new receiver for running state status.
         """
-        # Find all matching dispatches based on the type and collect them
         dispatches = [
             dispatch for dispatch in self._dispatches.values() if dispatch.type == type
         ]
 
-        # Create receiver with enough capacity to hold all matching dispatches
         receiver = self._running_state_status_channel.new_receiver(
-            limit=max(1, len(dispatches))
+            limit=max(30, len(dispatches))
         ).filter(lambda dispatch: dispatch.type == type)
 
-        if unify_running_intervals:
-
-            def _is_type_still_running(new_dispatch: Dispatch) -> bool:
-                """Merge time windows of running dispatches.
-
-                Any event that would cause a stop is filtered if at least one
-                dispatch of the same type is running.
-                """
-                if new_dispatch.started:
-                    return True
-
-                other_dispatches_running = any(
-                    dispatch.started
-                    for dispatch in self._dispatches.values()
-                    if dispatch.type == type
-                )
-                # If no other dispatches are running, we can allow the stop event
-                return not other_dispatches_running
-
-            receiver = receiver.filter(_is_type_still_running)
+        if merge_strategy:
+            receiver = receiver.filter(
+                functools.partial(merge_strategy.filter, self._dispatches)
+            )
 
-        # Send all matching dispatches to the receiver
         for dispatch in dispatches:
             await self._send_running_state_change(dispatch)
 

Diff for: src/frequenz/dispatch/_dispatcher.py

@@ -7,26 +7,29 @@
 from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
 
-from ._bg_service import DispatchScheduler
+from ._bg_service import DispatchScheduler, MergeStrategy
 from ._dispatch import Dispatch
 from ._event import DispatchEvent
+from ._merge_strategies import NoMerge
 
 
 class Dispatcher:
     """A highlevel interface for the dispatch API.
 
     This class provides a highlevel interface to the dispatch API.
-    It provides two channels:
+    It provides two receiver functions:
 
-    Lifecycle events:
-        A channel that sends a dispatch event message whenever a dispatch
-        is created, updated or deleted.
+    * [Lifecycle events receiver][frequenz.dispatch.Dispatcher.new_lifecycle_events_receiver]:
+        Receives an event whenever a dispatch is created, updated or deleted.
+    * [Running status change
+        receiver][frequenz.dispatch.Dispatcher.new_running_state_event_receiver]:
+        Receives an event whenever the running status of a dispatch changes.
+        The running status of a dispatch can change due to a variety of reasons,
+        such as but not limited to the dispatch being started, stopped, modified
+        or deleted or reaching its scheduled start or end time.
 
-    Running status change:
-        Sends a dispatch message whenever a dispatch is ready
-        to be executed according to the schedule or the running status of the
-        dispatch changed in a way that could potentially require the consumer to start,
-        stop or reconfigure itself.
+        Any change that could potentially require the consumer to start, stop or
+        reconfigure itself will cause a message to be sent.
 
     Example: Processing running state change dispatches
         ```python
@@ -200,7 +203,10 @@ def new_lifecycle_events_receiver(
         return self._bg_service.new_lifecycle_events_receiver(dispatch_type)
 
     async def new_running_state_event_receiver(
-        self, dispatch_type: str, *, unify_running_intervals: bool = True
+        self,
+        dispatch_type: str,
+        *,
+        merge_strategy: MergeStrategy = NoMerge(),
     ) -> Receiver[Dispatch]:
         """Return running state event receiver.
 
@@ -228,18 +234,30 @@ async def new_running_state_event_receiver(
          - The payload changed
          - The dispatch was deleted
 
-        If `unify_running_intervals` is True, running intervals from multiple
-        dispatches of the same type are considered as one continuous running
-        period. In this mode, any stop events are ignored as long as at least
-        one dispatch remains active.
+        `merge_strategy` is an instance of a class derived from
+        [`MergeStrategy`][frequenz.dispatch.MergeStrategy] Available strategies
+        are:
+
+        * [`MergeByType`][frequenz.dispatch.MergeByType] — merges all dispatches
+          of the same type
+        * [`MergeByTypeTarget`][frequenz.dispatch.MergeByTypeTarget] — merges all
+          dispatches of the same type and target
+        * [`NoMerge`][frequenz.dispatch.NoMerge] — no merging, just send all
+          events (default)
+
+        Running intervals from multiple dispatches will be merged, according to
+        the chosen strategy.
+
+        While merging, stop events are ignored as long as at least one
+        merge-criteria-matching dispatch remains active.
 
         Args:
             dispatch_type: The type of the dispatch to listen for.
-            unify_running_intervals: Whether to unify running intervals.
+            merge_strategy: The type of the strategy to merge running intervals.
 
         Returns:
             A new receiver for dispatches whose running status changed.
         """
         return await self._bg_service.new_running_state_event_receiver(
-            dispatch_type, unify_running_intervals=unify_running_intervals
+            dispatch_type, merge_strategy=merge_strategy
         )

Diff for: src/frequenz/dispatch/_merge_strategies.py

@@ -0,0 +1,75 @@
+# License: MIT
+# Copyright © 2025 Frequenz Energy-as-a-Service GmbH
+
+"""Different merge strategies for dispatch running state events."""
+
+import logging
+from abc import abstractmethod
+
+from typing_extensions import override
+
+from ._bg_service import MergeStrategy
+from ._dispatch import Dispatch
+
+
+class MergeByIdentity(MergeStrategy):
+    """Merge running intervals based on a dispatch configuration."""
+
+    @abstractmethod
+    def identity(self, dispatch: Dispatch) -> int:
+        """Identity function for the merge criteria."""
+
+    @override
+    def filter(self, dispatches: dict[int, Dispatch], dispatch: Dispatch) -> bool:
+        """Filter dispatches based on the merge strategy.
+
+        Keeps start events.
+        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)
+            return True
+
+        other_dispatches_running = any(
+            existing_dispatch.started
+            for existing_dispatch in dispatches.values()
+            if (
+                self.identity(existing_dispatch) == self.identity(dispatch)
+                and existing_dispatch.id != dispatch.id
+            )
+        )
+
+        logging.debug(
+            "stop event %s because other_dispatches_running=%s",
+            dispatch.id,
+            other_dispatches_running,
+        )
+        return not other_dispatches_running
+
+
+class MergeByType(MergeByIdentity):
+    """Merge running intervals based on the dispatch type."""
+
+    @override
+    def identity(self, dispatch: Dispatch) -> int:
+        """Identity function for the merge criteria."""
+        return hash(dispatch.type)
+
+
+class MergeByTypeTarget(MergeByType):
+    """Merge running intervals based on the dispatch type and target."""
+
+    @override
+    def identity(self, dispatch: Dispatch) -> int:
+        """Identity function for the merge criteria."""
+        return hash((dispatch.type, dispatch.target))
+
+
+class NoMerge(MergeByIdentity):
+    """Treat each dispatch individually. No merging."""
+
+    @override
+    def identity(self, dispatch: Dispatch) -> int:
+        """Identity function for the merge criteria."""
+        return dispatch.id