diff --git a/crates/openshell-server/src/persistence/postgres.rs b/crates/openshell-server/src/persistence/postgres.rs index cdbcc41dd8..071a71c0a3 100644 --- a/crates/openshell-server/src/persistence/postgres.rs +++ b/crates/openshell-server/src/persistence/postgres.rs @@ -1071,33 +1071,33 @@ FOR UPDATE let sandbox_payload: Vec = row.get("payload"); let current_version: i64 = row.try_get("resource_version").unwrap_or(1); let current_version = current_version.max(1).cast_unsigned(); - let (mut sandbox, sandbox_changed) = + // Every committed revision advances the sandbox version, even when the + // projection is a no-op. Callers use the version as their optimistic + // concurrency precondition, so a revision that left it untouched would let + // a second writer holding the previous value overwrite a newer policy. + let (mut sandbox, _projection_changed) = project_policy_revision_onto_sandbox(write, &sandbox_payload, current_version)?; - let resulting_version = if sandbox_changed { - let result = sqlx::query( - r" + let update = sqlx::query( + r" UPDATE objects SET payload = $2, updated_at_ms = $3, resource_version = resource_version + 1 WHERE object_type = 'sandbox' AND id = $1 AND resource_version = $4 ", - ) - .bind(&write.sandbox_id) - .bind(sandbox.encode_to_vec()) - .bind(now_ms) - .bind(i64::try_from(current_version).unwrap_or(i64::MAX)) - .execute(&mut *tx) - .await - .map_err(|e| map_db_error(&e))?; - if result.rows_affected() != 1 { - return Err(PersistenceError::Conflict { - current_resource_version: Some(current_version), - }); - } - current_version.saturating_add(1) - } else { - current_version - }; + ) + .bind(&write.sandbox_id) + .bind(sandbox.encode_to_vec()) + .bind(now_ms) + .bind(i64::try_from(current_version).unwrap_or(i64::MAX)) + .execute(&mut *tx) + .await + .map_err(|e| map_db_error(&e))?; + if update.rows_affected() != 1 { + return Err(PersistenceError::Conflict { + current_resource_version: Some(current_version), + }); + } + let resulting_version = current_version.saturating_add(1); sqlx::query( r" diff --git a/crates/openshell-server/src/persistence/sqlite.rs b/crates/openshell-server/src/persistence/sqlite.rs index ce269cd241..331fbbab22 100644 --- a/crates/openshell-server/src/persistence/sqlite.rs +++ b/crates/openshell-server/src/persistence/sqlite.rs @@ -1304,33 +1304,33 @@ WHERE "object_type" = 'sandbox' AND "id" = ?1 let sandbox_payload: Vec = row.get("payload"); let current_version: i64 = row.try_get("resource_version").unwrap_or(1); let current_version = current_version.max(1).cast_unsigned(); - let (mut sandbox, sandbox_changed) = + // Every committed revision advances the sandbox version, even when the + // projection is a no-op. Callers use the version as their optimistic + // concurrency precondition, so a revision that left it untouched would let + // a second writer holding the previous value overwrite a newer policy. + let (mut sandbox, _projection_changed) = project_policy_revision_onto_sandbox(write, &sandbox_payload, current_version)?; - let resulting_version = if sandbox_changed { - let result = sqlx::query( - r#" + let update = sqlx::query( + r#" UPDATE "objects" SET "payload" = ?2, "updated_at_ms" = ?3, "resource_version" = "resource_version" + 1 WHERE "object_type" = 'sandbox' AND "id" = ?1 AND "resource_version" = ?4 "#, - ) - .bind(&write.sandbox_id) - .bind(sandbox.encode_to_vec()) - .bind(now_ms) - .bind(i64::try_from(current_version).unwrap_or(i64::MAX)) - .execute(&mut *tx) - .await - .map_err(|e| map_db_error(&e))?; - if result.rows_affected() != 1 { - return Err(PersistenceError::Conflict { - current_resource_version: Some(current_version), - }); - } - current_version.saturating_add(1) - } else { - current_version - }; + ) + .bind(&write.sandbox_id) + .bind(sandbox.encode_to_vec()) + .bind(now_ms) + .bind(i64::try_from(current_version).unwrap_or(i64::MAX)) + .execute(&mut *tx) + .await + .map_err(|e| map_db_error(&e))?; + if update.rows_affected() != 1 { + return Err(PersistenceError::Conflict { + current_resource_version: Some(current_version), + }); + } + let resulting_version = current_version.saturating_add(1); sqlx::query( r#" diff --git a/crates/openshell-server/src/persistence/tests.rs b/crates/openshell-server/src/persistence/tests.rs index 7ba0bd1245..885c0d3a55 100644 --- a/crates/openshell-server/src/persistence/tests.rs +++ b/crates/openshell-server/src/persistence/tests.rs @@ -1458,6 +1458,20 @@ async fn policy_atomic_write_rolls_back_sandbox_when_revision_insert_conflicts() ); assert!(after.metadata.as_ref().unwrap().annotations.is_empty()); assert!(after.spec.as_ref().unwrap().policy.is_none()); + let latest = store + .get_latest_policy("sandbox-rollback") + .await + .unwrap() + .unwrap(); + assert_eq!(latest.version, 1); + assert_eq!(latest.id, "existing-policy"); + assert!( + store + .get("sandbox_policy", "conflicting-policy") + .await + .unwrap() + .is_none() + ); } #[tokio::test] @@ -1499,6 +1513,314 @@ async fn policy_atomic_write_persists_workspace() { assert_eq!(record.workspace, "my-workspace"); } +/// A committed revision must invalidate the caller's optimistic-concurrency +/// precondition even when it changes nothing on the sandbox. Policy revision +/// numbers are independent of the sandbox resource version, so a writer that +/// skips the version bump lets a second writer holding the previous value +/// commit policy content derived from a read that is now stale. +#[tokio::test] +async fn policy_atomic_write_advances_resource_version_without_projection_change() { + let store = test_store().await; + store + .put_message(&policy_test_sandbox("sandbox-noop", "noop")) + .await + .unwrap(); + let before = store + .get_message::("sandbox-noop") + .await + .unwrap() + .unwrap(); + let before_version = before.metadata.as_ref().unwrap().resource_version; + let policy = SandboxPolicy { + version: 1, + ..Default::default() + }; + + let updated = store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-noop-1".to_string(), + sandbox_id: "sandbox-noop".to_string(), + workspace: "default".to_string(), + version: 1, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-noop-1".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: before_version, + annotations: StdHashMap::new(), + backfill_policy: None, + }) + .await + .unwrap(); + assert_eq!( + updated.metadata.as_ref().unwrap().resource_version, + before_version + 1 + ); + + let row = store.get("sandbox", "sandbox-noop").await.unwrap().unwrap(); + assert_eq!(row.resource_version, before_version + 1); + let stored = Sandbox::decode(row.payload.as_slice()).unwrap(); + assert!(stored.spec.as_ref().unwrap().policy.is_none()); + assert!(stored.metadata.as_ref().unwrap().annotations.is_empty()); + + // The transaction rewrites the sandbox row, so its modification time is the + // revision's own creation time rather than the sandbox's original one. + let revision = store + .get_latest_policy("sandbox-noop") + .await + .unwrap() + .unwrap(); + assert_eq!(row.updated_at_ms, revision.created_at_ms); +} + +/// Backfilling a policy that already matches is the common no-op projection: +/// the gateway discovers the policy the sandbox is already running and +/// re-submits it. It is still a new revision, so it still has to advance the +/// version. +#[tokio::test] +async fn policy_atomic_write_advances_resource_version_with_identical_backfill_policy() { + let store = test_store().await; + let policy = SandboxPolicy { + version: 1, + ..Default::default() + }; + let mut sandbox = policy_test_sandbox("sandbox-identical", "identical"); + sandbox.spec = Some(SandboxSpec { + policy: Some(policy.clone()), + ..Default::default() + }); + store.put_message(&sandbox).await.unwrap(); + let before = store + .get_message::("sandbox-identical") + .await + .unwrap() + .unwrap(); + let before_version = before.metadata.as_ref().unwrap().resource_version; + + store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-identical-1".to_string(), + sandbox_id: "sandbox-identical".to_string(), + workspace: "default".to_string(), + version: 1, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-identical-1".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: before_version, + annotations: StdHashMap::new(), + backfill_policy: Some(policy.clone()), + }) + .await + .unwrap(); + + let row = store + .get("sandbox", "sandbox-identical") + .await + .unwrap() + .unwrap(); + assert_eq!(row.resource_version, before_version + 1); + let stored = Sandbox::decode(row.payload.as_slice()).unwrap(); + assert_eq!(stored.spec.as_ref().unwrap().policy.as_ref(), Some(&policy)); +} + +#[tokio::test] +async fn policy_atomic_write_rejects_stale_expected_resource_version() { + let store = test_store().await; + store + .put_message(&policy_test_sandbox("sandbox-stale", "stale")) + .await + .unwrap(); + let before = store + .get_message::("sandbox-stale") + .await + .unwrap() + .unwrap(); + let before_version = before.metadata.as_ref().unwrap().resource_version; + let policy = SandboxPolicy { + version: 1, + ..Default::default() + }; + + store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-stale-1".to_string(), + sandbox_id: "sandbox-stale".to_string(), + workspace: "default".to_string(), + version: 1, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-stale-1".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: before_version, + annotations: StdHashMap::new(), + backfill_policy: None, + }) + .await + .unwrap(); + + let error = store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-stale-2".to_string(), + sandbox_id: "sandbox-stale".to_string(), + workspace: "default".to_string(), + version: 2, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-stale-2".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: before_version, + annotations: StdHashMap::new(), + backfill_policy: None, + }) + .await + .unwrap_err(); + assert!( + matches!( + error, + PersistenceError::Conflict { + current_resource_version: Some(version) + } if version == before_version + 1 + ), + "{error:?}" + ); + + let latest = store + .get_latest_policy("sandbox-stale") + .await + .unwrap() + .unwrap(); + assert_eq!(latest.version, 1); + let after = store + .get_message::("sandbox-stale") + .await + .unwrap() + .unwrap(); + assert_eq!( + after.metadata.as_ref().unwrap().resource_version, + before_version + 1 + ); +} + +#[tokio::test] +async fn policy_atomic_write_succeeds_after_rereading_resource_version() { + let store = test_store().await; + store + .put_message(&policy_test_sandbox("sandbox-retry", "retry")) + .await + .unwrap(); + let before = store + .get_message::("sandbox-retry") + .await + .unwrap() + .unwrap(); + let before_version = before.metadata.as_ref().unwrap().resource_version; + let policy = SandboxPolicy { + version: 1, + ..Default::default() + }; + + store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-retry-1".to_string(), + sandbox_id: "sandbox-retry".to_string(), + workspace: "default".to_string(), + version: 1, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-retry-1".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: before_version, + annotations: StdHashMap::new(), + backfill_policy: None, + }) + .await + .unwrap(); + + let reread = store + .get_message::("sandbox-retry") + .await + .unwrap() + .unwrap(); + let reread_version = reread.metadata.as_ref().unwrap().resource_version; + assert_eq!(reread_version, before_version + 1); + + store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-retry-2".to_string(), + sandbox_id: "sandbox-retry".to_string(), + workspace: "default".to_string(), + version: 2, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-retry-2".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: reread_version, + annotations: StdHashMap::new(), + backfill_policy: None, + }) + .await + .unwrap(); + + let latest = store + .get_latest_policy("sandbox-retry") + .await + .unwrap() + .unwrap(); + assert_eq!(latest.version, 2); + let after = store + .get_message::("sandbox-retry") + .await + .unwrap() + .unwrap(); + assert_eq!( + after.metadata.as_ref().unwrap().resource_version, + before_version + 2 + ); +} + +/// An unconditional write still has to advance the version. Otherwise a client +/// that never reads a sandbox would keep submitting the same `0` precondition +/// forever while other writers' revisions went unnoticed. +#[tokio::test] +async fn policy_atomic_write_with_zero_expected_version_still_advances() { + let store = test_store().await; + store + .put_message(&policy_test_sandbox( + "sandbox-unconditional", + "unconditional", + )) + .await + .unwrap(); + let before = store + .get_message::("sandbox-unconditional") + .await + .unwrap() + .unwrap(); + let before_version = before.metadata.as_ref().unwrap().resource_version; + let policy = SandboxPolicy { + version: 1, + ..Default::default() + }; + + store + .put_policy_revision_atomic(&AtomicPolicyRevisionWrite { + id: "policy-unconditional-1".to_string(), + sandbox_id: "sandbox-unconditional".to_string(), + workspace: "default".to_string(), + version: 1, + policy_payload: policy.encode_to_vec(), + policy_hash: "hash-unconditional-1".to_string(), + provenance: StdHashMap::new(), + expected_resource_version: 0, + annotations: StdHashMap::new(), + backfill_policy: None, + }) + .await + .unwrap(); + + let row = store + .get("sandbox", "sandbox-unconditional") + .await + .unwrap() + .unwrap(); + assert_eq!(row.resource_version, before_version + 1); +} + #[tokio::test] async fn policy_put_and_get_latest() { let store = test_store().await;