feat(storage): add stream reconnect support for appendable upload - #6606
vsharonlynn wants to merge 5 commits into
Conversation
product-auto-label
Bot
added
the
api: storage
![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 introduces a new replay_buffer module to manage unacknowledged data chunks for bidirectional streaming writes, and adds a reconnect method to the Connector to handle stream reconnection and redirection. Feedback suggests updating the reconnect method to evaluate the error against the retry policy so that it fails early on permanent errors instead of attempting to reconnect unconditionally.
| pub async fn reconnect( | ||
| &mut self, | ||
| last_error: Error, | ||
| ) -> Result<(BidiWriteObjectResponse, Connection<T::Stream>)> { | ||
| if let Some(status) = gaxi::as_inner::as_inner::<gaxi::grpc::tonic::Status, _>(&last_error) | ||
| { | ||
| let mut guard = self.spec.lock().expect("never poisoned"); | ||
| guard.handle_redirect(status.clone()); | ||
| } | ||
| self.connect_attempt_loop().await | ||
| } |
There was a problem hiding this comment.
The reconnect method currently attempts to reconnect unconditionally, even if last_error is a permanent error (e.g., PermissionDenied). To avoid unnecessary connection attempts and fail early, we should evaluate the error against the retry policy and only proceed if the policy allows retrying.
pub async fn reconnect(
&mut self,
last_error: Error,
) -> Result<(BidiWriteObjectResponse, Connection<T::Stream>)> {
if let Some(status) = gaxi::as_inner::as_inner::<gaxi::grpc::tonic::Status, _>(&last_error)
{
let mut guard = self.spec.lock().expect("never poisoned");
guard.handle_redirect(status.clone());
}
let retry = RetryRedirect::new(self.options.retry_policy.clone());
let state = google_cloud_gax::retry_state::RetryState::new(true);
match retry.on_error(&state, last_error) {
google_cloud_gax::retry_result::RetryResult::Permanent(e)
| google_cloud_gax::retry_result::RetryResult::Exhausted(e) => Err(e),
google_cloud_gax::retry_result::RetryResult::Continue(_) => {
self.connect_attempt_loop().await
}
}
}
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #6606 +/- ##
==========================================
- Coverage 97.17% 97.17% -0.01%
==========================================
Files 328 329 +1
Lines 109802 110072 +270
==========================================
+ Hits 106702 106958 +256
- Misses 3100 3114 +14 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
db6a083 to
1926ec3
Compare
- Drop the unreachable expect() by reading the front chunk length before popping it, removing the redundant panic branch. - Relabel the // SAFETY: comment as // Invariant:, since Bytes::slice is safe Rust and the comment documents an algorithmic invariant, not an unsafe contract.
1926ec3 to
c860912
Compare
1 participant
Issue #5716 .
This PR follows PR #6605 .