Use a flag to identify if a ReceiverStoppedError is expected

If it happens because the fetcher is being stopped, then only log a
DEBUG message.

Signed-off-by: Elzbieta Kotulska <[email protected]>

Author: ela-kotulska-frequenz

src/frequenz/sdk/timeseries/formula_engine/_formula_steps.py

@@ -10,7 +10,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Generic
 
-from frequenz.channels import Receiver, ReceiverError
+from frequenz.channels import Receiver, ReceiverError, ReceiverStoppedError
 
 from .._base_types import QuantityT, Sample
 
@@ -400,6 +400,7 @@ def __init__(
         self._nones_are_zeros = nones_are_zeros
         self._fallback: FallbackMetricFetcher[QuantityT] | None = fallback
         self._latest_fallback_sample: Sample[QuantityT] | None = None
+        self._is_stopped = False
 
     @property
     def stream(self) -> Receiver[Sample[QuantityT]]:
@@ -416,6 +417,7 @@ async def stop(self) -> None:
         If metric fetcher is stopped, it can't be started again.
         There is no use-case now to start it again.
         """
+        self._is_stopped = True
         self.stream.close()
         if self._fallback:
             await self._fallback.stop()
@@ -436,6 +438,18 @@ async def _fetch_from_fallback(
     ) -> Sample[QuantityT] | None:
         try:
             return await fallback_fetcher.receive()
+        except ReceiverStoppedError:
+            if self._is_stopped:
+                _logger.debug(
+                    "Stream for fallback metric fetcher %s closed.",
+                    fallback_fetcher.name,
+                )
+            else:
+                _logger.error(
+                    "Failed to fetch next value from %s. Fallback stream closed.",
+                    self._name,
+                )
+            return None
         except ReceiverError[Any] as err:
             _logger.error(
                 "Failed to fetch next value from fallback stream %s: %s",
@@ -491,6 +505,12 @@ async def fetch_next(self) -> Sample[QuantityT] | None:
         Returns:
             The fetched Sample.
         """
+        if self._is_stopped:
+            _logger.error(
+                "Metric fetcher %s stopped. Can't fetch new value.", self._name
+            )
+            return None
+
         self._next_value = await self._fetch_next()
         return self._next_value
 
@@ -499,6 +519,14 @@ async def _fetch_next(self) -> Sample[QuantityT] | None:
         primary_value: Sample[QuantityT] | None = None
         try:
             primary_value = await self._stream.receive()
+        except ReceiverStoppedError:
+            if self._is_stopped:
+                _logger.debug("Stream for metric fetcher %s closed.", self._name)
+                return None
+            _logger.error(
+                "Failed to fetch next value from %s. Primary stream closed.",
+                self._name,
+            )
         except ReceiverError[Any] as err:
             _logger.error("Failed to fetch next value from %s: %s", self._name, err)