Trabaja con el control de simultaneidad optimista (OCC)

El control de simultaneidad optimista (OCC) es una estrategia para administrar recursos compartidos y evitar la pérdida de actualizaciones o condiciones de carrera cuando varios usuarios o procesos modifican el mismo recurso al mismo tiempo.

Por ejemplo, cuando trabajas con Identity and Access Management, una política de IAM es un recurso compartido adjunto a un recurso, como un proyecto, un bucket o un servicio. Sin el control de simultaneidad, las modificaciones simultáneas pueden reemplazar los cambios recientes en las políticas, lo que revoca de forma involuntaria el acceso legítimo o otorga permisos no autorizados.

Para implementar el OCC, los sistemas suelen usar un número de versión o un campo etag (etiqueta de entidad) en la estructura del recurso.

Cómo funciona el OCC

Supongamos que dos procesos, A y B, intentan actualizar un recurso compartido al mismo tiempo:

  1. El proceso A lee el estado actual del recurso.
  2. El proceso B lee el mismo estado actual.
  3. El proceso A modifica su copia y la vuelve a escribir en el servidor.
  4. El proceso B modifica su copia y la vuelve a escribir en el servidor.

Debido a que el proceso B reemplaza el recurso sin saber que el proceso A ya lo cambió, se pierden las actualizaciones del proceso A.

El OCC soluciona este problema con la introducción de una huella digital única que cambia cada vez que se modifica una entidad. En muchos sistemas, como IAM, esta huella digital es un etag. El servidor verifica esta etiqueta en cada escritura:

  1. Cuando lees el recurso, el servidor devuelve una etag (una huella digital única).
  2. Cuando vuelvas a enviar el recurso modificado, debes incluir el etag original.
  3. Si el etag almacenado no coincide con el etag que enviaste, lo que significa que otro proceso modificó el recurso después de que lo leíste, la operación de escritura falla con un error ABORTED o FAILED_PRECONDITION.

Esta falla indica al cliente que vuelva a intentar todo el proceso: volver a leer el estado actualizado, volver a aplicar los cambios y volver a intentar la escritura con el nuevo etag.

Implementa el bucle de OCC

El núcleo de una implementación de OCC es un bucle de reintentos que lee, modifica y guarda el recurso, y sale después de una escritura exitosa. Para evitar bucles infinitos durante una contención alta, configura una cantidad máxima razonable de intentos de reintento.

Pasos del bucle

En la siguiente tabla, se describe cada paso de un bucle de reintentos de OCC:

Paso Acción Ejemplo de implementación
Lectura Recupera el estado actual del recurso, incluido el etag. 
let mut policy = client
    .get_iam_policy()
    .set_resource(&secret_name)
    .send()
    .await?;
Modificar Aplica los cambios a la estructura local. 
policy.bindings.push(new_binding);
Escribe y verifica Intenta guardar el recurso modificado con el etag existente y verifica si hay códigos de error de simultaneidad. 
match client
    .set_iam_policy()
    .set_resource(&secret_name)
    .set_policy(policy)
    .send()
    .await
{
    Ok(p) => return Ok(p),
    Err(e) => { /* ... */ }
}
Éxito o reintento Si la escritura se realiza correctamente, sal del bucle. Si falla con un error de simultaneidad, incrementa el contador de reintentos y continúa el bucle desde el paso Lectura. -

Ejemplo

En el siguiente ejemplo, se muestra cómo implementar un bucle de OCC que actualiza una política de IAM en un recurso de Secret Manager otorgando a un miembro el rol roles/secretmanager.secretAccessor.

Para ejecutar este ejemplo, primero declara la dependencia en tu archivo Cargo.toml:

cargo add google-cloud-secretmanager-v1

Luego, implementa el bucle de reintentos de OCC:

const ROLE: &str = "roles/secretmanager.secretAccessor";
const ATTEMPTS: u32 = 5;

let secret_name = format!("projects/{project_id}/secrets/{secret_id}");
let client = SecretManagerService::builder().build().await?;
for _attempt in 0..ATTEMPTS {
    let mut current = client
        .get_iam_policy()
        .set_resource(&secret_name)
        .send()
        .await?;

    match current.bindings.iter_mut().find(|b| b.role == ROLE) {
        None => current
            .bindings
            .push(Binding::new().set_role(ROLE).set_members([member])),
        Some(b) => {
            if b.members.iter().any(|m| m == member) {
                return Ok(current);
            }
            b.members.push(member.to_string());
        }
    };
    let updated = client
        .set_iam_policy()
        .set_resource(&secret_name)
        .set_policy(current)
        .set_update_mask(FieldMask::default().set_paths(["bindings"]))
        .send()
        .await;
    match updated {
        Ok(p) => return Ok(p),
        Err(e)
            if e.status().is_some_and(|s| {
                s.code == Code::Aborted || s.code == Code::FailedPrecondition
            }) =>
        {
            continue;
        }
        Err(e) => return Err(e.into()),
    }
}
anyhow::bail!("could not set IAM policy after {ATTEMPTS} attempts")

Completar código

En el siguiente código, se muestra la función completa con todas las importaciones requeridas:

use google_cloud_gax::error::rpc::Code;
use google_cloud_iam_v1::model::Binding;
use google_cloud_iam_v1::model::Policy;
use google_cloud_secretmanager_v1::client::SecretManagerService;
use google_cloud_wkt::FieldMask;

/// Executes an Optimistic Concurrency Control (OCC) loop to safely update a resource.
///
/// This function demonstrates the core Read-Modify-Write-Retry pattern. It uses the secret manager
/// service and a hard-coded role. The principles apply to any other service or role.
///
/// # Parameters
/// * `project_id` The Google Cloud Project ID (e.g., "my-project-123").
/// * `secret_id` The Google Cloud Project ID (e.g., "my-secret").
/// * `member` The member to add (e.g., "user:user@example.com").
///
/// # Returns
/// The new IAM policy.
pub async fn sample(project_id: &str, secret_id: &str, member: &str) -> anyhow::Result<Policy> {
    const ROLE: &str = "roles/secretmanager.secretAccessor";
    const ATTEMPTS: u32 = 5;

    let secret_name = format!("projects/{project_id}/secrets/{secret_id}");
    let client = SecretManagerService::builder().build().await?;
    for _attempt in 0..ATTEMPTS {
        let mut current = client
            .get_iam_policy()
            .set_resource(&secret_name)
            .send()
            .await?;

        match current.bindings.iter_mut().find(|b| b.role == ROLE) {
            None => current
                .bindings
                .push(Binding::new().set_role(ROLE).set_members([member])),
            Some(b) => {
                if b.members.iter().any(|m| m == member) {
                    return Ok(current);
                }
                b.members.push(member.to_string());
            }
        };
        let updated = client
            .set_iam_policy()
            .set_resource(&secret_name)
            .set_policy(current)
            .set_update_mask(FieldMask::default().set_paths(["bindings"]))
            .send()
            .await;
        match updated {
            Ok(p) => return Ok(p),
            Err(e)
                if e.status().is_some_and(|s| {
                    s.code == Code::Aborted || s.code == Code::FailedPrecondition
                }) =>
            {
                continue;
            }
            Err(e) => return Err(e.into()),
        }
    }
    anyhow::bail!("could not set IAM policy after {ATTEMPTS} attempts")
}