Clean server close of bidi read stream treated as success with pending ranges #6688

Activity

1. coryan

   added

   type: bugError or flaw in code with unintended results or allowing sub-optimal usage patterns.Error or flaw in code with unintended results or allowing sub-optimal usage patterns.

   api: storageIssues related to the Cloud Storage API.Issues related to the Cloud Storage API.

   on Sep 7, 2026
2. james-media-anywhere commented on Nov 2, 2026

   

   james-media-anywhere

   on Nov 2, 2026

   More actions

   Confirming this: both halves reproduce deterministically against 1.21.0, and main still has
   the same code (src/storage/src/storage/bidi/transport.rs line 138,
   expect("worker never exits while ObjectDescriptor is live"), and the worker's handling of a
   clean end-of-stream are unchanged from the 1.21.0 release).

   Reproduction. A fake google.storage.v2.Storage (tonic) implementing only BidiReadObject,
   which ends the stream on cue:

   1. Open a descriptor, read one range successfully, have the service end the stream cleanly with
      nothing outstanding, then call read_range again. On 1.21.0 the call panics at the expect
      above: the worker treated the end-of-stream as a normal exit and dropped its request channel,
      while the ObjectDescriptor the caller holds is unchanged.
   2. Request a 204,800-byte range on an object with no full-object checksum, have the service
      deliver 65,536 bytes and then end the stream cleanly. The reader yields 65,536 bytes and
      reports success.

   Against the real service. We keep descriptors open between reads in a pool, so idle streams
   are common. Cloud Storage prunes an idle BidiReadObject stream after roughly six minutes; today
   that arrives as an HTTP/2 RST_STREAM(INTERNAL_ERROR) and the resume policy handles it. Over a
   68-hour soak (6.3 M reads against a real bucket) we counted 64 service-ended streams, all resumed.
   Case 1 bites when the resume itself fails, for example a connectivity loss longer than the retry
   time limit: the worker exits permanently, the descriptor looks live, and the next read panics on
   1.21.0. On 1.16.0 the same state showed as every later read on that descriptor timing out, which
   is how we first met it in a deployment in August.

   Patch. Our diff against the published 1.21.0, carried via [patch.crates-io], is in the details block at the end:

   • the worker fails every outstanding reader with UnexpectedEof when the service ends the stream
     with ranges pending (case 2);
   • read_range on a descriptor whose worker has exited returns a stream that yields an
     io::ErrorKind::BrokenPipe error instead of panicking (case 1);
   • ObjectDescriptor::is_closed(), added to the stub trait as a default method returning
     false, so a pool can skip a dead descriptor before reading; no existing API changes;
   • a tracing::warn! and two counters on the resume path, so service-ended streams are visible.
     That part is an enhancement and is filed separately as storage: no log or metric when a BidiReadObject stream is ended by the service and resumed #7345.

   Happy to turn this into a PR against main with the two cases as tests if that is useful.
   Otherwise we will drop the vendored copy when a release carries your fix.

   Related: #7344 (Recommended resume policy does not resume on ABORTED, the status the service
   uses to close an idle stream).

   Diff against the published google-cloud-storage 1.21.0 (6 files)

   diff -ru --exclude=.cargo-ok --exclude=.cargo_vcs_info.json --exclude=Cargo.toml.orig --exclude=Cargo.lock a/src/object_descriptor.rs b/src/object_descriptor.rs
   --- a/src/object_descriptor.rs	2006-07-24 02:21:28
   +++ b/src/object_descriptor.rs	2026-10-30 07:15:26
   @@ -134,6 +134,15 @@
            self.inner.headers()
        }
    
   +    /// Returns `true` once the service has ended the underlying stream.
   +    ///
   +    /// See [`crate::stub::ObjectDescriptor::is_closed`]. A closed descriptor cannot
   +    /// serve reads; open a fresh one with
   +    /// [`open_object()`][crate::client::Storage::open_object].
   +    pub fn is_closed(&self) -> bool {
   +        self.inner.is_closed()
   +    }
   +
        /// Create a new instance.
        ///
        /// Application developers should only need to create an `ObjectDescriptor`
   @@ -215,4 +224,20 @@
                async fn next(&mut self) -> Option<crate::Result<bytes::Bytes>>;
            }
        }
   +}
   +
   +/// How many times, in this process, a `BidiReadObject` stream was ended by the
   +/// service with a status and successfully re-opened with its read handle. A
   +/// resume is invisible to the caller otherwise; this is the only way to measure
   +/// how often the service ends streams.
   +pub fn resume_count() -> u64 {
   +    crate::storage::bidi::worker::RESUMES.load(std::sync::atomic::Ordering::Relaxed)
   +}
   +
   +/// How many service-ended streams were *not* resumed: the resume policy declined
   +/// (a non-transient status such as `ABORTED`) or the reconnect failed. Each one
   +/// leaves a descriptor that reports [`ObjectDescriptor::is_closed`] == false but
   +/// fails every read. Ended streams in total = `resume_count() + resume_failure_count()`.
   +pub fn resume_failure_count() -> u64 {
   +    crate::storage::bidi::worker::RESUME_FAILURES.load(std::sync::atomic::Ordering::Relaxed)
    }
   diff -ru --exclude=.cargo-ok --exclude=.cargo_vcs_info.json --exclude=Cargo.toml.orig --exclude=Cargo.lock a/src/storage/bidi/stub.rs b/src/storage/bidi/stub.rs
   --- a/src/storage/bidi/stub.rs	2006-07-24 02:21:28
   +++ b/src/storage/bidi/stub.rs	2026-10-28 15:53:35
   @@ -25,6 +25,7 @@
            fn object(&self) -> Object;
            async fn read_range(&self, range: ReadRange) -> ReadObjectResponse;
            fn headers(&self) -> HeaderMap;
   +        fn is_closed(&self) -> bool;
        }
    
        #[async_trait::async_trait]
   @@ -39,6 +40,10 @@
    
            fn headers(&self) -> HeaderMap {
                T::headers(self)
   +        }
   +
   +        fn is_closed(&self) -> bool {
   +            T::is_closed(self)
            }
        }
    }
   diff -ru --exclude=.cargo-ok --exclude=.cargo_vcs_info.json --exclude=Cargo.toml.orig --exclude=Cargo.lock a/src/storage/bidi/transport.rs b/src/storage/bidi/transport.rs
   --- a/src/storage/bidi/transport.rs	2006-07-24 02:21:28
   +++ b/src/storage/bidi/transport.rs	2026-10-28 15:53:35
   @@ -132,15 +132,25 @@
        async fn read_range(&self, range: ReadRange) -> ReadObjectResponse {
            let (active, response) =
                Self::map_range(range.0, &self.tx, &self.object, self.checksum.clone());
   -        self.tx
   -            .send(active)
   -            .await
   -            .expect("worker never exits while ObjectDescriptor is live");
   +        if let Err(tokio::sync::mpsc::error::SendError(mut active)) = self.tx.send(active).await {
   +            // The worker has exited: the service ended the stream — end-of-stream or
   +            // a permanent error — while no read was in flight. The descriptor cannot
   +            // serve this read; fail it on the returned stream rather than panic.
   +            let closed = std::io::Error::new(
   +                std::io::ErrorKind::BrokenPipe,
   +                "the service ended this BidiReadObject stream; the descriptor is closed",
   +            );
   +            active.interrupted(Arc::new(Error::io(closed))).await;
   +        }
            response
        }
    
        fn headers(&self) -> HeaderMap {
            self.headers.clone()
   +    }
   +
   +    fn is_closed(&self) -> bool {
   +        self.tx.is_closed()
        }
    }
    
   diff -ru --exclude=.cargo-ok --exclude=.cargo_vcs_info.json --exclude=Cargo.toml.orig --exclude=Cargo.lock a/src/storage/bidi/worker.rs b/src/storage/bidi/worker.rs
   --- a/src/storage/bidi/worker.rs	2006-07-24 02:21:28
   +++ b/src/storage/bidi/worker.rs	2026-10-30 07:15:26
   @@ -26,6 +26,14 @@
    type ReadResult<T> = std::result::Result<T, ReadError>;
    type LoopResult<T> = std::result::Result<T, Arc<crate::Error>>;
    
   +/// Number of times a `BidiReadObject` stream ended with a status and the worker
   +/// re-opened it with the saved read handle. Process-wide; see
   +/// [`crate::object_descriptor::resume_count`].
   +pub(crate) static RESUMES: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
   +/// Number of such re-opens that failed (the descriptor is dead afterwards).
   +pub(crate) static RESUME_FAILURES: std::sync::atomic::AtomicU64 =
   +    std::sync::atomic::AtomicU64::new(0);
   +
    #[derive(Debug)]
    pub struct Worker<C> {
        next_range_id: i64,
   @@ -63,6 +71,7 @@
        ) -> LoopResult<()> {
            let mut ranges = Vec::new();
            let (mut rx, mut tx) = (connection.rx, connection.tx);
   +        let mut ended_by_service = false;
            // Note how this loop only exits when the `requests` queue is
            // closed. A successfully closed stream and unrecoverable errors
            // return immediately.
   @@ -71,7 +80,10 @@
                    m = rx.next_message() => {
                        match self.handle_response(m).await {
                            // Successful end of stream, return without error.
   -                        None => break None,
   +                        None => {
   +                            ended_by_service = true;
   +                            break None;
   +                        }
                            // An unrecoverable in the stream or its data, return
                            // the error.
                            Some(Err(e)) => break Some(e),
   @@ -96,7 +108,21 @@
            drop(rx);
            drop(tx);
            let Some(e) = error else {
   -            // A successfully closed stream *and* there are no more readers.
   +            // A successfully closed stream. If the service ended it with ranges still
   +            // outstanding, those readers must see an error: left alone, their channels
   +            // end and a partial range is indistinguishable from a complete one.
   +            if ended_by_service {
   +                let outstanding = self.ranges.lock().await.len();
   +                if outstanding > 0 {
   +                    let truncated = std::io::Error::new(
   +                        std::io::ErrorKind::UnexpectedEof,
   +                        format!(
   +                            "the service ended the BidiReadObject stream with {outstanding} range(s) outstanding"
   +                        ),
   +                    );
   +                    self.close_readers(Arc::new(crate::Error::io(truncated))).await;
   +                }
   +            }
                return Ok(());
            };
            // Return errors for any future readers.
   @@ -161,14 +187,25 @@
                .iter()
                .map(|(id, r)| r.as_proto(*id))
                .collect();
   +        tracing::warn!(
   +            "BidiReadObject stream ended by the service with {} range(s) outstanding, attempting resume: {status:?}",
   +            ranges.len()
   +        );
            let (response, _headers, connection) = match self.connector.reconnect(status, ranges).await
            {
                Err(e) => {
   +                // Either the resume policy declined (a non-transient status such as
   +                // ABORTED) or the reconnect itself failed; both leave a dead descriptor.
   +                RESUME_FAILURES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
   +                tracing::warn!("BidiReadObject not resumed; the descriptor is now dead: {e:?}");
                    let error = Arc::new(e);
                    self.close_readers(error.clone()).await;
                    return Some(Err(error));
                }
   -            Ok(t) => t,
   +            Ok(t) => {
   +                RESUMES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
   +                t
   +            }
            };
            if let Err(e) = self.handle_ranges(response.object_data_ranges).await {
                // An error in the response. These are not recoverable.
   diff -ru --exclude=.cargo-ok --exclude=.cargo_vcs_info.json --exclude=Cargo.toml.orig --exclude=Cargo.lock a/src/storage/bidi.rs b/src/storage/bidi.rs
   --- a/src/storage/bidi.rs	2006-07-24 02:21:28
   +++ b/src/storage/bidi.rs	2026-10-29 13:45:38
   @@ -23,7 +23,7 @@
    mod retry_redirect;
    pub(crate) mod stub;
    pub(crate) mod transport;
   -mod worker;
   +pub(crate) mod worker;
    
    use crate::google::storage::v2::{BidiReadObjectRequest, BidiReadObjectResponse};
    use crate::request_options::RequestOptions;
   diff -ru --exclude=.cargo-ok --exclude=.cargo_vcs_info.json --exclude=Cargo.toml.orig --exclude=Cargo.lock a/src/storage/stub.rs b/src/storage/stub.rs
   --- a/src/storage/stub.rs	2006-07-24 02:21:28
   +++ b/src/storage/stub.rs	2026-10-28 15:53:35
   @@ -138,6 +138,20 @@
    
        /// The implementation for [ObjectDescriptor::headers][Descriptor::headers].
        fn headers(&self) -> HeaderMap;
   +
   +    /// Returns `true` once the service has ended the underlying stream.
   +    ///
   +    /// A BidiReadObject stream can be ended by the service at any time — a clean
   +    /// end-of-stream, or a permanent error — including while no read is in flight.
   +    /// After that, [`read_range()`][Self::read_range] can no longer be served on
   +    /// this descriptor: it returns a stream whose first item is an error. Callers
   +    /// that pool descriptors should check this before reuse and open a fresh
   +    /// descriptor instead.
   +    ///
   +    /// The default is `false`, for implementations without a notion of closing.
   +    fn is_closed(&self) -> bool {
   +        false
   +    }
    }
    
    /// Defines the trait used to implement [crate::appendable_object_writer::AppendableObjectWriter].
3. joshuatants

   added 4 commits that reference this issue on Nov 5, 2026

   fix(storage): fail bidi reads on unexpected server OK

   ...

   This issue will close once commit 4a51a38 is merged into the 'main' branch.

   4a51a38

   fix(storage): fail bidi reads on unexpected server OK

   ...

   This issue will close once commit 091d1cc is merged into the 'main' branch.

   091d1cc

   fix(storage): fail bidi reads on unexpected server OK

   ...

   This issue will close once commit f043050 is merged into the 'main' branch.

   f043050

   fix(storage): fail bidi reads on unexpected server OK

   ...

   This issue will close once commit efcb315 is merged into the 'main' branch.

   efcb315
4. joshuatants

   closed this as completedin #7382on Nov 8, 2026
5. joshuatants

   added a commit that references this issue on Nov 8, 2026

   fix(storage): fail bidi reads on unexpected server OK (googleapis#7382)

   ...

   This issue will close once commit c72da80 is merged into the 'main' branch.

   c72da80