Replace StreamStopped with StreamRetrying

The new event is sent only when the stream will retry reconnecting
instead of every time the stream stops, as this information is less
important. When the stream stops, 2 things can happen, it is retried or
it finishes. The new event covers the retry case, and the case when
it finishes is covered by the receiver being closed.

With the current approach, we also needed to consume from the retry
strategy even if no retry was going to be attempted, so with this change
we can now only consume from the retry strategy when we actually want
to retry.

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

Author: llucax

RELEASE_NOTES.md

@@ -19,8 +19,8 @@
         match msg:
             case StreamStarted():
                 print("Stream started")
-            case StreamStopped(delay, error):
-                print(f"Stream stopped, reason {error}, retry in {delay}")
+            case StreamRetrying(delay, error):
+                print(f"Stream stopped and will retry in {delay}: {error or 'closed'}")
             case StreamFatalError(error):
                 print(f"Stream will stop because of a fatal error: {error}")
             case int() as output:

src/frequenz/client/base/streaming.py

@@ -32,14 +32,18 @@ class StreamStarted:
 
 
 @dataclass(frozen=True)
-class StreamStopped:
+class StreamRetrying:
     """Event indicating that the stream has stopped."""
 
-    retry_time: timedelta | None = None
-    """Time to wait before retrying the stream, if applicable."""
+    delay: timedelta
+    """Time to wait before retrying to start the stream again."""
 
     exception: Exception | None = None
-    """The exception that caused the stream to stop, if any."""
+    """The exception that caused the stream to stop, if any.
+
+    If `None`, the stream was stopped without an error, e.g. the server closed the
+    stream.
+    """
 
 
 @dataclass(frozen=True)
@@ -50,7 +54,7 @@ class StreamFatalError:
     """The exception that caused the stream to stop."""
 
 
-StreamEvent: TypeAlias = StreamStarted | StreamStopped | StreamFatalError
+StreamEvent: TypeAlias = StreamStarted | StreamRetrying | StreamFatalError
 """Type alias for the events that can be sent over the stream."""
 
 
@@ -89,8 +93,8 @@ def async_range() -> AsyncIterable[int]:
             match msg:
                 case StreamStarted():
                     print("Stream started")
-                case StreamStopped(delay, error):
-                    print(f"Stream stopped, reason {error}, retry in {delay}")
+                case StreamRetrying(delay, error):
+                    print(f"Stream stopped and will retry in {delay}: {error or 'closed'}")
                 case StreamFatalError(error):
                     print(f"Stream will stop because of a fatal error: {error}")
                 case int() as output:
@@ -184,23 +188,14 @@ async def _run(self) -> None:
             except grpc.aio.AioRpcError as err:
                 error = err
 
-            interval = self._retry_strategy.next_interval()
-
-            await sender.send(
-                StreamStopped(
-                    retry_time=(
-                        timedelta(seconds=interval) if interval is not None else None
-                    ),
-                    exception=error,
-                )
-            )
-
             if error is None and not self._retry_on_exhausted_stream:
                 _logger.info(
                     "%s: connection closed, stream exhausted", self._stream_name
                 )
                 await self._channel.close()
                 break
+
+            interval = self._retry_strategy.next_interval()
             error_str = f"Error: {error}" if error else "Stream exhausted"
             if interval is None:
                 _logger.error(
@@ -220,4 +215,6 @@ async def _run(self) -> None:
                 interval,
                 error_str,
             )
+
+            await sender.send(StreamRetrying(timedelta(seconds=interval), error))
             await asyncio.sleep(interval)

tests/streaming/test_grpc_stream_broadcaster.py

@@ -19,8 +19,8 @@
 from frequenz.client.base.streaming import (
     StreamEvent,
     StreamFatalError,
+    StreamRetrying,
     StreamStarted,
-    StreamStopped,
 )
 
 
@@ -97,7 +97,7 @@ async def _split_message(
     events: list[StreamEvent] = []
     async for item in receiver:
         match item:
-            case StreamStarted() | StreamStopped() | StreamFatalError():
+            case StreamStarted() | StreamRetrying() | StreamFatalError():
                 events.append(item)
             case str():
                 items.append(item)
@@ -149,9 +149,7 @@ async def test_streaming_success_retry_on_exhausted(
         "transformed_3",
         "transformed_4",
     ]
-    assert events == [
-        StreamStopped(exception=None, retry_time=None),
-    ]
+    assert events == []
 
     assert caplog.record_tuples == [
         (
@@ -182,7 +180,7 @@ async def test_streaming_success(
         receiver_ready_event.set()
         items, events = await _split_message(receiver)
 
-    no_retry.next_interval.assert_called_once_with()
+    no_retry.next_interval.assert_not_called()
 
     assert items == [
         "transformed_0",
@@ -191,9 +189,7 @@ async def test_streaming_success(
         "transformed_3",
         "transformed_4",
     ]
-    assert events == [
-        StreamStopped(exception=None, retry_time=None),
-    ]
+    assert events == []
     assert caplog.record_tuples == [
         (
             "frequenz.client.base.streaming",
@@ -344,8 +340,7 @@ async def test_messages_on_retry(
     ]
     assert events == [
         StreamStarted(),
-        StreamStopped(timedelta(seconds=0.0), error),
+        StreamRetrying(timedelta(seconds=0.0), error),
         StreamStarted(),
-        StreamStopped(None, error),
         StreamFatalError(error),
     ]