使用乐观并发控制 (OCC)

乐观并发控制 (OCC) 是一种策略，用于管理共享资源，并防止在多个用户或进程同时修改同一资源时发生更新丢失或竞态条件。

例如，在使用 Identity and Access Management 时，IAM 政策是附加到项目、存储桶或服务等资源的共享资源。如果没有并发控制，同时进行的修改可能会覆盖最近的政策更改，从而无意中撤消合法访问权限或授予未经授权的权限。

为了实现 OCC，系统通常会在资源结构体中使用版本号或 etag（实体标记）字段。

OCC 的工作原理

假设有两个进程 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 重试循环中的每个步骤：

步骤 操作 实现示例
Read（读取） 提取当前资源状态，包括 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) => { /* ... */ }
}
成功或重试 如果写入成功，则退出循环。如果因并发错误而失败，请递增重试计数器，并从读取步骤继续循环。 -

示例

以下示例演示了如何实现 OCC 循环，该循环通过向成员授予 roles/secretmanager.secretAccessor 角色来更新 Secret Manager 密文资源上的 IAM 政策。

如需运行此示例，请先在 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")
}