Skip to content

feat(storage): Support GCS RCU Bidirectional Read Integration Tests - DRAFT #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
xlai20 wants to merge 6 commits into
base: main
Choose a base branch
from
Draft

feat(storage): Support GCS RCU Bidirectional Read Integration Tests - DRAFT #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
679a6fa
first draft: Test Suite 1 first 5 test cases
xlai20 Sep 21, 2026
b85e738
Change mod struct
xlai20 Sep 21, 2026
69ca28a
feat(storage): parameterize bidi read conformance tests into scenario…
xlai20 Sep 21, 2026
e3b88a2
cleaner test code design structure
xlai20 Sep 22, 2026
58d69e5
feat(storage): structure bidi_read tests into regional_standard, zona…
xlai20 Sep 22, 2026
54decfb
feat(storage): dedicated create and cleanup for rapid regional bucket…
xlai20 Sep 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
209 changes: 8 additions & 201 deletions tests/storage/src/bidi_read.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,204 +12,11 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use google_cloud_storage::client::Storage;
use google_cloud_storage::model_ext::ReadRange;

pub async fn run(bucket_name: &str) -> anyhow::Result<()> {
let client = Storage::builder().build().await?;
send(&client, bucket_name).await?;
send_and_read(&client, bucket_name).await?;
send_and_read_full(&client, bucket_name).await?;
send_and_read_md5(&client, bucket_name).await?;
send_and_read_gzip(&client, bucket_name).await?;
Ok(())
}

async fn send(client: &Storage, bucket_name: &str) -> anyhow::Result<()> {
let write = client
.write_object(
bucket_name,
"basic/source.txt",
String::from_iter((0..100_000).map(|_| 'a')),
)
.set_if_generation_match(0)
.send_unbuffered()
.await?;

let open = client.open_object(bucket_name, &write.name).send().await?;
tracing::info!("open returns: {open:?}");
let got = open.object();
let mut want = write.clone();
// This field is a mismatch, but both `Some(false)` and `None` represent
// the same value.
want.event_based_hold = want.event_based_hold.or(Some(false));
// There is a submillisecond difference, maybe rounding?
want.finalize_time = got.finalize_time;
assert_eq!(got, want);

let mut reader = open.read_range(ReadRange::head(100)).await;
let mut count = 0_usize;
while let Some(r) = reader.next().await.transpose()? {
tracing::info!("received {} bytes", r.len());
count += r.len();
}
assert_eq!(count, 100_usize);

Ok(())
}

pub async fn send_and_read(client: &Storage, bucket_name: &str) -> anyhow::Result<()> {
let payload = String::from_iter(('a'..='z').cycle().take(100_000));
let write = client
.write_object(bucket_name, "open_and_read/source.txt", payload.clone())
.set_if_generation_match(0)
.send_unbuffered()
.await?;

let (descriptor, mut reader) = client
.open_object(bucket_name, &write.name)
.send_and_read(ReadRange::tail(100))
.await?;
tracing::info!("object: {:?}", descriptor.object());
tracing::info!("headers: {:?}", descriptor.headers());
tracing::info!("reader: {:?}", reader);
let got = descriptor.object();
let mut want = write.clone();
// This field is a mismatch, but both `Some(false)` and `None` represent
// the same value.
want.event_based_hold = want.event_based_hold.or(Some(false));
// There is a submillisecond difference, maybe rounding?
want.finalize_time = got.finalize_time;
assert_eq!(got, want);

let mut data = Vec::new();
while let Some(r) = reader.next().await.transpose()? {
tracing::info!("received {} bytes", r.len());
data.extend_from_slice(&r);
}
assert_eq!(data, &payload.as_bytes()[(payload.len() - 100)..]);

Ok(())
}

