Cancel Cloud Storage object writes with errors

The Cloud Storage client in the Cloud Client Libraries for Rust writes objects by pulling payload chunks from a type that implements the StreamingSource trait. If your application detects an error condition while generating or streaming data, returning an error from your StreamingSource implementation stops the client library from finalizing the object creation.

This document shows you how to define a custom error type and implement StreamingSource to return that error mid-stream. You also verify that the failed upload returns your error without creating the object. If you're new to the client library, see Using Cloud Storage with Rust first.

Before you begin

Before running the code in this document, complete the following prerequisites:

Dependencies

Add the Cloud Storage client library and supporting crates to your Cargo.toml file:

cargo add google-cloud-storage anyhow bytes

Create a custom error type

To cancel an object write before finalization, your StreamingSource implementation must return an error that implements std::error::Error + Send + Sync + 'static. You can use any existing error type that meets these bounds or define a custom enum for your application. Complete the following steps to define a custom error enum:

  1. Define a custom error enum (MyError): 

    #[derive(Debug)]
pub enum MyError {
    ExpectedProblem,
    OhNoes,
}

  2. Implement the standard Error trait so the client library can preserve your error as the underlying cause: 

    impl std::error::Error for MyError {}

  3. Implement the Display trait required by std::error::Error: 

    impl std::fmt::Display for MyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ExpectedProblem => write!(f, "this kind of thing happens"),
            Self::OhNoes => write!(f, "oh noes! something terrible happened"),
        }
    }
}

Implement a custom streaming source

Next, create a type that generates synthetic chunks and returns an error after 41 chunks so the upload stops before completion. Complete the following steps:

  1. Define a struct to track the chunk counter: 

    #[derive(Debug, Default)]
struct MySource(u32);

  2. Implement StreamingSource for MySource, associating type Error = MyError and returning Some(Err(MyError::ExpectedProblem)) on iteration 42: 

    impl google_cloud_storage::streaming_source::StreamingSource for MySource {
    type Error = MyError;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0 += 1;
        match self.0 {
            42 => Some(Err(MyError::ExpectedProblem)),
            n if n > 42 => None,
            n => Some(Ok(bytes::Bytes::from_owner(format!(
                "test data for the example {n}\n"
            )))),
        }
    }
}

When next returns Some(Err(...)), the client library stops reading from the source and cancels the write request without finalizing the object.

Attempt the upload and inspect the error

After implementing StreamingSource, attempt a buffered object write and verify that the client returns a serialization error wrapping MyError. Because MySource is a single-pass source that doesn't implement Seek, start the write with send_buffered instead of send_unbuffered.

Inside an async function that accepts bucket_name in the format projects/_/buckets/BUCKET_ID, complete the following steps:

  1. Initialize a Storage client: 

    pub async fn attempt_upload(bucket_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::client::Storage;
    let client = Storage::builder().build().await?;

  2. Start the buffered object write using MySource::default() as the payload source: 

    let upload = client
    .write_object(bucket_name, "expect-error", MySource::default())
    .send_buffered()
    .await;

  3. Verify that the upload fails with a serialization error whose source error is MyError: 

    println!("Upload result {upload:?}");
let err = upload.expect_err("the source is supposed to terminate the upload");
assert!(err.is_serialization(), "{err:?}");
use std::error::Error as _;
assert!(err.source().is_some_and(|e| e.is::<MyError>()), "{err:?}");

  4. Return Ok(()) to complete the function: 

        Ok(())
}

When you run attempt_upload, it prints an Upload result Err(...) message containing MyError::ExpectedProblem. The assertions verify that the client stopped the upload and returned a serialization error wrapping MyError. Because the write was canceled before finalization, Cloud Storage doesn't create the expect-error object in the bucket.

Clean up

Even though the canceled write doesn't create an object, delete any test objects that you created for this document. Deleting test resources when you're finished avoids ongoing charges.

Complete code

The following example combines all the preceding steps into a complete module:

#[derive(Debug)]
pub enum MyError {
    ExpectedProblem,
    OhNoes,
}

impl std::error::Error for MyError {}

impl std::fmt::Display for MyError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ExpectedProblem => write!(f, "this kind of thing happens"),
            Self::OhNoes => write!(f, "oh noes! something terrible happened"),
        }
    }
}

#[derive(Debug, Default)]
struct MySource(u32);

impl google_cloud_storage::streaming_source::StreamingSource for MySource {
    type Error = MyError;
    async fn next(&mut self) -> Option<Result<bytes::Bytes, Self::Error>> {
        self.0 += 1;
        match self.0 {
            42 => Some(Err(MyError::ExpectedProblem)),
            n if n > 42 => None,
            n => Some(Ok(bytes::Bytes::from_owner(format!(
                "test data for the example {n}\n"
            )))),
        }
    }
}

pub async fn attempt_upload(bucket_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::client::Storage;
    let client = Storage::builder().build().await?;
    let upload = client
        .write_object(bucket_name, "expect-error", MySource::default())
        .send_buffered()
        .await;
    println!("Upload result {upload:?}");
    let err = upload.expect_err("the source is supposed to terminate the upload");
    assert!(err.is_serialization(), "{err:?}");
    use std::error::Error as _;
    assert!(err.source().is_some_and(|e| e.is::<MyError>()), "{err:?}");
    Ok(())
}

What's next