FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Add an idle timeout to GrpcStreamBroadcaster and stop it with its receivers by shsms · Pull Request #281 · frequenz-floss/frequenz-client-base-python · GitHub

Repository navigation

Add an idle timeout to GrpcStreamBroadcaster and stop it with its receivers - #281

Draft
shsms wants to merge 4 commits into
frequenz-floss:v0.x.xfrom
shsms:stream-idle-timeout
Draft

shsms wants to merge 4 commits into
frequenz-floss:v0.x.xfrom
shsms:stream-idle-timeout

Conversation

shsms commented Oct 20, 2026

Copy link
Copy Markdown
Contributor

After a reporting API outage, forecast streams stayed dead until a restart: a stalled call was never noticed, and abandoned broadcasters kept retrying forever. This adds an idle timeout to GrpcStreamBroadcaster, stops a broadcaster once every receiver it handed out is gone, and starts it again on the next new_receiver().

Changes

  • idle_timeout (default None, so off): a timer per message gap. A call that stalls is cancelled and goes through retry_strategy with a StreamIdleTimeoutError.
  • Receivers are tracked by a wrapper with weakref.finalize. Releases from other threads are handed back to the loop. When the last receiver is closed or garbage-collected, the run is cancelled, also while it is stalled or retrying. Note the order in new_receiver(): it counts the receiver before subscribing, because a garbage collection during subscribing could otherwise stop the run under the new receiver.
  • State is per run (_Run). After a stop for lack of receivers, new_receiver() starts a new run with its own channels and a fresh retry budget. An old run that is still winding down cannot end the new run's receivers. A stream that ended, gave up or failed by itself is not restarted, and nothing restarts after stop().
  • Fixes on the way:
    • Receivers now end on an unexpected error, and receivers with events get StreamFatalError.
    • stop() passes on its caller's cancellation and waits for every run.
    • An events receiver requested after the stream stopped no longer hangs.
  • Tests: 53 streaming tests. The new ones were checked to fail when the code they cover is removed. They also cover garbage collection while subscribing or counting, releases from another thread, and same-tick close and re-subscribe.
  • Known limit: a garbage collection inside the stream's own last step, before it decides to end, can still restart a stream that then ended. Whether a receiver dropped at that moment gets a restart depends on when the garbage collector runs.

Breaking

  • is_running is False once every receiver handed out is closed or garbage-collected, until the next new_receiver() starts the stream again.
  • new_receiver() returns a wrapper Receiver: len() on it and Merger.stop() no longer work; use close().

github-actions Bot added part:docs Affects the documentation part:tests Affects the unit, integration and performance (benchmarks) tests part:code Affects the code in general labels Oct 20, 2026
shsms added 4 commits October 21, 2026 17:04
The retry count of GrpcStreamBroadcaster was never reset, so a retry
limit counted every retry over the life of the stream. A long-running
stream with a limit gave up after that many lost connections, even when
it had recovered from each one.

Reset the count when a message arrives, so a limit counts the retries
since the last message.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
A server stream that stays open without sending blocked its consumers
forever. Cancel such a call and retry it like a failed one.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Abandoned streams kept retrying forever, so abandoned broadcasters piled
up calls during outages. Track the receivers handed out and stop once
all are closed or dropped, also while stalled or retrying.

Also end the receivers on any exit of the stream: an unexpected error
left them waiting forever. Now they end, and receivers with events get a
StreamFatalError. And let stop() pass on a cancellation of its caller,
which it used to swallow.

A receiver with events requested after the stream stopped now ends at
once too. If no receiver with events existed before, it used to wait on
a new event channel that was never closed.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
Some clients reuse a cached broadcaster without checking is_running, so
a stream stopped for lack of receivers would hand them a receiver that
ends without data. Start a new run instead. Each run already closes only
its own channels, so the old run winding down cannot end the new run's
receivers.

A stream that ended, gave up or failed by itself is not restarted.

Signed-off-by: Sahas Subramanian <sahas.subramanian@proton.me>
shsms force-pushed the stream-idle-timeout branch from da79257 to 1604390 Compare October 21, 2026 15:04

This branch has not been deployed

No deployments
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

part:code Affects the code in general part:docs Affects the documentation part:tests Affects the unit, integration and performance (benchmarks) tests

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant


Back | FazBrowse Home | New Git URL