diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index ec42277ab..b2fe32f99 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 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` +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 239feb74d..2754ac7b1 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,258 @@ 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 + } + + /// 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, + 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_if( + Sandbox::object_type(), + sandbox_id, + expected_resource_version, + ) + .await + { + Ok(true) => { + self.cleanup_sandbox_settings(sandbox_id, &record.name) + .await; + self.cleanup_removed_sandbox_state(sandbox_id).await; + Ok(true) + } + Ok(false) => { + 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 +1517,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 +1579,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); } @@ -1384,33 +1648,40 @@ impl ComputeRuntime { .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 + let removed = self .store .delete(Sandbox::object_type(), sandbox_id) .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); + 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(()) } - async fn cleanup_sandbox_owned_records(&self, sandbox: &Sandbox) { - self.cleanup_sandbox_ssh_sessions(sandbox.object_id()).await; + 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(()) + } - if let Err(e) = self + 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.object_name()) + .delete_by_name(SANDBOX_SETTINGS_OBJECT_TYPE, sandbox_name) .await { warn!( - sandbox_id = %sandbox.object_id(), - sandbox_name = %sandbox.object_name(), - error = %e, + sandbox_id, + sandbox_name, + error = %err, "Failed to delete sandbox settings during cleanup" ); } @@ -1437,6 +1708,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,19 +1739,22 @@ impl ComputeRuntime { snapshot: DriverSandbox, sweep_started_at_ms: i64, ) -> Result<(), String> { - 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(()); - }; + 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(()); - } + 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) @@ -1469,6 +1763,21 @@ impl ComputeRuntime { return Ok(()); }; + 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.resource_version != expected_resource_version + || existing.updated_at_ms > sweep_started_at_ms + { + return Ok(()); + } + self.apply_sandbox_update_locked(current, Some(existing)) .await } @@ -1479,42 +1788,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 +1865,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 +2207,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 +2234,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 +2560,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 +2573,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 +2797,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 +3029,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 +3079,62 @@ 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 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, + 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 +3156,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 +3541,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 +3561,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 +3583,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 +3599,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 +3643,920 @@ 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"); + + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; + 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"); + + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; + 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"); + + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; + 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"); + + remove_sandbox_owned_records_from_store(&runtime, &sandbox).await; + 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/e2e/rust/tests/sandbox_lifecycle.rs b/e2e/rust/tests/sandbox_lifecycle.rs index 01e89422e..48902e647 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"); }