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

@@ -12,6 +12,7 @@
 
 * This release now supports the sdk up to rc2000.
 * Less logs are now on `INFO` level, and more on `DEBUG` level, making the output less verbose.
+* 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,11 @@ 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-rc1302, < 1.0.0-rc2000",
+  "frequenz-sdk >= 1.0.0-rc1900, < 1.0.0-rc2000",
   "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-base >= 0.11.0, < 0.13.0",
+  "frequenz-client-base @ git+https://github.com/frequenz-floss/frequenz-client-base-python@refs/pull/146/head",
 ]
 dynamic = ["version"]
 

src/frequenz/dispatch/_bg_service.py

@@ -17,7 +17,9 @@
 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 StreamStartedEvent
 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
 
@@ -244,24 +246,40 @@ 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 StreamStartedEvent():
+                        _logger.info("Dispatch stream restarted, refreshing dispatches")
+                        await self._fetch()
+
+                    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)
+                                )
 
     async def _execute_scheduled_event(self, dispatch: Dispatch) -> None:
         """Execute a scheduled event.