Implement merge strategy based on TYPE+TARGET

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

Author: Marenz

src/frequenz/dispatch/__init__.py

@@ -15,6 +15,7 @@
 
 """
 
+from ._bg_service import UnifyStrategy
 from ._dispatch import Dispatch
 from ._dispatcher import Dispatcher
 from ._event import Created, Deleted, DispatchEvent, Updated
@@ -29,4 +30,5 @@
     "Dispatch",
     "DispatchManagingActor",
     "DispatchUpdate",
+    "UnifyStrategy",
 ]

src/frequenz/dispatch/_bg_service.py

@@ -7,7 +7,9 @@
 import logging
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
+from enum import Enum, auto
 from heapq import heappop, heappush
+from typing import Callable
 
 import grpc.aio
 from frequenz.channels import Broadcast, Receiver, select, selected_from
@@ -23,6 +25,16 @@
 """The logger for this module."""
 
 
+class UnifyStrategy(Enum):
+    """The strategy to unify running intervals."""
+
+    TYPE = auto()
+    """Unify running intervals based on the dispatch type."""
+
+    TYPE_TARGET = auto()
+    """Unify running intervals based on the dispatch type and target."""
+
+
 # pylint: disable=too-many-instance-attributes
 class DispatchScheduler(BackgroundService):
     """Dispatch background service.
@@ -119,54 +131,87 @@ 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, *, unify_running_intervals: UnifyStrategy | None = None
     ) -> 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.
+        If `unify_running_intervals` is set, running intervals from multiple
+        dispatches of the same type/type&target (depending on the chosen
+        strategy) are considered as one continuous running
+        period.
+        In this mode, stop events are ignored as long as at least one (criteria
+        matching) dispatch remains active.
 
         Args:
             type: The type of events to receive.
-            unify_running_intervals: Whether to unify running intervals.
-
+            unify_running_intervals: The strategy to unify running intervals.
         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))
         ).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.
+            def is_still_running_filter(
+                unify_strategy: UnifyStrategy,
+            ) -> Callable[[Dispatch], bool]:
+                """Create a filter function based on the provided UnifyStrategy.
+
+                Args:
+                    unify_strategy: The strategy to use for unifying running intervals.
 
-                Any event that would cause a stop is filtered if at least one
-                dispatch of the same type is running.
+                Returns:
+                    A callable that takes a Dispatch and returns True if it
+                    should be kept, False otherwise.
                 """
-                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
+                def filter_func(new_dispatch: Dispatch) -> bool:
+                    """Filter dispatches based on the provided unification strategy.
+
+                    Keeps start events.
+                    Keeps stop events only if no other dispatches matching the
+                    strategy's criteria are running.
+                    """
+                    if new_dispatch.started:
+                        return True
+
+                    # Define the criteria for checking other running dispatches
+                    # based on strategy
+                    if unify_strategy == UnifyStrategy.TYPE:
+                        # criteria = lambda dispatch: dispatch.type == type
+                        def criteria(dispatch: Dispatch) -> bool:
+                            return dispatch.type == type
+
+                    elif unify_strategy == UnifyStrategy.TYPE_TARGET:
+
+                        def criteria(dispatch: Dispatch) -> bool:
+                            return (
+                                dispatch.type == type
+                                and dispatch.target == new_dispatch.target
+                            )
+
+                    else:  # Default case (should not normally happen)
+                        assert False, f"Unknown unify strategy: {unify_strategy}"
+
+                    other_dispatches_running = any(
+                        dispatch.started
+                        for dispatch in self._dispatches.values()
+                        if criteria(dispatch)
+                    )
+
+                    return not other_dispatches_running
+
+                return filter_func
 
-            receiver = receiver.filter(_is_type_still_running)
+            receiver = receiver.filter(is_still_running_filter(unify_running_intervals))
+        # End of unify_running_intervals block
 
-        # Send all matching dispatches to the receiver
         for dispatch in dispatches:
             await self._send_running_state_change(dispatch)
 

src/frequenz/dispatch/_dispatcher.py

@@ -7,7 +7,7 @@
 from frequenz.channels import Receiver
 from frequenz.client.dispatch import Client
 
