diff --git a/architecture-baseline.toml b/architecture-baseline.toml index 8a1b4cffdb..f0d7208f7b 100644 --- a/architecture-baseline.toml +++ b/architecture-baseline.toml @@ -81,7 +81,7 @@ "crates/libs/sbx/mgr/server/src/db/state_transitions.rs" = 1140 "crates/libs/sbx/mgr/server/src/run.rs" = 1101 "crates/libs/sbx/mgr/server/src/sandbox_eviction.rs" = 1273 -"crates/libs/sbx/mgr/server/src/service/activation_support.rs" = 1592 +"crates/libs/sbx/mgr/server/src/service/activation_support.rs" = 1574 "crates/libs/sbx/mgr/server/src/service/auth_lookup.rs" = 1221 "crates/libs/sbx/mgr/server/src/service/cache_cleanup.rs" = 1455 "crates/libs/sbx/mgr/server/src/service/cleanup.rs" = 2704 diff --git a/crates/libs/sbx/mgr/server/src/db/state_write.rs b/crates/libs/sbx/mgr/server/src/db/state_write.rs index 827dd23db3..b81a4d2c5b 100644 --- a/crates/libs/sbx/mgr/server/src/db/state_write.rs +++ b/crates/libs/sbx/mgr/server/src/db/state_write.rs @@ -13,8 +13,7 @@ //! - A specialized compare-and-set writer's from-set is its SQL predicate, //! narrowed to the states its production callers reach. //! - A generic `update_state` caller writes from any state the generic guards -//! admit. Its edge lists the states the caller relies on, and -//! [`EdgeKind::KnownRace`] edges list the other states it can overwrite. +//! admit. Its edge lists the states the caller relies on. //! - An empty from-set is an insert: no row existed. //! - A write over a row already in the target state changes no state and has //! no edge. @@ -110,11 +109,6 @@ pub enum EdgeKind { Compensation, /// Startup, inventory, quarantine or cleanup repair. Recovery, - /// A stale or unvalidated write a generic `update_state` caller can make: - /// a concurrent transition moved the row after the caller read it, or the - /// caller never checked the state. Not intended, and `update_state_from` - /// refuses to take one: converting the caller removes the edge. - KnownRace, } /// One family of persisted transitions: every state in `from` to `to`. @@ -231,14 +225,15 @@ edge_table! { // Pause and resident resume. - /// `pause` and the suspend auto-pause: `update_state(Pausing)` after - /// `transition(Pausing)` on the read. + /// `pause` and the suspend auto-pause, after `transition(Pausing)` on the + /// read. PAUSE_BEGIN: Request, of(&[Started]) => Pausing; - /// `pause` and the suspend auto-pause: `update_state(Paused)`. + /// `pause` and the suspend auto-pause, after node-manager confirms the + /// pause. PAUSE_COMPLETE: Completion, of(&[Pausing]) => Paused; - /// `update_state(Started)` after a failed guest preparation in `pause` - /// and the suspend auto-pause; `update_state_best_effort(Started)` after a - /// node connect failure and in `pause_with_reconciliation`. + /// `pause` and the suspend auto-pause after a failed guest preparation; + /// `pause` after a node connect failure, and `pause_with_reconciliation`, + /// best effort. PAUSE_ROLLBACK: Compensation, of(&[Pausing]) => Started; /// `begin_resident_resume_with_policy`, and `begin_resident_resume`'s /// `update_state_from` without an activation policy. @@ -425,14 +420,6 @@ edge_table! { /// `restore_adopt_orphan_or_insert_sandbox_in_transaction`'s insert /// branch from orphan materialization of an interrupted restore. ORPHAN_RESTORE_INSERT: Recovery, INSERT => Restoring; - - // Known races. - - /// Pause rollback (`update_state(Started)`, best effort or not) after a - /// concurrent Stop settled `stopped`. - PAUSE_ROLLBACK_AFTER_STOP: KnownRace, of(&[Stopped]) => Started; - /// Pause completion (`update_state(Paused)`) after a concurrent Stop. - PAUSE_COMPLETE_AFTER_STOP: KnownRace, of(&[Stopped]) => Paused; } pub use edges::EDGES; @@ -522,8 +509,8 @@ pub enum StateWriteOutcome { /// autocommit it opens its own `IMMEDIATE` transaction, so the guard reads, /// the write and the job failure commit together; it never nests one. /// -/// Inserts and [`EdgeKind::KnownRace`] edges, and targets `update_state` -/// rejects (`quarantined`, `failed`), are `InvalidQuery`. A write that matches +/// Inserts, and targets `update_state` rejects (`quarantined`, `failed`), are +/// `InvalidQuery`. A write that matches /// no row although the row was read in a source state is /// `StatementChangedRows(0)`. pub fn update_state_from( @@ -531,7 +518,7 @@ pub fn update_state_from( id: &str, edge: &'static Edge, ) -> Result { - if edge.kind == EdgeKind::KnownRace || edge.from.is_empty() { + if edge.from.is_empty() { return Err(rusqlite::Error::InvalidQuery); } write_state_from(conn, id, edge.from, edge.to) diff --git a/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs b/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs index 533e7d4bb1..4d369200fb 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs @@ -14,11 +14,11 @@ use crate::state::SandboxState::{self, *}; enum Disagreement { /// A request edge writes the pair; `can_transition_to` forbids it. RequestForbidden, - /// Only completion, compensation, recovery or known-race edges write the - /// pair; `can_transition_to` forbids it. + /// Only completion, compensation or recovery edges write the pair; + /// `can_transition_to` forbids it. WriteForbidden, /// `can_transition_to` allows the pair, but only completion, - /// compensation, recovery or known-race edges write it. + /// compensation or recovery edges write it. AllowedNotRequested, /// `can_transition_to` allows the pair; no writer implements it. AllowedUnwritten, @@ -64,9 +64,7 @@ const ALLOWLIST: &[(SandboxState, SandboxState, Disagreement, &str)] = &[ (Stopping, CleanupRequired, WriteForbidden, "an interrupted stop is handed to cleanup"), (Stopping, CleanupFailed, WriteForbidden, "stop cleanup failed"), (Stopping, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), - (Stopped, Started, WriteForbidden, "known race: pause rollback after Stop"), (Stopped, Stopping, RequestForbidden, STOP_SETTLED_SOURCE), - (Stopped, Paused, WriteForbidden, "known race: pause completion after Stop"), (Stopped, Suspended, RequestForbidden, "warm start and process-snapshot rollback restage a stopped row"), (Stopped, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), (Deleting, CleanupRequired, WriteForbidden, "an interrupted delete is handed to cleanup"), diff --git a/crates/libs/sbx/mgr/server/src/db/tests/update_state_from.rs b/crates/libs/sbx/mgr/server/src/db/tests/update_state_from.rs index 68434a78f4..fb7f47b68d 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/update_state_from.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/update_state_from.rs @@ -11,8 +11,7 @@ use super::update_state_matrix::{ sweep_lifecycle_pairs_and_cleanup_shapes, sweep_quarantine_episode_and_missing_row, write_from, }; use crate::db::{ - EDGES, EdgeKind, StateSet, StateWriteOutcome, StateWriteRefusal, edges, test_conn, - update_state_from, + EDGES, StateSet, StateWriteOutcome, StateWriteRefusal, edges, test_conn, update_state_from, }; use crate::state::SandboxState; @@ -108,8 +107,7 @@ fn a_from_set_without_the_row_state_refuses_and_keeps_every_row() { } fn writable(edge: &crate::db::Edge) -> bool { - edge.kind() != EdgeKind::KnownRace - && !edge.from().is_empty() + !edge.from().is_empty() && !matches!(edge.to(), SandboxState::Quarantined | SandboxState::Failed) } @@ -137,25 +135,15 @@ fn an_edge_writes_its_target_only_from_its_sources() { } } -/// Known races, inserts and quarantine or failure targets are refused before -/// the row is read. Known races are tried over a source the generic guards -/// admit; inserts and quarantine targets over `started`. +/// Inserts and quarantine or failure targets are refused before the row is +/// read. Each is tried over `started`. #[test] -fn known_race_insert_and_rejected_target_edges_are_refused() { +fn insert_and_rejected_target_edges_are_refused() { let refused: Vec<_> = EDGES.iter().filter(|edge| !writable(edge)).collect(); - assert!( - refused - .iter() - .any(|edge| edge.kind() == EdgeKind::KnownRace) - ); assert!(refused.iter().any(|edge| edge.from().is_empty())); for edge in refused { let to = edge.to(); - let from = edge - .from() - .iter() - .find(|from| cell(PLAIN[index(*from)], to) == Expected::Applied) - .unwrap_or(SandboxState::Started); + let from = SandboxState::Started; let case = format!("{}: {from} -> {to}", edge.name()); let conn = test_conn(); let id = seed( diff --git a/crates/libs/sbx/mgr/server/src/service/activation_support.rs b/crates/libs/sbx/mgr/server/src/service/activation_support.rs index 821742a059..a54a1c8bed 100644 --- a/crates/libs/sbx/mgr/server/src/service/activation_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/activation_support.rs @@ -1816,42 +1816,6 @@ fn node_manager_failure_is_terminal(code: tonic::Code) -> bool { } } -pub(super) fn update_state_best_effort( - db: &Arc>, - sandbox_id: &str, - state: SandboxState, -) { - let updated = { - let Ok(db) = crate::node_state_sync::connection_or_signal_failure(db) else { - return; - }; - match crate::db::update_state(&db, sandbox_id, state) { - Ok(outcome) if outcome.is_applied() => true, - Ok(outcome) => { - tracing::warn!( - sandbox_id, - state = state.as_str(), - %outcome, - "sandbox state write was superseded" - ); - false - } - Err(error) => { - tracing::warn!( - sandbox_id, - state = state.as_str(), - %error, - "failed to update sandbox state" - ); - false - } - } - }; - if updated { - mark_sandbox_dirty(db, sandbox_id); - } -} - pub(super) async fn notify_resume_via_spaced_host( sandbox_id: &str, reason: pplx_space_spaced_host_client::proto::ResumeReason, diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs index 1d776f7a22..4153930c54 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs @@ -327,9 +327,9 @@ impl StateWriteRejected { } } -/// `update_state_best_effort` for [`crate::db::update_state_from`]: a refusal -/// or an error is logged and leaves the row as it is, and only an applied -/// write marks the sandbox dirty. `None` is a poisoned connection or a +/// The best-effort form of [`crate::db::update_state_from`]: a refusal or an +/// error is logged and leaves the row as it is, and only an applied write +/// marks the sandbox dirty. `None` is a poisoned connection or a /// database error. pub(super) fn update_state_from_best_effort( db: &Arc>, diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs index d68a35ae68..e41d515cfd 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs @@ -5,7 +5,7 @@ use super::activation_support::{ grpc_timeout_budget, operation_activation_attempt_id_from_wire, sandbox_event_trigger, space_api_route_identity_for_runtime_activation, space_api_route_identity_from_wake_start_context, space_api_route_identity_to_wire, - update_state_best_effort, wake_target_state_from_proto, + wake_target_state_from_proto, }; use super::cleanup::{ AsyncCleanupRequest, CleanupAdmission, CleanupDispatchDisposition, SynchronousCleanupRequest, diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume.rs index 072430c2d1..ea731f1b53 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume.rs @@ -2,11 +2,70 @@ //! that suspend preparation reuses. use super::*; +use crate::db::{Edge, edges}; +use crate::service::cleanup_support::{ + StateWriteRejected, require_state_write_from, state_write_result, update_state_from_best_effort, +}; const PAUSE_RECONCILIATION_ATTEMPTS: usize = 3; const PAUSE_RECONCILIATION_BACKOFF: Duration = Duration::from_millis(100); +/// Whether `rejected` found the row outside `edge`'s source states: another +/// operation, such as a Stop the cleanup fence admitted while pause held only +/// the artifact lock, moved the node-local SQLite `sandboxes.state` after +/// pause read it, and now owns the row. Decided on the found state, because +/// the generic guards answer a `stopping` row before the from-set does. +fn superseded(rejected: StateWriteRejected, edge: &Edge) -> bool { + match rejected { + StateWriteRejected::FromState { .. } => true, + StateWriteRejected::Guard { found, .. } => !edge.from().contains(found), + StateWriteRejected::Missing => false, + } +} + +/// Roll a `pausing` row back to `started`, best effort. Returns the rejection +/// when another operation moved the row out of `pausing`. +fn roll_back_pause_best_effort( + db: &Arc>, + sandbox_id: &str, +) -> Option { + let edge = &edges::PAUSE_ROLLBACK; + let rejected = + state_write_result(update_state_from_best_effort(db, sandbox_id, edge)?).err()?; + superseded(rejected, edge).then_some(rejected) +} + +#[cfg(test)] +type GuestPreparation = Arc< + dyn Fn( + String, + ) -> futures_util::future::BoxFuture< + 'static, + Result<(), crate::guest_filesystem_flush::GuestFlushError>, + > + Send + + Sync, +>; + +#[cfg(test)] +tokio::task_local! { + /// Replaces pause's guest snapshot preparation, which dials the host's + /// per-sandbox spaced-host socket, and lets a test act between pause's + /// state writes. + static GUEST_PREPARATION: GuestPreparation; +} + +async fn prepare_guest_for_pause( + sandbox_id: &str, + budget: GuestSnapshotPreparationBudget, +) -> Result<(), crate::guest_filesystem_flush::GuestFlushError> { + #[cfg(test)] + if let Ok(prepare) = GUEST_PREPARATION.try_with(Clone::clone) { + return prepare(sandbox_id.to_owned()).await; + } + prepare_guest_for_vm_snapshot_via_spaced_host(sandbox_id, budget).await +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(super) enum PauseReconciliationOutcome { Paused, @@ -53,6 +112,12 @@ async fn reconcile_state_after_unconfirmed_pause( outcome } +/// Pause the VM of a `pausing` row, reconciling unconfirmed responses against +/// the observed VM state. Returns `paused` for the caller to complete, or the +/// state the row was left in: `started` after a rollback, or `quarantined` +/// when the VM state stays unknown. `Err` is the rejection when the rollback +/// or the quarantine finds that another operation moved the row out of +/// `pausing`; that operation owns the row. #[expect( clippy::too_many_arguments, reason = "each argument is an independent pause input; the node-manager client and connector are injected" @@ -66,7 +131,7 @@ pub(super) async fn pause_with_reconciliation( guest_preparation_budget: GuestSnapshotPreparationBudget, node_manager: &crate::node_manager_transport::NodeManagerConnector, balloon_hint_timeout_ms: Option, -) -> SandboxState { +) -> Result { let sandbox_id_string = sandbox_id.to_string(); let mut last_reason = None; for attempt in 0..PAUSE_RECONCILIATION_ATTEMPTS { @@ -82,7 +147,7 @@ pub(super) async fn pause_with_reconciliation( }) .await { - Ok(_) => return SandboxState::Paused, + Ok(_) => return Ok(SandboxState::Paused), Err(error) => { let reason = format!("{reason_context}: {error}"); last_reason = Some(reason.clone()); @@ -92,12 +157,14 @@ pub(super) async fn pause_with_reconciliation( let retry_requires_prepare = matches!(&outcome, PauseReconciliationOutcome::Running); match outcome { - PauseReconciliationOutcome::Paused => return SandboxState::Paused, + PauseReconciliationOutcome::Paused => return Ok(SandboxState::Paused), PauseReconciliationOutcome::Running if attempt + 1 == PAUSE_RECONCILIATION_ATTEMPTS => { - update_state_best_effort(db, &sandbox_id_string, SandboxState::Started); - return SandboxState::Started; + return match roll_back_pause_best_effort(db, &sandbox_id_string) { + Some(rejected) => Err(rejected), + None => Ok(SandboxState::Started), + }; } PauseReconciliationOutcome::Running => {} PauseReconciliationOutcome::Unknown(detail) => { @@ -106,19 +173,19 @@ pub(super) async fn pause_with_reconciliation( } if attempt + 1 < PAUSE_RECONCILIATION_ATTEMPTS && retry_requires_prepare - && let Err(error) = prepare_guest_for_vm_snapshot_via_spaced_host( - &sandbox_id_string, - guest_preparation_budget, - ) - .await + && let Err(error) = + prepare_guest_for_pause(&sandbox_id_string, guest_preparation_budget).await { - update_state_best_effort(db, &sandbox_id_string, SandboxState::Started); + let superseded = roll_back_pause_best_effort(db, &sandbox_id_string); tracing::warn!( sandbox_id = %sandbox_id, %error, "pause retry guest preparation failed" ); - return SandboxState::Started; + return match superseded { + Some(rejected) => Err(rejected), + None => Ok(SandboxState::Started), + }; } } } @@ -128,14 +195,33 @@ pub(super) async fn pause_with_reconciliation( } let reason = last_reason.unwrap_or_else(|| reason_context.to_string()); - mark_quarantined_best_effort( + if !mark_quarantined_best_effort( db, &sandbox_id_string, SandboxState::Pausing, SandboxQuarantineReasonCode::PauseFailed, &reason, - ); - SandboxState::Quarantined + ) { + // The quarantine compare-and-set refuses a row that left `pausing`, + // which is how a Stop that killed the VM mid-pause surfaces here. It + // answers false only when its own write did not apply, so a + // `quarantined` row was quarantined by another writer. A failed read + // or a missing row still reports pause's quarantine. + let found = crate::node_state_sync::connection_or_signal_failure(db) + .ok() + .and_then(|db| { + crate::db::get_sandbox(&db, &sandbox_id_string) + .ok() + .flatten() + }) + .map(|row| row.state); + if let Some(found) = found + && found != SandboxState::Pausing + { + return Err(StateWriteRejected::FromState { found }); + } + } + Ok(SandboxState::Quarantined) } impl SandboxManagerService { @@ -170,26 +256,38 @@ impl SandboxManagerService { .map_err(Status::from)?; { + // `lookup_sandbox` and this write take the database mutex + // separately, so another worker (cleanup admission's blocking + // task) can commit in between. The edge's `started` from-set + // refuses whatever it left. let db = self.db_conn()?; - require_state_write( - crate::db::update_state(&db, &sandbox.id, SandboxState::Pausing), + require_state_write_from( + crate::db::update_state_from(&db, &sandbox.id, &edges::PAUSE_BEGIN), "begin pause", )?; } self.watchdog.record_lifecycle_edge(&sandbox.id).await; - if let Err(e) = prepare_guest_for_vm_snapshot_via_spaced_host( - &sandbox.id, - GuestSnapshotPreparationBudget::Standard, - ) - .await + if let Err(e) = + prepare_guest_for_pause(&sandbox.id, GuestSnapshotPreparationBudget::Standard).await { + const CONTEXT: &str = "roll back failed pause preparation"; let db = self.db_conn()?; - if let Err(error) = require_state_write( - crate::db::update_state(&db, &sandbox.id, SandboxState::Started), - "roll back failed pause preparation", - ) { - tracing::warn!(sandbox_id = %sandbox.id, %error, "failed to roll back pause state"); + let edge = &edges::PAUSE_ROLLBACK; + match crate::db::update_state_from(&db, &sandbox.id, edge).map(state_write_result) { + Ok(Ok(_)) => {} + // Guest preparation fails when a concurrent Stop kills the + // VM; that Stop, not the preparation, decided the outcome. + Ok(Err(rejected)) if superseded(rejected, edge) => { + return Err(rejected.into_status(CONTEXT)); + } + Ok(Err(rejected)) => { + let error = rejected.into_status(CONTEXT); + tracing::warn!(sandbox_id = %sandbox.id, %error, "failed to roll back pause state"); + } + Err(error) => { + tracing::warn!(sandbox_id = %sandbox.id, %error, "failed to roll back pause state"); + } } return Err(Status::internal(format!( "guest snapshot preparation before pause failed: {e}" @@ -199,7 +297,11 @@ impl SandboxManagerService { let mut node_client = match connect_node(&self.node_manager).await { Ok(node_client) => node_client, Err(e) => { - update_state_best_effort(&self.db, &sandbox.id, SandboxState::Started); + if let Some(rejected) = roll_back_pause_best_effort(&self.db, &sandbox.id) { + return Err( + rejected.into_status("roll back pause after node-manager connect failure") + ); + } return Err(Status::internal(format!( "cannot connect to node-manager: {e}" ))); @@ -221,8 +323,12 @@ impl SandboxManagerService { None, ) .await; - if pause_state != SandboxState::Paused { + if pause_state != Ok(SandboxState::Paused) { self.watchdog.record_lifecycle_edge(&sandbox.id).await; + } + let pause_state = + pause_state.map_err(|rejected| rejected.into_status("reconcile unconfirmed pause"))?; + if pause_state != SandboxState::Paused { return Err(Status::internal(format!( "pause failed; sandbox reconciled to {pause_state}" ))); @@ -230,8 +336,8 @@ impl SandboxManagerService { { let db = self.db_conn()?; - require_state_write( - crate::db::update_state(&db, &sandbox.id, SandboxState::Paused), + require_state_write_from( + crate::db::update_state_from(&db, &sandbox.id, &edges::PAUSE_COMPLETE), "complete pause", )?; } @@ -540,3 +646,6 @@ impl SandboxManagerService { Ok(response) } } + +#[cfg(test)] +mod supersession_tests; diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume/supersession_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume/supersession_tests.rs new file mode 100644 index 0000000000..6eb353eeb1 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/pause_resume/supersession_tests.rs @@ -0,0 +1,425 @@ +//! The Pause RPC against an operation that moves the node-local SQLite +//! `sandboxes.state` row while pause is between its state writes. +//! +//! Pause holds only the artifact lock while it awaits guest preparation and +//! node-manager. Stop takes the cleanup fence and admits a `pausing` row, so a +//! Stop can settle `stopped` in that window (plan finding 1). Every rollback, +//! completion and quarantine write pause makes then applies only over +//! `pausing`: any other row belongs to the operation that moved it, and Pause +//! answers `Aborted` without writing. +//! +//! [`GUEST_PREPARATION`] replaces guest preparation, which would dial the +//! host's spaced-host socket, and runs each test's interleaving at the +//! preparation call that precedes the write under test. node-manager is +//! unbound or a test-served fake. + +use std::sync::atomic::{AtomicUsize, Ordering}; + +use super::*; +use crate::db::tests::state_write_fixtures::{Snapshot, assert_unchanged, force_state, snapshot}; +use crate::guest_filesystem_flush::GuestFlushError; +use crate::node_manager_transport::fake::{FakeNodeManager, ServedNodeManager}; +use crate::service::activation_support::internal_request; +use crate::service::tests::service_tests::insert_test_sandbox_with_runtime; +use pplx_space_node_mgr_client::proto::{ + FirecrackerVmState as WireFirecrackerVmState, GetFirecrackerVmStateRequest, + GetFirecrackerVmStateResponse, PauseSandboxRequest, PauseSandboxResponse, +}; + +const WAIT: Duration = Duration::from_secs(30); + +/// The pause write a case reaches after `begin pause`. +#[derive(Clone, Copy, Debug)] +enum Exit { + /// Guest preparation fails, and the RPC rolls back to `started`. + PreparationFails, + /// node-manager is unreachable, and the RPC rolls back, best effort. + NodeUnreachable, + /// node-manager confirms the pause, and the RPC completes `paused`. + NodeConfirms, + /// The node pause fails with the VM running and the retry's guest + /// preparation fails: `pause_with_reconciliation` rolls back, best effort. + RetryPreparationFails, + /// Every node pause attempt fails with the VM running: + /// `pause_with_reconciliation` rolls back, best effort. + VmStaysRunning, + /// Every node pause attempt fails and the VM state cannot be read, as + /// when a Stop killed the VM: `pause_with_reconciliation` quarantines. + VmUnobservable, +} + +const EXITS: [Exit; 6] = [ + Exit::PreparationFails, + Exit::NodeUnreachable, + Exit::NodeConfirms, + Exit::RetryPreparationFails, + Exit::VmStaysRunning, + Exit::VmUnobservable, +]; + +impl Exit { + /// The guest preparation call after which the exit's write runs: pause's + /// own is the first, and each reconciliation retry adds one. + fn interleaved_preparation(self) -> usize { + match self { + Self::PreparationFails + | Self::NodeUnreachable + | Self::NodeConfirms + | Self::VmUnobservable => 1, + Self::RetryPreparationFails => 2, + Self::VmStaysRunning => 3, + } + } + + fn preparation_fails(self) -> bool { + match self { + Self::PreparationFails | Self::RetryPreparationFails => true, + Self::NodeUnreachable + | Self::NodeConfirms + | Self::VmStaysRunning + | Self::VmUnobservable => false, + } + } + + fn node_manager(self) -> Option { + match self { + Self::PreparationFails | Self::NodeUnreachable => None, + Self::NodeConfirms => Some( + FakeNodeManager::default() + .unary("PauseSandbox", |_: PauseSandboxRequest| { + Ok(PauseSandboxResponse {}) + }) + .serve(), + ), + Self::RetryPreparationFails | Self::VmStaysRunning => Some( + FakeNodeManager::default() + .unary("PauseSandbox", |_: PauseSandboxRequest| { + Err::(Status::unavailable("test pause failed")) + }) + .unary( + "GetFirecrackerVmState", + |_: GetFirecrackerVmStateRequest| { + Ok(GetFirecrackerVmStateResponse { + state: WireFirecrackerVmState::Running as i32, + }) + }, + ) + .serve(), + ), + Self::VmUnobservable => Some( + FakeNodeManager::default() + .unary("PauseSandbox", |_: PauseSandboxRequest| { + Err::(Status::unavailable("test pause failed")) + }) + .unary( + "GetFirecrackerVmState", + |_: GetFirecrackerVmStateRequest| { + Err::(Status::unavailable( + "test VM state unavailable", + )) + }, + ) + .serve(), + ), + } + } + + /// The status context of the exit's write. + fn context(self) -> &'static str { + match self { + Self::PreparationFails => "roll back failed pause preparation", + Self::NodeUnreachable => "roll back pause after node-manager connect failure", + Self::NodeConfirms => "complete pause", + Self::RetryPreparationFails | Self::VmStaysRunning | Self::VmUnobservable => { + "reconcile unconfirmed pause" + } + } + } +} + +/// What happens to the row at the interleaved guest preparation. +#[derive(Clone, Copy, Debug)] +enum Interleave { + /// Plan finding 1's Stop: cleanup admission's transaction for a + /// non-blocking Stop, then its cleanup completion. + Stop, + /// The row moves to the state with no other column changed. `pausing` + /// leaves it as pause wrote it. + Force(SandboxState), +} + +impl Interleave { + fn apply(self, conn: &rusqlite::Connection, id: &str) { + match self { + Self::Stop => { + let tx = crate::db::immediate_transaction(conn).unwrap(); + crate::db::update_state_superseding_activation_in_transaction( + &tx, + id, + SandboxState::Stopping, + ) + .unwrap(); + assert!( + crate::db::set_cleanup_intent(&tx, id, true, CleanupCompletion::MarkStopped) + .unwrap() + ); + tx.commit().unwrap(); + crate::db::complete_row_intent_cleanup_for_job( + conn, + id, + CleanupCompletion::MarkStopped, + None, + None, + None, + ) + .unwrap(); + } + Self::Force(SandboxState::Pausing) => {} + Self::Force(state) => force_state(conn, id, state), + } + } +} + +/// What Pause answers, and the row it leaves. +#[derive(Debug, Eq, PartialEq)] +enum Expected { + /// `Ok`, with the row `paused`. + Paused, + /// `Internal` with this message prefix, with the row rolled back to + /// `started`. + RolledBack(&'static str), + /// `Internal` reporting the quarantine, with the row `quarantined`. + Quarantined, + /// `Aborted` with this message, with every row as the interleaving left + /// it. + Rejected(String), +} + +/// The generic guard that refuses a `started` or `paused` write over `found`, +/// as `update_state`'s message names it, or `None` when only the from-set +/// refuses. Written out per state from #372's `PLAIN` grid for a row with no +/// cleanup intent or episode, not derived from the writer. +fn guard_refusal(found: SandboxState) -> Option<&'static str> { + match found { + SandboxState::Stopping + | SandboxState::Deleting + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed => Some("superseded_by_cleanup_episode"), + SandboxState::Quarantined => Some("blocked_by_quarantine"), + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopped + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::Failed => None, + } +} + +fn superseded(context: &str, found: SandboxState) -> Expected { + Expected::Rejected(format!( + "{context}: sandbox state write was superseded: sandbox is {found}" + )) +} + +/// Pause's answer when its write finds the row in `found`: only `pausing` is +/// pause's own row. +fn expected(exit: Exit, found: SandboxState) -> Expected { + if found == SandboxState::Pausing { + return match exit { + Exit::PreparationFails => { + Expected::RolledBack("guest snapshot preparation before pause failed: ") + } + Exit::NodeUnreachable => Expected::RolledBack("cannot connect to node-manager: "), + Exit::NodeConfirms => Expected::Paused, + Exit::RetryPreparationFails | Exit::VmStaysRunning => { + Expected::RolledBack("pause failed; sandbox reconciled to started") + } + Exit::VmUnobservable => Expected::Quarantined, + }; + } + match exit { + // The quarantine's compare-and-set reports only that it refused; + // pause re-reads the row and reports the state it found, including a + // quarantine another writer made. + Exit::VmUnobservable => match found { + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopping + | SandboxState::Stopped + | SandboxState::Deleting + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Quarantined + | SandboxState::Failed => superseded(exit.context(), found), + }, + Exit::PreparationFails + | Exit::NodeUnreachable + | Exit::NodeConfirms + | Exit::RetryPreparationFails + | Exit::VmStaysRunning => match guard_refusal(found) { + Some(outcome) => Expected::Rejected(format!( + "{}: sandbox state write was not applied: {outcome}", + exit.context() + )), + None => superseded(exit.context(), found), + }, + } +} + +/// Pause a `started` sandbox with a live runtime, running `interleave` at the +/// exit's guest preparation, and check the answer and the database. +async fn check_pause(exit: Exit, interleave: Interleave, found: SandboxState) { + let case = format!("{exit:?}, {interleave:?}"); + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + insert_test_sandbox_with_runtime(&conn, &id, SandboxState::Started); + let scratch = tempfile::tempdir().unwrap(); + let node_manager = exit.node_manager(); + let mut options = SandboxManagerServiceOptions::for_test_in(scratch.path()); + if let Some(node_manager) = &node_manager { + options.node_manager = node_manager.connector(); + } + let service = SandboxManagerService::new_with_options(conn, options); + + let calls = Arc::new(AtomicUsize::new(0)); + let interleaved = Arc::new(Mutex::new(None::)); + let prepare: GuestPreparation = { + let (calls, interleaved) = (Arc::clone(&calls), Arc::clone(&interleaved)); + let db = Arc::clone(&service.db); + Arc::new(move |sandbox_id| { + let call = calls.fetch_add(1, Ordering::SeqCst) + 1; + let last = call == exit.interleaved_preparation(); + if last { + let db = db.lock().unwrap(); + interleave.apply(&db, &sandbox_id); + *interleaved.lock().unwrap() = Some(snapshot(&db)); + } + let fails = last && exit.preparation_fails(); + Box::pin(async move { + if fails { + Err(GuestFlushError::Guest(Status::unavailable( + "test guest preparation failed", + ))) + } else { + Ok(()) + } + }) + }) + }; + + let result = tokio::time::timeout( + WAIT, + GUEST_PREPARATION.scope( + prepare, + service.pause(internal_request( + pplx_space_sbx_mgr_client::proto::PauseRequest { + sandbox_id: id.clone(), + }, + )), + ), + ) + .await + .unwrap_or_else(|_| panic!("{case}: Pause waited longer than {WAIT:?}")); + + assert_eq!( + calls.load(Ordering::SeqCst), + exit.interleaved_preparation(), + "{case}: guest preparation calls" + ); + let interleaved = interleaved + .lock() + .unwrap() + .take() + .unwrap_or_else(|| panic!("{case}: the interleaving never ran")); + assert_outcome( + &case, + expected(exit, found), + result, + &service.db_conn().unwrap(), + &id, + &interleaved, + ); + service.shutdown_token.cancel(); + service.task_tracker.close(); + tokio::time::timeout(WAIT, service.task_tracker.wait()) + .await + .unwrap_or_else(|_| panic!("{case}: service tasks outlived shutdown")); +} + +/// Check Pause's answer and the database against `expected`. +fn assert_outcome( + case: &str, + expected: Expected, + result: Result, Status>, + db: &rusqlite::Connection, + id: &str, + interleaved: &Snapshot, +) { + let state = crate::db::get_sandbox(db, id).unwrap().unwrap().state; + match expected { + Expected::Paused => { + assert_eq!(state, SandboxState::Paused, "{case}"); + assert!(result.is_ok(), "{case}: {result:?}"); + } + Expected::RolledBack(prefix) => { + assert_eq!(state, SandboxState::Started, "{case}"); + let status = result.expect_err(case); + assert_eq!(status.code(), tonic::Code::Internal, "{case}: {status:?}"); + assert!(status.message().starts_with(prefix), "{case}: {status:?}"); + } + Expected::Quarantined => { + assert_eq!(state, SandboxState::Quarantined, "{case}"); + let status = result.expect_err(case); + assert_eq!(status.code(), tonic::Code::Internal, "{case}: {status:?}"); + assert_eq!( + status.message(), + "pause failed; sandbox reconciled to quarantined", + "{case}" + ); + } + Expected::Rejected(message) => { + assert_unchanged(case, interleaved, db); + let status = result.expect_err(case); + assert_eq!(status.code(), tonic::Code::Aborted, "{case}: {status:?}"); + assert_eq!(status.message(), message, "{case}"); + } + } +} + +/// Plan finding 1: a Stop that settles `stopped` while pause awaits guest +/// preparation or node-manager keeps its row. Before pause wrote through its +/// edges, every exit overwrote it with `started` or `paused`, leaving a row +/// with no VM. +#[tokio::test] +async fn pause_leaves_a_row_a_concurrent_stop_settled() { + for exit in EXITS { + check_pause(exit, Interleave::Stop, SandboxState::Stopped).await; + } +} + +/// Every state the row can be in at each of pause's writes. Only `pausing` +/// is written; it is the accepting control for each exit. +#[tokio::test] +async fn pause_writes_only_over_its_own_pausing_row() { + for exit in EXITS { + for state in SandboxState::all() { + check_pause(exit, Interleave::Force(state), state).await; + } + } +} diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend.rs index 6cd02678fa..5179befa1d 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend.rs @@ -2,6 +2,7 @@ //! and suspend preparation once the locks are held. use super::*; +use crate::service::cleanup_support::StateWriteRejected; fn completed_suspend_retry_response( sandbox: &crate::db::SandboxRow, @@ -396,7 +397,7 @@ impl SandboxManagerService { .id .parse::() .map_err(|error| Status::internal(format!("invalid sandbox ID: {error}")))?; - let pause_state = pause_with_reconciliation( + let pause_result = pause_with_reconciliation( &mut node_client, sandbox.node_execution_fence(), &self.db, @@ -407,8 +408,20 @@ impl SandboxManagerService { self.config.balloon_hint_timeout_ms, ) .await; - if pause_state != SandboxState::Paused { + // Fail closed: a rejection means another operation owns the row, + // so only pause's own confirmed `paused` lets suspend continue. + if pause_result != Ok(SandboxState::Paused) { self.watchdog.record_lifecycle_edge(&sandbox.id).await; + // Report the state the row was left in. A missing row is never + // a rejection here: its rollback reads as `started`. + let pause_state = match pause_result { + Ok(state) => state, + Err( + StateWriteRejected::FromState { found } + | StateWriteRejected::Guard { found, .. }, + ) => found, + Err(StateWriteRejected::Missing) => SandboxState::Started, + }; let reason = format!("auto-pause failed; sandbox reconciled to {pause_state}"); pplx_space_metrics::snapshot::create_failed( create_kind, diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/tests.rs index f188dfce80..ee181a664a 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/tests.rs @@ -2011,7 +2011,7 @@ async fn confirmed_node_pause_reports_paused_and_carries_the_execution_fence() { stop.cancel(); server.await.unwrap().unwrap(); - assert_eq!(state, SandboxState::Paused); + assert_eq!(state, Ok(SandboxState::Paused)); // Every node call carries the caller's execution fence so node-manager // can refuse a pause aimed at a replaced execution. let requests = node_manager.requests.lock().unwrap();