Repository navigation
Conversation
|
Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA). View this failed invocation of the CLA check for more information. For the most up to date status, view the checks section at the bottom of the pull request. |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements a mechanism to drain and nack buffered responses when a message stream is closed or dropped, preventing messages already sent by the server but not yet read by the application from being delayed until their ack deadline expires. This is achieved by introducing a drain_buffered method on the underlying stream and calling it within drain_and_nack during stream closure and drop. Feedback on the changes suggests optimizing the drain_and_nack task to avoid unnecessary string clones when batching ack_ids by draining the vector directly instead of using chunks() and to_vec().
|
/gcbrun |
e7bab4b to
072b276
Compare
When a `MessageStream` shuts down or is dropped, it drops the gRPC stream. Any responses the server already sent but the client had not read yet were discarded with it. Their messages never reached lease management, so they were neither leased nor nacked, and the service only redelivered them once their ack deadline expired. With the default 60 second deadline, a message published around shutdown could stall for a minute. Now, before the stream is dropped (on shutdown, on a permanent error, or when the `MessageStream` is dropped), the client reads the responses already buffered on the stream, without waiting on the network, and nacks their messages with `ModifyAckDeadline(0)`. The drain stops at the first response that is not immediately available, at the end of the stream, at an error, or after 100 responses. The nacks go directly through the stub, because lease management never knew about these messages and may already have shut down. This applies to both `ShutdownBehavior` values. Messages that are still in flight on the network when the stream closes are not covered. The service still redelivers them after their ack deadline. Fixes googleapis#7412
Move the ack IDs into each nack request instead of copying them out of `chunks()`.
072b276 to
4b8d9ec
Compare
|
@dbolduc Sorry, I re-pushed with my own email. probably another |
|
/gcbrun |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #7413 +/- ##
==========================================
- Coverage 98.65% 98.65% -0.01%
==========================================
Files 334 334
Lines 145392 145604 +212
==========================================
+ Hits 143433 143642 +209
- Misses 1959 1962 +3 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Move the nack loop for drained messages into `nack_ack_ids()` and test that it splits the ack IDs into batches of at most 1000. Also test that dropping a `MessageStream` outside of a Tokio runtime does not panic.
|
@dbolduc Added the missing coverage, note that the |
|
/gcbrun |
I think |
|
@dbolduc Yay! All builds are passing now 🎉 |
When a
MessageStreamshuts down or is dropped, it drops the gRPC stream. Any responses the server already sent but the client had not read yet are discarded with it. Their messages never reach lease management, so they are neither leased nor nacked, and the service only redelivers them once their ack deadline expires. With the default 60 second deadline, a message published around shutdown can stall for a minute. We observed this in an application's end-to-end tests: a message published while a subscriber shut down was never yielded byMessageStream::next()in any process, and was redelivered exactly one ack deadline later.With this change, before the stream is dropped (on shutdown, on a permanent error, or when the
MessageStreamis dropped), the client reads the responses already buffered on the stream, without waiting on the network, and nacks their messages withModifyAckDeadline(0). The drain stops at the first response that is not immediately available, at the end of the stream, at an error, or after 100 responses, so it never blocks. The nacks go directly through the stub, because lease management never knew about these messages and may already have shut down. The nack RPCs run in a spawned task, soShutdownToken::shutdown()does not wait for them. This applies to bothShutdownBehaviorvalues.How this compares to other Pub/Sub clients: Go and Java do not drain. Go cancels the pull streams first on shutdown and nacks only the messages it has already read (
nackInventory). Java cancels the call inStreamingSubscriberConnection.doStop()withcloseSendWithError(CANCELLED). Both lose already-buffered responses the same way this crate did. Rust does drain because the cost is small and bounded: it only reads data the client already holds, it never waits, and it turns a full ack-deadline stall into an immediate redelivery for those messages.Limits: messages that are still in flight on the network when the stream closes are not covered, and are still redelivered after their ack deadline. If the application signals shutdown but neither calls
next()again nor drops theMessageStream, the drain happens only when it does one of those.Tests:
stream.rs:drain_bufferedreturns buffered responses and stops when the next one is pending, at the response limit, at the end of the stream, and at an error.message_stream.rs(against the mock gRPC server): a response buffered at shutdown is nacked, not dropped, underNackImmediatelyandWaitForProcessing, both throughnext()after shutdown and through dropping the stream. These tests time out without the fix. They wait 200 ms of real time for the responses to reach the client's buffers.Fixes #7412
🤖 Generated with Claude Code