Wait for dispatches to be fetched before creating a receiver

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

Author: Marenz

RELEASE_NOTES.md

@@ -9,7 +9,7 @@ This release introduces a more flexible and powerful mechanism for managing disp
 * `Dispatcher.start` is no longer `async`. Remove `await` when calling it.
 * Two properties have been replaced by methods that require a type as parameter.
     * `Dispatcher.lifecycle_events` has been replaced by the method `Dispatcher.new_lifecycle_events_receiver(self, dispatch_type: str)`.
-    * `Dispatcher.running_status_change` has been replaced by the method `Dispatcher.new_running_state_event_receiver(self, dispatch_type: str, merge_strategy: MergeStrategy)`.
+    * `Dispatcher.running_status_change` has been replaced by the method `Dispatcher.new_running_state_event_receiver(self, dispatch_type: str, merge_strategy: MergeStrategy, initial_fetch_timeout: timedelta)`.
 * The managing actor constructor no longer requires the `dispatch_type` parameter. Instead you're expected to pass the type to the new-receiver function.
 * The `DispatchManagingActor` class has been renamed to `DispatchActorsService`.
     * It's interface has been simplified and now only requires an actor factory and a running status receiver.
@@ -19,6 +19,7 @@ This release introduces a more flexible and powerful mechanism for managing disp
 
 ## New Features
 
+* A new parameter `initial_fetch_timeout` has been added to `Dispatcher.new_running_state_event_receiver` and related methods. It allows you to specify a timeout for the initial fetch of dispatches. After the timeout, startup continues as normal.
 * A new feature "merge strategy" (`MergeByType`, `MergeByTypeTarget`) has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
 * The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1600.
 * Actor management with dispatches has been simplified:

src/frequenz/dispatch/_bg_service.py

@@ -129,6 +129,9 @@ def __init__(
         always at index 0.
         """
 
+        self._initial_fetch_event = asyncio.Event()
+        """The initial fetch event."""
+
     # pylint: disable=redefined-builtin
     def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         """Create a new receiver for lifecycle events.
@@ -144,7 +147,11 @@ def new_lifecycle_events_receiver(self, type: str) -> Receiver[DispatchEvent]:
         )
 
     async def new_running_state_event_receiver(
-        self, type: str, *, merge_strategy: MergeStrategy | None = None
+        self,
+        type: str,
+        *,
+        merge_strategy: MergeStrategy | None = None,
+        initial_fetch_timeout: timedelta = timedelta(seconds=10),
     ) -> Receiver[Dispatch]:
         """Create a new receiver for running state events of the specified type.
 
@@ -168,12 +175,25 @@ async def new_running_state_event_receiver(
         While merging, stop events are ignored as long as at least one
         merge-criteria-matching dispatch remains active.
 
+        Will wait up to `initial_fetch_timeout` for the initial fetch to
+        complete to ensure that all dispatches are available but to also avoid
+        waiting indefinitely.
+
         Args:
             type: The type of events to receive.
             merge_strategy: The merge strategy to use.
+            initial_fetch_timeout: The timeout for the initial fetch.
+
         Returns:
             A new receiver for running state status.
         """
+
+        # Wait with timeout for the initial fetch to complete
+        await asyncio.wait_for(
+            self._initial_fetch_event.wait(),
+            timeout=initial_fetch_timeout.total_seconds(),
+        )
+
         # Find all matching dispatches based on the type and collect them
         dispatches = [
             dispatch for dispatch in self._dispatches.values() if dispatch.type == type
@@ -272,6 +292,8 @@ async def _fetch(self) -> None:
         This is used for the initial fetch and for re-fetching all dispatches
         if the connection was lost.
         """
+        self._initial_fetch_event.clear()
+
         old_dispatches = self._dispatches
         self._dispatches = {}
 
@@ -311,6 +333,8 @@ async def _fetch(self) -> None:
             dispatch._set_deleted()  # pylint: disable=protected-access
             await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
 
+        self._initial_fetch_event.set()
+
     async def _update_dispatch_schedule_and_notify(
         self, dispatch: Dispatch | None, old_dispatch: Dispatch | None
     ) -> None:

src/frequenz/dispatch/_dispatcher.py

@@ -8,6 +8,7 @@
 import asyncio
 import logging
 from asyncio import Event
+from datetime import timedelta
 from typing import Awaitable, Callable
 
 from frequenz.channels import Receiver
@@ -240,6 +241,7 @@ async def start_dispatching(
             [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
         ],
         merge_strategy: MergeStrategy | None = None,
+        initial_fetch_timeout: timedelta = timedelta(seconds=10),
     ) -> None:
         """Manage actors for a given dispatch type.
 
@@ -254,6 +256,10 @@ async def start_dispatching(
             dispatch_type: The type of the dispatch to manage.
             actor_factory: The factory to create actors.
             merge_strategy: The strategy to merge running intervals.
+            initial_fetch_timeout: Timeout for the initial fetch of dispatches.
+                After the timeout, the receiver will continue to start and
+                listen for new dispatches, while the initial fetch is still
+                running in the background.
         """
         dispatcher = self._actor_dispatchers.get(dispatch_type)
 
@@ -271,7 +277,9 @@ def id_identity(dispatch: Dispatch) -> int:
         dispatcher = ActorDispatcher(
             actor_factory=actor_factory,
             running_status_receiver=await self.new_running_state_event_receiver(
-                dispatch_type, merge_strategy=merge_strategy
+                dispatch_type,
+                merge_strategy=merge_strategy,
+                initial_fetch_timeout=initial_fetch_timeout,
             ),
             dispatch_identity=(
                 id_identity if merge_strategy is None else merge_strategy.identity
@@ -317,6 +325,7 @@ async def new_running_state_event_receiver(
         dispatch_type: str,
         *,
         merge_strategy: MergeStrategy | None = None,
+        initial_fetch_timeout: timedelta = timedelta(seconds=10),
     ) -> Receiver[Dispatch]:
         """Return running state event receiver.
 
@@ -363,10 +372,16 @@ async def new_running_state_event_receiver(
         Args:
             dispatch_type: The type of the dispatch to listen for.
             merge_strategy: The type of the strategy to merge running intervals.
+            initial_fetch_timeout: Timeout for the initial fetch of dispatches.
+                After the timeout, the receiver will continue to start and
+                listen for new dispatches, while the initial fetch is still
+                running in the background
 
         Returns:
             A new receiver for dispatches whose running status changed.
         """
         return await self._bg_service.new_running_state_event_receiver(
-            dispatch_type, merge_strategy=merge_strategy
+            dispatch_type,
+            merge_strategy=merge_strategy,
+            initial_fetch_timeout=initial_fetch_timeout,
         )