Reconnect stream if there has been no event in 5 minutes

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.
+* If there has been no dispatch event for 5 minutes, the dispatch client will force-reconnect the stream
 
 ## Bug Fixes
 

src/frequenz/dispatch/_bg_service.py

@@ -132,6 +132,9 @@ def __init__(
         self._initial_fetch_event = asyncio.Event()
         """The initial fetch event."""
 
+        self._last_stream_event = 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,29 @@ async def _run(self) -> None:
         await self._fetch()
 
         stream = self._client.stream(microgrid_id=self._microgrid_id)
+        max_event_inactivity = timedelta(minutes=5)
 
         # 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_stream_event + max_event_inactivity < now:
+                    _logger.info("No stream events for %s minutes, restarting stream")
+                    # This restarts the stream task but keeps the channel
+                    # pylint: disable=protected-access
+                    broadcaster = self._client._get_stream(self._microgrid_id)
+                    broadcaster._task.cancel()
+                    broadcaster._task = asyncio.create_task(self._run())
+                    self._last_stream_event = 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_stream_event = now
                 dispatch = Dispatch(selected.message.dispatch)
                 match selected.message.event:
                     case Event.CREATED: