diff --git a/architecture-baseline.toml b/architecture-baseline.toml index 77de4b6969..8a1b4cffdb 100644 --- a/architecture-baseline.toml +++ b/architecture-baseline.toml @@ -88,8 +88,8 @@ "crates/libs/sbx/mgr/server/src/service/create.rs" = 1195 "crates/libs/sbx/mgr/server/src/service/grpc_core.rs" = 1087 "crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend_snapshot.rs" = 1016 -"crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots.rs" = 1640 -"crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot.rs" = 1835 +"crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots.rs" = 1636 +"crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot.rs" = 1831 "crates/libs/sbx/mgr/server/src/service/mod.rs" = 1757 "crates/libs/sbx/mgr/server/src/service/network_runtime.rs" = 1112 "crates/libs/sbx/mgr/server/src/service/recovery.rs" = 2017 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 100cc886f3..827dd23db3 100644 --- a/crates/libs/sbx/mgr/server/src/db/state_write.rs +++ b/crates/libs/sbx/mgr/server/src/db/state_write.rs @@ -251,18 +251,22 @@ edge_table! { // Checkpoints and template base snapshots. - /// `do_checkpoint` and `create_template_base_snapshot` (from `paused`): - /// `update_state(Checkpointing)` checked and written under one guard. + /// `begin_checkpoint_state` from `do_checkpoint`, checked and written under + /// one guard. CHECKPOINT_BEGIN: Request, of(&[Started, Paused]) => Checkpointing; + /// `begin_template_base_snapshot_state` from + /// `create_template_base_snapshot`, checked and written under one guard. + TEMPLATE_SNAPSHOT_BEGIN: Request, of(&[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)`. + /// `complete_template_base_snapshot_state`. 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)`. + /// `rollback_template_base_snapshot_state` after a failed template base + /// snapshot. CHECKPOINT_ROLLBACK_PAUSED: Compensation, of(&[Checkpointing]) => Paused; // Suspend. @@ -274,9 +278,9 @@ edge_table! { 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. + /// `restage_rolled_back_full_snapshot`: the full process-snapshot rollback + /// restages its stopped row, so `start_from_checkpoint` can restore the + /// snapshot. A row that is already `suspended` is left as it is. PROCESS_SNAPSHOT_ROLLBACK_RESTAGE: Completion, of(&[Stopped]) => Suspended; // Stop and delete. @@ -429,14 +433,6 @@ edge_table! { 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; - /// 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; 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 edbf2db74d..c422437d98 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 @@ -658,7 +658,7 @@ pub(super) fn insert_suspended_with_artifact(conn: &Connection) -> (String, Loca (id, artifact) } -pub(super) fn decorate(conn: &Connection, id: &str, receipt: SandboxState) { +pub(crate) fn decorate(conn: &Connection, id: &str, receipt: SandboxState) { set_failure_reason(conn, id, "seeded failure"); seed_receipt(conn, id, receipt); backdate(conn, id); diff --git a/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs b/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs index 4ca292bd96..533e7d4bb1 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests/state_write_table.rs @@ -61,13 +61,12 @@ const ALLOWLIST: &[(SandboxState, SandboxState, Disagreement, &str)] = &[ (Started, Suspending, AllowedUnwritten, "suspend pauses a started row first"), (Started, Quarantined, AllowedNotRequested, RECOVERY_QUARANTINE), (Stopping, Stopped, AllowedNotRequested, COMPLETION_ONLY), - (Stopping, Suspended, WriteForbidden, "known race: 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, Paused, WriteForbidden, "known race: pause completion after Stop"), (Stopped, Suspended, RequestForbidden, "warm start and process-snapshot rollback restage a stopped row"), (Stopped, Quarantined, AllowedUnwritten, NO_QUARANTINE_WRITER), (Deleting, CleanupRequired, WriteForbidden, "an interrupted delete is handed to cleanup"), diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs index ac35beaf5c..1d776f7a22 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs @@ -331,10 +331,6 @@ impl StateWriteRejected { /// or an error is logged and leaves the row as it is, and only an applied /// write marks the sandbox dirty. `None` is a poisoned connection or a /// database error. -#[cfg_attr( - not(test), - expect(dead_code, reason = "the update_state_from conversions adopt it") -)] pub(super) fn update_state_from_best_effort( db: &Arc>, sandbox_id: &str, 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 5b6db037eb..0d8effd077 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 @@ -355,14 +355,14 @@ fn best_effort_refusals_change_nothing() { #[test] fn best_effort_database_error_is_none_and_changes_nothing() { - let edge: &'static Edge = &edges::TEMPLATE_SNAPSHOT_COMPLETE_AFTER_STOP; + let edge: &'static Edge = &edges::ORPHAN_SUSPENDED_INSERT; let db = Arc::new(Mutex::new(crate::db::test_conn())); let id = SandboxId::new().to_string(); insert_test_sandbox(&db.lock().unwrap(), &id, SandboxState::Stopped); let before = snapshot(&db.lock().unwrap()); assert_eq!(update_state_from_best_effort(&db, &id, edge), None); - assert_unchanged("known-race edge", &before, &db.lock().unwrap()); + assert_unchanged("insert edge", &before, &db.lock().unwrap()); } /// A poisoned connection is `None`, changes nothing and signals the failure: diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots.rs b/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots.rs index 2aa3731989..f8c2bcb9e0 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots.rs @@ -6,7 +6,7 @@ use super::cleanup::{ CleanupCompletionDisposition, CleanupDispatchDisposition, ObservedCleanupClaim, }; use super::cleanup_support::{ - CleanupCompletion, cleanup_sandbox_via_node, read_node_execution_fence, require_state_write, + CleanupCompletion, cleanup_sandbox_via_node, read_node_execution_fence, }; use super::create::StartFromCheckpointLaunch; use super::create_support::*; @@ -20,6 +20,7 @@ use pplx_space_events::{ emit_sandbox_event, }; use pplx_space_sbx_mgr_client::operations as spaced_ops; +use rollback_state::restage_rolled_back_full_snapshot; use std::str::FromStr; use tracing::Instrument; @@ -1687,14 +1688,7 @@ impl SandboxManagerService { "snapshot.json", )?; - // Set state to Suspended so start_from_checkpoint can proceed - { - let db = service.db_conn()?; - require_state_write( - crate::db::update_state(&db, ©_id, SandboxState::Suspended), - "complete process snapshot suspension", - )?; - } + restage_rolled_back_full_snapshot(&*service.db_conn()?, ©_id)?; Ok(artifact_guard) }) .await?; @@ -1781,6 +1775,8 @@ fn start_process_options( }) } +mod rollback_state; + #[cfg(test)] mod start_process_options_tests; diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/rollback_state.rs b/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/rollback_state.rs new file mode 100644 index 0000000000..f6a7e4a525 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/grpc_spaced_process_snapshots/rollback_state.rs @@ -0,0 +1,27 @@ +//! The full process-snapshot rollback's write to the node-local SQLite +//! `sandboxes.state`. + +use super::super::cleanup_support::{StateWriteRejected, state_write_result}; +use crate::db::edges; +use rusqlite::Connection; +use tonic::Status; + +/// Restage a full process-snapshot rollback's `stopped` row as `suspended`, so +/// `start_from_checkpoint` can restore the snapshot. A row that is already +/// `suspended` needs no write. Any other row was moved after the rollback +/// checked it, for example by a Stop admitted after the rollback's own stop, +/// and the rollback reports that it was superseded instead of overwriting it. +pub(super) fn restage_rolled_back_full_snapshot( + db: &Connection, + sandbox_id: &str, +) -> Result<(), Status> { + const CONTEXT: &str = "complete process snapshot suspension"; + let edge = &edges::PROCESS_SNAPSHOT_ROLLBACK_RESTAGE; + let outcome = crate::db::update_state_from(db, sandbox_id, edge) + .map_err(|error| Status::internal(format!("{CONTEXT}: {error}")))?; + match state_write_result(outcome) { + Ok(_) => Ok(()), + Err(StateWriteRejected::FromState { found }) if found == edge.to() => Ok(()), + Err(rejected) => Err(rejected.into_status(CONTEXT)), + } +} 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 0249779ee1..0828602b0b 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 @@ -343,3 +343,136 @@ fn rollback_replaces_optional_sidecars_or_reports_unremovable_stale_evidence() { } } } + +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 +/// `stopping` row and its completion still settles it. +#[test] +fn full_rollback_restage_leaves_a_stop_admitted_after_its_own_stop() { + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::service::tests::service_tests::insert_test_sandbox_with_runtime( + &conn, + &id, + SandboxState::Started, + ); + // The rollback's own stop. + admit_stop(&conn, &id); + complete_stop(&conn, &id); + assert_eq!(sandbox_state(&conn, &id), SandboxState::Stopped); + admit_stop(&conn, &id); + let before = crate::db::tests::database_rows(&conn); + + let error = restage_rolled_back_full_snapshot(&conn, &id) + .expect_err("the restage must not overwrite a concurrent Stop"); + + assert_eq!(error.code(), tonic::Code::Aborted); + assert_eq!( + error.message(), + "complete process snapshot suspension: sandbox state write was superseded: sandbox is stopping" + ); + assert_eq!(crate::db::tests::database_rows(&conn), before); + complete_stop(&conn, &id); + let row = crate::db::get_sandbox(&conn, &id).unwrap().unwrap(); + assert_eq!(row.state, SandboxState::Stopped); + assert_eq!(row.cleanup_completion, None); +} + +/// The restage writes only over a `stopped` row and accepts one that is +/// already `suspended` without a write. Every other row keeps every table +/// unchanged and the rollback reports why. Rows carry a failure reason, a +/// receipt for another target and backdated timestamps, which an in-place +/// rewrite of a `suspended` row would clear or restamp. +#[test] +fn full_rollback_restage_writes_only_over_a_stopped_row() { + enum Expected { + Restaged, + AlreadySuspended, + Refused(String), + } + for state in SandboxState::all() { + let expected = match state { + SandboxState::Stopped => Expected::Restaged, + SandboxState::Suspended => Expected::AlreadySuspended, + // The generic guard: a deleting row owns its `delete_row` cleanup. + SandboxState::Deleting => Expected::Refused( + "complete process snapshot suspension: sandbox state write was not applied: \ + superseded_by_cleanup_episode" + .to_string(), + ), + SandboxState::Quarantined => Expected::Refused( + "complete process snapshot suspension: sandbox state write was not applied: \ + blocked_by_quarantine" + .to_string(), + ), + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopping + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Failed => Expected::Refused(format!( + "complete process snapshot suspension: sandbox state write was superseded: \ + sandbox is {state}" + )), + }; + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::service::tests::service_tests::insert_test_sandbox_with_runtime(&conn, &id, state); + crate::db::tests::state_write_fixtures::decorate(&conn, &id, SandboxState::Started); + let before = crate::db::tests::database_rows(&conn); + + let result = restage_rolled_back_full_snapshot(&conn, &id); + + match expected { + Expected::Restaged => { + result.unwrap(); + assert_eq!(sandbox_state(&conn, &id), SandboxState::Suspended); + } + Expected::AlreadySuspended => { + result.unwrap(); + assert_eq!(crate::db::tests::database_rows(&conn), before, "{state}"); + } + Expected::Refused(message) => { + let error = result.expect_err("a refused restage must fail"); + assert_eq!(error.code(), tonic::Code::Aborted, "{state}"); + assert_eq!(error.message(), message, "{state}"); + assert_eq!(crate::db::tests::database_rows(&conn), before, "{state}"); + } + } + } +} diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot.rs b/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot.rs index aaf8a034ef..87d643ed1c 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_state_snapshot.rs @@ -1,12 +1,12 @@ use pplx_space_common_core::types::LocalArtifactId; -use super::activation_support::update_state_best_effort; use super::cache_cleanup::is_sandbox_running; -use super::cleanup_support::require_state_write; use super::create_support::*; use super::tail_support::*; use super::template_snapshot_support::{ + begin_template_base_snapshot_state, complete_template_base_snapshot_state, compute_template_base_snapshot_hash, publish_template_base_snapshot_dir, + rollback_template_base_snapshot_state, }; use super::validation::{canonical_batched_sandbox_ids, snapshot_type_from_proto}; use super::*; @@ -1785,10 +1785,7 @@ impl SandboxManagerService { .state .transition(SandboxState::Checkpointing) .map_err(Status::from)?; - require_state_write( - crate::db::update_state(&db, &sandbox.id, SandboxState::Checkpointing), - "begin template checkpoint", - )?; + begin_template_base_snapshot_state(&db, &sandbox.id)?; (sandbox, content_hash) }; @@ -1900,13 +1897,7 @@ impl SandboxManagerService { match snapshot_result { Ok((snapshot_hash, snapshot_memory_mib)) => { - { - let db = self.db_conn()?; - require_state_write( - crate::db::update_state(&db, &sandbox.id, SandboxState::Paused), - "complete template checkpoint", - )?; - } + complete_template_base_snapshot_state(&*self.db_conn()?, &sandbox.id)?; tracing::debug!( sandbox_id = %sandbox.id, template = %sandbox.template, @@ -1936,7 +1927,7 @@ impl SandboxManagerService { "template base snapshot checkpoint failed" ); let _ = std::fs::remove_dir_all(&staging_dir); - update_state_best_effort(&self.db, &sandbox.id, SandboxState::Paused); + rollback_template_base_snapshot_state(&self.db, &sandbox.id); Err(status) } } 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 ec50d61e1b..865ae09e80 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 @@ -133,3 +133,224 @@ async fn full_snapshot_capture_serializes_resume_on_wake_lock() { .unwrap(); resume_task.await.unwrap(); } + +fn sandbox_state(conn: &rusqlite::Connection, id: &str) -> SandboxState { + crate::db::get_sandbox(conn, id).unwrap().unwrap().state +} + +fn sandbox_inventory_dirty(conn: &rusqlite::Connection, id: &str) -> bool { + crate::db::list_node_dirty_refs(conn, 1000) + .unwrap() + .iter() + .any(|reference| { + reference.element_kind == "sandbox_inventory" && reference.element_value == id + }) +} + +/// 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 +/// revive the row as `paused`. +#[test] +fn template_base_snapshot_writes_leave_a_stop_that_completed_during_capture_stopped() { + for completion in [true, false] { + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::service::tests::service_tests::insert_test_sandbox_with_runtime( + &conn, + &id, + SandboxState::Paused, + ); + begin_template_base_snapshot_state(&conn, &id).unwrap(); + assert_eq!(sandbox_state(&conn, &id), SandboxState::Checkpointing); + stop_sandbox(&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)); + + if completion { + let error = complete_template_base_snapshot_state(&db.lock().unwrap(), &id) + .expect_err("completion must not revive a stopped row"); + assert_eq!(error.code(), tonic::Code::Aborted); + assert_eq!( + error.message(), + "complete template checkpoint: sandbox state write was superseded: sandbox is stopped" + ); + } else { + rollback_template_base_snapshot_state(&db, &id); + } + + assert_eq!( + crate::db::tests::database_rows(&db.lock().unwrap()), + before, + "completion={completion}" + ); + } +} + +/// Both writes return only a `checkpointing` row to `paused`. Every other row +/// keeps every table unchanged; the completion reports why to its caller. Only +/// the best-effort rollback marks the inventory dirty, and only when it writes. +/// Rows carry a failure reason, a receipt and backdated timestamps, so a write +/// is visible in every column it touches. +#[test] +fn template_base_snapshot_writes_return_only_a_checkpointing_row_to_paused() { + for state in SandboxState::all() { + let expected_error = match state { + SandboxState::Checkpointing => None, + // The generic guards: a stopping or cleanup row is never revived, + // and a deleting row owns its `delete_row` cleanup. + SandboxState::Stopping + | SandboxState::Deleting + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed => Some( + "complete template checkpoint: sandbox state write was not applied: \ + superseded_by_cleanup_episode" + .to_string(), + ), + SandboxState::Quarantined => Some( + "complete template checkpoint: sandbox state write was not applied: \ + blocked_by_quarantine" + .to_string(), + ), + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopped + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::Failed => Some(format!( + "complete template checkpoint: sandbox state write was superseded: sandbox is {state}" + )), + }; + for completion in [true, false] { + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::service::tests::service_tests::insert_test_sandbox_with_runtime( + &conn, &id, state, + ); + crate::db::tests::state_write_fixtures::decorate(&conn, &id, SandboxState::Started); + let before = crate::db::tests::database_rows(&conn); + let db = Arc::new(Mutex::new(conn)); + + let result = if completion { + complete_template_base_snapshot_state(&db.lock().unwrap(), &id) + } else { + rollback_template_base_snapshot_state(&db, &id); + Ok(()) + }; + + let conn = db.lock().unwrap(); + match &expected_error { + None => { + result.unwrap(); + assert_eq!(sandbox_state(&conn, &id), SandboxState::Paused); + assert_eq!( + sandbox_inventory_dirty(&conn, &id), + !completion, + "completion={completion}" + ); + } + Some(message) => { + if completion { + let error = result.expect_err("refused completion must fail"); + assert_eq!(error.code(), tonic::Code::Aborted, "{state}"); + assert_eq!(error.message(), message, "{state}"); + } + assert_eq!( + crate::db::tests::database_rows(&conn), + before, + "{state}, completion={completion}" + ); + } + } + } + } +} + +/// Template base-snapshot admission reads `paused` and writes `checkpointing` +/// under the same connection guard, so it writes only over a `paused` row. +/// Every other row keeps every table unchanged and reports why. +#[test] +fn template_base_snapshot_admission_writes_only_over_a_paused_row() { + for state in SandboxState::all() { + let expected_error = match state { + SandboxState::Paused => None, + // The generic guards: a stopping or cleanup row is never revived, + // and a deleting row owns its `delete_row` cleanup. + SandboxState::Stopping + | SandboxState::Deleting + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed => Some( + "begin template checkpoint: sandbox state write was not applied: \ + superseded_by_cleanup_episode" + .to_string(), + ), + SandboxState::Quarantined => Some( + "begin template checkpoint: sandbox state write was not applied: \ + blocked_by_quarantine" + .to_string(), + ), + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopped + | SandboxState::Pausing + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::Failed => Some(format!( + "begin template checkpoint: sandbox state write was superseded: sandbox is {state}" + )), + }; + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::service::tests::service_tests::insert_test_sandbox_with_runtime(&conn, &id, state); + let before = crate::db::tests::database_rows(&conn); + + let result = begin_template_base_snapshot_state(&conn, &id); + + match expected_error { + None => { + result.unwrap(); + assert_eq!(sandbox_state(&conn, &id), SandboxState::Checkpointing); + } + Some(message) => { + let error = result.expect_err("a refused admission must fail"); + assert_eq!(error.code(), tonic::Code::Aborted, "{state}"); + assert_eq!(error.message(), message, "{state}"); + assert_eq!(crate::db::tests::database_rows(&conn), before, "{state}"); + } + } + } +} diff --git a/crates/libs/sbx/mgr/server/src/service/template_snapshot_support.rs b/crates/libs/sbx/mgr/server/src/service/template_snapshot_support.rs index e8c8474c1c..20dd4add17 100644 --- a/crates/libs/sbx/mgr/server/src/service/template_snapshot_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/template_snapshot_support.rs @@ -1,3 +1,4 @@ +use super::cleanup_support::{require_state_write_from, update_state_from_best_effort}; use super::*; pub(super) fn compute_partial_sha256(path: &std::path::Path) -> std::io::Result { @@ -136,3 +137,49 @@ pub(super) fn publish_template_base_snapshot_dir( Ok(()) } + +/// Template base-snapshot admission moves the node-local SQLite +/// `sandboxes.state` from `paused` to `checkpointing`. The caller checks the +/// row under the same connection guard. +pub(super) fn begin_template_base_snapshot_state( + db: &rusqlite::Connection, + sandbox_id: &str, +) -> Result<(), Status> { + require_state_write_from( + crate::db::update_state_from(db, sandbox_id, &crate::db::edges::TEMPLATE_SNAPSHOT_BEGIN), + "begin template checkpoint", + ) +} + +/// Template base-snapshot completion returns `checkpointing` to `paused`. The +/// wake lock excludes neither Stop nor Start, so a Stop that completed during +/// capture or hashing leaves `stopped`, or whatever a later Start or Pause +/// moved it to; the completion reports that it was superseded instead of +/// overwriting the row. +pub(super) fn complete_template_base_snapshot_state( + db: &rusqlite::Connection, + sandbox_id: &str, +) -> Result<(), Status> { + require_state_write_from( + crate::db::update_state_from( + db, + sandbox_id, + &crate::db::edges::CHECKPOINT_COMPLETE_PAUSED, + ), + "complete template checkpoint", + ) +} + +/// After a failed template base snapshot, return `checkpointing` to `paused` +/// on a best-effort basis. A row another transition moved, such as a Stop, +/// keeps its state. +pub(super) fn rollback_template_base_snapshot_state( + db: &Arc>, + sandbox_id: &str, +) { + update_state_from_best_effort( + db, + sandbox_id, + &crate::db::edges::CHECKPOINT_ROLLBACK_PAUSED, + ); +} diff --git a/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs b/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs index 88e146209c..a45b1c3c02 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_scheduler.rs @@ -39,13 +39,40 @@ fn lock_db(ctx: &CheckpointContext) -> Result, -) -> Result<(), String> { - result +/// `do_checkpoint`'s admission to the node-local SQLite `sandboxes.state`: +/// `started` or `paused` to `checkpointing`, under the connection guard that +/// read the state. Guard refusals and a missing row keep `update_state`'s +/// message, and a from-state refusal reports the state found; this module's +/// errors are strings, so it cannot use the service's status mapping. +fn begin_checkpoint_state(db: &rusqlite::Connection, sandbox_id: &str) -> Result<(), String> { + use crate::db::{ + SandboxStateWriteNotApplied, SandboxStateWriteOutcome as Legacy, StateWriteOutcome, + StateWriteRefusal, + }; + let edge = &crate::db::edges::CHECKPOINT_BEGIN; + let outcome = match crate::db::update_state_from(db, sandbox_id, edge) .map_err(|error| format!("db error: {error}"))? - .require_applied() - .map_err(|error| error.to_string()) + { + StateWriteOutcome::Applied { .. } => return Ok(()), + StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::FromState, + } => { + return Err(format!( + "sandbox state write was superseded: sandbox is {found}" + )); + } + StateWriteOutcome::Refused { + refusal: StateWriteRefusal::Quarantined, + .. + } => Legacy::BlockedByQuarantine, + StateWriteOutcome::Refused { + refusal: StateWriteRefusal::CleanupOwned, + .. + } => Legacy::SupersededByCleanupEpisode, + StateWriteOutcome::Missing => Legacy::Missing, + }; + Err(SandboxStateWriteNotApplied { outcome }.to_string()) } pub(crate) fn write_file_atomic( @@ -832,11 +859,7 @@ pub async fn do_checkpoint( let previous_state = sandbox.state; // Transition to Checkpointing while still holding the lock - require_state_write(crate::db::update_state( - &db, - sandbox_id, - SandboxState::Checkpointing, - ))?; + begin_checkpoint_state(&db, sandbox_id)?; (previous_state, sandbox.node_execution_fence()) }; if previous_state == SandboxState::Started { 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 40e2ffa812..29d9261d16 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_scheduler/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_scheduler/tests.rs @@ -149,6 +149,59 @@ fn checkpoint_context(state: SandboxState) -> (CheckpointContext, String) { ) } +/// `do_checkpoint` reads `started` or `paused` and writes `checkpointing` under +/// the same connection guard, so its admission writes only from those states. +/// Every other row keeps every table unchanged and reports why. +#[test] +fn checkpoint_admission_writes_only_over_a_started_or_paused_row() { + for state in SandboxState::all() { + let expected_error = match state { + SandboxState::Started | SandboxState::Paused => None, + // The generic guards: a stopping or cleanup row is never revived, + // and a deleting row owns its `delete_row` cleanup. + SandboxState::Stopping + | SandboxState::Deleting + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed => Some( + "sandbox state write was not applied: superseded_by_cleanup_episode".to_string(), + ), + SandboxState::Quarantined => { + Some("sandbox state write was not applied: blocked_by_quarantine".to_string()) + } + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Stopped + | SandboxState::Pausing + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::Failed => Some(format!( + "sandbox state write was superseded: sandbox is {state}" + )), + }; + let (ctx, sandbox_id) = checkpoint_context(state); + let db = ctx.db.lock().unwrap(); + let before = crate::db::tests::database_rows(&db); + + let result = begin_checkpoint_state(&db, &sandbox_id); + + match expected_error { + None => { + result.unwrap(); + let row = crate::db::get_sandbox(&db, &sandbox_id).unwrap().unwrap(); + assert_eq!(row.state, SandboxState::Checkpointing, "{state}"); + } + Some(message) => { + assert_eq!(result.unwrap_err(), message, "{state}"); + assert_eq!(crate::db::tests::database_rows(&db), before, "{state}"); + } + } + } +} + #[tokio::test] async fn skipped_checkpoint_never_publishes_an_artifact() { let cases = SandboxState::all()