Refresh the dispatches after a stream restart.

To ensure we didn't miss any.

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

Author: Marenz

RELEASE_NOTES.md

@@ -10,7 +10,7 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* While the dispatch stream restarts we refresh our dispatch cache as well, to ensure we didn't miss any updates.
 
 ## Bug Fixes
 

pyproject.toml

@@ -38,9 +38,12 @@ dependencies = [
   # Make sure to update the version for cross-referencing also in the
   # mkdocs.yml file when changing the version here (look for the config key
   # plugins.mkdocstrings.handlers.python.import)
-  "frequenz-sdk >= 1.0.0-rc1900, < 1.0.0-rc2100",
+  #"frequenz-sdk >= 1.0.0-rc1900, < 1.0.0-rc2100",
+  "frequenz-sdk @ git+https://github.com/frequenz-floss/frequenz-sdk-python.git@refs/pull/1230/head",
   "frequenz-channels >= 1.6.1, < 2.0.0",
-  "frequenz-client-dispatch >= 0.10.1, < 0.11.0",
+  #  "frequenz-client-dispatch >= 0.10.2, < 0.11.0",
+  "frequenz-client-dispatch @ git+https://github.com/frequenz-floss/frequenz-client-dispatch-python.git@refs/pull/180/head",
+  "frequenz-client-base >= 0.11.0, < 0.12.0",
 ]
 dynamic = ["version"]
 

src/frequenz/dispatch/_bg_service.py

@@ -17,7 +17,13 @@
 import grpc.aio
 from frequenz.channels import Broadcast, Receiver, select, selected_from
 from frequenz.channels.timer import SkipMissedAndResync, Timer
+from frequenz.client.base.streaming import (
+    StreamFatalError,
+    StreamRetrying,
+    StreamStarted,
+)
 from frequenz.client.dispatch import DispatchApiClient
+from frequenz.client.dispatch.types import DispatchEvent as ApiDispatchEvent
 from frequenz.client.dispatch.types import Event
 from frequenz.sdk.actor import BackgroundService
 
@@ -233,7 +239,18 @@ async def _run(self) -> None:
         # Initial fetch
         await self._fetch()
 
-        stream = self._client.stream(microgrid_id=self._microgrid_id)
+        # pylint: disable-next=protected-access
+        streamer = self._client._get_stream(microgrid_id=self._microgrid_id)
+        stream = streamer.new_receiver(include_events=True)
+
+        # We track stream start events linked to retries to avoid re-fetching
+        # dispatches that were already retrieved during an initial stream start.
+        # The initial fetch gets all dispatches, and the StreamStarted event
+        # isn't always reliable due to parallel receiver creation and stream
+        # task initiation.
+        # This way we get a deterministic behavior where we only fetch
+        # dispatches once initially and then only when the stream is restarted.
+        is_retry_attempt = False
 
         # Streaming updates
         async for selected in select(self._next_event_timer, stream):
@@ -244,25 +261,50 @@ async def _run(self) -> None:
                     heappop(self._scheduled_events).dispatch
                 )
             elif selected_from(selected, stream):
-                _logger.debug("Received dispatch event: %s", selected.message)
-                dispatch = Dispatch(selected.message.dispatch)
-                match selected.message.event:
-                    case Event.CREATED:
-                        self._dispatches[dispatch.id] = dispatch
-                        await self._update_dispatch_schedule_and_notify(dispatch, None)
-                        await self._lifecycle_events_tx.send(Created(dispatch=dispatch))
-                    case Event.UPDATED:
-                        await self._update_dispatch_schedule_and_notify(
-                            dispatch, self._dispatches[dispatch.id]
-                        )
-                        self._dispatches[dispatch.id] = dispatch
-                        await self._lifecycle_events_tx.send(Updated(dispatch=dispatch))
-                    case Event.DELETED:
-                        self._dispatches.pop(dispatch.id)
-                        await self._update_dispatch_schedule_and_notify(None, dispatch)
-
-                        await self._lifecycle_events_tx.send(Deleted(dispatch=dispatch))
-
+                match selected.message:
+                    case ApiDispatchEvent():
+                        _logger.debug("Received dispatch event: %s", selected.message)
+                        dispatch = Dispatch(selected.message.dispatch)
+                        match selected.message.event:
+                            case Event.CREATED:
+                                self._dispatches[dispatch.id] = dispatch
+                                await self._update_dispatch_schedule_and_notify(
+                                    dispatch, None
+                                )
+                                await self._lifecycle_events_tx.send(
+                                    Created(dispatch=dispatch)
+                                )
+                            case Event.UPDATED:
+                                await self._update_dispatch_schedule_and_notify(
+                                    dispatch, self._dispatches[dispatch.id]
+                                )
+                                self._dispatches[dispatch.id] = dispatch
+                                await self._lifecycle_events_tx.send(
+                                    Updated(dispatch=dispatch)
+                                )
+                            case Event.DELETED:
+                                self._dispatches.pop(dispatch.id)
+                                await self._update_dispatch_schedule_and_notify(
+                                    None, dispatch
+                                )
+
+                                await self._lifecycle_events_tx.send(
+                                    Deleted(dispatch=dispatch)
+                                )
+
+                    case StreamRetrying():
+                        is_retry_attempt = True
+
+                    case StreamStarted():
+                        if is_retry_attempt:
+                            _logger.info(
+                                "Dispatch stream restarted, getting dispatches"
+                            )
+                            await self._fetch()
+                            is_retry_attempt = False
+
+                    case StreamFatalError():
+                        pass
     async def _execute_scheduled_event(self, dispatch: Dispatch) -> None:
         """Execute a scheduled event.