diff --git a/.agents/skills/debug-openshell-cluster/SKILL.md b/.agents/skills/debug-openshell-cluster/SKILL.md index 6a9bcd25b9..7a73131ea0 100644 --- a/.agents/skills/debug-openshell-cluster/SKILL.md +++ b/.agents/skills/debug-openshell-cluster/SKILL.md @@ -194,6 +194,13 @@ For source checkout development, restart the local gateway with: mise run gateway:docker ``` +During a graceful gateway restart, Docker, Podman, and VM sandboxes with +running intent should stop before the gateway exits and restart after it +returns. Check for `Stopped sandbox during gateway shutdown` and `Started +sandbox during gateway startup` in gateway logs. A sandbox explicitly stopped +through the CLI remains stopped. Kubernetes sandboxes are cluster-owned and do +not follow this local gateway lifecycle. + ### Step 5: Check Podman-Backed Gateways ```bash diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index 831be067ab..0b901ef3db 100644 --- a/architecture/compute-runtimes.md +++ b/architecture/compute-runtimes.md @@ -106,6 +106,15 @@ compute to zero, and VM retains its launch request and writable overlay beside a stop marker. Delete remains a separate operation that removes these resources. +On graceful gateway shutdown, persisted running intent for Docker, Podman, and +VM is stopped through the shared `StopSandbox` RPC before any gateway-managed +driver process exits. The gateway does not persist `Stopped` for this +infrastructure event. On startup, it reconciles the retained intent through the +shared idempotent `StartSandbox` RPC before watch processing begins. Explicitly +`Stopped` sandboxes are excluded from both sweeps. Kubernetes workloads are +cluster-owned and continue running without gateway shutdown or startup +lifecycle calls. + ## Deletion Lifecycle Lifecycle requests use per-sandbox gates to serialize stop, start, and diff --git a/crates/openshell-driver-docker/README.md b/crates/openshell-driver-docker/README.md index 48e56de7bc..fe705d0b50 100644 --- a/crates/openshell-driver-docker/README.md +++ b/crates/openshell-driver-docker/README.md @@ -26,6 +26,11 @@ policy. Start starts that same container, so files in the resolved OCI workspace remain available. A durably stopped sandbox is excluded from gateway startup recovery and stays stopped across gateway restarts. Delete continues to force-remove the container and clean up driver-owned material. +Graceful gateway shutdown sends `StopSandbox` for each sandbox whose persisted +phase requires running compute without changing that persisted intent. On +startup, the gateway sends an idempotent `StartSandbox` request for the same +sandboxes, restarting their retained containers. Explicitly stopped sandboxes +remain excluded. Before creating the container, the driver inspects the final sandbox image and captures its immutable image ID, raw OCI `Config.User`, and OCI diff --git a/crates/openshell-driver-docker/src/lib.rs b/crates/openshell-driver-docker/src/lib.rs index b1fb5ec224..cd1a0a8c51 100644 --- a/crates/openshell-driver-docker/src/lib.rs +++ b/crates/openshell-driver-docker/src/lib.rs @@ -1055,69 +1055,6 @@ impl DockerComputeDriver { } } - pub async fn stop_managed_containers_on_shutdown(&self) -> Result { - let containers = self.list_managed_container_summaries().await?; - let targets = containers - .into_iter() - .filter_map(|container| { - let state = container.state.unwrap_or(ContainerSummaryStateEnum::EMPTY); - if container_state_needs_shutdown_stop(state) { - summary_container_target(&container) - } else { - None - } - }) - .collect::>(); - let target_count = targets.len(); - let mut stopped = 0usize; - let mut failures = Vec::new(); - let stop_timeout_secs = self.config.stop_timeout_secs; - - let mut stop_results = futures::stream::iter(targets.into_iter().map(|target| { - let docker = self.docker.clone(); - async move { - let result = docker - .stop_container( - &target, - Some( - StopContainerOptionsBuilder::default() - .t(docker_stop_timeout_secs(stop_timeout_secs)) - .build(), - ), - ) - .await; - (target, result) - } - })) - .buffer_unordered(16); - - while let Some((target, result)) = stop_results.next().await { - match result { - Ok(()) => { - stopped += 1; - } - Err(err) if is_not_found_error(&err) || is_not_modified_error(&err) => {} - Err(err) => { - warn!( - container = %target, - error = %err, - "Failed to stop Docker sandbox container during shutdown" - ); - failures.push(target); - } - } - } - - if !failures.is_empty() { - return Err(Status::internal(format!( - "failed to stop {} of {target_count} Docker sandbox containers during shutdown", - failures.len() - ))); - } - - Ok(stopped) - } - async fn reserve_pending_sandbox(&self, sandbox: &DriverSandbox) -> Result<(), Status> { let mut pending = self.pending.lock().await; if pending @@ -3200,15 +3137,6 @@ fn summary_container_target(summary: &ContainerSummary) -> Option { .or_else(|| summary_container_name(summary)) } -fn container_state_needs_shutdown_stop(state: ContainerSummaryStateEnum) -> bool { - matches!( - state, - ContainerSummaryStateEnum::RUNNING - | ContainerSummaryStateEnum::RESTARTING - | ContainerSummaryStateEnum::PAUSED - ) -} - /// States from which a managed container can be brought back to running by /// `start_container`. Skip `Restarting` (already coming up), `Removing`, /// `Dead` (terminal), `Paused` (needs `unpause`, not `start`), and diff --git a/crates/openshell-driver-podman/README.md b/crates/openshell-driver-podman/README.md index 67ba07944e..aabf28c5b2 100644 --- a/crates/openshell-driver-podman/README.md +++ b/crates/openshell-driver-podman/README.md @@ -28,6 +28,12 @@ volume. Stopped managed containers remain visible through list and watch reconciliation. Delete remains responsible for removing the container, driver-owned secrets, and workspace volume. +Graceful gateway shutdown sends `StopSandbox` for each sandbox whose persisted +phase requires running compute without changing that persisted intent. On +startup, the gateway sends an idempotent `StartSandbox` request for the same +sandboxes, restarting their retained containers. Explicitly stopped sandboxes +remain excluded. + ## Architecture The Podman driver communicates with the Podman daemon over a Unix socket and diff --git a/crates/openshell-driver-vm/README.md b/crates/openshell-driver-vm/README.md index 10ffac2ba1..bf902013e1 100644 --- a/crates/openshell-driver-vm/README.md +++ b/crates/openshell-driver-vm/README.md @@ -194,10 +194,11 @@ during the first prepare. The driver also writes the accepted `DriverSandbox` launch request to `/sandboxes//sandbox.pb`. If the gateway restarts, it starts a -new VM driver process; that process scans the sandbox state directories, -restarts each persisted VM launcher, and preserves any existing `overlay.ext4` -instead of cloning a fresh overlay template. If a restart happened before the -overlay was created, the driver creates it during the start attempt. +new VM driver process. During graceful shutdown, the gateway first sends the +shared `StopSandbox` request for each persisted running-intent sandbox, which +stops its launcher while retaining the launch request and `overlay.ext4`. +After driver initialization, the gateway sends the idempotent `StartSandbox` +request for that retained intent. Explicitly stopped sandboxes remain excluded. Stop writes a marker in the sandbox state directory before terminating the launcher and releasing host GPU and network allocations. It retains diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index e09b63de09..82bf2e2c58 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -83,6 +83,9 @@ type SharedComputeDriver = use traced_driver::TracedDriver; +const LIFECYCLE_SWEEP_PAGE_SIZE: u32 = 1000; +const SHUTDOWN_STOP_CONCURRENCY: usize = 16; + /// Instrumenting wrapper around the compute driver. mod traced_driver { use std::future::Future; @@ -276,46 +279,6 @@ pub struct ComputeDriverInfoSnapshot { pub driver_version: String, } -#[tonic::async_trait] -trait ShutdownCleanup: Send + Sync { - async fn cleanup_on_shutdown(&self) -> Result<(), String>; -} - -#[tonic::async_trait] -#[cfg(not(target_os = "windows"))] -impl ShutdownCleanup for DockerComputeDriver { - async fn cleanup_on_shutdown(&self) -> Result<(), String> { - let stopped = self - .stop_managed_containers_on_shutdown() - .await - .map_err(|err| err.to_string())?; - info!( - stopped_containers = stopped, - "Stopped Docker sandbox containers during gateway shutdown" - ); - Ok(()) - } -} - -/// Start a single sandbox whose store record indicates it should be -/// running. Implemented by drivers (currently only Docker) where compute -/// resources do not auto-restart with the gateway. Returns `Ok(true)` if -/// the backend resource was found and started (or was already running), -/// `Ok(false)` if no backend resource exists. -#[tonic::async_trait] -trait StartupSandboxStarter: Send + Sync { - async fn start_sandbox(&self, sandbox_id: &str, sandbox_name: &str) -> Result; -} - -#[tonic::async_trait] -#[cfg(not(target_os = "windows"))] -impl StartupSandboxStarter for DockerComputeDriver { - async fn start_sandbox(&self, sandbox_id: &str, sandbox_name: &str) -> Result { - Self::start_sandbox(self, sandbox_id, sandbox_name) - .await - .map_err(|err| err.to_string()) - } -} /// Interval between store-vs-backend reconciliation sweeps. const RECONCILE_INTERVAL: Duration = Duration::from_secs(60); @@ -590,8 +553,6 @@ impl ComputeDriver for RemoteComputeDriver { pub struct ComputeRuntime { driver: TracedDriver, driver_info: ComputeDriverInfoSnapshot, - shutdown_cleanup: Option>, - startup_starter: Option>, driver_process: Option>, default_image: String, store: Arc, @@ -625,8 +586,6 @@ impl ComputeRuntime { async fn from_driver( driver_name: String, driver: SharedComputeDriver, - shutdown_cleanup: Option>, - startup_starter: Option>, driver_process: Option>, store: Arc, sandbox_index: SandboxIndex, @@ -711,8 +670,6 @@ impl ComputeRuntime { Ok(Self { driver: TracedDriver::new(driver, driver_name), driver_info, - shutdown_cleanup, - startup_starter, driver_process, default_image, store, @@ -762,19 +719,14 @@ impl ComputeRuntime { tracing_log_bus: TracingLogBus, supervisor_sessions: Arc, ) -> Result { - let driver = Arc::new( + let driver: SharedComputeDriver = Arc::new( DockerComputeDriver::new(&config, &docker_config) .await .map_err(|err| ComputeError::Message(err.to_string()))?, ); - let shutdown_cleanup: Arc = driver.clone(); - let startup_starter: Arc = driver.clone(); - let driver: SharedComputeDriver = driver; Self::from_driver( ComputeDriverKind::Docker.as_str().to_string(), driver, - Some(shutdown_cleanup), - Some(startup_starter), None, store, sandbox_index, @@ -804,8 +756,6 @@ impl ComputeRuntime { ComputeDriverKind::Kubernetes.as_str().to_string(), driver, None, - None, - None, store, sandbox_index, sandbox_watch_bus, @@ -828,8 +778,6 @@ impl ComputeRuntime { Self::from_driver( endpoint.name, driver, - None, - None, endpoint.driver_process, store, sandbox_index, @@ -857,8 +805,6 @@ impl ComputeRuntime { ComputeDriverKind::Podman.as_str().to_string(), driver, None, - None, - None, store, sandbox_index, sandbox_watch_bus, @@ -2070,54 +2016,162 @@ impl ComputeRuntime { } pub async fn cleanup_on_shutdown(&self) -> Result<(), String> { - let cleanup_result = match &self.shutdown_cleanup { - Some(cleanup) => cleanup.cleanup_on_shutdown().await, - None => Ok(()), - }; + let stop_result = self.stop_persisted_sandboxes_on_shutdown().await; + #[cfg(unix)] - let process_result = match &self.driver_process { - Some(process) => process.shutdown().await, - None => Ok(()), + let process_result = if let Some(process) = &self.driver_process { + process.shutdown().await + } else { + Ok(()) }; - cleanup_result?; - #[cfg(unix)] - process_result?; - Ok(()) + #[cfg(not(unix))] + let process_result: Result<(), String> = Ok(()); + + match (stop_result, process_result) { + (Ok(()), Ok(())) => Ok(()), + (Err(stop_err), Ok(())) => Err(stop_err), + (Ok(()), Err(process_err)) => Err(process_err), + (Err(stop_err), Err(process_err)) => Err(format!( + "{stop_err}; managed driver process shutdown failed: {process_err}" + )), + } + } + + /// Stop local compute during graceful gateway shutdown without changing + /// persisted lifecycle intent. + /// + /// An explicit sandbox stop persists `Stopped`; gateway shutdown does not. + /// Docker, Podman, and VM compute is stopped through the same public driver + /// RPC and restarted from the retained running-intent phase on gateway + /// startup. Kubernetes compute remains cluster-owned and is excluded. + async fn stop_persisted_sandboxes_on_shutdown(&self) -> Result<(), String> { + if !matches!( + self.driver_kind(), + Some(ComputeDriverKind::Docker | ComputeDriverKind::Podman | ComputeDriverKind::Vm) + ) { + return Ok(()); + } + + let sandbox_ids = self.list_persisted_sandbox_ids("gateway shutdown").await?; + + let outcomes = futures::stream::iter(sandbox_ids) + .map(|sandbox_id| async move { + let _lifecycle_guard = self.lifecycle_gates.lock_for(&sandbox_id).await; + let sandbox = match self.store.get_message::(&sandbox_id).await { + Ok(Some(sandbox)) => sandbox, + Ok(None) => return (0usize, 0usize), + Err(err) => { + warn!( + sandbox_id, + error = %err, + "Failed to re-read sandbox during gateway shutdown" + ); + return (0, 1); + } + }; + + let phase = + SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); + if !sandbox_phase_should_be_running(phase) { + return (0, 0); + } + + let sandbox_name = sandbox.object_name().to_string(); + match self + .driver + .call("driver.stop_sandbox", Some(&sandbox_id), |driver| { + let sandbox_id = sandbox_id.clone(); + let sandbox_name = sandbox_name.clone(); + async move { + driver + .stop_sandbox(Request::new(StopSandboxRequest { + sandbox_id, + sandbox_name, + })) + .await + } + }) + .await + { + Ok(_) => { + info!( + sandbox_id, + sandbox_name, + ?phase, + "Stopped sandbox during gateway shutdown" + ); + (1, 0) + } + Err(err) => { + warn!( + sandbox_id, + sandbox_name, + error = %err, + "Failed to stop sandbox during gateway shutdown" + ); + (0, 1) + } + } + }) + .buffer_unordered(SHUTDOWN_STOP_CONCURRENCY) + .collect::>() + .await; + let (stopped, failed) = outcomes + .into_iter() + .fold((0usize, 0usize), |(stopped, failed), outcome| { + (stopped + outcome.0, failed + outcome.1) + }); + + if stopped > 0 || failed > 0 { + info!(stopped, failed, "Sandbox shutdown stop sweep complete"); + } + if failed > 0 { + Err(format!( + "failed to stop {failed} sandbox(es) during gateway shutdown" + )) + } else { + Ok(()) + } } - /// Start sandboxes whose store records say they should be running. - /// Drivers that do not auto-restart compute resources across gateway - /// restarts (currently only Docker) implement `StartupSandboxStarter`. For - /// each sandbox in the store whose phase is not `Deleting` or - /// `Error`, we ask the driver to start the underlying resource. If - /// the driver reports that the resource no longer exists or fails to - /// start, the sandbox is moved to the `Error` phase so the failure - /// surfaces in the UI. + /// Reconcile running intent for local compute after a gateway restart. + /// + /// Docker and Podman resources can outlive the gateway but may have been + /// stopped by their runtime, while VM compute exits with its gateway-owned + /// driver process. `StartSandbox` is idempotent for all three drivers, so + /// call it for every persisted phase that requires running compute. Stable + /// stopped, deleting, and error states are deliberately left alone. /// /// Should be called once at gateway startup, before watchers spawn, /// so the watch loop sees the post-start state on its first poll. pub async fn start_persisted_sandboxes(&self) -> Result<(), String> { self.recover_persisted_lifecycle_transitions().await?; - let Some(startup_hook) = &self.startup_starter else { + if !matches!( + self.driver_kind(), + Some(ComputeDriverKind::Docker | ComputeDriverKind::Podman | ComputeDriverKind::Vm) + ) { return Ok(()); - }; + } - let records = self - .store - .list_by_type(Sandbox::object_type(), 1000, 0) - .await - .map_err(|e| e.to_string())?; + let sandbox_ids = self.list_persisted_sandbox_ids("gateway startup").await?; let mut started = 0usize; let mut missing = 0usize; let mut failed = 0usize; - for record in records { - let sandbox = match Sandbox::decode(record.payload.as_slice()) { - Ok(sandbox) => sandbox, + for sandbox_id in sandbox_ids { + let _lifecycle_guard = self.lifecycle_gates.lock_for(&sandbox_id).await; + let sandbox = match self.store.get_message::(&sandbox_id).await { + Ok(Some(sandbox)) => sandbox, + Ok(None) => continue, Err(err) => { - warn!(error = %err, "Failed to decode sandbox record during gateway startup"); + warn!( + sandbox_id, + error = %err, + "Failed to re-read sandbox during gateway startup" + ); + failed += 1; continue; } }; @@ -2127,11 +2181,24 @@ impl ComputeRuntime { continue; } - match startup_hook - .start_sandbox(sandbox.object_id(), sandbox.object_name()) + let sandbox_name = sandbox.object_name().to_string(); + match self + .driver + .call("driver.start_sandbox", Some(&sandbox_id), |driver| { + let sandbox_id = sandbox_id.clone(); + let sandbox_name = sandbox_name.clone(); + async move { + driver + .start_sandbox(Request::new(StartSandboxRequest { + sandbox_id, + sandbox_name, + })) + .await + } + }) .await { - Ok(true) => { + Ok(_) => { info!( sandbox_id = %sandbox.object_id(), sandbox_name = %sandbox.object_name(), @@ -2140,7 +2207,7 @@ impl ComputeRuntime { ); started += 1; } - Ok(false) => { + Err(err) if err.code() == Code::NotFound => { // Backend resource is gone but the store still // remembers the sandbox. Mark Error so the UI // surfaces the inconsistency; the reconcile loop @@ -2154,7 +2221,7 @@ impl ComputeRuntime { self.mark_sandbox_error( &sandbox, "BackendResourceMissing", - "Sandbox container disappeared while the gateway was offline", + "Sandbox compute resource disappeared while the gateway was offline", ) .await; missing += 1; @@ -2169,7 +2236,10 @@ impl ComputeRuntime { self.mark_sandbox_error( &sandbox, "StartFailed", - &format!("Failed to start sandbox during gateway startup: {err}"), + &format!( + "Failed to start sandbox during gateway startup: {}", + err.message() + ), ) .await; failed += 1; @@ -2189,16 +2259,20 @@ impl ComputeRuntime { } async fn recover_persisted_lifecycle_transitions(&self) -> Result<(), String> { - let records = self - .store - .list_by_type(Sandbox::object_type(), 1000, 0) - .await - .map_err(|e| e.to_string())?; - for record in records { - let sandbox = match Sandbox::decode(record.payload.as_slice()) { - Ok(sandbox) => sandbox, + let sandbox_ids = self + .list_persisted_sandbox_ids("lifecycle recovery") + .await?; + for sandbox_id in sandbox_ids { + let _lifecycle_guard = self.lifecycle_gates.lock_for(&sandbox_id).await; + let sandbox = match self.store.get_message::(&sandbox_id).await { + Ok(Some(sandbox)) => sandbox, + Ok(None) => continue, Err(err) => { - warn!(error = %err, "Failed to decode sandbox during lifecycle recovery"); + warn!( + sandbox_id, + error = %err, + "Failed to re-read sandbox during lifecycle recovery" + ); continue; } }; @@ -2285,6 +2359,28 @@ impl ComputeRuntime { Ok(()) } + async fn list_persisted_sandbox_ids(&self, operation: &str) -> Result, String> { + let mut sandbox_ids = Vec::new(); + let mut offset = 0u32; + loop { + let records = self + .store + .list_by_type(Sandbox::object_type(), LIFECYCLE_SWEEP_PAGE_SIZE, offset) + .await + .map_err(|err| format!("failed to list sandboxes for {operation}: {err}"))?; + let page_len = u32::try_from(records.len()) + .map_err(|_| format!("sandbox page size overflow during {operation}"))?; + sandbox_ids.extend(records.into_iter().map(|record| record.id)); + if page_len < LIFECYCLE_SWEEP_PAGE_SIZE { + break; + } + offset = offset + .checked_add(page_len) + .ok_or_else(|| format!("sandbox pagination offset overflow during {operation}"))?; + } + Ok(sandbox_ids) + } + async fn mark_sandbox_error(&self, sandbox: &Sandbox, reason: &str, message: &str) { let _guard = self.sync_lock.lock().await; let sandbox_id = sandbox.object_id().to_string(); @@ -3994,8 +4090,6 @@ pub async fn new_test_runtime_with_driver( driver_name: driver_name.to_string(), driver_version: "test".to_string(), }, - shutdown_cleanup: None, - startup_starter: None, driver_process: None, default_image: "openshell/sandbox:test".to_string(), store, @@ -4306,6 +4400,7 @@ mod tests { #[derive(Clone)] enum ControlledLifecycleOutcome { Ok, + NotFound, Error(&'static str), } @@ -4323,12 +4418,14 @@ mod tests { stop_release: Semaphore, stop_blocked: AtomicBool, stop_calls: AtomicUsize, + stop_requests: TestMutex>, stop_outcome: TestMutex, start_started: Notify, start_finished: Notify, start_release: Semaphore, start_blocked: AtomicBool, start_calls: AtomicUsize, + start_requests: TestMutex>, start_outcome: TestMutex, get_started: Notify, get_release: Semaphore, @@ -4353,12 +4450,14 @@ mod tests { stop_release: Semaphore::new(0), stop_blocked: AtomicBool::new(false), stop_calls: AtomicUsize::new(0), + stop_requests: TestMutex::new(Vec::new()), stop_outcome: TestMutex::new(ControlledLifecycleOutcome::Ok), start_started: Notify::new(), start_finished: Notify::new(), start_release: Semaphore::new(0), start_blocked: AtomicBool::new(false), start_calls: AtomicUsize::new(0), + start_requests: TestMutex::new(Vec::new()), start_outcome: TestMutex::new(ControlledLifecycleOutcome::Ok), get_started: Notify::new(), get_release: Semaphore::new(0), @@ -4432,10 +4531,24 @@ mod tests { self.stop_calls.load(Ordering::SeqCst) } + fn stop_requests(&self) -> Vec<(String, String)> { + self.stop_requests + .lock() + .expect("stop requests lock poisoned") + .clone() + } + fn start_calls(&self) -> usize { self.start_calls.load(Ordering::SeqCst) } + fn start_requests(&self) -> Vec<(String, String)> { + self.start_requests + .lock() + .expect("start requests lock poisoned") + .clone() + } + fn send_event(&self, event: WatchSandboxesEvent) { self.watch_tx .send(Ok(event)) @@ -4525,8 +4638,13 @@ mod tests { async fn stop_sandbox( &self, - _request: Request, + request: Request, ) -> Result, Status> { + let request = request.into_inner(); + self.stop_requests + .lock() + .expect("stop requests lock poisoned") + .push((request.sandbox_id, request.sandbox_name)); self.stop_calls.fetch_add(1, Ordering::SeqCst); self.stop_started.notify_one(); if self.stop_blocked.load(Ordering::SeqCst) { @@ -4544,14 +4662,20 @@ mod tests { .clone(); match outcome { ControlledLifecycleOutcome::Ok => Ok(tonic::Response::new(StopSandboxResponse {})), + ControlledLifecycleOutcome::NotFound => Err(Status::not_found("sandbox not found")), ControlledLifecycleOutcome::Error(message) => Err(Status::internal(message)), } } async fn start_sandbox( &self, - _request: Request, + request: Request, ) -> Result, Status> { + let request = request.into_inner(); + self.start_requests + .lock() + .expect("start requests lock poisoned") + .push((request.sandbox_id, request.sandbox_name)); self.start_calls.fetch_add(1, Ordering::SeqCst); self.start_started.notify_one(); if self.start_blocked.load(Ordering::SeqCst) { @@ -4569,6 +4693,7 @@ mod tests { .clone(); match outcome { ControlledLifecycleOutcome::Ok => Ok(tonic::Response::new(StartSandboxResponse {})), + ControlledLifecycleOutcome::NotFound => Err(Status::not_found("sandbox not found")), ControlledLifecycleOutcome::Error(message) => Err(Status::internal(message)), } } @@ -4631,23 +4756,21 @@ mod tests { } async fn test_runtime(driver: SharedComputeDriver) -> ComputeRuntime { - test_runtime_with_start(driver, None).await + test_runtime_for_driver(driver, "test-driver").await } - async fn test_runtime_with_start( + async fn test_runtime_for_driver( driver: SharedComputeDriver, - startup_starter: Option>, + driver_name: &str, ) -> ComputeRuntime { let store = Arc::new(Store::connect("sqlite::memory:").await.unwrap()); ComputeRuntime { driver: TracedDriver::new(driver, "test-driver".to_string()), driver_info: ComputeDriverInfoSnapshot { - name: "test-driver".to_string(), - driver_name: "test-driver".to_string(), + name: driver_name.to_string(), + driver_name: driver_name.to_string(), driver_version: "test".to_string(), }, - shutdown_cleanup: None, - startup_starter, driver_process: None, default_image: "openshell/sandbox:test".to_string(), store, @@ -8039,56 +8162,205 @@ mod tests { )); } - #[derive(Default)] - struct RecordingStart { - calls: Mutex>, - results: Mutex>>, + #[tokio::test] + async fn shutdown_stops_running_intent_without_changing_persisted_phase() { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), "docker").await; + + for (id, name, phase) in [ + ("sb-unspecified", "unspecified", SandboxPhase::Unspecified), + ("sb-prov", "prov", SandboxPhase::Provisioning), + ("sb-ready", "ready", SandboxPhase::Ready), + ("sb-starting", "starting", SandboxPhase::Starting), + ("sb-unknown", "unknown", SandboxPhase::Unknown), + ("sb-stopping", "stopping", SandboxPhase::Stopping), + ("sb-stopped", "stopped", SandboxPhase::Stopped), + ("sb-deleting", "deleting", SandboxPhase::Deleting), + ("sb-error", "error", SandboxPhase::Error), + ] { + runtime + .store + .put_message(&sandbox_record(id, name, phase)) + .await + .unwrap(); + } + + runtime + .stop_persisted_sandboxes_on_shutdown() + .await + .unwrap(); + + let mut called_ids = driver + .stop_requests() + .into_iter() + .map(|(id, _)| id) + .collect::>(); + called_ids.sort(); + assert_eq!( + called_ids, + vec![ + "sb-prov".to_string(), + "sb-ready".to_string(), + "sb-starting".to_string(), + "sb-unknown".to_string(), + "sb-unspecified".to_string(), + ] + ); + + let stored = runtime + .store + .get_message::("sb-ready") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready, + "gateway shutdown must retain logical running intent" + ); } - impl RecordingStart { - async fn set_result(&self, sandbox_id: &str, result: Result) { - self.results - .lock() + #[tokio::test] + async fn shutdown_stop_sweep_continues_after_driver_errors() { + let driver = ControlledDriver::new(); + driver.set_stop_outcome(ControlledLifecycleOutcome::Error("runtime angry")); + let runtime = test_runtime_for_driver(driver.clone(), "podman").await; + for (id, name) in [("sb-1", "one"), ("sb-2", "two")] { + runtime + .store + .put_message(&sandbox_record(id, name, SandboxPhase::Ready)) + .await + .unwrap(); + } + + let err = runtime + .stop_persisted_sandboxes_on_shutdown() + .await + .unwrap_err(); + + assert!(err.contains("failed to stop 2 sandbox(es)")); + assert_eq!(driver.stop_calls(), 2); + } + + #[tokio::test] + async fn shutdown_stop_sweep_bounds_driver_concurrency() { + let driver = ControlledDriver::new(); + driver.block_stop(); + let runtime = test_runtime_for_driver(driver.clone(), "docker").await; + for index in 0..=SHUTDOWN_STOP_CONCURRENCY { + runtime + .store + .put_message(&sandbox_record( + &format!("sb-{index}"), + &format!("sandbox-{index}"), + SandboxPhase::Ready, + )) .await - .insert(sandbox_id.to_string(), result); + .unwrap(); } - async fn calls(&self) -> Vec<(String, String)> { - self.calls.lock().await.clone() + let sweep_runtime = runtime.clone(); + let sweep = + tokio::spawn(async move { sweep_runtime.stop_persisted_sandboxes_on_shutdown().await }); + tokio::time::timeout(Duration::from_secs(1), async { + while driver.stop_calls() < SHUTDOWN_STOP_CONCURRENCY { + tokio::task::yield_now().await; + } + }) + .await + .expect("shutdown sweep did not fill its concurrency window"); + assert_eq!(driver.stop_calls(), SHUTDOWN_STOP_CONCURRENCY); + + for _ in 0..=SHUTDOWN_STOP_CONCURRENCY { + driver.release_stop(); } + tokio::time::timeout(Duration::from_secs(1), sweep) + .await + .expect("shutdown sweep did not finish") + .unwrap() + .unwrap(); + assert_eq!(driver.stop_calls(), SHUTDOWN_STOP_CONCURRENCY + 1); } - #[tonic::async_trait] - impl StartupSandboxStarter for RecordingStart { - async fn start_sandbox( - &self, - sandbox_id: &str, - sandbox_name: &str, - ) -> Result { - self.calls - .lock() + #[tokio::test] + async fn shutdown_stop_sweep_rechecks_intent_after_acquiring_gate() { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), "docker").await; + runtime + .store + .put_message(&sandbox_record("sb-1", "sandbox", SandboxPhase::Ready)) + .await + .unwrap(); + + let gate = runtime.lifecycle_gates.gate_for("sb-1"); + let guard = gate.clone().lock_owned().await; + let sweep_runtime = runtime.clone(); + let sweep = + tokio::spawn(async move { sweep_runtime.stop_persisted_sandboxes_on_shutdown().await }); + tokio::time::timeout(Duration::from_secs(1), async { + while Arc::strong_count(&gate) < 3 { + tokio::task::yield_now().await; + } + }) + .await + .expect("shutdown sweep did not wait on the lifecycle gate"); + + runtime + .store + .put_message(&sandbox_record("sb-1", "sandbox", SandboxPhase::Stopped)) + .await + .unwrap(); + drop(guard); + tokio::time::timeout(Duration::from_secs(1), sweep) + .await + .expect("shutdown sweep did not finish") + .unwrap() + .unwrap(); + assert_eq!(driver.stop_calls(), 0); + } + + #[tokio::test] + async fn shutdown_stop_sweep_runs_for_each_local_driver_only() { + for (driver_name, expected_calls) in [ + ("docker", 1), + ("podman", 1), + ("vm", 1), + ("kubernetes", 0), + ("extension", 0), + ] { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), driver_name).await; + runtime + .store + .put_message(&sandbox_record("sb-1", "sandbox", SandboxPhase::Ready)) .await - .push((sandbox_id.to_string(), sandbox_name.to_string())); - self.results - .lock() + .unwrap(); + + runtime + .stop_persisted_sandboxes_on_shutdown() .await - .get(sandbox_id) - .cloned() - .unwrap_or(Ok(true)) + .unwrap(); + + assert_eq!( + driver.stop_calls(), + expected_calls, + "unexpected shutdown behavior for {driver_name}" + ); } } #[tokio::test] async fn start_persisted_sandboxes_starts_running_phases() { - let start = Arc::new(RecordingStart::default()); - let runtime = - test_runtime_with_start(Arc::new(TestDriver::default()), Some(start.clone())).await; + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), "docker").await; for (id, name, phase) in [ ("sb-unspecified", "unspecified", SandboxPhase::Unspecified), ("sb-prov", "prov", SandboxPhase::Provisioning), ("sb-ready", "ready", SandboxPhase::Ready), ("sb-unknown", "unknown", SandboxPhase::Unknown), + ("sb-stopping", "stopping", SandboxPhase::Stopping), + ("sb-stopped", "stopped", SandboxPhase::Stopped), ("sb-deleting", "deleting", SandboxPhase::Deleting), ("sb-error", "error", SandboxPhase::Error), ] { @@ -8098,9 +8370,8 @@ mod tests { runtime.start_persisted_sandboxes().await.unwrap(); - let mut called_ids = start - .calls() - .await + let mut called_ids = driver + .start_requests() .into_iter() .map(|(id, _)| id) .collect::>(); @@ -8116,12 +8387,74 @@ mod tests { ); } + #[tokio::test] + async fn startup_sweep_rechecks_intent_after_acquiring_gate() { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), "docker").await; + runtime + .store + .put_message(&sandbox_record("sb-1", "sandbox", SandboxPhase::Ready)) + .await + .unwrap(); + + let gate = runtime.lifecycle_gates.gate_for("sb-1"); + let guard = gate.clone().lock_owned().await; + let sweep_runtime = runtime.clone(); + let sweep = tokio::spawn(async move { sweep_runtime.start_persisted_sandboxes().await }); + tokio::time::timeout(Duration::from_secs(1), async { + while Arc::strong_count(&gate) < 3 { + tokio::task::yield_now().await; + } + }) + .await + .expect("startup sweep did not wait on the lifecycle gate"); + + runtime + .store + .put_message(&sandbox_record("sb-1", "sandbox", SandboxPhase::Deleting)) + .await + .unwrap(); + drop(guard); + tokio::time::timeout(Duration::from_secs(1), sweep) + .await + .expect("startup sweep did not finish") + .unwrap() + .unwrap(); + assert_eq!(driver.start_calls(), 0); + } + + #[tokio::test] + async fn lifecycle_sweeps_page_through_all_persisted_sandboxes() { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), "docker").await; + let sandbox_count = LIFECYCLE_SWEEP_PAGE_SIZE + 1; + for index in 0..sandbox_count { + runtime + .store + .put_message(&sandbox_record( + &format!("sb-{index:04}"), + &format!("sandbox-{index:04}"), + SandboxPhase::Ready, + )) + .await + .unwrap(); + } + + runtime.start_persisted_sandboxes().await.unwrap(); + assert_eq!(driver.start_calls(), sandbox_count as usize); + + runtime + .stop_persisted_sandboxes_on_shutdown() + .await + .unwrap(); + assert_eq!(driver.stop_calls(), sandbox_count as usize); + } + #[tokio::test] async fn start_persisted_sandboxes_marks_missing_backend_as_error() { - let start = Arc::new(RecordingStart::default()); - start.set_result("sb-1", Ok(false)).await; - let runtime = - test_runtime_with_start(Arc::new(TestDriver::default()), Some(start.clone())).await; + let driver = ControlledDriver::new(); + driver.set_start_outcome(ControlledLifecycleOutcome::NotFound); + let runtime = test_runtime_for_driver(driver, "podman").await; let sandbox = sandbox_record("sb-1", "missing", SandboxPhase::Ready); runtime.store.put_message(&sandbox).await.unwrap(); @@ -8144,16 +8477,14 @@ mod tests { .and_then(|s| s.conditions.iter().find(|c| c.r#type == "Ready")) .expect("Ready condition present"); assert_eq!(ready.reason, "BackendResourceMissing"); + assert!(ready.message.contains("compute resource disappeared")); } #[tokio::test] async fn start_persisted_sandboxes_marks_failed_start_as_error() { - let start = Arc::new(RecordingStart::default()); - start - .set_result("sb-1", Err("docker daemon angry".to_string())) - .await; - let runtime = - test_runtime_with_start(Arc::new(TestDriver::default()), Some(start.clone())).await; + let driver = ControlledDriver::new(); + driver.set_start_outcome(ControlledLifecycleOutcome::Error("runtime angry")); + let runtime = test_runtime_for_driver(driver, "vm").await; let sandbox = sandbox_record("sb-1", "broken", SandboxPhase::Provisioning); runtime.store.put_message(&sandbox).await.unwrap(); @@ -8176,27 +8507,42 @@ mod tests { .and_then(|s| s.conditions.iter().find(|c| c.r#type == "Ready")) .expect("Ready condition present"); assert_eq!(ready.reason, "StartFailed"); - assert!(ready.message.contains("docker daemon angry")); + assert!(ready.message.contains("runtime angry")); } #[tokio::test] - async fn start_persisted_sandboxes_is_noop_without_start_hook() { - let runtime = test_runtime(Arc::new(TestDriver::default())).await; - let sandbox = sandbox_record("sb-1", "anywhere", SandboxPhase::Ready); - runtime.store.put_message(&sandbox).await.unwrap(); + async fn start_persisted_sandboxes_runs_for_each_local_driver() { + for driver_name in ["docker", "podman", "vm"] { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), driver_name).await; + let sandbox = sandbox_record("sb-1", "local", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); - runtime.start_persisted_sandboxes().await.unwrap(); + runtime.start_persisted_sandboxes().await.unwrap(); - let stored = runtime - .store - .get_message::("sb-1") - .await - .unwrap() - .unwrap(); - assert_eq!( - SandboxPhase::try_from(stored.phase()).unwrap(), - SandboxPhase::Ready - ); + assert_eq!( + driver.start_requests(), + vec![("sb-1".to_string(), "local".to_string())], + "{driver_name} should reconcile persisted running intent" + ); + } + } + + #[tokio::test] + async fn start_persisted_sandboxes_skips_kubernetes_and_extension_drivers() { + for driver_name in ["kubernetes", "extension"] { + let driver = ControlledDriver::new(); + let runtime = test_runtime_for_driver(driver.clone(), driver_name).await; + let sandbox = sandbox_record("sb-1", "remote", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + + runtime.start_persisted_sandboxes().await.unwrap(); + + assert!( + driver.start_requests().is_empty(), + "{driver_name} should not receive stable-running startup reconciliation" + ); + } } #[test] @@ -8251,8 +8597,6 @@ mod tests { "test-driver".to_string(), Arc::new(TestDriver::default()), None, - None, - None, store, SandboxIndex::new(), SandboxWatchBus::new(), diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index a6031a9bc6..979e372084 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -716,10 +716,9 @@ pub(crate) async fn run_server( let state = Arc::new(state); - // Start sandboxes that were stopped during the previous gateway - // shutdown so the running compute state matches the persisted store. - // Runs before watchers spawn so the watch loop sees the post-start - // snapshot on its first poll. + // Reconcile local-driver running intent before watchers spawn so their + // first snapshots observe the post-start backend state. Explicitly stopped + // sandboxes remain stopped. ensure_default_workspace(&store).await?; let gateway_listeners = bind_gateway_listeners( diff --git a/docs/reference/sandbox-compute-drivers.mdx b/docs/reference/sandbox-compute-drivers.mdx index 897e780c39..8405c1f3ba 100644 --- a/docs/reference/sandbox-compute-drivers.mdx +++ b/docs/reference/sandbox-compute-drivers.mdx @@ -18,6 +18,13 @@ Delete remains independent and removes compute plus driver-owned persistent state. While a sandbox is stopped, gateway access paths and exposed services remain unavailable. +Restarting the gateway preserves this intent. The gateway does not stop Docker +or Podman containers during shutdown. At startup it sends idempotent start +requests for Docker, Podman, and MicroVM sandboxes that were intended to run; +already-running resources are unchanged, retained stopped compute is restarted, +and explicitly stopped sandboxes remain stopped. Kubernetes workloads continue +running independently of the gateway process. + ## Configure a Compute Driver Configure the compute driver on the gateway. Current releases accept one driver per gateway. Set `compute_drivers` in the gateway TOML file: @@ -141,7 +148,10 @@ Select Docker with `compute_drivers = ["docker"]` in `[openshell.gateway]`. Conf Stop stops the existing Docker container without removing its writable layer or attached volumes. Start starts that same container. A durably stopped container stays stopped across gateway restart, and delete remains -responsible for removing it. +responsible for removing it. Graceful gateway shutdown stops running-intent +Docker containers through the driver RPC without recording an explicit +sandbox stop. On startup, the gateway reconciles that retained intent with an +idempotent start request. Explicitly stopped sandboxes remain stopped. For GPU-backed Docker sandboxes, configure Docker CDI before starting the gateway so OpenShell can detect the daemon capability. @@ -213,7 +223,11 @@ Podman sandboxes default to a 45-second graceful stop window before Podman escal Stop stops the existing Podman container while retaining its named workspace volume and driver-owned secrets. Start starts the same container. Delete is -the operation that removes the container and named volume. +the operation that removes the container and named volume. Graceful gateway +shutdown stops running-intent Podman containers through the driver RPC without +recording an explicit sandbox stop. On startup, the gateway reconciles that +retained intent with an idempotent start request while leaving explicitly +stopped sandboxes alone. For proxy-required networks, the Podman driver also accepts the corporate egress proxy keys `https_proxy`, `no_proxy`, `proxy_auth_file`, `proxy_auth_allow_insecure`, and `proxy_connect_by_hostname`. The supervisor chains policy-approved TLS tunnels through the proxy with HTTP CONNECT instead of dialing destinations directly. See the [Gateway Configuration File](./gateway-config) reference for the full contract, including the cleartext-credential acknowledgement and the validated-IP CONNECT behavior. @@ -285,7 +299,7 @@ The VM driver boots a cached immutable bootstrap ext4 root disk. When the reques VM sandbox creation follows the same progress model as Kubernetes-backed sandboxes. The gateway accepts the sandbox, then the VM driver publishes watch events while it resolves the image, prepares or reuses the bootstrap and prepared image caches, creates the writable overlay, and starts the VM launcher. -On gateway restart, the gateway starts a fresh VM driver process. The driver scans its state directory for accepted sandbox launch records, restarts those VMs, and reuses each sandbox's existing `overlay.ext4` so files written inside the sandbox remain available after the supervisor reconnects. +On graceful gateway shutdown, the gateway stops running-intent VMs through the driver RPC while retaining their launch records and writable overlays. On restart, the gateway starts a fresh VM driver process and reconciles the retained intent through the same idempotent start request used by the local container drivers. Running-intent VMs restart with their existing `overlay.ext4`, while explicitly stopped VMs remain stopped. Stopped VM state directories contain a marker that prevents startup from launching the VM. The driver retains `sandbox.pb`, `overlay.ext4`, and extension diff --git a/e2e/rust/src/harness/cli.rs b/e2e/rust/src/harness/cli.rs index 53392d752c..1c63ea392b 100644 --- a/e2e/rust/src/harness/cli.rs +++ b/e2e/rust/src/harness/cli.rs @@ -4,6 +4,7 @@ //! Shared CLI helpers for e2e tests that need to invoke `openshell` commands //! and poll for readiness. +use std::future::Future; use std::process::Stdio; use std::time::{Duration, Instant}; @@ -12,6 +13,24 @@ use tokio::time::sleep; use super::binary::openshell_cmd; use super::output::strip_ansi; +async fn poll_with_diagnostics(timeout: Duration, mut attempt: F) -> Result<(), String> +where + F: FnMut() -> Fut, + Fut: Future, +{ + let start = Instant::now(); + loop { + let (ready, last_output) = attempt().await; + if ready { + return Ok(()); + } + if start.elapsed() > timeout { + return Err(last_output); + } + sleep(Duration::from_secs(2)).await; + } +} + pub async fn run_cli(args: &[&str]) -> (String, i32) { let mut cmd = openshell_cmd(); cmd.args(args).stdout(Stdio::piped()).stderr(Stdio::piped()); @@ -25,30 +44,23 @@ pub async fn run_cli(args: &[&str]) -> (String, i32) { } pub async fn wait_for_healthy(timeout: Duration) -> Result<(), String> { - let start = Instant::now(); - let mut last_output: String; - - loop { + poll_with_diagnostics(timeout, || async { let (output, code) = run_cli(&["status"]).await; let clean = strip_ansi(&output); let lower = clean.to_lowercase(); - if code == 0 + let ready = code == 0 && (lower.contains("healthy") || lower.contains("running") - || lower.contains("connected")) - { - return Ok(()); - } - last_output = clean; - - if start.elapsed() > timeout { - return Err(format!( - "gateway did not become healthy within {}s. Last output:\n{last_output}", - timeout.as_secs() - )); - } - sleep(Duration::from_secs(2)).await; - } + || lower.contains("connected")); + (ready, clean) + }) + .await + .map_err(|last_output| { + format!( + "gateway did not become healthy within {}s. Last output:\n{last_output}", + timeout.as_secs() + ) + }) } pub async fn sandbox_names() -> Result, String> { @@ -66,42 +78,66 @@ pub async fn sandbox_names() -> Result, String> { .collect()) } +pub async fn wait_for_sandbox_phase( + sandbox_name: &str, + expected_phase: &str, + timeout: Duration, +) -> Result<(), String> { + poll_with_diagnostics(timeout, || async { + let (output, code) = run_cli(&["sandbox", "get", sandbox_name, "--output", "json"]).await; + let clean = strip_ansi(&output); + let phase_matches = serde_json::from_str::(&clean) + .ok() + .and_then(|value| { + value + .get("phase") + .and_then(|phase| phase.as_str()) + .map(str::to_owned) + }) + .is_some_and(|phase| phase == expected_phase); + (code == 0 && phase_matches, clean) + }) + .await + .map_err(|last_output| { + format!( + "sandbox '{sandbox_name}' did not reach phase '{expected_phase}' within {}s. Last output:\n{last_output}", + timeout.as_secs() + ) + }) +} + pub async fn wait_for_sandbox_exec_contains( sandbox_name: &str, command: &[&str], expected: &str, timeout: Duration, ) -> Result<(), String> { - let start = Instant::now(); - let mut last_output: String; - - loop { + poll_with_diagnostics(timeout, || async { let mut cmd = openshell_cmd(); cmd.args(["sandbox", "exec", "--name", sandbox_name, "--no-tty", "--"]) .args(command) .stdout(Stdio::piped()) .stderr(Stdio::piped()); - match cmd.output().await { + let (ready, last_output) = match cmd.output().await { Ok(output) => { let stdout = String::from_utf8_lossy(&output.stdout); let stderr = String::from_utf8_lossy(&output.stderr); - last_output = strip_ansi(&format!("{stdout}{stderr}")); - if output.status.success() && last_output.contains(expected) { - return Ok(()); - } - } - Err(err) => { - last_output = format!("failed to spawn openshell sandbox exec: {err}"); + let clean = strip_ansi(&format!("{stdout}{stderr}")); + (output.status.success() && clean.contains(expected), clean) } - } - - if start.elapsed() > timeout { - return Err(format!( - "sandbox '{sandbox_name}' exec did not produce '{expected}' within {}s. Last output:\n{last_output}", - timeout.as_secs() - )); - } - sleep(Duration::from_secs(2)).await; - } + Err(err) => ( + false, + format!("failed to spawn openshell sandbox exec: {err}"), + ), + }; + (ready, last_output) + }) + .await + .map_err(|last_output| { + format!( + "sandbox '{sandbox_name}' exec did not produce '{expected}' within {}s. Last output:\n{last_output}", + timeout.as_secs() + ) + }) } diff --git a/e2e/rust/tests/gateway_start.rs b/e2e/rust/tests/gateway_start.rs index 3984cf1907..cca35e3d59 100644 --- a/e2e/rust/tests/gateway_start.rs +++ b/e2e/rust/tests/gateway_start.rs @@ -3,7 +3,7 @@ #![cfg(feature = "e2e")] -//! E2E coverage for starting Docker sandboxes after a standalone gateway restart. +//! E2E coverage for reconciling Docker sandboxes after a standalone gateway restart. //! //! This intentionally targets the Docker-driver gateway started by //! `e2e/with-docker-gateway.sh`. Existing-endpoint E2E runs do not own the @@ -13,7 +13,8 @@ use std::process::{Command, Stdio}; use std::time::Duration; use openshell_e2e::harness::cli::{ - sandbox_names, wait_for_healthy, wait_for_sandbox_exec_contains, + run_cli, sandbox_names, wait_for_healthy, wait_for_sandbox_exec_contains, + wait_for_sandbox_phase, }; use openshell_e2e::harness::gateway::ManagedGateway; use openshell_e2e::harness::sandbox::SandboxGuard; @@ -21,6 +22,7 @@ use tokio::time::sleep; const MANAGED_BY_LABEL_FILTER: &str = "label=openshell.ai/managed-by=openshell"; const READY_MARKER: &str = "gateway-start-ready"; +const STOPPED_READY_MARKER: &str = "gateway-start-stopped-ready"; const START_FILE: &str = "/sandbox/gateway-start-state"; const SANDBOX_NAMESPACE_LABEL: &str = "openshell.ai/sandbox-namespace"; const SANDBOX_NAME_LABEL: &str = "openshell.ai/sandbox-name"; @@ -117,7 +119,7 @@ async fn wait_for_container_running( } #[tokio::test] -async fn docker_gateway_restart_starts_running_sandbox() { +async fn docker_gateway_restart_preserves_running_and_stopped_intent() { let Some(gateway) = ManagedGateway::from_env().expect("load managed e2e gateway metadata") else { eprintln!("Skipping gateway start test: e2e gateway is not managed by this test run"); @@ -155,10 +157,29 @@ async fn docker_gateway_restart_starts_running_sandbox() { .await .expect("sandbox container should be running before gateway restart"); + let stopped_script = format!("echo {STOPPED_READY_MARKER}; while true; do sleep 1; done"); + let mut stopped_sandbox = + SandboxGuard::create_keep(&["sh", "-lc", &stopped_script], STOPPED_READY_MARKER) + .await + .expect("create Docker sandbox that will remain stopped"); + let (stop_output, stop_code) = run_cli(&["sandbox", "stop", &stopped_sandbox.name]).await; + assert_eq!(stop_code, 0, "sandbox stop should succeed:\n{stop_output}"); + wait_for_sandbox_phase(&stopped_sandbox.name, "Stopped", Duration::from_secs(30)) + .await + .expect("sandbox should be stopped before gateway restart"); + wait_for_container_running( + &namespace, + &stopped_sandbox.name, + false, + Duration::from_secs(30), + ) + .await + .expect("stopped Docker sandbox container should not be running"); + gateway.stop().expect("stop e2e gateway"); - wait_for_container_running(&namespace, &sandbox.name, false, Duration::from_secs(120)) + wait_for_container_running(&namespace, &sandbox.name, false, Duration::from_secs(30)) .await - .expect("gateway shutdown should stop managed Docker sandboxes"); + .expect("gateway shutdown should stop a running-intent Docker sandbox"); gateway.start().expect("restart e2e gateway"); wait_for_healthy(Duration::from_secs(120)) @@ -166,7 +187,18 @@ async fn docker_gateway_restart_starts_running_sandbox() { .expect("gateway should become healthy after restart"); wait_for_container_running(&namespace, &sandbox.name, true, Duration::from_secs(120)) .await - .expect("gateway startup should start the Docker sandbox container"); + .expect("gateway startup should restart the running-intent Docker sandbox container"); + wait_for_sandbox_phase(&stopped_sandbox.name, "Stopped", Duration::from_secs(120)) + .await + .expect("explicitly stopped Docker sandbox should remain stopped after restart"); + wait_for_container_running( + &namespace, + &stopped_sandbox.name, + false, + Duration::from_secs(30), + ) + .await + .expect("gateway startup should not start an explicitly stopped Docker sandbox"); let names = sandbox_names().await.expect("list sandboxes after restart"); assert!( @@ -185,4 +217,5 @@ async fn docker_gateway_restart_starts_running_sandbox() { .expect("sandbox should become ready again with its state preserved"); sandbox.cleanup().await; + stopped_sandbox.cleanup().await; } diff --git a/e2e/rust/tests/podman_gateway_start.rs b/e2e/rust/tests/podman_gateway_start.rs index e72769cb80..9a047980ba 100644 --- a/e2e/rust/tests/podman_gateway_start.rs +++ b/e2e/rust/tests/podman_gateway_start.rs @@ -3,28 +3,95 @@ #![cfg(feature = "e2e-podman")] -//! Podman-specific E2E coverage for starting sandboxes after a standalone -//! gateway restart. -//! -//! Unlike the Docker driver, Podman does not stop sandbox containers when the -//! gateway process exits — the containers keep running and the restarted -//! gateway re-adopts them. This test follows the `vm_gateway_start.rs` -//! pattern: verify sandbox survival at the application level without asserting -//! intermediate container-state transitions. +//! Podman-specific E2E coverage for stopping and starting sandboxes across a +//! standalone gateway restart. +use std::process::{Command, Stdio}; use std::time::Duration; use openshell_e2e::harness::cli::{ - sandbox_names, wait_for_healthy, wait_for_sandbox_exec_contains, + run_cli, sandbox_names, wait_for_healthy, wait_for_sandbox_exec_contains, + wait_for_sandbox_phase, }; use openshell_e2e::harness::gateway::ManagedGateway; use openshell_e2e::harness::sandbox::SandboxGuard; +use tokio::time::sleep; const READY_MARKER: &str = "podman-gateway-start-ready"; +const STOPPED_READY_MARKER: &str = "podman-gateway-start-stopped-ready"; const START_FILE: &str = "/sandbox/podman-gateway-start-state"; +const MANAGED_BY_LABEL_FILTER: &str = "label=openshell.managed=true"; +const SANDBOX_NAME_LABEL: &str = "openshell.ai/sandbox-name"; + +fn sandbox_container_running(sandbox_name: &str) -> Result { + let sandbox_name_filter = format!("label={SANDBOX_NAME_LABEL}={sandbox_name}"); + let output = Command::new("podman") + .args(["ps", "-aq", "--filter", MANAGED_BY_LABEL_FILTER, "--filter"]) + .arg(sandbox_name_filter) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .output() + .map_err(|err| format!("failed to run podman ps: {err}"))?; + if !output.status.success() { + return Err(format!( + "podman ps failed: {}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + )); + } + let ids = String::from_utf8_lossy(&output.stdout) + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .map(str::to_string) + .collect::>(); + let [container_id] = ids.as_slice() else { + return Err(format!( + "expected one Podman container for sandbox '{sandbox_name}', found {ids:?}" + )); + }; + let output = Command::new("podman") + .args(["inspect", "-f", "{{.State.Running}}", container_id]) + .output() + .map_err(|err| format!("failed to run podman inspect: {err}"))?; + if !output.status.success() { + return Err(format!( + "podman inspect failed: {}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + )); + } + match String::from_utf8_lossy(&output.stdout).trim() { + "true" => Ok(true), + "false" => Ok(false), + state => Err(format!("unexpected Podman running state: {state}")), + } +} + +async fn wait_for_container_running( + sandbox_name: &str, + expected: bool, + timeout: Duration, +) -> Result<(), String> { + let start = std::time::Instant::now(); + let mut last_state; + loop { + match sandbox_container_running(sandbox_name) { + Ok(running) if running == expected => return Ok(()), + Ok(running) => last_state = format!("running={running}"), + Err(err) => last_state = err, + } + if start.elapsed() > timeout { + return Err(format!( + "Podman container for '{sandbox_name}' did not reach running={expected}: {last_state}" + )); + } + sleep(Duration::from_secs(1)).await; + } +} #[tokio::test] -async fn podman_gateway_restart_starts_running_sandbox() { +async fn podman_gateway_restart_preserves_running_and_stopped_intent() { if std::env::var("OPENSHELL_E2E_DRIVER").as_deref() != Ok("podman") { eprintln!("Skipping Podman gateway start test: e2e driver is not podman"); return; @@ -47,6 +114,9 @@ async fn podman_gateway_restart_starts_running_sandbox() { let mut sandbox = SandboxGuard::create_keep(&["sh", "-lc", &script], READY_MARKER) .await .expect("create long-running Podman sandbox"); + wait_for_container_running(&sandbox.name, true, Duration::from_secs(60)) + .await + .expect("Podman sandbox container should initially be running"); let before_restart = sandbox .exec(&["cat", START_FILE]) @@ -57,11 +127,28 @@ async fn podman_gateway_restart_starts_running_sandbox() { "sandbox state was not written before restart:\n{before_restart}" ); + let stopped_script = format!("echo {STOPPED_READY_MARKER}; while true; do sleep 1; done"); + let mut stopped_sandbox = + SandboxGuard::create_keep(&["sh", "-lc", &stopped_script], STOPPED_READY_MARKER) + .await + .expect("create Podman sandbox that will remain stopped"); + let (stop_output, stop_code) = run_cli(&["sandbox", "stop", &stopped_sandbox.name]).await; + assert_eq!(stop_code, 0, "sandbox stop should succeed:\n{stop_output}"); + wait_for_sandbox_phase(&stopped_sandbox.name, "Stopped", Duration::from_secs(120)) + .await + .expect("Podman sandbox should be stopped before gateway restart"); + gateway.stop().expect("stop e2e gateway"); + wait_for_container_running(&sandbox.name, false, Duration::from_secs(60)) + .await + .expect("gateway shutdown should stop the running-intent Podman sandbox"); gateway.start().expect("restart e2e gateway"); wait_for_healthy(Duration::from_secs(120)) .await .expect("gateway should become healthy after restart"); + wait_for_container_running(&sandbox.name, true, Duration::from_secs(120)) + .await + .expect("gateway startup should restart the running-intent Podman sandbox"); let names = sandbox_names().await.expect("list sandboxes after restart"); assert!( @@ -69,6 +156,9 @@ async fn podman_gateway_restart_starts_running_sandbox() { "sandbox '{}' should still be listed after gateway restart. Names: {names:?}", sandbox.name ); + wait_for_sandbox_phase(&stopped_sandbox.name, "Stopped", Duration::from_secs(120)) + .await + .expect("explicitly stopped Podman sandbox should remain stopped after restart"); wait_for_sandbox_exec_contains( &sandbox.name, @@ -80,4 +170,5 @@ async fn podman_gateway_restart_starts_running_sandbox() { .expect("Podman sandbox should become ready again with its state preserved"); sandbox.cleanup().await; + stopped_sandbox.cleanup().await; } diff --git a/e2e/rust/tests/vm_gateway_start.rs b/e2e/rust/tests/vm_gateway_start.rs index 923198668b..5a6399a0fd 100644 --- a/e2e/rust/tests/vm_gateway_start.rs +++ b/e2e/rust/tests/vm_gateway_start.rs @@ -9,19 +9,82 @@ //! This test is gated behind the `e2e-vm` feature because it requires the VM //! driver runtime prepared by `e2e/rust/e2e-vm.sh`. +use std::fs; +use std::path::PathBuf; use std::time::Duration; use openshell_e2e::harness::cli::{ - sandbox_names, wait_for_healthy, wait_for_sandbox_exec_contains, + run_cli, sandbox_names, wait_for_healthy, wait_for_sandbox_exec_contains, + wait_for_sandbox_phase, }; use openshell_e2e::harness::gateway::ManagedGateway; use openshell_e2e::harness::sandbox::SandboxGuard; +use prost::Message; +use tokio::time::sleep; const READY_MARKER: &str = "vm-gateway-start-ready"; +const STOPPED_READY_MARKER: &str = "vm-gateway-start-stopped-ready"; const START_FILE: &str = "/sandbox/vm-gateway-start-state"; +const VM_STATE_DIR_ENV: &str = "OPENSHELL_E2E_VM_STATE_DIR"; + +#[derive(Clone, PartialEq, Message)] +struct PersistedDriverSandbox { + #[prost(string, tag = "2")] + name: String, +} + +fn vm_sandbox_stopped(sandbox_name: &str) -> Result { + let state_dir = std::env::var_os(VM_STATE_DIR_ENV) + .map(PathBuf::from) + .ok_or_else(|| format!("{VM_STATE_DIR_ENV} must be set"))?; + let sandboxes_dir = state_dir.join("sandboxes"); + for entry in fs::read_dir(&sandboxes_dir) + .map_err(|err| format!("read '{}': {err}", sandboxes_dir.display()))? + { + let path = entry + .map_err(|err| format!("read VM sandbox entry: {err}"))? + .path(); + let request_path = path.join("sandbox.pb"); + let bytes = match fs::read(&request_path) { + Ok(bytes) => bytes, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue, + Err(err) => return Err(format!("read '{}': {err}", request_path.display())), + }; + let sandbox = PersistedDriverSandbox::decode(bytes.as_slice()) + .map_err(|err| format!("decode '{}': {err}", request_path.display()))?; + if sandbox.name == sandbox_name { + return Ok(path.join("stopped").exists()); + } + } + Err(format!( + "VM state for sandbox '{sandbox_name}' was not found" + )) +} + +async fn wait_for_vm_stopped_marker( + sandbox_name: &str, + expected: bool, + timeout: Duration, +) -> Result<(), String> { + let start = std::time::Instant::now(); + let mut last_state; + loop { + match vm_sandbox_stopped(sandbox_name) { + Ok(stopped) if stopped == expected => return Ok(()), + Ok(stopped) => last_state = format!("stopped={stopped}"), + Err(err) => last_state = err, + } + if start.elapsed() > timeout { + return Err(format!( + "VM '{sandbox_name}' did not reach stopped={expected}: {last_state}" + )); + } + sleep(Duration::from_secs(1)).await; + } +} #[tokio::test] -async fn vm_gateway_restart_starts_running_sandbox() { +async fn vm_gateway_restart_preserves_running_and_stopped_intent() { if std::env::var("OPENSHELL_E2E_DRIVER").as_deref() != Ok("vm") { eprintln!("Skipping VM gateway start test: e2e driver is not vm"); return; @@ -55,11 +118,28 @@ async fn vm_gateway_restart_starts_running_sandbox() { "VM sandbox state was not written before restart:\n{before_restart}" ); + let stopped_script = format!("echo {STOPPED_READY_MARKER}; while true; do sleep 1; done"); + let mut stopped_sandbox = + SandboxGuard::create_keep(&["sh", "-lc", &stopped_script], STOPPED_READY_MARKER) + .await + .expect("create VM sandbox that will remain stopped"); + let (stop_output, stop_code) = run_cli(&["sandbox", "stop", &stopped_sandbox.name]).await; + assert_eq!(stop_code, 0, "sandbox stop should succeed:\n{stop_output}"); + wait_for_sandbox_phase(&stopped_sandbox.name, "Stopped", Duration::from_secs(120)) + .await + .expect("VM sandbox should be stopped before gateway restart"); + gateway.stop().expect("stop e2e gateway"); + wait_for_vm_stopped_marker(&sandbox.name, true, Duration::from_secs(60)) + .await + .expect("gateway shutdown should stop the running-intent VM through its driver"); gateway.start().expect("restart e2e gateway"); wait_for_healthy(Duration::from_secs(120)) .await .expect("gateway should become healthy after restart"); + wait_for_vm_stopped_marker(&sandbox.name, false, Duration::from_secs(120)) + .await + .expect("gateway startup should restart the running-intent VM"); let names = sandbox_names().await.expect("list sandboxes after restart"); assert!( @@ -67,6 +147,9 @@ async fn vm_gateway_restart_starts_running_sandbox() { "sandbox '{}' should still be listed after gateway restart. Names: {names:?}", sandbox.name ); + wait_for_sandbox_phase(&stopped_sandbox.name, "Stopped", Duration::from_secs(120)) + .await + .expect("explicitly stopped VM sandbox should remain stopped after restart"); wait_for_sandbox_exec_contains( &sandbox.name, @@ -78,4 +161,5 @@ async fn vm_gateway_restart_starts_running_sandbox() { .expect("VM sandbox should become ready again with its overlay state preserved"); sandbox.cleanup().await; + stopped_sandbox.cleanup().await; } diff --git a/e2e/support/gateway-common.sh b/e2e/support/gateway-common.sh index d9f336b411..a3eef4bd18 100644 --- a/e2e/support/gateway-common.sh +++ b/e2e/support/gateway-common.sh @@ -273,6 +273,29 @@ e2e_stop_gateway() { if [ -n "${gateway_pid}" ] && kill -0 "${gateway_pid}" 2>/dev/null; then echo "Stopping openshell-gateway (pid ${gateway_pid})..." kill "${gateway_pid}" 2>/dev/null || true + + # A Rust E2E test may have restarted the gateway and updated the PID file. + # That replacement process is not a child of this shell, so `wait` returns + # immediately even though gateway shutdown (including its sandbox stop + # sweep) is still in progress. Poll until either the process exits or a + # child process becomes a zombie that the final `wait` can reap. + local attempts=0 + local process_state="" + while kill -0 "${gateway_pid}" 2>/dev/null && [ "${attempts}" -lt 120 ]; do + process_state="$(ps -p "${gateway_pid}" -o stat= 2>/dev/null || true)" + case "${process_state}" in + *Z*) break ;; + esac + sleep 0.5 + attempts=$((attempts + 1)) + done + if kill -0 "${gateway_pid}" 2>/dev/null; then + process_state="$(ps -p "${gateway_pid}" -o stat= 2>/dev/null || true)" + case "${process_state}" in + *Z*) ;; + *) kill -KILL "${gateway_pid}" 2>/dev/null || true ;; + esac + fi wait "${gateway_pid}" 2>/dev/null || true fi }