-from ._bg_service import DispatchScheduler
+from ._bg_service import DispatchScheduler, UnifyStrategy
 from ._dispatch import Dispatch
 from ._event import DispatchEvent
 
@@ -200,7 +200,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,
+        *,
+        unify_running_intervals: UnifyStrategy | None = None,
     ) -> Receiver[Dispatch]:
         """Return running state event receiver.
 
@@ -228,14 +231,16 @@ 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.
+        If `unify_running_intervals` is set, running intervals from multiple
+        dispatches of the same type/type&target (depending on the chosen
+        strategy) are considered as one continuous running
+        period.
+        In this mode, stop events are ignored as long as at least one (criteria
+        matching) dispatch remains active.
 
         Args:
             dispatch_type: The type of the dispatch to listen for.
-            unify_running_intervals: Whether to unify running intervals.
+            unify_running_intervals: The strategy to unify running intervals.
 
         Returns:
             A new receiver for dispatches whose running status changed.

tests/test_frequenz_dispatch.py

@@ -18,7 +18,14 @@
 from frequenz.client.dispatch.types import Dispatch as BaseDispatch
 from pytest import fixture
 
-from frequenz.dispatch import Created, Deleted, Dispatch, DispatchEvent, Updated
+from frequenz.dispatch import (
+    Created,
+    Deleted,
+    Dispatch,
+    DispatchEvent,
+    UnifyStrategy,
+    Updated,
+)
 from frequenz.dispatch._bg_service import DispatchScheduler
 
 
@@ -77,7 +84,7 @@ async def test_env() -> AsyncIterator[TestEnv]:
             service=service,
             lifecycle_events=service.new_lifecycle_events_receiver("TEST_TYPE"),
             running_state_change=await service.new_running_state_event_receiver(
-                "TEST_TYPE", unify_running_intervals=False
+                "TEST_TYPE", unify_running_intervals=None
             ),
             client=client,
             microgrid_id=microgrid_id,
@@ -520,7 +527,7 @@ async def test_multiple_dispatches_unify_running_intervals(
     service.start()
 
     receiver = await service.new_running_state_event_receiver(
-        "TEST_TYPE", unify_running_intervals=True
+        "TEST_TYPE", unify_running_intervals=UnifyStrategy.TYPE
     )
 
     # Create two overlapping dispatches
@@ -559,7 +566,7 @@ async def test_multiple_dispatches_unify_running_intervals(
     assert started1.started
     assert started2.started
 
-    # Stop dispatch2 first, but unify_running_intervals=True means as long as dispatch1 runs,
+    # Stop dispatch2 first, but unify_running_intervals=TYPE means as long as dispatch1 runs,
     # we do not send a stop event
     await client.update(
         microgrid_id=microgrid_id, dispatch_id=started2.id, new_fields={"active": False}
@@ -585,15 +592,15 @@ async def test_multiple_dispatches_sequential_intervals_unify(
     """Test that multiple dispatches are merged into a single running interval.
 
     Even if dispatches don't overlap but are consecutive,
-    unify_running_intervals=True should treat them as continuous if any event tries to stop.
+    unify_running_intervals=TPYE should treat them as continuous if any event tries to stop.
     """
     microgrid_id = randint(1, 100)
     client = FakeClient()
     service = DispatchScheduler(microgrid_id=microgrid_id, client=client)
     service.start()
 
     receiver = await service.new_running_state_event_receiver(
-        "TEST_TYPE", unify_running_intervals=True
+        "TEST_TYPE", unify_running_intervals=UnifyStrategy.TYPE
     )
 
     dispatch1 = replace(
@@ -653,9 +660,9 @@ async def test_at_least_one_running_filter(
     service = DispatchScheduler(microgrid_id=microgrid_id, client=client)
     service.start()
 
-    # unify_running_intervals is True, so we use merged intervals
+    # unify_running_intervals is TYPE, so we use merged intervals
     receiver = await service.new_running_state_event_receiver(
-        "TEST_TYPE", unify_running_intervals=True
+        "TEST_TYPE", unify_running_intervals=UnifyStrategy.TYPE
     )
 
     # Single dispatch that starts and stops normally