Only fetch relevant dispatches, excluding past ones

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

Author: Marenz

RELEASE_NOTES.md

@@ -15,6 +15,7 @@
 * Two new parameters were added to the `Dispatcher` constructor:
   * `sign_secret`: A secret key used for signing messages.
   * `auth_key`: An authentication key for the Dispatch API.
+* `Dispatcher` now only fetches ongoing dispatches, excluding completed ones, to optimize performance and relevance.
 
 ## Bug Fixes
 

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
   # plugins.mkdocstrings.handlers.python.import)
   "frequenz-sdk >= 1.0.0-rc2100, < 1.0.0-rc2200",
   "frequenz-channels >= 1.6.1, < 2.0.0",
-  "frequenz-client-dispatch >= 0.11.2, < 0.12.0",
+  "frequenz-client-dispatch >= 0.11.3, < 0.12.0",
   "frequenz-client-base >= 0.11.0, < 0.12.0",
 ]
 dynamic = ["version"]

src/frequenz/dispatch/_bg_service.py

@@ -370,14 +370,18 @@ async def _fetch(self, timer: Timer) -> None:
 
         try:
             _logger.debug("Fetching dispatches for microgrid %s", self._microgrid_id)
-            async for page in self._client.list(microgrid_id=self._microgrid_id):
+            now = datetime.now(timezone.utc)
+            async for page in self._client.list(
+                microgrid_id=self._microgrid_id, end_from=now - timedelta(seconds=5)
+            ):
                 for client_dispatch in page:
                     deleted_timestamp = self._deleted_dispatches.get(client_dispatch.id)
                     if (
                         deleted_timestamp
                         and client_dispatch.update_time < deleted_timestamp
                     ):
                         continue
+
                     dispatch = Dispatch(client_dispatch)
 
                     new_dispatches[dispatch.id] = dispatch

tests/test_frequenz_dispatch.py

@@ -4,6 +4,7 @@
 """Tests for the frequenz.dispatch.actor package."""
 
 import asyncio
+import logging
 from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from random import randint
@@ -476,9 +477,7 @@ async def test_dispatch_new_but_finished(
     )
     fake_time.shift(timedelta(seconds=1))
 
-    await asyncio.sleep(1)
-    # Process the lifecycle event caused by the old dispatch at startup
-    await test_env.lifecycle_events.receive()
+    # Expecting no event for a past and finished dispatch as it should be filtered out
 
     # Create another dispatch the normal way
     new_dispatch = generator.generate_dispatch()
@@ -514,7 +513,7 @@ async def test_notification_on_actor_start(
         running_dispatch,
         active=True,
         duration=timedelta(seconds=10),
-        start_time=_now() - timedelta(seconds=5),
+        start_time=_now() + timedelta(seconds=5),
         recurrence=RecurrenceRule(),
         type="TEST_TYPE",
     )
@@ -524,23 +523,31 @@ async def test_notification_on_actor_start(
         stopped_dispatch,
         active=False,
         duration=timedelta(seconds=5),
-        start_time=_now() - timedelta(seconds=5),
+        start_time=_now() + timedelta(seconds=5),
         recurrence=RecurrenceRule(),
         type="TEST_TYPE",
     )
+    logging.warning("Running dispatch: %s", running_dispatch)
+    logging.warning("Stopped dispatch: %s", stopped_dispatch)
+
     await test_env.service.stop()
 
+    logging.warning("Restarting service")
     # Create the dispatches
     test_env.client.set_dispatches(
         test_env.microgrid_id, [running_dispatch, stopped_dispatch]
     )
+    logging.warning("Dispatches set in client")
     test_env.service.start()
+    logging.warning("Service started")
 
-    fake_time.shift(timedelta(seconds=1))
-    await asyncio.sleep(1)
+    fake_time.shift(timedelta(seconds=7))
+    await asyncio.sleep(6)
+    logging.warning("Time shifted")
 
     # Expect notification of the running dispatch being ready to run
     ready_dispatch = await test_env.running_state_change.receive()
+    logging.warning("Received running state change: %s", ready_dispatch)
     assert ready_dispatch.started