diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2b168f26df..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,6 +14095,58 @@ mod tests { ); } + // 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_sweep_skips_a_sandbox_while_its_start_is_in_progress() { + 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( + "ContainerExited", + "Container exited with code 0", + ))); + 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(); + } + #[tokio::test] async fn live_container_exit_during_start_still_transitions_to_error() { for reason in [