Mit optimistischer Nebenläufigkeitserkennung (OCC) arbeiten

Die optimistische Nebenläufigkeitserkennung (Optimistic Concurrency Control, OCC) ist eine Strategie zur Verwaltung gemeinsam genutzter Ressourcen und zur Verhinderung von verlorenen Aktualisierungen oder Race-Bedingungen, wenn mehrere Nutzer oder Prozesse dieselbe Ressource gleichzeitig ändern.

Wenn Sie beispielsweise mit Identity and Access Management arbeiten, ist eine IAM-Richtlinie eine freigegebene Ressource, die an eine Ressource wie ein Projekt, einen Bucket oder einen Dienst angehängt ist. Ohne Nebenläufigkeitserkennung können gleichzeitige Änderungen aktuelle Richtlinienänderungen überschreiben, wodurch versehentlich legitimer Zugriff widerrufen oder nicht autorisierte Berechtigungen gewährt werden.

Zur Implementierung von OCC verwenden Systeme in der Regel eine Versionsnummer oder ein etag-Feld (Entity Tag) in der Ressourcenstruktur.

Funktionsweise von OCC

Angenommen, zwei Prozesse, A und B, versuchen, eine freigegebene Ressource gleichzeitig zu aktualisieren:

  1. Prozess A liest den aktuellen Status der Ressource.
  2. Prozess B liest denselben aktuellen Status.
  3. Prozess A ändert seine Kopie und schreibt sie zurück auf den Server.
  4. Prozess B ändert seine Kopie und schreibt sie zurück auf den Server.

Da Prozess B die Ressource überschreibt, ohne zu wissen, dass Prozess A sie bereits geändert hat, gehen die Aktualisierungen von Prozess A verloren.

OCC löst dieses Problem durch die Einführung eines eindeutigen Fingerabdrucks, der sich bei jeder Änderung einer Entität ändert. In vielen Systemen, z. B. IAM, ist dieser Fingerabdruck ein etag. Der Server prüft dieses Tag bei jedem Schreibvorgang:

  1. Wenn Sie die Ressource lesen, gibt der Server ein etag (einen eindeutigen Fingerabdruck) zurück.
  2. Wenn Sie die geänderte Ressource zurücksenden, müssen Sie die ursprüngliche etag angeben.
  3. Wenn die gespeicherte etag nicht mit der von Ihnen gesendeten etag übereinstimmt, d. h. ein anderer Prozess die Ressource nach dem Lesen geändert hat, schlägt der Schreibvorgang mit einem ABORTED- oder FAILED_PRECONDITION-Fehler fehl.

Dieser Fehler signalisiert dem Client, den gesamten Vorgang zu wiederholen: den aktualisierten Status noch einmal lesen, die Änderungen noch einmal anwenden und den Schreibvorgang mit dem neuen etag noch einmal versuchen.

OCC-Schleife implementieren

Der Kern einer OCC-Implementierung ist eine Wiederholungsschleife, die die Ressource liest, ändert und versucht, sie zu speichern. Nach einem erfolgreichen Schreibvorgang wird die Schleife beendet. Konfigurieren Sie eine angemessene maximale Anzahl von Wiederholungsversuchen, um Endlosschleifen bei hoher Konfliktrate zu vermeiden.

Schritte der Schleife

In der folgenden Tabelle werden die einzelnen Schritte in einer OCC-Wiederholungsschleife beschrieben:

Schritt Aktion Implementierungsbeispiel
Lesen Rufen Sie den aktuellen Ressourcenstatus ab, einschließlich des etag. 
let mut policy = client
    .get_iam_policy()
    .set_resource(&secret_name)
    .send()
    .await?;
Ändern Wenden Sie die Änderungen auf die lokale Struktur an. 
policy.bindings.push(new_binding);
Schreiben und prüfen Versuchen Sie, die geänderte Ressource mit dem vorhandenen etag zu speichern, und prüfen Sie, ob Fehlercodes für Nebenläufigkeit vorliegen. 
match client
    .set_iam_policy()
    .set_resource(&secret_name)
    .set_policy(policy)
    .send()
    .await
{
    Ok(p) => return Ok(p),
    Err(e) => { /* ... */ }
}
Erfolg oder Wiederholung Wenn der Schreibvorgang erfolgreich ist, wird die Schleife beendet. Wenn ein Nebenläufigkeitsfehler auftritt, erhöhen Sie den Wiederholungszähler und setzen Sie die Schleife ab dem Schritt Lesen fort. -

Beispiel

Im folgenden Beispiel wird gezeigt, wie Sie eine OCC-Schleife implementieren, die eine IAM-Richtlinie für eine Secret Manager-Secret-Ressource aktualisiert, indem einem Mitglied die Rolle roles/secretmanager.secretAccessor zugewiesen wird.

Wenn Sie dieses Beispiel ausführen möchten, deklarieren Sie zuerst die Abhängigkeit in Ihrer Cargo.toml-Datei:

cargo add google-cloud-secretmanager-v1

Implementieren Sie dann die OCC-Wiederholungsschleife:

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

Vollständiger Code

Der folgende Code zeigt die vollständige Funktion mit allen erforderlichen Importen:

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