Catch CancelledErrors properly

The method `exception()` in `Task` doesn't return `CancelledError` if
the task was cancelled, instead it raises the error, so if a task is
cancelled an exception would be raised unintentionally.

Instead we want to catch `CancelledError` and treat it like any other
task error for these tasks, are they are not supposed to be cancelled.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/actor/_data_sourcing/microgrid_api_source.py

@@ -413,11 +413,13 @@ async def clean_tasks(
             ) -> set[asyncio.Task[None]]:
                 done, pending = await asyncio.wait(sending_tasks, timeout=0)
                 for task in done:
-                    if error := task.exception():
-                        _logger.error(
+                    try:
+                        task.result()
+                    # pylint: disable-next=broad-except
+                    except (asyncio.CancelledError, Exception):
+                        _logger.exception(
                             "Error while processing message in task %s",
                             task.get_name(),
-                            exc_info=error,
                         )
                 return pending
 

src/frequenz/sdk/actor/_resampling.py

@@ -119,6 +119,7 @@ async def _run(self) -> None:
             tasks_to_cancel.add(subscriptions_task)
 
             while True:
+
                 resampling_task = asyncio.create_task(self._resampler.resample())
                 tasks_to_cancel.add(resampling_task)
                 done, _ = await asyncio.wait(
@@ -134,29 +135,35 @@ async def _run(self) -> None:
 
                 if resampling_task in done:
                     tasks_to_cancel.remove(resampling_task)
-                    # The resampler shouldn't end without an exception
-                    error = resampling_task.exception()
-                    assert (
-                        error is not None
-                    ), "The resample() function shouldn't exit normally."
-
-                    # We don't know what to do with something other than
-                    # ResamplingError, so propagate the exception if that is the
-                    # case.
-                    if not isinstance(error, ResamplingError):
-                        raise error
-                    for source, source_error in error.exceptions.items():
+                    # The resampler shouldn't be cancelled or end without an exception
+                    try:
+                        resampling_task.result()
+                    except ResamplingError as error:
+                        for source, source_error in error.exceptions.items():
+                            _logger.error(
+                                "Error resampling source %s, removing source...", source
+                            )
+                            removed = self._resampler.remove_timeseries(source)
+                            if not removed:
+                                _logger.error(
+                                    "Got an exception from an unknown source: "
+                                    "source=%r, exception=%r",
+                                    source,
+                                    source_error,
+                                )
+                    # pylint: disable-next=broad-except
+                    except (Exception, asyncio.CancelledError):
+                        # We don't know what to do with something other than
+                        # ResamplingError, so we log it, restart, and hope for the best.
+                        _logger.exception(
+                            "The resample() function got an unexpected error, restarting..."
+                        )
+                    else:
+                        # The resample function should not end normally, so we log it,
+                        # restart, and hope for the best.
                         _logger.error(
-                            "Error resampling source %s, removing source...", source
+                            "The resample() function ended without an exception, restarting..."
                         )
-                        removed = self._resampler.remove_timeseries(source)
-                        if not removed:
-                            _logger.warning(
-                                "Got an exception from an unknown source: "
-                                "source=%r, exception=%r",
-                                source,
-                                source_error,
-                            )
                     # The resampling_task will be re-created if we reached this point
         finally:
             await asyncio.gather(*[cancel_and_await(t) for t in tasks_to_cancel])

tests/timeseries/mock_microgrid.py

@@ -230,8 +230,12 @@ def start_mock_client(
         self.init_mock_client(initialize_cb)
 
         def _done_callback(task: asyncio.Task[None]) -> None:
-            if exc := task.exception():
-                raise SystemExit(f"Streaming task {task.get_name()!r} failed: {exc}")
+            try:
+                task.result()
+            except (asyncio.CancelledError, Exception) as exc:
+                raise SystemExit(
+                    f"Streaming task {task.get_name()!r} failed: {exc}"
+                ) from exc
 
         for component_id, coro in self._streaming_coros:
             task = asyncio.create_task(coro, name=f"component-id:{component_id}")