Add missing retry strategy reset in streaming

[llucax]

After the streaming started successfully we should reset the retry strategy, so next time there is a failure we can retry from the beginning of the retry strategy again.

[Marenz]

Can we make a test for that?

[shsms]

I have a feeling we need to make a flag and put the StreamStartedEvent and the retry_strategy reset inside the loop.

Until we try to iterate over the call, we don't know if it was successful.

[shsms]

Especially for some kinds of issues, like the streaming method on the API is failing before sending any data.

[Marenz]

I mean, true, but for me this is good enough

[shsms]

StreamStarted sounds like there were messages. StreamOpened might be better in this case.

[llucax]

Mmm, it looks like we need to think a bit more about this, but I think it is very subtle when to consider a stream being started. For me if the stream request was sent, it started, and any errors after that is a normal retry. In any case, discussing this could take some time and effort, and I think that is a separate discussion and could be a future improvement, but it is a bug not ever reset the retry strategy. So I suggest to merge this an open a different issue for this if we want to change it, because also I think I already spent more time than I should have improving the streaming events 😬, so unless someone else is willing to pick it up, I don't see much happening beyond this PR and #149.

[llucax]

what do you mean?

[Marenz]

I thought it's my fault, misunderstanding

[llucax]

I was thinking about this and I don't think putting the reset (let's leave the event for a separate issue, as this PR is not touching on that, but I think the arguments for reset applies to the event too) is really not an option. A stream could be idle for a long time until something is sent through it. If errors are really delayed until the returned iterator is awaited, then maybe the only option would be to try to get the first element with some very short timeout, but I'm not sure if we have any guarantees. For example if the other end disappeared, I'm not sure any errors will be raised until the connection really times out for good.

[llucax]

AI overlords suggest using await call.initial_metadata() to "survey for errors" (or wait_for_connection(), but this one is marked as experimental, and not sure if it is only about the underlying channel connection or if it will also catch other errors that happen after the connection, like a NOT FOUND or something like that).

[llucax]

If the AI overlords are not hallucinating, it seems like it should work.

sequenceDiagram
    actor Client
    actor Server

    %% Phase 0: Client-Side Setup
    Note over Client: Phase 0: Client-Side Setup
    Client->>Client: Creates gRPC channel
    Client->>Client: Instantiates service stub

    %% Phase 1: RPC Invocation & Connection Establishment
    Note over Client: Phase 1: RPC Invocation & Connection Establishment
    Client->>Client: Calls unary_stream method (call = stub.Method())
    Client->>Client: Executes await call.initial_metadata()

    opt TCP/TLS Connection Not Established
        Client->>Server: TCP: SYN
        Server-->>Client: TCP: SYN-ACK
        Client->>Server: TCP: ACK
        Note over Client,Server: TCP Connection Established

        Client->>Server: TLS: ClientHello
        Server-->>Client: TLS: ServerHello, Certificate, etc.
        Client->>Server: TLS: KeyExchange, ChangeCipherSpec
        Server-->>Client: TLS: ChangeCipherSpec
        Note over Client,Server: TLS Handshake Complete

        Client->>Server: HTTP/2: Connection Preface (Magic + SETTINGS)
        Server-->>Client: HTTP/2: SETTINGS frame
        Note over Client,Server: HTTP/2 Connection Initialized
    end

    Note over Client,Server: RPC Request over established HTTP/2
    Client->>Server: HTTP/2: HEADERS (method, metadata, :status=200, content-type=application/grpc)
    Client->>Server: HTTP/2: DATA (serialized request_message, END_STREAM=true)
    Server-->>Client: HTTP/2: HEADERS (initial server metadata, END_STREAM=false)
    Note over Client: await call.initial_metadata() completes

    %% Phase 2: Server Streaming Responses
    Note over Client,Server: Phase 2: Server Streaming Responses
    loop For each message in stream
        Note over Client: Enters/awaits in `async for msg in call`
        Server-->>Client: HTTP/2: DATA (serialized response_message_N, END_STREAM=false)
        Client->>Client: Receives and deserializes msg_N
    end

    %% Phase 3: Stream Termination (Normal)
    Note over Client,Server: Phase 3: Stream Termination (Normal)
    Note over Client: Awaits in `async for` (server finished sending)
    Server-->>Client: HTTP/2: HEADERS (Trailers: grpc-status=OK, END_STREAM=true)
    Note over Client: `async for` loop terminates
    Client->>Client: Accesses call.code(), call.details()

    %% Alternate Flow: Error During Stream
    alt Error During Stream
        Note over Client,Server: Alternate Flow: Error During Stream
        Note over Client: Awaits next message in `async for msg in call`
        Server-->>Client: HTTP/2: HEADERS (Trailers: grpc-status=ERROR_CODE, grpc-message, END_STREAM=true)
        Note over Client: AioRpcError raised
        Client->>Client: Accesses call.code(), call.details()
    end

