[shsms]

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().