From 0f18d50c78cdb7ae46861508c6641342028ad665 Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Thu, 16 Jul 2026 22:01:15 -0700 Subject: [PATCH 1/4] fix(server): unblock sandbox deletion events Signed-off-by: Piotr Mlocek --- architecture/compute-runtimes.md | 24 + crates/openshell-server/src/compute/mod.rs | 2189 +++++++++++++++-- .../openshell-server/src/persistence/mod.rs | 40 + .../src/persistence/postgres.rs | 87 + .../src/persistence/sqlite.rs | 82 + .../openshell-server/src/persistence/tests.rs | 207 ++ e2e/rust/tests/sandbox_lifecycle.rs | 55 +- 7 files changed, 2425 insertions(+), 259 deletions(-) diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index ec42277ab6..010d8a9e05 100644 --- a/architecture/compute-runtimes.md +++ b/architecture/compute-runtimes.md @@ -30,6 +30,30 @@ The gateway records driver identity and version from the startup capability response. Elevated gateway info reports that initialized driver snapshot instead of re-querying drivers on each request. +## Deletion Lifecycle + +Delete requests use per-sandbox gates to serialize duplicate driver calls. A +request resolves the name once and remains bound to that stable ID. The only +combined lock order is delete gate, then the gateway-wide state guard; external +driver calls run without the global guard. + +Watcher events do not acquire delete gates. Exact resource-version checks allow +them to interleave safely: status snapshots are no-ops for `Deleting` rows, +deleted events are idempotent, and snapshots for absent rows are ignored. + +An accepted delete (`deleted = true`) is finalized by the watcher. If the +backend is already absent (`deleted = false`), the request removes gateway state +synchronously. Sandbox rows and name-scoped settings are removed atomically; +SSH sessions, indexes, and watch/log buses are cleaned only after confirmed +removal. + +The request workflow runs in an owned task so client cancellation cannot strand +a mutation. A gateway restart does not resume or reissue a persisted `Deleting` +operation. If the backend completed the delete, watcher or reconciliation +cleanup removes the row. If the backend never accepted the call, the row can +remain `Deleting`; crash-safe retries require a durable operation generation +that the compute driver honors. + ## Runtime Summary | Runtime | Best fit | Sandbox boundary | Notes | diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 239feb74d5..daa618f08c 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -44,7 +44,7 @@ use std::fmt; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::pin::Pin; -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex, Weak}; use std::time::Duration; #[cfg(unix)] use tokio::net::UnixStream; @@ -60,6 +60,60 @@ type SharedComputeDriver = const DELETE_PHASE_CAS_RETRY_LIMIT: usize = 3; +/// Serializes request-side deletes for the same stable sandbox ID. +/// +/// Watch events deliberately do not use these gates, so a slow driver delete +/// cannot block the sequential watch loop. Weak values let entries disappear +/// after the last request using a sandbox's gate completes. +#[derive(Debug, Default)] +struct DeleteGateRegistry { + gates: StdMutex>>>, +} + +impl DeleteGateRegistry { + fn gate_for(&self, sandbox_id: &str) -> Arc> { + let mut gates = self + .gates + .lock() + .expect("sandbox delete gate registry lock poisoned"); + gates.retain(|_, gate| gate.strong_count() > 0); + + if let Some(gate) = gates.get(sandbox_id).and_then(Weak::upgrade) { + return gate; + } + + let gate = Arc::new(Mutex::new(())); + gates.insert(sandbox_id.to_string(), Arc::downgrade(&gate)); + gate + } + + #[cfg(test)] + fn entry_count(&self) -> usize { + self.gates + .lock() + .expect("sandbox delete gate registry lock poisoned") + .len() + } +} + +#[derive(Debug)] +struct DeleteTransition { + /// Snapshot restored if the driver result is ambiguous and no newer write + /// has replaced the exact `Deleting` version. + previous: Sandbox, + /// Exact durable `Deleting` snapshot owned by this request. + deleting: Sandbox, +} + +/// Result of trying to claim ownership of a sandbox's delete operation. +#[derive(Debug)] +enum BeginDelete { + /// This request persisted `Deleting` and owns the driver call and recovery. + Started(Box), + /// Another request already owns or completed the driver-side delete call. + AlreadyDeleting, +} + #[derive(Debug, Clone)] pub struct ComputeDriverInfoSnapshot { /// Gateway-selected driver name used for routing and `driver_config` keys. @@ -280,6 +334,7 @@ pub struct ComputeRuntime { tracing_log_bus: TracingLogBus, supervisor_sessions: Arc, sync_lock: Arc>, + delete_gates: Arc, gateway_bind_addresses: Vec, replica_id: String, } @@ -336,6 +391,7 @@ impl ComputeRuntime { tracing_log_bus, supervisor_sessions, sync_lock: Arc::new(Mutex::new(())), + delete_gates: Arc::new(DeleteGateRegistry::default()), gateway_bind_addresses, replica_id: lease::replica_id(), }) @@ -592,58 +648,104 @@ impl ComputeRuntime { } pub async fn delete_sandbox(&self, name: &str) -> Result { - let _guard = self.sync_lock.lock().await; + // The owned task keeps the delete workflow running if the client drops + // its request after the gateway has durably persisted `Deleting`. We + // still await it here so the API remains synchronous for live clients. + let runtime = self.clone(); + let name = name.to_string(); + tokio::spawn(async move { runtime.delete_sandbox_inner(&name).await }) + .await + .map_err(|err| { + Status::internal(format!( + "sandbox delete worker terminated unexpectedly: {err}" + )) + })? + } - // Resolve sandbox ID from name - let sandbox = self + async fn delete_sandbox_inner(&self, name: &str) -> Result { + // Resolve the name once. Everything after this point is bound to the + // resulting stable ID, so name reuse cannot redirect this request to a + // newly-created sandbox. + let candidate = self .store .get_message_by_name::(name) .await + .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))? + .ok_or_else(|| Status::not_found("sandbox not found"))?; + let candidate_id = candidate.object_id().to_string(); + let delete_gate = self.delete_gates.gate_for(&candidate_id); + let _delete_guard = delete_gate.lock().await; + + let guard = self.sync_lock.lock().await; + let current = self + .store + .get_message::(&candidate_id) + .await .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))?; - - let Some(sandbox) = sandbox else { - return Err(Status::not_found("sandbox not found")); + let Some(current) = current else { + // A delete that owned this ID's gate completed while this request + // waited. A different sandbox may now use the old name; this + // request acknowledges only the disappearance of its original ID. + self.cleanup_removed_sandbox_state(&candidate_id).await; + return Ok(true); }; + if current.object_name() != name { + return Err(Status::aborted( + "sandbox name changed while the delete request was waiting; retry explicitly", + )); + } - let id = sandbox.object_id().to_string(); - - let sandbox = self.set_sandbox_phase_deleting_with_retry(&id).await?; + // `Started` carries both sides of the CAS transition: the durable + // `Deleting` row used to fence recovery, and the prior row used only + // for exact-version rollback after an ambiguous driver failure. + let transition = match self + .begin_sandbox_delete_with_initial_snapshot(&candidate_id, Some(current)) + .await? + { + BeginDelete::AlreadyDeleting => return Ok(true), + BeginDelete::Started(transition) => *transition, + }; - self.sandbox_index.update_from_sandbox(&sandbox); - self.sandbox_watch_bus.notify(&id); - self.cleanup_sandbox_owned_records(&sandbox).await; + self.sandbox_index.update_from_sandbox(&transition.deleting); + self.sandbox_watch_bus.notify(&candidate_id); + drop(guard); - let deleted = self + let result = self .driver .delete_sandbox(Request::new(DeleteSandboxRequest { - sandbox_id: sandbox.object_id().to_string(), - sandbox_name: sandbox.object_name().to_string(), + sandbox_id: transition.deleting.object_id().to_string(), + sandbox_name: transition.deleting.object_name().to_string(), })) - .await - .map(|response| response.into_inner().deleted) - .map_err(|err| Status::internal(format!("delete sandbox failed: {}", err.message())))?; + .await; - if !deleted && let Err(e) = self.store.delete(Sandbox::object_type(), &id).await { - warn!(sandbox_id = %id, error = %e, "Failed to clean up store after delete"); + match result { + Ok(response) => { + let deleted = response.into_inner().deleted; + if deleted { + self.cleanup_local_state_if_sandbox_absent(&candidate_id) + .await?; + } else if !self.remove_deleting_sandbox_record(&candidate_id).await { + return Err(Status::internal( + "compute resource was absent, but gateway cleanup did not complete", + )); + } + Ok(deleted) + } + Err(err) => { + self.recover_failed_delete(&transition).await; + Err(Status::internal(format!( + "delete sandbox failed: {}", + err.message() + ))) + } } - - self.cleanup_sandbox_state(&id); - Ok(deleted) - } - - async fn set_sandbox_phase_deleting_with_retry( - &self, - sandbox_id: &str, - ) -> Result { - self.set_sandbox_phase_deleting_with_initial_snapshot(sandbox_id, None) - .await } - async fn set_sandbox_phase_deleting_with_initial_snapshot( + async fn begin_sandbox_delete_with_initial_snapshot( &self, sandbox_id: &str, mut initial_snapshot: Option, - ) -> Result { + ) -> Result { let operation = "set sandbox phase to Deleting"; for attempt in 1..=DELETE_PHASE_CAS_RETRY_LIMIT { @@ -657,18 +759,29 @@ impl ComputeRuntime { .ok_or_else(|| Status::not_found("sandbox not found"))?, }; + if SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown) + == SandboxPhase::Deleting + { + return Ok(BeginDelete::AlreadyDeleting); + } + + let previous = sandbox.clone(); + match self .write_sandbox_phase_deleting_from_snapshot(sandbox) .await { - Ok(sandbox) => { + Ok(deleting) => { if attempt > 1 { debug!( sandbox_id, attempt, "Retried sandbox delete phase transition after CAS conflict" ); } - return Ok(sandbox); + return Ok(BeginDelete::Started(Box::new(DeleteTransition { + previous, + deleting, + }))); } Err(crate::persistence::PersistenceError::Conflict { current_resource_version, @@ -694,6 +807,264 @@ impl ComputeRuntime { unreachable!("delete phase retry loop always returns") } + /// Removes a durable `Deleting` row after the driver confirms that its + /// compute resource is already absent (`deleted = false`). + /// + /// Benign resource-version changes are retried, but cleanup stops if the + /// row leaves `Deleting`; that state belongs to a concurrent writer. + async fn remove_deleting_sandbox_record(&self, sandbox_id: &str) -> bool { + let _guard = self.sync_lock.lock().await; + for attempt in 1..=DELETE_PHASE_CAS_RETRY_LIMIT { + let record = match self.store.get(Sandbox::object_type(), sandbox_id).await { + Ok(Some(record)) => record, + Ok(None) => { + self.cleanup_removed_sandbox_state(sandbox_id).await; + return true; + } + Err(err) => { + warn!( + sandbox_id, + error = %err, + "Failed to fetch sandbox after the compute resource was absent" + ); + return false; + } + }; + let sandbox = match decode_sandbox_record(&record) { + Ok(sandbox) => sandbox, + Err(err) => { + warn!(sandbox_id, error = %err, "Failed to decode sandbox during cleanup"); + return false; + } + }; + if SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown) + != SandboxPhase::Deleting + { + debug!( + sandbox_id, + "Skipped cleanup of a sandbox no longer deleting" + ); + return false; + } + + match self + .remove_sandbox_record_if_version_locked(sandbox_id, record.resource_version) + .await + { + Ok(true) => return true, + Ok(false) if attempt < DELETE_PHASE_CAS_RETRY_LIMIT => { + tokio::task::yield_now().await; + } + Ok(false) => return false, + Err(err) => { + warn!( + sandbox_id, + error = %err, + "Failed to clean up store after the compute resource was absent" + ); + return false; + } + } + } + + false + } + + /// Atomically removes the sandbox and its name-scoped settings only when + /// the expected resource version still owns the row. The caller holds + /// `sync_lock`; successful or already-completed durable removal also + /// clears this replica's index, sessions, and watch/log buses. + async fn remove_sandbox_record_if_version_locked( + &self, + sandbox_id: &str, + expected_resource_version: u64, + ) -> Result { + let Some(record) = self + .store + .get(Sandbox::object_type(), sandbox_id) + .await + .map_err(|err| err.to_string())? + else { + self.cleanup_removed_sandbox_state(sandbox_id).await; + return Ok(false); + }; + + if record.resource_version != expected_resource_version { + debug!( + sandbox_id, + expected_resource_version, + current_resource_version = record.resource_version, + "Skipped sandbox cleanup after a concurrent state change" + ); + return Ok(false); + } + + match self + .store + .delete_with_name_scoped_if_version( + Sandbox::object_type(), + sandbox_id, + expected_resource_version, + SANDBOX_SETTINGS_OBJECT_TYPE, + ) + .await + { + Ok(true) => { + self.cleanup_removed_sandbox_state(sandbox_id).await; + Ok(true) + } + Ok(false) => { + if self + .store + .get(Sandbox::object_type(), sandbox_id) + .await + .map_err(|err| err.to_string())? + .is_none() + { + self.cleanup_removed_sandbox_state(sandbox_id).await; + } + Ok(false) + } + Err(crate::persistence::PersistenceError::Conflict { + current_resource_version, + }) => { + debug!( + sandbox_id, + expected_resource_version, + current_resource_version, + "Skipped sandbox cleanup after a concurrent state change" + ); + Ok(false) + } + Err(err) => Err(err.to_string()), + } + } + + /// Resolves an ambiguous driver delete error without overwriting newer + /// gateway state. + /// + /// The external lookup runs without `sync_lock`. Recovery then uses the + /// exact `Deleting` resource version to apply one of three outcomes: + /// reconcile an observed backend snapshot, remove a confirmed-absent + /// backend, or restore the pre-delete snapshot when lookup is inconclusive. + async fn recover_failed_delete(&self, transition: &DeleteTransition) { + let sandbox_id = transition.deleting.object_id(); + let sandbox_name = transition.deleting.object_name(); + let deleting_resource_version = sandbox_resource_version(&transition.deleting); + + // The driver lookup is deliberately outside the process-wide guard. + let observed = self.get_driver_sandbox(sandbox_id, sandbox_name).await; + let _guard = self.sync_lock.lock().await; + + match observed { + Ok(Some(snapshot)) if snapshot.id == sandbox_id && snapshot.status.is_some() => { + let session_connected = self.supervisor_sessions.has_session(sandbox_id); + match self + .store + .update_message_cas::( + sandbox_id, + deleting_resource_version, + |sandbox| { + apply_driver_snapshot(sandbox, &snapshot, session_connected); + }, + ) + .await + { + Ok(recovered) => { + self.sandbox_index.update_from_sandbox(&recovered); + self.sandbox_watch_bus.notify(sandbox_id); + } + Err(err) => { + self.handle_delete_recovery_write_failure( + sandbox_id, + err, + "reconcile observed backend snapshot", + ) + .await; + } + } + } + Ok(None) => { + match self + .remove_sandbox_record_if_version_locked(sandbox_id, deleting_resource_version) + .await + { + Ok(_) => {} + Err(err) => { + debug!( + sandbox_id, + error = %err, + "Skipped absent-backend recovery after a concurrent state change" + ); + } + } + } + Ok(Some(_)) | Err(_) => { + let previous = transition.previous.clone(); + match self + .store + .update_message_cas::( + sandbox_id, + deleting_resource_version, + |sandbox| *sandbox = previous.clone(), + ) + .await + { + Ok(recovered) => { + self.sandbox_index.update_from_sandbox(&recovered); + self.sandbox_watch_bus.notify(sandbox_id); + } + Err(err) => { + self.handle_delete_recovery_write_failure( + sandbox_id, + err, + "restore pre-delete snapshot", + ) + .await; + } + } + } + } + } + + /// Handles a failed recovery CAS while the caller holds `sync_lock`. + /// Another replica may have removed the durable row during the external + /// driver lookup; in that case this replica still needs local cleanup. + async fn handle_delete_recovery_write_failure( + &self, + sandbox_id: &str, + error: crate::persistence::PersistenceError, + recovery_action: &'static str, + ) { + match self.store.get(Sandbox::object_type(), sandbox_id).await { + Ok(None) => { + debug!( + sandbox_id, + recovery_action, + "Delete recovery found the row removed by another replica; cleaning local state" + ); + self.cleanup_removed_sandbox_state(sandbox_id).await; + } + Ok(Some(_)) => { + debug!( + sandbox_id, + recovery_action, + error = %error, + "Skipped delete recovery after a concurrent state change" + ); + } + Err(fetch_error) => { + warn!( + sandbox_id, + recovery_action, + error = %error, + fetch_error = %fetch_error, + "Failed to verify sandbox state after delete recovery conflict" + ); + } + } + } + async fn write_sandbox_phase_deleting_from_snapshot( &self, mut sandbox: Sandbox, @@ -1152,147 +1523,36 @@ impl ComputeRuntime { incoming: DriverSandbox, existing_record: Option, ) -> Result<(), String> { - let existing = existing_record - .as_ref() - .map(decode_sandbox_record) - .transpose()?; - - // If no existing record, create initial sandbox (first watch event for this sandbox) - if existing.is_none() { - use crate::persistence::WriteCondition; - let now_ms = openshell_core::time::now_ms(); - - let session_connected = self.supervisor_sessions.has_session(&incoming.id); - let mut phase = derive_phase(incoming.status.as_ref()); - let sandbox_name = incoming.name.clone(); - - let supervisor_promoted = session_connected - && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); - if supervisor_promoted { - phase = SandboxPhase::Ready; - } - - let mut status = incoming - .status - .as_ref() - .map(|s| public_status_from_driver(s, phase, 0)); - rewrite_user_facing_conditions(&mut status, None); - if supervisor_promoted { - ensure_supervisor_ready_status(&mut status, &sandbox_name); - } - let mut sandbox = Sandbox { - metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { - id: incoming.id.clone(), - name: sandbox_name, - created_at_ms: now_ms, - labels: std::collections::HashMap::new(), - resource_version: 0, - annotations: std::collections::HashMap::new(), - }), - spec: None, - status, - }; - sandbox.set_phase(phase as i32); - - self.store - .put_if( - Sandbox::object_type(), - &incoming.id, - sandbox.object_name(), - &sandbox.encode_to_vec(), - None, - WriteCondition::MustCreate, - ) - .await - .map_err(|e| match e { - crate::persistence::PersistenceError::Conflict { - current_resource_version, - } => format!( - "concurrent modification detected during sandbox creation (current resource_version: {})", - current_resource_version - .map_or_else(|| "unknown".to_string(), |v| v.to_string()) - ), - other => other.to_string(), - })?; + let Some(existing_record) = existing_record else { + // The gateway store is authoritative. API creation persists the + // complete row before asking the driver to create its resource, so + // an unknown snapshot is either stale or unmanaged. + debug!( + sandbox_id = %incoming.id, + sandbox_name = %incoming.name, + "Ignoring driver snapshot for sandbox absent from gateway store" + ); + return Ok(()); + }; + let existing = decode_sandbox_record(&existing_record)?; - self.sandbox_index.update_from_sandbox(&sandbox); - self.sandbox_watch_bus.notify(sandbox.object_id()); + if SandboxPhase::try_from(existing.phase()).unwrap_or(SandboxPhase::Unknown) + == SandboxPhase::Deleting + { + // Ordinary driver snapshots cannot recover or regress a durable + // deletion. Delete-failure recovery is explicit and version-bound. return Ok(()); } // Single-attempt CAS: on conflict, the next watch event will naturally retry let session_connected = self.supervisor_sessions.has_session(&incoming.id); - let sandbox_name = incoming.name.clone(); - let sandbox = self .store - .update_message_cas::(&incoming.id, 0, |sandbox| { - let old_phase = - SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); - let mut phase = incoming - .status - .as_ref() - .map_or(old_phase, |status| derive_phase(Some(status))); - let supervisor_promoted = session_connected - && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); - if supervisor_promoted { - phase = SandboxPhase::Ready; - } - - let cpv = sandbox.current_policy_version(); - let mut status = incoming - .status - .as_ref() - .map(|s| public_status_from_driver(s, phase, cpv)) - .or_else(|| sandbox.status.clone()); - rewrite_user_facing_conditions(&mut status, sandbox.spec.as_ref()); - if supervisor_promoted { - ensure_supervisor_ready_status(&mut status, &sandbox_name); - } - - if let Some(s) = status.as_mut() - && s.sandbox_name.is_empty() - { - s.sandbox_name.clone_from(&sandbox_name); - } - - if old_phase != phase { - info!( - sandbox_id = %incoming.id, - sandbox_name = %sandbox_name, - old_phase = ?old_phase, - new_phase = ?phase, - "Sandbox phase changed" - ); - } - - if phase == SandboxPhase::Error - && let Some(ref status) = status - { - for condition in &status.conditions { - if condition.r#type == "Ready" - && condition.status.eq_ignore_ascii_case("false") - && is_terminal_failure_reason(&condition.reason) - { - warn!( - sandbox_id = %incoming.id, - sandbox_name = %sandbox_name, - reason = %condition.reason, - message = %condition.message, - "Sandbox failed to become ready" - ); - } - } - } - - // Update metadata fields - if let Some(metadata) = sandbox.metadata.as_mut() { - metadata.name.clone_from(&sandbox_name); - } - sandbox.status = status; - sandbox.set_phase(phase as i32); - sandbox.set_current_policy_version(cpv); - }) + .update_message_cas::( + &incoming.id, + existing_record.resource_version, + |sandbox| apply_driver_snapshot(sandbox, &incoming, session_connected), + ) .await .map_err(|e| match e { crate::persistence::PersistenceError::Conflict { @@ -1325,23 +1585,33 @@ impl ComputeRuntime { ) -> Result<(), String> { let _guard = self.sync_lock.lock().await; + let Some(existing) = self + .store + .get_message::(sandbox_id) + .await + .map_err(|err| err.to_string())? + else { + return Ok(()); + }; + let current_phase = + SandboxPhase::try_from(existing.phase()).unwrap_or(SandboxPhase::Unknown); + if current_phase == SandboxPhase::Deleting || current_phase == SandboxPhase::Error { + return Ok(()); + } + if !connected && current_phase != SandboxPhase::Ready { + return Ok(()); + } + let expected_resource_version = sandbox_resource_version(&existing); + // Use CAS to update sandbox phase based on supervisor session state let result = self .store - .update_message_cas::(sandbox_id, 0, |sandbox| { - let current_phase = - SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); - - // Skip if sandbox is in terminal state - if current_phase == SandboxPhase::Deleting || current_phase == SandboxPhase::Error { - return; - } - + .update_message_cas::(sandbox_id, expected_resource_version, |sandbox| { let sandbox_name = sandbox.object_name().to_string(); if connected { ensure_supervisor_ready_status(&mut sandbox.status, &sandbox_name); sandbox.set_phase(SandboxPhase::Ready as i32); - } else if current_phase == SandboxPhase::Ready { + } else { ensure_supervisor_not_ready_status(&mut sandbox.status, &sandbox_name); sandbox.set_phase(SandboxPhase::Provisioning as i32); } @@ -1379,42 +1649,28 @@ impl ComputeRuntime { } async fn apply_deleted_locked(&self, sandbox_id: &str) -> Result<(), String> { - let sandbox = self - .store - .get_message::(sandbox_id) - .await - .map_err(|e| e.to_string())?; - if let Some(sandbox) = sandbox.as_ref() { - self.cleanup_sandbox_owned_records(sandbox).await; - } - - let _ = self - .store - .delete(Sandbox::object_type(), sandbox_id) + self.store + .delete_with_name_scoped( + Sandbox::object_type(), + sandbox_id, + SANDBOX_SETTINGS_OBJECT_TYPE, + ) .await .map_err(|e| e.to_string())?; - self.sandbox_index.remove_sandbox(sandbox_id); - self.sandbox_watch_bus.notify(sandbox_id); - self.cleanup_sandbox_state(sandbox_id); + self.cleanup_removed_sandbox_state(sandbox_id).await; Ok(()) } - async fn cleanup_sandbox_owned_records(&self, sandbox: &Sandbox) { - self.cleanup_sandbox_ssh_sessions(sandbox.object_id()).await; - - if let Err(e) = self - .store - .delete_by_name(SANDBOX_SETTINGS_OBJECT_TYPE, sandbox.object_name()) - .await - { - warn!( - sandbox_id = %sandbox.object_id(), - sandbox_name = %sandbox.object_name(), - error = %e, - "Failed to delete sandbox settings during cleanup" - ); - } - } + async fn apply_deleted_if_version_locked( + &self, + sandbox: &Sandbox, + expected_resource_version: u64, + ) -> Result<(), String> { + let sandbox_id = sandbox.object_id(); + self.remove_sandbox_record_if_version_locked(sandbox_id, expected_resource_version) + .await?; + Ok(()) + } async fn cleanup_sandbox_ssh_sessions(&self, sandbox_id: &str) { if let Ok(records) = self.store.list(SshSession::object_type(), 1000, 0).await { @@ -1437,6 +1693,26 @@ impl ComputeRuntime { } } + async fn cleanup_local_state_if_sandbox_absent(&self, sandbox_id: &str) -> Result<(), Status> { + let _guard = self.sync_lock.lock().await; + let record = self + .store + .get(Sandbox::object_type(), sandbox_id) + .await + .map_err(|err| Status::internal(format!("fetch sandbox failed: {err}")))?; + if record.is_none() { + self.cleanup_removed_sandbox_state(sandbox_id).await; + } + Ok(()) + } + + async fn cleanup_removed_sandbox_state(&self, sandbox_id: &str) { + self.cleanup_sandbox_ssh_sessions(sandbox_id).await; + self.sandbox_index.remove_sandbox(sandbox_id); + self.sandbox_watch_bus.notify(sandbox_id); + self.cleanup_sandbox_state(sandbox_id); + } + fn cleanup_sandbox_state(&self, sandbox_id: &str) { self.tracing_log_bus.remove(sandbox_id); self.tracing_log_bus.platform_event_bus.remove(sandbox_id); @@ -1448,6 +1724,30 @@ impl ComputeRuntime { snapshot: DriverSandbox, sweep_started_at_ms: i64, ) -> Result<(), String> { + let expected_resource_version = { + let _guard = self.sync_lock.lock().await; + let Some(existing) = self + .store + .get(Sandbox::object_type(), &snapshot.id) + .await + .map_err(|e| e.to_string())? + else { + return Ok(()); + }; + + if existing.updated_at_ms > sweep_started_at_ms { + return Ok(()); + } + existing.resource_version + }; + + let Some(current) = self + .get_driver_sandbox(&snapshot.id, &snapshot.name) + .await? + else { + return Ok(()); + }; + let _guard = self.sync_lock.lock().await; let Some(existing) = self .store @@ -1457,18 +1757,12 @@ impl ComputeRuntime { else { return Ok(()); }; - - if existing.updated_at_ms > sweep_started_at_ms { + if existing.resource_version != expected_resource_version + || existing.updated_at_ms > sweep_started_at_ms + { return Ok(()); } - let Some(current) = self - .get_driver_sandbox(&snapshot.id, &snapshot.name) - .await? - else { - return Ok(()); - }; - self.apply_sandbox_update_locked(current, Some(existing)) .await } @@ -1479,42 +1773,68 @@ impl ComputeRuntime { sweep_started_at_ms: i64, grace_ms: i64, ) -> Result<(), String> { + let (sandbox_id, sandbox_name, expected_resource_version, age_ms) = { + let _guard = self.sync_lock.lock().await; + let Some(current_record) = self + .store + .get(Sandbox::object_type(), &record.id) + .await + .map_err(|e| e.to_string())? + else { + return Ok(()); + }; + + if current_record.updated_at_ms > sweep_started_at_ms { + return Ok(()); + } + + let sandbox = decode_sandbox_record(¤t_record)?; + let age_ms = + openshell_core::time::now_ms().saturating_sub(current_record.created_at_ms); + if age_ms < grace_ms { + return Ok(()); + } + + ( + sandbox.object_id().to_string(), + sandbox.object_name().to_string(), + current_record.resource_version, + age_ms, + ) + }; + + let current = self.get_driver_sandbox(&sandbox_id, &sandbox_name).await?; + let _guard = self.sync_lock.lock().await; let Some(current_record) = self .store - .get(Sandbox::object_type(), &record.id) + .get(Sandbox::object_type(), &sandbox_id) .await .map_err(|e| e.to_string())? else { return Ok(()); }; - - if current_record.updated_at_ms > sweep_started_at_ms { - return Ok(()); - } - - let sandbox = decode_sandbox_record(¤t_record)?; - let age_ms = openshell_core::time::now_ms().saturating_sub(current_record.created_at_ms); - if age_ms < grace_ms { + if current_record.resource_version != expected_resource_version + || current_record.updated_at_ms > sweep_started_at_ms + { return Ok(()); } - if let Some(current) = self - .get_driver_sandbox(sandbox.object_id(), sandbox.object_name()) - .await? - { + if let Some(current) = current { return self .apply_sandbox_update_locked(current, Some(current_record)) .await; } + let sandbox = decode_sandbox_record(¤t_record)?; info!( - sandbox_id = %sandbox.object_id(), - sandbox_name = %sandbox.object_name(), + sandbox_id = %sandbox_id, + sandbox_name = %sandbox_name, age_secs = age_ms / 1000, "Removing sandbox from store after it disappeared from the compute driver snapshot" ); - self.apply_deleted_locked(sandbox.object_id()).await + self.apply_deleted_if_version_locked(&sandbox, expected_resource_version) + .await } async fn get_driver_sandbox( @@ -1530,7 +1850,18 @@ impl ComputeRuntime { })) .await { - Ok(response) => Ok(response.into_inner().sandbox), + Ok(response) => { + let sandbox = response.into_inner().sandbox; + if let Some(sandbox) = sandbox.as_ref() + && sandbox.id != sandbox_id + { + return Err(format!( + "compute driver returned sandbox '{}' for requested id '{sandbox_id}'", + sandbox.id + )); + } + Ok(sandbox) + } Err(status) if status.code() == Code::NotFound => Ok(None), Err(status) => Err(status.to_string()), } @@ -1861,6 +2192,13 @@ fn decode_sandbox_record(record: &ObjectRecord) -> Result { Sandbox::decode(record.payload.as_slice()).map_err(|e| e.to_string()) } +fn sandbox_resource_version(sandbox: &Sandbox) -> u64 { + sandbox + .metadata + .as_ref() + .map_or(0, |metadata| metadata.resource_version) +} + fn public_status_from_driver( status: &DriverSandboxStatus, phase: SandboxPhase, @@ -1881,6 +2219,73 @@ fn public_status_from_driver( } } +fn apply_driver_snapshot(sandbox: &mut Sandbox, incoming: &DriverSandbox, session_connected: bool) { + let old_phase = SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); + let mut phase = incoming + .status + .as_ref() + .map_or(old_phase, |status| derive_phase(Some(status))); + let sandbox_name = &incoming.name; + let supervisor_promoted = + session_connected && matches!(phase, SandboxPhase::Provisioning | SandboxPhase::Unknown); + if supervisor_promoted { + phase = SandboxPhase::Ready; + } + + let cpv = sandbox.current_policy_version(); + let mut status = incoming + .status + .as_ref() + .map(|status| public_status_from_driver(status, phase, cpv)) + .or_else(|| sandbox.status.clone()); + rewrite_user_facing_conditions(&mut status, sandbox.spec.as_ref()); + if supervisor_promoted { + ensure_supervisor_ready_status(&mut status, sandbox_name); + } + + if let Some(status) = status.as_mut() + && status.sandbox_name.is_empty() + { + status.sandbox_name.clone_from(sandbox_name); + } + + if old_phase != phase { + info!( + sandbox_id = %incoming.id, + sandbox_name = %sandbox_name, + old_phase = ?old_phase, + new_phase = ?phase, + "Sandbox phase changed" + ); + } + + if phase == SandboxPhase::Error + && let Some(ref status) = status + { + for condition in &status.conditions { + if condition.r#type == "Ready" + && condition.status.eq_ignore_ascii_case("false") + && is_terminal_failure_reason(&condition.reason) + { + warn!( + sandbox_id = %incoming.id, + sandbox_name = %sandbox_name, + reason = %condition.reason, + message = %condition.message, + "Sandbox failed to become ready" + ); + } + } + } + + if let Some(metadata) = sandbox.metadata.as_mut() { + metadata.name.clone_from(sandbox_name); + } + sandbox.status = status; + sandbox.set_phase(phase as i32); + sandbox.set_current_policy_version(cpv); +} + fn ensure_supervisor_ready_status(status: &mut Option, sandbox_name: &str) { upsert_ready_condition( status, @@ -2140,6 +2545,7 @@ pub async fn new_test_runtime(store: Arc) -> ComputeRuntime { tracing_log_bus: TracingLogBus::new(), supervisor_sessions: Arc::new(SupervisorSessionRegistry::new()), sync_lock: Arc::new(Mutex::new(())), + delete_gates: Arc::new(DeleteGateRegistry::default()), gateway_bind_addresses: Vec::new(), replica_id: "test-replica".to_string(), } @@ -2152,10 +2558,13 @@ mod tests { use openshell_core::proto::compute::v1::{ CreateSandboxResponse, DeleteSandboxResponse, GetCapabilitiesResponse, GetSandboxRequest, GetSandboxResponse, StopSandboxRequest, StopSandboxResponse, ValidateSandboxCreateResponse, + WatchSandboxesDeletedEvent, WatchSandboxesSandboxEvent, }; use std::collections::HashMap; - use std::sync::Arc; - use tokio::sync::{mpsc, oneshot}; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::sync::{Arc, Mutex as TestMutex}; + use tokio::sync::{Notify, Semaphore, mpsc, oneshot}; + use tokio_stream::wrappers::UnboundedReceiverStream; fn string_value(value: &str) -> prost_types::Value { prost_types::Value { @@ -2373,6 +2782,212 @@ mod tests { } } + #[derive(Clone)] + enum ControlledDeleteOutcome { + Ok(bool), + Error(&'static str), + } + + #[derive(Clone)] + enum ControlledGetOutcome { + Sandbox(Box), + Missing, + Error(&'static str), + } + + struct ControlledDriver { + watch_tx: mpsc::UnboundedSender>, + watch_rx: TestMutex>>>, + watch_started: Notify, + delete_started: Notify, + delete_release: Semaphore, + delete_blocked: AtomicBool, + delete_calls: AtomicUsize, + delete_outcome: TestMutex, + get_started: Notify, + get_release: Semaphore, + get_blocked: AtomicBool, + get_outcome: TestMutex, + } + + impl ControlledDriver { + fn new() -> Arc { + let (watch_tx, watch_rx) = mpsc::unbounded_channel(); + Arc::new(Self { + watch_tx, + watch_rx: TestMutex::new(Some(watch_rx)), + watch_started: Notify::new(), + delete_started: Notify::new(), + delete_release: Semaphore::new(0), + delete_blocked: AtomicBool::new(false), + delete_calls: AtomicUsize::new(0), + delete_outcome: TestMutex::new(ControlledDeleteOutcome::Ok(true)), + get_started: Notify::new(), + get_release: Semaphore::new(0), + get_blocked: AtomicBool::new(false), + get_outcome: TestMutex::new(ControlledGetOutcome::Missing), + }) + } + + fn block_delete(&self) { + self.delete_blocked.store(true, Ordering::SeqCst); + } + + fn release_delete(&self) { + self.delete_release.add_permits(1); + } + + fn block_get(&self) { + self.get_blocked.store(true, Ordering::SeqCst); + } + + fn release_get(&self) { + self.get_release.add_permits(1); + } + + fn set_delete_outcome(&self, outcome: ControlledDeleteOutcome) { + *self + .delete_outcome + .lock() + .expect("delete outcome lock poisoned") = outcome; + } + + fn set_get_outcome(&self, outcome: ControlledGetOutcome) { + *self.get_outcome.lock().expect("get outcome lock poisoned") = outcome; + } + + fn delete_calls(&self) -> usize { + self.delete_calls.load(Ordering::SeqCst) + } + + fn send_event(&self, event: WatchSandboxesEvent) { + self.watch_tx + .send(Ok(event)) + .expect("watch loop should still be receiving events"); + } + } + + #[tonic::async_trait] + impl ComputeDriver for ControlledDriver { + type WatchSandboxesStream = DriverWatchStream; + + async fn get_capabilities( + &self, + _request: Request, + ) -> Result, Status> { + Ok(tonic::Response::new(GetCapabilitiesResponse { + driver_name: "controlled-test-driver".to_string(), + driver_version: "test".to_string(), + default_image: "openshell/sandbox:test".to_string(), + })) + } + + async fn validate_sandbox_create( + &self, + _request: Request, + ) -> Result, Status> { + Ok(tonic::Response::new(ValidateSandboxCreateResponse {})) + } + + async fn get_sandbox( + &self, + _request: Request, + ) -> Result, Status> { + self.get_started.notify_one(); + if self.get_blocked.load(Ordering::SeqCst) { + self.get_release + .acquire() + .await + .expect("get release semaphore closed") + .forget(); + } + let outcome = self + .get_outcome + .lock() + .expect("get outcome lock poisoned") + .clone(); + match outcome { + ControlledGetOutcome::Sandbox(sandbox) => { + Ok(tonic::Response::new(GetSandboxResponse { + sandbox: Some(*sandbox), + })) + } + ControlledGetOutcome::Missing => Err(Status::not_found("sandbox not found")), + ControlledGetOutcome::Error(message) => Err(Status::internal(message)), + } + } + + async fn list_sandboxes( + &self, + _request: Request, + ) -> Result< + tonic::Response, + Status, + > { + Ok(tonic::Response::new( + openshell_core::proto::compute::v1::ListSandboxesResponse { + sandboxes: Vec::new(), + }, + )) + } + + async fn create_sandbox( + &self, + _request: Request, + ) -> Result, Status> { + Ok(tonic::Response::new(CreateSandboxResponse {})) + } + + async fn stop_sandbox( + &self, + _request: Request, + ) -> Result, Status> { + Ok(tonic::Response::new(StopSandboxResponse {})) + } + + async fn delete_sandbox( + &self, + _request: Request, + ) -> Result, Status> { + self.delete_calls.fetch_add(1, Ordering::SeqCst); + self.delete_started.notify_one(); + if self.delete_blocked.load(Ordering::SeqCst) { + self.delete_release + .acquire() + .await + .expect("delete release semaphore closed") + .forget(); + } + let outcome = self + .delete_outcome + .lock() + .expect("delete outcome lock poisoned") + .clone(); + match outcome { + ControlledDeleteOutcome::Ok(deleted) => { + Ok(tonic::Response::new(DeleteSandboxResponse { deleted })) + } + ControlledDeleteOutcome::Error(message) => Err(Status::internal(message)), + } + } + + async fn watch_sandboxes( + &self, + _request: Request, + ) -> Result, Status> { + let receiver = self + .watch_rx + .lock() + .expect("watch receiver lock poisoned") + .take() + .ok_or_else(|| Status::failed_precondition("watch already started"))?; + self.watch_started.notify_one(); + Ok(tonic::Response::new(Box::pin( + UnboundedReceiverStream::new(receiver), + ))) + } + } + async fn test_runtime(driver: SharedComputeDriver) -> ComputeRuntime { test_runtime_with_resume(driver, None).await } @@ -2399,6 +3014,7 @@ mod tests { tracing_log_bus: TracingLogBus::new(), supervisor_sessions: Arc::new(SupervisorSessionRegistry::new()), sync_lock: Arc::new(Mutex::new(())), + delete_gates: Arc::new(DeleteGateRegistry::default()), gateway_bind_addresses: Vec::new(), replica_id: "test-replica".to_string(), } @@ -2448,6 +3064,49 @@ mod tests { } } + async fn seed_sandbox_owned_records(runtime: &ComputeRuntime, sandbox: &Sandbox) -> SshSession { + runtime + .store + .put( + SANDBOX_SETTINGS_OBJECT_TYPE, + &format!("settings-{}", sandbox.object_id()), + sandbox.object_name(), + br#"{"revision":1,"settings":{}}"#, + None, + ) + .await + .unwrap(); + let session = ssh_session_record("owned", sandbox.object_id()); + runtime.store.put_message(&session).await.unwrap(); + session + } + + async fn assert_sandbox_owned_records( + runtime: &ComputeRuntime, + sandbox: &Sandbox, + session: &SshSession, + expected: bool, + ) { + assert_eq!( + runtime + .store + .get_by_name(SANDBOX_SETTINGS_OBJECT_TYPE, sandbox.object_name()) + .await + .unwrap() + .is_some(), + expected + ); + assert_eq!( + runtime + .store + .get_message::(session.object_id()) + .await + .unwrap() + .is_some(), + expected + ); + } + fn make_driver_condition(reason: &str, message: &str) -> DriverCondition { DriverCondition { r#type: "Ready".to_string(), @@ -2469,12 +3128,90 @@ mod tests { } } + fn ready_driver_sandbox(id: &str, name: &str) -> DriverSandbox { + DriverSandbox { + id: id.to_string(), + name: name.to_string(), + namespace: "default".to_string(), + spec: None, + status: Some(DriverSandboxStatus { + sandbox_name: name.to_string(), + instance_id: format!("{name}-pod"), + agent_fd: String::new(), + sandbox_fd: String::new(), + conditions: vec![DriverCondition { + r#type: "Ready".to_string(), + status: "True".to_string(), + reason: "DependenciesReady".to_string(), + message: "Sandbox is ready".to_string(), + last_transition_time: String::new(), + }], + deleting: false, + }), + } + } + + fn sandbox_watch_event(sandbox: DriverSandbox) -> WatchSandboxesEvent { + WatchSandboxesEvent { + payload: Some(watch_sandboxes_event::Payload::Sandbox( + WatchSandboxesSandboxEvent { + sandbox: Some(sandbox), + }, + )), + } + } + + fn deleted_watch_event(sandbox_id: &str) -> WatchSandboxesEvent { + WatchSandboxesEvent { + payload: Some(watch_sandboxes_event::Payload::Deleted( + WatchSandboxesDeletedEvent { + sandbox_id: sandbox_id.to_string(), + }, + )), + } + } + + async fn start_watch_loop( + runtime: &ComputeRuntime, + driver: &ControlledDriver, + ) -> (watch::Sender, tokio::task::JoinHandle<()>) { + let (shutdown_tx, shutdown_rx) = watch::channel(false); + let runtime = Arc::new(runtime.clone()); + let handle = tokio::spawn(async move { runtime.watch_loop(shutdown_rx).await }); + tokio::time::timeout(Duration::from_secs(1), driver.watch_started.notified()) + .await + .expect("watch loop did not start"); + (shutdown_tx, handle) + } + + async fn stop_watch_loop( + shutdown_tx: watch::Sender, + handle: tokio::task::JoinHandle<()>, + ) { + shutdown_tx.send(true).unwrap(); + tokio::time::timeout(Duration::from_secs(1), handle) + .await + .expect("watch loop did not stop") + .unwrap(); + } + #[tokio::test] async fn sqlite_store_is_single_replica() { let store = Arc::new(Store::connect("sqlite::memory:").await.unwrap()); assert!(store.is_single_replica()); } + #[test] + fn delete_gate_registry_removes_stale_entries() { + let registry = DeleteGateRegistry::default(); + let first = registry.gate_for("sb-1"); + assert_eq!(registry.entry_count(), 1); + drop(first); + + let _second = registry.gate_for("sb-2"); + assert_eq!(registry.entry_count(), 1); + } + #[test] fn terminal_failure_treats_unknown_reasons_as_terminal() { let terminal_cases = [ @@ -2776,7 +3513,7 @@ mod tests { } #[tokio::test] - async fn set_sandbox_phase_deleting_retries_after_stale_snapshot_conflict() { + async fn begin_sandbox_delete_retries_after_stale_snapshot_conflict() { let runtime = test_runtime(Arc::new(TestDriver::default())).await; let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); runtime.store.put_message(&sandbox).await.unwrap(); @@ -2796,10 +3533,15 @@ mod tests { .await .unwrap(); - let updated = runtime - .set_sandbox_phase_deleting_with_initial_snapshot("sb-1", Some(stale_snapshot)) + let transition = runtime + .begin_sandbox_delete_with_initial_snapshot("sb-1", Some(stale_snapshot)) .await .unwrap(); + let BeginDelete::Started(transition) = transition else { + panic!("expected a new delete transition"); + }; + let transition = *transition; + let updated = transition.deleting; assert_eq!( SandboxPhase::try_from(updated.phase()).unwrap(), @@ -2813,6 +3555,7 @@ mod tests { .map_or(0, |metadata| metadata.resource_version), 3 ); + assert_eq!(transition.previous.current_policy_version(), 7); let stored = runtime .store @@ -2828,10 +3571,17 @@ mod tests { } #[tokio::test] - async fn apply_sandbox_update_allows_delete_failures_to_recover() { + async fn apply_sandbox_update_is_noop_while_deleting() { let runtime = test_runtime(Arc::new(TestDriver::default())).await; let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Deleting); runtime.store.put_message(&sandbox).await.unwrap(); + let before = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); runtime .apply_sandbox_update(DriverSandbox { @@ -2865,7 +3615,960 @@ mod tests { .unwrap(); assert_eq!( SandboxPhase::try_from(stored.phase()).unwrap(), - SandboxPhase::Ready + SandboxPhase::Deleting + ); + assert_eq!( + sandbox_resource_version(&stored), + sandbox_resource_version(&before) + ); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + } + + #[tokio::test] + async fn supervisor_session_change_is_noop_while_deleting() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Deleting); + runtime.store.put_message(&sandbox).await.unwrap(); + let before = runtime + .store + .get(Sandbox::object_type(), "sb-1") + .await + .unwrap() + .unwrap(); + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); + + runtime + .supervisor_session_disconnected("sb-1") + .await + .unwrap(); + + let after = runtime + .store + .get(Sandbox::object_type(), "sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!(after.resource_version, before.resource_version); + assert_eq!(after.payload, before.payload); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + } + + #[tokio::test] + async fn unknown_watch_snapshot_does_not_create_gateway_state() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-unknown"); + + runtime + .apply_sandbox_update(ready_driver_sandbox("sb-unknown", "unmanaged")) + .await + .unwrap(); + + assert!( + runtime + .store + .get_message::("sb-unknown") + .await + .unwrap() + .is_none() + ); + assert!( + runtime + .sandbox_index + .sandbox_id_for_sandbox_name("unmanaged") + .is_none() + ); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + } + + #[tokio::test] + async fn watcher_update_preserves_complete_api_created_spec() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + sandbox.spec = Some(SandboxSpec { + log_level: "debug".to_string(), + template: Some(SandboxTemplate { + image: "example.test/sandbox:complete".to_string(), + ..Default::default() + }), + ..Default::default() + }); + + runtime.create_sandbox(sandbox, None).await.unwrap(); + runtime + .apply_sandbox_update(ready_driver_sandbox("sb-1", "sandbox-a")) + .await + .unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + stored + .spec + .as_ref() + .and_then(|spec| spec.template.as_ref()) + .map(|template| template.image.as_str()), + Some("example.test/sandbox:complete") + ); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + } + + #[tokio::test] + async fn blocked_delete_does_not_delay_unrelated_deleted_watch_event() { + let driver = ControlledDriver::new(); + driver.block_delete(); + let runtime = test_runtime(driver.clone()).await; + + for sandbox in [ + sandbox_record("sb-a", "sandbox-a", SandboxPhase::Ready), + sandbox_record("sb-b", "sandbox-b", SandboxPhase::Ready), + ] { + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + } + + let (shutdown_tx, watch_handle) = start_watch_loop(&runtime, &driver).await; + let mut sandbox_b_rx = runtime.sandbox_watch_bus.subscribe("sb-b"); + let delete_runtime = runtime.clone(); + let delete_handle = + tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + + let before = runtime + .store + .get(Sandbox::object_type(), "sb-a") + .await + .unwrap() + .unwrap(); + let mut stale_snapshot = ready_driver_sandbox("sb-a", "sandbox-a"); + stale_snapshot.status = Some(make_driver_status(make_driver_condition( + "ContainerExited", + "container exited during deletion", + ))); + driver.send_event(sandbox_watch_event(stale_snapshot)); + driver.send_event(deleted_watch_event("sb-b")); + + tokio::time::timeout(Duration::from_secs(1), sandbox_b_rx.recv()) + .await + .expect("sandbox B event was blocked by sandbox A deletion") + .expect("sandbox B watch bus closed before notification"); + assert!( + runtime + .store + .get_message::("sb-b") + .await + .unwrap() + .is_none() + ); + let after = runtime + .store + .get(Sandbox::object_type(), "sb-a") + .await + .unwrap() + .unwrap(); + assert_eq!(after.resource_version, before.resource_version); + assert_eq!(after.payload, before.payload); + + driver.release_delete(); + assert!( + tokio::time::timeout(Duration::from_secs(1), delete_handle) + .await + .expect("delete did not finish") + .unwrap() + .unwrap() + ); + stop_watch_loop(shutdown_tx, watch_handle).await; + } + + #[tokio::test] + async fn concurrent_duplicate_deletes_call_driver_once() { + let driver = ControlledDriver::new(); + driver.block_delete(); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let first_runtime = runtime.clone(); + let first = tokio::spawn(async move { first_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("first delete did not reach the driver"); + + let second_runtime = runtime.clone(); + let second = tokio::spawn(async move { second_runtime.delete_sandbox("sandbox-a").await }); + driver.release_delete(); + + assert!( + tokio::time::timeout(Duration::from_secs(1), first) + .await + .unwrap() + .unwrap() + .unwrap() + ); + assert!( + tokio::time::timeout(Duration::from_secs(1), second) + .await + .unwrap() + .unwrap() + .unwrap() + ); + assert_eq!(driver.delete_calls(), 1); + } + + #[tokio::test] + async fn waiting_delete_does_not_retarget_a_reused_name() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + let original = sandbox_record("sb-original", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&original).await.unwrap(); + + // Hold the original ID's gate so the request resolves the name and + // then waits before it can revalidate the durable row. + let delete_gate = runtime.delete_gates.gate_for(original.object_id()); + let delete_guard = delete_gate.lock().await; + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), async { + while Arc::strong_count(&delete_gate) < 2 { + tokio::task::yield_now().await; + } + }) + .await + .expect("delete did not start waiting on the original ID gate"); + + runtime + .store + .delete(Sandbox::object_type(), original.object_id()) + .await + .unwrap(); + let replacement = sandbox_record("sb-replacement", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&replacement).await.unwrap(); + drop(delete_guard); + + assert!(delete.await.unwrap().unwrap()); + assert_eq!(driver.delete_calls(), 0); + assert!( + runtime + .store + .get_message::(replacement.object_id()) + .await + .unwrap() + .is_some() + ); + } + + #[tokio::test] + async fn request_cancellation_does_not_cancel_the_delete_worker() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Ok(false)); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_runtime = runtime.clone(); + let request = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + + request.abort(); + assert!(request.await.unwrap_err().is_cancelled()); + driver.release_delete(); + + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("detached delete worker did not finish cleanup"); + assert_eq!(driver.delete_calls(), 1); + } + + #[tokio::test] + async fn already_absent_driver_resource_is_removed_synchronously() { + let driver = ControlledDriver::new(); + driver.set_delete_outcome(ControlledDeleteOutcome::Ok(false)); + let runtime = test_runtime(driver).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + + assert!(!runtime.delete_sandbox("sandbox-a").await.unwrap()); + assert!( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + assert!( + runtime + .sandbox_index + .sandbox_id_for_sandbox_name("sandbox-a") + .is_none() + ); + + runtime + .apply_sandbox_update(ready_driver_sandbox("sb-1", "sandbox-a")) + .await + .unwrap(); + assert!( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + } + + #[tokio::test] + async fn already_absent_delete_retries_cleanup_after_a_deleting_version_change() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Ok(false)); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + runtime + .store + .update_message_cas::("sb-1", 0, |sandbox| { + sandbox.set_current_policy_version(9); + }) + .await + .unwrap(); + + driver.release_delete(); + assert!(!delete.await.unwrap().unwrap()); + assert!( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn already_absent_delete_reports_incomplete_gateway_cleanup() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Ok(false)); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + runtime + .store + .update_message_cas::("sb-1", 0, |sandbox| { + sandbox.set_phase(SandboxPhase::Ready as i32); + }) + .await + .unwrap(); + + driver.release_delete(); + let error = delete.await.unwrap().unwrap_err(); + assert_eq!(error.code(), Code::Internal); + assert_eq!( + SandboxPhase::try_from( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap() + .phase() + ) + .unwrap(), + SandboxPhase::Ready + ); + } + + #[tokio::test] + async fn accepted_driver_delete_leaves_removal_to_watcher() { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); + + assert!(runtime.delete_sandbox("sandbox-a").await.unwrap()); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Deleting + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, true).await; + assert!(watch_rx.try_recv().is_ok()); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + + runtime.apply_deleted("sb-1").await.unwrap(); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + assert!(watch_rx.try_recv().is_ok()); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Closed) + )); + } + + #[tokio::test] + async fn accepted_delete_cleans_local_state_after_another_replica_removes_row() { + let driver = ControlledDriver::new(); + driver.block_delete(); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + tokio::time::timeout(Duration::from_secs(1), watch_rx.recv()) + .await + .expect("deleting notification was not sent") + .expect("watch bus closed before the deleting notification"); + + assert!( + runtime + .store + .delete_with_name_scoped( + Sandbox::object_type(), + "sb-1", + SANDBOX_SETTINGS_OBJECT_TYPE, + ) + .await + .unwrap() + ); + driver.release_delete(); + + assert!( + tokio::time::timeout(Duration::from_secs(1), delete) + .await + .expect("delete did not finish") + .unwrap() + .unwrap() + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + assert!( + runtime + .sandbox_index + .sandbox_id_for_sandbox_name("sandbox-a") + .is_none() + ); + assert!(watch_rx.try_recv().is_ok()); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Closed) + )); + } + + #[tokio::test] + async fn absent_delete_cleans_local_state_after_another_replica_removes_row() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Ok(false)); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + tokio::time::timeout(Duration::from_secs(1), watch_rx.recv()) + .await + .expect("deleting notification was not sent") + .expect("watch bus closed before the deleting notification"); + + assert!( + runtime + .store + .delete_with_name_scoped( + Sandbox::object_type(), + "sb-1", + SANDBOX_SETTINGS_OBJECT_TYPE, + ) + .await + .unwrap() + ); + driver.release_delete(); + + assert!( + !tokio::time::timeout(Duration::from_secs(1), delete) + .await + .expect("delete did not finish") + .unwrap() + .unwrap() + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + assert!( + runtime + .sandbox_index + .sandbox_id_for_sandbox_name("sandbox-a") + .is_none() + ); + assert!(watch_rx.try_recv().is_ok()); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Closed) + )); + } + + #[tokio::test] + async fn delete_error_with_absent_backend_removes_gateway_row() { + let driver = ControlledDriver::new(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + let runtime = test_runtime(driver).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + + runtime.delete_sandbox("sandbox-a").await.unwrap_err(); + + assert!( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + } + + #[tokio::test] + async fn delete_error_cleans_local_state_after_another_replica_removes_row() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + tokio::time::timeout(Duration::from_secs(1), watch_rx.recv()) + .await + .expect("deleting notification was not sent") + .expect("watch bus closed before the deleting notification"); + + assert!( + runtime + .store + .delete_with_name_scoped( + Sandbox::object_type(), + "sb-1", + SANDBOX_SETTINGS_OBJECT_TYPE, + ) + .await + .unwrap() + ); + driver.release_delete(); + + tokio::time::timeout(Duration::from_secs(1), delete) + .await + .expect("delete did not finish") + .unwrap() + .unwrap_err(); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + assert!( + runtime + .sandbox_index + .sandbox_id_for_sandbox_name("sandbox-a") + .is_none() + ); + assert!(watch_rx.try_recv().is_ok()); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Closed) + )); + } + + async fn assert_recovery_cleans_local_state_after_row_removed_during_lookup( + get_outcome: ControlledGetOutcome, + ) { + let driver = ControlledDriver::new(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + driver.set_get_outcome(get_outcome); + driver.block_get(); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + let mut watch_rx = runtime.sandbox_watch_bus.subscribe("sb-1"); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.get_started.notified()) + .await + .expect("delete recovery did not reach the driver lookup"); + tokio::time::timeout(Duration::from_secs(1), watch_rx.recv()) + .await + .expect("deleting notification was not sent") + .expect("watch bus closed before the deleting notification"); + + assert!( + runtime + .store + .delete_with_name_scoped( + Sandbox::object_type(), + "sb-1", + SANDBOX_SETTINGS_OBJECT_TYPE, + ) + .await + .unwrap() + ); + driver.release_get(); + + tokio::time::timeout(Duration::from_secs(1), delete) + .await + .expect("delete did not finish") + .unwrap() + .unwrap_err(); + assert_sandbox_owned_records(&runtime, &sandbox, &session, false).await; + assert!( + runtime + .sandbox_index + .sandbox_id_for_sandbox_name("sandbox-a") + .is_none() + ); + assert!(watch_rx.try_recv().is_ok()); + assert!(matches!( + watch_rx.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Closed) + )); + } + + #[tokio::test] + async fn snapshot_recovery_cleans_local_state_after_another_replica_removes_row() { + assert_recovery_cleans_local_state_after_row_removed_during_lookup( + ControlledGetOutcome::Sandbox(Box::new(ready_driver_sandbox("sb-1", "sandbox-a"))), + ) + .await; + } + + #[tokio::test] + async fn rollback_recovery_cleans_local_state_after_another_replica_removes_row() { + assert_recovery_cleans_local_state_after_row_removed_during_lookup( + ControlledGetOutcome::Error("lookup failed"), + ) + .await; + } + + #[tokio::test] + async fn delete_error_recovers_from_current_driver_snapshot() { + let driver = ControlledDriver::new(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new( + ready_driver_sandbox("sb-1", "sandbox-a"), + ))); + let runtime = test_runtime(driver).await; + let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + sandbox.spec = Some(SandboxSpec { + log_level: "debug".to_string(), + ..Default::default() + }); + runtime.store.put_message(&sandbox).await.unwrap(); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + + let error = runtime.delete_sandbox("sandbox-a").await.unwrap_err(); + assert_eq!(error.code(), Code::Internal); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, true).await; + assert_eq!( + stored.spec.as_ref().map(|spec| spec.log_level.as_str()), + Some("debug") + ); + } + + #[tokio::test] + async fn delete_error_rolls_back_when_driver_lookup_fails() { + let driver = ControlledDriver::new(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + driver.set_get_outcome(ControlledGetOutcome::Error("lookup failed")); + let runtime = test_runtime(driver).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + let session = seed_sandbox_owned_records(&runtime, &sandbox).await; + + runtime.delete_sandbox("sandbox-a").await.unwrap_err(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + assert_sandbox_owned_records(&runtime, &sandbox, &session, true).await; + } + + #[tokio::test] + async fn delete_error_recovery_does_not_overwrite_concurrent_cas_update() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new( + ready_driver_sandbox("sb-1", "sandbox-a"), + ))); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + runtime + .store + .update_message_cas::("sb-1", 0, |sandbox| { + sandbox.set_current_policy_version(9); + }) + .await + .unwrap(); + + driver.release_delete(); + tokio::time::timeout(Duration::from_secs(1), delete) + .await + .unwrap() + .unwrap() + .unwrap_err(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Deleting + ); + assert_eq!(stored.current_policy_version(), 9); + } + + #[tokio::test] + async fn driver_completion_tolerates_watcher_removing_row_in_flight() { + let driver = ControlledDriver::new(); + driver.block_delete(); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + runtime.apply_deleted("sb-1").await.unwrap(); + + driver.release_delete(); + assert!( + tokio::time::timeout(Duration::from_secs(1), delete) + .await + .unwrap() + .unwrap() + .unwrap() + ); + assert!( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn delete_error_does_not_resurrect_row_removed_by_watcher() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new( + ready_driver_sandbox("sb-1", "sandbox-a"), + ))); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_runtime = runtime.clone(); + let delete = tokio::spawn(async move { delete_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("delete did not reach the driver"); + runtime.apply_deleted("sb-1").await.unwrap(); + + driver.release_delete(); + tokio::time::timeout(Duration::from_secs(1), delete) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert!( + runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn blocked_reconciliation_lookup_does_not_delay_watch_events() { + let driver = ControlledDriver::new(); + driver.block_get(); + let snapshot = ready_driver_sandbox("sb-a", "sandbox-a"); + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(snapshot.clone()))); + let runtime = test_runtime(driver.clone()).await; + for sandbox in [ + sandbox_record("sb-a", "sandbox-a", SandboxPhase::Provisioning), + sandbox_record("sb-b", "sandbox-b", SandboxPhase::Ready), + ] { + runtime.store.put_message(&sandbox).await.unwrap(); + } + + let (shutdown_tx, watch_handle) = start_watch_loop(&runtime, &driver).await; + let mut sandbox_b_rx = runtime.sandbox_watch_bus.subscribe("sb-b"); + let reconcile_runtime = runtime.clone(); + let reconcile = tokio::spawn(async move { + reconcile_runtime + .reconcile_snapshot_sandbox(snapshot, i64::MAX) + .await + }); + tokio::time::timeout(Duration::from_secs(1), driver.get_started.notified()) + .await + .expect("reconciliation did not reach the driver lookup"); + + driver.send_event(deleted_watch_event("sb-b")); + tokio::time::timeout(Duration::from_secs(1), sandbox_b_rx.recv()) + .await + .expect("watch event was blocked by reconciliation lookup") + .expect("sandbox B watch bus closed before notification"); + assert!( + runtime + .store + .get_message::("sb-b") + .await + .unwrap() + .is_none() + ); + + driver.release_get(); + tokio::time::timeout(Duration::from_secs(1), reconcile) + .await + .unwrap() + .unwrap() + .unwrap(); + stop_watch_loop(shutdown_tx, watch_handle).await; + } + + #[tokio::test] + async fn non_deleting_container_exit_still_transitions_to_error() { + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + let mut exited = ready_driver_sandbox("sb-1", "sandbox-a"); + exited.status = Some(make_driver_status(make_driver_condition( + "ContainerExited", + "container exited unexpectedly", + ))); + + runtime.apply_sandbox_update(exited).await.unwrap(); + + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Error ); } diff --git a/crates/openshell-server/src/persistence/mod.rs b/crates/openshell-server/src/persistence/mod.rs index 8b6172c1c6..7dad899f3b 100644 --- a/crates/openshell-server/src/persistence/mod.rs +++ b/crates/openshell-server/src/persistence/mod.rs @@ -261,6 +261,46 @@ impl Store { store_dispatch!(self.delete_if(object_type, id, expected_resource_version)) } + /// Atomically delete an object and the object of `scoped_type` that has + /// the same name, subject to an exact parent resource-version match. + /// + /// The parent name is captured by the same database statement that + /// removes the records. This prevents a concurrent name reuse from + /// causing the newly-created name-scoped object to be removed. + /// + /// # Returns + /// * `Ok(true)` - The parent and any matching name-scoped object were deleted + /// * `Ok(false)` - The parent was not found + /// * `Err(Conflict)` - The parent exists with a different resource version + pub async fn delete_with_name_scoped_if_version( + &self, + parent_type: &str, + parent_id: &str, + expected_resource_version: u64, + scoped_type: &str, + ) -> PersistenceResult { + store_dispatch!(self.delete_with_name_scoped_if_version( + parent_type, + parent_id, + expected_resource_version, + scoped_type + )) + } + + /// Atomically delete an object and the object of `scoped_type` that has + /// the same name. + /// + /// The parent name is captured by the same database statement that + /// removes the records, preventing cleanup from racing with name reuse. + pub async fn delete_with_name_scoped( + &self, + parent_type: &str, + parent_id: &str, + scoped_type: &str, + ) -> PersistenceResult { + store_dispatch!(self.delete_with_name_scoped(parent_type, parent_id, scoped_type)) + } + /// Insert or update a generic named object with an application-owned scope. pub async fn put_scoped( &self, diff --git a/crates/openshell-server/src/persistence/postgres.rs b/crates/openshell-server/src/persistence/postgres.rs index 12299805e9..18960d54c6 100644 --- a/crates/openshell-server/src/persistence/postgres.rs +++ b/crates/openshell-server/src/persistence/postgres.rs @@ -244,6 +244,93 @@ WHERE object_type = $1 AND id = $2 AND resource_version = $3 } } + pub async fn delete_with_name_scoped_if_version( + &self, + parent_type: &str, + parent_id: &str, + expected_resource_version: u64, + scoped_type: &str, + ) -> PersistenceResult { + // Feed the sibling delete exclusively from the exact parent delete's + // RETURNING output. A stale resource version therefore cannot remove + // settings, and both deletes commit as one statement. + let row = sqlx::query( + r" +WITH deleted_parent AS ( + DELETE FROM objects + WHERE object_type = $1 AND id = $2 AND resource_version = $3 + RETURNING name +), +deleted_scoped AS ( + DELETE FROM objects + WHERE object_type = $4 + AND name IN (SELECT name FROM deleted_parent) + AND NOT (object_type = $1 AND id = $2) + RETURNING id +) +SELECT + EXISTS (SELECT 1 FROM deleted_parent) AS parent_deleted, + (SELECT COUNT(*) FROM deleted_scoped) AS scoped_deleted +", + ) + .bind(parent_type) + .bind(parent_id) + .bind(i64::try_from(expected_resource_version).unwrap_or(i64::MAX)) + .bind(scoped_type) + .fetch_one(&self.pool) + .await + .map_err(|e| map_db_error(&e))?; + + let parent_deleted: bool = row.get("parent_deleted"); + if parent_deleted { + Ok(true) + } else { + let existing = self.get(parent_type, parent_id).await?; + if let Some(record) = existing { + Err(PersistenceError::Conflict { + current_resource_version: Some(record.resource_version), + }) + } else { + Ok(false) + } + } + } + + pub async fn delete_with_name_scoped( + &self, + parent_type: &str, + parent_id: &str, + scoped_type: &str, + ) -> PersistenceResult { + let row = sqlx::query( + r" +WITH deleted_parent AS ( + DELETE FROM objects + WHERE object_type = $1 AND id = $2 + RETURNING name +), +deleted_scoped AS ( + DELETE FROM objects + WHERE object_type = $3 + AND name IN (SELECT name FROM deleted_parent) + AND NOT (object_type = $1 AND id = $2) + RETURNING id +) +SELECT + EXISTS (SELECT 1 FROM deleted_parent) AS parent_deleted, + (SELECT COUNT(*) FROM deleted_scoped) AS scoped_deleted +", + ) + .bind(parent_type) + .bind(parent_id) + .bind(scoped_type) + .fetch_one(&self.pool) + .await + .map_err(|e| map_db_error(&e))?; + + Ok(row.get("parent_deleted")) + } + pub async fn put_scoped( &self, object_type: &str, diff --git a/crates/openshell-server/src/persistence/sqlite.rs b/crates/openshell-server/src/persistence/sqlite.rs index 57d5c20e31..b0a60ede64 100644 --- a/crates/openshell-server/src/persistence/sqlite.rs +++ b/crates/openshell-server/src/persistence/sqlite.rs @@ -265,6 +265,88 @@ WHERE "object_type" = ?1 AND "id" = ?2 AND "resource_version" = ?3 } } + pub async fn delete_with_name_scoped_if_version( + &self, + parent_type: &str, + parent_id: &str, + expected_resource_version: u64, + scoped_type: &str, + ) -> PersistenceResult { + // SQLite serializes writers. Materializing the matching parent inside + // this DELETE captures its name before either row is removed, while + // keeping both removals in the same write statement. + let result = sqlx::query( + r#" +WITH "parent"("name") AS MATERIALIZED ( + SELECT "name" + FROM "objects" + WHERE "object_type" = ?1 AND "id" = ?2 AND "resource_version" = ?3 +) +DELETE FROM "objects" +WHERE ( + "object_type" = ?1 AND "id" = ?2 AND "resource_version" = ?3 +) +OR ( + "object_type" = ?4 + AND "name" IN (SELECT "name" FROM "parent") +) +"#, + ) + .bind(parent_type) + .bind(parent_id) + .bind(i64::try_from(expected_resource_version).unwrap_or(i64::MAX)) + .bind(scoped_type) + .execute(&self.pool) + .await + .map_err(|e| map_db_error(&e))?; + + if result.rows_affected() > 0 { + Ok(true) + } else { + let existing = self.get(parent_type, parent_id).await?; + if let Some(record) = existing { + Err(PersistenceError::Conflict { + current_resource_version: Some(record.resource_version), + }) + } else { + Ok(false) + } + } + } + + pub async fn delete_with_name_scoped( + &self, + parent_type: &str, + parent_id: &str, + scoped_type: &str, + ) -> PersistenceResult { + let result = sqlx::query( + r#" +WITH "parent"("name") AS MATERIALIZED ( + SELECT "name" + FROM "objects" + WHERE "object_type" = ?1 AND "id" = ?2 +) +DELETE FROM "objects" +WHERE ( + "object_type" = ?1 AND "id" = ?2 +) +OR ( + "object_type" = ?3 + AND "name" IN (SELECT "name" FROM "parent") +) +"#, + ) + .bind(parent_type) + .bind(parent_id) + .bind(scoped_type) + .execute(&self.pool) + .await + .map_err(|e| map_db_error(&e))?; + + Ok(result.rows_affected() > 0) + } + pub async fn put_scoped( &self, object_type: &str, diff --git a/crates/openshell-server/src/persistence/tests.rs b/crates/openshell-server/src/persistence/tests.rs index c92f63827a..99e9959ff4 100644 --- a/crates/openshell-server/src/persistence/tests.rs +++ b/crates/openshell-server/src/persistence/tests.rs @@ -1250,6 +1250,213 @@ async fn cas_delete_if_fails_with_wrong_version() { assert_eq!(record.resource_version, 1); } +#[tokio::test] +async fn delete_with_name_scoped_if_version_removes_matching_pair_only() { + use super::WriteCondition; + + let store = test_store().await; + + store + .put_if( + "sandbox", + "sandbox-id-1", + "sandbox-1", + b"sandbox", + None, + WriteCondition::MustCreate, + ) + .await + .unwrap(); + store + .put( + "sandbox_settings", + "settings-id-1", + "sandbox-1", + b"matching settings", + None, + ) + .await + .unwrap(); + store + .put( + "sandbox_settings", + "settings-id-2", + "sandbox-2", + b"unrelated settings", + None, + ) + .await + .unwrap(); + + let deleted = store + .delete_with_name_scoped_if_version("sandbox", "sandbox-id-1", 1, "sandbox_settings") + .await + .unwrap(); + + assert!(deleted); + assert!( + store + .get("sandbox", "sandbox-id-1") + .await + .unwrap() + .is_none() + ); + assert!( + store + .get("sandbox_settings", "settings-id-1") + .await + .unwrap() + .is_none() + ); + assert_eq!( + store + .get("sandbox_settings", "settings-id-2") + .await + .unwrap() + .unwrap() + .payload, + b"unrelated settings" + ); + + // A settings row that reuses the old name after parent removal must not + // be selected by a later cleanup attempt for the absent parent. + store + .put( + "sandbox_settings", + "settings-id-reused", + "sandbox-1", + b"reused-name settings", + None, + ) + .await + .unwrap(); + assert!( + !store + .delete_with_name_scoped_if_version("sandbox", "sandbox-id-1", 1, "sandbox_settings",) + .await + .unwrap() + ); + assert!( + store + .get("sandbox_settings", "settings-id-reused") + .await + .unwrap() + .is_some() + ); +} + +#[tokio::test] +async fn delete_with_name_scoped_if_version_conflict_preserves_both() { + use super::WriteCondition; + + let store = test_store().await; + + store + .put_if( + "sandbox", + "sandbox-id-1", + "sandbox-1", + b"sandbox", + None, + WriteCondition::MustCreate, + ) + .await + .unwrap(); + store + .put( + "sandbox_settings", + "settings-id-1", + "sandbox-1", + b"matching settings", + None, + ) + .await + .unwrap(); + + let result = store + .delete_with_name_scoped_if_version("sandbox", "sandbox-id-1", 99, "sandbox_settings") + .await; + + assert!(matches!( + result, + Err(PersistenceError::Conflict { + current_resource_version: Some(1) + }) + )); + assert!( + store + .get("sandbox", "sandbox-id-1") + .await + .unwrap() + .is_some() + ); + assert!( + store + .get("sandbox_settings", "settings-id-1") + .await + .unwrap() + .is_some() + ); +} + +#[tokio::test] +async fn delete_with_name_scoped_removes_matching_pair_only() { + let store = test_store().await; + + store + .put("sandbox", "sandbox-id-1", "sandbox-1", b"sandbox", None) + .await + .unwrap(); + store + .put( + "sandbox_settings", + "settings-id-1", + "sandbox-1", + b"matching settings", + None, + ) + .await + .unwrap(); + store + .put( + "sandbox_settings", + "settings-id-2", + "sandbox-2", + b"unrelated settings", + None, + ) + .await + .unwrap(); + + assert!( + store + .delete_with_name_scoped("sandbox", "sandbox-id-1", "sandbox_settings") + .await + .unwrap() + ); + assert!( + store + .get("sandbox", "sandbox-id-1") + .await + .unwrap() + .is_none() + ); + assert!( + store + .get("sandbox_settings", "settings-id-1") + .await + .unwrap() + .is_none() + ); + assert!( + store + .get("sandbox_settings", "settings-id-2") + .await + .unwrap() + .is_some() + ); +} + #[tokio::test] async fn cas_resource_version_increments() { use super::WriteCondition; diff --git a/e2e/rust/tests/sandbox_lifecycle.rs b/e2e/rust/tests/sandbox_lifecycle.rs index 01e89422e0..48902e6471 100644 --- a/e2e/rust/tests/sandbox_lifecycle.rs +++ b/e2e/rust/tests/sandbox_lifecycle.rs @@ -8,7 +8,10 @@ use std::time::Duration; use openshell_e2e::harness::binary::{openshell_cmd, openshell_tty_cmd}; use openshell_e2e::harness::output::{extract_field, strip_ansi}; -use tokio::time::sleep; +use tokio::time::{Instant, sleep}; + +const SANDBOX_PRESENCE_TIMEOUT: Duration = Duration::from_secs(30); +const SANDBOX_LIST_POLL_INTERVAL: Duration = Duration::from_millis(500); fn normalize_output(output: &str) -> String { let stripped = strip_ansi(output).replace('\r', ""); @@ -59,6 +62,28 @@ async fn sandbox_list_names() -> Vec { .collect() } +async fn assert_sandbox_presence_eventually( + sandbox_name: &str, + should_exist: bool, +) -> Result<(), Vec> { + let deadline = Instant::now() + SANDBOX_PRESENCE_TIMEOUT; + + loop { + let sandbox_names = sandbox_list_names().await; + let exists = sandbox_names.iter().any(|name| name == sandbox_name); + if exists == should_exist { + return Ok(()); + } + + let now = Instant::now(); + if now >= deadline { + return Err(sandbox_names); + } + + sleep(SANDBOX_LIST_POLL_INTERVAL.min(deadline - now)).await; + } +} + async fn delete_sandbox(name: &str) { let mut cmd = openshell_cmd(); cmd.args(["sandbox", "delete", name]) @@ -90,15 +115,15 @@ async fn sandbox_create_keeps_sandbox_after_tty_command_by_default() { let sandbox_name = extract_sandbox_name(&combined).expect("sandbox name should be present in output"); - for _ in 0..20 { - if sandbox_list_names().await.contains(&sandbox_name) { - delete_sandbox(&sandbox_name).await; - return; - } - sleep(Duration::from_millis(500)).await; + if let Err(last_sandbox_list) = assert_sandbox_presence_eventually(&sandbox_name, true).await { + delete_sandbox(&sandbox_name).await; + panic!( + "sandbox {sandbox_name} should still exist by default after {SANDBOX_PRESENCE_TIMEOUT:?}; \ + last observed sandbox list: {last_sandbox_list:?}" + ); } - panic!("sandbox {sandbox_name} should still exist by default"); + delete_sandbox(&sandbox_name).await; } #[tokio::test] @@ -124,13 +149,11 @@ async fn sandbox_create_with_no_keep_cleans_up_after_tty_command() { let sandbox_name = extract_sandbox_name(&combined).expect("sandbox name should be present in output"); - for _ in 0..20 { - if !sandbox_list_names().await.contains(&sandbox_name) { - return; - } - sleep(Duration::from_millis(500)).await; + if let Err(last_sandbox_list) = assert_sandbox_presence_eventually(&sandbox_name, false).await { + delete_sandbox(&sandbox_name).await; + panic!( + "sandbox {sandbox_name} should have been deleted automatically after \ + {SANDBOX_PRESENCE_TIMEOUT:?}; last observed sandbox list: {last_sandbox_list:?}" + ); } - - delete_sandbox(&sandbox_name).await; - panic!("sandbox {sandbox_name} should have been deleted automatically"); } From 8283eecff739992796381daf88841b2fe2f09714 Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Thu, 16 Jul 2026 22:24:33 -0700 Subject: [PATCH 2/4] refactor(server): simplify sandbox settings cleanup Signed-off-by: Piotr Mlocek --- architecture/compute-runtimes.md | 6 +- crates/openshell-server/src/compute/mod.rs | 118 +++++----- .../openshell-server/src/persistence/mod.rs | 40 ---- .../src/persistence/postgres.rs | 87 -------- .../src/persistence/sqlite.rs | 82 ------- .../openshell-server/src/persistence/tests.rs | 207 ------------------ 6 files changed, 56 insertions(+), 484 deletions(-) diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index 010d8a9e05..b2fe32f998 100644 --- a/architecture/compute-runtimes.md +++ b/architecture/compute-runtimes.md @@ -43,9 +43,9 @@ deleted events are idempotent, and snapshots for absent rows are ignored. An accepted delete (`deleted = true`) is finalized by the watcher. If the backend is already absent (`deleted = false`), the request removes gateway state -synchronously. Sandbox rows and name-scoped settings are removed atomically; -SSH sessions, indexes, and watch/log buses are cleaned only after confirmed -removal. +synchronously. Sandbox row removal remains bound to the stable ID and resource +version. Settings retain their existing best-effort name-based cleanup; SSH +sessions, indexes, and watch/log buses are cleaned after confirmed removal. The request workflow runs in an owned task so client cancellation cannot strand a mutation. A gateway restart does not resume or reissue a persisted `Deleting` diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index daa618f08c..2754ac7b1e 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -870,10 +870,11 @@ impl ComputeRuntime { false } - /// Atomically removes the sandbox and its name-scoped settings only when - /// the expected resource version still owns the row. The caller holds - /// `sync_lock`; successful or already-completed durable removal also - /// clears this replica's index, sessions, and watch/log buses. + /// Removes the sandbox by stable ID only when the expected resource + /// version still owns the row. The caller holds `sync_lock`; a successful + /// delete also performs best-effort settings cleanup, while successful or + /// already-completed removal clears this replica's index, sessions, and + /// watch/log buses. async fn remove_sandbox_record_if_version_locked( &self, sandbox_id: &str, @@ -901,28 +902,21 @@ impl ComputeRuntime { match self .store - .delete_with_name_scoped_if_version( + .delete_if( Sandbox::object_type(), sandbox_id, expected_resource_version, - SANDBOX_SETTINGS_OBJECT_TYPE, ) .await { Ok(true) => { + self.cleanup_sandbox_settings(sandbox_id, &record.name) + .await; self.cleanup_removed_sandbox_state(sandbox_id).await; Ok(true) } Ok(false) => { - if self - .store - .get(Sandbox::object_type(), sandbox_id) - .await - .map_err(|err| err.to_string())? - .is_none() - { - self.cleanup_removed_sandbox_state(sandbox_id).await; - } + self.cleanup_removed_sandbox_state(sandbox_id).await; Ok(false) } Err(crate::persistence::PersistenceError::Conflict { @@ -1649,14 +1643,20 @@ impl ComputeRuntime { } async fn apply_deleted_locked(&self, sandbox_id: &str) -> Result<(), String> { - self.store - .delete_with_name_scoped( - Sandbox::object_type(), - sandbox_id, - SANDBOX_SETTINGS_OBJECT_TYPE, - ) + let sandbox = self + .store + .get_message::(sandbox_id) + .await + .map_err(|e| e.to_string())?; + let removed = self + .store + .delete(Sandbox::object_type(), sandbox_id) .await .map_err(|e| e.to_string())?; + if removed && let Some(sandbox) = sandbox.as_ref() { + self.cleanup_sandbox_settings(sandbox_id, sandbox.object_name()) + .await; + } self.cleanup_removed_sandbox_state(sandbox_id).await; Ok(()) } @@ -1672,6 +1672,21 @@ impl ComputeRuntime { Ok(()) } + async fn cleanup_sandbox_settings(&self, sandbox_id: &str, sandbox_name: &str) { + if let Err(err) = self + .store + .delete_by_name(SANDBOX_SETTINGS_OBJECT_TYPE, sandbox_name) + .await + { + warn!( + sandbox_id, + sandbox_name, + error = %err, + "Failed to delete sandbox settings during cleanup" + ); + } + } + async fn cleanup_sandbox_ssh_sessions(&self, sandbox_id: &str) { if let Ok(records) = self.store.list(SshSession::object_type(), 1000, 0).await { for record in records { @@ -3081,6 +3096,19 @@ mod tests { session } + async fn remove_sandbox_owned_records_from_store(runtime: &ComputeRuntime, sandbox: &Sandbox) { + runtime + .store + .delete_by_name(SANDBOX_SETTINGS_OBJECT_TYPE, sandbox.object_name()) + .await + .unwrap(); + runtime + .store + .delete(Sandbox::object_type(), sandbox.object_id()) + .await + .unwrap(); + } + async fn assert_sandbox_owned_records( runtime: &ComputeRuntime, sandbox: &Sandbox, @@ -4086,17 +4114,7 @@ mod tests { .expect("deleting notification was not sent") .expect("watch bus closed before the deleting notification"); - assert!( - runtime - .store - .delete_with_name_scoped( - Sandbox::object_type(), - "sb-1", - SANDBOX_SETTINGS_OBJECT_TYPE, - ) - .await - .unwrap() - ); + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; driver.release_delete(); assert!( @@ -4142,17 +4160,7 @@ mod tests { .expect("deleting notification was not sent") .expect("watch bus closed before the deleting notification"); - assert!( - runtime - .store - .delete_with_name_scoped( - Sandbox::object_type(), - "sb-1", - SANDBOX_SETTINGS_OBJECT_TYPE, - ) - .await - .unwrap() - ); + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; driver.release_delete(); assert!( @@ -4220,17 +4228,7 @@ mod tests { .expect("deleting notification was not sent") .expect("watch bus closed before the deleting notification"); - assert!( - runtime - .store - .delete_with_name_scoped( - Sandbox::object_type(), - "sb-1", - SANDBOX_SETTINGS_OBJECT_TYPE, - ) - .await - .unwrap() - ); + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; driver.release_delete(); tokio::time::timeout(Duration::from_secs(1), delete) @@ -4276,17 +4274,7 @@ mod tests { .expect("deleting notification was not sent") .expect("watch bus closed before the deleting notification"); - assert!( - runtime - .store - .delete_with_name_scoped( - Sandbox::object_type(), - "sb-1", - SANDBOX_SETTINGS_OBJECT_TYPE, - ) - .await - .unwrap() - ); + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; driver.release_get(); tokio::time::timeout(Duration::from_secs(1), delete) diff --git a/crates/openshell-server/src/persistence/mod.rs b/crates/openshell-server/src/persistence/mod.rs index 7dad899f3b..8b6172c1c6 100644 --- a/crates/openshell-server/src/persistence/mod.rs +++ b/crates/openshell-server/src/persistence/mod.rs @@ -261,46 +261,6 @@ impl Store { store_dispatch!(self.delete_if(object_type, id, expected_resource_version)) } - /// Atomically delete an object and the object of `scoped_type` that has - /// the same name, subject to an exact parent resource-version match. - /// - /// The parent name is captured by the same database statement that - /// removes the records. This prevents a concurrent name reuse from - /// causing the newly-created name-scoped object to be removed. - /// - /// # Returns - /// * `Ok(true)` - The parent and any matching name-scoped object were deleted - /// * `Ok(false)` - The parent was not found - /// * `Err(Conflict)` - The parent exists with a different resource version - pub async fn delete_with_name_scoped_if_version( - &self, - parent_type: &str, - parent_id: &str, - expected_resource_version: u64, - scoped_type: &str, - ) -> PersistenceResult { - store_dispatch!(self.delete_with_name_scoped_if_version( - parent_type, - parent_id, - expected_resource_version, - scoped_type - )) - } - - /// Atomically delete an object and the object of `scoped_type` that has - /// the same name. - /// - /// The parent name is captured by the same database statement that - /// removes the records, preventing cleanup from racing with name reuse. - pub async fn delete_with_name_scoped( - &self, - parent_type: &str, - parent_id: &str, - scoped_type: &str, - ) -> PersistenceResult { - store_dispatch!(self.delete_with_name_scoped(parent_type, parent_id, scoped_type)) - } - /// Insert or update a generic named object with an application-owned scope. pub async fn put_scoped( &self, diff --git a/crates/openshell-server/src/persistence/postgres.rs b/crates/openshell-server/src/persistence/postgres.rs index 18960d54c6..12299805e9 100644 --- a/crates/openshell-server/src/persistence/postgres.rs +++ b/crates/openshell-server/src/persistence/postgres.rs @@ -244,93 +244,6 @@ WHERE object_type = $1 AND id = $2 AND resource_version = $3 } } - pub async fn delete_with_name_scoped_if_version( - &self, - parent_type: &str, - parent_id: &str, - expected_resource_version: u64, - scoped_type: &str, - ) -> PersistenceResult { - // Feed the sibling delete exclusively from the exact parent delete's - // RETURNING output. A stale resource version therefore cannot remove - // settings, and both deletes commit as one statement. - let row = sqlx::query( - r" -WITH deleted_parent AS ( - DELETE FROM objects - WHERE object_type = $1 AND id = $2 AND resource_version = $3 - RETURNING name -), -deleted_scoped AS ( - DELETE FROM objects - WHERE object_type = $4 - AND name IN (SELECT name FROM deleted_parent) - AND NOT (object_type = $1 AND id = $2) - RETURNING id -) -SELECT - EXISTS (SELECT 1 FROM deleted_parent) AS parent_deleted, - (SELECT COUNT(*) FROM deleted_scoped) AS scoped_deleted -", - ) - .bind(parent_type) - .bind(parent_id) - .bind(i64::try_from(expected_resource_version).unwrap_or(i64::MAX)) - .bind(scoped_type) - .fetch_one(&self.pool) - .await - .map_err(|e| map_db_error(&e))?; - - let parent_deleted: bool = row.get("parent_deleted"); - if parent_deleted { - Ok(true) - } else { - let existing = self.get(parent_type, parent_id).await?; - if let Some(record) = existing { - Err(PersistenceError::Conflict { - current_resource_version: Some(record.resource_version), - }) - } else { - Ok(false) - } - } - } - - pub async fn delete_with_name_scoped( - &self, - parent_type: &str, - parent_id: &str, - scoped_type: &str, - ) -> PersistenceResult { - let row = sqlx::query( - r" -WITH deleted_parent AS ( - DELETE FROM objects - WHERE object_type = $1 AND id = $2 - RETURNING name -), -deleted_scoped AS ( - DELETE FROM objects - WHERE object_type = $3 - AND name IN (SELECT name FROM deleted_parent) - AND NOT (object_type = $1 AND id = $2) - RETURNING id -) -SELECT - EXISTS (SELECT 1 FROM deleted_parent) AS parent_deleted, - (SELECT COUNT(*) FROM deleted_scoped) AS scoped_deleted -", - ) - .bind(parent_type) - .bind(parent_id) - .bind(scoped_type) - .fetch_one(&self.pool) - .await - .map_err(|e| map_db_error(&e))?; - - Ok(row.get("parent_deleted")) - } - pub async fn put_scoped( &self, object_type: &str, diff --git a/crates/openshell-server/src/persistence/sqlite.rs b/crates/openshell-server/src/persistence/sqlite.rs index b0a60ede64..57d5c20e31 100644 --- a/crates/openshell-server/src/persistence/sqlite.rs +++ b/crates/openshell-server/src/persistence/sqlite.rs @@ -265,88 +265,6 @@ WHERE "object_type" = ?1 AND "id" = ?2 AND "resource_version" = ?3 } } - pub async fn delete_with_name_scoped_if_version( - &self, - parent_type: &str, - parent_id: &str, - expected_resource_version: u64, - scoped_type: &str, - ) -> PersistenceResult { - // SQLite serializes writers. Materializing the matching parent inside - // this DELETE captures its name before either row is removed, while - // keeping both removals in the same write statement. - let result = sqlx::query( - r#" -WITH "parent"("name") AS MATERIALIZED ( - SELECT "name" - FROM "objects" - WHERE "object_type" = ?1 AND "id" = ?2 AND "resource_version" = ?3 -) -DELETE FROM "objects" -WHERE ( - "object_type" = ?1 AND "id" = ?2 AND "resource_version" = ?3 -) -OR ( - "object_type" = ?4 - AND "name" IN (SELECT "name" FROM "parent") -) -"#, - ) - .bind(parent_type) - .bind(parent_id) - .bind(i64::try_from(expected_resource_version).unwrap_or(i64::MAX)) - .bind(scoped_type) - .execute(&self.pool) - .await - .map_err(|e| map_db_error(&e))?; - - if result.rows_affected() > 0 { - Ok(true) - } else { - let existing = self.get(parent_type, parent_id).await?; - if let Some(record) = existing { - Err(PersistenceError::Conflict { - current_resource_version: Some(record.resource_version), - }) - } else { - Ok(false) - } - } - } - - pub async fn delete_with_name_scoped( - &self, - parent_type: &str, - parent_id: &str, - scoped_type: &str, - ) -> PersistenceResult { - let result = sqlx::query( - r#" -WITH "parent"("name") AS MATERIALIZED ( - SELECT "name" - FROM "objects" - WHERE "object_type" = ?1 AND "id" = ?2 -) -DELETE FROM "objects" -WHERE ( - "object_type" = ?1 AND "id" = ?2 -) -OR ( - "object_type" = ?3 - AND "name" IN (SELECT "name" FROM "parent") -) -"#, - ) - .bind(parent_type) - .bind(parent_id) - .bind(scoped_type) - .execute(&self.pool) - .await - .map_err(|e| map_db_error(&e))?; - - Ok(result.rows_affected() > 0) - } - pub async fn put_scoped( &self, object_type: &str, diff --git a/crates/openshell-server/src/persistence/tests.rs b/crates/openshell-server/src/persistence/tests.rs index 99e9959ff4..c92f63827a 100644 --- a/crates/openshell-server/src/persistence/tests.rs +++ b/crates/openshell-server/src/persistence/tests.rs @@ -1250,213 +1250,6 @@ async fn cas_delete_if_fails_with_wrong_version() { assert_eq!(record.resource_version, 1); } -#[tokio::test] -async fn delete_with_name_scoped_if_version_removes_matching_pair_only() { - use super::WriteCondition; - - let store = test_store().await; - - store - .put_if( - "sandbox", - "sandbox-id-1", - "sandbox-1", - b"sandbox", - None, - WriteCondition::MustCreate, - ) - .await - .unwrap(); - store - .put( - "sandbox_settings", - "settings-id-1", - "sandbox-1", - b"matching settings", - None, - ) - .await - .unwrap(); - store - .put( - "sandbox_settings", - "settings-id-2", - "sandbox-2", - b"unrelated settings", - None, - ) - .await - .unwrap(); - - let deleted = store - .delete_with_name_scoped_if_version("sandbox", "sandbox-id-1", 1, "sandbox_settings") - .await - .unwrap(); - - assert!(deleted); - assert!( - store - .get("sandbox", "sandbox-id-1") - .await - .unwrap() - .is_none() - ); - assert!( - store - .get("sandbox_settings", "settings-id-1") - .await - .unwrap() - .is_none() - ); - assert_eq!( - store - .get("sandbox_settings", "settings-id-2") - .await - .unwrap() - .unwrap() - .payload, - b"unrelated settings" - ); - - // A settings row that reuses the old name after parent removal must not - // be selected by a later cleanup attempt for the absent parent. - store - .put( - "sandbox_settings", - "settings-id-reused", - "sandbox-1", - b"reused-name settings", - None, - ) - .await - .unwrap(); - assert!( - !store - .delete_with_name_scoped_if_version("sandbox", "sandbox-id-1", 1, "sandbox_settings",) - .await - .unwrap() - ); - assert!( - store - .get("sandbox_settings", "settings-id-reused") - .await - .unwrap() - .is_some() - ); -} - -#[tokio::test] -async fn delete_with_name_scoped_if_version_conflict_preserves_both() { - use super::WriteCondition; - - let store = test_store().await; - - store - .put_if( - "sandbox", - "sandbox-id-1", - "sandbox-1", - b"sandbox", - None, - WriteCondition::MustCreate, - ) - .await - .unwrap(); - store - .put( - "sandbox_settings", - "settings-id-1", - "sandbox-1", - b"matching settings", - None, - ) - .await - .unwrap(); - - let result = store - .delete_with_name_scoped_if_version("sandbox", "sandbox-id-1", 99, "sandbox_settings") - .await; - - assert!(matches!( - result, - Err(PersistenceError::Conflict { - current_resource_version: Some(1) - }) - )); - assert!( - store - .get("sandbox", "sandbox-id-1") - .await - .unwrap() - .is_some() - ); - assert!( - store - .get("sandbox_settings", "settings-id-1") - .await - .unwrap() - .is_some() - ); -} - -#[tokio::test] -async fn delete_with_name_scoped_removes_matching_pair_only() { - let store = test_store().await; - - store - .put("sandbox", "sandbox-id-1", "sandbox-1", b"sandbox", None) - .await - .unwrap(); - store - .put( - "sandbox_settings", - "settings-id-1", - "sandbox-1", - b"matching settings", - None, - ) - .await - .unwrap(); - store - .put( - "sandbox_settings", - "settings-id-2", - "sandbox-2", - b"unrelated settings", - None, - ) - .await - .unwrap(); - - assert!( - store - .delete_with_name_scoped("sandbox", "sandbox-id-1", "sandbox_settings") - .await - .unwrap() - ); - assert!( - store - .get("sandbox", "sandbox-id-1") - .await - .unwrap() - .is_none() - ); - assert!( - store - .get("sandbox_settings", "settings-id-1") - .await - .unwrap() - .is_none() - ); - assert!( - store - .get("sandbox_settings", "settings-id-2") - .await - .unwrap() - .is_some() - ); -} - #[tokio::test] async fn cas_resource_version_increments() { use super::WriteCondition; From 6a15fd94590d705bfe40773334640046a28050d4 Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Fri, 17 Jul 2026 11:04:23 -0700 Subject: [PATCH 3/4] refactor(server): make delete lock ordering explicit Signed-off-by: Piotr Mlocek --- crates/openshell-server/src/compute/mod.rs | 65 ++++++++++++++++++---- 1 file changed, 53 insertions(+), 12 deletions(-) diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2754ac7b1e..1f34942040 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -71,6 +71,13 @@ struct DeleteGateRegistry { } impl DeleteGateRegistry { + async fn lock_for(&self, sandbox_id: &str) -> SandboxDeleteGuard { + let gate = self.gate_for(sandbox_id); + SandboxDeleteGuard { + _guard: gate.lock_owned().await, + } + } + fn gate_for(&self, sandbox_id: &str) -> Arc> { let mut gates = self .gates @@ -96,6 +103,16 @@ impl DeleteGateRegistry { } } +/// Proof that the current delete operation holds its sandbox-ID gate. +/// +/// Delete code must acquire this guard before taking `ComputeRuntime::sync_lock`. +/// Passing it to `lock_global_for_delete` makes that ordering visible at every +/// global-lock acquisition in the delete path. +#[derive(Debug)] +struct SandboxDeleteGuard { + _guard: tokio::sync::OwnedMutexGuard<()>, +} + #[derive(Debug)] struct DeleteTransition { /// Snapshot restored if the driver result is ambiguous and no newer write @@ -407,6 +424,16 @@ impl ComputeRuntime { self.sync_lock.clone().lock_owned().await } + /// Acquires the process-wide lock for code that already holds the + /// sandbox-ID delete gate. The guard parameter documents and enforces that + /// delete-path callers acquire locks in delete-gate -> global-lock order. + async fn lock_global_for_delete( + &self, + _delete_guard: &SandboxDeleteGuard, + ) -> tokio::sync::OwnedMutexGuard<()> { + self.sync_lock.clone().lock_owned().await + } + pub async fn new_docker( config: openshell_core::Config, docker_config: DockerComputeConfig, @@ -673,10 +700,9 @@ impl ComputeRuntime { .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))? .ok_or_else(|| Status::not_found("sandbox not found"))?; let candidate_id = candidate.object_id().to_string(); - let delete_gate = self.delete_gates.gate_for(&candidate_id); - let _delete_guard = delete_gate.lock().await; + let delete_guard = self.delete_gates.lock_for(&candidate_id).await; - let guard = self.sync_lock.lock().await; + let guard = self.lock_global_for_delete(&delete_guard).await; let current = self .store .get_message::(&candidate_id) @@ -722,9 +748,12 @@ impl ComputeRuntime { Ok(response) => { let deleted = response.into_inner().deleted; if deleted { - self.cleanup_local_state_if_sandbox_absent(&candidate_id) + self.cleanup_local_state_if_sandbox_absent(&delete_guard, &candidate_id) .await?; - } else if !self.remove_deleting_sandbox_record(&candidate_id).await { + } else if !self + .remove_deleting_sandbox_record(&delete_guard, &candidate_id) + .await + { return Err(Status::internal( "compute resource was absent, but gateway cleanup did not complete", )); @@ -732,7 +761,7 @@ impl ComputeRuntime { Ok(deleted) } Err(err) => { - self.recover_failed_delete(&transition).await; + self.recover_failed_delete(&delete_guard, &transition).await; Err(Status::internal(format!( "delete sandbox failed: {}", err.message() @@ -812,8 +841,12 @@ impl ComputeRuntime { /// /// Benign resource-version changes are retried, but cleanup stops if the /// row leaves `Deleting`; that state belongs to a concurrent writer. - async fn remove_deleting_sandbox_record(&self, sandbox_id: &str) -> bool { - let _guard = self.sync_lock.lock().await; + async fn remove_deleting_sandbox_record( + &self, + delete_guard: &SandboxDeleteGuard, + sandbox_id: &str, + ) -> bool { + let _guard = self.lock_global_for_delete(delete_guard).await; for attempt in 1..=DELETE_PHASE_CAS_RETRY_LIMIT { let record = match self.store.get(Sandbox::object_type(), sandbox_id).await { Ok(Some(record)) => record, @@ -941,14 +974,18 @@ impl ComputeRuntime { /// exact `Deleting` resource version to apply one of three outcomes: /// reconcile an observed backend snapshot, remove a confirmed-absent /// backend, or restore the pre-delete snapshot when lookup is inconclusive. - async fn recover_failed_delete(&self, transition: &DeleteTransition) { + async fn recover_failed_delete( + &self, + delete_guard: &SandboxDeleteGuard, + transition: &DeleteTransition, + ) { let sandbox_id = transition.deleting.object_id(); let sandbox_name = transition.deleting.object_name(); let deleting_resource_version = sandbox_resource_version(&transition.deleting); // The driver lookup is deliberately outside the process-wide guard. let observed = self.get_driver_sandbox(sandbox_id, sandbox_name).await; - let _guard = self.sync_lock.lock().await; + let _guard = self.lock_global_for_delete(delete_guard).await; match observed { Ok(Some(snapshot)) if snapshot.id == sandbox_id && snapshot.status.is_some() => { @@ -1708,8 +1745,12 @@ impl ComputeRuntime { } } - async fn cleanup_local_state_if_sandbox_absent(&self, sandbox_id: &str) -> Result<(), Status> { - let _guard = self.sync_lock.lock().await; + async fn cleanup_local_state_if_sandbox_absent( + &self, + delete_guard: &SandboxDeleteGuard, + sandbox_id: &str, + ) -> Result<(), Status> { + let _guard = self.lock_global_for_delete(delete_guard).await; let record = self .store .get(Sandbox::object_type(), sandbox_id) From 335961d85d62374664240225bbe8cbea831434bc Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Fri, 17 Jul 2026 11:58:12 -0700 Subject: [PATCH 4/4] fix(server): bind delete worker to stable sandbox id Signed-off-by: Piotr Mlocek --- architecture/compute-runtimes.md | 9 +- crates/openshell-server/src/compute/mod.rs | 111 +++++++++++++++++---- 2 files changed, 97 insertions(+), 23 deletions(-) diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index b2fe32f998..0e938f5624 100644 --- a/architecture/compute-runtimes.md +++ b/architecture/compute-runtimes.md @@ -32,11 +32,16 @@ of re-querying drivers on each request. ## Deletion Lifecycle -Delete requests use per-sandbox gates to serialize duplicate driver calls. A -request resolves the name once and remains bound to that stable ID. The only +Delete requests use per-sandbox gates to serialize delete attempts. A request +resolves the name once and remains bound to that stable ID. The only combined lock order is delete gate, then the gateway-wide state guard; external driver calls run without the global guard. +Delete gates are process-local and do not coordinate gateway replicas. They +serialize attempts rather than share results: if one attempt fails and recovery +restores a deletable state, a request waiting on the gate may retry the driver. +Persisted resource-version checks remain the cross-replica safety boundary. + Watcher events do not acquire delete gates. Exact resource-version checks allow them to interleave safely: status snapshots are no-ops for `Deleting` rows, deleted events are idempotent, and snapshots for absent rows are ignored. diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 1f34942040..726121f431 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -113,6 +113,12 @@ struct SandboxDeleteGuard { _guard: tokio::sync::OwnedMutexGuard<()>, } +#[derive(Debug)] +struct SandboxDeleteTarget { + sandbox_id: String, + sandbox_name: String, +} + #[derive(Debug)] struct DeleteTransition { /// Snapshot restored if the driver result is ambiguous and no newer write @@ -675,12 +681,26 @@ impl ComputeRuntime { } pub async fn delete_sandbox(&self, name: &str) -> Result { + // Resolve the request to a stable identity before spawning the owned + // worker. Cancellation before this lookup completes is harmless + // because no mutation has occurred; after the lookup, the worker can + // never retarget a replacement sandbox that reuses the name. + let candidate = self + .store + .get_message_by_name::(name) + .await + .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))? + .ok_or_else(|| Status::not_found("sandbox not found"))?; + let target = SandboxDeleteTarget { + sandbox_id: candidate.object_id().to_string(), + sandbox_name: candidate.object_name().to_string(), + }; + // The owned task keeps the delete workflow running if the client drops // its request after the gateway has durably persisted `Deleting`. We // still await it here so the API remains synchronous for live clients. let runtime = self.clone(); - let name = name.to_string(); - tokio::spawn(async move { runtime.delete_sandbox_inner(&name).await }) + tokio::spawn(async move { runtime.delete_sandbox_inner(target).await }) .await .map_err(|err| { Status::internal(format!( @@ -689,33 +709,23 @@ impl ComputeRuntime { })? } - async fn delete_sandbox_inner(&self, name: &str) -> Result { - // Resolve the name once. Everything after this point is bound to the - // resulting stable ID, so name reuse cannot redirect this request to a - // newly-created sandbox. - let candidate = self - .store - .get_message_by_name::(name) - .await - .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))? - .ok_or_else(|| Status::not_found("sandbox not found"))?; - let candidate_id = candidate.object_id().to_string(); - let delete_guard = self.delete_gates.lock_for(&candidate_id).await; + async fn delete_sandbox_inner(&self, target: SandboxDeleteTarget) -> Result { + let delete_guard = self.delete_gates.lock_for(&target.sandbox_id).await; let guard = self.lock_global_for_delete(&delete_guard).await; let current = self .store - .get_message::(&candidate_id) + .get_message::(&target.sandbox_id) .await .map_err(|e| Status::internal(format!("fetch sandbox failed: {e}")))?; let Some(current) = current else { // A delete that owned this ID's gate completed while this request // waited. A different sandbox may now use the old name; this // request acknowledges only the disappearance of its original ID. - self.cleanup_removed_sandbox_state(&candidate_id).await; + self.cleanup_removed_sandbox_state(&target.sandbox_id).await; return Ok(true); }; - if current.object_name() != name { + if current.object_name() != target.sandbox_name { return Err(Status::aborted( "sandbox name changed while the delete request was waiting; retry explicitly", )); @@ -725,7 +735,7 @@ impl ComputeRuntime { // `Deleting` row used to fence recovery, and the prior row used only // for exact-version rollback after an ambiguous driver failure. let transition = match self - .begin_sandbox_delete_with_initial_snapshot(&candidate_id, Some(current)) + .begin_sandbox_delete_with_initial_snapshot(&target.sandbox_id, Some(current)) .await? { BeginDelete::AlreadyDeleting => return Ok(true), @@ -733,7 +743,7 @@ impl ComputeRuntime { }; self.sandbox_index.update_from_sandbox(&transition.deleting); - self.sandbox_watch_bus.notify(&candidate_id); + self.sandbox_watch_bus.notify(&target.sandbox_id); drop(guard); let result = self @@ -748,10 +758,10 @@ impl ComputeRuntime { Ok(response) => { let deleted = response.into_inner().deleted; if deleted { - self.cleanup_local_state_if_sandbox_absent(&delete_guard, &candidate_id) + self.cleanup_local_state_if_sandbox_absent(&delete_guard, &target.sandbox_id) .await?; } else if !self - .remove_deleting_sandbox_record(&delete_guard, &candidate_id) + .remove_deleting_sandbox_record(&delete_guard, &target.sandbox_id) .await { return Err(Status::internal( @@ -3901,6 +3911,65 @@ mod tests { assert_eq!(driver.delete_calls(), 1); } + #[tokio::test] + async fn waiting_delete_retries_after_leader_failure_recovery() { + let driver = ControlledDriver::new(); + driver.block_delete(); + driver.set_delete_outcome(ControlledDeleteOutcome::Error("delete failed")); + driver.set_get_outcome(ControlledGetOutcome::Error("lookup failed")); + let runtime = test_runtime(driver.clone()).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + let delete_gate = runtime.delete_gates.gate_for(sandbox.object_id()); + let first_runtime = runtime.clone(); + let first = tokio::spawn(async move { first_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("first delete did not reach the driver"); + + let second_runtime = runtime.clone(); + let second = tokio::spawn(async move { second_runtime.delete_sandbox("sandbox-a").await }); + tokio::time::timeout(Duration::from_secs(1), async { + while Arc::strong_count(&delete_gate) < 3 { + tokio::task::yield_now().await; + } + }) + .await + .expect("second delete did not start waiting on the sandbox gate"); + + driver.release_delete(); + let first_error = tokio::time::timeout(Duration::from_secs(1), first) + .await + .expect("first delete did not finish") + .unwrap() + .unwrap_err(); + assert!(first_error.message().contains("delete failed")); + + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await + .expect("waiting delete did not retry the driver call"); + driver.set_delete_outcome(ControlledDeleteOutcome::Ok(false)); + driver.release_delete(); + + assert!( + !tokio::time::timeout(Duration::from_secs(1), second) + .await + .expect("second delete did not finish") + .unwrap() + .unwrap() + ); + assert_eq!(driver.delete_calls(), 2); + assert!( + runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn waiting_delete_does_not_retarget_a_reused_name() { let driver = ControlledDriver::new();