Loading

[llucax]

Updated, still the test needs to be fixed because to do this we can't take an arbitrary async iterator anymore, we need to take a proper UnaryStreamCall to be able to call initial_metadata().

[llucax]

OK, this is ready for review again.

After considering where to put the reset(), I think the best place is after the first message is received. Otherwise a stream that doesn't receive anything and is exhausted will still be considered successfully connected, and the retry will be reset. That seems weird, particularly in the scenario where retry_on_exhausted_stream=True, but in general if the server is saying the stream is over again and again, never being able to receive any messages it sounds to me like something is wrong and we should do a back-off between tries (and eventually give up if the number of retries are limited) instead of resetting the strategy.

I didn't change when the StreamStarted event is emitted though, but I added the await for the initial metadata before sending the event. Again I think this is probably the best trade-off between knowing with some certainty that the streaming successfully started without having to wait for the first message, which could take a long time for low-frequency streams (like dispatch updates).

[Copilot]

Pull Request Overview

This PR fixes the retry strategy behavior in the GrpcStreamBroadcaster by resetting the retry strategy after a successful stream start, and improves the timing of the StreamStarted event.

• Reset retry strategy after receiving the first message to prevent accumulating delays across retries
• Wait for initial metadata before firing StreamStarted event for better connection indication
• Update channel closing methods and dependency version

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
src/frequenz/client/base/streaming.py	Add retry strategy reset logic and improve StreamStarted event timing
tests/streaming/test_grpc_stream_broadcaster.py	Refactor test mocks and add test for retry strategy reset behavior
pyproject.toml	Update frequenz-channels dependency version requirement
RELEASE_NOTES.md	Document bug fixes and breaking changes

---

_{Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.}

[llucax]

@shsms do you want to give it a look too as you also reviewed in the past?

[shsms]

Just a comment for future improvement.

[shsms]

But here they can pass False from a variable and still get typed StreamEvent | OutputT. And there's another overload above that would give OutputT when users send Literal[False]. I understand that it is unusable with literals and that in practice we're just adding support for variables while not providing any narrowing. But still, I feel it is a bad idea to do any overloads based on literals and maybe there's a type based solution to this, with sentinels, etc. But I can't think of a nice scheme at the moment, so happy to postpone.

[llucax]

No, we are providing narrowing when using a literal, so x.new_receiver(include_events=False) (which is the default) should be typed -> OutputT. The problem is we can properly narrow the return type when a variable is passed. So I would say it is half-useful, but since this half will probably be used in 95% of the cases (the default), I think it is still pretty useful.

g: GrpcStreamBroadcaster[int, str] = GrpcStreamBroadcaster[int, str](
    stream_name="test",
    stream_method=None,  # type: ignore
)
reveal_type(obj: g.new_receiver())

Gives:

Type of "g.new_receiver()" is "Receiver[str]"

[shsms]

Oh that's true. It is a problem only when people start to mess with things. :D

[llucax]

Yeah, I hit the issue only because I parametrized a test, so I started passing a variable to include_events, but I don't think it will be a common case in practice.