Add opt-in contiguous dispatch handovers

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
+
+- 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.

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",
 ]

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",
 ]

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":

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

src/frequenz/dispatch/_dispatcher.py

@@ -312,14 +312,15 @@ 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,
         *,
         actor_factory: Callable[
             [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