Use MicrogridId, DispatchId and implement DispatchActorId

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

Author: Marenz

src/frequenz/dispatch/_actor_dispatcher.py

@@ -13,7 +13,9 @@
 from frequenz.channels import Broadcast, Receiver, Sender, select
 from frequenz.channels.timer import SkipMissedAndDrift, Timer
 from frequenz.client.common.microgrid.components import ComponentCategory, ComponentId
+from frequenz.client.dispatch.types import DispatchId
 from frequenz.client.dispatch.types import TargetComponents as ClientTargetComponents
+from frequenz.core.id import BaseId
 from frequenz.sdk.actor import Actor, BackgroundService
 
 from ._dispatch import Dispatch
@@ -27,6 +29,18 @@
 """
 
 
+class DispatchActorId(BaseId, str_prefix="DA"):
+    """ID for a dispatch actor."""
+
+    def __init__(self, dispatch_id: DispatchId | int) -> None:
+        """Initialize the DispatchActorId.
+
+        Args:
+            dispatch_id: The ID of the dispatch this actor is associated with.
+        """
+        super().__init__(int(dispatch_id))
+
+
 @dataclass(frozen=True, kw_only=True)
 class DispatchInfo:
     """Event emitted when the dispatch changes."""
@@ -162,7 +176,7 @@ def __init__(  # pylint: disable=too-many-arguments, too-many-positional-argumen
             [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
         ],
         running_status_receiver: Receiver[Dispatch],
-        dispatch_identity: Callable[[Dispatch], int] | None = None,
+        dispatch_identity: Callable[[Dispatch], DispatchActorId] | None = None,
         retry_interval: timedelta = timedelta(seconds=60),
     ) -> None:
         """Initialize the dispatch handler.
