feat(storage): add stream reconnect support for appendable upload

[vsharonlynn]

Issue #5716 .

This PR follows PR #6605 .

[gemini-code-assist]

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.

[gemini-code-assist]

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]

Codecov Report

❌ Patch coverage is 98.34711% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 96.53%. Comparing base (7684ebe) to head (1926ec3).
⚠️ Report is 13 commits behind head on main.

Files with missing lines	Patch %	Lines
src/storage/src/storage/bidi_write/connector.rs	96.93%	3 Missing ⚠️
...rc/storage/src/storage/bidi_write/replay_buffer.rs	99.30%	1 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##             main    #6606    +/-   ##
========================================
  Coverage   96.52%   96.53%            
========================================
  Files         304      305     +1     
  Lines       92054    92296   +242     
========================================
+ Hits        88858    89096   +238     
- Misses       3196     3200     +4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.