Reconnect stream if there has been no event in 5 minutes

Signed-off-by: Mathias L. Baumann <mathias.baumann@frequenz.com>

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.
+* Introduced a timeout to the stream combined with a periodic refresh on reconnect. This allows the stream to be more resilient against network issues.
 
 ## Bug Fixes
 

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
   # plugins.mkdocstrings.handlers.python.import)
   "frequenz-sdk >= 1.0.0-rc1302, < 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",
 ]
 dynamic = ["version"]
 

src/frequenz/dispatch/_bg_service.py

@@ -132,6 +132,9 @@ def __init__(
         self._initial_fetch_event = asyncio.Event()
         """The initial fetch event."""
 
+        self._last_refresh = datetime.now(timezone.utc)
+        """Time of the last stream event, used to detect connection issues."""
+
     async def wait_for_initialization(self) -> None:
         """Wait for the initial fetch to complete."""
         await self._initial_fetch_event.wait()
@@ -234,17 +237,30 @@ async def _run(self) -> None:
         await self._fetch()
 
         stream = self._client.stream(microgrid_id=self._microgrid_id)
+        stream_reconnect_period = self._client.stream_timeout
 
         # Streaming updates
         async for selected in select(self._next_event_timer, stream):
+            now = datetime.now(timezone.utc)
             if selected_from(selected, self._next_event_timer):
+                if self._last_refresh + stream_reconnect_period < now:
+                    _logger.info(
+                        "Periodic refresh of dispatches for microgrid %s",
+                        self._microgrid_id,
+                    )
+                    # When _client.stream reconnects after the timeout, perform
+                    # a refresh to ensure we didn't miss any updates
+                    await self._fetch()
+                    self._last_refresh = now
+                    # pylint: enable=protected-access
                 if not self._scheduled_events:
                     continue
                 await self._execute_scheduled_event(
                     heappop(self._scheduled_events).dispatch
                 )
             elif selected_from(selected, stream):
                 _logger.debug("Received dispatch event: %s", selected.message)
+                self._last_refresh = now
                 dispatch = Dispatch(selected.message.dispatch)
                 match selected.message.event:
                     case Event.CREATED: