使用開放式並行控制 (OCC)

樂觀並行控制 (OCC) 是一種策略，用於管理共用資源，並在多位使用者或多個程序同時修改相同資源時，防止更新遺失或競爭狀況。

舉例來說，使用 Identity and Access Management 時，IAM 政策是附加至專案、bucket 或服務等資源的共用資源。如果沒有並行控制，同時修改可能會覆寫最近的政策變更，不小心撤銷合法存取權或授予未經授權的權限。

如要實作 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 重試迴圈中的每個步驟：

步驟 動作 導入範例
閱讀 擷取目前的資源狀態，包括 etag。 
let mut policy = client
    .get_iam_policy()
    .set_resource(&secret_name)
    .send()
    .await?;
修改 將變更套用至本機 struct。 
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")
}