From 8646f74b21e9fa865653d0aad5c159c2b9ed8ade Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Tue, 13 Oct 2026 18:26:16 +0000 Subject: [PATCH] sbx-mgr: add the persisted state-transition table and update_state_from (no callers) Add db/state_write.rs for the node-local SQLite sandboxes.state column: - StateSet, a bitset over the 18 lifecycle states. - EDGES: 73 data-only edge constants (from-set, target, kind) built from every production writer at this head, each naming its writer functions. Kinds are Request, Completion, Compensation, Recovery and KnownRace (the stale or unvalidated writes generic update_state callers can make). SandboxState::can_transition_to stays the RPC-admission machine. - update_state_from(conn, id, &'static Edge) -> Applied { from } | Refused { found, refusal } | Missing. Call sites name a table edge; KnownRace and insert edges are refused. It applies update_state's quarantine and cleanup-ownership guards, now generic_state_write_refusal in this module, then writes with the edge's from-set predicate in the UPDATE itself. A missed write from a source state is an invariant error. It runs in a caller's transaction or opens its own IMMEDIATE one in autocommit, never a nested one. It has no production callers yet. update_state calls the moved guard; its behavior is unchanged, and state_transitions.rs shrinks below its archcheck cap, which is lowered. Tests: - With every state expected, the writer matches update_state on every cell of the existing outcome matrix and its whole-database diffs, standalone and inside a transaction, by running the matrix's own grids and sweeps against both writers. - A from-set without the row's state refuses with every row unchanged, for every (from, to) pair, with accepting controls; every writable edge writes only from its own sources; refused edge kinds change nothing. - The invariant error and the autocommit atomicity of a deleting write and its teardown-job failure are pinned with temporary triggers. - Every disagreement between the edge table and can_transition_to (85 pairs) is in a checked-in allowlist with a reason, so a new one fails. The sbx-mgr AGENTS.md writer rule now points new transitions at update_state_from with an edge, or a specialized compare-and-set writer. --- architecture-baseline.toml | 2 +- crates/libs/sbx/mgr/server/AGENTS.md | 13 +- crates/libs/sbx/mgr/server/src/db.rs | 2 + .../mgr/server/src/db/state_transitions.rs | 60 +- .../libs/sbx/mgr/server/src/db/state_write.rs | 660 ++++++++++++++++++ crates/libs/sbx/mgr/server/src/db/tests.rs | 2 + .../server/src/db/tests/state_write_table.rs | 205 ++++++ .../server/src/db/tests/update_state_from.rs | 225 ++++++ .../src/db/tests/update_state_matrix.rs | 204 ++++-- 9 files changed, 1271 insertions(+), 102 deletions(-) create mode 100644 crates/libs/sbx/mgr/server/src/db/state_write.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/update_state_from.rs diff --git a/architecture-baseline.toml b/architecture-baseline.toml index c013e6a03d..6325c7c3a6 100644 --- a/architecture-baseline.toml +++ b/architecture-baseline.toml @@ -78,7 +78,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" = 1156 +"crates/libs/sbx/mgr/server/src/db/state_transitions.rs" = 1140 "crates/libs/sbx/mgr/server/src/run.rs" = 1101 "crates/libs/sbx/mgr/server/src/sandbox_eviction.rs" = 1273 "crates/libs/sbx/mgr/server/src/service/activation_support.rs" = 1592 diff --git a/crates/libs/sbx/mgr/server/AGENTS.md b/crates/libs/sbx/mgr/server/AGENTS.md index abb6e1b600..7b373b449b 100644 --- a/crates/libs/sbx/mgr/server/AGENTS.md +++ b/crates/libs/sbx/mgr/server/AGENTS.md @@ -2,10 +2,15 @@ ## Changes -- Write `sandboxes.state` only through `src/db` writers. A new transition uses - or adds a compare-and-set writer in `src/db`; no generic writer compares - against the caller's observed state. A refused write must leave every row - unchanged. +- 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 + `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. - Foreign keys are off: register a new sandbox child table in `delete_sandbox_child_rows` (`src/db.rs`). - `src/service/adapter.rs` implements the gRPC trait; most handler bodies live diff --git a/crates/libs/sbx/mgr/server/src/db.rs b/crates/libs/sbx/mgr/server/src/db.rs index 181c97c080..d05965bf95 100644 --- a/crates/libs/sbx/mgr/server/src/db.rs +++ b/crates/libs/sbx/mgr/server/src/db.rs @@ -75,6 +75,7 @@ mod restore_transfer_scopes; mod sandboxes; mod startup_recovery; mod state_transitions; +mod state_write; mod teardown_suspend; pub use auto_lifecycle::*; pub use cleanup_episodes::*; @@ -88,6 +89,7 @@ pub use restore_transfer_scopes::*; pub use sandboxes::*; pub use startup_recovery::*; pub use state_transitions::*; +pub use state_write::*; pub use teardown_suspend::*; // Most of the no-progress vocabulary is durable: its rendered form is persisted 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 3447a80602..0c5fec5a9f 100644 --- a/crates/libs/sbx/mgr/server/src/db/state_transitions.rs +++ b/crates/libs/sbx/mgr/server/src/db/state_transitions.rs @@ -110,47 +110,25 @@ pub fn update_state( let previous_state_value = previous_state .as_deref() .and_then(|state| state.parse::().ok()); - if generic_writer_transition_blocked(previous_state_value, state) { - tracing::warn!( - sandbox_id = %id, - target_state = %state, - "ignored generic sandbox state update because sandbox is quarantined" - ); - return Ok(SandboxStateWriteOutcome::BlockedByQuarantine); - } - if cleanup_episode(conn, id)? - .is_some_and(|episode| !episode.completion.deletes_row() || episode.requested_at.is_some()) - || (matches!( - previous_state_value, - Some( - SandboxState::Deleting - | SandboxState::CleanupRequired - | SandboxState::CleanupFailed - ) - ) && state != SandboxState::Deleting - && cleanup_intent_current(conn, id, CleanupCompletion::DeleteRow)?) - || (matches!( - previous_state_value, - Some( - SandboxState::Stopping - | SandboxState::CleanupRequired - | SandboxState::CleanupFailed - ) - ) && matches!( - state, - SandboxState::Started - | SandboxState::Paused - | SandboxState::Pausing - | SandboxState::Resuming - | SandboxState::Checkpointing - )) - { - tracing::warn!( - sandbox_id = %id, - target_state = %state, - "ignored generic sandbox state update because cleanup owns the transition" - ); - return Ok(SandboxStateWriteOutcome::SupersededByCleanupEpisode); + 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( diff --git a/crates/libs/sbx/mgr/server/src/db/state_write.rs b/crates/libs/sbx/mgr/server/src/db/state_write.rs new file mode 100644 index 0000000000..debafe6a2f --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/state_write.rs @@ -0,0 +1,660 @@ +//! 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. +//! +//! `SandboxState::can_transition_to` is the RPC-admission machine. [`EDGES`] +//! is a separate inventory of the transitions the database writers implement: +//! one entry per writer purpose, naming the writer functions and their +//! callers. Constructors are private, and `update_state_from` takes a +//! `&'static Edge`, so its call sites cannot invent one. Nothing checks the +//! table against the specialized writers. +//! +//! 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, and +//! [`EdgeKind::KnownRace`] edges list the other states it can overwrite. +//! - 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. +//! - Deleting the row is not a state write and has no edge. +//! - Writers with only test callers are excluded. + +use super::*; + +/// A set of lifecycle states, as a bitset in `SandboxState::all()` order. +#[derive(Clone, Copy, Eq, PartialEq)] +pub struct StateSet(u32); + +impl StateSet { + pub const EMPTY: Self = Self(0); + pub const ALL: Self = Self((1 << 18) - 1); + + pub const fn of(states: &[SandboxState]) -> Self { + let mut set = Self::EMPTY; + let mut rest = states; + while let [state, tail @ ..] = rest { + set = set.with(*state); + rest = tail; + } + set + } + + pub const fn with(self, state: SandboxState) -> Self { + Self(self.0 | bit(state)) + } + + pub const fn without(self, state: SandboxState) -> Self { + Self(self.0 & !bit(state)) + } + + pub const fn contains(self, state: SandboxState) -> bool { + self.0 & bit(state) != 0 + } + + pub const fn is_empty(self) -> bool { + self.0 == 0 + } + + /// The members in `SandboxState::all()` order. + pub fn iter(self) -> impl Iterator { + SandboxState::all() + .into_iter() + .filter(move |state| self.contains(*state)) + } +} + +impl fmt::Debug for StateSet { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_set() + .entries(self.iter().map(|state| state.as_str())) + .finish() + } +} + +/// Wildcard-free, so a new lifecycle state does not compile until it has a +/// bit, and `StateSet::ALL` must then grow with it. +const fn bit(state: SandboxState) -> u32 { + 1 << match state { + SandboxState::Creating => 0, + SandboxState::Created => 1, + SandboxState::Starting => 2, + SandboxState::Started => 3, + SandboxState::Stopping => 4, + SandboxState::Stopped => 5, + SandboxState::Deleting => 6, + SandboxState::Pausing => 7, + SandboxState::Paused => 8, + SandboxState::Resuming => 9, + SandboxState::Checkpointing => 10, + SandboxState::Suspending => 11, + SandboxState::Suspended => 12, + SandboxState::Restoring => 13, + SandboxState::CleanupRequired => 14, + SandboxState::CleanupFailed => 15, + SandboxState::Quarantined => 16, + SandboxState::Failed => 17, + } +} + +/// Why a writer implements an edge. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum EdgeKind { + /// Admission of an operation: a client RPC, or the node's own scheduler, + /// auto-lifecycle or eviction acting as one. + Request, + /// An in-flight operation finished. + Completion, + /// Rollback to the operation's entry state after a failure. + Compensation, + /// Startup, inventory, quarantine or cleanup repair. + Recovery, + /// A stale or unvalidated write a generic `update_state` caller can make: + /// a concurrent transition moved the row after the caller read it, or the + /// caller never checked the state. Not intended, and `update_state_from` + /// refuses to take one: converting the caller removes the edge. + KnownRace, +} + +/// One family of persisted transitions: every state in `from` to `to`. +#[derive(Debug)] +pub struct Edge { + name: &'static str, + from: StateSet, + to: SandboxState, + kind: EdgeKind, +} + +impl Edge { + pub const fn name(&self) -> &'static str { + self.name + } + + /// Empty for an insert. + pub const fn from(&self) -> StateSet { + self.from + } + + pub const fn to(&self) -> SandboxState { + self.to + } + + pub const fn kind(&self) -> EdgeKind { + self.kind + } +} + +macro_rules! edge_table { + ($($(#[doc = $doc:literal])+ $name:ident: $kind:ident, $from:expr => $to:ident;)+) => { + /// The persisted-edge constants. Each names its writer functions. + pub mod edges { + use super::{Edge, EdgeKind, StateSet}; + use crate::state::SandboxState::*; + + const fn of(states: &[crate::state::SandboxState]) -> StateSet { + StateSet::of(states) + } + + const INSERT: StateSet = StateSet::EMPTY; + + $( + $(#[doc = $doc])+ + pub const $name: Edge = Edge { + name: stringify!($name), + from: $from, + to: $to, + kind: EdgeKind::$kind, + }; + )+ + + /// Every persisted edge, in declaration order. + pub const EDGES: &[Edge] = &[$($name),+]; + } + }; +} + +edge_table! { + // Create, pool clone and import. + + /// `insert_sandbox_with_cleanup_episode_and_route_evidence` (create, + /// pool clone) and `transfer::reserve_import` insert a `creating` row with + /// an armed cleanup episode. + CREATE_INSERT: Request, INSERT => Creating; + /// `create_with_id`: `update_state(Created)`. + CREATE_COMPLETE: Completion, of(&[Creating]) => Created; + /// `create_with_id` in base-snapshot mode: `update_state(Suspended)` + /// after its own `created` write. + CREATE_BASE_SNAPSHOT_COMPLETE: Completion, of(&[Created]) => Suspended; + /// `complete_pool_clone` and `complete_import`, under their armed + /// episodes. + CLONE_OR_IMPORT_COMPLETE_SUSPENDED: Completion, of(&[Creating]) => Suspended; + /// `complete_import` of a stopped export. + IMPORT_COMPLETE_STOPPED: Completion, of(&[Creating]) => Stopped; + + // Cold start and restore activation. + + /// `begin_activation_with_route_evidence(ColdStart)` from + /// `start_sandbox_with_authority`. + COLD_START_BEGIN: Request, of(&[Created, Stopped, Suspended]) => Starting; + /// `begin_activation_with_route_evidence(Restore)` from + /// `start_from_checkpoint`. + RESTORE_BEGIN: Request, of(&[Suspended]) => Restoring; + /// `try_start_from_warm_snapshot`: `update_state(Suspended)` when the row + /// is not already `suspended`, before the cold path. + WARM_START_RESTAGE: Request, of(&[Created, Stopped]) => Suspended; + /// `insert_restoring_sandbox_row_from_snapshot` through + /// `restore_adopt_orphan_or_insert_sandbox_in_transaction`: a + /// cross-node restore for a sandbox this node has no row for. + CROSS_NODE_RESTORE_INSERT: Request, INSERT => Restoring; + /// `mark_start_succeeded`. + START_COMPLETE: Completion, of(&[Starting]) => Started; + /// `mark_restore_succeeded`. + RESTORE_COMPLETE_STARTED: Completion, of(&[Restoring]) => Started; + /// `mark_restore_succeeded` with a paused target. + RESTORE_COMPLETE_PAUSED: Completion, of(&[Restoring]) => Paused; + /// `publish_restored_suspended_artifact` and + /// `publish_restored_suspended_without_artifact`: cross-node restore + /// staging published a suspended row. + CROSS_NODE_RESTORE_PUBLISHED: Completion, of(&[Restoring]) => Suspended; + /// `publish_restored_disk_as_new`: a restore-as-new published a stopped + /// disk. + RESTORED_DISK_AS_NEW: Completion, of(&[Restoring]) => Stopped; + /// `rollback_starting_episode` to a `created` entry state. + START_ROLLBACK_CREATED: Compensation, of(&[Starting]) => Created; + /// `rollback_starting_episode` to a `stopped` entry state. + START_ROLLBACK_STOPPED: Compensation, of(&[Starting]) => Stopped; + /// `rollback_starting_episode` to a `suspended` entry state. + START_ROLLBACK_SUSPENDED: Compensation, of(&[Starting]) => Suspended; + /// `rollback_restoring_episode` from `start_from_checkpoint`. + RESTORE_ROLLBACK: Compensation, of(&[Restoring]) => Suspended; + + // Pause and resident resume. + + /// `pause` and the suspend auto-pause: `update_state(Pausing)` after + /// `transition(Pausing)` on the read. + PAUSE_BEGIN: Request, of(&[Started]) => Pausing; + /// `pause` and the suspend auto-pause: `update_state(Paused)`. + PAUSE_COMPLETE: Completion, of(&[Pausing]) => Paused; + /// `update_state(Started)` after a failed guest preparation in `pause` + /// and the suspend auto-pause; `update_state_best_effort(Started)` after a + /// node connect failure and in `pause_with_reconciliation`. + PAUSE_ROLLBACK: Compensation, of(&[Pausing]) => Started; + /// `begin_resident_resume_with_policy`, and `resume_after_wake_lock`'s + /// `update_state(Resuming)` without an activation policy. + RESUME_BEGIN: Request, of(&[Paused]) => Resuming; + /// `restore_resident_resume_state(Started)` through + /// `rollback_activation_state`. + RESUME_COMPLETE: Completion, of(&[Resuming]) => Started; + /// `restore_resident_resume_state(Paused)` after a failed resume. + RESUME_ROLLBACK: Compensation, of(&[Resuming]) => Paused; + + // Checkpoints and template base snapshots. + + /// `do_checkpoint` and `create_template_base_snapshot` (from `paused`): + /// `update_state(Checkpointing)` checked and written under one guard. + CHECKPOINT_BEGIN: Request, of(&[Started, Paused]) => Checkpointing; + /// `restore_checkpoint_state` from `finish_checkpoint_success`. + CHECKPOINT_COMPLETE_STARTED: Completion, of(&[Checkpointing]) => Started; + /// `restore_checkpoint_state` from `finish_checkpoint_success`, and + /// `create_template_base_snapshot`'s `update_state(Paused)`. + CHECKPOINT_COMPLETE_PAUSED: Completion, of(&[Checkpointing]) => Paused; + /// `restore_checkpoint_state_best_effort` after a failed checkpoint. + CHECKPOINT_ROLLBACK_STARTED: Compensation, of(&[Checkpointing]) => Started; + /// `restore_checkpoint_state_best_effort`, and + /// `create_template_base_snapshot`'s `update_state_best_effort(Paused)`. + CHECKPOINT_ROLLBACK_PAUSED: Compensation, of(&[Checkpointing]) => Paused; + + // Suspend. + + /// `begin_suspend` and `begin_teardown_suspend`. + SUSPEND_BEGIN: Request, of(&[Paused]) => Suspending; + /// `complete_row_intent_cleanup_for_job(MarkSuspended)` with a published + /// suspend artifact. + SUSPEND_COMPLETE: Completion, of(&[Suspending]) => Suspended; + /// `rollback_suspending_to_paused` from the suspend snapshot path. + SUSPEND_ROLLBACK: Compensation, of(&[Suspending]) => Paused; + /// The full process-snapshot rollback in `grpc_spaced_process_snapshots`: + /// `update_state(Suspended)` after its own stop, so `start_from_checkpoint` + /// can restore the snapshot. + PROCESS_SNAPSHOT_ROLLBACK_RESTAGE: Completion, of(&[Stopped]) => Suspended; + + // 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`. + STOP_ADMISSION: Request, + StateSet::ALL + .without(Creating) + .without(Stopping) + .without(Deleting) + .without(Quarantined) => Stopping; + /// `complete_row_intent_cleanup_for_job(MarkStopped)` and + /// `publish_stopped_artifact`. + STOP_COMPLETE: Completion, of(&[Stopping]) => Stopped; + /// `run_due_auto_lifecycle`'s idle Stop of a `suspended` row: + /// `update_state(Stopped)`, 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. + DELETE_ADMISSION: Request, StateSet::ALL.without(Deleting) => Deleting; + /// The eviction claims: `claim_stopped_sandbox_for_eviction`, + /// `claim_suspended_sandbox_for_eviction`, + /// `claim_failed_restoring_sandbox_for_eviction`, + /// `claim_teardown_suspend_failed_sandbox_for_eviction` and + /// `claim_quarantined_sandbox_for_eviction`. + EVICTION_CLAIM: Request, + of(&[Stopped, Suspended, Restoring, Suspending, Quarantined]) => Deleting; + + // Cleanup hand-off and completion. + + /// `request_cleanup_episode`: the activation cleanup guard + /// (`schedule_activation_cleanup`), startup recovery and cleanup recovery + /// hand an armed episode to cleanup. Create's guard can fire after its own + /// `created` write, or after the base-snapshot `suspended` write when + /// `complete_create` fails. + ACTIVATION_CLEANUP_REQUEST: Recovery, + of(&[Creating, Created, Starting, Suspended, Restoring]) => CleanupRequired; + /// `mark_cleanup_required` from startup recovery (interrupted + /// `starting`, `stopping`, `deleting`, `suspending` and `restoring` rows), + /// cleanup recovery, and a restore that finds a pending `mark_suspended` + /// intent; `mark_cleanup_required_if_no_intent` for an ambiguous restore + /// retry; `mark_cleanup_required_if_no_intent_in_state` from cleanup + /// recovery. + CLEANUP_HANDOFF: Recovery, + of(&[Starting, Stopping, Deleting, Suspending, Restoring, CleanupFailed]) + => CleanupRequired; + /// `pin_suspended_artifact_for_cleanup`: startup suspend recovery pins a + /// staged artifact and hands the row to cleanup. + SUSPEND_ARTIFACT_CLEANUP: Recovery, of(&[Suspending]) => CleanupRequired; + /// `mark_cleanup_failed` (from `mark_async_cleanup_failed`, for rows whose + /// intent the completion owns) and `record_cleanup_episode_failure`. + CLEANUP_FAILURE: Recovery, + of(&[Stopping, Suspending, Deleting, CleanupRequired]) => CleanupFailed; + /// `complete_cleanup_episode_for_job` and + /// `complete_row_intent_cleanup_for_job` through `apply_cleanup_state` + /// for a `mark_created` completion. + CLEANUP_COMPLETE_CREATED: Recovery, of(&[CleanupRequired, CleanupFailed]) => Created; + /// As above for `mark_stopped`, and `publish_stopped_artifact` for a + /// cleanup row with a `mark_stopped` intent. + CLEANUP_COMPLETE_STOPPED: Recovery, of(&[CleanupRequired, CleanupFailed]) => Stopped; + /// As above for `mark_suspended`. + CLEANUP_COMPLETE_SUSPENDED: Recovery, of(&[CleanupRequired, CleanupFailed]) => Suspended; + /// `quarantine_cleanup_episode_for_job`, and `apply_cleanup_state` for a + /// `quarantine` completion or a `mark_suspended` completion without a + /// suspend artifact. + CLEANUP_QUARANTINE: Recovery, of(&[CleanupRequired, CleanupFailed]) => Quarantined; + + // Quarantine entry and exit. + + /// `mark_quarantined_best_effort` and `mark_quarantined_from_state`: an + /// unconfirmed pause (`pause_with_reconciliation`, startup pause + /// recovery), a failed checkpoint, a failed process-snapshot stop, and a + /// failed resident resume. + IN_FLIGHT_QUARANTINE: Recovery, + of(&[Pausing, Stopping, Resuming, Checkpointing]) => Quarantined; + /// `mark_suspending_quarantined` (suspend snapshot path and startup + /// suspend recovery), and + /// `complete_row_intent_cleanup_for_job(MarkSuspended)` without a suspend + /// artifact. + SUSPEND_QUARANTINE: Recovery, of(&[Suspending]) => Quarantined; + /// `settle_startup_resident(Quarantined)`: a `started` row with no live + /// VM, and resident rows whose VM contradicts or cannot prove the state. + STARTUP_RESIDENT_QUARANTINE: Recovery, + of(&[Started, Paused, Resuming, Checkpointing]) => Quarantined; + /// `insert_quarantined_sandbox_in_transaction` from orphan + /// materialization. + ORPHAN_QUARANTINE_INSERT: Recovery, INSERT => Quarantined; + /// `recover_pause_failed_quarantine` from the watchdog. + PAUSE_FAILED_QUARANTINE_RECOVERY_STARTED: Recovery, of(&[Quarantined]) => Started; + /// `recover_pause_failed_quarantine` with a paused VM. + PAUSE_FAILED_QUARANTINE_RECOVERY_PAUSED: Recovery, of(&[Quarantined]) => Paused; + /// `adopt_quarantine_cleanup_episode`: quarantine recovery hands a pinned + /// suspend artifact to cleanup. + QUARANTINE_CLEANUP_ADOPTION: Recovery, of(&[Quarantined]) => CleanupRequired; + /// `rearm_prepared_clone_disposal`: Delete of a quarantined pool clone + /// re-requests its disposal episode instead of writing `deleting`. + POOL_CLONE_DISPOSAL_REARM: Request, of(&[Quarantined]) => CleanupRequired; + /// `restore_adopt_orphan_or_insert_sandbox_in_transaction`'s conflict + /// branch: orphan materialization or a cross-node restore insert adopts + /// the synthetic orphan row an earlier pass quarantined. + ORPHAN_RESTORE_ADOPTION: Recovery, of(&[Quarantined]) => Restoring; + + // Startup and inventory repair. + + /// `recover_interrupted_create`: `update_state(Created)`. + INTERRUPTED_CREATE_RECOVERY: Recovery, of(&[Creating]) => Created; + /// `recover_unlaunched_start`: `update_state(Stopped)` when a rootfs + /// exists. + UNLAUNCHED_START_RECOVERY_STOPPED: Recovery, of(&[Starting]) => Stopped; + /// `recover_unlaunched_start`: `update_state(Created)` otherwise. + UNLAUNCHED_START_RECOVERY_CREATED: Recovery, of(&[Starting]) => Created; + /// `recover_interrupted_pause_state` from startup pause recovery. + INTERRUPTED_PAUSE_RECOVERY_STARTED: Recovery, of(&[Pausing]) => Started; + /// `recover_interrupted_pause_state` with a paused VM. + INTERRUPTED_PAUSE_RECOVERY_PAUSED: Recovery, of(&[Pausing]) => Paused; + /// `settle_startup_resident(Started)`: the VM runs. + STARTUP_RESIDENT_STARTED: Recovery, of(&[Resuming, Checkpointing]) => Started; + /// `settle_startup_resident(Paused)`: an interrupted resume left the VM + /// paused. + STARTUP_RESIDENT_PAUSED: Recovery, of(&[Resuming]) => Paused; + /// `rollback_suspending_to_paused` from startup suspend recovery when the + /// VM is live without a usable artifact. + STARTUP_SUSPEND_ROLLBACK: Recovery, of(&[Suspending]) => Paused; + /// `recover_interrupted_startup_states`' `update_state(Suspended)`, and + /// `publish_orphan_quarantine_recovered_artifact`. + INTERRUPTED_RESTORE_RECOVERY: Recovery, of(&[Restoring]) => Suspended; + /// Orphan materialization: `insert_sandbox` of a `cleanup_required` row, + /// which `mark_cleanup_required` then gives its intent. + ORPHAN_CLEANUP_INSERT: Recovery, INSERT => CleanupRequired; + /// `insert_recovered_suspended_sandbox_in_transaction` from orphan + /// materialization. + ORPHAN_SUSPENDED_INSERT: Recovery, INSERT => Suspended; + /// `restore_adopt_orphan_or_insert_sandbox_in_transaction`'s insert + /// branch from orphan materialization of an interrupted restore. + ORPHAN_RESTORE_INSERT: Recovery, INSERT => Restoring; + + // Known races. + + /// Pause rollback (`update_state(Started)`, best effort or not) after a + /// concurrent Stop settled `stopped`. + PAUSE_ROLLBACK_AFTER_STOP: KnownRace, of(&[Stopped]) => Started; + /// Pause completion (`update_state(Paused)`) after a concurrent Stop. + PAUSE_COMPLETE_AFTER_STOP: KnownRace, of(&[Stopped]) => Paused; + /// `resume_after_wake_lock`'s policy-free `update_state(Resuming)` after + /// a Stop completed during the pre-wake policy refresh. + RESUME_BEGIN_AFTER_STOP: KnownRace, of(&[Stopped]) => Resuming; + /// `try_start_from_warm_snapshot`'s `update_state(Suspended)`. An + /// empty-checkpoint warm Start rejects only `failed` before it, so it + /// writes from every state the generic guards admit, with no concurrency + /// needed; a row with a current activation episode answers + /// `CleanupOwned` instead. + WARM_START_UNVALIDATED: KnownRace, + StateSet::ALL + .without(Created) + .without(Stopped) + .without(Suspended) + .without(Deleting) + .without(Quarantined) + .without(Failed) => Suspended; + /// The full process-snapshot rollback's `update_state(Suspended)` after a + /// Stop admitted while it restaged the snapshot. Its own stop leaves no + /// runtime, so the second Stop is reached only in the window between + /// Stop's runtime gate and the cleanup fence, or with a leaked runtime. + PROCESS_SNAPSHOT_ROLLBACK_AFTER_STOP: KnownRace, of(&[Stopping]) => Suspended; + /// `create_template_base_snapshot`'s `update_state(Paused)` after a Stop + /// completed during capture. + TEMPLATE_SNAPSHOT_COMPLETE_AFTER_STOP: KnownRace, of(&[Stopped]) => Paused; +} + +pub use edges::EDGES; + +/// 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, +} + +/// 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`. +pub(super) fn generic_state_write_refusal( + conn: &Connection, + id: &str, + previous: Option, + target: SandboxState, +) -> Result> { + if generic_writer_transition_blocked(previous, target) { + return Ok(Some(StateWriteRefusal::Quarantined)); + } + let cleanup_owned = cleanup_episode(conn, id)? + .is_some_and(|episode| !episode.completion.deletes_row() || episode.requested_at.is_some()) + || (matches!( + previous, + Some( + SandboxState::Deleting + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + ) + ) && target != SandboxState::Deleting + && cleanup_intent_current(conn, id, CleanupCompletion::DeleteRow)?) + || (matches!( + previous, + Some( + SandboxState::Stopping + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + ) + ) && matches!( + target, + SandboxState::Started + | SandboxState::Paused + | SandboxState::Pausing + | SandboxState::Resuming + | SandboxState::Checkpointing + )); + Ok(cleanup_owned.then_some(StateWriteRefusal::CleanupOwned)) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[must_use = "callers must distinguish an applied state write from a refused one"] +pub enum StateWriteOutcome { + Applied { + from: SandboxState, + }, + Refused { + found: SandboxState, + refusal: StateWriteRefusal, + }, + Missing, +} + +/// 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. +/// +/// 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. +/// +/// Inserts and [`EdgeKind::KnownRace`] edges, and targets `update_state` +/// rejects (`quarantined`, `failed`), are `InvalidQuery`. A write that matches +/// no row although the row was read in a source state is +/// `StatementChangedRows(0)`. +pub fn update_state_from( + conn: &Connection, + id: &str, + edge: &'static Edge, +) -> Result { + if edge.kind == EdgeKind::KnownRace || edge.from.is_empty() { + return Err(rusqlite::Error::InvalidQuery); + } + write_state_from(conn, id, edge.from, edge.to) +} + +/// [`update_state_from`] over an arbitrary from-set, for tests that sweep +/// from-sets no edge has. +#[cfg(test)] +pub(crate) fn update_state_from_states( + conn: &Connection, + id: &str, + from: StateSet, + to: SandboxState, +) -> Result { + write_state_from(conn, id, from, to) +} + +fn write_state_from( + conn: &Connection, + id: &str, + from: StateSet, + to: SandboxState, +) -> Result { + if matches!(to, SandboxState::Quarantined | SandboxState::Failed) { + return Err(rusqlite::Error::InvalidQuery); + } + if conn.is_autocommit() { + let tx = immediate_transaction(conn)?; + let outcome = write_state_from_in_transaction(&tx, id, from, to)?; + tx.commit()?; + return Ok(outcome); + } + write_state_from_in_transaction(conn, id, from, to) +} + +fn write_state_from_in_transaction( + conn: &Connection, + id: &str, + from: StateSet, + to: SandboxState, +) -> Result { + let Some(found) = current_state_label(conn, id)? else { + return Ok(StateWriteOutcome::Missing); + }; + 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 }); + } + let from_labels = + serde_json::to_string(&from.iter().map(|state| state.as_str()).collect::>()) + .map_err(|error| rusqlite::Error::ToSqlConversionFailure(Box::new(error)))?; + 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 state IN (SELECT value FROM json_each(?4))", + params![to.as_str(), now()?, id, from_labels], + )?; + if changed == 0 { + // The row was read as `found` in this transaction, so only a source + // set that excludes it can miss. + if from.contains(found) { + return Err(rusqlite::Error::StatementChangedRows(0)); + } + return Ok(StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::FromState, + }); + } + if to == SandboxState::Deleting { + teardown_jobs::fail_active(conn, id, TeardownSuspendFailure::Deleting)?; + } + Ok(StateWriteOutcome::Applied { from: found }) +} diff --git a/crates/libs/sbx/mgr/server/src/db/tests.rs b/crates/libs/sbx/mgr/server/src/db/tests.rs index 104b764350..831eeae62d 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests.rs @@ -10025,5 +10025,7 @@ mod accepted_shapes_single_statement; mod cleanup_quarantine_refusal; mod local_network_policy_clear; pub(crate) mod state_write_fixtures; +mod state_write_table; mod thin_writer_coverage; +mod update_state_from; mod update_state_matrix; diff --git a/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs b/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs new file mode 100644 index 0000000000..70c85aaf4c --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs @@ -0,0 +1,205 @@ +//! The persisted-edge table of the node-local SQLite `sandboxes.state` against +//! `SandboxState::can_transition_to`, the RPC-admission machine. +//! +//! Every pair on which the two disagree is listed in [`ALLOWLIST`] with a +//! reason, so a new writer edge, a changed from-set, or a changed state +//! machine fails here until the disagreement is recorded. + +use std::collections::BTreeMap; + +use crate::db::{EDGES, EdgeKind}; +use crate::state::SandboxState::{self, *}; + +#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)] +enum Disagreement { + /// A request edge writes the pair; `can_transition_to` forbids it. + RequestForbidden, + /// Only completion, compensation, recovery or known-race edges write the + /// pair; `can_transition_to` forbids it. + WriteForbidden, + /// `can_transition_to` allows the pair, but only completion, + /// compensation, recovery or known-race edges write it. + AllowedNotRequested, + /// `can_transition_to` allows the pair; no writer implements it. + AllowedUnwritten, +} + +use Disagreement::*; + +/// Stop resolves its target only while the runtime is live; admission does not +/// re-check it, and Stop does not consult `can_transition_to`. +const STOP_LIVE_RUNTIME: &str = "Stop admits a live runtime in this transient state"; +const STOP_SETTLED_SOURCE: &str = + "Stop admission, only in the runtime gate-to-fence window or with a leaked runtime"; +const STOP_RUNTIME_OWNER: &str = "Stop admits a row in this state while its runtime is live"; +/// `can_transition_to` lists an operation's completion; only its completion +/// writer implements it. +const COMPLETION_ONLY: &str = "an in-flight operation's completion, not an admission"; +const RECOVERY_QUARANTINE: &str = "only recovery writers enter quarantine"; +const NO_QUARANTINE_WRITER: &str = "no production writer quarantines a row in this state"; +const CLEANUP_SETTLES_INTENT: &str = "cleanup completion settles the persisted intent's target"; + +#[rustfmt::skip] +const ALLOWLIST: &[(SandboxState, SandboxState, Disagreement, &str)] = &[ + (Creating, Created, AllowedNotRequested, COMPLETION_ONLY), + (Creating, Stopped, WriteForbidden, "import completion of a stopped export"), + (Creating, Deleting, RequestForbidden, "Delete admission supersedes the create episode"), + (Creating, Suspended, WriteForbidden, "pool clone and import completion; known race: unvalidated warm start"), + (Creating, CleanupRequired, WriteForbidden, "an interrupted create is handed to cleanup"), + (Creating, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (Created, Stopping, RequestForbidden, STOP_SETTLED_SOURCE), + (Created, Suspended, RequestForbidden, "warm start restages a created row as suspended"), + (Created, CleanupRequired, WriteForbidden, "create's cleanup guard fires after its created write"), + (Created, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (Starting, Created, WriteForbidden, "cold-start rollback and startup recovery restore the entry state"), + (Starting, Started, AllowedNotRequested, COMPLETION_ONLY), + (Starting, Stopping, RequestForbidden, STOP_LIVE_RUNTIME), + (Starting, Stopped, WriteForbidden, "cold-start rollback and startup recovery restore the entry state"), + (Starting, Suspended, WriteForbidden, "cold-start rollback restores a suspended entry state; known race: warm start"), + (Starting, CleanupRequired, WriteForbidden, "an interrupted cold start is handed to cleanup"), + (Starting, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (Started, Suspending, AllowedUnwritten, "suspend pauses a started row first"), + (Started, Suspended, WriteForbidden, "known race: unvalidated warm start"), + (Started, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Stopping, Stopped, AllowedNotRequested, COMPLETION_ONLY), + (Stopping, Suspended, WriteForbidden, "known races: unvalidated warm start; process-snapshot rollback in the gate-to-fence window"), + (Stopping, CleanupRequired, WriteForbidden, "an interrupted stop is handed to cleanup"), + (Stopping, CleanupFailed, WriteForbidden, "stop cleanup failed"), + (Stopping, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Stopped, Started, WriteForbidden, "known race: pause rollback after Stop"), + (Stopped, Stopping, RequestForbidden, STOP_SETTLED_SOURCE), + (Stopped, Paused, WriteForbidden, "known races: pause and template-snapshot completion after Stop"), + (Stopped, Resuming, WriteForbidden, "known race: policy-free resume after Stop"), + (Stopped, Suspended, RequestForbidden, "warm start and process-snapshot rollback restage a stopped row"), + (Stopped, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (Deleting, CleanupRequired, WriteForbidden, "an interrupted delete is handed to cleanup"), + (Deleting, CleanupFailed, WriteForbidden, "delete cleanup failed"), + (Deleting, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (Pausing, Started, WriteForbidden, "pause rollback and startup pause recovery"), + (Pausing, Stopping, RequestForbidden, STOP_LIVE_RUNTIME), + (Pausing, Paused, AllowedNotRequested, COMPLETION_ONLY), + (Pausing, Suspended, WriteForbidden, "known race: unvalidated warm start"), + (Pausing, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Paused, Suspended, WriteForbidden, "known race: unvalidated warm start"), + (Paused, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Resuming, Started, AllowedNotRequested, COMPLETION_ONLY), + (Resuming, Stopping, RequestForbidden, STOP_LIVE_RUNTIME), + (Resuming, Paused, WriteForbidden, "resume rollback and startup resident recovery"), + (Resuming, Suspended, WriteForbidden, "known race: unvalidated warm start"), + (Resuming, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Checkpointing, Started, AllowedNotRequested, COMPLETION_ONLY), + (Checkpointing, Stopping, RequestForbidden, STOP_LIVE_RUNTIME), + (Checkpointing, Paused, AllowedNotRequested, COMPLETION_ONLY), + (Checkpointing, Suspended, WriteForbidden, "known race: unvalidated warm start"), + (Checkpointing, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Suspending, Stopping, RequestForbidden, STOP_LIVE_RUNTIME), + (Suspending, Paused, WriteForbidden, "suspend rollback and startup suspend recovery"), + (Suspending, Suspended, AllowedNotRequested, COMPLETION_ONLY), + (Suspending, CleanupRequired, WriteForbidden, "an interrupted suspend is handed to cleanup"), + (Suspending, CleanupFailed, WriteForbidden, "suspend cleanup failed"), + (Suspending, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Suspended, Stopping, RequestForbidden, STOP_SETTLED_SOURCE), + (Suspended, Stopped, RequestForbidden, "auto-lifecycle idle Stop of a suspended row bypasses every RPC"), + (Suspended, CleanupRequired, WriteForbidden, "base-snapshot create's cleanup guard fires after its suspended write"), + (Suspended, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (Restoring, Started, AllowedNotRequested, COMPLETION_ONLY), + (Restoring, Stopping, RequestForbidden, STOP_LIVE_RUNTIME), + (Restoring, Stopped, WriteForbidden, "restore-as-new publishes a stopped disk"), + (Restoring, Paused, AllowedNotRequested, COMPLETION_ONLY), + (Restoring, Suspended, WriteForbidden, "restore rollback, cross-node publication, restore recovery; known race: warm start"), + (Restoring, CleanupRequired, WriteForbidden, "an interrupted or ambiguous restore is handed to cleanup"), + (Restoring, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), + (CleanupRequired, Created, WriteForbidden, CLEANUP_SETTLES_INTENT), + (CleanupRequired, Stopping, RequestForbidden, STOP_RUNTIME_OWNER), + (CleanupRequired, Stopped, WriteForbidden, CLEANUP_SETTLES_INTENT), + (CleanupRequired, Suspended, WriteForbidden, "cleanup completion settles a mark_suspended intent; known race: warm start"), + (CleanupRequired, CleanupFailed, AllowedNotRequested, "only cleanup records its own failure"), + (CleanupRequired, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (CleanupFailed, Created, WriteForbidden, CLEANUP_SETTLES_INTENT), + (CleanupFailed, Stopping, RequestForbidden, STOP_RUNTIME_OWNER), + (CleanupFailed, Stopped, WriteForbidden, CLEANUP_SETTLES_INTENT), + (CleanupFailed, Suspended, WriteForbidden, "cleanup completion settles a mark_suspended intent; known race: warm start"), + (CleanupFailed, CleanupRequired, AllowedNotRequested, "only cleanup recovery re-requests cleanup"), + (CleanupFailed, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), + (Quarantined, Started, WriteForbidden, "pause-failed quarantine recovery"), + (Quarantined, Paused, WriteForbidden, "pause-failed quarantine recovery"), + (Quarantined, Restoring, WriteForbidden, "a restore adopts a quarantined synthetic orphan"), + (Quarantined, CleanupRequired, RequestForbidden, "Delete of a quarantined pool clone re-arms its disposal"), + (Failed, Stopping, RequestForbidden, STOP_RUNTIME_OWNER), +]; + +fn position(state: SandboxState) -> usize { + SandboxState::all() + .into_iter() + .position(|candidate| candidate == state) + .unwrap() +} + +fn classify(from: SandboxState, to: SandboxState) -> Option { + let kinds: Vec = EDGES + .iter() + .filter(|edge| edge.to() == to && edge.from().contains(from)) + .map(|edge| edge.kind()) + .collect(); + let requested = kinds.contains(&EdgeKind::Request); + match (from.can_transition_to(to), requested, !kinds.is_empty()) { + (true, true, _) | (false, false, false) => None, + (true, false, true) => Some(AllowedNotRequested), + (true, false, false) => Some(AllowedUnwritten), + (false, true, _) => Some(RequestForbidden), + (false, false, true) => Some(WriteForbidden), + } +} + +#[test] +fn persisted_edges_disagree_with_the_admission_machine_only_as_allowlisted() { + let mut actual = BTreeMap::new(); + for from in SandboxState::all() { + for to in SandboxState::all() { + if from == to { + continue; + } + if let Some(disagreement) = classify(from, to) { + actual.insert((position(from), position(to)), (from, to, disagreement)); + } + } + } + let mut listed = BTreeMap::new(); + for &(from, to, disagreement, reason) in ALLOWLIST { + assert!(!reason.trim().is_empty(), "{from} -> {to} needs a reason"); + assert!( + listed + .insert((position(from), position(to)), (from, to, disagreement)) + .is_none(), + "{from} -> {to} is listed twice" + ); + } + let unlisted: Vec = actual + .iter() + .filter(|(key, entry)| listed.get(*key) != Some(*entry)) + .map(|(_, (from, to, disagreement))| format!("({from:?}, {to:?}, {disagreement:?}, \"\"),")) + .collect(); + let stale: Vec = listed + .iter() + .filter(|(key, entry)| actual.get(*key) != Some(*entry)) + .map(|(_, (from, to, disagreement))| format!("{from:?} -> {to:?} ({disagreement:?})")) + .collect(); + assert!( + unlisted.is_empty() && stale.is_empty(), + "the persisted-edge table and can_transition_to disagree outside the allowlist.\n\ + Unlisted disagreements:\n{}\nAllowlist entries that no longer hold:\n{}", + unlisted.join("\n"), + stale.join("\n"), + ); +} + +#[test] +fn no_edge_lists_its_own_target_as_a_source() { + for edge in EDGES { + assert!( + !edge.from().contains(edge.to()), + "{}: rewriting a row in its target state is not a transition", + edge.name() + ); + } +} 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 new file mode 100644 index 0000000000..68434a78f4 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/update_state_from.rs @@ -0,0 +1,225 @@ +//! `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. + +use super::state_write_fixtures::{assert_diff, assert_unchanged, snapshot, test_start}; +use super::update_state_matrix::{ + Expected, PLAIN, Receipt, Shape, Writer, applied_diff, cell, index, seed, + sweep_lifecycle_pairs_and_cleanup_shapes, sweep_quarantine_episode_and_missing_row, write_from, +}; +use crate::db::{ + EDGES, EdgeKind, StateSet, StateWriteOutcome, StateWriteRefusal, edges, test_conn, + update_state_from, +}; +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() { + for in_transaction in IN_TRANSACTION { + sweep_lifecycle_pairs_and_cleanup_shapes(Writer::UpdateStateFromAll { in_transaction }); + } +} + +#[test] +fn expecting_every_state_matches_update_state_over_a_quarantine_episode_and_a_missing_row() { + for in_transaction in IN_TRANSACTION { + sweep_quarantine_episode_and_missing_row(Writer::UpdateStateFromAll { in_transaction }); + } +} + +/// The from-sets each case passes for a row in `from`, and whether each one +/// contains it: two that exclude it and two accepting controls. +fn from_sets(from: SandboxState) -> [(&'static str, StateSet, bool); 4] { + [ + ("every other state", StateSet::ALL.without(from), false), + ("no state", StateSet::EMPTY, false), + ("only the row's state", StateSet::of(&[from]), true), + ("every state", StateSet::ALL, true), + ] +} + +/// What writing `to` over a plain row in `from` returns when the from-set +/// does or does not contain `from`, or `None` for a target the generic +/// writers reject before reading the row. The generic guards answer before +/// the expectation does, so their refusals do not depend on the from-set. +fn expected_outcome( + from: SandboxState, + to: SandboxState, + from_set_contains_row: bool, +) -> Option { + let refused = |refusal| StateWriteOutcome::Refused { + found: from, + refusal, + }; + 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::Rejected => None, + Expected::Missing => panic!("a seeded row cannot be missing"), + } +} + +#[test] +fn a_from_set_without_the_row_state_refuses_and_keeps_every_row() { + for in_transaction in IN_TRANSACTION { + for receipt_kind in Receipt::ALL { + for from in SandboxState::all() { + for to in SandboxState::all() { + let receipt = receipt_kind.target(to); + for (label, from_set, contains) in from_sets(from) { + let case = format!( + "in transaction: {in_transaction}, {receipt} receipt, \ + expecting {label}: {from} -> {to}" + ); + let conn = test_conn(); + let id = seed(&conn, Shape::Plain, from, receipt); + let before = snapshot(&conn); + let start = test_start(); + let result = write_from(&conn, &id, from_set, to, in_transaction); + match (expected_outcome(from, to, contains), result) { + (Some(expected), Ok(outcome)) => { + assert_eq!(outcome, expected, "{case}"); + if matches!(outcome, StateWriteOutcome::Applied { .. }) { + let applied = applied_diff(&id, from, to, receipt); + assert_diff(&case, &before, &snapshot(&conn), start, &applied); + } else { + assert_unchanged(&case, &before, &conn); + } + } + (None, Err(rusqlite::Error::InvalidQuery)) => { + assert_unchanged(&case, &before, &conn); + } + (expected, result) => { + panic!("{case}: expected {expected:?}, got {result:?}") + } + } + } + } + } + } + } +} + +fn writable(edge: &crate::db::Edge) -> bool { + edge.kind() != EdgeKind::KnownRace + && !edge.from().is_empty() + && !matches!(edge.to(), SandboxState::Quarantined | SandboxState::Failed) +} + +#[test] +fn an_edge_writes_its_target_only_from_its_sources() { + for edge in EDGES.iter().filter(|edge| writable(edge)) { + let to = edge.to(); + for from in SandboxState::all() { + let receipt = Receipt::ForTarget.target(to); + let case = format!("{}: {from} -> {to}", edge.name()); + let conn = test_conn(); + let id = seed(&conn, Shape::Plain, from, receipt); + let before = snapshot(&conn); + let start = test_start(); + let expected = expected_outcome(from, to, edge.from().contains(from)).unwrap(); + let outcome = update_state_from(&conn, &id, edge).unwrap(); + assert_eq!(outcome, expected, "{case}"); + if matches!(outcome, StateWriteOutcome::Applied { .. }) { + let applied = applied_diff(&id, from, to, receipt); + assert_diff(&case, &before, &snapshot(&conn), start, &applied); + } else { + assert_unchanged(&case, &before, &conn); + } + } + } +} + +/// Known races, inserts and quarantine or failure targets are refused before +/// the row is read. Known races are tried over a source the generic guards +/// admit; inserts and quarantine targets over `started`. +#[test] +fn known_race_insert_and_rejected_target_edges_are_refused() { + let refused: Vec<_> = EDGES.iter().filter(|edge| !writable(edge)).collect(); + assert!( + refused + .iter() + .any(|edge| edge.kind() == EdgeKind::KnownRace) + ); + assert!(refused.iter().any(|edge| edge.from().is_empty())); + for edge in refused { + let to = edge.to(); + let from = edge + .from() + .iter() + .find(|from| cell(PLAIN[index(*from)], to) == Expected::Applied) + .unwrap_or(SandboxState::Started); + let case = format!("{}: {from} -> {to}", edge.name()); + let conn = test_conn(); + let id = seed( + &conn, + Shape::Plain, + from, + Receipt::ForAnotherState.target(to), + ); + let before = snapshot(&conn); + match update_state_from(&conn, &id, edge) { + Err(rusqlite::Error::InvalidQuery) => assert_unchanged(&case, &before, &conn), + other => panic!("{case}: expected InvalidQuery, got {other:?}"), + } + } +} + +/// A write that matches no row although the row was read in a source state +/// is an invariant failure, not a refusal. A trigger that skips the update +/// stands in for SQL that drifted from the from-set. +#[test] +fn a_missed_write_from_a_source_state_is_an_invariant_error() { + let conn = test_conn(); + let id = seed( + &conn, + Shape::Plain, + SandboxState::Pausing, + SandboxState::Started, + ); + conn.execute_batch( + "CREATE TEMP TRIGGER skip_state_write BEFORE UPDATE OF state ON main.sandboxes + BEGIN SELECT RAISE(IGNORE); END;", + ) + .unwrap(); + let before = snapshot(&conn); + match update_state_from(&conn, &id, &edges::PAUSE_ROLLBACK) { + Err(rusqlite::Error::StatementChangedRows(0)) => {} + other => panic!("expected an invariant error, got {other:?}"), + } + assert_unchanged("skipped pause rollback", &before, &conn); +} + +/// Standalone, a `deleting` write and the teardown jobs it fails commit +/// together: when failing the job fails, the state write rolls back too. +#[test] +fn a_standalone_deletion_rolls_back_when_its_job_failure_does() { + let conn = test_conn(); + let id = seed( + &conn, + Shape::Plain, + SandboxState::Stopped, + SandboxState::Stopped, + ); + conn.execute_batch( + "CREATE TEMP TRIGGER refuse_job_update BEFORE UPDATE ON main.jobs + BEGIN SELECT RAISE(ABORT, 'job update refused'); END;", + ) + .unwrap(); + let before = snapshot(&conn); + assert!(conn.is_autocommit()); + match update_state_from(&conn, &id, &edges::DELETE_ADMISSION) { + Err(rusqlite::Error::SqliteFailure(_, Some(message))) => { + assert!(message.contains("job update refused"), "{message}"); + } + other => panic!("expected the job failure to fail the write, got {other:?}"), + } + assert_unchanged("deletion with a failing job update", &before, &conn); +} 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 a49d7940ad..d8e4d4e36d 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 @@ -7,6 +7,9 @@ //! 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. use pplx_space_common_core::types::{ SandboxCleanupEpisodeId, SandboxId, SandboxQuarantineReasonCode, @@ -20,14 +23,15 @@ use super::state_write_fixtures::{ }; use crate::db::{ CleanupCompletion, CleanupEpisodeRequestOutcome, CleanupEpisodeSpec, SandboxStateWriteOutcome, - begin_teardown_suspend, insert_cleanup_episode, mark_suspending_quarantined, - request_cleanup_episode, set_cleanup_intent, test_conn, update_state, + StateSet, 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, }; use crate::state::SandboxState; /// The grids' row and column order. Wildcard-free, so a new lifecycle state /// does not compile until every grid has a row and a column for it. -const fn index(state: SandboxState) -> usize { +pub(super) const fn index(state: SandboxState) -> usize { match state { SandboxState::Creating => 0, SandboxState::Created => 1, @@ -52,10 +56,10 @@ const fn index(state: SandboxState) -> usize { /// `A` applied, `Q` blocked by quarantine, `S` superseded by cleanup /// ownership, `X` rejected before reading the row (`InvalidQuery`). -type Grid = [&'static str; 18]; +pub(super) type Grid = [&'static str; 18]; /// No cleanup intent and no cleanup episode. -const PLAIN: Grid = [ +pub(super) const PLAIN: Grid = [ // to: cre crd stg std stp spd del pau psd res chk ssg ssd rst crq crf qua fai /* creating */ "A A A A A A A A A A A A A A A A X X", @@ -139,7 +143,7 @@ const QUARANTINE_EPISODE: &str = const MISSING: &str = "M M M M M M M M M M M M M M M M X X"; #[derive(Clone, Copy, Debug, Eq, PartialEq)] -enum Expected { +pub(super) enum Expected { Applied, Quarantine, Superseded, @@ -147,7 +151,7 @@ enum Expected { Rejected, } -fn cell(row: &str, to: SandboxState) -> Expected { +pub(super) fn cell(row: &str, to: SandboxState) -> Expected { let cells: Vec<&str> = row.split_whitespace().collect(); assert_eq!(cells.len(), 18, "grid row {row:?}"); match cells[index(to)] { @@ -161,7 +165,7 @@ fn cell(row: &str, to: SandboxState) -> Expected { } #[derive(Clone, Copy, Debug)] -enum Shape { +pub(super) enum Shape { Plain, DeleteRowIntent, ArmedCreateEpisode, @@ -239,7 +243,7 @@ impl Shape { /// Which transition receipt a seeded row carries relative to the target. #[derive(Clone, Copy, Debug)] -enum Receipt { +pub(super) enum Receipt { /// A receipt for `to` itself when the table admits one, which an applied /// write keeps; otherwise a `started` receipt, which it clears. ForTarget, @@ -249,9 +253,9 @@ enum Receipt { } impl Receipt { - const ALL: [Self; 2] = [Self::ForTarget, Self::ForAnotherState]; + pub(super) const ALL: [Self; 2] = [Self::ForTarget, Self::ForAnotherState]; - fn target(self, to: SandboxState) -> SandboxState { + pub(super) fn target(self, to: SandboxState) -> SandboxState { let receipt_capable = match to { SandboxState::Created | SandboxState::Started @@ -286,7 +290,12 @@ impl Receipt { /// a deletion fails), a failure reason, a transition receipt for `receipt`, a /// remote restore's columns when `from` is `restoring`, and backdated /// timestamps. -fn seed(conn: &Connection, shape: Shape, from: SandboxState, receipt: SandboxState) -> String { +pub(super) fn seed( + conn: &Connection, + shape: Shape, + from: SandboxState, + receipt: SandboxState, +) -> String { let id = insert_row(conn, SandboxState::Paused); begin_teardown_suspend(conn, &id, generation(7)).unwrap(); shape.install(conn, &id); @@ -297,9 +306,10 @@ fn seed(conn: &Connection, shape: Shape, from: SandboxState, receipt: SandboxSta id } -/// The row diff an applied `update_state(to)` makes to a [`seed`] row whose -/// receipt targets `receipt`. -fn applied_diff( +/// The row diff an applied generic write of `to` makes to a [`seed`] row whose +/// receipt targets `receipt`. The teardown job's failure is the deletion side +/// effect; every other change is to the sandbox row. +pub(super) fn applied_diff( id: &str, from: SandboxState, to: SandboxState, @@ -319,47 +329,118 @@ fn applied_diff( if receipt != to { cells.extend(receipt_cleared(receipt)); } - ExpectedDiff::new().sandbox(id, cells) + let diff = ExpectedDiff::new().sandbox(id, cells); + if to != SandboxState::Deleting { + return diff; + } + diff.updated( + "jobs", + ("subject_id", id), + [ + ("status", text("failed")), + ("error_kind", text("permanent")), + ("error", text("sandbox is deleting")), + ("completed_at", Cell::Fresh), + ], + ) } -fn assert_outcome( - case: &str, - result: rusqlite::Result, - expected: Expected, -) { - match (expected, result) { - (Expected::Applied, Ok(SandboxStateWriteOutcome::Applied)) - | (Expected::Quarantine, Ok(SandboxStateWriteOutcome::BlockedByQuarantine)) - | (Expected::Superseded, Ok(SandboxStateWriteOutcome::SupersededByCleanupEpisode)) - | (Expected::Missing, Ok(SandboxStateWriteOutcome::Missing)) - | (Expected::Rejected, Err(rusqlite::Error::InvalidQuery)) => {} - (expected, result) => panic!("{case}: expected {expected:?}, got {result:?}"), +/// A generic writer the grids characterize. +#[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, + }, +} + +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. + fn write( + self, + case: &str, + conn: &Connection, + id: &str, + 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, + 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:?}"), + } + } + } + } +} + +/// `update_state_from` over `from`, standalone or inside a caller's +/// transaction that commits afterwards. +pub(super) fn write_from( + conn: &Connection, + id: &str, + from: StateSet, + to: SandboxState, + in_transaction: bool, +) -> rusqlite::Result { + if !in_transaction { + return update_state_from_states(conn, id, from, to); } + let tx = conn.unchecked_transaction()?; + let outcome = update_state_from_states(&tx, id, from, to)?; + tx.commit()?; + Ok(outcome) } /// Run one write and assert both its outcome and its row diff: an applied /// write changes exactly `applied`, any other outcome changes nothing. fn check( + writer: Writer, case: &str, conn: &Connection, id: &str, - to: SandboxState, + (seeded, to): (Option, SandboxState), expected: Expected, applied: ExpectedDiff, ) { + let case = format!("{writer:?}, {case}"); let before: Snapshot = snapshot(conn); let start = test_start(); - let result = update_state(conn, id, to); - assert_outcome(case, result, expected); + let outcome = writer.write(&case, conn, id, seeded, to); + assert_eq!(outcome, expected, "{case}"); if expected == Expected::Applied { - assert_diff(case, &before, &snapshot(conn), start, &applied); + assert_diff(&case, &before, &snapshot(conn), start, &applied); } else { - assert_unchanged(case, &before, conn); + assert_unchanged(&case, &before, conn); } } -#[test] -fn generic_state_write_outcomes_cover_every_lifecycle_pair_and_cleanup_shape() { +pub(super) fn sweep_lifecycle_pairs_and_cleanup_shapes(writer: Writer) { for (shape, receipt_kind) in Shape::ALL .into_iter() .flat_map(|shape| Receipt::ALL.map(|receipt| (shape, receipt))) @@ -371,30 +452,22 @@ fn generic_state_write_outcomes_cover_every_lifecycle_pair_and_cleanup_shape() { let conn = test_conn(); let id = seed(&conn, shape, from, receipt); let expected = cell(shape.grid()[index(from)], to); - // The teardown job's failure is `update_state`'s deletion side - // effect; every other change is to the sandbox row. - let job_failure = expected == Expected::Applied && to == SandboxState::Deleting; - let mut applied = applied_diff(&id, from, to, receipt); - if job_failure { - applied = applied.updated( - "jobs", - ("subject_id", &id), - [ - ("status", text("failed")), - ("error_kind", text("permanent")), - ("error", text("sandbox is deleting")), - ("completed_at", Cell::Fresh), - ], - ); - } - check(&case, &conn, &id, to, expected, applied); + let applied = applied_diff(&id, from, to, receipt); + check( + writer, + &case, + &conn, + &id, + (Some(from), to), + expected, + applied, + ); } } } } -#[test] -fn generic_state_write_outcomes_over_a_quarantine_episode_and_a_missing_row() { +pub(super) fn sweep_quarantine_episode_and_missing_row(writer: Writer) { for to in SandboxState::all() { let case = format!("quarantine episode: quarantined -> {to}"); let conn = test_conn(); @@ -413,22 +486,41 @@ fn generic_state_write_outcomes_over_a_quarantine_episode_and_a_missing_row() { ("failure_reason", NULL), ], ); - check(&case, &conn, &id, to, cell(QUARANTINE_EPISODE, to), applied); + check( + writer, + &case, + &conn, + &id, + (Some(SandboxState::Quarantined), to), + cell(QUARANTINE_EPISODE, to), + applied, + ); let case = format!("missing row -> {to}"); let conn = test_conn(); let id = SandboxId::new().to_string(); check( + writer, &case, &conn, &id, - to, + (None, to), cell(MISSING, to), ExpectedDiff::new(), ); } } +#[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];