Skip to content

Add missing retry strategy reset in streaming #148

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Sep 2, 2025
Merged

Add missing retry strategy reset in streaming #148

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
7293ab6
Clear release notes
llucax Aug 28, 2025
23137ba
Add missing retry strategy reset in streaming
llucax May 28, 2025
043e121
Wait for the initial metadata before sending `StreamStarted` event
llucax Aug 28, 2025
bcecf65
Use the local ready event instead of passing it as argument
llucax Aug 28, 2025
a345940
Fix mocking of the `stream_method` in the `ok_helper` fixture
llucax Aug 28, 2025
30ee943
Fix mocking of error tests
llucax Aug 28, 2025
eef9557
Remove wrong extra messages from test
llucax Aug 28, 2025
99b6853
Add test for failures in `initial_metadata()`
llucax Aug 28, 2025
bc5fe5f
Don't require using literals for `new_receiver()`
llucax Aug 28, 2025
2711173
Make `num_success` keyword-only
llucax Aug 28, 2025
a7be10a
Remove unused private class
llucax Aug 28, 2025
1159958
Test that the retry strategy is reset after a successful start
llucax Aug 28, 2025
4575451
Stop using the deprecated 'close' method for channels
llucax Aug 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/frequenz/client/base/streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,7 @@ def new_receiver(
*,
maxsize: int = 50,
warn_on_overflow: bool = True,
include_events: Literal[True],
include_events: bool,
) -> channels.Receiver[StreamEvent | OutputT]: ...
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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]"

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.


def new_receiver(
Expand All @@ -208,7 +208,7 @@ def new_receiver(
maxsize: int = 50,
warn_on_overflow: bool = True,
include_events: bool = False,
) -> channels.Receiver[OutputT] | channels.Receiver[StreamEvent | OutputT]:
) -> channels.Receiver[StreamEvent | OutputT]:
"""Create a new receiver for the stream.

Args:
Expand Down
5 changes: 2 additions & 3 deletions tests/streaming/test_grpc_stream_broadcaster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
from collections.abc import AsyncIterator, Callable
from contextlib import AsyncExitStack
from datetime import timedelta
from typing import Literal
from unittest import mock

import grpc
Expand Down Expand Up @@ -308,7 +307,7 @@ async def test_streaming_error( # pylint: disable=too-many-arguments
async def test_retry_next_interval_zero( # pylint: disable=too-many-arguments
receiver_ready_event: asyncio.Event, # pylint: disable=redefined-outer-name
caplog: pytest.LogCaptureFixture,
include_events: Literal[True] | Literal[False],
include_events: bool,
) -> None:
"""Test retry logic when next_interval returns 0."""
caplog.set_level(logging.WARNING)
Expand Down Expand Up @@ -363,7 +362,7 @@ async def test_retry_next_interval_zero( # pylint: disable=too-many-arguments
)
async def test_messages_on_retry(
receiver_ready_event: asyncio.Event, # pylint: disable=redefined-outer-name
include_events: Literal[True] | Literal[False],
include_events: bool,
error_in_metadata: bool,
) -> None:
"""Test that messages are sent on retry."""
Expand Down