@@ -176,15 +190,15 @@ def __init__(  # pylint: disable=too-many-arguments, too-many-positional-argumen
             retry_interval: How long to wait until trying to start failed actors again.
         """
         super().__init__()
-        self._dispatch_identity: Callable[[Dispatch], int] = (
-            dispatch_identity if dispatch_identity else lambda d: d.id
+        self._dispatch_identity: Callable[[Dispatch], DispatchActorId] = (
+            dispatch_identity if dispatch_identity else lambda d: DispatchActorId(d.id)
         )
 
         self._dispatch_rx = running_status_receiver
         self._retry_timer_rx = Timer(retry_interval, SkipMissedAndDrift())
         self._actor_factory = actor_factory
-        self._actors: dict[int, ActorDispatcher.ActorAndChannel] = {}
-        self._failed_dispatches: dict[int, Dispatch] = {}
+        self._actors: dict[DispatchActorId, ActorDispatcher.ActorAndChannel] = {}
+        self._failed_dispatches: dict[DispatchActorId, Dispatch] = {}
         """Failed dispatches that will be retried later."""
 
     def start(self) -> None:

src/frequenz/dispatch/_bg_service.py

@@ -23,11 +23,13 @@
     StreamRetrying,
     StreamStarted,
 )
+from frequenz.client.common.microgrid import MicrogridId
 from frequenz.client.dispatch import DispatchApiClient
 from frequenz.client.dispatch.types import DispatchEvent as ApiDispatchEvent
-from frequenz.client.dispatch.types import Event
+from frequenz.client.dispatch.types import DispatchId, Event
 from frequenz.sdk.actor import BackgroundService
 
+from ._actor_dispatcher import DispatchActorId
 from ._dispatch import Dispatch
 from ._event import Created, Deleted, DispatchEvent, Updated
 
@@ -39,11 +41,13 @@ class MergeStrategy(ABC):
     """Base class for strategies to merge running intervals."""
 
     @abstractmethod
-    def identity(self, dispatch: Dispatch) -> int:
+    def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
 
     @abstractmethod
-    def filter(self, dispatches: Mapping[int, Dispatch], dispatch: Dispatch) -> bool:
+    def filter(
+        self, dispatches: Mapping[DispatchId, Dispatch], dispatch: Dispatch
+    ) -> bool:
         """Filter dispatches based on the strategy.
 
         Args:
@@ -81,7 +85,7 @@ class QueueItem:
         to consider the start event when deciding whether to execute the
         stop event.
         """
-        dispatch_id: int
+        dispatch_id: DispatchId
         dispatch: Dispatch = field(compare=False)
 
         def __init__(
@@ -96,7 +100,7 @@ def __init__(
     # pylint: disable=too-many-arguments
     def __init__(
         self,
-        microgrid_id: int,
+        microgrid_id: MicrogridId,
         client: DispatchApiClient,
     ) -> None:
         """Initialize the background service.
@@ -108,7 +112,7 @@ def __init__(
         super().__init__(name="dispatch")
 
         self._client = client
-        self._dispatches: dict[int, Dispatch] = {}
+        self._dispatches: dict[DispatchId, Dispatch] = {}
         self._microgrid_id = microgrid_id
 
         self._lifecycle_events_channel = Broadcast[DispatchEvent](

src/frequenz/dispatch/_dispatcher.py

@@ -12,11 +12,12 @@
 from typing import Awaitable, Callable, Self
 
 from frequenz.channels import Receiver
+from frequenz.client.common.microgrid import MicrogridId
 from frequenz.client.dispatch import DispatchApiClient
 from frequenz.sdk.actor import Actor, BackgroundService
 from typing_extensions import override
 
-from ._actor_dispatcher import ActorDispatcher, DispatchInfo
+from ._actor_dispatcher import ActorDispatcher, DispatchActorId, DispatchInfo
 from ._bg_service import DispatchScheduler, MergeStrategy
 from ._dispatch import Dispatch
 from ._event import DispatchEvent
@@ -202,7 +203,7 @@ async def run():
     def __init__(
         self,
         *,
-        microgrid_id: int,
+        microgrid_id: MicrogridId,
         server_url: str,
         key: str,
         call_timeout: timedelta = timedelta(seconds=60),
@@ -328,8 +329,8 @@ async def start_managing(
 
         self._empty_event.clear()
 
-        def id_identity(dispatch: Dispatch) -> int:
-            return dispatch.id
+        def id_identity(dispatch: Dispatch) -> DispatchActorId:
+            return DispatchActorId(dispatch.id)
 
         dispatcher = ActorDispatcher(
             actor_factory=actor_factory,

src/frequenz/dispatch/_merge_strategies.py

@@ -5,23 +5,34 @@
 
 import logging
 from collections.abc import Mapping
+from sys import maxsize
+from typing import Any
 
+from frequenz.client.dispatch.types import DispatchId
 from typing_extensions import override
 
+from ._actor_dispatcher import DispatchActorId
 from ._bg_service import MergeStrategy
 from ._dispatch import Dispatch
 
 
+def _hash_positive(args: Any) -> int:
+    """Make a positive hash."""
+    return hash(args) + maxsize + 1
+
+
 class MergeByType(MergeStrategy):
     """Merge running intervals based on the dispatch type."""
 
     @override
-    def identity(self, dispatch: Dispatch) -> int:
+    def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
-        return hash(dispatch.type)
+        return DispatchActorId(_hash_positive(dispatch.type))
 
     @override
-    def filter(self, dispatches: Mapping[int, Dispatch], dispatch: Dispatch) -> bool:
+    def filter(
+        self, dispatches: Mapping[DispatchId, Dispatch], dispatch: Dispatch
+    ) -> bool:
         """Filter dispatches based on the merge strategy.
 
         Keeps start events.
@@ -53,6 +64,6 @@ class MergeByTypeTarget(MergeByType):
     """Merge running intervals based on the dispatch type and target."""
 
     @override
-    def identity(self, dispatch: Dispatch) -> int:
+    def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
-        return hash((dispatch.type, tuple(dispatch.target)))
+        return DispatchActorId(_hash_positive((dispatch.type, tuple(dispatch.target))))

tests/test_frequenz_dispatch.py

@@ -13,6 +13,7 @@
 import pytest
 import time_machine
 from frequenz.channels import Receiver
+from frequenz.client.common.microgrid import MicrogridId
 from frequenz.client.dispatch.recurrence import Frequency, RecurrenceRule
 from frequenz.client.dispatch.test.client import FakeClient, to_create_params
 from frequenz.client.dispatch.test.generator import DispatchGenerator
@@ -67,14 +68,14 @@ class _TestEnv:
     """The receiver for ready dispatches."""
     client: FakeClient
     """The fake client for the actor."""
-    microgrid_id: int
+    microgrid_id: MicrogridId
     """The microgrid id."""
 
 
 @fixture
 async def test_env() -> AsyncIterator[_TestEnv]:
     """Return an actor test environment."""
-    microgrid_id = randint(1, 100)
+    microgrid_id = MicrogridId(randint(1, 100))
     client = FakeClient()
 
     service = DispatchScheduler(
@@ -475,11 +476,10 @@ async def test_dispatch_new_but_finished(
     )
     fake_time.shift(timedelta(seconds=1))
 
+    await asyncio.sleep(1)
     # Process the lifecycle event caused by the old dispatch at startup
     await test_env.lifecycle_events.receive()
 
-    await asyncio.sleep(1)
-
     # Create another dispatch the normal way
     new_dispatch = generator.generate_dispatch()
     new_dispatch = replace(
@@ -551,7 +551,7 @@ async def test_multiple_dispatches_merge_running_intervals(
     merge_strategy: MergeStrategy,
 ) -> None:
     """Test that multiple dispatches are merged into a single running interval."""
-    microgrid_id = randint(1, 100)
+    microgrid_id = MicrogridId(randint(1, 100))
     client = FakeClient()
     service = DispatchScheduler(
         microgrid_id=microgrid_id,
@@ -633,7 +633,7 @@ async def test_multiple_dispatches_sequential_intervals_merge(
     Even if dispatches don't overlap but are consecutive,
     merge_running_intervals=TPYE should treat them as continuous if any event tries to stop.
     """
-    microgrid_id = randint(1, 100)
+    microgrid_id = MicrogridId(randint(1, 100))
     client = FakeClient()
     service = DispatchScheduler(microgrid_id=microgrid_id, client=client)
     service.start()
@@ -699,7 +699,7 @@ async def test_at_least_one_running_filter(
     merge_strategy: MergeStrategy,
 ) -> None:
     """Test scenarios directly tied to the _at_least_one_running logic."""
-    microgrid_id = randint(1, 100)
+    microgrid_id = MicrogridId(randint(1, 100))
     client = FakeClient()
     service = DispatchScheduler(microgrid_id=microgrid_id, client=client)
     service.start()