diff --git a/architecture-baseline.toml b/architecture-baseline.toml index 11c9808111..b15a7fb9dd 100644 --- a/architecture-baseline.toml +++ b/architecture-baseline.toml @@ -80,7 +80,7 @@ "crates/libs/sbx/mgr/server/src/db/cleanup_episodes.rs" = 1833 "crates/libs/sbx/mgr/server/src/db/quarantine_recovery.rs" = 1042 "crates/libs/sbx/mgr/server/src/db/sandboxes.rs" = 1553 -"crates/libs/sbx/mgr/server/src/db/state_transitions.rs" = 1140 +"crates/libs/sbx/mgr/server/src/db/state_transitions.rs" = 1018 "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" = 1574 diff --git a/crates/libs/sbx/mgr/server/AGENTS.md b/crates/libs/sbx/mgr/server/AGENTS.md index d070989387..1c956e6f1b 100644 --- a/crates/libs/sbx/mgr/server/AGENTS.md +++ b/crates/libs/sbx/mgr/server/AGENTS.md @@ -2,12 +2,11 @@ ## Changes -- Write `sandboxes.state` only through `src/db` writers. A new transition - calls `update_state_from` with its edge, or adds a specialized - compare-and-set writer in `src/db`. Do not add `update_state` callers: it - overwrites any state its cleanup and quarantine guards admit. Record the - edge in `src/db/state_write.rs`, and add any new disagreement with - `can_transition_to`, with a reason, to `ALLOWLIST` in +- Write the node-local SQLite `sandboxes.state` only through `src/db`: every + persisted write goes through `update_state_from` with an explicit edge, or + through a named specialized writer. A new edge in `src/db/state_write.rs` + needs a kind and a doc comment naming its writer; add any new disagreement + with `can_transition_to`, with a reason, to `ALLOWLIST` in `src/db/tests/state_write_table.rs`. Nothing checks the table against the writers; the `&'static Edge` argument does that at call sites. A refused write must leave every row unchanged. Map an `update_state_from` outcome to diff --git a/crates/libs/sbx/mgr/server/src/db/startup_recovery.rs b/crates/libs/sbx/mgr/server/src/db/startup_recovery.rs index b07ea0dc1b..e4f4ef8f0c 100644 --- a/crates/libs/sbx/mgr/server/src/db/startup_recovery.rs +++ b/crates/libs/sbx/mgr/server/src/db/startup_recovery.rs @@ -371,11 +371,22 @@ pub enum RecoverStartupStatesError { CleanupRequest(#[from] RequestCleanupEpisodeError), #[error("armed cleanup episode has unsupported activation state {0}")] UnsupportedActivationState(SandboxState), - #[error("startup recovery write {edge} was not applied: {outcome:?}")] - StateWriteNotApplied { + #[error("startup recovery write {edge} was refused: sandbox is {found}, {}", refusal_text(.refusal))] + StateWriteRefused { edge: &'static str, - outcome: StateWriteOutcome, + found: SandboxState, + refusal: StateWriteRefusal, }, + #[error("startup recovery write {edge} found no sandbox row")] + StateWriteMissing { edge: &'static str }, +} + +fn refusal_text(refusal: &StateWriteRefusal) -> &'static str { + match refusal { + StateWriteRefusal::FromState => "outside the edge's source states", + StateWriteRefusal::Guard(StateWriteGuard::Quarantined) => "blocked by quarantine", + StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned) => "owned by cleanup", + } } /// [`update_state_from`] for a startup recovery writer that requires its @@ -393,9 +404,15 @@ pub fn require_startup_state_write( ) -> std::result::Result<(), RecoverStartupStatesError> { match update_state_from(conn, id, edge)? { StateWriteOutcome::Applied { .. } => Ok(()), - outcome => Err(RecoverStartupStatesError::StateWriteNotApplied { - edge: edge.name(), - outcome, - }), + StateWriteOutcome::Refused { found, refusal } => { + Err(RecoverStartupStatesError::StateWriteRefused { + edge: edge.name(), + found, + refusal, + }) + } + StateWriteOutcome::Missing => { + Err(RecoverStartupStatesError::StateWriteMissing { edge: edge.name() }) + } } } diff --git a/crates/libs/sbx/mgr/server/src/db/state_transitions.rs b/crates/libs/sbx/mgr/server/src/db/state_transitions.rs index 0c5fec5a9f..d373b145b8 100644 --- a/crates/libs/sbx/mgr/server/src/db/state_transitions.rs +++ b/crates/libs/sbx/mgr/server/src/db/state_transitions.rs @@ -48,136 +48,6 @@ pub enum SupersedeActivationError { StateWriteNotApplied(#[from] SandboxStateWriteNotApplied), } -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -#[must_use = "callers must distinguish an applied state write from a superseded one"] -pub enum SandboxStateWriteOutcome { - Applied, - Missing, - BlockedByQuarantine, - SupersededByCleanupEpisode, -} - -impl SandboxStateWriteOutcome { - pub const fn as_str(self) -> &'static str { - match self { - Self::Applied => "applied", - Self::Missing => "missing", - Self::BlockedByQuarantine => "blocked_by_quarantine", - Self::SupersededByCleanupEpisode => "superseded_by_cleanup_episode", - } - } - - pub const fn is_applied(self) -> bool { - matches!(self, Self::Applied) - } - - pub fn require_applied(self) -> std::result::Result<(), SandboxStateWriteNotApplied> { - if self.is_applied() { - Ok(()) - } else { - Err(SandboxStateWriteNotApplied { outcome: self }) - } - } -} - -impl fmt::Display for SandboxStateWriteOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.write_str(self.as_str()) - } -} - -#[derive(Debug, thiserror::Error)] -#[error("sandbox state write was not applied: {outcome}")] -pub struct SandboxStateWriteNotApplied { - pub outcome: SandboxStateWriteOutcome, -} - -pub fn update_state( - conn: &Connection, - id: &str, - state: SandboxState, -) -> Result { - if matches!(state, SandboxState::Quarantined | SandboxState::Failed) { - return Err(rusqlite::Error::InvalidQuery); - } - if state == SandboxState::Deleting && conn.is_autocommit() { - let tx = immediate_transaction(conn)?; - let outcome = update_state(&tx, id, state)?; - tx.commit()?; - return Ok(outcome); - } - let previous_state = current_state_label(conn, id)?; - let previous_state_value = previous_state - .as_deref() - .and_then(|state| state.parse::().ok()); - match generic_state_write_refusal(conn, id, previous_state_value, state)? { - Some(StateWriteRefusal::Quarantined) => { - tracing::warn!( - sandbox_id = %id, - target_state = %state, - "ignored generic sandbox state update because sandbox is quarantined" - ); - return Ok(SandboxStateWriteOutcome::BlockedByQuarantine); - } - // The generic guards never answer `FromState`. - Some(StateWriteRefusal::CleanupOwned | StateWriteRefusal::FromState) => { - tracing::warn!( - sandbox_id = %id, - target_state = %state, - "ignored generic sandbox state update because cleanup owns the transition" - ); - return Ok(SandboxStateWriteOutcome::SupersededByCleanupEpisode); - } - None => {} - } - - let changed = conn.execute( - "UPDATE sandboxes - SET state = ?1, - state_changed_at = CASE WHEN state IS NOT ?1 THEN ?2 ELSE state_changed_at END, - failure_reason = NULL, - restore_origin = CASE WHEN ?1 = 'restoring' THEN restore_origin ELSE NULL END, - restore_phase = CASE WHEN ?1 = 'restoring' THEN restore_phase ELSE NULL END, - restore_storage_uri = CASE WHEN ?1 = 'restoring' THEN restore_storage_uri ELSE NULL END, - transition_receipt_id = CASE - WHEN ?1 = transition_receipt_target_state THEN transition_receipt_id - ELSE NULL - END, - transition_receipt_kind = CASE - WHEN ?1 = transition_receipt_target_state THEN transition_receipt_kind - ELSE NULL - END, - transition_receipt_target_state = CASE - WHEN ?1 = transition_receipt_target_state THEN transition_receipt_target_state - ELSE NULL - END, - transition_receipt_execution_generation = CASE - WHEN ?1 = transition_receipt_target_state THEN transition_receipt_execution_generation - ELSE NULL - END, - transition_receipt_activation_attempt_id = CASE - WHEN ?1 = transition_receipt_target_state THEN transition_receipt_activation_attempt_id - ELSE NULL - END, - transition_receipt_reason_code = CASE - WHEN ?1 = transition_receipt_target_state THEN transition_receipt_reason_code - ELSE NULL - END, - updated_at = ?2 - WHERE id = ?3 - AND (?1 = 'deleting' OR state != 'quarantined')", - params![state.as_str(), now()?, id], - )?; - if changed != 0 && state == SandboxState::Deleting { - teardown_jobs::fail_active(conn, id, TeardownSuspendFailure::Deleting)?; - } - Ok(if changed == 0 { - SandboxStateWriteOutcome::Missing - } else { - SandboxStateWriteOutcome::Applied - }) -} - pub(crate) fn restore_checkpoint_state( conn: &Connection, id: &str, @@ -199,9 +69,10 @@ pub(crate) fn restore_checkpoint_state( Ok(changed == 1) } -/// Fence policy installation and Paused -> Resuming in one transaction. -/// An absent generation matches only an identity-free legacy resident; it is -/// never a wildcard for an execution that has acquired a generation. +/// Fence policy installation and `paused` -> `resuming` in one transaction: +/// [`edges::RESUME_BEGIN`] refuses any other state. An absent generation +/// matches only an identity-free legacy resident; it is never a wildcard for an +/// execution that has acquired a generation. pub(crate) fn begin_resident_resume_with_policy( conn: &Connection, id: &str, @@ -212,13 +83,12 @@ pub(crate) fn begin_resident_resume_with_policy( let Some(sandbox) = get_sandbox(&tx, id)? else { return Ok(false); }; - if sandbox.state != SandboxState::Paused - || sandbox.space_api_route_identity.execution_generation != generation - { + if sandbox.space_api_route_identity.execution_generation != generation { return Ok(false); } - if update_state(&tx, id, SandboxState::Resuming)? != SandboxStateWriteOutcome::Applied { - return Ok(false); + match update_state_from(&tx, id, &edges::RESUME_BEGIN)? { + StateWriteOutcome::Applied { .. } => {} + StateWriteOutcome::Refused { .. } | StateWriteOutcome::Missing => return Ok(false), } install_activation_auto_lifecycle(&tx, id, policy)?; tx.commit()?; @@ -311,7 +181,7 @@ pub(crate) fn update_state_superseding_activation_in_transaction( params![id, episode.completion.as_str(), episode.keep_filesystem], )?; } - update_state(conn, id, state)?.require_applied()?; + write_cleanup_admission_state(conn, id, state)??; Ok(()) } 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 ea60915897..efcb224a71 100644 --- a/crates/libs/sbx/mgr/server/src/db/state_write.rs +++ b/crates/libs/sbx/mgr/server/src/db/state_write.rs @@ -1,6 +1,7 @@ //! The transitions sbx-manager persists in the node-local SQLite //! `sandboxes.state` column, and `update_state_from`, the generic writer that -//! compares against the caller's expected states. +//! compares against the caller's expected states. Every other writer of the +//! column is a named specialized writer in `src/db`. //! //! `SandboxState::can_transition_to` is the RPC-admission machine. [`EDGES`] //! is a separate inventory of the transitions the database writers implement: @@ -12,8 +13,8 @@ //! The table follows these conventions: //! - 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. +//! - An `update_state_from` edge's from-set is exactly what its callers +//! legitimately see. //! - 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. @@ -235,8 +236,9 @@ edge_table! { /// `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. + /// `begin_resident_resume_with_policy`'s `update_state_from` inside its + /// policy transaction, and `begin_resident_resume`'s without an activation + /// policy. RESUME_BEGIN: Request, of(&[Paused]) => Resuming; /// `restore_resident_resume_state(Started)` through /// `rollback_activation_state`. @@ -281,14 +283,14 @@ edge_table! { // Stop and delete. /// Stop admission: `admit_lifecycle_cleanup_fenced` through - /// `update_state_superseding_activation_in_transaction(Stopping)`. Stop - /// does not consult `transition`, but resolves its target only while the - /// runtime is live (`fc.pid`), and admission does not re-check it. A - /// settled `created`, `stopped` or `suspended` source is therefore - /// reached only in the window between that gate and the cleanup fence, - /// or with a leaked runtime; a `creating` row has none. Kill admits only - /// `started` and `paused` through the same writer, and the generic guards - /// refuse `deleting` and `quarantined`. + /// `update_state_superseding_activation_in_transaction(Stopping)` and + /// `write_cleanup_admission_state`. Stop does not consult `transition`, + /// but resolves its target only while the runtime is live (`fc.pid`), and + /// admission does not re-check it. A settled `created`, `stopped` or + /// `suspended` source is therefore reached only in the window between + /// that gate and the cleanup fence, or with a leaked runtime; a `creating` + /// row has none. Kill admits only `started` and `paused` through the same + /// writer, and the generic guards refuse `deleting` and `quarantined`. STOP_ADMISSION: Request, StateSet::ALL .without(Creating) @@ -302,8 +304,9 @@ edge_table! { /// `stop_suspended_for_auto_lifecycle`, which bypasses every RPC. AUTO_LIFECYCLE_STOP_SUSPENDED: Request, of(&[Suspended]) => Stopped; /// Delete admission: `admit_lifecycle_cleanup_fenced` through - /// `update_state_superseding_activation_in_transaction(Deleting)`, from - /// the Delete RPC and auto-lifecycle Delete. + /// `update_state_superseding_activation_in_transaction(Deleting)` and + /// `write_cleanup_admission_state`, from the Delete RPC and auto-lifecycle + /// Delete. DELETE_ADMISSION: Request, StateSet::ALL.without(Deleting) => Deleting; /// The eviction claims: `claim_stopped_sandbox_for_eviction`, /// `claim_suspended_sandbox_for_eviction`, @@ -424,32 +427,54 @@ edge_table! { pub use edges::EDGES; +/// A generic guard that refused a write. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum StateWriteGuard { + /// Only deletion leaves `quarantined`. + Quarantined, + /// Cleanup owns the row. + CleanupOwned, +} + /// Why [`update_state_from`] did not write. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum StateWriteRefusal { /// The row is not in any of the edge's source states. FromState, - /// `update_state`'s quarantine guard: only deletion leaves `quarantined`. - Quarantined, - /// `update_state`'s cleanup-ownership guard. - CleanupOwned, + Guard(StateWriteGuard), } -/// The guards `update_state` and `update_state_from` share. Quarantine answers -/// first: only deletion leaves `quarantined`. Cleanup then owns the row while -/// an episode that keeps the row exists or has been requested, while a -/// cleanup state carries a current `delete_row` intent (except for deletion -/// itself), and against any target that would revive a `stopping` or cleanup -/// row. An armed `delete_row` create episode deliberately does not block: -/// create completion writes through it. Never answers `FromState`. +/// A guard refusal or a missing row, in the text clients receive. +#[derive(Clone, Copy, Debug, Eq, PartialEq, thiserror::Error)] +pub enum SandboxStateWriteNotApplied { + #[error("sandbox state write was not applied: {}", guard_label(.0))] + Guard(StateWriteGuard), + #[error("sandbox state write was not applied: missing")] + Missing, +} + +fn guard_label(guard: &StateWriteGuard) -> &'static str { + match guard { + StateWriteGuard::Quarantined => "blocked_by_quarantine", + StateWriteGuard::CleanupOwned => "superseded_by_cleanup_episode", + } +} + +/// The guards every generic write of the node-local SQLite `sandboxes.state` +/// shares. Quarantine answers first: only deletion leaves `quarantined`. +/// Cleanup then owns the row while an episode that keeps the row exists or has +/// been requested, while a cleanup state carries a current `delete_row` intent +/// (except for deletion itself), and against any target that would revive a +/// `stopping` or cleanup row. An armed `delete_row` create episode +/// deliberately does not block: create completion writes through it. pub(super) fn generic_state_write_refusal( conn: &Connection, id: &str, previous: Option, target: SandboxState, -) -> Result> { +) -> Result> { if generic_writer_transition_blocked(previous, target) { - return Ok(Some(StateWriteRefusal::Quarantined)); + return Ok(Some(StateWriteGuard::Quarantined)); } let cleanup_owned = cleanup_episode(conn, id)? .is_some_and(|episode| !episode.completion.deletes_row() || episode.requested_at.is_some()) @@ -477,7 +502,7 @@ pub(super) fn generic_state_write_refusal( | SandboxState::Resuming | SandboxState::Checkpointing )); - Ok(cleanup_owned.then_some(StateWriteRefusal::CleanupOwned)) + Ok(cleanup_owned.then_some(StateWriteGuard::CleanupOwned)) } #[derive(Clone, Copy, Debug, Eq, PartialEq)] @@ -495,23 +520,19 @@ pub enum StateWriteOutcome { /// Write `edge.to()` only if the row is in one of `edge.from()`. /// -/// This is `update_state` with an expected-state predicate, part of the -/// `UPDATE` itself. The generic guards answer first, so every write -/// `update_state` refuses is refused for the same reason, and converting an -/// `update_state` caller changes only rows outside the edge's from-set. A -/// guard refusal does not say whether the row was also outside the from-set: -/// a caller that must tell the two apart checks `from.contains(found)`, not -/// the refusal. +/// The expected-state predicate is part of the `UPDATE` itself. The generic +/// guards answer first, so a guard refusal does not say whether the row was +/// also outside the from-set: a caller that must tell the two apart checks +/// `from.contains(found)`, not the refusal. /// -/// Side effects match `update_state`: the failure reason, restore columns and -/// non-matching receipts are cleared, and a `deleting` write fails the active -/// teardown jobs. Inside a caller's transaction this runs in it. In -/// autocommit it opens its own `IMMEDIATE` transaction, so the guard reads, -/// the write and the job failure commit together; it never nests one. +/// The failure reason, restore columns and receipts for another target are +/// cleared, and a `deleting` write fails the active teardown jobs. Inside a +/// caller's transaction this runs in it. In 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 targets `update_state` rejects (`quarantined`, `failed`), are -/// `InvalidQuery`. A write that matches -/// no row although the row was read in a source state is +/// Inserts and `quarantined` or `failed` targets 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( conn: &Connection, @@ -536,6 +557,52 @@ pub(crate) fn update_state_from_states( write_state_from(conn, id, from, to) } +/// Stop and Delete admission's write, for +/// `update_state_superseding_activation_in_transaction` alone. It writes over +/// every state the generic guards admit, so it is not an `update_state_from` +/// edge: admission also re-admits a row already in its target, a `stopping` +/// row for a repeated Stop of a live runtime and a `deleting` row for a +/// repeated Delete, and no edge lists its own target. The guards refuse a Stop +/// of a `deleting` or `quarantined` row. [`edges::STOP_ADMISSION`] and +/// [`edges::DELETE_ADMISSION`] list the transitions it makes. +pub(super) fn write_cleanup_admission_state( + conn: &Connection, + id: &str, + target: SandboxState, +) -> Result> { + if !matches!(target, SandboxState::Stopping | SandboxState::Deleting) { + return Err(rusqlite::Error::InvalidQuery); + } + match write_state_from(conn, id, StateSet::ALL, target)? { + StateWriteOutcome::Applied { .. } => Ok(Ok(())), + StateWriteOutcome::Refused { + refusal: StateWriteRefusal::Guard(guard), + .. + } => { + match guard { + StateWriteGuard::Quarantined => tracing::warn!( + sandbox_id = %id, + target_state = %target, + "ignored generic sandbox state update because sandbox is quarantined" + ), + StateWriteGuard::CleanupOwned => tracing::warn!( + sandbox_id = %id, + target_state = %target, + "ignored generic sandbox state update because cleanup owns the transition" + ), + } + Ok(Err(SandboxStateWriteNotApplied::Guard(guard))) + } + // `ALL` holds the row's state, so a write that missed it is the + // invariant error `write_state_from` returns, never this refusal. + StateWriteOutcome::Refused { + refusal: StateWriteRefusal::FromState, + .. + } => Err(rusqlite::Error::StatementChangedRows(0)), + StateWriteOutcome::Missing => Ok(Err(SandboxStateWriteNotApplied::Missing)), + } +} + fn write_state_from( conn: &Connection, id: &str, @@ -566,8 +633,11 @@ fn write_state_from_in_transaction( let found = found.parse::().map_err(|error| { rusqlite::Error::FromSqlConversionFailure(0, Type::Text, Box::new(error)) })?; - if let Some(refusal) = generic_state_write_refusal(conn, id, Some(found), to)? { - return Ok(StateWriteOutcome::Refused { found, refusal }); + if let Some(guard) = generic_state_write_refusal(conn, id, Some(found), to)? { + return Ok(StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::Guard(guard), + }); } let from_labels = serde_json::to_string(&from.iter().map(|state| state.as_str()).collect::>()) diff --git a/crates/libs/sbx/mgr/server/src/db/tests.rs b/crates/libs/sbx/mgr/server/src/db/tests.rs index 831eeae62d..e3041c1deb 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests.rs @@ -62,10 +62,7 @@ fn node_execution_cleanup_acknowledgements_are_retryable_and_fenced() { .unwrap(); let staged = node_execution_fence(&conn, &id).unwrap().unwrap(); assert!(staged.previous().is_none()); - assert_eq!( - update_state(&conn, &id, SandboxState::Created).unwrap(), - SandboxStateWriteOutcome::Applied - ); + state_write_fixtures::apply_edge(&conn, &id, &edges::CREATE_COMPLETE); assert!(complete_create(&conn, &id, staging_cleanup, false).unwrap()); let activate = || { let episode = begin_activation( @@ -275,14 +272,17 @@ fn cleanup_intent_cannot_be_replaced_by_suspend_or_resident_completion() { ); assert!(set_cleanup_intent(&conn, &id, false, CleanupCompletion::DeleteRow).unwrap()); for stale_completion in [ - SandboxState::Pausing, - SandboxState::Paused, - SandboxState::Started, - SandboxState::Resuming, + &edges::PAUSE_BEGIN, + &edges::PAUSE_COMPLETE, + &edges::PAUSE_ROLLBACK, + &edges::RESUME_BEGIN, ] { assert_eq!( - update_state(&conn, &id, stale_completion).unwrap(), - SandboxStateWriteOutcome::SupersededByCleanupEpisode + update_state_from(&conn, &id, stale_completion).unwrap(), + StateWriteOutcome::Refused { + found: cleanup_state, + refusal: StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned), + } ); assert!(!begin_suspend(&conn, &id).unwrap()); let row = get_sandbox(&conn, &id).unwrap().unwrap(); @@ -320,7 +320,8 @@ fn application_state_write_updates_only_state_changed_at() { .unwrap(); // Act - let outcome = update_state(&conn, &sandbox_id, SandboxState::Paused).unwrap(); + let outcome = + update_state_from_states(&conn, &sandbox_id, StateSet::ALL, SandboxState::Paused).unwrap(); let after_transition: (String, String) = conn .query_row( "SELECT state_changed_at, updated_at FROM sandboxes WHERE id = ?1", @@ -328,7 +329,8 @@ fn application_state_write_updates_only_state_changed_at() { |row| Ok((row.get(0)?, row.get(1)?)), ) .unwrap(); - let retry_outcome = update_state(&conn, &sandbox_id, SandboxState::Paused).unwrap(); + let retry_outcome = + update_state_from_states(&conn, &sandbox_id, StateSet::ALL, SandboxState::Paused).unwrap(); let after_retry: String = conn .query_row( "SELECT state_changed_at FROM sandboxes WHERE id = ?1", @@ -351,9 +353,19 @@ fn application_state_write_updates_only_state_changed_at() { ) .unwrap(); assert_ne!(after_transition.0, "123"); - assert_eq!(outcome, SandboxStateWriteOutcome::Applied); + assert_eq!( + outcome, + StateWriteOutcome::Applied { + from: SandboxState::Started + } + ); assert_eq!(after_transition.0, after_transition.1); - assert_eq!(retry_outcome, SandboxStateWriteOutcome::Applied); + assert_eq!( + retry_outcome, + StateWriteOutcome::Applied { + from: SandboxState::Paused + } + ); assert_eq!(after_retry, after_transition.0); assert_eq!(after_unrelated_write, after_transition.0); } @@ -391,10 +403,12 @@ fn every_allowed_state_transition_tracks_the_change_time() { ) .unwrap_or_else(|error| panic!("{from} -> {to}: {error}")); } else { - update_state(&conn, &sandbox_id, to) - .unwrap() - .require_applied() - .unwrap(); + assert_eq!( + update_state_from_states(&conn, &sandbox_id, StateSet::of(&[from]), to) + .unwrap(), + StateWriteOutcome::Applied { from }, + "{from} -> {to}" + ); } // Assert @@ -2329,10 +2343,7 @@ fn suspended_cleanup_completion_dedupes_prior_suspend_row() { assert!(get_local_artifact(&conn, a1).unwrap().is_some()); // Simulate a wake-and-resuspend cycle, including its stable paused state. - assert_eq!( - update_state(&conn, "sandbox-1", SandboxState::Paused).unwrap(), - SandboxStateWriteOutcome::Applied - ); + state_write_fixtures::force_state(&conn, "sandbox-1", SandboxState::Paused); assert!(begin_suspend(&conn, "sandbox-1").unwrap()); let a2 = LocalArtifactId::new(); complete_row_intent_cleanup( @@ -2563,10 +2574,7 @@ fn stopped_publication_commits_disk_identity_with_terminal_state() { &conn, TestSandboxInsert::new("sandbox-1", "sandbox1", SandboxState::Started), ); - update_state(&conn, "sandbox-1", SandboxState::Stopping) - .unwrap() - .require_applied() - .unwrap(); + state_write_fixtures::apply_edge(&conn, "sandbox-1", &edges::STOP_ADMISSION); let local_artifact_id = LocalArtifactId::new(); publish_stopped_artifact( @@ -6615,10 +6623,7 @@ fn start_arms_access_idle_only_after_guest_readiness() { restored.auto_lifecycle_state.conditions[&AutoLifecycleConditionSetId::AccessIdle].epoch_at; assert_eq!(restored_epoch, None); - assert_eq!( - update_state(&conn, "restore-ready-sandbox", SandboxState::Resuming).unwrap(), - SandboxStateWriteOutcome::Applied - ); + state_write_fixtures::apply_edge(&conn, "restore-ready-sandbox", &edges::RESUME_BEGIN); assert!( restore_resident_resume_state(&conn, "restore-ready-sandbox", SandboxState::Started,) .unwrap() @@ -6630,10 +6635,7 @@ fn start_arms_access_idle_only_after_guest_readiness() { resumed.auto_lifecycle_state.conditions[&AutoLifecycleConditionSetId::AccessIdle].epoch_at; assert!(resumed_epoch.is_some()); - assert_eq!( - update_state(&conn, "restore-ready-sandbox", SandboxState::Resuming).unwrap(), - SandboxStateWriteOutcome::Applied - ); + state_write_fixtures::force_state(&conn, "restore-ready-sandbox", SandboxState::Resuming); assert!( restore_resident_resume_state(&conn, "restore-ready-sandbox", SandboxState::Started,) .unwrap() @@ -6839,10 +6841,6 @@ fn generic_lifecycle_writers_reject_failed_targets() { &conn, TestSandboxInsert::new("starting", "starting", SandboxState::Starting), ); - assert!(matches!( - update_state(&conn, "starting", SandboxState::Failed), - Err(rusqlite::Error::InvalidQuery) - )); assert!(matches!( rollback_starting(&conn, "starting", SandboxState::Failed), Err(rusqlite::Error::InvalidQuery) @@ -6895,8 +6893,11 @@ fn cleanup_recovery_owns_failure_clearance_not_stale_resident_completion() { ) .unwrap(); assert_eq!( - update_state(&conn, "sandbox-1", SandboxState::Started).unwrap(), - SandboxStateWriteOutcome::SupersededByCleanupEpisode + update_state_from(&conn, "sandbox-1", &edges::PAUSE_ROLLBACK).unwrap(), + StateWriteOutcome::Refused { + found: SandboxState::CleanupFailed, + refusal: StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned), + } ); complete_row_intent_cleanup(&conn, "sandbox-1", CleanupCompletion::MarkStopped, None).unwrap(); @@ -6924,8 +6925,11 @@ fn quarantined_state_is_not_exited_by_generic_failure_writers() { .unwrap(); assert_eq!( - update_state(&conn, "sandbox-1", SandboxState::Started).unwrap(), - SandboxStateWriteOutcome::BlockedByQuarantine, + update_state_from(&conn, "sandbox-1", &edges::PAUSE_ROLLBACK).unwrap(), + StateWriteOutcome::Refused { + found: SandboxState::Quarantined, + refusal: StateWriteRefusal::Guard(StateWriteGuard::Quarantined), + } ); let row = get_sandbox(&conn, "sandbox-1").unwrap().unwrap(); @@ -6935,10 +6939,7 @@ fn quarantined_state_is_not_exited_by_generic_failure_writers() { Some("runtime evidence is inconsistent") ); - update_state(&conn, "sandbox-1", SandboxState::Deleting) - .unwrap() - .require_applied() - .unwrap(); + state_write_fixtures::apply_edge(&conn, "sandbox-1", &edges::DELETE_ADMISSION); let row = get_sandbox(&conn, "sandbox-1").unwrap().unwrap(); assert_eq!(row.state, SandboxState::Deleting); } diff --git a/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs index 2e11c7417b..6f3dddfd63 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs @@ -16,8 +16,8 @@ use pplx_space_common_core::types::{ use rusqlite::{Connection, params}; use super::state_write_fixtures::{ - Cell, ExpectedDiff, NULL, assert_accepts, backdate, decorate, generation, insert_creating, - insert_creating_with, insert_row, insert_suspended_with_artifact, int, + Cell, ExpectedDiff, NULL, apply_edge, assert_accepts, backdate, decorate, generation, + insert_creating, insert_creating_with, insert_row, insert_suspended_with_artifact, int, quarantine_history_entered, quarantine_state_entered, receipt_cleared, runtime_cleared, seed_receipt, seed_runtime, snapshot_path, suspend_artifact_inserted, text, }; @@ -27,10 +27,10 @@ use crate::db::{ CleanupEpisodeRequestOutcome, CompleteRowIntentCleanupDisposition, begin_activation_with_route_evidence, begin_suspend, charge_prepared_clone_disposal_attempt, complete_cleanup_episode_for_job, complete_create, complete_pool_clone, - complete_row_intent_cleanup_for_job, mark_cleanup_required, mark_restore_succeeded, + complete_row_intent_cleanup_for_job, edges, mark_cleanup_required, mark_restore_succeeded, mark_start_succeeded, quarantine_cleanup_episode_for_job, rearm_prepared_clone_disposal, record_cleanup_episode_failure, request_cleanup_episode, rollback_restoring_episode, - rollback_starting_episode, set_cleanup_intent, test_conn, update_state, + rollback_starting_episode, set_cleanup_intent, test_conn, }; use crate::state::SandboxState; @@ -299,10 +299,10 @@ fn activation_completion_publishes_the_target_and_retires_its_episode() { CleanupCompletion::DeleteRow, Some((generation(7), attempt())), ); - assert_eq!( - update_state(&conn, &id, settled).unwrap(), - crate::db::SandboxStateWriteOutcome::Applied - ); + apply_edge(&conn, &id, &edges::CREATE_COMPLETE); + if settled == SandboxState::Suspended { + apply_edge(&conn, &id, &edges::CREATE_BASE_SNAPSHOT_COMPLETE); + } seed_receipt(&conn, &id, SandboxState::Stopped); backdate(&conn, &id); let mut cells = vec![ diff --git a/crates/libs/sbx/mgr/server/src/db/tests/state_write_fixtures.rs b/crates/libs/sbx/mgr/server/src/db/tests/state_write_fixtures.rs index c422437d98..9362c65a1f 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/state_write_fixtures.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/state_write_fixtures.rs @@ -18,9 +18,11 @@ use rusqlite::{Connection, params}; use super::{TestSandboxInsert, insert_test_sandbox}; use crate::db::{ - CleanupCompletion, CleanupEpisodeSpec, InsertSandboxParams, ORPHANED_SANDBOX_OWNER, - ORPHANED_SANDBOX_TEMPLATE, PolicyAuthority, begin_suspend, complete_row_intent_cleanup_for_job, - ensure_teardown_suspend_operation, insert_sandbox_with_cleanup_episode_and_route_evidence, + CleanupCompletion, CleanupEpisodeSpec, Edge, InsertSandboxParams, ORPHANED_SANDBOX_OWNER, + ORPHANED_SANDBOX_TEMPLATE, PolicyAuthority, StateWriteOutcome, begin_suspend, + complete_row_intent_cleanup_for_job, ensure_teardown_suspend_operation, immediate_transaction, + insert_sandbox_with_cleanup_episode_and_route_evidence, set_cleanup_intent, update_state_from, + update_state_superseding_activation_in_transaction, }; use crate::state::SandboxState; @@ -387,6 +389,38 @@ pub(crate) fn force_state(conn: &Connection, id: &str, state: SandboxState) { .unwrap(); } +/// Write `edge` with `update_state_from`, which must apply. +pub(crate) fn apply_edge(conn: &Connection, id: &str, edge: &'static Edge) { + let outcome = update_state_from(conn, id, edge).unwrap(); + assert!( + matches!(outcome, StateWriteOutcome::Applied { .. }), + "{}: {outcome:?}", + edge.name() + ); +} + +/// A non-blocking Stop's admission transaction, as the cleanup fence runs it: +/// supersede any activation, write `stopping` and persist the `mark_stopped` +/// intent. +pub(crate) fn admit_stop(conn: &Connection, id: &str) { + let tx = immediate_transaction(conn).unwrap(); + update_state_superseding_activation_in_transaction(&tx, id, SandboxState::Stopping).unwrap(); + assert!(set_cleanup_intent(&tx, id, true, CleanupCompletion::MarkStopped).unwrap()); + tx.commit().unwrap(); +} + +/// Stop's row-intent completion of [`admit_stop`]: `stopping` to `stopped`. +pub(crate) fn complete_stop(conn: &Connection, id: &str) { + complete_row_intent_cleanup_for_job(conn, id, CleanupCompletion::MarkStopped, None, None, None) + .unwrap(); +} + +/// A non-blocking Stop from admission to completion. +pub(crate) fn stop(conn: &Connection, id: &str) { + admit_stop(conn, id); + complete_stop(conn, id); +} + pub(super) const RESTORE_STORAGE_URI: &str = "s3://snapshots/suspend/restore?region=us-east-1"; /// The remote snapshot a seeded cross-node restore stages from. diff --git a/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs b/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs index 340c2a29de..affa145063 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs @@ -13,19 +13,20 @@ use super::state_write_fixtures::{ Cell, ExpectedDiff, NULL, RESTORE_STORAGE_URI, Snapshot, as_synthetic_orphan, assert_accepts, assert_unchanged, backdate, decorate, endpoint_proof, execution_metadata, generation, insert_creating, insert_row, int, quarantine_history_entered, quarantine_state_entered, - receipt_cleared, restore_source, seed_receipt, set_failure_reason, snapshot, snapshot_path, - text, with_sandbox_params, + receipt_cleared, restore_source, seed_receipt, seed_teardown_job, set_failure_reason, snapshot, + snapshot_path, teardown_job_failed, text, with_sandbox_params, }; use crate::db::{ ActivationKind, ActivationRouteAdmission, CleanupCompletion, CleanupEpisodeSpec, - QuarantineAdoptionOutcome, RestoreAdoptOrphanInsertError, SandboxStateWriteOutcome, - StartupResidentSettlement, SupersedeActivationError, adopt_quarantine_cleanup_episode, - begin_activation_with_route_evidence, begin_teardown_suspend, get_sandbox, - immediate_transaction, insert_cleanup_episode, insert_quarantined_sandbox_in_transaction, - insert_restore_transfer_scope, mark_local_artifact_backup_completed, - pin_suspended_artifact_for_cleanup, request_cleanup_episode, - restore_adopt_orphan_or_insert_sandbox, restore_checkpoint_state, settle_startup_resident, - stage_teardown_suspend_artifact, test_conn, update_state_superseding_activation_in_transaction, + QuarantineAdoptionOutcome, RestoreAdoptOrphanInsertError, SandboxStateWriteNotApplied, + StartupResidentSettlement, StateWriteGuard, SupersedeActivationError, + adopt_quarantine_cleanup_episode, begin_activation_with_route_evidence, begin_teardown_suspend, + get_sandbox, immediate_transaction, insert_cleanup_episode, + insert_quarantined_sandbox_in_transaction, insert_restore_transfer_scope, + mark_local_artifact_backup_completed, pin_suspended_artifact_for_cleanup, + request_cleanup_episode, restore_adopt_orphan_or_insert_sandbox, restore_checkpoint_state, + settle_startup_resident, stage_teardown_suspend_artifact, test_conn, + update_state_superseding_activation_in_transaction, }; use crate::restore_operation::RemoteRestoreOperation; use crate::state::SandboxState; @@ -115,8 +116,10 @@ fn superseded(target: SandboxState) -> Vec<(&'static str, Cell)> { ] } -/// Per-source-state outcomes are `update_state`'s, pinned by -/// `update_state_matrix`; these cases pin what supersession adds around it. +/// The write answers each source state as `update_state_from` expecting every +/// state does, which `update_state_matrix` pins; these cases pin what +/// supersession adds around it, and admission's full write over a row already +/// in its target. #[test] fn activation_supersession_retires_the_episode_before_the_generic_write() { for target in [SandboxState::Stopping, SandboxState::Deleting] { @@ -190,8 +193,9 @@ fn activation_supersession_retires_the_episode_before_the_generic_write() { |result| { matches!( result, - Err(SupersedeActivationError::StateWriteNotApplied(error)) - if error.outcome == SandboxStateWriteOutcome::BlockedByQuarantine + Err(SupersedeActivationError::StateWriteNotApplied( + SandboxStateWriteNotApplied::Guard(StateWriteGuard::Quarantined) + )) ) }, ); @@ -246,6 +250,31 @@ fn activation_supersession_retires_the_episode_before_the_generic_write() { ); } + // Admission re-admits a row already in its target with the whole + // generic write: the failure reason and a receipt for another state + // are cleared and `updated_at` advances, while `state_changed_at` + // keeps the original transition. Re-admitting `deleting` also fails + // the active teardown jobs. + let readmitted = insert_row(&conn, target); + if target == SandboxState::Deleting { + seed_teardown_job(&conn, &readmitted); + } + decorate(&conn, &readmitted, SandboxState::Started); + let mut cells = vec![("failure_reason", NULL), ("updated_at", Cell::NowText)]; + cells.extend(receipt_cleared(SandboxState::Started)); + let expected = if target == SandboxState::Deleting { + teardown_job_failed(&readmitted, "sandbox is deleting") + } else { + ExpectedDiff::new() + } + .sandbox(&readmitted, cells); + assert_accepts( + &format!("{target} re-admitted"), + &conn, + || supersede(&conn, &readmitted, target).unwrap(), + expected, + ); + refuses( &format!("missing row -> {target}"), &conn, @@ -253,8 +282,9 @@ fn activation_supersession_retires_the_episode_before_the_generic_write() { |result| { matches!( result, - Err(SupersedeActivationError::StateWriteNotApplied(error)) - if error.outcome == SandboxStateWriteOutcome::Missing + Err(SupersedeActivationError::StateWriteNotApplied( + SandboxStateWriteNotApplied::Missing + )) ) }, ); 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 fb7f47b68d..1513fea2a1 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 @@ -1,9 +1,8 @@ //! `update_state_from`, the expected-state generic writer of the node-local //! SQLite `sandboxes.state`: with every state expected it answers each cell of -//! `update_state`'s matrix identically, and a from-set that excludes the row's -//! state refuses with the row unchanged. Each runs standalone and inside a -//! caller's transaction. An edge binds the from-set and target a call site -//! writes. +//! the generic guards' matrix, and a from-set that excludes the row's state +//! refuses with the row unchanged. Each runs standalone and inside a caller's +//! transaction. An edge binds the from-set and target a call site writes. use super::state_write_fixtures::{assert_diff, assert_unchanged, snapshot, test_start}; use super::update_state_matrix::{ @@ -11,23 +10,24 @@ use super::update_state_matrix::{ sweep_lifecycle_pairs_and_cleanup_shapes, sweep_quarantine_episode_and_missing_row, write_from, }; use crate::db::{ - EDGES, StateSet, StateWriteOutcome, StateWriteRefusal, edges, test_conn, update_state_from, + EDGES, StateSet, StateWriteGuard, StateWriteOutcome, StateWriteRefusal, edges, test_conn, + update_state_from, }; use crate::state::SandboxState; const IN_TRANSACTION: [bool; 2] = [false, true]; #[test] -fn expecting_every_state_matches_update_state_on_every_lifecycle_pair_and_cleanup_shape() { +fn expecting_every_state_answers_every_lifecycle_pair_and_cleanup_shape() { for in_transaction in IN_TRANSACTION { - sweep_lifecycle_pairs_and_cleanup_shapes(Writer::UpdateStateFromAll { in_transaction }); + sweep_lifecycle_pairs_and_cleanup_shapes(Writer { in_transaction }); } } #[test] -fn expecting_every_state_matches_update_state_over_a_quarantine_episode_and_a_missing_row() { +fn expecting_every_state_answers_a_quarantine_episode_and_a_missing_row() { for in_transaction in IN_TRANSACTION { - sweep_quarantine_episode_and_missing_row(Writer::UpdateStateFromAll { in_transaction }); + sweep_quarantine_episode_and_missing_row(Writer { in_transaction }); } } @@ -58,8 +58,12 @@ fn expected_outcome( match cell(PLAIN[index(from)], to) { Expected::Applied if from_set_contains_row => Some(StateWriteOutcome::Applied { from }), Expected::Applied => Some(refused(StateWriteRefusal::FromState)), - Expected::Quarantine => Some(refused(StateWriteRefusal::Quarantined)), - Expected::Superseded => Some(refused(StateWriteRefusal::CleanupOwned)), + Expected::Quarantine => Some(refused(StateWriteRefusal::Guard( + StateWriteGuard::Quarantined, + ))), + Expected::Superseded => Some(refused(StateWriteRefusal::Guard( + StateWriteGuard::CleanupOwned, + ))), Expected::Rejected => None, Expected::Missing => panic!("a seeded row cannot be missing"), } diff --git a/crates/libs/sbx/mgr/server/src/db/tests/update_state_matrix.rs b/crates/libs/sbx/mgr/server/src/db/tests/update_state_matrix.rs index d8e4d4e36d..3be1b01ab6 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/update_state_matrix.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/update_state_matrix.rs @@ -1,15 +1,13 @@ //! Every outcome of the generic node-local SQLite `sandboxes.state` writer, -//! `update_state`, over all 18×18 `(from, to)` lifecycle pairs and each -//! cleanup-ownership shape that changes its answer, once with a transition -//! receipt for the target and once with a receipt for another state. +//! `update_state_from` with every state expected, over all 18×18 `(from, to)` +//! lifecycle pairs and each cleanup-ownership shape that changes its answer, +//! once with a transition receipt for the target and once with a receipt for +//! another state. With every state expected only the generic guards refuse, so +//! the grids pin those guards. //! //! The expected outcomes are written out below as one grid per shape, not -//! derived from `SandboxState::can_transition_to`: `update_state` deliberately -//! has no expected-state predicate, so the grids record which writes it applies -//! today, including edges the state machine forbids. -//! -//! The same grids and row diffs drive `update_state_from` with every state -//! expected, which must answer every cell exactly as `update_state` does. +//! derived from `SandboxState::can_transition_to`: the grids record which +//! writes the guards admit, including edges the state machine forbids. use pplx_space_common_core::types::{ SandboxCleanupEpisodeId, SandboxId, SandboxQuarantineReasonCode, @@ -22,10 +20,10 @@ use super::state_write_fixtures::{ test_start, text, }; use crate::db::{ - CleanupCompletion, CleanupEpisodeRequestOutcome, CleanupEpisodeSpec, SandboxStateWriteOutcome, - StateSet, StateWriteOutcome, StateWriteRefusal, begin_teardown_suspend, insert_cleanup_episode, + CleanupCompletion, CleanupEpisodeRequestOutcome, CleanupEpisodeSpec, StateSet, StateWriteGuard, + StateWriteOutcome, StateWriteRefusal, begin_teardown_suspend, insert_cleanup_episode, mark_suspending_quarantined, request_cleanup_episode, set_cleanup_intent, test_conn, - update_state, update_state_from_states, + update_state_from_states, }; use crate::state::SandboxState; @@ -345,21 +343,17 @@ pub(super) fn applied_diff( ) } -/// A generic writer the grids characterize. +/// `update_state_from` with every state expected, standalone or inside a +/// caller's transaction. #[derive(Clone, Copy, Debug)] -pub(super) enum Writer { - UpdateState, - /// `update_state_from` with every state expected, standalone or inside a - /// caller's transaction. - UpdateStateFromAll { - in_transaction: bool, - }, +pub(super) struct Writer { + pub(super) in_transaction: bool, } impl Writer { /// Write `to` and name the outcome in the grids' terms. `seeded` is the - /// row's state, which `update_state_from` must report as found or as the - /// applied write's origin. + /// row's state, which the writer must report as found or as the applied + /// write's origin. fn write( self, case: &str, @@ -368,34 +362,27 @@ impl Writer { seeded: Option, to: SandboxState, ) -> Expected { - match self { - Self::UpdateState => match update_state(conn, id, to) { - Ok(SandboxStateWriteOutcome::Applied) => Expected::Applied, - Ok(SandboxStateWriteOutcome::BlockedByQuarantine) => Expected::Quarantine, - Ok(SandboxStateWriteOutcome::SupersededByCleanupEpisode) => Expected::Superseded, - Ok(SandboxStateWriteOutcome::Missing) => Expected::Missing, - Err(rusqlite::Error::InvalidQuery) => Expected::Rejected, + let (observed, expected) = + match write_from(conn, id, StateSet::ALL, to, self.in_transaction) { + Ok(StateWriteOutcome::Applied { from }) => (Some(from), Expected::Applied), + Ok(StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::Guard(StateWriteGuard::Quarantined), + }) => (Some(found), Expected::Quarantine), + Ok(StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned), + }) => (Some(found), Expected::Superseded), + Ok(StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::FromState, + }) => panic!("{case}: every state is expected, yet {found} was refused"), + Ok(StateWriteOutcome::Missing) => (None, Expected::Missing), + Err(rusqlite::Error::InvalidQuery) => return Expected::Rejected, Err(error) => panic!("{case}: {error}"), - }, - Self::UpdateStateFromAll { in_transaction } => { - match write_from(conn, id, StateSet::ALL, to, in_transaction) { - Ok(StateWriteOutcome::Applied { from }) if Some(from) == seeded => { - Expected::Applied - } - Ok(StateWriteOutcome::Refused { - found, - refusal: StateWriteRefusal::Quarantined, - }) if Some(found) == seeded => Expected::Quarantine, - Ok(StateWriteOutcome::Refused { - found, - refusal: StateWriteRefusal::CleanupOwned, - }) if Some(found) == seeded => Expected::Superseded, - Ok(StateWriteOutcome::Missing) if seeded.is_none() => Expected::Missing, - Err(rusqlite::Error::InvalidQuery) => Expected::Rejected, - other => panic!("{case}: seeded {seeded:?}, got {other:?}"), - } - } - } + }; + assert_eq!(observed, seeded, "{case}: the outcome names the seeded row"); + expected } } @@ -511,16 +498,6 @@ pub(super) fn sweep_quarantine_episode_and_missing_row(writer: Writer) { } } -#[test] -fn generic_state_write_outcomes_cover_every_lifecycle_pair_and_cleanup_shape() { - sweep_lifecycle_pairs_and_cleanup_shapes(Writer::UpdateState); -} - -#[test] -fn generic_state_write_outcomes_over_a_quarantine_episode_and_a_missing_row() { - sweep_quarantine_episode_and_missing_row(Writer::UpdateState); -} - #[test] fn every_lifecycle_state_has_one_grid_position() { let mut seen = [false; 18]; diff --git a/crates/libs/sbx/mgr/server/src/service/auth_lookup/resume_begin/tests.rs b/crates/libs/sbx/mgr/server/src/service/auth_lookup/resume_begin/tests.rs index 6f61633fdb..824b67210e 100644 --- a/crates/libs/sbx/mgr/server/src/service/auth_lookup/resume_begin/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/auth_lookup/resume_begin/tests.rs @@ -19,7 +19,6 @@ use pplx_space_node_mgr_client::proto::{ResumeSandboxRequest, ResumeSandboxRespo use tonic::{Code, Status}; use super::BEGIN_BARRIER; -use crate::db::CleanupCompletion; use crate::db::tests::state_write_fixtures::snapshot; use crate::node_manager_transport::fake::FakeNodeManager; use crate::proto::ResumeRequest; @@ -157,29 +156,7 @@ fn move_row(conn: &rusqlite::Connection, id: &str, found: Found) { }; match state { SandboxState::Paused => {} - SandboxState::Stopped => { - 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(); - } + SandboxState::Stopped => crate::db::tests::state_write_fixtures::stop(conn, id), state => { // The schema ties the restore columns to `restoring`. conn.execute( diff --git a/crates/libs/sbx/mgr/server/src/service/auth_lookup/tests.rs b/crates/libs/sbx/mgr/server/src/service/auth_lookup/tests.rs index 210e7f8413..0eaafa8edf 100644 --- a/crates/libs/sbx/mgr/server/src/service/auth_lookup/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/auth_lookup/tests.rs @@ -43,10 +43,7 @@ async fn gateway_identity_mapping_preserves_exact_pid_and_lifecycle_authority() crate::db::get_sandbox(&conn, &id).unwrap().unwrap().uid, Some(owner) ); - crate::db::update_state(&conn, &id, SandboxState::Stopped) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state(&conn, &id, SandboxState::Stopped); } service .lookup_sandbox_by_net_gateway_process(4242, isolated) diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup.rs b/crates/libs/sbx/mgr/server/src/service/cleanup.rs index 5daa54ba7b..ef43961324 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup.rs @@ -378,8 +378,6 @@ pub(super) enum CleanupCompletionError { Database(#[from] rusqlite::Error), #[error(transparent)] RowIntent(#[from] crate::db::CompleteRowIntentCleanupError), - #[error(transparent)] - StateWriteNotApplied(#[from] crate::db::SandboxStateWriteNotApplied), } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -1236,9 +1234,9 @@ impl SandboxManagerService { Err(crate::db::SupersedeActivationError::QuarantineRecoveryInProgress) => { return Err(CleanupAdmissionError::QuarantineRecoveryInProgress { sandbox_id }); } - Err(crate::db::SupersedeActivationError::StateWriteNotApplied(error)) - if error.outcome == crate::db::SandboxStateWriteOutcome::Missing => - { + Err(crate::db::SupersedeActivationError::StateWriteNotApplied( + crate::db::SandboxStateWriteNotApplied::Missing, + )) => { if mode.auto_lifecycle_delete_expectation().is_some() { return Err(Status::unavailable( "sandbox disappeared during auto-lifecycle delete admission", diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs b/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs index ac71d11df7..39b79f6271 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs @@ -1,10 +1,13 @@ -//! Stop's cleanup admission from every node-local SQLite `sandboxes.state` -//! source state, and the Stop RPC's runtime gate in front of it. +//! Stop's and Delete's cleanup admission from every node-local SQLite +//! `sandboxes.state` source state, and the Stop RPC's runtime gate in front of +//! it. //! //! Stop admission does not consult `SandboxState::transition`, unlike Kill: it -//! supersedes any activation and writes `stopping` through the generic writer, -//! so every state the generic writer does not refuse is admitted. The -//! expected outcomes are written out here, not derived from that writer. +//! supersedes any activation and writes `stopping` over every state the generic +//! guards do not refuse, a `stopping` row included. Delete admission writes +//! `deleting` the same way, over a `deleting` row too, unless the row's suspend +//! state has not reached the cluster. The expected outcomes are written out +//! here, not derived from that writer. //! //! The Stop RPC admits only a sandbox whose runtime is live: it resolves its //! target with `running_only`, which selects nothing unless the sandbox's @@ -23,7 +26,10 @@ use pplx_space_common_core::types::SandboxId; use pplx_space_common_svc::PeerIdentity; use tonic::Request; -use super::{CleanupAdmission, SandboxManagerService, SandboxManagerServiceOptions, StopRequest}; +use super::{ + CleanupAdmission, CleanupAdmissionError, SandboxManagerService, SandboxManagerServiceOptions, + StopRequest, +}; use crate::db::CleanupCompletion; use crate::db::tests::state_write_fixtures::{assert_unchanged, snapshot}; use crate::service::tests::service_tests::insert_test_sandbox; @@ -80,13 +86,14 @@ fn insert_row(conn: &rusqlite::Connection, state: SandboxState) -> SandboxId { #[derive(Clone, Copy, Debug, Eq, PartialEq)] enum Admission { - /// Admitted, and the Stop RPC can reach it for a live runtime. + /// Admitted, and the RPC can reach it: Stop only for a live runtime. Admitted, /// Admitted, but only by calling admission directly: the state has no /// Firecracker process, so the Stop RPC's runtime gate selects nothing. AdmittedOnlyDirectly, - /// Refused with `Unavailable`. - Refused, + /// Refused: Stop with `Unavailable` and this message, Delete (`None`) + /// because the row's suspend state has not reached the cluster. + Refused(Option<&'static str>), } /// What Stop admission answers for a plain row in `state`. Wildcard-free, so @@ -109,60 +116,107 @@ fn admission(state: SandboxState) -> Admission { | SandboxState::Created | SandboxState::Stopped | SandboxState::Suspended => Admission::AdmittedOnlyDirectly, - // The generic writer leaves a `deleting` row to its delete and a + // The generic guards leave a `deleting` row to its delete and a // `quarantined` row to quarantine. - SandboxState::Deleting | SandboxState::Quarantined => Admission::Refused, + SandboxState::Deleting => Admission::Refused(Some( + "cleanup lifecycle claim was superseded: sandbox state write was not applied: \ + superseded_by_cleanup_episode", + )), + SandboxState::Quarantined => Admission::Refused(Some( + "cleanup lifecycle claim was superseded: sandbox state write was not applied: \ + blocked_by_quarantine", + )), } } -/// Assert one admission's outcome: an admitted Stop leaves the row `stopping` -/// with its intent and cleanup job; a refused one changes nothing. -fn check_admission( - case: &str, - service: &SandboxManagerService, - id: &str, - state: SandboxState, - persist_intent: bool, - result: Result, - before: &crate::db::tests::state_write_fixtures::Snapshot, -) { - let db = service.db_conn().unwrap(); - if admission(state) == Admission::Refused { - let status = match result { - Err(error) => error.into_status(), - Ok(_) => panic!("{case}: expected a refusal"), - }; - assert_eq!( - status.code(), - tonic::Code::Unavailable, - "{case}: {status:?}" - ); - assert_unchanged(case, before, &db); - return; +/// What Delete admission answers for a plain row in `state`. Wildcard-free, +/// so a new lifecycle state does not compile until it is placed. +fn delete_admission(state: SandboxState) -> Admission { + match state { + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopping + | SandboxState::Stopped + | SandboxState::Deleting + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Quarantined + | SandboxState::Failed => Admission::Admitted, + // Its suspend state has not reached the cluster yet. + SandboxState::Suspending => Admission::Refused(None), } - assert!( - matches!(result, Ok(CleanupAdmission::Persisted { .. })), - "{case}: expected a persisted admission" - ); - let row = crate::db::get_sandbox(&db, id).unwrap().unwrap(); - assert_eq!(row.state, SandboxState::Stopping, "{case}"); - assert_eq!( - (row.cleanup_completion, row.cleanup_keep_filesystem), - if persist_intent { - (Some(CleanupCompletion::MarkStopped), Some(true)) - } else { - (None, None) - }, - "{case}" - ); - assert!( - pplx_space_common_jobs::db::find_active_job(&db, crate::db::cleanup_jobs::KIND, id) - .unwrap() - .is_some(), - "{case}: admission persists its cleanup job" - ); } +/// The lifecycle cleanup an admission sweep runs. +#[derive(Clone, Copy, Debug)] +enum Cleanup { + Stop, + Delete, +} + +impl Cleanup { + fn completion(self) -> CleanupCompletion { + match self { + Self::Stop => CleanupCompletion::MarkStopped, + Self::Delete => CleanupCompletion::DeleteRow, + } + } + + fn target(self) -> SandboxState { + match self { + Self::Stop => SandboxState::Stopping, + Self::Delete => SandboxState::Deleting, + } + } + + fn keep_filesystem(self) -> bool { + match self { + Self::Stop => true, + Self::Delete => false, + } + } + + fn admission(self, state: SandboxState) -> Admission { + match self { + Self::Stop => admission(state), + Self::Delete => delete_admission(state), + } + } + + /// Whether `result` is this cleanup's refusal of a row in `state`; for + /// Stop, with exactly `message`. + fn refused( + self, + state: SandboxState, + message: Option<&str>, + result: Result, + ) -> bool { + match (self, result) { + (Self::Stop, Err(error)) => { + let status = error.into_status(); + (status.code(), Some(status.message())) == (tonic::Code::Unavailable, message) + } + ( + Self::Delete, + Err(CleanupAdmissionError::SuspendPublicationPending { state: found, .. }), + ) => found == state, + (Self::Delete, Err(_)) | (Self::Stop | Self::Delete, Ok(_)) => false, + } + } +} + +/// Admit `cleanup` over a plain row in every source state, with and without +/// a persisted intent. An admitted cleanup leaves the row in its target state +/// with its intent and cleanup job; a refused one changes nothing. +/// /// Admission never consults the shutdown token, but the cleanup executor it /// starts does not start once the service is shut down. Each case cancels the /// token and waits for the service's own tasks to exit before admitting, so @@ -170,23 +224,16 @@ fn check_admission( /// whole-database comparison sees admission's writes alone, and an admitted /// row stays as admission left it. The empty task tracker after admission /// checks that. -#[tokio::test(flavor = "current_thread")] -async fn stop_admission_answers_every_source_state() { - if !crate::service::tests::enter_route_test_process(&[ - ("EPPO_SDK_KEY", None), - ("SPACE_FLAGD_ENABLED", None), - ]) { - return; - } - // A non-blocking Stop persists its intent; a blocking one does not. +async fn sweep_every_source_state(cleanup: Cleanup) { for persist_intent in [true, false] { for state in SandboxState::all() { let case = format!( - "{state} ({:?}), persist intent {persist_intent}", - admission(state) + "{cleanup:?} of {state} ({:?}), persist intent {persist_intent}", + cleanup.admission(state) ); let conn = crate::db::test_conn(); let sandbox_id = insert_row(&conn, state); + let id = sandbox_id.to_string(); let scratch = tempfile::tempdir().unwrap(); let service = isolated_service(conn, scratch.path()); // The service's own background tasks exit on cancellation; once @@ -205,7 +252,7 @@ async fn stop_admission_answers_every_source_state() { WAIT, service.admit_selected_cleanup( sandbox_id, - CleanupCompletion::MarkStopped, + cleanup.completion(), false, persist_intent, None, @@ -219,20 +266,69 @@ async fn stop_admission_answers_every_source_state() { "{case}: no cleanup executor started" ); - check_admission( - &case, - &service, - &sandbox_id.to_string(), - state, - persist_intent, - result, - &before, - ); + { + let db = service.db_conn().unwrap(); + if let Admission::Refused(message) = cleanup.admission(state) { + assert!( + cleanup.refused(state, message, result), + "{case}: expected a refusal" + ); + assert_unchanged(&case, &before, &db); + } else { + assert!( + matches!(result, Ok(CleanupAdmission::Persisted { .. })), + "{case}: expected a persisted admission" + ); + let row = crate::db::get_sandbox(&db, &id).unwrap().unwrap(); + assert_eq!(row.state, cleanup.target(), "{case}"); + assert_eq!( + (row.cleanup_completion, row.cleanup_keep_filesystem), + if persist_intent { + (Some(cleanup.completion()), Some(cleanup.keep_filesystem())) + } else { + (None, None) + }, + "{case}" + ); + assert!( + pplx_space_common_jobs::db::find_active_job( + &db, + crate::db::cleanup_jobs::KIND, + &id + ) + .unwrap() + .is_some(), + "{case}: admission persists its cleanup job" + ); + } + } shutdown(service).await; } } } +#[tokio::test(flavor = "current_thread")] +async fn stop_admission_answers_every_source_state() { + if !crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) { + return; + } + sweep_every_source_state(Cleanup::Stop).await; +} + +#[tokio::test(flavor = "current_thread")] +async fn delete_admission_answers_every_source_state() { + if !crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) { + return; + } + sweep_every_source_state(Cleanup::Delete).await; +} + /// The Stop RPC selects nothing without a live runtime, whatever the row's /// state, and admits nothing. A live runtime is admitted, as /// `stop_selects_only_a_live_runtime_and_both_treat_an_empty_template_filter_as_none` 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 92d76d073a..c2546b8f37 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs @@ -224,30 +224,13 @@ pub(super) async fn service_cleanup_via_node_with_options( pub(super) use crate::db::CleanupCompletion; -#[cfg_attr( - not(test), - expect( - dead_code, - reason = "the update_state parity tests call it; production writes use update_state_from" - ) -)] -pub(super) fn require_state_write( - result: rusqlite::Result, - context: &'static str, -) -> Result<(), Status> { - let outcome = result.map_err(|error| Status::internal(format!("{context}: {error}")))?; - outcome - .require_applied() - .map_err(|error| Status::aborted(format!("{context}: {error}"))) -} - -/// [`require_state_write`] for [`crate::db::update_state_from`]. A database +/// Map a [`crate::db::update_state_from`] result to a status. A database /// error is `Internal`. Every refusal is `Aborted`: a from-state refusal /// reports the state another transition left the row in, and a generic-guard -/// refusal or a missing row keeps `update_state`'s message. +/// refusal or a missing row renders [`crate::db::SandboxStateWriteNotApplied`]. pub(super) fn require_state_write_from( result: rusqlite::Result, - context: &'static str, + context: &str, ) -> Result<(), Status> { let outcome = result.map_err(|error| Status::internal(format!("{context}: {error}")))?; state_write_result(outcome) @@ -264,22 +247,15 @@ pub(super) enum StateWriteRejected { FromState { found: SandboxState, }, - /// A generic guard refused, as it refuses `update_state`. The guards - /// answer first, so `found` may also be outside the edge's source states. + /// A generic guard refused. The guards answer first, so `found` may also + /// be outside the edge's source states. Guard { - refusal: StateWriteGuard, + refusal: crate::db::StateWriteGuard, found: SandboxState, }, Missing, } -/// The generic guards `update_state` and `update_state_from` share. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub(super) enum StateWriteGuard { - Quarantined, - CleanupOwned, -} - pub(super) fn state_write_result( outcome: crate::db::StateWriteOutcome, ) -> Result { @@ -292,45 +268,25 @@ pub(super) fn state_write_result( } => Err(StateWriteRejected::FromState { found }), StateWriteOutcome::Refused { found, - refusal: StateWriteRefusal::Quarantined, - } => Err(StateWriteRejected::Guard { - refusal: StateWriteGuard::Quarantined, - found, - }), - StateWriteOutcome::Refused { - found, - refusal: StateWriteRefusal::CleanupOwned, - } => Err(StateWriteRejected::Guard { - refusal: StateWriteGuard::CleanupOwned, - found, - }), + refusal: StateWriteRefusal::Guard(refusal), + } => Err(StateWriteRejected::Guard { refusal, found }), StateWriteOutcome::Missing => Err(StateWriteRejected::Missing), } } impl StateWriteRejected { pub(super) fn into_status(self, context: &str) -> Status { - use crate::db::{SandboxStateWriteNotApplied, SandboxStateWriteOutcome}; - let outcome = match self { + use crate::db::SandboxStateWriteNotApplied; + let not_applied = match self { Self::FromState { found } => { return Status::aborted(format!( "{context}: sandbox state write was superseded: sandbox is {found}" )); } - Self::Guard { - refusal: StateWriteGuard::Quarantined, - .. - } => SandboxStateWriteOutcome::BlockedByQuarantine, - Self::Guard { - refusal: StateWriteGuard::CleanupOwned, - .. - } => SandboxStateWriteOutcome::SupersededByCleanupEpisode, - Self::Missing => SandboxStateWriteOutcome::Missing, + Self::Guard { refusal, .. } => SandboxStateWriteNotApplied::Guard(refusal), + Self::Missing => SandboxStateWriteNotApplied::Missing, }; - Status::aborted(format!( - "{context}: {}", - SandboxStateWriteNotApplied { outcome } - )) + Status::aborted(format!("{context}: {not_applied}")) } } @@ -353,7 +309,7 @@ pub(super) fn update_state_from_best_effort( mark_sandbox_dirty(db, sandbox_id); Some(outcome) } - Ok(outcome @ (StateWriteOutcome::Refused { .. } | StateWriteOutcome::Missing)) => { + Ok(outcome @ StateWriteOutcome::Refused { .. }) => { tracing::warn!( sandbox_id, state = edge.to().as_str(), @@ -362,6 +318,14 @@ pub(super) fn update_state_from_best_effort( ); Some(outcome) } + Ok(outcome @ StateWriteOutcome::Missing) => { + tracing::warn!( + sandbox_id, + state = edge.to().as_str(), + "sandbox state write found no sandbox row" + ); + Some(outcome) + } Err(error) => { tracing::warn!( sandbox_id, diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs b/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs index 0d8effd077..6a24001476 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs @@ -14,13 +14,10 @@ use tokio_util::task::TaskTracker; use tonic::Code; use super::{ - StateWriteGuard, StateWriteRejected, require_state_write, require_state_write_from, - state_write_result, update_state_from_best_effort, + StateWriteRejected, require_state_write_from, state_write_result, update_state_from_best_effort, }; use crate::db::tests::state_write_fixtures::{assert_unchanged, snapshot}; -use crate::db::{ - Edge, SandboxStateWriteOutcome, StateSet, StateWriteOutcome, StateWriteRefusal, edges, -}; +use crate::db::{Edge, StateSet, StateWriteGuard, StateWriteOutcome, StateWriteRefusal, edges}; use crate::service::drain::{CriticalTaskGuard, DrainState}; use crate::service::tests::service_tests::insert_test_sandbox; use crate::state::SandboxState; @@ -31,19 +28,21 @@ const CONTEXT: &str = "complete synthetic transition"; /// until it joins the chain. fn refusals() -> Vec { let next = |refusal| match refusal { - StateWriteRefusal::FromState => Some(StateWriteRefusal::Quarantined), - StateWriteRefusal::Quarantined => Some(StateWriteRefusal::CleanupOwned), - StateWriteRefusal::CleanupOwned => None, + StateWriteRefusal::FromState => { + Some(StateWriteRefusal::Guard(StateWriteGuard::Quarantined)) + } + StateWriteRefusal::Guard(StateWriteGuard::Quarantined) => { + Some(StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned)) + } + StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned) => None, }; std::iter::successors(Some(StateWriteRefusal::FromState), |refusal| next(*refusal)).collect() } -/// The messages `require_state_write` returned for `update_state` before -/// `update_state_from` existed, copied from 8f12bd6: it formats -/// `"{context}: {SandboxStateWriteNotApplied}"` (`service/cleanup_support.rs`) -/// over `SandboxStateWriteOutcome::as_str` (`db/state_transitions.rs`), and -/// `update_state`'s guards answer `BlockedByQuarantine` and -/// `SupersededByCleanupEpisode` there. +/// The messages the retired generic `update_state` caller path returned, +/// copied from 8f12bd6: `"{context}: {SandboxStateWriteNotApplied}"` over +/// `SandboxStateWriteOutcome::as_str`. Clients still receive them for a guard +/// refusal and a missing row. const BLOCKED_BY_QUARANTINE: &str = "complete synthetic transition: sandbox state write was not applied: blocked_by_quarantine"; const SUPERSEDED_BY_CLEANUP_EPISODE: &str = "complete synthetic transition: sandbox state write \ @@ -69,11 +68,11 @@ fn expected_rejection(outcome: StateWriteOutcome) -> Result Ok(from), StateWriteOutcome::Refused { found, refusal } => Err(match refusal { StateWriteRefusal::FromState => StateWriteRejected::FromState { found }, - StateWriteRefusal::Quarantined => StateWriteRejected::Guard { + StateWriteRefusal::Guard(StateWriteGuard::Quarantined) => StateWriteRejected::Guard { refusal: StateWriteGuard::Quarantined, found, }, - StateWriteRefusal::CleanupOwned => StateWriteRejected::Guard { + StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned) => StateWriteRejected::Guard { refusal: StateWriteGuard::CleanupOwned, found, }, @@ -90,8 +89,10 @@ fn expected_status(outcome: StateWriteOutcome) -> Result<(), (Code, String)> { Code::Aborted, format!("{CONTEXT}: sandbox state write was superseded: sandbox is {found}"), ), - StateWriteRefusal::Quarantined => (Code::Aborted, BLOCKED_BY_QUARANTINE.to_owned()), - StateWriteRefusal::CleanupOwned => { + StateWriteRefusal::Guard(StateWriteGuard::Quarantined) => { + (Code::Aborted, BLOCKED_BY_QUARANTINE.to_owned()) + } + StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned) => { (Code::Aborted, SUPERSEDED_BY_CLEANUP_EPISODE.to_owned()) } }), @@ -129,49 +130,18 @@ fn every_state_write_outcome_has_its_rejection_and_status() { } } -/// The literals above are still what the `update_state` caller path returns. #[test] -fn update_state_outcomes_keep_their_statuses() { - for (outcome, expected) in [ - (SandboxStateWriteOutcome::Applied, Ok(())), - ( - SandboxStateWriteOutcome::BlockedByQuarantine, - Err(BLOCKED_BY_QUARANTINE), - ), - ( - SandboxStateWriteOutcome::SupersededByCleanupEpisode, - Err(SUPERSEDED_BY_CLEANUP_EPISODE), - ), - (SandboxStateWriteOutcome::Missing, Err(MISSING)), - ] { - assert_eq!( - code_and_message(require_state_write(Ok(outcome), CONTEXT)), - expected.map_err(|message| (Code::Aborted, message.to_owned())), - "{outcome}" - ); - } -} - -#[test] -fn a_database_error_is_internal_as_for_update_state() { - let from = code_and_message(require_state_write_from( - Err(rusqlite::Error::InvalidQuery), - CONTEXT, - )); +fn a_database_error_is_internal() { assert_eq!( - from, + code_and_message(require_state_write_from( + Err(rusqlite::Error::InvalidQuery), + CONTEXT, + )), Err(( Code::Internal, "complete synthetic transition: Query is not read-only".to_owned() )) ); - assert_eq!( - from, - code_and_message(require_state_write( - Err(rusqlite::Error::InvalidQuery), - CONTEXT - )) - ); } /// A row each generic guard refuses `target` over, or no row. @@ -244,10 +214,10 @@ const GUARDED_ROWS: [GuardedRow; 4] = [ }, ]; -/// For the same row, a guard refusal or a missing row maps to the status the -/// `update_state` caller path returned, even with the row's state expected. +/// A guard refusal or a missing row maps to the status the retired generic +/// caller path returned, even with the row's state expected. #[test] -fn guard_refusals_and_a_missing_row_keep_update_state_statuses() { +fn guard_refusals_and_a_missing_row_keep_their_statuses() { for row in GUARDED_ROWS { let conn = crate::db::test_conn(); let id = SandboxId::new().to_string(); @@ -267,18 +237,12 @@ fn guard_refusals_and_a_missing_row_keep_update_state_statuses() { "{}", row.name ); - let from = code_and_message(require_state_write_from(Ok(outcome), CONTEXT)); - let generic = code_and_message(require_state_write( - crate::db::update_state(&conn, &id, row.target), - CONTEXT, - )); assert_eq!( - from, + code_and_message(require_state_write_from(Ok(outcome), CONTEXT)), Err((Code::Aborted, row.message.to_owned())), "{}", row.name ); - assert_eq!(from, generic, "{}", row.name); assert_unchanged(row.name, &before, &conn); } } @@ -332,7 +296,7 @@ fn best_effort_refusals_change_nothing() { Some(SandboxState::Stopping), StateWriteOutcome::Refused { found: SandboxState::Stopping, - refusal: StateWriteRefusal::CleanupOwned, + refusal: StateWriteRefusal::Guard(StateWriteGuard::CleanupOwned), }, ), ("missing row", None, StateWriteOutcome::Missing), diff --git a/crates/libs/sbx/mgr/server/src/service/create_support.rs b/crates/libs/sbx/mgr/server/src/service/create_support.rs index 9837548cbe..d7e31dc6c5 100644 --- a/crates/libs/sbx/mgr/server/src/service/create_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/create_support.rs @@ -342,6 +342,12 @@ pub(super) fn admit_warm_start(sandbox: &crate::db::SandboxRow) -> Result<(), St /// the meantime needs no write. A row another transition moved anywhere else /// is `FailedPrecondition`, as [`admit_warm_start`] answers it; a generic-guard /// refusal or a missing row keeps `require_state_write_from`'s `Aborted`. +/// A guard refusal of a `suspended` row (`Guard { found: Suspended }`) stays +/// `Aborted`, unlike `FromState { Suspended }`: the guard means cleanup or +/// quarantine owns the row, so it is not a staged row the restore may claim. +/// A transient row answers `FailedPrecondition`, where a cold Start answers +/// `Unavailable`, deliberately: warm start does not wait out a transition, it +/// refuses a row it cannot restage. pub(super) fn write_warm_start_restaged(db: &rusqlite::Connection, id: &str) -> Result<(), Status> { const CONTEXT: &str = "prepare recovered sandbox restore"; let outcome = crate::db::update_state_from(db, id, &crate::db::edges::WARM_START_RESTAGE) diff --git a/crates/libs/sbx/mgr/server/src/service/create_support/state_write_tests.rs b/crates/libs/sbx/mgr/server/src/service/create_support/state_write_tests.rs index 2429d68f0d..ef511c6fea 100644 --- a/crates/libs/sbx/mgr/server/src/service/create_support/state_write_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/create_support/state_write_tests.rs @@ -10,7 +10,7 @@ use pplx_space_common_core::types::SandboxQuarantineReasonCode; use super::*; use crate::db::tests::state_write_fixtures::{ - assert_unchanged, backdate, force_state, insert_creating, insert_row, seed_receipt, + admit_stop, assert_unchanged, backdate, force_state, insert_creating, insert_row, seed_receipt, set_failure_reason, snapshot, }; use crate::db::{ @@ -18,16 +18,6 @@ use crate::db::{ CompleteRowIntentCleanupDisposition, }; -/// A non-blocking Stop's admission transaction: supersede the activation, -/// write `stopping` and persist the `mark_stopped` intent. -fn admit_stop(conn: &rusqlite::Connection, id: &str) { - 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(); -} - fn state(conn: &rusqlite::Connection, id: &str) -> SandboxState { crate::db::get_sandbox(conn, id).unwrap().unwrap().state } @@ -260,7 +250,7 @@ fn check(case: &str, conn: &rusqlite::Connection, id: &str, write: Write, expect } } -/// Every row carries the side columns a same-state `update_state` write would +/// Every row carries the side columns a same-state generic write would /// change: a failure reason, a receipt for another target and backdated /// timestamps. So a writer that rewrites a row it should leave alone, such as /// an already-`suspended` warm-start row, is visible in the row. diff --git a/crates/libs/sbx/mgr/server/src/service/gateway_authority/tests.rs b/crates/libs/sbx/mgr/server/src/service/gateway_authority/tests.rs index 21a67afdd5..7bf47330a6 100644 --- a/crates/libs/sbx/mgr/server/src/service/gateway_authority/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/gateway_authority/tests.rs @@ -458,7 +458,7 @@ async fn gateway_authority_rechecks_ownership_after_await_but_not_ordinary_polic crate::db::update_netgtw_pid(&db, &id.to_string(), sandbox.uid.unwrap(), sandbox.netgtw_pid.unwrap() + 1).unwrap(); } "pool" => { crate::db::set_pool_template(&db, &id.to_string(), Some("warm-pool")).unwrap(); } - "stopping" => { assert!(matches!(crate::db::update_state(&db, &id.to_string(), SandboxState::Stopping).unwrap(), crate::db::SandboxStateWriteOutcome::Applied)); } + "stopping" => crate::db::tests::state_write_fixtures::apply_edge(&db, &id.to_string(), &crate::db::edges::STOP_ADMISSION), "cleanup" => { crate::db::mark_cleanup_required(&db, &id.to_string(), true, crate::db::CleanupCompletion::MarkSuspended, "restore failed").unwrap(); } "policy" => { crate::db::update_network_policy_override(&db, &id.to_string(), Some(&NetworkPolicyV1::deny_all()), 2, crate::db::PolicyAuthority::Cluster, false, false).unwrap(); } _ => unreachable!(), diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core/tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core/tests.rs index 595061be7c..71157b0608 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_core/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core/tests.rs @@ -77,10 +77,11 @@ async fn legacy_delete_after_durable_delete_does_not_dispatch_node_cleanup() { ); { let db = service.db.lock().unwrap(); - crate::db::update_state(&db, &id, SandboxState::Deleting) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &db, + &id, + &crate::db::edges::DELETE_ADMISSION, + ); crate::db::set_cleanup_intent(&db, &id, false, CleanupCompletion::DeleteRow).unwrap(); crate::db::complete_row_intent_cleanup(&db, &id, CleanupCompletion::DeleteRow, None) .unwrap(); 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 f19a5cfd3a..0871472e74 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 @@ -204,9 +204,11 @@ pub(super) async fn pause_with_reconciliation( ) { // 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. + // also answers false on a poisoned connection or a database error, so + // the row is read again: a row found outside `pausing`, a + // `quarantined` one included, was moved by another writer. A failed + // read, a missing row or a row still `pausing` reports pause's + // quarantine. let found = crate::node_state_sync::connection_or_signal_failure(db) .ok() .and_then(|db| { 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 index 6eb353eeb1..138e790a4f 100644 --- 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 @@ -16,7 +16,9 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use super::*; -use crate::db::tests::state_write_fixtures::{Snapshot, assert_unchanged, force_state, snapshot}; +use crate::db::tests::state_write_fixtures::{ + Snapshot, assert_unchanged, force_state, snapshot, stop, +}; use crate::guest_filesystem_flush::GuestFlushError; use crate::node_manager_transport::fake::{FakeNodeManager, ServedNodeManager}; use crate::service::activation_support::internal_request; @@ -151,29 +153,7 @@ enum Interleave { 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::Stop => stop(conn, id), Self::Force(SandboxState::Pausing) => {} Self::Force(state) => force_state(conn, id, state), } @@ -196,7 +176,7 @@ enum Expected { } /// 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 +/// as the generic guards' 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> { 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 74c37cf9dc..89ec86c2d7 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 @@ -368,6 +368,7 @@ impl SandboxManagerService { let guest_preparation_budget = context.guest_preparation_budget(); if let Err(e) = prepare_guest_for_pause(&sandbox.id, guest_preparation_budget).await { let db = self.db_conn()?; + // Unreachable under suspend's fence and locks; a refusal answers `Internal`. if let Err(error) = require_state_write_from( crate::db::update_state_from( &db, @@ -442,6 +443,7 @@ impl SandboxManagerService { "fc_pause_failed", reason.clone(), ); + // A refusal is unreachable under suspend's fence and locks; it answers `Internal`. return Err(Status::internal(reason)); } { diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend/tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend/tests.rs index 1c55ae5f4a..f65798c8dd 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend/tests.rs @@ -94,8 +94,8 @@ enum Expected { } /// The generic guard that refuses a `pausing`, `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 +/// over `found`, as the generic guards' 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 { 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 ee181a664a..14a2e3c6e8 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 @@ -210,10 +210,11 @@ async fn artifact_alias_waits_on_canonical_lock_and_rereads_after_admission() { let pending = service.lock_artifact_for_sandbox("01939e3e1111"); tokio::pin!(pending); assert!(futures_util::poll!(&mut pending).is_pending()); - crate::db::update_state(&service.db_conn().unwrap(), id, SandboxState::Stopped) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &service.db_conn().unwrap(), + id, + &crate::db::edges::AUTO_LIFECYCLE_STOP_SUSPENDED, + ); drop(lock); let (row, guard) = pending.await.unwrap(); assert_eq!(row.id, id); @@ -1383,10 +1384,11 @@ fn deleting_sandbox_with_closed_teardown_suspend( .status, crate::db::TeardownSuspendOperationStatus::Running ); - crate::db::update_state(conn, &id, SandboxState::Deleting) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + conn, + &id, + &crate::db::edges::DELETE_ADMISSION, + ); // The transition fails the job before any shim request arrives. assert_eq!( crate::db::latest_teardown_suspend_operation(conn, &id) @@ -1796,9 +1798,10 @@ async fn teardown_suspend_pause_failure_recovery_allows_same_generation_retry() let original = crate::db::latest_teardown_suspend_operation(&conn, &id) .unwrap() .unwrap(); - assert_eq!( - crate::db::update_state(&conn, &id, SandboxState::Pausing).unwrap(), - crate::db::SandboxStateWriteOutcome::Applied + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &id, + &crate::db::edges::PAUSE_BEGIN, ); crate::db::mark_quarantined_from_state( &conn, @@ -2264,10 +2267,11 @@ async fn suspend_completion_refuses_a_superseded_row_before_runtime_cleanup() { match supersession { Supersession::CleanupActive => {} Supersession::Deleted => { - crate::db::update_state(&conn, &id, SandboxState::Deleting) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &id, + &crate::db::edges::DELETE_ADMISSION, + ); commit_sandbox_delete(&conn, &id); assert!(crate::db::get_sandbox(&conn, &id).unwrap().is_none()); } diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone/tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone/tests.rs index 3d60248eb2..68a8f41746 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_pool_wait_clone/tests.rs @@ -338,10 +338,7 @@ async fn pool_clone_and_adoption_keep_donor_ownership_exclusive() { { let db = service.db.lock().unwrap(); assert!(crate::db::complete_pool_clone(&db, &id, cleanup_id).unwrap()); - crate::db::update_state(&db, &id, SandboxState::Paused) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state(&db, &id, SandboxState::Paused); crate::db::update_user(&db, &id, &TEMPLATE_MANAGER_SANDBOX_OWNER).unwrap(); crate::db::set_pool_template(&db, &id, Some("seed")).unwrap(); } diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/tests.rs index 0828602b0b..9107b51998 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/tests.rs @@ -1,4 +1,5 @@ use super::*; +use crate::db::tests::state_write_fixtures::{admit_stop, complete_stop}; #[test] fn snapshot_inventory_does_not_return_partial_success_for_oversized_sidecars() { @@ -348,28 +349,6 @@ fn sandbox_state(conn: &rusqlite::Connection, id: &str) -> SandboxState { crate::db::get_sandbox(conn, id).unwrap().unwrap().state } -/// Stop admission as the cleanup fence runs it, with a non-blocking Stop's -/// persisted intent. -fn admit_stop(conn: &rusqlite::Connection, id: &str) { - 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(); -} - -fn complete_stop(conn: &rusqlite::Connection, id: &str) { - crate::db::complete_row_intent_cleanup_for_job( - conn, - id, - CleanupCompletion::MarkStopped, - None, - None, - None, - ) - .unwrap(); -} - /// The full rollback holds the artifact lock, which Stop does not take, while /// it rolls the disk back and copies the snapshot. A Stop admitted in that /// window must not be overwritten by the restage to `suspended`: it keeps its diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot/tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot/tests.rs index 865ae09e80..e6e44a9dbd 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot/tests.rs @@ -147,28 +147,6 @@ fn sandbox_inventory_dirty(conn: &rusqlite::Connection, id: &str) -> bool { }) } -/// Stop admission as the cleanup fence runs it, with a non-blocking Stop's -/// persisted intent, then Stop's row-intent completion. -fn stop_sandbox(conn: &rusqlite::Connection, id: &str) { - 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, crate::db::CleanupCompletion::MarkStopped) - .unwrap() - ); - tx.commit().unwrap(); - crate::db::complete_row_intent_cleanup_for_job( - conn, - id, - crate::db::CleanupCompletion::MarkStopped, - None, - None, - None, - ) - .unwrap(); -} - /// The template base-snapshot completion and its failure rollback hold only /// the wake lock, which Stop does not take. A Stop that completes while the /// snapshot is captured or hashed leaves `stopped`, and neither write may @@ -185,7 +163,7 @@ fn template_base_snapshot_writes_leave_a_stop_that_completed_during_capture_stop ); begin_template_base_snapshot_state(&conn, &id).unwrap(); assert_eq!(sandbox_state(&conn, &id), SandboxState::Checkpointing); - stop_sandbox(&conn, &id); + crate::db::tests::state_write_fixtures::stop(&conn, &id); assert_eq!(sandbox_state(&conn, &id), SandboxState::Stopped); let before = crate::db::tests::database_rows(&conn); let db = Arc::new(Mutex::new(conn)); diff --git a/crates/libs/sbx/mgr/server/src/service/reconcile/tests.rs b/crates/libs/sbx/mgr/server/src/service/reconcile/tests.rs index 3bceb571ea..6eec651ebc 100644 --- a/crates/libs/sbx/mgr/server/src/service/reconcile/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/reconcile/tests.rs @@ -953,3 +953,53 @@ async fn stop_admitted_during_auto_lifecycle_stop_keeps_its_completion() { "the Stop's completion clears the runtime it stopped" ); } + +/// A Delete admitted between auto-lifecycle Stop's re-read and its write +/// leaves a `deleting` row with a current `delete_row` intent, which the +/// cleanup-ownership guard refuses. The Stop skips without recording the +/// transition or reporting an error, and the delete keeps its row and intent. +#[tokio::test] +async fn delete_admitted_during_auto_lifecycle_stop_keeps_its_row() { + let scratch = tempfile::tempdir().unwrap(); + let service = isolated_service(scratch.path()); + let id = SandboxId::new().to_string(); + insert_due_suspended_sandbox(&service, &id, 10); + + let (result, before) = stop_suspended_with_concurrent_write(&service, &id, |db| { + // A non-blocking Delete's admission transaction. + let admission = crate::db::immediate_transaction(db).unwrap(); + crate::db::update_state_superseding_activation_in_transaction( + &admission, + &id, + SandboxState::Deleting, + ) + .unwrap(); + assert!( + crate::db::set_cleanup_intent( + &admission, + &id, + false, + crate::db::CleanupCompletion::DeleteRow + ) + .unwrap() + ); + admission.commit().unwrap(); + }) + .await; + + assert!(result.is_ok(), "{result:?}"); + let db = service.db.lock().unwrap(); + crate::db::tests::state_write_fixtures::assert_unchanged( + "auto-lifecycle Stop after Delete admission", + &before, + &db, + ); + let row = crate::db::get_sandbox(&db, &id).unwrap().unwrap(); + assert_eq!( + (row.state, row.cleanup_completion), + ( + SandboxState::Deleting, + Some(crate::db::CleanupCompletion::DeleteRow) + ) + ); +} diff --git a/crates/libs/sbx/mgr/server/src/service/recovery/staging_cleanup/tests.rs b/crates/libs/sbx/mgr/server/src/service/recovery/staging_cleanup/tests.rs index 292d093b9b..011d7bfa2a 100644 --- a/crates/libs/sbx/mgr/server/src/service/recovery/staging_cleanup/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/recovery/staging_cleanup/tests.rs @@ -140,10 +140,11 @@ async fn staging_cleanup_rediscovery_survives_rollback_failed_deletion_and_resta SandboxState::Started, SandboxState::Paused, ] { - crate::db::update_state(&service.db.lock().unwrap(), &id, state) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state( + &service.db.lock().unwrap(), + &id, + state, + ); assert_eq!(sweep(&service, cache.path()), expected); assert!( suspend_snapshot_publication::find_complete_suspend_snapshot_staging_dir( @@ -165,10 +166,11 @@ async fn staging_cleanup_rediscovery_survives_rollback_failed_deletion_and_resta fresh_id, b"new attempt", ); - crate::db::update_state(&service.db.lock().unwrap(), &id, SandboxState::Suspending) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state( + &service.db.lock().unwrap(), + &id, + SandboxState::Suspending, + ); let _wake = service .auto_lifecycle .wake_lock(&id) diff --git a/crates/libs/sbx/mgr/server/src/service/route_identity_tests.rs b/crates/libs/sbx/mgr/server/src/service/route_identity_tests.rs index 6ed112eda9..26bdd45543 100644 --- a/crates/libs/sbx/mgr/server/src/service/route_identity_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/route_identity_tests.rs @@ -307,9 +307,10 @@ async fn identity_equal_adoption_does_not_overwrite_an_owned_execution() { SandboxState::Started } ExecutionOwner::LifecycleTransition => { - assert_eq!( - crate::db::update_state(&conn, &id, SandboxState::Stopping).unwrap(), - crate::db::SandboxStateWriteOutcome::Applied + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &id, + &crate::db::edges::STOP_ADMISSION, ); SandboxState::Stopping } diff --git a/crates/libs/sbx/mgr/server/src/service/tests.rs b/crates/libs/sbx/mgr/server/src/service/tests.rs index 4eaa5f3cda..84b96924cb 100644 --- a/crates/libs/sbx/mgr/server/src/service/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/tests.rs @@ -1906,10 +1906,15 @@ pub(crate) mod service_tests { .as_ref(), ) .unwrap(); - assert!( - crate::db::update_state(&db, &sandbox_id_string, SandboxState::Suspended) - .unwrap() - .is_applied() + crate::db::tests::state_write_fixtures::apply_edge( + &db, + &sandbox_id_string, + &crate::db::edges::CREATE_COMPLETE, + ); + crate::db::tests::state_write_fixtures::apply_edge( + &db, + &sandbox_id_string, + &crate::db::edges::CREATE_BASE_SNAPSHOT_COMPLETE, ); assert!(crate::db::complete_create(&db, &sandbox_id_string, cleanup_id, true).unwrap()); assert!(crate::route_activation_claims::claim_exists(&db, &sandbox_id_string).unwrap()); @@ -5366,10 +5371,11 @@ pub(crate) mod service_tests { let conn = crate::db::test_conn(); let sandbox_id = SandboxId::new().to_string(); insert_test_sandbox(&conn, &sandbox_id, SandboxState::Starting); - crate::db::update_state(&conn, &sandbox_id, SandboxState::Deleting) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &sandbox_id, + &crate::db::edges::DELETE_ADMISSION, + ); crate::db::set_cleanup_intent( &conn, &sandbox_id, @@ -6695,6 +6701,23 @@ pub(crate) mod service_tests { insert_test_sandbox_with_runtime(conn, id, SandboxState::Started); } + /// Move a live `started` row to `restoring`, keeping its empty restore + /// columns. No production edge writes this. + fn restage_started_as_restoring(conn: &rusqlite::Connection, id: &str) { + assert_eq!( + crate::db::update_state_from_states( + conn, + id, + crate::db::StateSet::of(&[SandboxState::Started]), + SandboxState::Restoring, + ) + .unwrap(), + crate::db::StateWriteOutcome::Applied { + from: SandboxState::Started + } + ); + } + pub(crate) fn complete_runtime_endpoint_identity(conn: &rusqlite::Connection, id: &str) { conn.execute( "UPDATE sandboxes @@ -10228,10 +10251,11 @@ pub(crate) mod service_tests { .expect("delete must reach the cleanup claim"); { let db = service.db.lock().expect("mutex poisoned"); - crate::db::update_state(&db, &sandbox_id_string, SandboxState::Suspending) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state( + &db, + &sandbox_id_string, + SandboxState::Suspending, + ); } drop(cleanup_blocker); @@ -10701,10 +10725,11 @@ pub(crate) mod service_tests { let sandbox_id = SandboxId::new(); let sandbox_id_string = sandbox_id.to_string(); insert_test_sandbox(&conn, &sandbox_id_string, SandboxState::Started); - crate::db::update_state(&conn, &sandbox_id_string, SandboxState::Paused) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state( + &conn, + &sandbox_id_string, + SandboxState::Paused, + ); let service = SandboxManagerService::for_test(conn); let result = service @@ -10922,10 +10947,11 @@ pub(crate) mod service_tests { runtime_fields(ExecutionGeneration::new(2).unwrap(), 7), ) .unwrap(); - crate::db::update_state(&conn, &sandbox_id, SandboxState::Stopping) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &sandbox_id, + &crate::db::edges::STOP_ADMISSION, + ); crate::db::set_cleanup_intent( &conn, &sandbox_id, @@ -11048,10 +11074,11 @@ pub(crate) mod service_tests { runtime_fields(deleted_generation, 7), ) .unwrap(); - crate::db::update_state(conn, sandbox_id, SandboxState::Deleting) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + conn, + sandbox_id, + &crate::db::edges::DELETE_ADMISSION, + ); crate::db::set_cleanup_intent( conn, sandbox_id, @@ -11762,10 +11789,11 @@ pub(crate) mod service_tests { ExecutionGeneration::new(1).unwrap(), ) .unwrap(); - crate::db::update_state(&conn, &sandbox_id.to_string(), SandboxState::Paused) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &sandbox_id.to_string(), + &crate::db::edges::SUSPEND_ROLLBACK, + ); crate::db::update_runtime_fields( &conn, sandbox_id, @@ -14869,10 +14897,11 @@ pub(crate) mod service_tests { }; { let db = service.db.lock().expect("mutex poisoned"); - crate::db::update_state(&db, &sandbox_id, SandboxState::Paused) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &db, + &sandbox_id, + &crate::db::edges::SUSPEND_ROLLBACK, + ); } let cache_dir = tempfile::tempdir().unwrap(); @@ -16487,10 +16516,11 @@ pub(crate) mod service_tests { let service = SandboxManagerService::for_test(conn); { let db = service.db.lock().expect("mutex poisoned"); - crate::db::update_state(&db, &sandbox_id, SandboxState::Suspended) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::force_state( + &db, + &sandbox_id, + SandboxState::Suspended, + ); } let stale = seed_due_access_idle_sandbox_state(&service, &sandbox_id, &policy, 300); @@ -17123,10 +17153,7 @@ pub(crate) mod service_tests { let conn = crate::db::test_conn(); let sandbox_id = SandboxId::new().to_string(); insert_started_test_sandbox(&conn, &sandbox_id); - crate::db::update_state(&conn, &sandbox_id, SandboxState::Restoring) - .unwrap() - .require_applied() - .unwrap(); + restage_started_as_restoring(&conn, &sandbox_id); // The test service's node-manager refuses every dial, so the // retirement's cleanup cannot start and the row keeps the intent it // persisted. @@ -17814,10 +17841,11 @@ pub(crate) mod service_tests { let sandbox_id = SandboxId::new().to_string(); insert_test_sandbox(&conn, &sandbox_id, SandboxState::Suspended); register_full_suspend_artifact(&conn, &sandbox_id); - crate::db::update_state(&conn, &sandbox_id, SandboxState::Restoring) - .unwrap() - .require_applied() - .unwrap(); + crate::db::tests::state_write_fixtures::apply_edge( + &conn, + &sandbox_id, + &crate::db::edges::RESTORE_BEGIN, + ); let service = service_with_runtime_probe(conn, false); let _ = service @@ -18133,10 +18161,7 @@ pub(crate) mod service_tests { let complete_but_dead_id = Uuid::now_v7().to_string(); insert_started_test_sandbox(&conn, &complete_but_dead_id); - crate::db::update_state(&conn, &complete_but_dead_id, SandboxState::Restoring) - .unwrap() - .require_applied() - .unwrap(); + restage_started_as_restoring(&conn, &complete_but_dead_id); let complete_but_dead = crate::db::get_sandbox(&conn, &complete_but_dead_id) .unwrap() .expect("sandbox row"); diff --git a/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs b/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs index a45b1c3c02..ad3d51d6d9 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs @@ -41,16 +41,14 @@ fn lock_db(ctx: &CheckpointContext) -> Result Result<(), String> { - use crate::db::{ - SandboxStateWriteNotApplied, SandboxStateWriteOutcome as Legacy, StateWriteOutcome, - StateWriteRefusal, - }; + use crate::db::{SandboxStateWriteNotApplied, StateWriteOutcome, StateWriteRefusal}; let edge = &crate::db::edges::CHECKPOINT_BEGIN; - let outcome = match crate::db::update_state_from(db, sandbox_id, edge) + let not_applied = match crate::db::update_state_from(db, sandbox_id, edge) .map_err(|error| format!("db error: {error}"))? { StateWriteOutcome::Applied { .. } => return Ok(()), @@ -63,16 +61,12 @@ fn begin_checkpoint_state(db: &rusqlite::Connection, sandbox_id: &str) -> Result )); } StateWriteOutcome::Refused { - refusal: StateWriteRefusal::Quarantined, - .. - } => Legacy::BlockedByQuarantine, - StateWriteOutcome::Refused { - refusal: StateWriteRefusal::CleanupOwned, + refusal: StateWriteRefusal::Guard(guard), .. - } => Legacy::SupersededByCleanupEpisode, - StateWriteOutcome::Missing => Legacy::Missing, + } => SandboxStateWriteNotApplied::Guard(guard), + StateWriteOutcome::Missing => SandboxStateWriteNotApplied::Missing, }; - Err(SandboxStateWriteNotApplied { outcome }.to_string()) + Err(not_applied.to_string()) } pub(crate) fn write_file_atomic( diff --git a/crates/libs/sbx/mgr/server/src/snapshot_scheduler/tests.rs b/crates/libs/sbx/mgr/server/src/snapshot_scheduler/tests.rs index 29d9261d16..7f1c697332 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_scheduler/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_scheduler/tests.rs @@ -202,6 +202,19 @@ fn checkpoint_admission_writes_only_over_a_started_or_paused_row() { } } +/// A missing row keeps the text the retired generic writer's outcome rendered. +#[test] +fn checkpoint_admission_of_a_missing_row_reports_it() { + let db = crate::db::test_conn(); + let before = crate::db::tests::database_rows(&db); + + assert_eq!( + begin_checkpoint_state(&db, &SandboxId::new().to_string()).unwrap_err(), + "sandbox state write was not applied: missing" + ); + assert_eq!(crate::db::tests::database_rows(&db), before); +} + #[tokio::test] async fn skipped_checkpoint_never_publishes_an_artifact() { let cases = SandboxState::all()