Consider dry_run status in dispatch merging

Dispatches with different `dry_run` values are now handled by separate actor instances.

This change modifies the `MergeByType` and `MergeByTypeTarget` merge strategies to include the `dry_run` status in the identity function. Previously, two dispatches with the same type and target, but different `dry_run` statuses, would be merged into a single actor instance. This could lead to unexpected behavior, as dry-run dispatches are intended for testing and should not affect production actors.

With this change, dispatches are only merged if they have the same type, target (for `MergeByTypeTarget`), and `dry_run` status.

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

Author: Marenz

RELEASE_NOTES.md

@@ -10,6 +10,7 @@
 
 ## New Features
 
+* `dry_run` status is now considered when merging dispatches. Dispatches with different `dry_run` values will no longer be merged, ensuring that dry-run and operational dispatches are handled by separate actors.
 * Two new parameters were added to the `Dispatcher` constructor:
   * `sign_secret`: A secret key used for signing messages.
   * `auth_key`: An authentication key for the Dispatch API.

src/frequenz/dispatch/_dispatcher.py

@@ -333,11 +333,11 @@ async def start_managing(
         This also decides how instances are mapped from dispatches to actors:
 
         * [`MergeByType`][frequenz.dispatch.MergeByType] — All dispatches map to
-        one single instance identified by the dispatch type.
+        one single instance identified by the dispatch type and dry_run status.
         * [`MergeByTypeTarget`][frequenz.dispatch.MergeByTypeTarget] — A
-        dispatch maps to an instance identified by the dispatch type and target.
-        So different dispatches with equal type and target will map to the same
-        instance.
+        dispatch maps to an instance identified by the dispatch type, dry_run status
+        and target. So different dispatches with equal type and target will map to
+        the same instance.
         * `None` — No merging, each dispatch maps to a separate instance.
 
         Args:

src/frequenz/dispatch/_merge_strategies.py

@@ -30,7 +30,7 @@ class MergeByType(MergeStrategy):
     @override
     def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
-        return DispatchActorId(_hash_positive(dispatch.type))
+        return DispatchActorId(_hash_positive((dispatch.type, dispatch.dry_run)))
 
     @override
     def filter(
@@ -88,4 +88,6 @@ class MergeByTypeTarget(MergeByType):
     @override
     def identity(self, dispatch: Dispatch) -> DispatchActorId:
         """Identity function for the merge criteria."""
-        return DispatchActorId(_hash_positive((dispatch.type, tuple(dispatch.target))))
+        return DispatchActorId(
+            _hash_positive((dispatch.type, dispatch.dry_run, tuple(dispatch.target)))
+        )

tests/test_frequenz_dispatch.py

@@ -781,3 +781,78 @@ async def test_at_least_one_running_filter(
     await asyncio.sleep(1)
     stopped_b = await receiver.receive()
     assert not stopped_b.started
+
+
+@pytest.mark.parametrize(
+    "merge_strategy",
+    [
+        MergeByType(),
+        MergeByTypeTarget(),
+    ],
+)
+async def test_dry_run_dispatches_not_merged(
+    fake_time: time_machine.Coordinates,
+    generator: DispatchGenerator,
+    merge_strategy: MergeStrategy,
+) -> None:
+    """Test that dispatches with different dry_run values are not merged."""
+    microgrid_id = MicrogridId(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", merge_strategy=merge_strategy
+    )
+
+    # Create two dispatches with same type and target, but different dry_run
+    dispatch1 = replace(
+        generator.generate_dispatch(),
+        active=True,
+        duration=timedelta(seconds=10),
+        target=TargetIds(1, 2),
+        start_time=_now() + timedelta(seconds=5),
+        recurrence=RecurrenceRule(),
+        type="TEST_TYPE",
+        dry_run=False,
+    )
+    dispatch2 = replace(
+        dispatch1,
+        dry_run=True,
+    )
+
+    lifecycle_events = service.new_lifecycle_events_receiver("TEST_TYPE")
+
+    await client.create(**to_create_params(microgrid_id, dispatch1))
+    await client.create(**to_create_params(microgrid_id, dispatch2))
+
+    # Wait for both to be registered
+    await lifecycle_events.receive()
+    await lifecycle_events.receive()
+
+    # Move time forward to start both dispatches
+    fake_time.shift(timedelta(seconds=6))
+    await asyncio.sleep(1)
+
+    started1 = await receiver.receive()
+    started2 = await receiver.receive()
+
+    assert started1.started
+    assert started2.started
+    assert started1.dry_run != started2.dry_run
+
+    # Move time forward to the end of the dispatches
+    fake_time.shift(timedelta(seconds=10))
+    await asyncio.sleep(1)
+
+    stopped1 = await receiver.receive()
+    stopped2 = await receiver.receive()
+
+    assert not stopped1.started
+    assert not stopped2.started
+    assert stopped1.dry_run != stopped2.dry_run
+
+    await service.stop()