From 1984a72419053c16a5ce11f59f7d86d26d0694bc Mon Sep 17 00:00:00 2001 From: Matthew Grossman Date: Wed, 4 Nov 2026 10:09:26 -0800 Subject: [PATCH 1/4] fix(podman): keep a polled stale exit from failing an in-progress start The gateway's reconcile sweep applies polled snapshots without waiting for an in-progress start. During a stop/start restart, get_sandbox and list_sandboxes built the stopped workload from its list entry, so the previous run's exit was reported as ContainerExited and moved the sandbox from Starting to Error. While a start is in progress, inspect a stopped workload before reporting it. If it is running again, or still shows the exit recorded before the start, report it as ContainerStarting. A new exit, and any exit once the start has returned, is still reported. The start is tracked by a guard that clears the in-progress mark on every return path. Closes #5258 Signed-off-by: Matthew Grossman --- crates/openshell-driver-podman/src/driver.rs | 170 ++++++++++++++- crates/openshell-driver-podman/src/watcher.rs | 193 +++++++++++++++--- crates/openshell-server/src/compute/mod.rs | 63 ++++++ 3 files changed, 391 insertions(+), 35 deletions(-) diff --git a/crates/openshell-driver-podman/src/driver.rs b/crates/openshell-driver-podman/src/driver.rs index c70cc6eca1..0ac89d66fe 100644 --- a/crates/openshell-driver-podman/src/driver.rs +++ b/crates/openshell-driver-podman/src/driver.rs @@ -2002,6 +2002,7 @@ impl PodmanComputeDriver { durable_sandbox: Option<&DriverSandbox>, ) -> Result<(), ComputeDriverError> { let span_status = openshell_otel::ErrorStatusGuard::current(); + let _start = self.lifecycle_event_fences.begin_start(sandbox_id); // Verify before inspecting or modifying either member of the pair. // The durable snapshot is used only if a stopped generation must be // reconciled; it never comes from Podman's mutable inspect response. @@ -2328,14 +2329,55 @@ impl PodmanComputeDriver { .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) .or_else(|| driver_sandbox_from_list_entry(entry))) } else { - Ok(driver_sandbox_from_list_entry(entry)) + Ok(self.driver_sandbox_from_polled_entry(entry).await) } } + /// Build a polled snapshot from a list entry that was not inspected as a + /// running workload. A workload whose start is in progress is reported as + /// starting unless it shows a new exit, because the gateway applies polled + /// snapshots without waiting for the start to finish. + async fn driver_sandbox_from_polled_entry( + &self, + entry: &ContainerListEntry, + ) -> Option { + if matches!(entry.state.as_str(), "exited" | "stopped") + && let Some(sandbox_id) = entry.labels.get(LABEL_SANDBOX_ID) + && self.lifecycle_event_fences.start_in_progress(sandbox_id) + { + match self.client.inspect_container(&entry.id).await { + Ok(inspect) + if self + .lifecycle_event_fences + .polled_workload_still_starting(sandbox_id, &inspect.state) => + { + debug!( + sandbox_id = %sandbox_id, + container = %entry.id, + status = %inspect.state.status, + "Reporting polled sandbox as starting while its start is in progress" + ); + return watcher::starting_driver_sandbox_from_list_entry(entry); + } + Ok(_) => {} + Err(error) => { + warn!( + sandbox_id = %sandbox_id, + container = %entry.id, + error = %error, + "Failed to inspect stopped container during start, falling back to list entry" + ); + } + } + } + driver_sandbox_from_list_entry(entry) + } + /// List all managed sandboxes. /// - /// Only inspects running containers (to get health status). Non-running - /// containers are built directly from the list entry data. + /// Inspects running containers (to get health status) and stopped + /// containers whose start is in progress. Other containers are built + /// directly from the list entry data. pub async fn list_sandboxes(&self) -> Result, ComputeDriverError> { let entries = self .client @@ -2365,7 +2407,7 @@ impl PodmanComputeDriver { } } // Non-running containers (or inspect fallback): build from list data. - if let Some(sandbox) = driver_sandbox_from_list_entry(entry) { + if let Some(sandbox) = self.driver_sandbox_from_polled_entry(entry).await { sandboxes.push(sandbox); } } @@ -3064,6 +3106,126 @@ mod tests { let _ = fs::remove_file(start_socket); } + const PREVIOUS_EXIT: &str = "2026-08-12T16:39:13.123456789Z"; + const EXITED_WORKLOAD_LIST: &str = r#"[{"Id":"ctr-1","Names":["openshell-sandbox-demo"],"State":"exited","Labels":{"openshell.ai/sandbox-id":"sandbox-1","openshell.ai/sandbox-name":"demo","openshell.ai/sandbox-workspace":"default"}}]"#; + + fn exited_workload_inspect(finished_at: &str) -> StubResponse { + StubResponse::new( + StatusCode::OK, + format!( + r#"{{"Id":"ctr-1","Name":"openshell-sandbox-demo","State":{{"Status":"exited","Running":false,"ExitCode":0,"FinishedAt":"{finished_at}"}},"Config":{{}}}}"# + ), + ) + } + + fn ready_condition(sandbox: Option) -> (String, String) { + let condition = sandbox + .expect("sandbox should be reported") + .status + .expect("sandbox should have a status") + .conditions + .into_iter() + .find(|condition| condition.r#type == "Ready") + .expect("sandbox should have a Ready condition"); + (condition.status, condition.reason) + } + + #[tokio::test] + async fn polled_snapshot_during_restart_does_not_report_the_previous_exit() { + let (socket, requests, handle) = spawn_podman_stub( + "poll-during-start", + vec![ + StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), + exited_workload_inspect(PREVIOUS_EXIT), + StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), + exited_workload_inspect(PREVIOUS_EXIT), + // The workload starts between the list and the inspect. + StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), + StubResponse::new( + StatusCode::OK, + r#"{"Id":"ctr-1","Name":"openshell-sandbox-demo","State":{"Status":"running","Running":true,"StartedAt":"2026-08-12T16:39:14Z","FinishedAt":"2026-08-12T16:39:13.123456789Z"},"Config":{}}"#, + ), + ], + ); + let driver = test_driver(socket.clone()); + driver + .lifecycle_event_fences + .record_previous_exit("sandbox-1", Some(PREVIOUS_EXIT)); + let _start = driver.lifecycle_event_fences.begin_start("sandbox-1"); + let starting = ("False".to_string(), "ContainerStarting".to_string()); + + assert_eq!( + ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), + starting + ); + let mut listed = driver.list_sandboxes().await.unwrap(); + assert_eq!(listed.len(), 1); + assert_eq!(ready_condition(listed.pop()), starting); + assert_eq!( + ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), + starting + ); + + handle.await.unwrap(); + assert_eq!(requests.lock().unwrap().len(), 6); + let _ = fs::remove_file(socket); + } + + #[tokio::test] + async fn polled_snapshot_reports_a_new_exit_and_any_exit_after_start() { + let (socket, requests, handle) = spawn_podman_stub( + "poll-new-exit", + vec![ + StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), + exited_workload_inspect("2026-08-12T16:40:00.5Z"), + // Once the start has returned, the list entry is reported + // without an inspect, including a failed start's previous exit. + StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), + StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), + ], + ); + let driver = test_driver(socket.clone()); + driver + .lifecycle_event_fences + .record_previous_exit("sandbox-1", Some(PREVIOUS_EXIT)); + let exited = ("False".to_string(), "ContainerExited".to_string()); + + let start = driver.lifecycle_event_fences.begin_start("sandbox-1"); + assert_eq!( + ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), + exited + ); + drop(start); + assert_eq!( + ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), + exited + ); + let mut listed = driver.list_sandboxes().await.unwrap(); + assert_eq!(ready_condition(listed.pop()), exited); + + handle.await.unwrap(); + assert_eq!(requests.lock().unwrap().len(), 4); + let _ = fs::remove_file(socket); + } + + #[tokio::test] + async fn failed_start_is_no_longer_in_progress() { + let (socket, _requests, handle) = spawn_podman_stub( + "failed-start", + vec![StubResponse::new(StatusCode::OK, "[]")], + ); + let driver = test_driver(socket.clone()); + + let result = driver + .start_sandbox("sandbox-1", "generation-1", &[], None) + .await; + + assert!(matches!(result, Err(ComputeDriverError::NotFound))); + assert!(!driver.lifecycle_event_fences.start_in_progress("sandbox-1")); + handle.await.unwrap(); + let _ = fs::remove_file(socket); + } + #[tokio::test] async fn stop_waits_for_the_container_to_leave_stopping_state() { let (socket, requests, handle) = spawn_podman_stub( diff --git a/crates/openshell-driver-podman/src/watcher.rs b/crates/openshell-driver-podman/src/watcher.rs index 9a508e52c3..730246bef1 100644 --- a/crates/openshell-driver-podman/src/watcher.rs +++ b/crates/openshell-driver-podman/src/watcher.rs @@ -16,9 +16,9 @@ use openshell_core::proto::compute::v1::{ DriverCondition, DriverSandbox, DriverSandboxStatus, WatchSandboxesDeletedEvent, WatchSandboxesEvent, WatchSandboxesSandboxEvent, watch_sandboxes_event, }; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::pin::Pin; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, MutexGuard}; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tracing::{debug, info, warn}; @@ -43,30 +43,70 @@ pub type WatchStream = /// the sandbox from `Starting` to `Error`. #[derive(Clone, Debug, Default)] pub struct LifecycleEventFences { - previous_finished_at: Arc>>, + state: Arc>, +} + +#[derive(Debug, Default)] +struct LifecycleFenceState { + previous_finished_at: HashMap, + starts_in_progress: HashSet, +} + +/// Marks a sandbox start as in progress until it is dropped. +#[must_use = "the start is only in progress while this guard is held"] +pub struct StartInProgress { + fences: LifecycleEventFences, + sandbox_id: String, +} + +impl Drop for StartInProgress { + fn drop(&mut self) { + self.fences + .lock() + .starts_in_progress + .remove(&self.sandbox_id); + } } impl LifecycleEventFences { - pub fn record_previous_exit(&self, sandbox_id: &str, finished_at: Option<&str>) { - let mut fences = self - .previous_finished_at + fn lock(&self) -> MutexGuard<'_, LifecycleFenceState> { + self.state .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + pub fn begin_start(&self, sandbox_id: &str) -> StartInProgress { + self.lock() + .starts_in_progress + .insert(sandbox_id.to_string()); + StartInProgress { + fences: self.clone(), + sandbox_id: sandbox_id.to_string(), + } + } + + pub fn start_in_progress(&self, sandbox_id: &str) -> bool { + self.lock().starts_in_progress.contains(sandbox_id) + } + + pub fn record_previous_exit(&self, sandbox_id: &str, finished_at: Option<&str>) { + let mut state = self.lock(); match finished_at.filter(|finished_at| !finished_at.is_empty()) { Some(finished_at) => { - fences.insert(sandbox_id.to_string(), finished_at.to_string()); + state + .previous_finished_at + .insert(sandbox_id.to_string(), finished_at.to_string()); } None => { - fences.remove(sandbox_id); + state.previous_finished_at.remove(sandbox_id); } } } pub fn remove(&self, sandbox_id: &str) { - self.previous_finished_at - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .remove(sandbox_id); + let mut state = self.lock(); + state.previous_finished_at.remove(sandbox_id); + state.starts_in_progress.remove(sandbox_id); } fn matches_previous_exit( @@ -84,12 +124,37 @@ impl LifecycleEventFences { let Some(finished_at) = state.finished_at.as_deref() else { return false; }; - self.previous_finished_at - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) + self.lock() + .previous_finished_at .get(sandbox_id) .is_some_and(|previous| previous == finished_at) } + + /// Whether a polled workload that the container list reported as not + /// running is still starting rather than failed. + /// + /// The gateway's reconcile sweep polls sandboxes without waiting for an + /// in-progress start. Until the start runs the workload again, inspect + /// still shows the exit recorded before the start, and the workload can + /// also be running again by the time it is inspected. Neither is a failure + /// of the new run. An exit with a different `FinishedAt` is, and once the + /// start has returned every exit is reported. + pub fn polled_workload_still_starting(&self, sandbox_id: &str, state: &ContainerState) -> bool { + let fences = self.lock(); + if !fences.starts_in_progress.contains(sandbox_id) { + return false; + } + match state.status.as_str() { + "running" => true, + "exited" | "stopped" => state.finished_at.as_deref().is_some_and(|finished_at| { + fences + .previous_finished_at + .get(sandbox_id) + .is_some_and(|previous| previous == finished_at) + }), + _ => false, + } + } } /// Build a `WatchSandboxesEvent` carrying a sandbox snapshot. @@ -493,14 +558,6 @@ pub fn driver_sandbox_from_inspect(inspect: &ContainerInspect) -> Option Option { - let sandbox_id = entry.labels.get(LABEL_SANDBOX_ID)?.clone(); - let sandbox_name = entry - .labels - .get(LABEL_SANDBOX_NAME) - .cloned() - .unwrap_or_default(); - let workspace = entry.labels.get(LABEL_SANDBOX_WORKSPACE).cloned()?; - let (reason, status_str, message) = match entry.state.as_str() { "running" => ( CONDITION_RUNNING, @@ -513,13 +570,8 @@ pub fn driver_sandbox_from_list_entry(entry: &ContainerListEntry) -> Option ("ContainerRemoving", "False", String::new()), _ => ("Unknown", "Unknown", String::new()), }; - - Some(build_driver_sandbox( - sandbox_id, - sandbox_name, - workspace, - entry.names.first().cloned().unwrap_or_default(), - short_id(&entry.id), + driver_sandbox_from_list_entry_with_condition( + entry, DriverCondition { r#type: "Ready".to_string(), status: status_str.to_string(), @@ -527,6 +579,45 @@ pub fn driver_sandbox_from_list_entry(entry: &ContainerListEntry) -> Option Option { + driver_sandbox_from_list_entry_with_condition( + entry, + DriverCondition { + r#type: "Ready".to_string(), + status: "False".to_string(), + reason: CONDITION_STARTING.to_string(), + message: "Container start in progress".to_string(), + transition_time: None, + }, + ) +} + +fn driver_sandbox_from_list_entry_with_condition( + entry: &ContainerListEntry, + condition: DriverCondition, +) -> Option { + let sandbox_id = entry.labels.get(LABEL_SANDBOX_ID)?.clone(); + let sandbox_name = entry + .labels + .get(LABEL_SANDBOX_NAME) + .cloned() + .unwrap_or_default(); + let workspace = entry.labels.get(LABEL_SANDBOX_WORKSPACE).cloned()?; + + Some(build_driver_sandbox( + sandbox_id, + sandbox_name, + workspace, + entry.names.first().cloned().unwrap_or_default(), + short_id(&entry.id), + condition, entry.state == "removing", )) } @@ -769,6 +860,46 @@ mod tests { )); } + #[test] + fn lifecycle_fence_reports_polled_previous_exit_as_starting_only_during_start() { + let fences = LifecycleEventFences::default(); + fences.record_previous_exit("sandbox-1", Some("2026-08-12T16:39:13.123456789Z")); + let previous_exit = ContainerState { + status: "exited".to_string(), + running: false, + exit_code: 0, + oom_killed: false, + health: None, + started_at: Some("2026-08-12T16:38:58Z".to_string()), + finished_at: Some("2026-08-12T16:39:13.123456789Z".to_string()), + supervisor_exit_code: None, + startup_diagnostic: None, + }; + let mut new_exit = previous_exit.clone(); + new_exit.finished_at = Some("2026-08-12T16:39:13.5Z".to_string()); + let mut running = previous_exit.clone(); + running.status = "running".to_string(); + running.running = true; + + // A stopped sandbox, or one whose start has already failed, reports + // the exit. + assert!(!fences.polled_workload_still_starting("sandbox-1", &previous_exit)); + + let start = fences.begin_start("sandbox-1"); + assert!(fences.polled_workload_still_starting("sandbox-1", &previous_exit)); + assert!(fences.polled_workload_still_starting("sandbox-1", &running)); + assert!(!fences.polled_workload_still_starting("sandbox-1", &new_exit)); + assert!(!fences.polled_workload_still_starting("sandbox-2", &previous_exit)); + drop(start); + assert!(!fences.start_in_progress("sandbox-1")); + assert!(!fences.polled_workload_still_starting("sandbox-1", &previous_exit)); + + let _start = fences.begin_start("sandbox-1"); + fences.remove("sandbox-1"); + assert!(!fences.start_in_progress("sandbox-1")); + assert!(!fences.polled_workload_still_starting("sandbox-1", &previous_exit)); + } + #[test] fn condition_healthy_container() { let state = ContainerState { diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2b168f26df..7104fe1aef 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -14068,6 +14068,69 @@ mod tests { ); } + // Gateway side of the Podman start contract. The reconcile sweep applies + // polled snapshots without waiting for an in-progress start. While a start + // is in progress, the Podman driver reports a stopped workload that still + // shows its previous exit as ContainerStarting. Such a snapshot keeps the + // sandbox Starting, and the start completes. This passes without the + // #5258 fix; the driver test + // polled_snapshot_during_restart_does_not_report_the_previous_exit covers + // the regression. + #[tokio::test] + async fn reconcile_snapshot_during_start_keeps_the_sandbox_starting() { + let driver = ControlledDriver::new(); + driver.block_start(); + let sandbox = sandbox_record( + "sb-start-reconcile", + "sandbox-start-reconcile", + SandboxPhase::Stopped, + ); + let mut polled = ready_driver_sandbox(sandbox.object_id(), sandbox.object_name()); + polled.status = Some(make_driver_status(make_driver_condition( + "ContainerStarting", + "Container start in progress", + ))); + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(polled.clone()))); + let runtime = test_runtime(driver.clone()).await; + runtime.store.put_message(&sandbox).await.unwrap(); + + let start_runtime = runtime.clone(); + let sandbox_name = sandbox.object_name().to_string(); + let start = + tokio::spawn( + async move { start_runtime.start_sandbox("default", &sandbox_name).await }, + ); + tokio::time::timeout(Duration::from_secs(1), driver.start_started.notified()) + .await + .expect("start did not reach the driver"); + + tokio::time::timeout( + Duration::from_secs(1), + runtime.reconcile_snapshot_sandbox(polled, i64::MAX), + ) + .await + .expect("reconciliation must not wait for the start") + .unwrap(); + let phase_during_start = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap() + .phase(); + assert_eq!(phase_during_start, SandboxPhase::Starting as i32); + + driver.release_start(); + start.await.unwrap().unwrap(); + let stored = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert_eq!(stored.phase(), SandboxPhase::Starting as i32); + } + #[tokio::test] async fn live_container_exit_during_start_still_transitions_to_error() { for reason in [ From c15cece39a92ddbc69bab79f3f0324f2c1ab6c29 Mon Sep 17 00:00:00 2001 From: Matthew Grossman Date: Wed, 4 Nov 2026 10:09:59 -0800 Subject: [PATCH 2/4] docs(podman): describe how restarts fence the previous exit Signed-off-by: Matthew Grossman --- crates/openshell-driver-podman/README.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/openshell-driver-podman/README.md b/crates/openshell-driver-podman/README.md index 1c63bb8f37..dcd606ae2c 100644 --- a/crates/openshell-driver-podman/README.md +++ b/crates/openshell-driver-podman/README.md @@ -131,6 +131,9 @@ the consumed sandbox bootstrap from a copy in the supervisor's private filesystem, verifies the fence, and starts the same pair. A failed supervisor start stops the workload. Delete removes the companion first, then the workload, channel, workspace, and driver-owned secrets. User-owned volumes are retained. +The previous run's exit must not fail a restart. Watch ignores stop events that +still show the exit recorded before the start. Until the start returns, polled +snapshots report such a workload as starting. A new exit is still reported. Only workload containers appear in sandbox list/watch results. Readiness uses the supervisor's private health socket; there is no shell, legacy marker, or From 03162dbadc0b160fb58af819c2c3dc28756898c2 Mon Sep 17 00:00:00 2001 From: Matthew Grossman Date: Wed, 4 Nov 2026 11:18:13 -0800 Subject: [PATCH 3/4] fix(podman): keep a restart from containing its own workload A restart starts the workload before its exited supervisor. Watch reconciliation, start events, and polled snapshots could inspect the workload in that gap, see a running workload beside an exited supervisor, and stop it. Record the supervisor exit that the start observed. While that start is in progress, an exited supervisor with the recorded FinishedAt reports the workload as starting instead of containing it. A missing supervisor, a new supervisor exit, and any exit after the start returns are still contained. Signed-off-by: Matthew Grossman --- crates/openshell-driver-podman/README.md | 6 +- crates/openshell-driver-podman/src/driver.rs | 25 ++- crates/openshell-driver-podman/src/watcher.rs | 154 ++++++++++++++++-- 3 files changed, 166 insertions(+), 19 deletions(-) diff --git a/crates/openshell-driver-podman/README.md b/crates/openshell-driver-podman/README.md index dcd606ae2c..2c0808ea5e 100644 --- a/crates/openshell-driver-podman/README.md +++ b/crates/openshell-driver-podman/README.md @@ -138,8 +138,10 @@ snapshots report such a workload as starting. A new exit is still reported. Only workload containers appear in sandbox list/watch results. Readiness uses the supervisor's private health socket; there is no shell, legacy marker, or TCP-listener shortcut. Watch reconciliation and supervisor exit/removal events -stop a running workload whose companion is unavailable. The gateway also -requires the authenticated supervisor session before publishing Ready. +stop a running workload whose companion is unavailable. A restart starts the +workload before its exited companion, so the companion exit that the start +recorded is not contained until the start returns. The gateway also requires +the authenticated supervisor session before publishing Ready. ## Mounts, GPUs, and configuration diff --git a/crates/openshell-driver-podman/src/driver.rs b/crates/openshell-driver-podman/src/driver.rs index 0ac89d66fe..a79d1507f9 100644 --- a/crates/openshell-driver-podman/src/driver.rs +++ b/crates/openshell-driver-podman/src/driver.rs @@ -2099,8 +2099,11 @@ impl PodmanComputeDriver { self.client .stop_container(&supervisor, self.config.stop_timeout_secs) .await?; - self.wait_for_container_stopped(sandbox_id, &supervisor) + let supervisor_finished_at = self + .wait_for_container_stopped(sandbox_id, &supervisor) .await?; + self.lifecycle_event_fences + .record_supervisor_restart(sandbox_id, supervisor_finished_at.as_deref()); let archive = self .client .copy_from_container(&supervisor, crate::isolation::RESTART_METADATA_PATH) @@ -2323,11 +2326,13 @@ impl PodmanComputeDriver { return Ok(None); }; if entry.state == "running" { - Ok(watcher::inspect_workload(&self.client, &entry.id) - .await - .ok() - .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) - .or_else(|| driver_sandbox_from_list_entry(entry))) + Ok( + watcher::inspect_workload(&self.client, &self.lifecycle_event_fences, &entry.id) + .await + .ok() + .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) + .or_else(|| driver_sandbox_from_list_entry(entry)), + ) } else { Ok(self.driver_sandbox_from_polled_entry(entry).await) } @@ -2389,7 +2394,13 @@ impl PodmanComputeDriver { for entry in &entries { if entry.state == "running" { // Running containers need inspect for health check status. - match watcher::inspect_workload(&self.client, &entry.id).await { + match watcher::inspect_workload( + &self.client, + &self.lifecycle_event_fences, + &entry.id, + ) + .await + { Ok(inspect) => { if let Some(sandbox) = driver_sandbox_from_inspect(&inspect) { sandboxes.push(sandbox); diff --git a/crates/openshell-driver-podman/src/watcher.rs b/crates/openshell-driver-podman/src/watcher.rs index 730246bef1..c1e77d0dbe 100644 --- a/crates/openshell-driver-podman/src/watcher.rs +++ b/crates/openshell-driver-podman/src/watcher.rs @@ -16,7 +16,7 @@ use openshell_core::proto::compute::v1::{ DriverCondition, DriverSandbox, DriverSandboxStatus, WatchSandboxesDeletedEvent, WatchSandboxesEvent, WatchSandboxesSandboxEvent, watch_sandboxes_event, }; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::pin::Pin; use std::sync::{Arc, Mutex, MutexGuard}; use tokio::sync::mpsc; @@ -49,7 +49,13 @@ pub struct LifecycleEventFences { #[derive(Debug, Default)] struct LifecycleFenceState { previous_finished_at: HashMap, - starts_in_progress: HashSet, + starts_in_progress: HashMap, +} + +#[derive(Debug, Default)] +struct InProgressStart { + /// The supervisor exit that the start observed before restarting it. + supervisor_finished_at: Option, } /// Marks a sandbox start as in progress until it is dropped. @@ -78,7 +84,7 @@ impl LifecycleEventFences { pub fn begin_start(&self, sandbox_id: &str) -> StartInProgress { self.lock() .starts_in_progress - .insert(sandbox_id.to_string()); + .insert(sandbox_id.to_string(), InProgressStart::default()); StartInProgress { fences: self.clone(), sandbox_id: sandbox_id.to_string(), @@ -86,7 +92,34 @@ impl LifecycleEventFences { } pub fn start_in_progress(&self, sandbox_id: &str) -> bool { - self.lock().starts_in_progress.contains(sandbox_id) + self.lock().starts_in_progress.contains_key(sandbox_id) + } + + /// Record the supervisor exit that an in-progress start is about to + /// restart. See [`Self::supervisor_restart_in_progress`]. + pub fn record_supervisor_restart(&self, sandbox_id: &str, finished_at: Option<&str>) { + if let Some(start) = self.lock().starts_in_progress.get_mut(sandbox_id) { + start.supervisor_finished_at = finished_at + .filter(|finished_at| !finished_at.is_empty()) + .map(str::to_string); + } + } + + /// Whether an exited supervisor is the one that the in-progress start is + /// restarting. A restart starts the workload before its supervisor, so the + /// workload must not be contained for that exit. Any other exit is. + pub fn supervisor_restart_in_progress(&self, sandbox_id: &str, state: &ContainerState) -> bool { + if !matches!(state.status.as_str(), "exited" | "stopped") { + return false; + } + let Some(finished_at) = state.finished_at.as_deref() else { + return false; + }; + self.lock() + .starts_in_progress + .get(sandbox_id) + .and_then(|start| start.supervisor_finished_at.as_deref()) + .is_some_and(|recorded| recorded == finished_at) } pub fn record_previous_exit(&self, sandbox_id: &str, finished_at: Option<&str>) { @@ -141,7 +174,7 @@ impl LifecycleEventFences { /// start has returned every exit is reported. pub fn polled_workload_still_starting(&self, sandbox_id: &str, state: &ContainerState) -> bool { let fences = self.lock(); - if !fences.starts_in_progress.contains(sandbox_id) { + if !fences.starts_in_progress.contains_key(sandbox_id) { return false; } match state.status.as_str() { @@ -214,7 +247,7 @@ pub async fn start_watch( // health check status — matching the same condition derivation used // for live events. if entry.state == "running" { - match inspect_workload(&client, &entry.id).await { + match inspect_workload(&client, &lifecycle_event_fences, &entry.id).await { Ok(inspect) => { if let Some(sandbox) = driver_sandbox_from_inspect(&inspect) { if tx.send(Ok(sandbox_event(sandbox))).await.is_err() { @@ -330,7 +363,7 @@ async fn map_podman_event( .await .ok()?; let workload = workloads.first()?; - return inspect_workload(client, &workload.id) + return inspect_workload(client, lifecycle_event_fences, &workload.id) .await .ok() .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) @@ -341,7 +374,7 @@ async fn map_podman_event( "remove" => Some(deleted_event(sandbox_id.clone())), "create" | "start" | "stop" | "die" | "health_status" => { // Inspect the container to get current state. - match inspect_workload(client, container_id).await { + match inspect_workload(client, lifecycle_event_fences, container_id).await { Ok(inspect) => { if lifecycle_event_fences.matches_previous_exit( event, @@ -421,8 +454,13 @@ async fn map_podman_event( /// A workload is ready only when its independent supervisor is healthy. This /// check runs both on watch reconciliation and on events, and contains a lost /// supervisor even when the gateway missed the original exit event. +/// +/// A restart starts the workload before its exited companion. The exit that +/// the start recorded is not contained: the companion is about to start, and +/// the start stops the workload itself if the companion fails to start. pub async fn inspect_workload( client: &PodmanClient, + lifecycle_event_fences: &LifecycleEventFences, id: &str, ) -> Result { let mut workload = client.inspect_container(id).await?; @@ -457,6 +495,19 @@ pub async fn inspect_workload( status: "starting".into(), }); } + Ok(supervisor) + if lifecycle_event_fences + .supervisor_restart_in_progress(sandbox_id, &supervisor.state) => + { + debug!( + sandbox_id = %sandbox_id, + workload = %workload.id, + "Not containing workload while its start restarts the supervisor" + ); + workload.state.health = Some(HealthState { + status: "starting".into(), + }); + } // Both containers exist before initial start. A missing or exited // companion therefore requires containment, including after a // gateway restart that missed the original Podman exit event. @@ -732,7 +783,9 @@ mod tests { ], ); let client = PodmanClient::new(path.clone()); - let inspected = inspect_workload(&client, "workload").await.unwrap(); + let inspected = inspect_workload(&client, &LifecycleEventFences::default(), "workload") + .await + .unwrap(); assert!(!inspected.state.running); handle.await.unwrap(); assert!( @@ -763,7 +816,9 @@ mod tests { ], ); let client = PodmanClient::new(path.clone()); - let inspected = inspect_workload(&client, "workload").await.unwrap(); + let inspected = inspect_workload(&client, &LifecycleEventFences::default(), "workload") + .await + .unwrap(); assert!(inspected.state.running); assert_eq!(inspected.state.health.unwrap().status, "starting"); handle.await.unwrap(); @@ -771,6 +826,85 @@ mod tests { let _ = std::fs::remove_file(path); } + #[tokio::test] + async fn supervisor_exit_recorded_by_a_restart_does_not_contain_the_workload() { + use crate::test_utils::{StubResponse, spawn_podman_stub}; + use hyper::StatusCode; + const RECORDED_EXIT: &str = "2026-08-12T16:39:13.123456789Z"; + const RUNNING_WORKLOAD: &str = r#"{"Id":"workload","Name":"workload","State":{"Status":"running","Running":true},"Config":{"Labels":{"openshell.ai/sandbox-id":"test","openshell.ai/isolation-role":"sandbox"}}}"#; + const STOPPED_WORKLOAD: &str = r#"{"Id":"workload","Name":"workload","State":{"Status":"exited","Running":false},"Config":{}}"#; + let exited_supervisor = |finished_at: &str| { + StubResponse::new( + StatusCode::OK, + format!( + r#"{{"Id":"supervisor","Name":"supervisor","State":{{"Status":"exited","Running":false,"ExitCode":0,"FinishedAt":"{finished_at}"}},"Config":{{}}}}"# + ), + ) + }; + let contained = |finished_at: &str| { + vec![ + StubResponse::new(StatusCode::OK, RUNNING_WORKLOAD), + exited_supervisor(finished_at), + StubResponse::new(StatusCode::NO_CONTENT, ""), + StubResponse::new(StatusCode::OK, STOPPED_WORKLOAD), + ] + }; + let mut responses = contained(RECORDED_EXIT); + responses.extend([ + StubResponse::new(StatusCode::OK, RUNNING_WORKLOAD), + exited_supervisor(RECORDED_EXIT), + ]); + responses.extend(contained("2026-08-12T16:39:14Z")); + responses.extend(contained(RECORDED_EXIT)); + let (path, requests, handle) = spawn_podman_stub("restarting-supervisor", responses); + let client = PodmanClient::new(path.clone()); + let fences = LifecycleEventFences::default(); + let stops = || { + requests + .lock() + .unwrap() + .iter() + .filter(|request| request.ends_with("/libpod/containers/workload/stop?timeout=0")) + .count() + }; + + // Before the start has stopped and recorded the supervisor, its exit + // is contained. + let start = fences.begin_start("test"); + let inspected = inspect_workload(&client, &fences, "workload") + .await + .unwrap(); + assert!(!inspected.state.running); + assert_eq!(stops(), 1); + + // A restart starts the workload before the supervisor it recorded. + fences.record_supervisor_restart("test", Some(RECORDED_EXIT)); + let inspected = inspect_workload(&client, &fences, "workload") + .await + .unwrap(); + assert!(inspected.state.running); + assert_eq!(inspected.state.health.unwrap().status, "starting"); + assert_eq!(stops(), 1); + + // A new supervisor exit is contained during the start. + let inspected = inspect_workload(&client, &fences, "workload") + .await + .unwrap(); + assert!(!inspected.state.running); + assert_eq!(stops(), 2); + + // Once the start has returned, the recorded exit is contained. + drop(start); + let inspected = inspect_workload(&client, &fences, "workload") + .await + .unwrap(); + assert!(!inspected.state.running); + assert_eq!(stops(), 3); + + handle.await.unwrap(); + let _ = std::fs::remove_file(path); + } + fn podman_event(action: &str, sandbox_id: &str, time_nano: i64) -> PodmanEvent { podman_event_for_container(action, sandbox_id, "container-1", time_nano) } From 1e64885b613cfb20c260e7dcd36dfdece258f7f4 Mon Sep 17 00:00:00 2001 From: Matthew Grossman Date: Wed, 4 Nov 2026 22:03:33 -0800 Subject: [PATCH 4/4] fix(server): skip the reconcile poll of a sandbox whose start is in progress StartSandbox holds the sandbox lifecycle gate for the whole driver call, and the workload can still show its previous exit until the driver starts it. The watch path already waits on that gate for Starting sandboxes; the reconcile sweep did not, so a poll could move a restarting sandbox from Starting to Error. Skip such a sandbox and let the next tick or event report the settled state. This replaces the Podman-specific fence of polled snapshots. Signed-off-by: Matthew Grossman --- crates/openshell-driver-podman/README.md | 9 +- crates/openshell-driver-podman/src/driver.rs | 195 +--------- crates/openshell-driver-podman/src/watcher.rs | 335 ++---------------- crates/openshell-server/src/compute/mod.rs | 58 +-- 4 files changed, 85 insertions(+), 512 deletions(-) diff --git a/crates/openshell-driver-podman/README.md b/crates/openshell-driver-podman/README.md index 2c0808ea5e..1c63bb8f37 100644 --- a/crates/openshell-driver-podman/README.md +++ b/crates/openshell-driver-podman/README.md @@ -131,17 +131,12 @@ the consumed sandbox bootstrap from a copy in the supervisor's private filesystem, verifies the fence, and starts the same pair. A failed supervisor start stops the workload. Delete removes the companion first, then the workload, channel, workspace, and driver-owned secrets. User-owned volumes are retained. -The previous run's exit must not fail a restart. Watch ignores stop events that -still show the exit recorded before the start. Until the start returns, polled -snapshots report such a workload as starting. A new exit is still reported. Only workload containers appear in sandbox list/watch results. Readiness uses the supervisor's private health socket; there is no shell, legacy marker, or TCP-listener shortcut. Watch reconciliation and supervisor exit/removal events -stop a running workload whose companion is unavailable. A restart starts the -workload before its exited companion, so the companion exit that the start -recorded is not contained until the start returns. The gateway also requires -the authenticated supervisor session before publishing Ready. +stop a running workload whose companion is unavailable. The gateway also +requires the authenticated supervisor session before publishing Ready. ## Mounts, GPUs, and configuration diff --git a/crates/openshell-driver-podman/src/driver.rs b/crates/openshell-driver-podman/src/driver.rs index a79d1507f9..c70cc6eca1 100644 --- a/crates/openshell-driver-podman/src/driver.rs +++ b/crates/openshell-driver-podman/src/driver.rs @@ -2002,7 +2002,6 @@ impl PodmanComputeDriver { durable_sandbox: Option<&DriverSandbox>, ) -> Result<(), ComputeDriverError> { let span_status = openshell_otel::ErrorStatusGuard::current(); - let _start = self.lifecycle_event_fences.begin_start(sandbox_id); // Verify before inspecting or modifying either member of the pair. // The durable snapshot is used only if a stopped generation must be // reconciled; it never comes from Podman's mutable inspect response. @@ -2099,11 +2098,8 @@ impl PodmanComputeDriver { self.client .stop_container(&supervisor, self.config.stop_timeout_secs) .await?; - let supervisor_finished_at = self - .wait_for_container_stopped(sandbox_id, &supervisor) + self.wait_for_container_stopped(sandbox_id, &supervisor) .await?; - self.lifecycle_event_fences - .record_supervisor_restart(sandbox_id, supervisor_finished_at.as_deref()); let archive = self .client .copy_from_container(&supervisor, crate::isolation::RESTART_METADATA_PATH) @@ -2326,63 +2322,20 @@ impl PodmanComputeDriver { return Ok(None); }; if entry.state == "running" { - Ok( - watcher::inspect_workload(&self.client, &self.lifecycle_event_fences, &entry.id) - .await - .ok() - .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) - .or_else(|| driver_sandbox_from_list_entry(entry)), - ) + Ok(watcher::inspect_workload(&self.client, &entry.id) + .await + .ok() + .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) + .or_else(|| driver_sandbox_from_list_entry(entry))) } else { - Ok(self.driver_sandbox_from_polled_entry(entry).await) - } - } - - /// Build a polled snapshot from a list entry that was not inspected as a - /// running workload. A workload whose start is in progress is reported as - /// starting unless it shows a new exit, because the gateway applies polled - /// snapshots without waiting for the start to finish. - async fn driver_sandbox_from_polled_entry( - &self, - entry: &ContainerListEntry, - ) -> Option { - if matches!(entry.state.as_str(), "exited" | "stopped") - && let Some(sandbox_id) = entry.labels.get(LABEL_SANDBOX_ID) - && self.lifecycle_event_fences.start_in_progress(sandbox_id) - { - match self.client.inspect_container(&entry.id).await { - Ok(inspect) - if self - .lifecycle_event_fences - .polled_workload_still_starting(sandbox_id, &inspect.state) => - { - debug!( - sandbox_id = %sandbox_id, - container = %entry.id, - status = %inspect.state.status, - "Reporting polled sandbox as starting while its start is in progress" - ); - return watcher::starting_driver_sandbox_from_list_entry(entry); - } - Ok(_) => {} - Err(error) => { - warn!( - sandbox_id = %sandbox_id, - container = %entry.id, - error = %error, - "Failed to inspect stopped container during start, falling back to list entry" - ); - } - } + Ok(driver_sandbox_from_list_entry(entry)) } - driver_sandbox_from_list_entry(entry) } /// List all managed sandboxes. /// - /// Inspects running containers (to get health status) and stopped - /// containers whose start is in progress. Other containers are built - /// directly from the list entry data. + /// Only inspects running containers (to get health status). Non-running + /// containers are built directly from the list entry data. pub async fn list_sandboxes(&self) -> Result, ComputeDriverError> { let entries = self .client @@ -2394,13 +2347,7 @@ impl PodmanComputeDriver { for entry in &entries { if entry.state == "running" { // Running containers need inspect for health check status. - match watcher::inspect_workload( - &self.client, - &self.lifecycle_event_fences, - &entry.id, - ) - .await - { + match watcher::inspect_workload(&self.client, &entry.id).await { Ok(inspect) => { if let Some(sandbox) = driver_sandbox_from_inspect(&inspect) { sandboxes.push(sandbox); @@ -2418,7 +2365,7 @@ impl PodmanComputeDriver { } } // Non-running containers (or inspect fallback): build from list data. - if let Some(sandbox) = self.driver_sandbox_from_polled_entry(entry).await { + if let Some(sandbox) = driver_sandbox_from_list_entry(entry) { sandboxes.push(sandbox); } } @@ -3117,126 +3064,6 @@ mod tests { let _ = fs::remove_file(start_socket); } - const PREVIOUS_EXIT: &str = "2026-08-12T16:39:13.123456789Z"; - const EXITED_WORKLOAD_LIST: &str = r#"[{"Id":"ctr-1","Names":["openshell-sandbox-demo"],"State":"exited","Labels":{"openshell.ai/sandbox-id":"sandbox-1","openshell.ai/sandbox-name":"demo","openshell.ai/sandbox-workspace":"default"}}]"#; - - fn exited_workload_inspect(finished_at: &str) -> StubResponse { - StubResponse::new( - StatusCode::OK, - format!( - r#"{{"Id":"ctr-1","Name":"openshell-sandbox-demo","State":{{"Status":"exited","Running":false,"ExitCode":0,"FinishedAt":"{finished_at}"}},"Config":{{}}}}"# - ), - ) - } - - fn ready_condition(sandbox: Option) -> (String, String) { - let condition = sandbox - .expect("sandbox should be reported") - .status - .expect("sandbox should have a status") - .conditions - .into_iter() - .find(|condition| condition.r#type == "Ready") - .expect("sandbox should have a Ready condition"); - (condition.status, condition.reason) - } - - #[tokio::test] - async fn polled_snapshot_during_restart_does_not_report_the_previous_exit() { - let (socket, requests, handle) = spawn_podman_stub( - "poll-during-start", - vec![ - StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), - exited_workload_inspect(PREVIOUS_EXIT), - StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), - exited_workload_inspect(PREVIOUS_EXIT), - // The workload starts between the list and the inspect. - StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), - StubResponse::new( - StatusCode::OK, - r#"{"Id":"ctr-1","Name":"openshell-sandbox-demo","State":{"Status":"running","Running":true,"StartedAt":"2026-08-12T16:39:14Z","FinishedAt":"2026-08-12T16:39:13.123456789Z"},"Config":{}}"#, - ), - ], - ); - let driver = test_driver(socket.clone()); - driver - .lifecycle_event_fences - .record_previous_exit("sandbox-1", Some(PREVIOUS_EXIT)); - let _start = driver.lifecycle_event_fences.begin_start("sandbox-1"); - let starting = ("False".to_string(), "ContainerStarting".to_string()); - - assert_eq!( - ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), - starting - ); - let mut listed = driver.list_sandboxes().await.unwrap(); - assert_eq!(listed.len(), 1); - assert_eq!(ready_condition(listed.pop()), starting); - assert_eq!( - ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), - starting - ); - - handle.await.unwrap(); - assert_eq!(requests.lock().unwrap().len(), 6); - let _ = fs::remove_file(socket); - } - - #[tokio::test] - async fn polled_snapshot_reports_a_new_exit_and_any_exit_after_start() { - let (socket, requests, handle) = spawn_podman_stub( - "poll-new-exit", - vec![ - StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), - exited_workload_inspect("2026-08-12T16:40:00.5Z"), - // Once the start has returned, the list entry is reported - // without an inspect, including a failed start's previous exit. - StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), - StubResponse::new(StatusCode::OK, EXITED_WORKLOAD_LIST), - ], - ); - let driver = test_driver(socket.clone()); - driver - .lifecycle_event_fences - .record_previous_exit("sandbox-1", Some(PREVIOUS_EXIT)); - let exited = ("False".to_string(), "ContainerExited".to_string()); - - let start = driver.lifecycle_event_fences.begin_start("sandbox-1"); - assert_eq!( - ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), - exited - ); - drop(start); - assert_eq!( - ready_condition(driver.get_sandbox("sandbox-1").await.unwrap()), - exited - ); - let mut listed = driver.list_sandboxes().await.unwrap(); - assert_eq!(ready_condition(listed.pop()), exited); - - handle.await.unwrap(); - assert_eq!(requests.lock().unwrap().len(), 4); - let _ = fs::remove_file(socket); - } - - #[tokio::test] - async fn failed_start_is_no_longer_in_progress() { - let (socket, _requests, handle) = spawn_podman_stub( - "failed-start", - vec![StubResponse::new(StatusCode::OK, "[]")], - ); - let driver = test_driver(socket.clone()); - - let result = driver - .start_sandbox("sandbox-1", "generation-1", &[], None) - .await; - - assert!(matches!(result, Err(ComputeDriverError::NotFound))); - assert!(!driver.lifecycle_event_fences.start_in_progress("sandbox-1")); - handle.await.unwrap(); - let _ = fs::remove_file(socket); - } - #[tokio::test] async fn stop_waits_for_the_container_to_leave_stopping_state() { let (socket, requests, handle) = spawn_podman_stub( diff --git a/crates/openshell-driver-podman/src/watcher.rs b/crates/openshell-driver-podman/src/watcher.rs index c1e77d0dbe..9a508e52c3 100644 --- a/crates/openshell-driver-podman/src/watcher.rs +++ b/crates/openshell-driver-podman/src/watcher.rs @@ -18,7 +18,7 @@ use openshell_core::proto::compute::v1::{ }; use std::collections::HashMap; use std::pin::Pin; -use std::sync::{Arc, Mutex, MutexGuard}; +use std::sync::{Arc, Mutex}; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tracing::{debug, info, warn}; @@ -43,103 +43,30 @@ pub type WatchStream = /// the sandbox from `Starting` to `Error`. #[derive(Clone, Debug, Default)] pub struct LifecycleEventFences { - state: Arc>, -} - -#[derive(Debug, Default)] -struct LifecycleFenceState { - previous_finished_at: HashMap, - starts_in_progress: HashMap, -} - -#[derive(Debug, Default)] -struct InProgressStart { - /// The supervisor exit that the start observed before restarting it. - supervisor_finished_at: Option, -} - -/// Marks a sandbox start as in progress until it is dropped. -#[must_use = "the start is only in progress while this guard is held"] -pub struct StartInProgress { - fences: LifecycleEventFences, - sandbox_id: String, -} - -impl Drop for StartInProgress { - fn drop(&mut self) { - self.fences - .lock() - .starts_in_progress - .remove(&self.sandbox_id); - } + previous_finished_at: Arc>>, } impl LifecycleEventFences { - fn lock(&self) -> MutexGuard<'_, LifecycleFenceState> { - self.state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - } - - pub fn begin_start(&self, sandbox_id: &str) -> StartInProgress { - self.lock() - .starts_in_progress - .insert(sandbox_id.to_string(), InProgressStart::default()); - StartInProgress { - fences: self.clone(), - sandbox_id: sandbox_id.to_string(), - } - } - - pub fn start_in_progress(&self, sandbox_id: &str) -> bool { - self.lock().starts_in_progress.contains_key(sandbox_id) - } - - /// Record the supervisor exit that an in-progress start is about to - /// restart. See [`Self::supervisor_restart_in_progress`]. - pub fn record_supervisor_restart(&self, sandbox_id: &str, finished_at: Option<&str>) { - if let Some(start) = self.lock().starts_in_progress.get_mut(sandbox_id) { - start.supervisor_finished_at = finished_at - .filter(|finished_at| !finished_at.is_empty()) - .map(str::to_string); - } - } - - /// Whether an exited supervisor is the one that the in-progress start is - /// restarting. A restart starts the workload before its supervisor, so the - /// workload must not be contained for that exit. Any other exit is. - pub fn supervisor_restart_in_progress(&self, sandbox_id: &str, state: &ContainerState) -> bool { - if !matches!(state.status.as_str(), "exited" | "stopped") { - return false; - } - let Some(finished_at) = state.finished_at.as_deref() else { - return false; - }; - self.lock() - .starts_in_progress - .get(sandbox_id) - .and_then(|start| start.supervisor_finished_at.as_deref()) - .is_some_and(|recorded| recorded == finished_at) - } - pub fn record_previous_exit(&self, sandbox_id: &str, finished_at: Option<&str>) { - let mut state = self.lock(); + let mut fences = self + .previous_finished_at + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); match finished_at.filter(|finished_at| !finished_at.is_empty()) { Some(finished_at) => { - state - .previous_finished_at - .insert(sandbox_id.to_string(), finished_at.to_string()); + fences.insert(sandbox_id.to_string(), finished_at.to_string()); } None => { - state.previous_finished_at.remove(sandbox_id); + fences.remove(sandbox_id); } } } pub fn remove(&self, sandbox_id: &str) { - let mut state = self.lock(); - state.previous_finished_at.remove(sandbox_id); - state.starts_in_progress.remove(sandbox_id); + self.previous_finished_at + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(sandbox_id); } fn matches_previous_exit( @@ -157,37 +84,12 @@ impl LifecycleEventFences { let Some(finished_at) = state.finished_at.as_deref() else { return false; }; - self.lock() - .previous_finished_at + self.previous_finished_at + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) .get(sandbox_id) .is_some_and(|previous| previous == finished_at) } - - /// Whether a polled workload that the container list reported as not - /// running is still starting rather than failed. - /// - /// The gateway's reconcile sweep polls sandboxes without waiting for an - /// in-progress start. Until the start runs the workload again, inspect - /// still shows the exit recorded before the start, and the workload can - /// also be running again by the time it is inspected. Neither is a failure - /// of the new run. An exit with a different `FinishedAt` is, and once the - /// start has returned every exit is reported. - pub fn polled_workload_still_starting(&self, sandbox_id: &str, state: &ContainerState) -> bool { - let fences = self.lock(); - if !fences.starts_in_progress.contains_key(sandbox_id) { - return false; - } - match state.status.as_str() { - "running" => true, - "exited" | "stopped" => state.finished_at.as_deref().is_some_and(|finished_at| { - fences - .previous_finished_at - .get(sandbox_id) - .is_some_and(|previous| previous == finished_at) - }), - _ => false, - } - } } /// Build a `WatchSandboxesEvent` carrying a sandbox snapshot. @@ -247,7 +149,7 @@ pub async fn start_watch( // health check status — matching the same condition derivation used // for live events. if entry.state == "running" { - match inspect_workload(&client, &lifecycle_event_fences, &entry.id).await { + match inspect_workload(&client, &entry.id).await { Ok(inspect) => { if let Some(sandbox) = driver_sandbox_from_inspect(&inspect) { if tx.send(Ok(sandbox_event(sandbox))).await.is_err() { @@ -363,7 +265,7 @@ async fn map_podman_event( .await .ok()?; let workload = workloads.first()?; - return inspect_workload(client, lifecycle_event_fences, &workload.id) + return inspect_workload(client, &workload.id) .await .ok() .and_then(|inspect| driver_sandbox_from_inspect(&inspect)) @@ -374,7 +276,7 @@ async fn map_podman_event( "remove" => Some(deleted_event(sandbox_id.clone())), "create" | "start" | "stop" | "die" | "health_status" => { // Inspect the container to get current state. - match inspect_workload(client, lifecycle_event_fences, container_id).await { + match inspect_workload(client, container_id).await { Ok(inspect) => { if lifecycle_event_fences.matches_previous_exit( event, @@ -454,13 +356,8 @@ async fn map_podman_event( /// A workload is ready only when its independent supervisor is healthy. This /// check runs both on watch reconciliation and on events, and contains a lost /// supervisor even when the gateway missed the original exit event. -/// -/// A restart starts the workload before its exited companion. The exit that -/// the start recorded is not contained: the companion is about to start, and -/// the start stops the workload itself if the companion fails to start. pub async fn inspect_workload( client: &PodmanClient, - lifecycle_event_fences: &LifecycleEventFences, id: &str, ) -> Result { let mut workload = client.inspect_container(id).await?; @@ -495,19 +392,6 @@ pub async fn inspect_workload( status: "starting".into(), }); } - Ok(supervisor) - if lifecycle_event_fences - .supervisor_restart_in_progress(sandbox_id, &supervisor.state) => - { - debug!( - sandbox_id = %sandbox_id, - workload = %workload.id, - "Not containing workload while its start restarts the supervisor" - ); - workload.state.health = Some(HealthState { - status: "starting".into(), - }); - } // Both containers exist before initial start. A missing or exited // companion therefore requires containment, including after a // gateway restart that missed the original Podman exit event. @@ -609,6 +493,14 @@ pub fn driver_sandbox_from_inspect(inspect: &ContainerInspect) -> Option Option { + let sandbox_id = entry.labels.get(LABEL_SANDBOX_ID)?.clone(); + let sandbox_name = entry + .labels + .get(LABEL_SANDBOX_NAME) + .cloned() + .unwrap_or_default(); + let workspace = entry.labels.get(LABEL_SANDBOX_WORKSPACE).cloned()?; + let (reason, status_str, message) = match entry.state.as_str() { "running" => ( CONDITION_RUNNING, @@ -621,46 +513,6 @@ pub fn driver_sandbox_from_list_entry(entry: &ContainerListEntry) -> Option ("ContainerRemoving", "False", String::new()), _ => ("Unknown", "Unknown", String::new()), }; - driver_sandbox_from_list_entry_with_condition( - entry, - DriverCondition { - r#type: "Ready".to_string(), - status: status_str.to_string(), - reason: reason.to_string(), - message, - transition_time: None, - }, - ) -} - -/// Build a `DriverSandbox` that reports a start in progress for a container -/// list entry. See [`LifecycleEventFences::polled_workload_still_starting`]. -pub fn starting_driver_sandbox_from_list_entry( - entry: &ContainerListEntry, -) -> Option { - driver_sandbox_from_list_entry_with_condition( - entry, - DriverCondition { - r#type: "Ready".to_string(), - status: "False".to_string(), - reason: CONDITION_STARTING.to_string(), - message: "Container start in progress".to_string(), - transition_time: None, - }, - ) -} - -fn driver_sandbox_from_list_entry_with_condition( - entry: &ContainerListEntry, - condition: DriverCondition, -) -> Option { - let sandbox_id = entry.labels.get(LABEL_SANDBOX_ID)?.clone(); - let sandbox_name = entry - .labels - .get(LABEL_SANDBOX_NAME) - .cloned() - .unwrap_or_default(); - let workspace = entry.labels.get(LABEL_SANDBOX_WORKSPACE).cloned()?; Some(build_driver_sandbox( sandbox_id, @@ -668,7 +520,13 @@ fn driver_sandbox_from_list_entry_with_condition( workspace, entry.names.first().cloned().unwrap_or_default(), short_id(&entry.id), - condition, + DriverCondition { + r#type: "Ready".to_string(), + status: status_str.to_string(), + reason: reason.to_string(), + message, + transition_time: None, + }, entry.state == "removing", )) } @@ -783,9 +641,7 @@ mod tests { ], ); let client = PodmanClient::new(path.clone()); - let inspected = inspect_workload(&client, &LifecycleEventFences::default(), "workload") - .await - .unwrap(); + let inspected = inspect_workload(&client, "workload").await.unwrap(); assert!(!inspected.state.running); handle.await.unwrap(); assert!( @@ -816,9 +672,7 @@ mod tests { ], ); let client = PodmanClient::new(path.clone()); - let inspected = inspect_workload(&client, &LifecycleEventFences::default(), "workload") - .await - .unwrap(); + let inspected = inspect_workload(&client, "workload").await.unwrap(); assert!(inspected.state.running); assert_eq!(inspected.state.health.unwrap().status, "starting"); handle.await.unwrap(); @@ -826,85 +680,6 @@ mod tests { let _ = std::fs::remove_file(path); } - #[tokio::test] - async fn supervisor_exit_recorded_by_a_restart_does_not_contain_the_workload() { - use crate::test_utils::{StubResponse, spawn_podman_stub}; - use hyper::StatusCode; - const RECORDED_EXIT: &str = "2026-08-12T16:39:13.123456789Z"; - const RUNNING_WORKLOAD: &str = r#"{"Id":"workload","Name":"workload","State":{"Status":"running","Running":true},"Config":{"Labels":{"openshell.ai/sandbox-id":"test","openshell.ai/isolation-role":"sandbox"}}}"#; - const STOPPED_WORKLOAD: &str = r#"{"Id":"workload","Name":"workload","State":{"Status":"exited","Running":false},"Config":{}}"#; - let exited_supervisor = |finished_at: &str| { - StubResponse::new( - StatusCode::OK, - format!( - r#"{{"Id":"supervisor","Name":"supervisor","State":{{"Status":"exited","Running":false,"ExitCode":0,"FinishedAt":"{finished_at}"}},"Config":{{}}}}"# - ), - ) - }; - let contained = |finished_at: &str| { - vec![ - StubResponse::new(StatusCode::OK, RUNNING_WORKLOAD), - exited_supervisor(finished_at), - StubResponse::new(StatusCode::NO_CONTENT, ""), - StubResponse::new(StatusCode::OK, STOPPED_WORKLOAD), - ] - }; - let mut responses = contained(RECORDED_EXIT); - responses.extend([ - StubResponse::new(StatusCode::OK, RUNNING_WORKLOAD), - exited_supervisor(RECORDED_EXIT), - ]); - responses.extend(contained("2026-08-12T16:39:14Z")); - responses.extend(contained(RECORDED_EXIT)); - let (path, requests, handle) = spawn_podman_stub("restarting-supervisor", responses); - let client = PodmanClient::new(path.clone()); - let fences = LifecycleEventFences::default(); - let stops = || { - requests - .lock() - .unwrap() - .iter() - .filter(|request| request.ends_with("/libpod/containers/workload/stop?timeout=0")) - .count() - }; - - // Before the start has stopped and recorded the supervisor, its exit - // is contained. - let start = fences.begin_start("test"); - let inspected = inspect_workload(&client, &fences, "workload") - .await - .unwrap(); - assert!(!inspected.state.running); - assert_eq!(stops(), 1); - - // A restart starts the workload before the supervisor it recorded. - fences.record_supervisor_restart("test", Some(RECORDED_EXIT)); - let inspected = inspect_workload(&client, &fences, "workload") - .await - .unwrap(); - assert!(inspected.state.running); - assert_eq!(inspected.state.health.unwrap().status, "starting"); - assert_eq!(stops(), 1); - - // A new supervisor exit is contained during the start. - let inspected = inspect_workload(&client, &fences, "workload") - .await - .unwrap(); - assert!(!inspected.state.running); - assert_eq!(stops(), 2); - - // Once the start has returned, the recorded exit is contained. - drop(start); - let inspected = inspect_workload(&client, &fences, "workload") - .await - .unwrap(); - assert!(!inspected.state.running); - assert_eq!(stops(), 3); - - handle.await.unwrap(); - let _ = std::fs::remove_file(path); - } - fn podman_event(action: &str, sandbox_id: &str, time_nano: i64) -> PodmanEvent { podman_event_for_container(action, sandbox_id, "container-1", time_nano) } @@ -994,46 +769,6 @@ mod tests { )); } - #[test] - fn lifecycle_fence_reports_polled_previous_exit_as_starting_only_during_start() { - let fences = LifecycleEventFences::default(); - fences.record_previous_exit("sandbox-1", Some("2026-08-12T16:39:13.123456789Z")); - let previous_exit = ContainerState { - status: "exited".to_string(), - running: false, - exit_code: 0, - oom_killed: false, - health: None, - started_at: Some("2026-08-12T16:38:58Z".to_string()), - finished_at: Some("2026-08-12T16:39:13.123456789Z".to_string()), - supervisor_exit_code: None, - startup_diagnostic: None, - }; - let mut new_exit = previous_exit.clone(); - new_exit.finished_at = Some("2026-08-12T16:39:13.5Z".to_string()); - let mut running = previous_exit.clone(); - running.status = "running".to_string(); - running.running = true; - - // A stopped sandbox, or one whose start has already failed, reports - // the exit. - assert!(!fences.polled_workload_still_starting("sandbox-1", &previous_exit)); - - let start = fences.begin_start("sandbox-1"); - assert!(fences.polled_workload_still_starting("sandbox-1", &previous_exit)); - assert!(fences.polled_workload_still_starting("sandbox-1", &running)); - assert!(!fences.polled_workload_still_starting("sandbox-1", &new_exit)); - assert!(!fences.polled_workload_still_starting("sandbox-2", &previous_exit)); - drop(start); - assert!(!fences.start_in_progress("sandbox-1")); - assert!(!fences.polled_workload_still_starting("sandbox-1", &previous_exit)); - - let _start = fences.begin_start("sandbox-1"); - fences.remove("sandbox-1"); - assert!(!fences.start_in_progress("sandbox-1")); - assert!(!fences.polled_workload_still_starting("sandbox-1", &previous_exit)); - } - #[test] fn condition_healthy_container() { let state = ContainerState { diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 7104fe1aef..24874dd3aa 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -6563,12 +6563,32 @@ impl ComputeRuntime { self.sandbox_watch_bus.remove(sandbox_id); } + /// Lifecycle gate for a sweep that observes a `Starting` sandbox. + /// + /// `StartSandbox` holds the gate for the whole driver call, during which + /// the workload can still show its previous exit. The sweep skips the + /// sandbox (`Err`) instead of waiting, and the next tick or watch event + /// reports the settled state. The sweep holds the gate while it polls. + fn sweep_start_gate( + &self, + record: &ObjectRecord, + ) -> Result, String> { + let sandbox = decode_sandbox_record(record)?; + if sandbox.phase() != SandboxPhase::Starting as i32 { + return Ok(None); + } + self.lifecycle_gates + .try_lock_for(sandbox.object_id()) + .map(Some) + .ok_or_else(|| "start in progress".to_string()) + } + async fn reconcile_snapshot_sandbox( &self, snapshot: DriverSandbox, sweep_started_at_ms: i64, ) -> Result<(), String> { - let expected_resource_version = { + let (expected_resource_version, _start_gate) = { let _guard = self.lock_sandbox_local(&snapshot.id).await; let Some(existing) = self .store @@ -6582,7 +6602,10 @@ impl ComputeRuntime { if existing.updated_at_ms > sweep_started_at_ms { return Ok(()); } - existing.resource_version + let Ok(start_gate) = self.sweep_start_gate(&existing) else { + return Ok(()); + }; + (existing.resource_version, start_gate) }; let Some(current) = self @@ -6617,7 +6640,7 @@ impl ComputeRuntime { sweep_started_at_ms: i64, grace_ms: i64, ) -> Result<(), String> { - let (sandbox_id, sandbox_name, expected_resource_version, age_ms) = { + let (sandbox_id, sandbox_name, expected_resource_version, age_ms, _start_gate) = { let _guard = self.lock_sandbox_local(&record.id).await; let Some(current_record) = self .store @@ -6645,12 +6668,16 @@ impl ComputeRuntime { if age_ms < grace_ms { return Ok(()); } + let Ok(start_gate) = self.sweep_start_gate(¤t_record) else { + return Ok(()); + }; ( sandbox.object_id().to_string(), sandbox.object_name().to_string(), current_record.resource_version, age_ms, + start_gate, ) }; @@ -14068,16 +14095,12 @@ mod tests { ); } - // Gateway side of the Podman start contract. The reconcile sweep applies - // polled snapshots without waiting for an in-progress start. While a start - // is in progress, the Podman driver reports a stopped workload that still - // shows its previous exit as ContainerStarting. Such a snapshot keeps the - // sandbox Starting, and the start completes. This passes without the - // #5258 fix; the driver test - // polled_snapshot_during_restart_does_not_report_the_previous_exit covers - // the regression. + // Regression for #5258. The reconcile sweep polls the driver without + // waiting for a start, and a restarting workload still shows its previous + // exit until the driver starts it. The sweep skips a sandbox whose start + // holds the lifecycle gate; the settled state is reported afterwards. #[tokio::test] - async fn reconcile_snapshot_during_start_keeps_the_sandbox_starting() { + async fn reconcile_sweep_skips_a_sandbox_while_its_start_is_in_progress() { let driver = ControlledDriver::new(); driver.block_start(); let sandbox = sandbox_record( @@ -14087,8 +14110,8 @@ mod tests { ); let mut polled = ready_driver_sandbox(sandbox.object_id(), sandbox.object_name()); polled.status = Some(make_driver_status(make_driver_condition( - "ContainerStarting", - "Container start in progress", + "ContainerExited", + "Container exited with code 0", ))); driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(polled.clone()))); let runtime = test_runtime(driver.clone()).await; @@ -14122,13 +14145,6 @@ mod tests { driver.release_start(); start.await.unwrap().unwrap(); - let stored = runtime - .store - .get_message::(sandbox.object_id()) - .await - .unwrap() - .unwrap(); - assert_eq!(stored.phase(), SandboxPhase::Starting as i32); } #[tokio::test]