Cancela las escrituras de objetos de Cloud Storage con errores

El cliente de Cloud Storage en las bibliotecas cliente de Cloud para Rust escribe objetos extrayendo fragmentos de carga útil de un tipo que implementa el atributo StreamingSource. Si tu aplicación detecta una condición de error mientras genera o transmite datos, la devolución de un error de tu implementación de StreamingSource impide que la biblioteca cliente finalice la creación del objeto.

En este documento, se muestra cómo definir un tipo de error personalizado y, luego, implementar StreamingSource para devolver ese error a mitad de la transmisión. También debes verificar que la carga con errores devuelva tu error sin crear el objeto. Si es la primera vez que usas la biblioteca cliente, consulta Usa Cloud Storage con Rust primero.

Antes de comenzar

Antes de ejecutar el código de este documento, completa los siguientes requisitos previos:

Dependencias

Agrega la biblioteca cliente de Cloud Storage y los crates de compatibilidad a tu archivo Cargo.toml:

cargo add google-cloud-storage anyhow bytes

Crea un tipo de error personalizado

Para cancelar una escritura de objeto antes de la finalización, tu implementación de StreamingSource debe mostrar un error que implemente std::error::Error + Send + Sync + 'static. Puedes usar cualquier tipo de error existente que cumpla con estos límites o definir una enumeración personalizada para tu aplicación. Sigue estos pasos para definir una enumeración de errores personalizada:

  1. Define un enum de error personalizado (MyError): 

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

  2. Implementa el atributo estándar Error para que la biblioteca cliente pueda conservar tu error como la causa subyacente: 

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

  3. Implementa el atributo Display que requiere 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"),
        }
    }
}

Implementa una fuente de transmisión personalizada

A continuación, crea un tipo que genere fragmentos sintéticos y devuelva un error después de 41 fragmentos para que la carga se detenga antes de completarse. Completa los siguientes pasos:

  1. Define un struct para hacer un seguimiento del contador de fragmentos: 

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

  2. Implementa StreamingSource para MySource, asociando type Error = MyError y devolviendo Some(Err(MyError::ExpectedProblem)) en la iteración 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"
            )))),
        }
    }
}

Cuando next devuelve Some(Err(...)), la biblioteca cliente deja de leer desde la fuente y cancela la solicitud de escritura sin finalizar el objeto.

Intenta realizar la carga y, luego, inspecciona el error

Después de implementar StreamingSource, intenta escribir un objeto almacenado en búfer y verifica que el cliente devuelva un error de serialización que contenga MyError. Debido a que MySource es una fuente de un solo paso que no implementa Seek, inicia la escritura con send_buffered en lugar de send_unbuffered.

Dentro de una función async que acepte bucket_name en el formato projects/_/buckets/BUCKET_ID, completa los siguientes pasos:

  1. Inicializa un cliente de Storage: 

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

  2. Inicia la escritura del objeto almacenado en búfer con MySource::default() como la fuente de carga útil: 

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

  3. Verifica que la carga falle con un error de serialización cuyo error de origen sea 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. Devuelve Ok(()) para completar la función: 

        Ok(())
}

Cuando ejecutas attempt_upload, se imprime un mensaje Upload result Err(...) que contiene MyError::ExpectedProblem. Las aserciones verifican que el cliente detuvo la carga y devolvió un error de serialización que une MyError. Debido a que la escritura se canceló antes de la finalización, Cloud Storage no crea el objeto expect-error en el bucket.

Realiza una limpieza

Aunque la escritura cancelada no crea un objeto, borra los objetos de prueba que creaste para este documento. Borra los recursos de prueba cuando termines para evitar cargos continuos.

Completar código

En el siguiente ejemplo, se combinan todos los pasos anteriores en un módulo completo:

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

¿Qué sigue?