From e20da6cbb18c7176ce1462232531a91133c790b8 Mon Sep 17 00:00:00 2001 From: "Mathias L. Baumann" Date: Thu, 15 Oct 2026 18:35:20 +0200 Subject: [PATCH 1/4] Support contiguous dispatch scheduling Signed-off-by: Mathias L. Baumann --- src/frequenz/dispatch/_bg_service.py | 40 +++++++++++++++++++++++++++- tests/test_dispatcher_scheduler.py | 25 +++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/src/frequenz/dispatch/_bg_service.py b/src/frequenz/dispatch/_bg_service.py index 54bbe5f7..9c874dcf 100644 --- a/src/frequenz/dispatch/_bg_service.py +++ b/src/frequenz/dispatch/_bg_service.py @@ -14,6 +14,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from heapq import heapify, heappop, heappush +from types import MappingProxyType import grpc.aio from frequenz.channels import Broadcast, Receiver, select, selected_from @@ -157,6 +158,26 @@ async def wait_for_initialization(self) -> None: """Wait for the initial fetch to complete.""" await self._initial_fetch_event.wait() + def dispatches_snapshot(self) -> Mapping[DispatchId, Dispatch]: + """Return a READ-ONLY, live view of every dispatch currently known. + + This is the authoritative data this scheduler already fetches ahead of + execution (initial `list()` plus the live CREATE/UPDATE/DELETE stream) + -- intended for `ActorDispatcher`'s opt-in `dispatches_snapshot` + parameter, so it can resolve "is there already a known, started, + exact-continuation successor for this dispatch" at the moment a stop + event is being processed, without this scheduler ever having to decide + whether to suppress or forward the stop itself: every stop is always + forwarded; `ActorDispatcher` is the only place that decides what to do + with it. + + Returns: + A live, read-only mapping (a `MappingProxyType` over the same + underlying dict this scheduler mutates -- callers see + updates, but cannot make any themselves). + """ + return MappingProxyType(self._dispatches) + # pylint: disable=redefined-builtin def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]: """Create a new receiver for lifecycle events. @@ -545,7 +566,24 @@ def _update_timer(self, timer: Timer) -> None: """Update the timer to the next event.""" if self._scheduled_events: due_at: datetime = self._scheduled_events[0].time - timer.reset(interval=due_at - datetime.now(timezone.utc)) + interval = due_at - datetime.now(timezone.utc) + # Two scheduled events landing at (or within microseconds of) the + # exact same instant -- e.g. one dispatch's stop and another's + # start sharing a boundary -- is a normal, expected shape (see + # `QueueItem`'s own docstring on same-time start/stop ordering). + # Once the first of the pair is processed, this recomputed + # interval for the second can be exactly zero or slightly + # negative. `Timer.reset()` accepts that value without complaint, + # but a zero interval later crashes the timer's OWN drift + # correction with a `ZeroDivisionError` in + # `MissedTickPolicy.calculate_next_tick_time` (`drift % interval`). + # Floor it to the smallest positive interval `Timer` itself + # accepts, instead: the due event is already due, so it fires on + # the very next tick either way, but doesn't poison the timer's + # internal state. + min_interval = timedelta(microseconds=1) + interval = max(interval, min_interval) + timer.reset(interval=interval) _logger.debug("Next event scheduled at %s", self._scheduled_events[0].time) def _remove_scheduled(self, dispatch: Dispatch) -> "QueueItem | None": diff --git a/tests/test_dispatcher_scheduler.py b/tests/test_dispatcher_scheduler.py index 833c9f6a..379d5521 100644 --- a/tests/test_dispatcher_scheduler.py +++ b/tests/test_dispatcher_scheduler.py @@ -14,11 +14,13 @@ import async_solipsism import time_machine from frequenz.channels import Receiver +from frequenz.channels.timer import Timer from frequenz.client.common.microgrid import MicrogridId from frequenz.client.dispatch.recurrence import RecurrenceRule from frequenz.client.dispatch.test.client import FakeClient, to_create_params from frequenz.client.dispatch.test.generator import DispatchGenerator from pytest import fixture +from pytest_mock import MockerFixture from frequenz.dispatch import Dispatch, DispatchEvent from frequenz.dispatch._bg_service import DispatchScheduler @@ -50,6 +52,29 @@ def _now() -> datetime: return datetime.now(tz=timezone.utc) +def test_update_timer_floors_already_due_interval( + generator: DispatchGenerator, + microgrid_id: MicrogridId, + client: FakeClient, + mocker: MockerFixture, +) -> None: + """Keep a same-boundary follow-up timer interval strictly positive.""" + scheduler = DispatchScheduler(microgrid_id=microgrid_id, client=client) + dispatch = Dispatch(generator.generate_dispatch()) + scheduler._scheduled_events.append( # pylint: disable=protected-access + DispatchScheduler.QueueItem( + time=_now() - timedelta(microseconds=1), + dispatch=dispatch, + stop_event=False, + ) + ) + timer = mocker.create_autospec(Timer, instance=True) + + scheduler._update_timer(timer) # pylint: disable=protected-access + + assert timer.reset.call_args.kwargs["interval"] == timedelta(microseconds=1) + + @fixture def microgrid_id() -> MicrogridId: """Return a random microgrid ID.""" From 45ef357debd1f8353a0c888026f842e38e87e99c Mon Sep 17 00:00:00 2001 From: "Mathias L. Baumann" Date: Thu, 15 Oct 2026 18:35:39 +0200 Subject: [PATCH 2/4] Add opt-in contiguous dispatch handovers Signed-off-by: Mathias L. Baumann --- src/frequenz/dispatch/__init__.py | 5 +- src/frequenz/dispatch/_actor_dispatcher.py | 524 +++++++++++++++++-- src/frequenz/dispatch/_continuation.py | 156 ++++++ src/frequenz/dispatch/_dispatcher.py | 33 +- tests/test_continuation.py | 178 +++++++ tests/test_continuation_realclock.py | 573 +++++++++++++++++++++ 6 files changed, 1426 insertions(+), 43 deletions(-) create mode 100644 src/frequenz/dispatch/_continuation.py create mode 100644 tests/test_continuation.py create mode 100644 tests/test_continuation_realclock.py diff --git a/src/frequenz/dispatch/__init__.py b/src/frequenz/dispatch/__init__.py index d91f4002..29e906e1 100644 --- a/src/frequenz/dispatch/__init__.py +++ b/src/frequenz/dispatch/__init__.py @@ -15,8 +15,9 @@ """ -from ._actor_dispatcher import ActorDispatcher, DispatchInfo +from ._actor_dispatcher import ActorDispatcher, ContinuationCapable, DispatchInfo from ._bg_service import MergeStrategy +from ._continuation import is_exact_continuation from ._dispatch import Dispatch from ._dispatcher import Dispatcher from ._event import Created, Deleted, DispatchEvent, Updated @@ -30,8 +31,10 @@ "Updated", "Dispatch", "ActorDispatcher", + "ContinuationCapable", "DispatchInfo", "MergeStrategy", "MergeByType", "MergeByTypeTarget", + "is_exact_continuation", ] diff --git a/src/frequenz/dispatch/_actor_dispatcher.py b/src/frequenz/dispatch/_actor_dispatcher.py index 9c31b523..d38571db 100644 --- a/src/frequenz/dispatch/_actor_dispatcher.py +++ b/src/frequenz/dispatch/_actor_dispatcher.py @@ -5,10 +5,10 @@ import asyncio import logging -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass -from datetime import timedelta -from typing import Any, Awaitable +from datetime import datetime, timedelta, timezone +from typing import Any, Awaitable, Protocol, runtime_checkable from warnings import warn from frequenz.channels import Broadcast, Receiver, Sender, select @@ -23,6 +23,44 @@ _logger = logging.getLogger(__name__) +@runtime_checkable +class ContinuationCapable(Protocol): + """An actor that can explicitly accept or refuse a continuation update. + + `ActorDispatcher` only ever reuses an existing actor for an unrelated + dispatch (see `continuation_predicate`) when that actor's CLASS implements + this protocol -- checked with `isinstance()` BEFORE anything is sent and + BEFORE the predecessor's actor is left running under the assumption it + will pick the update up. An actor that does not implement this (e.g. an + existing actor written before this contract existed) is never silently + left stale: `ActorDispatcher` stops it and starts a fresh actor for the + new dispatch instead, exactly as if no continuation had been attempted. + """ + + async def accept_continuation(self, update: "DispatchInfo") -> bool: + """Attempt to accept a continuation update. + + If this returns `True`, the actor MUST have adopted `update` as its + new current dispatch and submitted the payload's new proposal through + its existing proposal pool before returning. This is proof of actor + state adoption and proposal submission, not a promise that physical + power has already changed or that downstream aggregation has already + completed synchronously. + + Args: + update: The continuation update, with `update.continuation_of` + equal to the dispatch id this actor is currently running + under, and `update._src` the new dispatch to adopt. + + Returns: + `True` if the new dispatch state was adopted and its proposal was + submitted; `False` to refuse, in which case the caller treats + this exactly as if the actor were not continuation-capable at + all (it will be stopped, not left running with a stale + payload). + """ + + class DispatchActorId(BaseId, str_prefix="DA"): """An ID for a dispatch actor.""" @@ -50,17 +88,35 @@ def components(self) -> TargetComponents: return self.target target: TargetComponents - """The target components.""" + """Target components to be used.""" dry_run: bool """Whether this is a dry run.""" options: dict[str, Any] - """The additional options.""" + """Additional options.""" _src: Dispatch """The dispatch that triggered this update.""" + continuation_of: DispatchId | None + """If set, this update continues a PREVIOUS dispatch with this id. + + Only ever set by `ActorDispatcher` itself, when it has decided (via an + explicit, opt-in `continuation_predicate`) that `_src` is an exact, + compatible continuation of a dispatch the receiving actor was already + running under. A continuation-AWARE actor MAY accept this update and adopt + `_src` as its new dispatch -- after validating that `continuation_of` + matches the dispatch id it is currently running under -- even though + `_src.id` differs from that id. + + An actor that does not look at this field is unaffected: it defaults to + `None`, and an actor that only ever checks `_src.id` against its own + dispatch id (as every actor does today) keeps rejecting cross-id updates + exactly as before. This field cannot by itself weaken any existing guard; + an actor has to explicitly opt in to honouring it. + """ + def __init__( self, *, @@ -69,15 +125,18 @@ def __init__( dry_run: bool, options: dict[str, Any], _src: Dispatch, + continuation_of: DispatchId | None = None, ) -> None: - """Initialize a new instance. + """Initialize the DispatchInfo. Args: - target: The target components to be used. + target: Target components to be used. components: Deprecated alias for `target`. dry_run: Whether this is a dry run. - options: The additional options. + options: Additional options. _src: The dispatch that triggered this update. + continuation_of: If set, the id of a previous dispatch this update + continues. See the `continuation_of` attribute for details. Raises: ValueError: If both `target` and `components` are set, or if neither is set. @@ -100,9 +159,12 @@ def __init__( object.__setattr__(self, "dry_run", dry_run) object.__setattr__(self, "options", options) object.__setattr__(self, "_src", _src) + object.__setattr__(self, "continuation_of", continuation_of) -class ActorDispatcher(BackgroundService): +class ActorDispatcher( + BackgroundService +): # pylint: disable=too-many-instance-attributes """A helper class to manage actors based on dispatches. Example: @@ -142,7 +204,7 @@ def new_with_dispatch( @override async def _run(self) -> None: - other_recv: Receiver[Any] = ... + other_recv = Broadcast[Any](name="other").new_receiver() if self._dispatch_updates_receiver is None: async for msg in other_recv: @@ -203,7 +265,7 @@ async def main(): ``` """ - @dataclass(frozen=True, kw_only=True) + @dataclass(kw_only=True) class ActorAndChannel: """An actor and its dispatch update sender.""" @@ -216,6 +278,16 @@ class ActorAndChannel: sender: Sender[DispatchInfo] """The sender for dispatch updates.""" + current_dispatch: Dispatch + """The most recent dispatch this actor is known to be running under. + + Updated on every ordinary same-identity update AND on every successful + continuation, so that `continuation_predicate` always compares a new + dispatch against the FRESHEST known state of a running actor (this is + what makes multi-step continuation chains work: a third dispatch is + compared against the second, not the first). + """ + def __init__( # pylint: disable=too-many-arguments, too-many-positional-arguments self, actor_factory: Callable[ @@ -224,6 +296,9 @@ def __init__( # pylint: disable=too-many-arguments, too-many-positional-argumen running_status_receiver: Receiver[Dispatch], dispatch_identity: Callable[[Dispatch], DispatchActorId] | None = None, retry_interval: timedelta = timedelta(seconds=60), + continuation_predicate: Callable[[Dispatch, Dispatch], bool] | None = None, + dispatches_snapshot: Callable[[], Mapping[DispatchId, Dispatch]] | None = None, + continuation_accept_timeout: timedelta = timedelta(seconds=5), ) -> None: """Initialize the dispatch handler. @@ -231,14 +306,66 @@ def __init__( # pylint: disable=too-many-arguments, too-many-positional-argumen actor_factory: A callable that creates an actor with some initial dispatch information. running_status_receiver: The receiver for dispatch running status changes. + EVERY event (including every stop) is always processed here; + nothing is ever filtered out upstream on this class's behalf -- + see `continuation_predicate`/`dispatches_snapshot` for why. dispatch_identity: A function to identify to which actor a dispatch refers. By default, it uses the dispatch ID. retry_interval: How long to wait until trying to start failed actors again. + continuation_predicate: Opt-in, default `None` (no behaviour change). + Called as `continuation_predicate(current_dispatch, new_dispatch)`. + Used in two symmetric places. Both require a UNIQUE PAIR: + the candidate predecessor must match exactly one successor, + and that successor must match exactly one predecessor. A + one-sided unique match is insufficient. + * On a START event for a dispatch that does NOT already match an + existing actor by `dispatch_identity`: if it returns `True` for + EXACTLY ONE currently-running actor, and that actor's + dispatch has EXACTLY ONE matching successor in + `dispatches_snapshot`, that dispatch's start is + routed to the EXISTING actor as a continuation (see + `DispatchInfo.continuation_of`) instead of creating a second, + independent actor ("start-first" ordering). + * On a STOP event for a dispatch with a running actor: if + `dispatches_snapshot` finds EXACTLY ONE other, + already-started, not-yet-registered dispatch that + `continuation_predicate` says this one continues INTO, and + that successor has EXACTLY ONE matching predecessor in the + same snapshot, the stop is resolved as a continuation into + that dispatch instead ("stop-first" ordering) -- see + `dispatches_snapshot`. + In both places, zero or more than one match means no + continuation is attempted: an ambiguous match is never guessed + at, and a dispatch with no known successor/predecessor is + handled exactly as it always was. + dispatches_snapshot: Opt-in, default `None`. A zero-argument + callable returning a live, READ-ONLY view of every dispatch + currently known to the event source (e.g. + `DispatchScheduler.dispatches_snapshot`) -- used ONLY for the + pair-uniqueness checks above. This class never + mutates anything through it, and never uses it to delay or + suppress a stop: by the time `_stop_actor` is called, the stop + is already being acted on; this is purely input to deciding + WHAT to do about it, synchronously, in the same call. Required + (alongside `continuation_predicate`) for either ordering to be + resolved. Without it, continuation fails closed because the + opposite side of the pair cannot be proven unique. + continuation_accept_timeout: An upper bound on how long a + candidate actor's `accept_continuation` may run before it is + treated as a refusal (same fallback as a `False` return or a + raised exception: the predecessor is stopped and a fresh actor + is started instead). Bounds how long one hanging or + badly-behaved actor can stall this whole service's `_run()` + loop -- every other dispatch is queued behind it until this + resolves one way or another. """ super().__init__() self._dispatch_identity: Callable[[Dispatch], DispatchActorId] = ( dispatch_identity if dispatch_identity else lambda d: DispatchActorId(d.id) ) + self._continuation_predicate = continuation_predicate + self._dispatches_snapshot = dispatches_snapshot + self._continuation_accept_timeout = continuation_accept_timeout self._dispatch_rx = running_status_receiver self._retry_timer_rx = Timer(retry_interval, SkipMissedAndDrift()) @@ -251,6 +378,284 @@ def start(self) -> None: """Start the background service.""" self._tasks.add(asyncio.create_task(self._run())) + def _find_unique_continuation_candidate( + self, dispatch: Dispatch + ) -> tuple[DispatchActorId, "ActorDispatcher.ActorAndChannel"] | None: + """Find the single running actor `dispatch` exactly continues, if any. + + Args: + dispatch: The newly-arrived dispatch, not already matched by + `dispatch_identity` to an existing actor. + + Returns: + The `(identity, actor_and_channel)` of the unique continuation + pair, or `None` if either side of the pair is not unique. + """ + if self._continuation_predicate is None or self._dispatches_snapshot is None: + return None + + candidates = [ + (identity, actor_and_channel) + for identity, actor_and_channel in self._actors.items() + if self._continuation_predicate( + actor_and_channel.current_dispatch, dispatch + ) + ] + + if len(candidates) != 1: + if len(candidates) > 1: + _logger.warning( + "Dispatch %s matches %d continuation predecessor candidates " + "(%s); refusing to guess, starting an independent actor instead", + dispatch.id, + len(candidates), + [identity for identity, _ in candidates], + ) + return None + + predecessor_identity, predecessor = candidates[0] + snapshot = self._dispatches_snapshot() + predecessor_candidates = self._matching_predecessors(dispatch, snapshot) + if ( + len(predecessor_candidates) != 1 + or predecessor_candidates[0].id != predecessor.current_dispatch.id + ): + _logger.warning( + "Dispatch %s matches %d snapshot predecessor candidates (%s); " + "refusing a one-sided actor match and starting an independent " + "actor instead", + dispatch.id, + len(predecessor_candidates), + [candidate.id for candidate in predecessor_candidates], + ) + return None + + successor_candidates = self._matching_successors( + predecessor.current_dispatch, snapshot + ) + if len(successor_candidates) != 1 or successor_candidates[0].id != dispatch.id: + _logger.warning( + "Dispatch %s's predecessor %s matches %d successor candidates " + "(%s); refusing a one-sided match and starting an independent " + "actor instead", + dispatch.id, + predecessor.current_dispatch.id, + len(successor_candidates), + [candidate.id for candidate in successor_candidates], + ) + return None + + return predecessor_identity, predecessor + + def _matching_successors( + self, predecessor: Dispatch, snapshot: Mapping[DispatchId, Dispatch] + ) -> list[Dispatch]: + """Return all already-started snapshot successors of `predecessor`.""" + assert self._continuation_predicate is not None + now = datetime.now(timezone.utc) + return [ + other + for other in snapshot.values() + if other.id != predecessor.id + and other.started_at(now) + and self._continuation_predicate(predecessor, other) + ] + + def _matching_predecessors( + self, successor: Dispatch, snapshot: Mapping[DispatchId, Dispatch] + ) -> list[Dispatch]: + """Return all snapshot predecessors exactly continued by `successor`.""" + assert self._continuation_predicate is not None + return [ + other + for other in snapshot.values() + if other.id != successor.id + and self._continuation_predicate(other, successor) + ] + + def _find_unique_continuation_successor( + self, predecessor: Dispatch + ) -> Dispatch | None: + """Find the unique started successor of `predecessor`, if any. + + This uses the live `dispatches_snapshot`. Called when `predecessor`'s + own stop is about to be processed, so this + can resolve "stop-first" ordering WITHOUT waiting for the successor's + own start event to arrive later -- when it does, it simply hits the + ordinary same-identity "already running" update path, because the + actor has, by then, already been re-keyed onto it. + + Args: + predecessor: The dispatch whose stop is being processed. + + Returns: + The unique successor dispatch, or `None` if there is no + `continuation_predicate`/`dispatches_snapshot` configured, or + zero, or more than one, matching candidate. + """ + if self._continuation_predicate is None or self._dispatches_snapshot is None: + return None + + snapshot = self._dispatches_snapshot() + candidates = self._matching_successors(predecessor, snapshot) + + if len(candidates) != 1: + if len(candidates) > 1: + _logger.warning( + "Dispatch %s has %d ambiguous exact-continuation successor " + "candidates (%s) in the known-dispatches snapshot; refusing " + "to guess, stopping normally instead", + predecessor.id, + len(candidates), + [candidate.id for candidate in candidates], + ) + return None + + successor = candidates[0] + predecessor_candidates = self._matching_predecessors(successor, snapshot) + if ( + len(predecessor_candidates) != 1 + or predecessor_candidates[0].id != predecessor.id + ): + _logger.warning( + "Dispatch %s's successor %s matches %d predecessor candidates " + "(%s); refusing a one-sided match and stopping normally instead", + predecessor.id, + successor.id, + len(predecessor_candidates), + [candidate.id for candidate in predecessor_candidates], + ) + return None + + # A successor whose identity already has its OWN registered actor must + # never be re-keyed into from here too -- that would orphan or conflict + # with the actor it already has. + if self._dispatch_identity(successor) in self._actors: + return None + + return successor + + async def _continue_actor( + self, + predecessor_identity: DispatchActorId, + predecessor: "ActorDispatcher.ActorAndChannel", + dispatch: Dispatch, + ) -> bool: + """Route `dispatch` to `predecessor`'s actor as a continuation. + + Fall back to ordinary stop/start if the actor cannot safely accept it. + + Args: + predecessor_identity: The predecessor actor's current identity key. + predecessor: The predecessor actor-and-channel entry. + dispatch: The new dispatch continuing the predecessor. + + Returns: + `True` if the continuation was delivered AND the actor confirmed it + adopted the new dispatch state and submitted its new proposal + through the existing pool (the actor is re-indexed under the + new dispatch's identity). `False` in every other case -- the + predecessor is NOT left stale: it has already been explicitly + stopped here, and the caller should start a fresh, independent + actor for `dispatch` if it hasn't already been stopped as part + of the caller's own normal flow (see `_stop_actor`, which calls + this and, on `False`, has nothing further to do since the + actor is already gone). + """ + if not isinstance(predecessor.actor, ContinuationCapable): + _logger.info( + "Dispatch %s exactly continues dispatch %s, but its actor does " + "not implement the continuation-acceptance contract " + "(ContinuationCapable.accept_continuation): stopping it and " + "starting %s fresh, instead of leaving it running with a stale " + "payload.", + dispatch.id, + predecessor.current_dispatch.id, + dispatch.id, + ) + return await self._reject_continuation(predecessor_identity, predecessor) + + continuation_update = DispatchInfo( + target=dispatch.target, + dry_run=dispatch.dry_run, + options=dispatch.payload, + _src=dispatch, + continuation_of=predecessor.current_dispatch.id, + ) + + try: + accepted = await asyncio.wait_for( + predecessor.actor.accept_continuation(continuation_update), + timeout=self._continuation_accept_timeout.total_seconds(), + ) + except TimeoutError: + _logger.error( + "Dispatch %s's actor did not respond to a continuation from " + "dispatch %s within %s; stopping it instead of letting it " + "stall this service's _run() loop indefinitely, and starting " + "%s fresh", + predecessor.current_dispatch.id, + dispatch.id, + self._continuation_accept_timeout, + dispatch.id, + ) + return await self._reject_continuation(predecessor_identity, predecessor) + except Exception as e: # pylint: disable=broad-except + _logger.error( + "Failed to deliver continuation update from dispatch %s to " + "dispatch %s; stopping the predecessor's actor instead of leaving " + "its proposal orphaned, and starting %s fresh", + predecessor.current_dispatch.id, + dispatch.id, + dispatch.id, + exc_info=e, + ) + return await self._reject_continuation(predecessor_identity, predecessor) + + if not accepted: + _logger.info( + "Dispatch %s's actor explicitly refused the continuation from " + "dispatch %s; stopping it and starting %s fresh", + predecessor.current_dispatch.id, + dispatch.id, + dispatch.id, + ) + return await self._reject_continuation(predecessor_identity, predecessor) + + new_identity = self._dispatch_identity(dispatch) + del self._actors[predecessor_identity] + predecessor.current_dispatch = dispatch + self._actors[new_identity] = predecessor + _logger.info( + "Dispatch %s continues dispatch %s: reused its actor instead of " + "stopping and restarting", + dispatch.id, + continuation_update.continuation_of, + ) + return True + + async def _reject_continuation( + self, + predecessor_identity: DispatchActorId, + predecessor: "ActorDispatcher.ActorAndChannel", + ) -> bool: + """Stop and remove the predecessor actor. + + This retains ordinary stop/start behaviour as if no continuation had + been attempted. + + Args: + predecessor_identity: The predecessor actor's current identity key. + predecessor: The predecessor actor-and-channel entry to stop. + + Returns: + Always `False`, so callers can `return await self._reject_continuation(...)`. + """ + del self._actors[predecessor_identity] + await predecessor.actor.stop("continuation not accepted") + await predecessor.channel.close() + return False + async def _start_actor(self, dispatch: Dispatch) -> None: """Start the actor the given dispatch refers to. @@ -269,56 +674,93 @@ async def _start_actor(self, dispatch: Dispatch) -> None: if actor_and_channel: await actor_and_channel.sender.send(dispatch_update) + actor_and_channel.current_dispatch = dispatch _logger.info( "Actor for dispatch type %r is already running, " "sent a dispatch update instead of creating a new actor", dispatch.type, ) - else: - try: - _logger.info("Starting actor for dispatch type %r", dispatch.type) - channel = Broadcast[DispatchInfo]( - name=f"dispatch_updates_channel_instance={identity}", - resend_latest=True, - ) - actor = await self._actor_factory( - dispatch_update, - channel.new_receiver(limit=1, warn_on_overflow=False), - ) + return + + candidate = self._find_unique_continuation_candidate(dispatch) + if candidate is not None: + predecessor_identity, predecessor = candidate + if await self._continue_actor(predecessor_identity, predecessor, dispatch): + return + # Continuation delivery failed and the predecessor has already been + # cleaned up (see `_continue_actor`) -- fall through to start fresh. + + try: + _logger.info("Starting actor for dispatch type %r", dispatch.type) + channel = Broadcast[DispatchInfo]( + name=f"dispatch_updates_channel_instance={identity}", + resend_latest=True, + ) + actor = await self._actor_factory( + dispatch_update, + channel.new_receiver(limit=1, warn_on_overflow=False), + ) - actor.start() + actor.start() - except Exception as e: # pylint: disable=broad-except - _logger.error( - "Failed to start actor for dispatch type %r", - dispatch.type, - exc_info=e, - ) - self._failed_dispatches[identity] = 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() - ) + except Exception as e: # pylint: disable=broad-except + _logger.error( + "Failed to start actor for dispatch type %r", + dispatch.type, + exc_info=e, + ) + self._failed_dispatches[identity] = 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(), + current_dispatch=dispatch, + ) async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None: - """Stop the actor for the given dispatch. + """Stop the actor or continue it into a unique started successor. + + This stop is NEVER withheld or delayed while that is being decided: + every stop this class receives is one it was always going to receive + (nothing is suppressed upstream), and the decision below is made + synchronously, in this one call, using only data already known at + this instant. Args: stopping_dispatch: The dispatch that is stopping the actor. msg: The message to be passed to the actors being stopped. """ identity = self._dispatch_identity(stopping_dispatch) + actor_and_channel = self._actors.get(identity) - 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) - await actor_and_channel.channel.close() - else: + if actor_and_channel is None: _logger.warning( "Actor for dispatch type %r is not running, ignoring stop request", stopping_dispatch.type, ) + return + + successor = self._find_unique_continuation_successor(stopping_dispatch) + if successor is not None: + if await self._continue_actor(identity, actor_and_channel, successor): + _logger.info( + "Dispatch %s's stop resolved into a continuation of " + "dispatch %s instead of stopping", + stopping_dispatch.id, + successor.id, + ) + return + # `_continue_actor` already stopped and removed the actor on any + # failure path (capability check, refusal, timeout, exception) -- + # nothing further to do here. + return + + del self._actors[identity] + _logger.info("Stopping actor for dispatch type %r", stopping_dispatch.type) + await actor_and_channel.actor.stop(msg) + await actor_and_channel.channel.close() async def _run(self) -> None: """Run the background service.""" diff --git a/src/frequenz/dispatch/_continuation.py b/src/frequenz/dispatch/_continuation.py new file mode 100644 index 00000000..5255565d --- /dev/null +++ b/src/frequenz/dispatch/_continuation.py @@ -0,0 +1,156 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Opt-in support for exactly-adjacent, compatible dispatch continuations. + +This module is deliberately separate from `_merge_strategies.py`: `MergeByType` +and `MergeByTypeTarget` merge dispatches that merely share (type[, target], +dry_run), with no regard for whether their time windows are adjacent or +overlapping. That is the wrong tool for "the next back-to-back dispatch window +is really a continuation of the previous one" -- a genuine, independent overlap +on the same target must NOT be treated as a continuation (see +`is_exact_continuation`). Nothing here changes `MergeByType`/`MergeByTypeTarget` +or `ActorDispatcher`'s default (per-dispatch-ID) behaviour; everything is opt-in. + +There is deliberately no scheduler-side filter/strategy in this module. An +earlier version of this prototype had one (suppressing a predecessor's stop +event when a continuation was already known) -- that is gone: suppressing an +event before any consumer decides what to do with it creates an orphan class +of its own whenever the consumer turns out unable to use what was suppressed +(deletion racing the suppression, a hanging/refusing actor, an unsupported +actor type, ...). Every stop is now always forwarded; `ActorDispatcher` is the +only place that decides, synchronously, at the moment it is about to act on a +stop, whether a unique already-known successor lets it re-key instead of +stopping -- see `ActorDispatcher.dispatches_snapshot`/`continuation_predicate`. +""" + +from __future__ import annotations + +from frequenz.client.dispatch.types import ( + BatteryType, + EvChargerType, + InverterType, + TargetCategories, + TargetIds, +) + +from ._dispatch import Dispatch + + +def _canonical_target(target: object) -> frozenset[tuple[object, ...]]: + """Canonicalize a dispatch target into an order-independent, class-aware set. + + `TargetIds`/`TargetCategories` are already `frozenset`s, so their OWN + `==` is already order-independent -- but comparing them via + `tuple(target_a) == tuple(target_b)` throws that away and relies on two + separately-constructed frozensets happening to iterate in the same order, + which CPython does not guarantee for arbitrary construction paths. This + canonicalizes to a plain `frozenset` of hashable keys instead, so + comparison is unconditionally order-independent. + + For `TargetCategories`, each key is built from `TargetCategory.category2` + (the library's OWN normalization of a category/type element to a single + `ElectricalComponentCategory` value -- it deliberately treats the legacy + `ComponentCategory.BATTERY` and `ElectricalComponentCategory.BATTERY` as + the SAME real-world category, by value, and this canonicalization + respects that rather than fighting it), plus, when the element names a + specific sub-type (`BatteryType`/`EvChargerType`/`InverterType`), that + sub-type's own class AND value. A plain "any battery" target is therefore + NOT conflated with a specific "LI_ION battery" target (continuation + requires the same specificity, not just the same broad category) -- but + `ComponentCategory.BATTERY` and `ElectricalComponentCategory.BATTERY` on + their own ARE treated as the same target, matching the library's own + `category2` semantics. The sub-type is tagged with its own class name as + well as its value so that, were any of `BatteryType`/`EvChargerType`/ + `InverterType` to ever become an `IntEnum` (today none are plain `Enum`, + so this cannot yet happen), two different sub-types that happened to + share a raw integer value still could not be conflated. + + Args: + target: The dispatch target to canonicalize (a `TargetIds` or + `TargetCategories` instance). + + Raises: + TypeError: If `target` is neither a `TargetIds` nor a + `TargetCategories`. + + Returns: + An order-independent, class-aware, hashable representation of + `target`, safe to compare with `==` against another call's result. + """ + match target: + case TargetIds(): + return frozenset(("id", int(component_id)) for component_id in target) + case TargetCategories(): + keys: set[tuple[object, ...]] = set() + for category in target: + specific = ( + category.target + if isinstance( + category.target, (BatteryType, EvChargerType, InverterType) + ) + else None + ) + keys.add( + ( + "category", + category.category2, + type(specific).__qualname__ if specific is not None else None, + specific.value if specific is not None else None, + ) + ) + return frozenset(keys) + case _: + raise TypeError( + f"Unsupported target type for continuation comparison: {type(target)!r}" + ) + + +def is_exact_continuation(predecessor: Dispatch, successor: Dispatch) -> bool: + """Return whether `successor` exactly, compatibly continues `predecessor`. + + A successor continues a predecessor when, and only when: + * they have the same dispatch `type`; + * they have the same `dry_run` mode; + * they resolve to the same `target` (compared order-independently); + * the successor starts EXACTLY when the predecessor ends (`successor.start_time + == predecessor.until`) -- a strict equality, not `<=`. A successor that starts + before the predecessor ends is an OVERLAP, not a continuation, and must be + treated as a fully independent dispatch. + + Args: + predecessor: The dispatch that may be continued. + successor: The dispatch that may continue it. + + Returns: + True if `successor` is an exact, compatible continuation of `predecessor`. + """ + if predecessor.type != successor.type: + return False + if predecessor.dry_run != successor.dry_run: + return False + if _canonical_target(predecessor.target) != _canonical_target(successor.target): + return False + + if not predecessor.active: + # An explicit cancellation is authoritative and is never silently + # "continued through" by a later dispatch, even a coincidentally + # adjacent one -- see the module docstring and `until`'s own behaviour + # (it returns None once `active` is False). + return False + + if predecessor.duration is None: + # Runs indefinitely until explicitly stopped: there is no scheduled end + # for a successor to be adjacent to. NOTE: `Dispatch.until` raises + # ValueError for an active, duration-less dispatch rather than + # returning None -- this check exists specifically to avoid calling it + # in that case. + return False + + predecessor_until = predecessor.until + if predecessor_until is None: + # Should not happen given the two guards above, but avoid ever treating + # a missing boundary as a match. + return False + + return successor.start_time == predecessor_until diff --git a/src/frequenz/dispatch/_dispatcher.py b/src/frequenz/dispatch/_dispatcher.py index 77762c8c..cab4c112 100644 --- a/src/frequenz/dispatch/_dispatcher.py +++ b/src/frequenz/dispatch/_dispatcher.py @@ -312,7 +312,7 @@ def is_managed(self, dispatch_type: str) -> bool: """ return dispatch_type in self._actor_dispatchers - async def start_managing( + async def start_managing( # pylint: disable=too-many-arguments self, dispatch_type: str, *, @@ -320,6 +320,7 @@ async def start_managing( [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor] ], merge_strategy: MergeStrategy | None = None, + continuation_predicate: Callable[[Dispatch, Dispatch], bool] | None = None, retry_interval: timedelta = timedelta(seconds=60), ) -> None: """Manage actors for a given dispatch type. @@ -345,10 +346,34 @@ async def start_managing( the same instance. * `None` — No merging, each dispatch maps to a separate instance. + `continuation_predicate` is a DIFFERENT, orthogonal opt-in mechanism, + not a `merge_strategy`, and may be combined freely with any of the + above (or with `None`). It does NOT filter or suppress anything at + the scheduler/receiver layer -- every event, including every stop, + is always forwarded to the `ActorDispatcher` this method builds. + Passing it wires that `ActorDispatcher` with BOTH the predicate and a + read-only, live view of this service's own known-dispatches snapshot + (`DispatchScheduler.dispatches_snapshot`), so it can require a unique + predecessor-successor PAIR in either event order. One predecessor + matching two successors, or one successor matching two predecessors, + is always ambiguous and never guessed at. Stop-side resolution still + happens AT THE MOMENT that dispatch's own stop is being processed, + without ever needing to delay or withhold that stop while deciding. See + `ActorDispatcher`'s `continuation_predicate`/`dispatches_snapshot` + parameters for the full contract, including the symmetric + start-first-ordering case and the `ContinuationCapable` capability + check that governs whether an actor is ever reused this way at all. + Args: dispatch_type: The type of the dispatch to manage. actor_factory: The factory to create actors. merge_strategy: The strategy to merge running intervals. + continuation_predicate: An opt-in predicate letting a dispatch's + stop or start be resolved as a continuation into/from a + unique, compatible, exactly-adjacent dispatch instead of an + ordinary stop or an independent new actor (normally + `is_exact_continuation`). See `ActorDispatcher`'s own + parameter of the same name for the full contract. retry_interval: The retry interval to use when actor creation fails. """ dispatcher = self._actor_dispatchers.get(dispatch_type) @@ -374,6 +399,12 @@ def id_identity(dispatch: Dispatch) -> DispatchActorId: id_identity if merge_strategy is None else merge_strategy.identity ), retry_interval=retry_interval, + continuation_predicate=continuation_predicate, + dispatches_snapshot=( + self._bg_service.dispatches_snapshot + if continuation_predicate is not None + else None + ), ) self._actor_dispatchers[dispatch_type] = dispatcher diff --git a/tests/test_continuation.py b/tests/test_continuation.py new file mode 100644 index 00000000..883d0ee9 --- /dev/null +++ b/tests/test_continuation.py @@ -0,0 +1,178 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Unit tests for `is_exact_continuation`'s target-equality semantics. + +These are pure, synchronous tests of `_canonical_target`/`is_exact_continuation` +-- no scheduler, no actors, no event loop. Synthetic IDs only. +""" + +from dataclasses import replace +from datetime import datetime, timedelta, timezone + +from frequenz.client.common.microgrid.components import ComponentCategory +from frequenz.client.common.microgrid.electrical_components import ( + ElectricalComponentCategory, +) +from frequenz.client.dispatch.recurrence import RecurrenceRule +from frequenz.client.dispatch.test.generator import DispatchGenerator +from frequenz.client.dispatch.types import ( + BatteryType, + DispatchId, + EvChargerType, + TargetCategories, + TargetIds, +) + +from frequenz.dispatch import Dispatch, is_exact_continuation +from frequenz.dispatch._continuation import _canonical_target + + +def _make_dispatch( + dispatch_id: int, + *, + start_time: datetime, + duration: timedelta | None, + target: TargetIds | TargetCategories, + active: bool = True, + dry_run: bool = False, + dispatch_type: str = "TEST", +) -> Dispatch: + # pylint: disable=too-many-arguments + generator = DispatchGenerator() + spec = replace( + generator.generate_dispatch(), + id=DispatchId(dispatch_id), + type=dispatch_type, + start_time=start_time, + duration=duration, + target=target, + active=active, + dry_run=dry_run, + recurrence=RecurrenceRule(), + ) + return Dispatch(spec) + + +UTC = timezone.utc + + +# --- _canonical_target: order independence --- + + +def test_reordered_target_ids_are_equal() -> None: + """Canonicalize reordered `TargetIds` equally.""" + assert _canonical_target(TargetIds(1, 2, 3)) == _canonical_target( + TargetIds(3, 1, 2) + ) + assert _canonical_target(TargetIds(5, 7)) == _canonical_target(TargetIds(7, 5)) + + +def test_reordered_target_categories_are_equal() -> None: + """Same for a multi-element `TargetCategories`.""" + a = TargetCategories( + ElectricalComponentCategory.BATTERY, ElectricalComponentCategory.INVERTER + ) + b = TargetCategories( + ElectricalComponentCategory.INVERTER, ElectricalComponentCategory.BATTERY + ) + assert _canonical_target(a) == _canonical_target(b) + + +# --- _canonical_target: class-aware, not bare-value, equality --- + + +def test_legacy_and_new_category_enum_for_the_same_category_are_equal() -> None: + """Canonicalize legacy and current spellings of a category equally.""" + legacy = TargetCategories(ComponentCategory.BATTERY) + current = TargetCategories(ElectricalComponentCategory.BATTERY) + assert _canonical_target(legacy) == _canonical_target(current) + + +def test_same_numeric_value_different_kind_is_not_equal() -> None: + """Keep equal numeric values of different target kinds distinct. + + A plain category target and a MORE SPECIFIC sub-type target must NOT be + conflated, even when the broad category matches. A `TargetIds` + whose id happens to numerically match a category's enum value must not be + conflated with that category either (different container kind entirely). + """ + plain_battery = TargetCategories(ElectricalComponentCategory.BATTERY) + specific_li_ion = TargetCategories(BatteryType.LI_ION) + assert _canonical_target(plain_battery) != _canonical_target( + specific_li_ion + ), "A plain 'any battery' target must not equal a specific battery sub-type target" + + battery_value = ElectricalComponentCategory.BATTERY.value + ids_with_matching_number = TargetIds(battery_value) + assert _canonical_target(ids_with_matching_number) != _canonical_target( + plain_battery + ), ( + "A TargetIds whose id numerically matches a category's enum value must " + "not be conflated with that category -- different container kind" + ) + + +def test_different_specific_subtypes_are_not_equal() -> None: + """Two different, equally-specific sub-types are not conflated with each other.""" + li_ion = TargetCategories(BatteryType.LI_ION) + assert _canonical_target(li_ion) != _canonical_target( + TargetCategories(EvChargerType.AC) + ) + + +# --- is_exact_continuation: end-to-end with the target comparison --- + + +def test_continuation_allowed_across_legacy_and_new_category_enum() -> None: + """Allow continuation across legacy and current category spellings.""" + boundary = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) + predecessor = _make_dispatch( + 1, + start_time=boundary - timedelta(minutes=10), + duration=timedelta(minutes=10), + target=TargetCategories(ComponentCategory.BATTERY), + ) + successor = _make_dispatch( + 2, + start_time=boundary, + duration=None, + target=TargetCategories(ElectricalComponentCategory.BATTERY), + ) + assert is_exact_continuation(predecessor, successor) + + +def test_continuation_refused_for_differently_specific_targets() -> None: + """Refuse continuation between differently specific targets.""" + boundary = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) + predecessor = _make_dispatch( + 1, + start_time=boundary - timedelta(minutes=10), + duration=timedelta(minutes=10), + target=TargetCategories(ElectricalComponentCategory.BATTERY), + ) + successor = _make_dispatch( + 2, + start_time=boundary, + duration=None, + target=TargetCategories(BatteryType.LI_ION), + ) + assert not is_exact_continuation(predecessor, successor) + + +def test_continuation_allowed_with_reordered_multi_id_target() -> None: + """Allow continuation with reordered component IDs.""" + boundary = datetime(2026, 1, 1, 12, 0, tzinfo=UTC) + predecessor = _make_dispatch( + 1, + start_time=boundary - timedelta(minutes=10), + duration=timedelta(minutes=10), + target=TargetIds(1, 2, 3), + ) + successor = _make_dispatch( + 2, + start_time=boundary, + duration=None, + target=TargetIds(3, 2, 1), + ) + assert is_exact_continuation(predecessor, successor) diff --git a/tests/test_continuation_realclock.py b/tests/test_continuation_realclock.py new file mode 100644 index 00000000..2a886a07 --- /dev/null +++ b/tests/test_continuation_realclock.py @@ -0,0 +1,573 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Real-clock, real-scheduler integration tests for the continuation prototype. + +Deliberately NO `time_machine`, NO `async_solipsism`: this uses the DEFAULT +asyncio event loop and genuine `datetime.now()`, with SHORT real dispatch +durations (a few hundred milliseconds), so the whole suite still runs in a +few seconds while every clock involved -- `asyncio`'s own loop time and +`DispatchScheduler`'s `Timer` -- is the SAME, single, real clock. This sidesteps +the `time_machine`/`async_solipsism`/`frequenz.channels.timer.Timer` +interaction that made `test_continuation_integration.py` unreliable; it is not +a replacement for that file's intent (real `Dispatcher.start_managing` + real +`DispatchScheduler` + the official `FakeClient` test double, end-to-end, +exercising deletion, ambiguity and coincident boundaries), just a different, +real-time way of driving the same real components. + +Assertions are event-based (`asyncio.wait_for` polling a condition), never a +fixed sleep-and-hope. + +Synthetic component ID 1 only. No network/API/host calls. +""" + +# pylint: disable=protected-access + +import asyncio +from collections.abc import Generator +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, cast +from unittest.mock import patch + +import pytest +from frequenz.channels import Receiver +from frequenz.client.common.microgrid import MicrogridId +from frequenz.client.dispatch.test.client import FakeClient +from frequenz.client.dispatch.types import TargetIds +from frequenz.sdk.actor import Actor + +from frequenz.dispatch import Dispatcher, DispatchInfo, is_exact_continuation +from frequenz.dispatch._actor_dispatcher import ActorDispatcher, DispatchActorId + +REAL_EVENT_LOOP_POLICY = asyncio.get_event_loop_policy() + + +@pytest.fixture +def event_loop_policy() -> Generator[asyncio.AbstractEventLoopPolicy, None, None]: + """Pin the real default asyncio event loop policy for this file. + + `test_frequenz_dispatch.py` defines an `event_loop_policy` fixture that + calls `asyncio.set_event_loop_policy(async_solipsism.EventLoopPolicy())` + -- a GLOBAL, process-wide mutation it never restores. pytest-asyncio's own + default `event_loop_policy` fixture just returns whatever the current + global policy is, so once that other file's tests have run in the same + process, every later file silently inherits a solipsism (manually-ticked, + never-really-advancing) loop -- exactly the shape of hang this file is + designed not to have. Defining this fixture here pins the real default + regardless of what ran before, independent of execution order. Restore + the previous global policy explicitly so this fixture cannot leak its own + process-wide mutation into tests that run after this file. + """ + previous_policy = asyncio.get_event_loop_policy() + asyncio.set_event_loop_policy(REAL_EVENT_LOOP_POLICY) + try: + yield REAL_EVENT_LOOP_POLICY + finally: + asyncio.set_event_loop_policy(previous_policy) + + +# The real dispatch API (via FakeClient, matching the real protobuf wire +# format) rounds `start_time` to the nearest second and `duration` with +# `round(seconds)` (confirmed empirically: a 400ms duration round-trips as +# exactly 0, a sub-second start_time has its microseconds truncated) -- so +# anything sub-second here is silently lossy. These are whole seconds, +# deliberately: short enough for a fast suite, long enough to round-trip +# exactly. +DURATION = timedelta(seconds=2) +STARTUP_MARGIN = timedelta(seconds=1) + + +def _now() -> datetime: + return datetime.now(tz=timezone.utc) + + +async def _wait_until( + predicate: Any, *, timeout: float = 3.0, interval: float = 0.01 +) -> None: + """Poll `predicate()` (real, short sleeps) until true, or raise.""" + + async def _poll() -> None: + while not predicate(): + await asyncio.sleep(interval) + + await asyncio.wait_for(_poll(), timeout=timeout) + + +class RecordingActor(Actor): + """Track the dispatch ID, received updates, and proposed power. + + The power is a `target_power_w` payload field stored directly. This actor + is NOT continuation-aware: it has no + `accept_continuation` method, so `isinstance(actor, ContinuationCapable)` + is `False`. + """ + + def __init__( + self, initial_dispatch: DispatchInfo, receiver: Receiver[DispatchInfo] + ) -> None: + """Initialize the recording actor.""" + super().__init__(name=f"actor-for-{initial_dispatch._src.id}") + self.current_dispatch_id = initial_dispatch._src.id + self.proposed_power_w: float | None = initial_dispatch.options.get( + "target_power_w" + ) + self.stop_called = False + self.receiver = receiver + + async def _run(self) -> None: + async for update in self.receiver: + if update._src.id == self.current_dispatch_id: + self.proposed_power_w = update.options.get("target_power_w") + + async def stop(self, msg: str | None = None) -> None: + """Stop, recording that it happened and withdrawing (like a real actor).""" + self.stop_called = True + self.proposed_power_w = None + await super().stop(msg) + + @classmethod + async def create( + cls, initial_dispatch: DispatchInfo, receiver: Receiver[DispatchInfo] + ) -> "RecordingActor": + """Create a new actor.""" + return cls(initial_dispatch, receiver) + + +class ContinuationAwareActor(RecordingActor): + """Implements `ContinuationCapable.accept_continuation`.""" + + async def accept_continuation(self, update: DispatchInfo) -> bool: + """Accept a continuation update, immediately applying its payload.""" + if update.continuation_of != self.current_dispatch_id: + return False + self.current_dispatch_id = update._src.id + self.proposed_power_w = update.options.get("target_power_w") + return True + + +@dataclass +class _Env: + dispatcher: Dispatcher + client: FakeClient + microgrid_id: MicrogridId + + +@pytest.fixture +async def env() -> _Env: + """Create a dispatcher on the real-time asyncio event loop. + + The dispatcher is backed by the official `FakeClient`, with no + time-machine or async-solipsism clock. + """ + + class _FakeClientWithSignature(FakeClient): + # pylint: disable=too-many-arguments,unused-argument + def __init__( + self, + *, + server_url: str, + auth_key: str | None = None, + key: str | None = None, + sign_secret: str | None = None, + call_timeout: timedelta, + stream_timeout: timedelta, + ) -> None: + super().__init__() + + # pylint: enable=too-many-arguments,unused-argument + + mid = MicrogridId(1) + with patch( + "frequenz.dispatch._dispatcher.DispatchApiClient", _FakeClientWithSignature + ): + dispatcher = Dispatcher( + microgrid_id=mid, server_url="grpc://test-url", key="test-key" + ) + dispatcher.start() + await asyncio.sleep(0.2) # let its background _run() task actually start + return _Env( + dispatcher=dispatcher, + client=cast(FakeClient, dispatcher.client), + microgrid_id=mid, + ) + + +def _actor_dispatcher_for( + dispatcher: Dispatcher, dispatch_type: str +) -> ActorDispatcher: + return dispatcher._actor_dispatchers[dispatch_type] + + +async def test_exact_adjacency_end_to_end_real_scheduler(env: _Env) -> None: + """Continue an exactly adjacent dispatch with the real scheduler. + + The predecessor naturally starts, naturally reaches its + own scheduled end, and the real scheduler's own Timer fires the stop -- + which `ActorDispatcher` resolves into a continuation using the real + `dispatches_snapshot`, all without this test ever manually constructing + or feeding any event. + """ + dispatch_type = "REALCLOCK_EXACT_ADJACENCY" + await env.dispatcher.start_managing( + dispatch_type=dispatch_type, + actor_factory=ContinuationAwareActor.create, + continuation_predicate=is_exact_continuation, + ) + actor_dispatcher = _actor_dispatcher_for(env.dispatcher, dispatch_type) + + start = _now() + STARTUP_MARGIN + boundary = start + DURATION + predecessor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 1_200_000.0}, + ) + successor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=boundary, + duration=None, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 900_000.0}, # deliberately different + ) + + await _wait_until( + lambda: DispatchActorId(predecessor.id) in actor_dispatcher._actors + ) + actor = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(predecessor.id)].actor, + ) + await _wait_until(lambda: actor.proposed_power_w == 1_200_000.0) + + # Real time passes naturally past `boundary` -- the real scheduler's own + # Timer (with the floor-fix) fires the real stop event for `predecessor`. + await _wait_until( + lambda: actor.current_dispatch_id == successor.id + and actor.proposed_power_w == 900_000.0, + timeout=5.0, + ) + + assert ( + not actor.stop_called + ), "Expected the actor to be reused via continuation, never stopped" + assert len(actor_dispatcher._actors) == 1, "Expected exactly one live actor, reused" + + +async def test_successor_deleted_after_continuation_real_scheduler(env: _Env) -> None: + """Stop the reused actor when its successor is deleted.""" + dispatch_type = "REALCLOCK_SUCCESSOR_DELETED" + await env.dispatcher.start_managing( + dispatch_type=dispatch_type, + actor_factory=ContinuationAwareActor.create, + continuation_predicate=is_exact_continuation, + ) + actor_dispatcher = _actor_dispatcher_for(env.dispatcher, dispatch_type) + + start = _now() + STARTUP_MARGIN + boundary = start + DURATION + predecessor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={}, + ) + successor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=boundary, + duration=None, + target=TargetIds(1), + active=True, + dry_run=False, + payload={}, + ) + + await _wait_until( + lambda: DispatchActorId(predecessor.id) in actor_dispatcher._actors + ) + await _wait_until( + lambda: len(actor_dispatcher._actors) == 1 + and next(iter(actor_dispatcher._actors.values())).current_dispatch.id + == successor.id, + timeout=5.0, + ) + reused_actor = cast( + ContinuationAwareActor, next(iter(actor_dispatcher._actors.values())).actor + ) + + await env.client.delete(microgrid_id=env.microgrid_id, dispatch_id=successor.id) + + await _wait_until(lambda: reused_actor.stop_called, timeout=3.0) + await _wait_until(lambda: len(actor_dispatcher._actors) == 0, timeout=3.0) + + +async def test_ambiguous_successor_real_scheduler(env: _Env) -> None: + """Refuse two successors that both continue the same predecessor. + + The scheduler orders START before STOP at a shared instant. Pairwise + uniqueness therefore has to reject this ambiguity on the start side, + before either successor can arbitrarily claim the predecessor's actor. + Both successors must start independently and the predecessor must stop, + withdrawing its old proposal. + """ + dispatch_type = "REALCLOCK_AMBIGUOUS_SUCCESSOR" + await env.dispatcher.start_managing( + dispatch_type=dispatch_type, + actor_factory=ContinuationAwareActor.create, + continuation_predicate=is_exact_continuation, + ) + actor_dispatcher = _actor_dispatcher_for(env.dispatcher, dispatch_type) + + start = _now() + STARTUP_MARGIN + boundary = start + DURATION + predecessor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 1_200_000.0}, + ) + successor_a = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=boundary, + duration=None, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 900_000.0}, + ) + successor_b = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=boundary, + duration=None, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 800_000.0}, + ) + + await _wait_until( + lambda: DispatchActorId(predecessor.id) in actor_dispatcher._actors + ) + predecessor_actor = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(predecessor.id)].actor, + ) + + # Whichever race resolution occurs (see the docstring above), it must + # settle into EXACTLY two live actors/identities, one correctly + # representing each successor's own payload -- never fewer (an orphan) + # and never more (a double-claim), and never both successors pointing at + # the SAME actor (a corrupted shared re-key). + await _wait_until( + lambda: len(actor_dispatcher._actors) == 2 + and DispatchActorId(successor_a.id) in actor_dispatcher._actors + and DispatchActorId(successor_b.id) in actor_dispatcher._actors, + timeout=5.0, + ) + + actor_a = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(successor_a.id)].actor, + ) + actor_b = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(successor_b.id)].actor, + ) + assert ( + actor_a is not actor_b + ), "The two successors must never share the same re-keyed actor" + assert predecessor_actor is not actor_a + assert predecessor_actor is not actor_b + assert predecessor_actor.stop_called + assert predecessor_actor.proposed_power_w is None + assert actor_a.current_dispatch_id == successor_a.id + assert actor_b.current_dispatch_id == successor_b.id + assert actor_a.proposed_power_w == 900_000.0 + assert actor_b.proposed_power_w == 800_000.0 + + +async def test_ambiguous_predecessor_real_scheduler(env: _Env) -> None: + """Refuse two predecessors that both match one successor. + + The successor must start independently, then both predecessor actors must + stop and withdraw their own proposals. Stop-side processing must not + allow the second predecessor to claim the successor merely because the + first predecessor has already been stopped. + """ + dispatch_type = "REALCLOCK_AMBIGUOUS_PREDECESSOR" + await env.dispatcher.start_managing( + dispatch_type=dispatch_type, + actor_factory=ContinuationAwareActor.create, + continuation_predicate=is_exact_continuation, + ) + actor_dispatcher = _actor_dispatcher_for(env.dispatcher, dispatch_type) + + start = _now() + STARTUP_MARGIN + boundary = start + DURATION + predecessor_a = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 1_200_000.0}, + ) + predecessor_b = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 800_000.0}, + ) + successor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=boundary, + duration=None, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 900_000.0}, + ) + + await _wait_until( + lambda: DispatchActorId(predecessor_a.id) in actor_dispatcher._actors + and DispatchActorId(predecessor_b.id) in actor_dispatcher._actors + ) + actor_a = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(predecessor_a.id)].actor, + ) + actor_b = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(predecessor_b.id)].actor, + ) + + await _wait_until( + lambda: len(actor_dispatcher._actors) == 1 + and DispatchActorId(successor.id) in actor_dispatcher._actors, + timeout=5.0, + ) + successor_actor = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(successor.id)].actor, + ) + + assert successor_actor is not actor_a + assert successor_actor is not actor_b + assert successor_actor.current_dispatch_id == successor.id + assert successor_actor.proposed_power_w == 900_000.0 + assert actor_a.stop_called and actor_a.proposed_power_w is None + assert actor_b.stop_called and actor_b.proposed_power_w is None + + +async def test_snapshot_predecessor_without_actor_still_makes_pair_ambiguous( + env: _Env, +) -> None: + """Count snapshot predecessors even when one actor failed to start. + + Start-side matching must not reuse the sole registered predecessor actor + when another known dispatch also matches the successor but has no actor. + """ + dispatch_type = "REALCLOCK_SNAPSHOT_ONLY_PREDECESSOR" + + async def selective_factory( + initial_dispatch: DispatchInfo, receiver: Receiver[DispatchInfo] + ) -> ContinuationAwareActor: + if initial_dispatch.options.get("fail_start"): + raise RuntimeError("synthetic actor start failure") + return cast( + ContinuationAwareActor, + await ContinuationAwareActor.create(initial_dispatch, receiver), + ) + + await env.dispatcher.start_managing( + dispatch_type=dispatch_type, + actor_factory=selective_factory, + continuation_predicate=is_exact_continuation, + ) + actor_dispatcher = _actor_dispatcher_for(env.dispatcher, dispatch_type) + + start = _now() + STARTUP_MARGIN + boundary = start + DURATION + registered_predecessor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 1_200_000.0}, + ) + snapshot_only_predecessor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=start, + duration=DURATION, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 800_000.0, "fail_start": True}, + ) + successor = await env.client.create( + microgrid_id=env.microgrid_id, + type=dispatch_type, + start_time=boundary, + duration=None, + target=TargetIds(1), + active=True, + dry_run=False, + payload={"target_power_w": 900_000.0}, + ) + + registered_identity = DispatchActorId(registered_predecessor.id) + failed_identity = DispatchActorId(snapshot_only_predecessor.id) + await _wait_until( + lambda: registered_identity in actor_dispatcher._actors + and failed_identity in actor_dispatcher._failed_dispatches + ) + predecessor_actor = cast( + ContinuationAwareActor, + actor_dispatcher._actors[registered_identity].actor, + ) + + await _wait_until( + lambda: len(actor_dispatcher._actors) == 1 + and DispatchActorId(successor.id) in actor_dispatcher._actors, + timeout=5.0, + ) + successor_actor = cast( + ContinuationAwareActor, + actor_dispatcher._actors[DispatchActorId(successor.id)].actor, + ) + + assert successor_actor is not predecessor_actor + assert successor_actor.current_dispatch_id == successor.id + assert successor_actor.proposed_power_w == 900_000.0 + assert predecessor_actor.stop_called + assert predecessor_actor.proposed_power_w is None From dc12de45767f09429154a47801753fb3dfd095c6 Mon Sep 17 00:00:00 2001 From: "Mathias L. Baumann" Date: Thu, 15 Oct 2026 18:39:35 +0200 Subject: [PATCH 3/4] Add release notes for dispatch handovers Signed-off-by: Mathias L. Baumann --- RELEASE_NOTES.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 6b1698ed..c586d256 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 + +- Added opt-in support for reusing a continuation-capable actor across exactly adjacent, compatible dispatches. Callers must provide a continuation predicate; ambiguous or unsupported handovers retain the normal stop/start behavior. ## 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`. +- Prevented scheduler crashes when coincident events produced a non-positive timer interval. From e1fc534a2d31a7aa5968439a37742d0c78c1ea57 Mon Sep 17 00:00:00 2001 From: "Mathias L. Baumann" Date: Thu, 15 Oct 2026 18:44:43 +0200 Subject: [PATCH 4/4] Update mkdocstrings for pymdown compatibility Signed-off-by: Mathias L. Baumann --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index d944604c..9df37046 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,7 +70,7 @@ dev-mkdocs = [ "mkdocs-literate-nav == 0.6.3", "mkdocs-macros-plugin == 1.5.0", "mkdocs-material == 9.7.7", - "mkdocstrings[python] == 1.0.6", + "mkdocstrings[python] == 1.0.7", "mkdocstrings-python == 2.0.8", "frequenz-repo-config[lib] == 0.19.0", ]