Stop actors when dispatch merge identities change

Remember each running dispatch's routed identity. If an update changes it, stop the old actor unless another running dispatch still uses it.

Use the merge strategy's existing filter for managed actors and clear stale retries.

Signed-off-by: Mathias L. Baumann <mathias.baumann@frequenz.com>

Author: Marenz

RELEASE_NOTES.md

@@ -1,10 +1,17 @@
 # Dispatch Highlevel Interface Release Notes
 
-## Enhancements
+## Summary
 
-- Improved docstring documentation across the project.
+<!-- Here goes a general summary of what this release is about -->
+
+## Upgrading
+
+<!-- Here goes notes on how to upgrade from previous versions, including deprecations and what they should be replaced with -->
+
+## New Features
+
+<!-- Here goes the main new features and examples or instructions on how to use them -->
 
 ## Bug Fixes
 
-* Fixed documentation cross-references to dependencies by updating the mkdocstrings inventories to match the minimum dependency versions and adding the missing `frequenz-client-base` inventory.
-* Declared the missing direct dependencies `frequenz-client-common`, `frequenz-core` and `grpcio`, which were previously only pulled in indirectly through `frequenz-client-dispatch`.
+* Fixed actors being left running when an update changes a dispatch's merge identity.

src/frequenz/dispatch/_actor_dispatcher.py

@@ -5,7 +5,8 @@
 
 import asyncio
 import logging
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
+from copy import copy
 from dataclasses import dataclass
 from datetime import timedelta
 from typing import Any, Awaitable
@@ -19,6 +20,7 @@
 from typing_extensions import deprecated
 
 from ._dispatch import Dispatch
+from ._merge_helpers import _other_running_dispatches
 
 _logger = logging.getLogger(__name__)
 
