Repository navigation
Conversation
) * Return the next message after a streaming subscription reconnects After a transient stream error (UNAVAILABLE, UNKNOWN, INTERNAL), next_message() reconnected and then returned None instead of reading from the new stream. It now reads from the new stream, and raises after MAX_RECONNECT_ATTEMPTS reconnects in one call. The async client does the same and now also retries INTERNAL. The request iterator of a failed stream kept waiting on the shared send queue, so after a reconnect it took the next ack and sent it to the dead stream. Each stream now has its own send queue, and closing a stream ends its request iterator. close() now stays closed: a reconnect already in progress no longer reopens the stream. The close function from subscribe_with_handler waits up to SUBSCRIPTION_CLOSE_TIMEOUT_SECONDS for the handler thread, so test_subscribe_topic_with_handler no longer leaves that thread running and its coverage no longer changes between runs. Fixes #1230 Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Signed-off-by: Casper Nielsen <casper@diagrid.io> * Address review: end of stream, stale acks and reconnect backoff A stream the server ends cleanly (StopIteration in the sync client, grpc.aio.EOF in the async one) now goes through the same bounded reconnect as UNAVAILABLE, UNKNOWN and INTERNAL, instead of failing with "Error while fetching message". The async docstring no longer promises None at end of stream. Reconnects inside one next_message() call now wait a short jittered exponential backoff between attempts. close() ends that wait at once. After MAX_RECONNECT_ATTEMPTS the clients raise the new StreamReconnectError, a subclass of Exception. The async start() now raises StreamInactiveError when close() runs during the initial read, as the sync one does. A cancellation while the subscription is still open is re-raised. Each message remembers the send queue of the stream it came from, and respond() drops a response to a message from a replaced stream, with a debug log. The runtime only logs an error for such an ack, but it never has to see it now. The sync close function logs a warning when the handler thread is still running after SUBSCRIPTION_CLOSE_TIMEOUT_SECONDS. The async subscription logs through a module logger instead of print(). The unreachable queue.Empty handler is gone. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Signed-off-by: Casper Nielsen <casper@diagrid.io> * fix: retry failed reconnects and release the request iterator on start failure - Sync start() opens the RPC outside the stream lock and re-checks close() afterwards, cancelling the new stream if the subscription was closed. - A failed initial read now puts the None sentinel on the stream's send queue, so gRPC's request-consumer thread exits (sync and async), and the async client clears _stream_active like the sync client does. - A reconnect whose new stream fails with UNAVAILABLE/UNKNOWN/INTERNAL now counts as an attempt and waits for the backoff, instead of escaping next_message(). - Async start() re-raises CancelledError when the caller's own task is being cancelled (Python 3.11+), instead of turning it into StreamInactiveError. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Casper Nielsen <casper@diagrid.io> * docs: describe streaming subscription reconnects and StreamReconnectError Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Casper Nielsen <casper@diagrid.io> * fix: use a general, lazily formatted warning when a subscription response fails Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Signed-off-by: Casper Nielsen <casper@diagrid.io> --------- Signed-off-by: Casper Nielsen <casper@diagrid.io> Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Co-authored-by: Sam Kendziora <sam@diagrid.io> (cherry picked from commit 77beac2) Signed-off-by: dapr-bot <dapr-bot@users.noreply.github.com>
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## release-1.19 #1269 +/- ##
===============================================
Coverage ? 85.83%
===============================================
Files ? 163
Lines ? 12908
Branches ? 0
===============================================
Hits ? 11080
Misses ? 1828
Partials ? 0 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
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
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Backport 77beac2 from #1231.