Fix scheduler pacing and stop condition logic

Resolved incorrect pacing and stop condition handling in the Scheduler
class.

- Replaced inaccurate duration checks with explicit `_start_time`
  tracking.
- Introduced `_time_remaining()` and `_available_sleep_window()` for
  precise timing.
- Refactored sleep logic into `_pace()` to account for task duration.
- Ensured scheduler stops exactly when configured `duration` is reached.
- Improved logging clarity and added test correction for edge timing
  behavior.

Signed-off-by: cyiallou - Costas <[email protected]>

Author: cyiallou

RELEASE_NOTES.md

@@ -15,3 +15,4 @@
 ## Bug Fixes
 
 - Updated the Solar Maintenance notebook to fix the expected environment variable name for the reporting server url.
+- Fixed a bug in the notification `Scheduler` where tasks could overrun the configured duration due to imprecise sleep and stop logic. The scheduler now correctly tracks elapsed time, respects task execution duration, and stops reliably after the intended interval.

src/frequenz/lib/notebooks/notification_service.py

@@ -333,7 +333,7 @@ def __init__(self, config: SchedulerConfig) -> None:
         self._task_name: str | None = None
         self._stop_event = threading.Event()
         self._thread: threading.Thread | None = None
-        self._time_awoke: float = 0.0  # time when the scheduler awoke from sleep
+        self._start_time: float | None = None
 
     def start(self, task: Callable[..., None], **kwargs: Any) -> None:
         """Start the scheduler for a given task.
@@ -363,13 +363,15 @@ def stop(self) -> None:
         """Stop the scheduler."""
         if self._thread is not None:
             if self._thread.is_alive():
-                _log.info("Stopping scheduler for %s", self._task_name)
+                _log.info("Stopping scheduler for task '%s'", self._task_name)
                 self._stop_event.set()
                 if not self._stop_event.is_set():
-                    _log.error("Failed to stop scheduler for %s", self._task_name)
+                    _log.error(
+                        "Failed to stop scheduler for task '%s'", self._task_name
+                    )
         else:
             _log.warning(
-                "Attempted to stop scheduler for %s, but no active thread was found.",
+                "Attempted to stop scheduler for task '%s', but no active thread was found.",
                 self._task_name,
             )
         _log.info("Scheduler successfully stopped")
@@ -380,42 +382,38 @@ def _run_task(self, kwargs: dict[str, Any]) -> None:
         Args:
             kwargs: Arguments to pass to the task.
         """
-        start_time = time.time()
+        self._start_time = time.time()
         if self._config.send_immediately:
-            self._execute_task(kwargs)
+            elapsed = self._execute_task(kwargs)
+            self._pace(elapsed)
         else:
             _log.info(
                 "Waiting for first interval before sending the first notification."
             )
-            self._stop_event.wait(self._config.interval)
-            self._time_awoke = time.time()
-
-        while not self._stop_event.is_set():
-            if self._should_stop(start_time):
-                break
-            self._execute_task(kwargs)
-
-    def _should_stop(self, start_time: float) -> bool:
-        """Determine if the scheduler should stop.
-
-        Args:
-            start_time: The time the scheduler started.
-
-        Returns:
-            True if the scheduler should stop, False otherwise.
-        """
-        if (
+            self._pace(0)
+        while not self._should_stop():
+            elapsed = self._execute_task(kwargs)
+            self._pace(elapsed)
+        _log.info("Scheduler stopping: stop condition met.")
+        self.stop()
+
+    def _should_stop(self) -> bool:
+        """Return True if the scheduler should stop."""
+        _log.debug("Checking if scheduler for task '%s' should stop.", self._task_name)
+        return self._stop_event.is_set() or (
             self._config.duration is not None
-            and (time.time() - self._time_awoke - start_time) >= self._config.duration
-        ):
-            return True
-        return False
+            and self._start_time is not None
+            and self._time_remaining() <= 0
+        )
 
-    def _execute_task(self, kwargs: dict[str, Any]) -> None:
+    def _execute_task(self, kwargs: dict[str, Any]) -> float:
         """Execute the scheduled task and handle interval waiting.
 
         Args:
             kwargs: Arguments to pass to the task.
+
+        Returns:
+            The time taken to execute the task in seconds.
         """
         task_start_time = time.time()
         try:
@@ -429,14 +427,44 @@ def _execute_task(self, kwargs: dict[str, Any]) -> None:
             )
         finally:
             task_elapsed = time.time() - task_start_time
-            sleep_duration = max(0, self._config.interval - task_elapsed)
-            _log.info(
-                "Scheduled execution completed for %s. Sleeping for %d seconds.",
+            _log.debug(
+                "Execution of task '%s' completed in %.2f seconds.",
                 self._task_name,
-                sleep_duration,
+                task_elapsed,
             )
-            self._stop_event.wait(sleep_duration)
-            self._time_awoke = time.time()
+        return task_elapsed
+
+    def _time_remaining(self) -> float:
+        """Return the remaining time before the scheduler should stop.
+
+        Returns:
+            A float indicating the number of seconds remaining until the
+            configured duration is exceeded. If no duration is configured,
+            returns float('inf') to represent an unbounded schedule.
+        """
+        if self._config.duration is None or self._start_time is None:
+            return float("inf")
+        return max(0.0, self._config.duration - (time.time() - self._start_time))
+
+    def _available_sleep_window(self) -> float:
+        """Calculate the maximum allowed sleep time given the interval and remaining time."""
+        return min(self._config.interval, self._time_remaining())
+
+    def _pace(self, elapsed_task_time: float) -> None:
+        """Sleep for interval minus task duration, bounded by duration limit.
+
+        Args:
+            elapsed_task_time: Time taken by the task in seconds.
+        """
+        sleep_duration = self._available_sleep_window()
+        if sleep_duration < self._config.interval:
+            actual_sleep = max(0, sleep_duration)
+        else:
+            actual_sleep = max(0, sleep_duration - elapsed_task_time)
+        if self._stop_event.is_set():
+            return
+        _log.info("Sleeping for %.2f seconds before next task execution.", actual_sleep)
+        self._stop_event.wait(actual_sleep)
 
 
 class BaseNotification:

tests/test_notification_service.py

@@ -97,14 +97,20 @@ def test_scheduler_behavior(config: SchedulerConfig, mock_send: MagicMock) -> No
     scheduler.start(mock_send)
     if config.duration:
         time.sleep(config.duration + 0.5)
+        correction = -1  # -1 to account for the time lost in executing the task
     else:
         time.sleep(test_time_if_no_duration)
+        correction = 0
         scheduler.stop()
 
     expected_calls = (
-        (config.duration if config.duration else test_time_if_no_duration)
-        // config.interval
-    ) + (1 if config.send_immediately else 0)
+        (
+            (config.duration if config.duration else test_time_if_no_duration)
+            // config.interval
+        )
+        + (1 if config.send_immediately else 0)
+        + correction
+    )
     assert mock_send.call_count == expected_calls