diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2b168f26df..b6bfc9062d 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -6563,6 +6563,22 @@ impl ComputeRuntime { self.sandbox_watch_bus.remove(sandbox_id); } + /// Whether a sweep observes a `Starting` sandbox with an active start. + /// + /// `StartSandbox` holds the gate for the whole driver call, during which + /// the workload can still show its previous exit. The sweep skips the + /// sandbox instead of waiting, and the next tick or watch event + /// reports the settled state. Probe before and after polling without + /// retaining the gate, so Deleted watches can still dispatch cleanup. + fn sweep_start_in_progress(&self, record: &ObjectRecord) -> Result { + let sandbox = decode_sandbox_record(record)?; + Ok(sandbox.phase() == SandboxPhase::Starting as i32 + && self + .lifecycle_gates + .try_lock_for(sandbox.object_id()) + .is_none()) + } + async fn reconcile_snapshot_sandbox( &self, snapshot: DriverSandbox, @@ -6582,6 +6598,9 @@ impl ComputeRuntime { if existing.updated_at_ms > sweep_started_at_ms { return Ok(()); } + if self.sweep_start_in_progress(&existing)? { + return Ok(()); + } existing.resource_version }; @@ -6603,6 +6622,7 @@ impl ComputeRuntime { }; if existing.resource_version != expected_resource_version || existing.updated_at_ms > sweep_started_at_ms + || self.sweep_start_in_progress(&existing)? { return Ok(()); } @@ -6645,6 +6665,9 @@ impl ComputeRuntime { if age_ms < grace_ms { return Ok(()); } + if self.sweep_start_in_progress(¤t_record)? { + return Ok(()); + } ( sandbox.object_id().to_string(), @@ -6667,6 +6690,7 @@ impl ComputeRuntime { }; if current_record.resource_version != expected_resource_version || current_record.updated_at_ms > sweep_started_at_ms + || self.sweep_start_in_progress(¤t_record)? { return Ok(()); } @@ -14068,6 +14092,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 [ @@ -15888,51 +15964,86 @@ mod tests { #[tokio::test] async fn blocked_reconciliation_lookup_does_not_delay_watch_events() { - let driver = ControlledDriver::new(); - driver.block_get(); - let snapshot = ready_driver_sandbox("sb-a", "sandbox-a"); - driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(snapshot.clone()))); - let runtime = test_runtime(driver.clone()).await; - for sandbox in [ - sandbox_record("sb-a", "sandbox-a", SandboxPhase::Provisioning), - sandbox_record("sb-b", "sandbox-b", SandboxPhase::Ready), + for (observation, deleted_id) in [ + ("snapshot", "sb-b"), + ("snapshot", "sb-a"), + ("prune", "sb-a"), ] { - runtime.store.put_message(&sandbox).await.unwrap(); - } + let driver = ControlledDriver::new(); + driver.block_get(); + let snapshot = ready_driver_sandbox("sb-a", "sandbox-a"); + driver.set_get_outcome(if deleted_id == "sb-b" { + ControlledGetOutcome::Sandbox(Box::new(snapshot.clone())) + } else if observation == "snapshot" { + ControlledGetOutcome::Missing + } else { + ControlledGetOutcome::Error("lookup failed after the Deleted watch") + }); + let runtime = test_runtime(driver.clone()).await; + for sandbox in [ + sandbox_record("sb-a", "sandbox-a", SandboxPhase::Starting), + sandbox_record("sb-b", "sandbox-b", SandboxPhase::Ready), + ] { + runtime.store.put_message(&sandbox).await.unwrap(); + } - let (shutdown_tx, watch_handle) = start_watch_loop(&runtime, &driver).await; - let mut sandbox_b_rx = runtime.sandbox_watch_bus.subscribe("sb-b"); - let reconcile_runtime = runtime.clone(); - let reconcile = tokio::spawn(async move { - reconcile_runtime - .reconcile_snapshot_sandbox(snapshot, i64::MAX) + let (shutdown_tx, watch_handle) = start_watch_loop(&runtime, &driver).await; + let mut deleted_rx = runtime.sandbox_watch_bus.subscribe(deleted_id); + let reconcile_runtime = runtime.clone(); + let reconcile = tokio::spawn(async move { + if observation == "prune" { + let record = reconcile_runtime + .store + .get(Sandbox::object_type(), "sb-a") + .await + .unwrap() + .unwrap(); + return reconcile_runtime + .prune_missing_sandbox(record, i64::MAX, 0) + .await; + } + reconcile_runtime + .reconcile_snapshot_sandbox(snapshot, i64::MAX) + .await + }); + tokio::time::timeout(Duration::from_secs(1), driver.get_started.notified()) .await - }); - tokio::time::timeout(Duration::from_secs(1), driver.get_started.notified()) - .await - .expect("reconciliation did not reach the driver lookup"); + .expect("reconciliation did not reach the driver lookup"); - driver.send_event(deleted_watch_event("sb-b")); - tokio::time::timeout(Duration::from_secs(1), sandbox_b_rx.recv()) - .await - .expect("watch event was blocked by reconciliation lookup") - .expect("sandbox B watch bus closed before notification"); - assert!( - runtime - .store - .get_message::("sb-b") + driver.send_event(deleted_watch_event(deleted_id)); + tokio::time::timeout(Duration::from_secs(1), deleted_rx.recv()) + .await + .expect("watch event was blocked by reconciliation lookup") + .expect("sandbox watch bus closed before notification"); + assert!( + runtime + .store + .get_message::(deleted_id) + .await + .unwrap() + .is_none() + ); + let cleanup = + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await; + driver.release_get(); + let reconciled = tokio::time::timeout(Duration::from_secs(1), reconcile) .await .unwrap() - .is_none() - ); - - driver.release_get(); - tokio::time::timeout(Duration::from_secs(1), reconcile) - .await - .unwrap() - .unwrap() - .unwrap(); - stop_watch_loop(shutdown_tx, watch_handle).await; + .unwrap(); + if observation == "prune" { + assert!( + reconciled + .unwrap_err() + .contains("lookup failed after the Deleted watch") + ); + } else { + reconciled.unwrap(); + } + cleanup.expect("the sweep must not suppress deletion-watch driver cleanup"); + assert_eq!(driver.delete_calls(), 1); + stop_watch_loop(shutdown_tx, watch_handle).await; + } } #[tokio::test]