@@ -102,6 +104,7 @@ def __init__(
         object.__setattr__(self, "_src", _src)
 
 
+# pylint: disable=too-many-instance-attributes
 class ActorDispatcher(BackgroundService):
     """A helper class to manage actors based on dispatches.
 
@@ -246,16 +249,34 @@ def __init__(  # pylint: disable=too-many-arguments, too-many-positional-argumen
         self._actors: dict[DispatchActorId, ActorDispatcher.ActorAndChannel] = {}
         self._failed_dispatches: dict[DispatchActorId, Dispatch] = {}
         """Failed dispatches that will be retried later."""
+        self._dispatch_identities: dict[DispatchId, DispatchActorId] = {}
+        """The identity to which each running dispatch was last routed."""
+        self._running_dispatches: dict[DispatchId, Dispatch] = {}
+        """The latest version of each running dispatch."""
+        self._dispatch_filter: (
+            Callable[[Mapping[DispatchId, Dispatch], Dispatch], bool] | None
+        ) = None
+        self._known_dispatches: Mapping[DispatchId, Dispatch] | None = None
 
     def start(self) -> None:
         """Start the background service."""
         self._tasks.add(asyncio.create_task(self._run()))
 
-    async def _start_actor(self, dispatch: Dispatch) -> None:
+    def _set_merge_context(
+        self,
+        dispatch_filter: Callable[[Mapping[DispatchId, Dispatch], Dispatch], bool],
+        known_dispatches: Mapping[DispatchId, Dispatch],
+    ) -> None:
+        """Set the merge filter and its complete dispatch context."""
+        self._dispatch_filter = dispatch_filter
+        self._known_dispatches = known_dispatches
+
+    async def _start_actor(self, dispatch: Dispatch, identity: DispatchActorId) -> None:
         """Start the actor the given dispatch refers to.
 
         Args:
             dispatch: The dispatch to start the actor for.
+            identity: The identity to route the dispatch to.
         """
         dispatch_update = DispatchInfo(
             target=dispatch.target,
@@ -264,7 +285,6 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
             _src=dispatch,
         )
 
-        identity = self._dispatch_identity(dispatch)
         actor_and_channel = self._actors.get(identity)
 
         if actor_and_channel:
@@ -301,15 +321,19 @@ async def _start_actor(self, dispatch: Dispatch) -> None:
                     actor=actor, channel=channel, sender=channel.new_sender()
                 )
 
-    async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None:
+    async def _stop_actor(
+        self,
+        stopping_dispatch: Dispatch,
+        identity: DispatchActorId,
+        msg: str,
+    ) -> None:
         """Stop the actor for the given dispatch.
 
         Args:
             stopping_dispatch: The dispatch that is stopping the actor.
+            identity: The identity of the actor to stop.
             msg: The message to be passed to the actors being stopped.
         """
-        identity = self._dispatch_identity(stopping_dispatch)
-
         if actor_and_channel := self._actors.pop(identity, None):
             _logger.info("Stopping actor for dispatch type %r", stopping_dispatch.type)
             await actor_and_channel.actor.stop(msg)
@@ -339,6 +363,27 @@ async def _run(self) -> None:
             elif self._dispatch_rx.triggered(selected):
                 await self._handle_dispatch(selected.message)
 
+    def _should_stop_actor(
+        self,
+        dispatch: Dispatch,
+        identity: DispatchActorId,
+        *,
+        check_local_dispatches: bool = False,
+    ) -> bool:
+        """Return whether the actor for an identity should stop."""
+        if self._dispatch_filter is not None and self._known_dispatches is not None:
+            return self._dispatch_filter(self._known_dispatches, dispatch)
+
+        if not check_local_dispatches:
+            return True
+
+        return not _other_running_dispatches(
+            self._running_dispatches,
+            dispatch.id,
+            identity,
+            self._dispatch_identity,
+        )
+
     async def _handle_dispatch(self, dispatch: Dispatch) -> None:
         """Process a dispatch to start, update, or stop an actor.
 
@@ -349,11 +394,43 @@ async def _handle_dispatch(self, dispatch: Dispatch) -> None:
         Args:
             dispatch: The dispatch to handle.
         """
-        identity = self._dispatch_identity(dispatch)
-        if identity in self._failed_dispatches:
-            self._failed_dispatches.pop(identity)
-
         if dispatch.started:
-            await self._start_actor(dispatch)
+            identity = self._dispatch_identity(dispatch)
+            previous_identity = self._dispatch_identities.get(dispatch.id)
+
+            if previous_identity is not None and previous_identity != identity:
+                previous_dispatch = self._running_dispatches[dispatch.id]
+                failed_dispatch = self._failed_dispatches.get(previous_identity)
+                if failed_dispatch is not None and failed_dispatch.id == dispatch.id:
+                    self._failed_dispatches.pop(previous_identity)
+
+                stopping_dispatch = copy(previous_dispatch)
+                # pylint: disable-next=protected-access
+                stopping_dispatch._set_deleted()
+                if self._should_stop_actor(
+                    stopping_dispatch,
+                    previous_identity,
+                    check_local_dispatches=True,
+                ):
+                    await self._stop_actor(
+                        dispatch,
+                        previous_identity,
+                        "Dispatch identity changed",
+                    )
+
+            self._dispatch_identities[dispatch.id] = identity
+            self._running_dispatches[dispatch.id] = dispatch
+            self._failed_dispatches.pop(identity, None)
+            await self._start_actor(dispatch, identity)
         else:
-            await self._stop_actor(dispatch, "Dispatch stopped")
+            identity = self._dispatch_identities.pop(
+                dispatch.id, self._dispatch_identity(dispatch)
+            )
+            self._running_dispatches.pop(dispatch.id, None)
+
+            failed_dispatch = self._failed_dispatches.get(identity)
+            if failed_dispatch is not None and failed_dispatch.id == dispatch.id:
+                self._failed_dispatches.pop(identity)
+
+            if self._should_stop_actor(dispatch, identity):
+                await self._stop_actor(dispatch, identity, "Dispatch stopped")

src/frequenz/dispatch/_dispatcher.py

@@ -368,14 +368,23 @@ def id_identity(dispatch: Dispatch) -> DispatchActorId:
             actor_factory=actor_factory,
             running_status_receiver=await self.new_running_state_event_receiver(
                 dispatch_type,
-                merge_strategy=merge_strategy,
+                merge_strategy=None,
             ),
             dispatch_identity=(
                 id_identity if merge_strategy is None else merge_strategy.identity
             ),
             retry_interval=retry_interval,
         )
 
+        if merge_strategy is not None:
+            # The filter needs the same complete context used by public receivers.
+            # pylint: disable=protected-access
+            dispatcher._set_merge_context(
+                merge_strategy.filter,
+                self._bg_service._dispatches,
+            )
+            # pylint: enable=protected-access
+
         self._actor_dispatchers[dispatch_type] = dispatcher
         dispatcher.start()
 

src/frequenz/dispatch/_merge_helpers.py

@@ -0,0 +1,33 @@
+# License: MIT
+# Copyright © 2026 Frequenz Energy-as-a-Service GmbH
+
+"""Helpers shared by dispatch merge handling."""
+
+from collections.abc import Callable, Mapping
+from datetime import datetime, timezone
+
+from frequenz.client.dispatch.types import DispatchId
+
+from ._dispatch import Dispatch
+
+
+def _other_running_dispatches(
+    dispatches: Mapping[DispatchId, Dispatch],
+    dispatch_id: DispatchId,
+    identity: object,
+    dispatch_identity: Callable[[Dispatch], object],
+    *,
+    now: datetime | None = None,
+) -> list[Dispatch]:
+    """Return other running dispatches mapped to an identity."""
+    if now is None:
+        now = datetime.now(tz=timezone.utc)
+    return [
+        dispatch
+        for dispatch in dispatches.values()
+        if (
+            dispatch.id != dispatch_id
+            and dispatch.started_at(now)
+            and dispatch_identity(dispatch) == identity
+        )
+    ]

src/frequenz/dispatch/_merge_strategies.py

@@ -15,6 +15,7 @@
 from ._actor_dispatcher import DispatchActorId
 from ._bg_service import MergeStrategy
 from ._dispatch import Dispatch
+from ._merge_helpers import _other_running_dispatches
 
 _logger = logging.getLogger(__name__)
 
@@ -61,19 +62,14 @@ def filter(
             _logger.debug("Keeping start event %s", dispatch.id)
             return True
 
-        running_dispatch_list = [
-            existing_dispatch
-            for existing_dispatch in dispatches.values()
-            if (
-                self.identity(existing_dispatch) == self.identity(dispatch)
-                and existing_dispatch.id != dispatch.id
-            )
-        ]
-
-        other_dispatches_running = any(
-            running_dispatch.started_at(now)
-            for running_dispatch in running_dispatch_list
+        running_dispatch_list = _other_running_dispatches(
+            dispatches,
+            dispatch.id,
+            self.identity(dispatch),
+            self.identity,
+            now=now,
         )
+        other_dispatches_running = bool(running_dispatch_list)
 
         _logger.debug(
             "%s stop event %s because other_dispatches_running=%s",