Add new method Dispatcher.refresh() to fetch latest dispatches

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

Author: Marenz

RELEASE_NOTES.md

@@ -14,7 +14,7 @@
 
 ## New Features
 
-<!-- Here goes the main new features and examples or instructions on how to use them -->
+* A new method `Dispatcher.refresh()` was added, for when you need to refresh the dispatcher's state.
 
 ## Bug Fixes
 

src/frequenz/dispatch/_dispatcher.py

@@ -210,6 +210,10 @@ async def start(self) -> None:
         """Start the actor."""
         self._actor.start()
 
+    async def refresh(self) -> None:
+        """Re-fetch all dispatches."""
+        await self._actor.fetch()
+
     @property
     def client(self) -> Client:
         """Return the client."""

src/frequenz/dispatch/actor.py

@@ -22,6 +22,7 @@
 """The logger for this module."""
 
 
+# pylint: disable=too-many-instance-attributes
 class DispatchingActor(Actor):
     """Dispatch actor.
 
@@ -84,12 +85,14 @@ def __init__(
         always at index 0.
         """
 
+        self._currently_fetching = False
+
     async def _run(self) -> None:
         """Run the actor."""
         _logger.info("Starting dispatch actor for microgrid %s", self._microgrid_id)
 
         # Initial fetch
-        await self._fetch()
+        await self.fetch()
 
         stream = self._client.stream(microgrid_id=self._microgrid_id)
 
@@ -151,52 +154,58 @@ async def _execute_scheduled_event(self, dispatch: Dispatch) -> None:
 
         self._update_timer()
 
-    async def _fetch(self) -> None:
+    async def fetch(self) -> None:
         """Fetch all relevant dispatches using list.
 
         This is used for the initial fetch and for re-fetching all dispatches
         if the connection was lost.
         """
-        old_dispatches = self._dispatches
-        self._dispatches = {}
-
-        try:
-            _logger.info("Fetching dispatches for microgrid %s", self._microgrid_id)
-            async for page in self._client.list(microgrid_id=self._microgrid_id):
-                for client_dispatch in page:
-                    dispatch = Dispatch(client_dispatch)
-
-                    self._dispatches[dispatch.id] = Dispatch(client_dispatch)
-                    old_dispatch = old_dispatches.pop(dispatch.id, None)
-                    if not old_dispatch:
-                        _logger.info("New dispatch: %s", dispatch)
-                        await self._update_dispatch_schedule_and_notify(dispatch, None)
-                        await self._lifecycle_updates_sender.send(
-                            Created(dispatch=dispatch)
-                        )
-                    elif dispatch.update_time != old_dispatch.update_time:
-                        _logger.info("Updated dispatch: %s", dispatch)
-                        await self._update_dispatch_schedule_and_notify(
-                            dispatch, old_dispatch
-                        )
-                        await self._lifecycle_updates_sender.send(
-                            Updated(dispatch=dispatch)
-                        )
-
-        except grpc.aio.AioRpcError as error:
-            _logger.error("Error fetching dispatches: %s", error)
-            self._dispatches = old_dispatches
+        if self._currently_fetching:
+            _logger.debug("Already fetching dispatches, skipping")
             return
 
-        for dispatch in old_dispatches.values():
-            _logger.info("Deleted dispatch: %s", dispatch)
-            await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
-            await self._update_dispatch_schedule_and_notify(None, dispatch)
-
-            # Set deleted only here as it influences the result of dispatch.started
-            # which is used in above in _running_state_change
-            dispatch._set_deleted()  # pylint: disable=protected-access
-            await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
+        try:
+            self._currently_fetching = True
+
+            old_dispatches = self._dispatches
+            self._dispatches = {}
+
+            try:
+                _logger.info("Fetching dispatches for microgrid %s", self._microgrid_id)
+                async for page in self._client.list(microgrid_id=self._microgrid_id):
+                    for client_dispatch in page:
+                        dispatch = Dispatch(client_dispatch)
+
+                        self._dispatches[dispatch.id] = Dispatch(client_dispatch)
+                        old_dispatch = old_dispatches.pop(dispatch.id, None)
+                        if not old_dispatch:
+                            _logger.info("New dispatch: %s", dispatch)
+                            await self._update_dispatch_schedule_and_notify(
+                                dispatch, None
+                            )
+                            await self._lifecycle_updates_sender.send(
+                                Created(dispatch=dispatch)
+                            )
+                        elif dispatch.update_time != old_dispatch.update_time:
+                            _logger.info("Updated dispatch: %s", dispatch)
+                            await self._update_dispatch_schedule_and_notify(
+                                dispatch, old_dispatch
+                            )
+                            await self._lifecycle_updates_sender.send(
+                                Updated(dispatch=dispatch)
+                            )
+
+            except grpc.aio.AioRpcError as error:
+                _logger.error("Error fetching dispatches: %s", error)
+                self._dispatches = old_dispatches
+                return
+
+            for dispatch in old_dispatches.values():
+                _logger.info("Deleted dispatch: %s", dispatch)
+                await self._update_dispatch_schedule_and_notify(None, dispatch)
+                await self._lifecycle_updates_sender.send(Deleted(dispatch=dispatch))
+        finally:
+            self._currently_fetching = False
 
     async def _update_dispatch_schedule_and_notify(
         self, dispatch: Dispatch | None, old_dispatch: Dispatch | None

tests/test_frequenz_dispatch.py

@@ -144,8 +144,6 @@ async def _test_new_dispatch_created(
             assert False, "Expected a created event"
         case Created(dispatch):
             received = Dispatch(update_dispatch(sample, dispatch))
-            received._set_running_status_notified()  # pylint: disable=protected-access
-            dispatch._set_running_status_notified()  # pylint: disable=protected-access
             assert dispatch == received
 
     return dispatch
@@ -184,10 +182,7 @@ async def test_existing_dispatch_updated(
         case Created(dispatch) | Deleted(dispatch):
             assert False, f"Expected an updated event, got {dispatch_event}"
         case Updated(dispatch):
-            assert dispatch == Dispatch(
-                updated,
-                running_state_change_synced=dispatch.running_state_change_synced,
-            )
+            assert dispatch == Dispatch(updated)
 
     await asyncio.sleep(1)
 
@@ -212,7 +207,6 @@ async def test_existing_dispatch_deleted(
             assert False, "Expected a deleted event"
         case Deleted(dispatch):
             sample._set_deleted()  # pylint: disable=protected-access
-            dispatch._set_running_status_notified()  # pylint: disable=protected-access
             assert dispatch == sample
 
 
@@ -352,9 +346,6 @@ async def test_dispatch_schedule(
     # Expect notification of the dispatch being ready to run
     ready_dispatch = await actor_env.running_state_change.receive()
 
-    # Set flag we expect to be different to compare the dispatch with the one received
-    dispatch._set_running_status_notified()  # pylint: disable=protected-access
-
     assert ready_dispatch == dispatch
 
     assert dispatch.duration is not None