pub async fn send_and_read_md5(client: &Storage, bucket_name: &str) -> anyhow::Result<()> {
let payload = String::from_iter(('a'..='z').cycle().take(100_000));
let write = client
.write_object(bucket_name, "open_and_read_md5/source.txt", payload.clone())
.set_if_generation_match(0)
.send_unbuffered()
.await?;

let (descriptor, mut reader) = client
.open_object(bucket_name, &write.name)
.compute_md5()
.send_and_read(ReadRange::all())
.await?;
tracing::info!("object: {:?}", descriptor.object());
tracing::info!("headers: {:?}", descriptor.headers());
tracing::info!("reader: {:?}", reader);
let got = descriptor.object();
let mut want = write.clone();
want.event_based_hold = want.event_based_hold.or(Some(false));
want.finalize_time = got.finalize_time;
assert_eq!(got, want);

let mut data = Vec::new();
while let Some(r) = reader.next().await.transpose()? {
tracing::info!("received {} bytes", r.len());
data.extend_from_slice(&r);
}
assert_eq!(data, payload.as_bytes());

Ok(())
}

pub async fn send_and_read_full(client: &Storage, bucket_name: &str) -> anyhow::Result<()> {
let payload = String::from_iter(('a'..='z').cycle().take(100_000));
let write = client
.write_object(
bucket_name,
"open_and_read_full/source.txt",
payload.clone(),
)
.set_if_generation_match(0)
.send_unbuffered()
.await?;

let (descriptor, mut reader) = client
.open_object(bucket_name, &write.name)
.send_and_read(ReadRange::all())
.await?;
tracing::info!("object: {:?}", descriptor.object());
tracing::info!("headers: {:?}", descriptor.headers());
tracing::info!("reader: {:?}", reader);
let got = descriptor.object();
let mut want = write.clone();
want.event_based_hold = want.event_based_hold.or(Some(false));
want.finalize_time = got.finalize_time;
assert_eq!(got, want);

let mut data = Vec::new();
while let Some(r) = reader.next().await.transpose()? {
tracing::info!("received {} bytes", r.len());
data.extend_from_slice(&r);
}
assert_eq!(data, payload.as_bytes());

Ok(())
}

/// This test verifies the checksum validation behavior for gzip-encoded objects
/// over the gRPC Bidi read stream.
///
/// Unlike the JSON REST API, which often transcodes (decompresses) gzip objects
/// on the fly, the gRPC Bidi read stream delivers the raw, compressed bytes directly.
/// Because no on-the-fly decompression occurs, the CRC32C checksum of the received
/// chunks will naturally match the server's stored checksum of the compressed object.
///
/// We explicitly expect `RangeReader`'s automatic checksum validation to succeed
/// without throwing a `ChecksumMismatch` error, proving that we do not need to
/// bypass checksum validation for `content-encoding: gzip` objects in gRPC.
pub async fn send_and_read_gzip(client: &Storage, bucket_name: &str) -> anyhow::Result<()> {
use std::io::Write;
let payload = String::from_iter(('a'..='z').cycle().take(100_000));

// Compress the payload
let mut e = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
e.write_all(payload.as_bytes())?;
let compressed_payload = e.finish()?;

let write = client
.write_object(
bucket_name,
"open_and_read_gzip/source.txt",
bytes::Bytes::from_owner(compressed_payload.clone()),
)
.set_if_generation_match(0)
.set_content_encoding("gzip")
.send_unbuffered()
.await?;

let (descriptor, mut reader) = client
.open_object(bucket_name, &write.name)
.send_and_read(ReadRange::all())
.await?;
tracing::info!("object: {:?}", descriptor.object());
tracing::info!("headers: {:?}", descriptor.headers());
tracing::info!("reader: {:?}", reader);
let got = descriptor.object();
let mut want = write.clone();
want.event_based_hold = want.event_based_hold.or(Some(false));
want.finalize_time = got.finalize_time;
assert_eq!(got, want);

let mut data = Vec::new();
while let Some(r) = reader.next().await.transpose()? {
tracing::info!("received {} bytes", r.len());
data.extend_from_slice(&r);
}
// Verify we received the EXACT compressed payload, meaning gRPC did not decompress it.
assert_eq!(data, compressed_payload);

Ok(())
}
//! Bidirectional Read Integration Tests.
//!
//! Separated into two distinct test suites:
//! - [conformance]: Formal cross-SDK conformance tests for Bidi Read sessions (Suite 1).
//! - [features]: Client builder options, encodings, and backward-compatibility regression tests.

pub mod conformance;
pub mod features;
Loading