diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 6b1698ed..3892a765 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,10 +1,17 @@ # Dispatch Highlevel Interface Release Notes -## Enhancements +## Summary -- Improved docstring documentation across the project. + + +## Upgrading + + + +## New Features + + ## 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. diff --git a/src/frequenz/dispatch/_actor_dispatcher.py b/src/frequenz/dispatch/_actor_dispatcher.py index a8a8bc5b..be135a5e 100644 --- a/src/frequenz/dispatch/_actor_dispatcher.py +++ b/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__) @@ -216,6 +218,27 @@ class ActorAndChannel: sender: Sender[DispatchInfo] """The sender for dispatch updates.""" + @dataclass(frozen=True, kw_only=True) + class ActorStatus: + """The running actor or failed dispatch for one actor identity.""" + + actor_and_channel: "ActorDispatcher.ActorAndChannel | None" = None + failed_dispatch: Dispatch | None = None + + @dataclass(frozen=True, kw_only=True) + class DispatchStatus: + """The latest dispatch and identity routed for one dispatch ID.""" + + dispatch: Dispatch + identity: DispatchActorId + + @dataclass(frozen=True, kw_only=True) + class MergeContext: + """The merge filter and the dispatches it evaluates.""" + + dispatch_filter: Callable[[Mapping[DispatchId, Dispatch], Dispatch], bool] + known_dispatches: Mapping[DispatchId, Dispatch] + def __init__( # pylint: disable=too-many-arguments, too-many-positional-arguments self, actor_factory: Callable[ @@ -243,19 +266,31 @@ def __init__( # pylint: disable=too-many-arguments, too-many-positional-argumen self._dispatch_rx = running_status_receiver self._retry_timer_rx = Timer(retry_interval, SkipMissedAndDrift()) self._actor_factory = actor_factory - self._actors: dict[DispatchActorId, ActorDispatcher.ActorAndChannel] = {} - self._failed_dispatches: dict[DispatchActorId, Dispatch] = {} - """Failed dispatches that will be retried later.""" + self._actors: dict[DispatchActorId, ActorDispatcher.ActorStatus] = {} + self._dispatch_status: dict[DispatchId, ActorDispatcher.DispatchStatus] = {} + self._merge_context: ActorDispatcher.MergeContext | 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._merge_context = ActorDispatcher.MergeContext( + dispatch_filter=dispatch_filter, + 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,8 +299,10 @@ async def _start_actor(self, dispatch: Dispatch) -> None: _src=dispatch, ) - identity = self._dispatch_identity(dispatch) - actor_and_channel = self._actors.get(identity) + actor_status = self._actors.get(identity) + actor_and_channel = ( + actor_status.actor_and_channel if actor_status is not None else None + ) if actor_and_channel: await actor_and_channel.sender.send(dispatch_update) @@ -294,23 +331,37 @@ async def _start_actor(self, dispatch: Dispatch) -> None: dispatch.type, exc_info=e, ) - self._failed_dispatches[identity] = dispatch + self._actors[identity] = ActorDispatcher.ActorStatus( + failed_dispatch=dispatch + ) else: # No exception occurred, so we can add the actor to the list - self._actors[identity] = ActorDispatcher.ActorAndChannel( - actor=actor, channel=channel, sender=channel.new_sender() + self._actors[identity] = ActorDispatcher.ActorStatus( + actor_and_channel=ActorDispatcher.ActorAndChannel( + 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): + actor_status = self._actors.pop(identity, None) + actor_and_channel = ( + actor_status.actor_and_channel if actor_status is not None else None + ) + if actor_and_channel is not None: _logger.info("Stopping actor for dispatch type %r", stopping_dispatch.type) await actor_and_channel.actor.stop(msg) await actor_and_channel.channel.close() @@ -324,21 +375,49 @@ async def _run(self) -> None: """Run the background service.""" async for selected in select(self._retry_timer_rx, self._dispatch_rx): if self._retry_timer_rx.triggered(selected): - if not self._failed_dispatches: + failed_dispatches = [ + actor_status.failed_dispatch + for actor_status in self._actors.values() + if actor_status.failed_dispatch is not None + ] + if not failed_dispatches: continue _logger.info( "Retrying %d failed actor starts", - len(self._failed_dispatches), + len(failed_dispatches), ) - keys = list(self._failed_dispatches.keys()) - for identity in keys: - dispatch = self._failed_dispatches[identity] - + for dispatch in failed_dispatches: await self._handle_dispatch(dispatch) 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._merge_context is not None: + return self._merge_context.dispatch_filter( + self._merge_context.known_dispatches, dispatch + ) + + if not check_local_dispatches: + return True + + return not _other_running_dispatches( + { + dispatch_id: status.dispatch + for dispatch_id, status in self._dispatch_status.items() + }, + 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 +428,58 @@ 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_status = self._dispatch_status.get(dispatch.id) + previous_identity = ( + previous_status.identity if previous_status is not None else None + ) + + if previous_identity is not None and previous_identity != identity: + assert previous_status is not None + actor_status = self._actors.get(previous_identity) + failed_dispatch = ( + actor_status.failed_dispatch if actor_status is not None else None + ) + if failed_dispatch is not None and failed_dispatch.id == dispatch.id: + self._actors.pop(previous_identity) + + stopping_dispatch = copy(previous_status.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_status[dispatch.id] = ActorDispatcher.DispatchStatus( + dispatch=dispatch, + identity=identity, + ) + actor_status = self._actors.get(identity) + if actor_status is not None and actor_status.failed_dispatch is not None: + self._actors.pop(identity) + await self._start_actor(dispatch, identity) else: - await self._stop_actor(dispatch, "Dispatch stopped") + dispatch_status = self._dispatch_status.pop(dispatch.id, None) + identity = ( + dispatch_status.identity + if dispatch_status is not None + else self._dispatch_identity(dispatch) + ) + + actor_status = self._actors.get(identity) + failed_dispatch = ( + actor_status.failed_dispatch if actor_status is not None else None + ) + if failed_dispatch is not None and failed_dispatch.id == dispatch.id: + self._actors.pop(identity) + + if self._should_stop_actor(dispatch, identity): + await self._stop_actor(dispatch, identity, "Dispatch stopped") diff --git a/src/frequenz/dispatch/_dispatcher.py b/src/frequenz/dispatch/_dispatcher.py index 77762c8c..9351fdd2 100644 --- a/src/frequenz/dispatch/_dispatcher.py +++ b/src/frequenz/dispatch/_dispatcher.py @@ -368,7 +368,7 @@ 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 @@ -376,6 +376,15 @@ def id_identity(dispatch: Dispatch) -> DispatchActorId: 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() diff --git a/src/frequenz/dispatch/_merge_helpers.py b/src/frequenz/dispatch/_merge_helpers.py new file mode 100644 index 00000000..d6d240ed --- /dev/null +++ b/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 + ) + ] diff --git a/src/frequenz/dispatch/_merge_strategies.py b/src/frequenz/dispatch/_merge_strategies.py index 7c6b1635..2b0d2a98 100644 --- a/src/frequenz/dispatch/_merge_strategies.py +++ b/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,16 @@ 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 + # The receiver is filtered by type only; dry-run dispatches have a + # different identity and must not suppress each other's stop events. + 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", diff --git a/tests/test_managing_actor.py b/tests/test_managing_actor.py index 13adf64b..df76f253 100644 --- a/tests/test_managing_actor.py +++ b/tests/test_managing_actor.py @@ -15,12 +15,12 @@ import time_machine from frequenz.channels import Broadcast, Receiver, Sender from frequenz.client.common.microgrid import MicrogridId -from frequenz.client.common.microgrid.components import ComponentId +from frequenz.client.common.microgrid.components import ComponentCategory, ComponentId from frequenz.client.dispatch import recurrence from frequenz.client.dispatch.recurrence import Frequency, RecurrenceRule from frequenz.client.dispatch.test.client import FakeClient from frequenz.client.dispatch.test.generator import DispatchGenerator -from frequenz.client.dispatch.types import DispatchId, TargetIds +from frequenz.client.dispatch.types import DispatchId, TargetCategories, TargetIds from frequenz.sdk.actor import Actor from pytest import fixture @@ -111,7 +111,9 @@ def actor(self, identity: DispatchActorId | int) -> MockActor: # pylint: disable=protected-access assert identity in self.actors_service._actors - return cast(MockActor, self.actors_service._actors[identity].actor) + actor_and_channel = self.actors_service._actors[identity].actor_and_channel + assert actor_and_channel is not None + return cast(MockActor, actor_and_channel.actor) # pylint: enable=protected-access def is_running(self, identity: int) -> bool: @@ -123,6 +125,77 @@ def is_running(self, identity: int) -> bool: return self.actor(identity).is_running +@dataclass +class _IdentityTestEnv: + """Test environment for merge identity changes.""" + + actors_service: ActorDispatcher + actors: list[MockActor] + generator: DispatchGenerator + + async def handle(self, dispatch: Dispatch) -> None: + """Handle a dispatch and let any actor task start.""" + # pylint: disable-next=protected-access + await self.actors_service._handle_dispatch(dispatch) + await asyncio.sleep(0) + + +@fixture +async def identity_test_env( + generator: DispatchGenerator, +) -> AsyncIterator[_IdentityTestEnv]: + """Create an actor dispatcher using target-based merge identities.""" + channel = Broadcast[Dispatch](name="identity change test channel") + actors: list[MockActor] = [] + + async def actor_factory( + initial_dispatch: DispatchInfo, receiver: Receiver[DispatchInfo] + ) -> MockActor: + actor = MockActor(initial_dispatch, receiver) + actors.append(actor) + return actor + + actors_service = ActorDispatcher( + actor_factory=actor_factory, + running_status_receiver=channel.new_receiver(), + dispatch_identity=MergeByTypeTarget().identity, + ) + + yield _IdentityTestEnv(actors_service, actors, generator) + + # pylint: disable=protected-access + for actor_status in list(actors_service._actors.values()): + actor_and_channel = actor_status.actor_and_channel + if actor_and_channel is None: + continue + await actor_and_channel.actor.stop("Test finished") + await actor_and_channel.channel.close() + # pylint: enable=protected-access + + +def _identity_test_dispatch( + generator: DispatchGenerator, + *, + dispatch_id: int, + target: ComponentCategory, + active: bool = True, +) -> Dispatch: + """Create a dispatch for merge identity tests.""" + return Dispatch( + replace( + generator.generate_dispatch(), + id=DispatchId(dispatch_id), + type="IDENTITY_TEST", + active=active, + dry_run=False, + start_time=_now() - timedelta(minutes=1), + duration=timedelta(minutes=10), + target=TargetCategories(target), + recurrence=RecurrenceRule(), + ) + ) + + @fixture async def test_env() -> AsyncIterator[_TestEnv]: """Create a test environment.""" @@ -234,6 +307,226 @@ async def test_start_failed( assert test_env.actor(1).is_running is True +async def test_identity_change_stops_old_actor_and_final_stop_stops_new_actor( + identity_test_env: _IdentityTestEnv, +) -> None: + """Test changing a merge identity doesn't orphan the old actor.""" + battery_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.BATTERY, + ) + inverter_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.INVERTER, + ) + inverter_stop = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.INVERTER, + active=False, + ) + + await identity_test_env.handle(battery_dispatch) + battery_actor = identity_test_env.actors[-1] + await identity_test_env.handle(inverter_dispatch) + inverter_actor = identity_test_env.actors[-1] + + assert battery_actor.is_running is False + assert inverter_actor.is_running is True + + await identity_test_env.handle(inverter_stop) + + assert inverter_actor.is_running is False + # pylint: disable-next=protected-access + assert not identity_test_env.actors_service._actors + + +async def test_identity_change_keeps_shared_old_actor_running( + identity_test_env: _IdentityTestEnv, +) -> None: + """Test an identity change preserves an actor shared by another dispatch.""" + strategy = MergeByTypeTarget() + known_dispatches: dict[DispatchId, Dispatch] = {} + # Use the same filter and dispatch context wired by Dispatcher.start_managing(). + # pylint: disable-next=protected-access + identity_test_env.actors_service._set_merge_context( + strategy.filter, known_dispatches + ) + + first_battery_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.BATTERY, + ) + second_battery_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=2, + target=ComponentCategory.BATTERY, + ) + inverter_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.INVERTER, + ) + + known_dispatches[first_battery_dispatch.id] = first_battery_dispatch + await identity_test_env.handle(first_battery_dispatch) + battery_actor = identity_test_env.actors[-1] + known_dispatches[second_battery_dispatch.id] = second_battery_dispatch + await identity_test_env.handle(second_battery_dispatch) + known_dispatches[inverter_dispatch.id] = inverter_dispatch + await identity_test_env.handle(inverter_dispatch) + + assert battery_actor.is_running is True + assert identity_test_env.actors[-1].is_running is True + # pylint: disable-next=protected-access + assert len(identity_test_env.actors_service._actors) == 2 + + second_battery_stop = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=2, + target=ComponentCategory.BATTERY, + active=False, + ) + known_dispatches[second_battery_stop.id] = second_battery_stop + await identity_test_env.handle(second_battery_stop) + + assert battery_actor.is_running is False + + +async def test_filtered_receiver_final_stop_ignores_stale_local_dispatches( + identity_test_env: _IdentityTestEnv, +) -> None: + """Test a filtered final stop keeps the receiver's old stop semantics.""" + first_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.BATTERY, + ) + second_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=2, + target=ComponentCategory.BATTERY, + ) + second_stop = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=2, + target=ComponentCategory.BATTERY, + active=False, + ) + + await identity_test_env.handle(first_dispatch) + actor = identity_test_env.actors[-1] + await identity_test_env.handle(second_dispatch) + + # A filtered receiver suppresses the first dispatch's stop event. + await identity_test_env.handle(second_stop) + + assert actor.is_running is False + # pylint: disable-next=protected-access + assert not identity_test_env.actors_service._actors + + +async def test_identity_can_change_back_and_forth( + identity_test_env: _IdentityTestEnv, +) -> None: + """Test repeated identity changes stop each superseded actor.""" + battery_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.BATTERY, + ) + inverter_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.INVERTER, + ) + + await identity_test_env.handle(battery_dispatch) + first_battery_actor = identity_test_env.actors[-1] + await identity_test_env.handle(inverter_dispatch) + inverter_actor = identity_test_env.actors[-1] + await identity_test_env.handle(battery_dispatch) + second_battery_actor = identity_test_env.actors[-1] + + assert first_battery_actor.is_running is False + assert inverter_actor.is_running is False + assert second_battery_actor.is_running is True + # pylint: disable-next=protected-access + assert len(identity_test_env.actors_service._actors) == 1 + + +async def test_identity_change_discards_failed_start_retry( + identity_test_env: _IdentityTestEnv, +) -> None: + """Test an identity change removes the old identity's failed start.""" + # pylint: disable=protected-access + working_factory = identity_test_env.actors_service._actor_factory + attempts = 0 + + async def fail_once( + initial_dispatch: DispatchInfo, receiver: Receiver[DispatchInfo] + ) -> MockActor: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise ValueError("Failed to create actor") + return cast(MockActor, await working_factory(initial_dispatch, receiver)) + + identity_test_env.actors_service._actor_factory = fail_once + + battery_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.BATTERY, + ) + inverter_dispatch = _identity_test_dispatch( + identity_test_env.generator, + dispatch_id=1, + target=ComponentCategory.INVERTER, + ) + old_identity = MergeByTypeTarget().identity(battery_dispatch) + + await identity_test_env.handle(battery_dispatch) + actor_status = identity_test_env.actors_service._actors[old_identity] + assert actor_status.failed_dispatch is not None + + await identity_test_env.handle(inverter_dispatch) + + assert old_identity not in identity_test_env.actors_service._actors + assert identity_test_env.actors[-1].is_running is True + # pylint: enable=protected-access + + +async def test_default_identity_does_not_change_with_target( + test_env: _TestEnv, +) -> None: + """Test the default dispatch-ID identity is unaffected by target updates.""" + battery_dispatch = _identity_test_dispatch( + test_env.generator, + dispatch_id=42, + target=ComponentCategory.BATTERY, + ) + inverter_dispatch = _identity_test_dispatch( + test_env.generator, + dispatch_id=42, + target=ComponentCategory.INVERTER, + ) + + await test_env.running_status_sender.send(battery_dispatch) + await asyncio.sleep(0.1) + actor = test_env.actor(42) + await test_env.running_status_sender.send(inverter_dispatch) + await asyncio.sleep(0.1) + + assert test_env.actor(42) is actor + assert actor.is_running is True + # pylint: disable-next=protected-access + assert len(test_env.actors_service._actors) == 1 + + def test_heapq_dispatch_compare(test_env: _TestEnv) -> None: """Test that the heapq compare function works.""" dispatch1 = test_env.generator.generate_dispatch() @@ -371,7 +664,7 @@ async def new_mock_receiver( merge_strategy: MergeStrategy | None, ) -> Receiver[Dispatch]: assert dispatch_type == "MANAGE_TEST" - assert merge_strategy is strategy + assert merge_strategy is None return channel.new_receiver() with patch(