Catch and log transform errors in the GrpcStreamBroadcaster (#203)

Author: shsms

RELEASE_NOTES.md

@@ -14,4 +14,4 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+- This fixes a bug in the `GrpcStreamBroadcaster` that was causing exceptions in the transform method to be silenced.

src/frequenz/client/base/retry.py

@@ -9,6 +9,8 @@
 from copy import deepcopy
 from typing import Self
 
+from typing_extensions import override
+
 DEFAULT_RETRY_INTERVAL = 3.0
 """Default retry interval, in seconds."""
 
@@ -98,6 +100,7 @@ def __init__(
 
         self._count = 0
 
+    @override
     def next_interval(self) -> float | None:
         """Return the time to wait before the next retry.
 
@@ -153,6 +156,7 @@ def __init__(  # pylint: disable=too-many-arguments
 
         self._count = 0
 
+    @override
     def next_interval(self) -> float | None:
         """Return the time to wait before the next retry.
 

src/frequenz/client/base/streaming.py

@@ -291,7 +291,17 @@ async def _run(self) -> None:
 
                 async for msg in call:
                     first_message_received = True
-                    await data_sender.send(self._transform(msg))
+                    try:
+                        transformed = self._transform(msg)
+                    except Exception:  # pylint: disable=broad-exception-caught
+                        _logger.exception(
+                            "%s: error transforming message: %s",
+                            self._stream_name,
+                            msg,
+                        )
+                        continue
+
+                    await data_sender.send(transformed)
 
             except grpc.aio.AioRpcError as err:
                 error = err

tests/streaming/test_grpc_stream_broadcaster.py

@@ -293,6 +293,86 @@ async def test_streaming_error(  # pylint: disable=too-many-arguments
     ]
 
 
+async def test_streaming_transform_error(  # pylint: disable=too-many-arguments
+    no_retry: mock.MagicMock,  # pylint: disable=redefined-outer-name
+    receiver_ready_event: asyncio.Event,  # pylint: disable=redefined-outer-name
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test streaming transform errors."""
+    caplog.set_level(logging.INFO)
+
+    def transform_err(x: int) -> str:
+        """Mock transform with err for odd values."""
+        if x % 2 == 1:
+            raise ValueError("No, you transform.")
+        return f"transformed_{x}"
+
+    async def asynciter() -> AsyncIterator[int]:
+        """Mock async iterator."""
+        await receiver_ready_event.wait()
+        for i in range(5):
+            yield i
+            await asyncio.sleep(0)  # Yield control to the event loop
+
+    rpc_mock = mock.MagicMock(
+        name="ok_helper_method",
+        side_effect=lambda: unary_stream_call_mock(
+            "ok_helper_unary_stream_call", asynciter
+        ),
+    )
+
+    helper = streaming.GrpcStreamBroadcaster(
+        stream_name="test_helper",
+        stream_method=rpc_mock,
+        transform=transform_err,
+        retry_strategy=no_retry,
+    )
+
+    items: list[str] = []
+    async with AsyncExitStack() as stack:
+        stack.push_async_callback(helper.stop)
+
+        receiver = helper.new_receiver()
+        receiver_ready_event.set()
+        items, _ = await _split_message(receiver)
+
+    assert items == [
+        "transformed_0",
+        "transformed_2",
+        "transformed_4",
+    ]
+
+    assert caplog.record_tuples == [
+        (
+            "frequenz.client.base.streaming",
+            logging.INFO,
+            "test_helper: starting to stream",
+        ),
+        # LogCaptureFixture can't capture tracebacks, so only the error message
+        # is checked.
+        (
+            "frequenz.client.base.streaming",
+            logging.ERROR,
+            "test_helper: error transforming message: 1",
+        ),
+        (
+            "frequenz.client.base.streaming",
+            logging.ERROR,
+            "test_helper: error transforming message: 3",
+        ),
+        (
+            "frequenz.client.base.streaming",
+            logging.INFO,
+            "test_helper: connection closed, stream exhausted",
+        ),
+        (
+            "frequenz.client.base.streaming",
+            logging.INFO,
+            "test_helper: stopping the stream",
+        ),
+    ]
+
+
 @pytest.mark.parametrize("include_events", [True, False])
 async def test_retry_next_interval_zero(  # pylint: disable=too-many-arguments
     receiver_ready_event: asyncio.Event,  # pylint: disable=redefined-outer-name