オプティミスティック同時実行制御（OCC）の使用

オプティミスティック同時実行制御（OCC）は、共有リソースを管理し、複数のユーザーまたはプロセスが同じリソースを同時に変更した場合に、更新の損失や競合状態を防ぐための戦略です。

たとえば、Identity and Access Management を使用する場合、IAM ポリシーはプロジェクト、バケット、サービスなどのリソースに接続された共有リソースです。同時実行制御がないと、同時変更によって最近のポリシー変更が上書きされ、正当なアクセス権が誤って取り消されたり、不正な権限が付与されたりする可能性があります。

OCC を実装するために、システムは通常、リソース構造体のバージョン番号または etag（エンティティ タグ）フィールドを使用します。

OCC の仕組み

2 つのプロセス A と B が、共有リソースを同時に更新しようとしているとします。

  1. プロセス A がリソースの現在の状態を読み取ります。
  2. プロセス B が同じ現在の状態を読み取ります。
  3. プロセス A がコピーを変更し、サーバーに書き戻します。
  4. プロセス B がコピーを変更してサーバーに書き戻します。

プロセス B は、プロセス A がすでにリソースを変更していることを認識せずにリソースを上書きするため、プロセス A の更新は失われます。

OCC は、エンティティが変更されるたびに変化する一意のフィンガープリントを導入することで、この問題を解決します。IAM などの多くのシステムでは、このフィンガープリントは etag です。サーバーは、書き込みのたびにこのタグをチェックします。

  1. リソースを読み取ると、サーバーは etag（一意のフィンガープリント）を返します。
  2. 変更したリソースを返送するときは、元の etag を含める必要があります。
  3. 保存されている etag が送信した etag と一致しない場合（つまり、読み取り後に別のプロセスによってリソースが変更された場合）、書き込みオペレーションは失敗し、ABORTED エラーまたは FAILED_PRECONDITION エラーが発生します。

このエラーは、クライアントにプロセス全体の再試行を指示します。更新された状態を再度読み取り、変更を再適用して、新しい etag で書き込みを再試行します。

OCC ループを実装する

OCC 実装の中核は、リソースの読み取り、変更、保存を試みる再試行ループです。書き込みが成功すると、このループは終了します。競合が多い場合に無限ループが発生しないように、再試行の最大数を適切な値に設定します。

ループのステップ

次の表に、OCC 再試行ループの各ステップを示します。

ステップ アクション 実装例
読み取り etag を含む現在のリソースの状態を取得します。 
let mut policy = client
    .get_iam_policy()
    .set_resource(&secret_name)
    .send()
    .await?;
変更 変更をローカル構造体に適用します。 
policy.bindings.push(new_binding);
作成と確認 既存の etag を使用して、変更されたリソースの保存を試み、同時実行エラーコードを確認します。 
match client
    .set_iam_policy()
    .set_resource(&secret_name)
    .set_policy(policy)
    .send()
    .await
{
    Ok(p) => return Ok(p),
    Err(e) => { /* ... */ }
}
成功または再試行 書き込みが成功した場合は、ループを終了します。同時実行エラーで失敗した場合は、再試行カウンタをインクリメントし、読み取りステップからループを続行します。 -

例

次の例は、メンバーに roles/secretmanager.secretAccessor ロールを付与して、Secret Manager のシークレット リソースの IAM ポリシーを更新する OCC ループを実装する方法を示しています。

この例を実行するには、まず Cargo.toml ファイルで依存関係を宣言します。

cargo add google-cloud-secretmanager-v1

次に、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")

コードを補完する

次のコードは、必要なインポートをすべて含む完全な関数を示しています。

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