エラーが発生した Cloud Storage オブジェクトの書き込みをキャンセルする

Rust 用 Cloud クライアント ライブラリの Cloud Storage クライアントは、StreamingSource 特性を実装する型からペイロード チャンクを pull してオブジェクトを書き込みます。データの生成中またはストリーミング中にアプリケーションでエラー状態が検出された場合、StreamingSource 実装からエラーを返すと、クライアント ライブラリによるオブジェクト作成の完了が停止します。

このドキュメントでは、カスタム エラータイプを定義し、StreamingSource を実装して、ストリームの途中でそのエラーを返す方法について説明します。また、アップロードに失敗すると、オブジェクトが作成されずにエラーが返されることも確認します。クライアント ライブラリを初めて使用する場合は、まず Rust で Cloud Storage を使用するをご覧ください。

始める前に

このドキュメントのコードを実行する前に、次の前提条件を満たしてください。

依存関係

Cloud Storage クライアント ライブラリとサポート クレートを Cargo.toml ファイルに追加します。

cargo add google-cloud-storage anyhow bytes

カスタム エラータイプを作成する

ファイナライズ前にオブジェクトの書き込みをキャンセルするには、StreamingSource の実装で std::error::Error + Send + Sync + 'static を実装するエラーを返す必要があります。これらの境界を満たす既存のエラータイプを使用することも、アプリケーションにカスタム列挙型を定義することもできます。カスタムエラー列挙型を定義する手順は次のとおりです。

  1. カスタムエラー列挙型（MyError）を定義します。 

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

  2. クライアント ライブラリがエラーを根本原因として保持できるように、標準の Error トレイトを実装します。 

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

  3. std::error::Error に必要な Display トレイトを実装します。 

    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"),
        }
    }
}

カスタム ストリーミング ソースを実装する

次に、合成チャンクを生成し、41 チャンク後にエラーを返すタイプを作成して、アップロードが完了する前に停止するようにします。次の手順を行います。

  1. チャンク カウンタを追跡する構造体を定義します。 

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

  2. MySource の StreamingSource を実装し、type Error = MyError を関連付けて、イテレーション 42 で Some(Err(MyError::ExpectedProblem)) を返します。 

    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"
            )))),
        }
    }
}

next が Some(Err(...)) を返すと、クライアント ライブラリはソースからの読み取りを停止し、オブジェクトを確定せずに書き込みリクエストをキャンセルします。

アップロードを試行してエラーを調べる

StreamingSource を実装したら、バッファリングされたオブジェクトの書き込みを試行し、クライアントが MyError をラップするシリアル化エラーを返すことを確認します。MySource は Seek を実装しないシングルパス ソースであるため、send_unbuffered ではなく send_buffered で書き込みを開始します。

projects/_/buckets/BUCKET_ID 形式の bucket_name を受け入れる async 関数内で、次の操作を行います。

  1. Storage クライアントを初期化します。 

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

  2. MySource::default() をペイロード ソースとして使用して、バッファリングされたオブジェクトの書き込みを開始します。

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

  3. ソースエラーが 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. Ok(()) を返して関数を完了します。 

        Ok(())
}

attempt_upload を実行すると、MyError::ExpectedProblem を含む Upload result Err(...) メッセージが出力されます。アサーションは、クライアントがアップロードを停止し、MyError をラップするシリアル化エラーを返したことを確認します。書き込みはファイナライズ前にキャンセルされたため、Cloud Storage はバケットに expect-error オブジェクトを作成しません。

クリーンアップ

キャンセルされた書き込みではオブジェクトは作成されませんが、このドキュメント用に作成したテスト オブジェクトを削除します。テストが完了したら、テストリソースを削除して、継続的な課金を回避します。

コードを補完する

次の例では、上記のすべてのステップを組み合わせて完全なモジュールにしています。

#[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(())
}

次のステップ