From 6d16ffbbbafac6e75e476bf85c6291f9585c9440 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Tue, 13 Oct 2026 04:28:33 +0000 Subject: [PATCH] sbx-mgr: characterize the state writers before the state-write helper (tests only) Pin today's behavior of sbx-mgr's writers of the node-local SQLite `sandboxes.state` (and their cleanup-episode and `sandbox_quarantine_*` companion rows), so the planned `update_state_from` helper and its caller conversions can prove they are behavior-preserving. Nothing here touches the control-plane Postgres `public.sandboxes.lifecycle`. - `update_state` outcome matrix: every (from, to) pair of the 18 lifecycle states, for a plain row, a `delete_row` intent, an armed create episode, an armed cold-start episode and a requested episode. Each pair runs once with a transition receipt for the target and once with a receipt for another state. The matrix also covers a row the production quarantine writer quarantined and a missing row. Each case asserts the outcome and the whole-database diff. The expected outcomes are literal grids indexed by a wildcard-free match over `SandboxState`, not derived from `can_transition_to`. - Whole-database accepted-shape diffs for each specialized writer that still has a production caller, keyed by primary key and diffed column by column, so a later move onto a shared helper must stay column-identical. Seeds carry a failure reason where production can, a transition receipt, a runtime with its spaced endpoint proof, route identity, artifact pins and an active teardown-suspend job. - Direct tests, with the accepting case and each refusing predicate in its own case, for restore_checkpoint_state, update_state_superseding_activation_in_transaction, settle_startup_resident, restore_adopt_orphan_or_insert_sandbox, adopt_quarantine_cleanup_episode and pin_suspended_artifact_for_cleanup. - Stop's cleanup admission function from every source state, with and without a persisted intent: admitted from 16 states, refused from `deleting` and `quarantined`. `creating`, `created`, `stopped` and `suspended` are admission-only. The Stop RPC selects only a live runtime, and from every state it admits nothing without one. The new DB modules live under `db/tests/`. The Stop module is wired by one `#[cfg(test)] mod` line in `service/cleanup.rs` and builds its service with `SandboxManagerServiceOptions::for_test_in`. Known races in today's writers are documented in the pull request, not asserted. No production code changes. --- crates/libs/sbx/mgr/server/src/db/tests.rs | 6 + .../tests/accepted_shapes_cleanup_episodes.rs | 1001 +++++++++++++++++ .../db/tests/accepted_shapes_quarantine.rs | 490 ++++++++ .../tests/accepted_shapes_single_statement.rs | 476 ++++++++ .../src/db/tests/state_write_fixtures.rs | 839 ++++++++++++++ .../src/db/tests/thin_writer_coverage.rs | 993 ++++++++++++++++ .../src/db/tests/update_state_matrix.rs | 440 ++++++++ .../sbx/mgr/server/src/service/cleanup.rs | 3 + .../service/cleanup/stop_admission_tests.rs | 276 +++++ 9 files changed, 4524 insertions(+) create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_quarantine.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_single_statement.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/state_write_fixtures.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs create mode 100644 crates/libs/sbx/mgr/server/src/db/tests/update_state_matrix.rs create mode 100644 crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs diff --git a/crates/libs/sbx/mgr/server/src/db/tests.rs b/crates/libs/sbx/mgr/server/src/db/tests.rs index 3a6870c5c1..104b764350 100644 --- a/crates/libs/sbx/mgr/server/src/db/tests.rs +++ b/crates/libs/sbx/mgr/server/src/db/tests.rs @@ -10019,5 +10019,11 @@ fn teardown_suspend_reconciliation_skips_healthy_and_retained_jobs_and_bounds_cl } } +mod accepted_shapes_cleanup_episodes; +mod accepted_shapes_quarantine; +mod accepted_shapes_single_statement; mod cleanup_quarantine_refusal; mod local_network_policy_clear; +pub(crate) mod state_write_fixtures; +mod thin_writer_coverage; +mod update_state_matrix; diff --git a/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs new file mode 100644 index 0000000000..2e11c7417b --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_cleanup_episodes.rs @@ -0,0 +1,1001 @@ +//! The complete row diff each cleanup-episode-bound writer of the node-local +//! SQLite `sandboxes.state` makes when it accepts: activation admission, +//! completion and rollback, cleanup request, failure and quarantine, and the +//! cleanup completions that settle a row. +//! +//! Refusals are pinned by `state_writer_refusal` and `cleanup_quarantine_refusal`. +//! These cases seed the side columns the writer might touch and assert the +//! whole-database diff, so moving a writer onto a shared helper must stay +//! column-identical. Auto-lifecycle is disabled on every seeded row, so +//! arming after activation writes nothing here; other tests cover arming. + +use pplx_space_common_core::types::{ + LocalArtifactId, SandboxActivationAttemptId, SandboxCleanupEpisodeId, + SandboxQuarantineReasonCode, +}; +use rusqlite::{Connection, params}; + +use super::state_write_fixtures::{ + Cell, ExpectedDiff, NULL, assert_accepts, backdate, decorate, generation, insert_creating, + insert_creating_with, insert_row, insert_suspended_with_artifact, int, + quarantine_history_entered, quarantine_state_entered, receipt_cleared, runtime_cleared, + seed_receipt, seed_runtime, snapshot_path, suspend_artifact_inserted, text, +}; +use crate::db::transfer::complete_import; +use crate::db::{ + ActivationKind, ActivationRouteAdmission, CleanupCompletion, CleanupEpisode, + CleanupEpisodeRequestOutcome, CompleteRowIntentCleanupDisposition, + begin_activation_with_route_evidence, begin_suspend, charge_prepared_clone_disposal_attempt, + complete_cleanup_episode_for_job, complete_create, complete_pool_clone, + complete_row_intent_cleanup_for_job, mark_cleanup_required, mark_restore_succeeded, + mark_start_succeeded, quarantine_cleanup_episode_for_job, rearm_prepared_clone_disposal, + record_cleanup_episode_failure, request_cleanup_episode, rollback_restoring_episode, + rollback_starting_episode, set_cleanup_intent, test_conn, update_state, +}; +use crate::state::SandboxState; + +const ATTEMPT: &str = "019a0000-0000-7000-8000-0000000000c1"; + +fn attempt() -> SandboxActivationAttemptId { + ATTEMPT.parse().unwrap() +} + +fn true_() -> Cell { + int(1) +} + +fn false_() -> Cell { + int(0) +} + +fn activate( + conn: &Connection, + id: &str, + from: SandboxState, + kind: ActivationKind, +) -> CleanupEpisode { + begin_activation_with_route_evidence( + conn, + id, + from, + kind, + Some(generation(7)), + Some(attempt()), + ActivationRouteAdmission::default(), + ) + .unwrap() + .unwrap() +} + +/// Every column of the armed episode activation admission inserts. +fn activation_episode( + id: &str, + completion: CleanupCompletion, + from: SandboxState, + activation: SandboxState, +) -> Vec<(&'static str, Cell)> { + vec![ + ("sandbox_id", text(id)), + ("cleanup_id", Cell::Fresh), + ("completion", text(completion.as_str())), + ("keep_filesystem", true_()), + ("phase", text("armed")), + ("entered_from_state", text(from.as_str())), + ("activation_state", text(activation.as_str())), + ("requested_at", NULL), + ("quarantine_reason_code", NULL), + ("failure_message", NULL), + ("last_cleanup_error", NULL), + ("execution_generation", int(7)), + ("activation_attempt_id", text(ATTEMPT)), + ("attempt_count", int(0)), + ("created_at", Cell::NowInteger), + ("recovery_local_artifact_id", NULL), + ("recovery_execution_generation", NULL), + ("prepared_clone_root", NULL), + ("operation_activation_attempt_id", NULL), + ] +} + +fn receipt_set( + receipt_id: SandboxCleanupEpisodeId, + kind: &str, + target: SandboxState, + reason: Option<&str>, +) -> Vec<(&'static str, Cell)> { + vec![ + ("transition_receipt_id", text(receipt_id.to_string())), + ("transition_receipt_kind", text(kind)), + ("transition_receipt_target_state", text(target.as_str())), + ("transition_receipt_execution_generation", int(7)), + ("transition_receipt_activation_attempt_id", text(ATTEMPT)), + ("transition_receipt_reason_code", reason.map_or(NULL, text)), + ] +} + +/// Mark the row's node execution as having owned an allocation, so admission +/// must rotate it into `previous_node_execution_id`. +fn seed_live_node_execution(conn: &Connection, id: &str) -> String { + conn.execute( + "UPDATE sandboxes SET node_execution_retired = 0 WHERE id = ?1", + params![id], + ) + .unwrap(); + conn.query_row( + "SELECT node_execution_id FROM sandboxes WHERE id = ?1", + params![id], + |row| row.get(0), + ) + .unwrap() +} + +fn node_execution_rotated(previous: &str) -> Vec<(&'static str, Cell)> { + vec![ + ("previous_node_execution_id", text(previous)), + ("node_execution_id", Cell::Fresh), + ] +} + +#[test] +fn activation_admission_claims_the_row_and_arms_one_episode() { + let conn = test_conn(); + let (suspended_restore, _) = insert_suspended_with_artifact(&conn); + for (kind, from, completion) in [ + ( + ActivationKind::ColdStart, + SandboxState::Created, + CleanupCompletion::MarkCreated, + ), + ( + ActivationKind::ColdStart, + SandboxState::Stopped, + CleanupCompletion::MarkStopped, + ), + ( + ActivationKind::ColdStart, + SandboxState::Suspended, + CleanupCompletion::MarkSuspended, + ), + ( + ActivationKind::Restore, + SandboxState::Suspended, + CleanupCompletion::MarkSuspended, + ), + ] { + let case = format!("{kind:?} from {from}"); + let id = if kind == ActivationKind::Restore { + suspended_restore.clone() + } else { + insert_row(&conn, from) + }; + let previous_execution = seed_live_node_execution(&conn, &id); + decorate(&conn, &id, SandboxState::Started); + let activation = kind.activation_state(); + let mut cells = vec![ + ("state", text(activation.as_str())), + ("failure_reason", NULL), + ("cleanup_completion", text(completion.as_str())), + ("cleanup_keep_filesystem", true_()), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + if kind == ActivationKind::Restore { + cells.push(("restore_origin", text("local"))); + cells.push(("restore_phase", text("activation"))); + } + cells.extend(node_execution_rotated(&previous_execution)); + cells.extend(receipt_cleared(SandboxState::Started)); + assert_accepts( + &case, + &conn, + || { + activate(&conn, &id, from, kind); + }, + ExpectedDiff::new().sandbox(&id, cells).inserted( + "sandbox_cleanup_episodes", + activation_episode(&id, completion, from, activation), + ), + ); + } +} + +#[test] +fn activation_completion_publishes_the_target_and_retires_its_episode() { + let conn = test_conn(); + + let starting = insert_row(&conn, SandboxState::Suspended); + let start_episode = activate( + &conn, + &starting, + SandboxState::Suspended, + ActivationKind::ColdStart, + ); + conn.execute( + "UPDATE sandboxes + SET current_local_artifact_id = ?2, + current_local_artifact_kind = 'full', + current_local_artifact_origin = 'suspend' + WHERE id = ?1", + params![starting, LocalArtifactId::new().to_string()], + ) + .unwrap(); + decorate(&conn, &starting, SandboxState::Stopped); + let mut cells = vec![ + ("state", text("started")), + ("failure_reason", NULL), + ("current_local_artifact_id", NULL), + ("current_local_artifact_kind", NULL), + ("current_local_artifact_origin", NULL), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(receipt_set( + start_episode.id, + "activation_completed", + SandboxState::Started, + None, + )); + assert_accepts( + "start succeeded", + &conn, + || assert!(mark_start_succeeded(&conn, &starting, Some(start_episode.id)).unwrap()), + ExpectedDiff::new() + .sandbox(&starting, cells) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &starting)), + ); + + for target in [SandboxState::Started, SandboxState::Paused] { + let (restoring, _) = insert_suspended_with_artifact(&conn); + let episode = activate( + &conn, + &restoring, + SandboxState::Suspended, + ActivationKind::Restore, + ); + decorate(&conn, &restoring, SandboxState::Stopped); + let mut cells = vec![ + ("state", text(target.as_str())), + ("failure_reason", NULL), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(receipt_set( + episode.id, + "activation_completed", + target, + None, + )); + assert_accepts( + &format!("restore succeeded to {target}"), + &conn, + || { + assert!( + mark_restore_succeeded(&conn, &restoring, target, Some(episode.id)).unwrap() + ) + }, + ExpectedDiff::new() + .sandbox(&restoring, cells) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &restoring)), + ); + } + + // Create completion moves no state itself: create's own generic write has + // already settled the row, and completion retires the armed episode. A + // `creating` row never carries a failure reason, so none is seeded. A + // create that does not start publishes an activation receipt, which only + // a `created` row can carry. + for publish in [false, true] { + for settled in [SandboxState::Created, SandboxState::Suspended] { + // Production create episodes carry the route's generation and + // attempt, so only the flag decides whether a receipt is written. + let (id, cleanup_id) = insert_creating_with( + &conn, + CleanupCompletion::DeleteRow, + Some((generation(7), attempt())), + ); + assert_eq!( + update_state(&conn, &id, settled).unwrap(), + crate::db::SandboxStateWriteOutcome::Applied + ); + seed_receipt(&conn, &id, SandboxState::Stopped); + backdate(&conn, &id); + let mut cells = vec![ + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("updated_at", Cell::NowText), + ]; + if publish && settled == SandboxState::Created { + cells.extend(receipt_set( + cleanup_id, + "activation_completed", + settled, + None, + )); + } + assert_accepts( + &format!("create completed as {settled}, publish {publish}"), + &conn, + || assert!(complete_create(&conn, &id, cleanup_id, publish).unwrap()), + ExpectedDiff::new() + .sandbox(&id, cells) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &id)), + ); + } + } + + let (clone, clone_cleanup) = insert_creating(&conn, CleanupCompletion::DeletePoolClone); + seed_receipt(&conn, &clone, SandboxState::Stopped); + backdate(&conn, &clone); + assert_accepts( + "pool clone published", + &conn, + || assert!(complete_pool_clone(&conn, &clone, clone_cleanup).unwrap()), + ExpectedDiff::new() + .sandbox( + &clone, + [ + ("state", text("suspended")), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &clone)), + ); + + for suspended in [false, true] { + let (import, import_cleanup) = insert_creating(&conn, CleanupCompletion::DeleteRow); + seed_receipt(&conn, &import, SandboxState::Started); + backdate(&conn, &import); + let target = if suspended { "suspended" } else { "stopped" }; + assert_accepts( + &format!("import completed as {target}"), + &conn, + || assert!(complete_import(&conn, &import, import_cleanup, suspended).unwrap()), + ExpectedDiff::new() + .sandbox( + &import, + [ + ("state", text(target)), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &import)), + ); + } +} + +#[test] +fn activation_rollback_returns_the_row_and_retires_its_episode() { + let conn = test_conn(); + let starting = insert_row(&conn, SandboxState::Stopped); + let start_episode = activate( + &conn, + &starting, + SandboxState::Stopped, + ActivationKind::ColdStart, + ); + decorate(&conn, &starting, SandboxState::Started); + assert_accepts( + "starting rollback", + &conn, + || { + assert!( + rollback_starting_episode( + &conn, + &starting, + SandboxState::Stopped, + start_episode.id + ) + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox( + &starting, + [ + ("state", text("stopped")), + ("failure_reason", NULL), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &starting)), + ); + + let (restoring, _) = insert_suspended_with_artifact(&conn); + let restore_episode = activate( + &conn, + &restoring, + SandboxState::Suspended, + ActivationKind::Restore, + ); + decorate(&conn, &restoring, SandboxState::Started); + assert_accepts( + "restoring rollback", + &conn, + || { + assert!( + rollback_restoring_episode( + &conn, + &restoring, + SandboxState::Suspended, + restore_episode.id + ) + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox( + &restoring, + [ + ("state", text("suspended")), + ("failure_reason", NULL), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &restoring)), + ); +} + +fn requested_episode( + reason: SandboxQuarantineReasonCode, + message: &str, +) -> Vec<(&'static str, Cell)> { + vec![ + ("phase", text("requested")), + ("requested_at", Cell::NowInteger), + ("quarantine_reason_code", text(reason.as_str())), + ("failure_message", text(message)), + ] +} + +fn request(conn: &Connection, id: &str, cleanup_id: SandboxCleanupEpisodeId, message: &str) { + assert_eq!( + request_cleanup_episode( + conn, + id, + cleanup_id, + SandboxQuarantineReasonCode::ColdActivationInterrupted, + message, + ) + .unwrap(), + CleanupEpisodeRequestOutcome::Requested + ); +} + +#[test] +fn cleanup_request_failure_and_quarantine_advance_the_episode_and_the_row() { + let conn = test_conn(); + let reason = SandboxQuarantineReasonCode::ColdActivationInterrupted; + + let (creating, create_cleanup) = insert_creating(&conn, CleanupCompletion::DeleteRow); + seed_receipt(&conn, &creating, SandboxState::Started); + backdate(&conn, &creating); + assert_accepts( + "request from creating", + &conn, + || request(&conn, &creating, create_cleanup, "create interrupted"), + ExpectedDiff::new() + .sandbox( + &creating, + [ + ("state", text("cleanup_required")), + ("failure_reason", text("create interrupted")), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .updated( + "sandbox_cleanup_episodes", + ("sandbox_id", &creating), + requested_episode(reason, "create interrupted"), + ), + ); + + let (restoring, _) = insert_suspended_with_artifact(&conn); + let restore_episode = activate( + &conn, + &restoring, + SandboxState::Suspended, + ActivationKind::Restore, + ); + decorate(&conn, &restoring, SandboxState::Started); + assert_accepts( + "request from restoring", + &conn, + || request(&conn, &restoring, restore_episode.id, "restore interrupted"), + ExpectedDiff::new() + .sandbox( + &restoring, + [ + ("state", text("cleanup_required")), + ("failure_reason", text("restore interrupted")), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .updated( + "sandbox_cleanup_episodes", + ("sandbox_id", &restoring), + requested_episode(reason, "restore interrupted"), + ), + ); + + backdate(&conn, &creating); + assert_accepts( + "failure from cleanup_required", + &conn, + || { + assert!( + record_cleanup_episode_failure(&conn, &creating, create_cleanup, "delete failed") + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox( + &creating, + [ + ("state", text("cleanup_failed")), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .updated( + "sandbox_cleanup_episodes", + ("sandbox_id", &creating), + [ + ("last_cleanup_error", text("delete failed")), + ("attempt_count", int(1)), + ], + ), + ); + + // The requested episode survives a repeated failure on a `cleanup_failed` + // row, which keeps its state and change time. + backdate(&conn, &creating); + assert_accepts( + "failure from cleanup_failed", + &conn, + || { + assert!( + record_cleanup_episode_failure( + &conn, + &creating, + create_cleanup, + "delete failed again" + ) + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox(&creating, [("updated_at", Cell::NowText)]) + .updated( + "sandbox_cleanup_episodes", + ("sandbox_id", &creating), + [ + ("last_cleanup_error", text("delete failed again")), + ("attempt_count", int(2)), + ], + ), + ); + + backdate(&conn, &restoring); + let quarantine_reason = + "activation compensation cannot safely reach mark_suspended; unsafe_evidence=pid reused"; + assert_accepts( + "quarantine of an activation cleanup", + &conn, + || { + assert!( + quarantine_cleanup_episode_for_job( + &conn, + &restoring, + restore_episode.id, + "pid reused", + None + ) + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox( + &restoring, + [ + ("state", text("quarantined")), + ("failure_reason", text(quarantine_reason)), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &restoring)) + .inserted( + "sandbox_quarantine_state", + quarantine_state_entered(&restoring, "cleanup_required", reason, quarantine_reason), + ) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&restoring, reason, quarantine_reason), + ), + ); + + // A pool-clone disposal keeps its episode through quarantine, so an + // operator delete can rearm it with its original identity. + // Its budget is charged and its attempt failed first, so the rearm's + // resets of both are visible. + let (clone, clone_cleanup) = insert_creating(&conn, CleanupCompletion::DeletePoolClone); + request(&conn, &clone, clone_cleanup, "clone interrupted"); + assert!(charge_prepared_clone_disposal_attempt(&conn, &clone, clone_cleanup).unwrap()); + assert!( + record_cleanup_episode_failure(&conn, &clone, clone_cleanup, "clone root busy").unwrap() + ); + backdate(&conn, &clone); + let clone_reason = + "activation compensation cannot safely reach delete_pool_clone; unsafe_evidence=root busy"; + assert_accepts( + "quarantine of a pool-clone disposal", + &conn, + || { + assert!( + quarantine_cleanup_episode_for_job(&conn, &clone, clone_cleanup, "root busy", None) + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox( + &clone, + [ + ("state", text("quarantined")), + ("failure_reason", text(clone_reason)), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .inserted( + "sandbox_quarantine_state", + quarantine_state_entered(&clone, "cleanup_failed", reason, clone_reason), + ) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&clone, reason, clone_reason), + ), + ); + + backdate(&conn, &clone); + conn.execute( + "UPDATE sandbox_cleanup_episodes SET requested_at = 1 WHERE sandbox_id = ?1", + params![clone], + ) + .unwrap(); + // The disposal's `sandbox_quarantine_state` row stays while the row is + // `cleanup_required`: pinned as current behavior. + assert_accepts( + "rearm a quarantined pool-clone disposal", + &conn, + || assert!(rearm_prepared_clone_disposal(&conn, &clone, clone_cleanup).unwrap()), + ExpectedDiff::new() + .sandbox( + &clone, + [ + ("state", text("cleanup_required")), + ("cleanup_completion", text("delete_pool_clone")), + ("cleanup_keep_filesystem", false_()), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .updated( + "sandbox_cleanup_episodes", + ("sandbox_id", &clone), + [ + ("requested_at", Cell::NowInteger), + ("quarantine_reason_code", text("create_interrupted")), + ("attempt_count", int(0)), + ("last_cleanup_error", NULL), + ], + ), + ); +} + +/// Complete the row's owned `completion` intent the way the cleanup worker +/// does after node release, with no job id. +fn complete_intent( + conn: &Connection, + id: &str, + completion: CleanupCompletion, + artifact: Option<(LocalArtifactId, &str)>, +) -> CompleteRowIntentCleanupDisposition { + complete_row_intent_cleanup_for_job(conn, id, completion, artifact, None, None).unwrap() +} + +/// The runtime columns a cleanup completion clears from a launched row, with +/// the generation and pod kept as teardown proof unless a receipt replaces it. +fn launched_runtime_cleared(keep_teardown_proof: bool) -> Vec<(&'static str, Cell)> { + let retained: &[&str] = if keep_teardown_proof { + &[ + "space_api_route_execution_generation", + "space_api_route_pod_uid", + ] + } else { + &[] + }; + runtime_cleared(retained) +} + +#[test] +fn episode_completion_applies_its_target_and_retires_the_episode() { + let reason = SandboxQuarantineReasonCode::ColdActivationInterrupted; + for from in [ + SandboxState::Created, + SandboxState::Stopped, + SandboxState::Suspended, + ] { + let conn = test_conn(); + let case = format!("cold start from {from} rolled back"); + let (id, _) = if from == SandboxState::Suspended { + insert_suspended_with_artifact(&conn) + } else { + (insert_row(&conn, from), LocalArtifactId::new()) + }; + let episode = activate(&conn, &id, from, ActivationKind::ColdStart); + seed_runtime(&conn, &id); + request(&conn, &id, episode.id, "launch failed"); + conn.execute( + "UPDATE sandboxes SET suspend_predecessor_artifact_id = ?2 WHERE id = ?1", + params![id, LocalArtifactId::new().to_string()], + ) + .unwrap(); + backdate(&conn, &id); + let mut cells = vec![ + ("state", text(from.as_str())), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("suspend_predecessor_artifact_id", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + // A rollback receipt replaces the teardown proof a plain stop keeps; + // a suspended row that still pins its suspend artifact keeps it. + cells.extend(launched_runtime_cleared(from == SandboxState::Suspended)); + cells.extend(receipt_set( + episode.id, + "activation_rollback", + from, + Some(reason.as_str()), + )); + assert_accepts( + &case, + &conn, + || { + assert!( + complete_cleanup_episode_for_job(&conn, &id, episode.id, None) + .unwrap() + .is_some() + ) + }, + ExpectedDiff::new() + .sandbox(&id, cells) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &id)), + ); + } + + for completion in [ + CleanupCompletion::DeleteRow, + CleanupCompletion::DeletePoolClone, + ] { + let conn = test_conn(); + let (id, cleanup_id) = insert_creating(&conn, completion); + request(&conn, &id, cleanup_id, "create interrupted"); + assert_accepts( + &format!("{completion} completion"), + &conn, + || { + assert!( + complete_cleanup_episode_for_job(&conn, &id, cleanup_id, None) + .unwrap() + .is_some() + ) + }, + ExpectedDiff::new() + .deleted("sandboxes", ("id", &id)) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &id)), + ); + } +} + +#[test] +fn row_intent_completion_applies_the_owned_intent() { + let conn = test_conn(); + + let stopping = insert_row(&conn, SandboxState::Stopping); + assert!(set_cleanup_intent(&conn, &stopping, true, CleanupCompletion::MarkStopped).unwrap()); + seed_runtime(&conn, &stopping); + decorate(&conn, &stopping, SandboxState::Started); + let mut cells = vec![ + ("state", text("stopped")), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(launched_runtime_cleared(true)); + cells.extend(receipt_cleared(SandboxState::Started)); + assert_accepts( + "stop", + &conn, + || { + assert_eq!( + complete_intent(&conn, &stopping, CleanupCompletion::MarkStopped, None), + CompleteRowIntentCleanupDisposition::Applied + ) + }, + ExpectedDiff::new().sandbox(&stopping, cells), + ); + + let suspending = insert_row(&conn, SandboxState::Paused); + assert!(begin_suspend(&conn, &suspending).unwrap()); + seed_runtime(&conn, &suspending); + conn.execute( + "UPDATE sandboxes SET suspend_predecessor_artifact_id = ?2 WHERE id = ?1", + params![suspending, LocalArtifactId::new().to_string()], + ) + .unwrap(); + decorate(&conn, &suspending, SandboxState::Suspended); + let artifact = LocalArtifactId::new(); + let mut cells = vec![ + ("state", text("suspended")), + ("failure_reason", NULL), + ("current_local_artifact_id", text(artifact.to_string())), + ("current_local_artifact_kind", text("full")), + ("current_local_artifact_origin", text("suspend")), + ("suspend_predecessor_artifact_id", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(launched_runtime_cleared(true)); + assert_accepts( + "suspend with its published artifact", + &conn, + || { + assert_eq!( + complete_intent( + &conn, + &suspending, + CleanupCompletion::MarkSuspended, + Some((artifact, &snapshot_path(&suspending))) + ), + CompleteRowIntentCleanupDisposition::Applied + ) + }, + ExpectedDiff::new().sandbox(&suspending, cells).inserted( + "local_artifacts", + suspend_artifact_inserted(&suspending, artifact), + ), + ); + + // Without a published or registered artifact the suspend cannot settle: + // the row is quarantined instead. + let lost = insert_row(&conn, SandboxState::Paused); + assert!(begin_suspend(&conn, &lost).unwrap()); + seed_runtime(&conn, &lost); + decorate(&conn, &lost, SandboxState::Suspended); + let lost_reason = crate::db::SUSPEND_PUBLISH_FAILED_NO_ARTIFACT_RUNTIME_GONE; + let mut cells = vec![ + ("state", text("quarantined")), + ("failure_reason", text(lost_reason)), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(launched_runtime_cleared(false)); + cells.extend(receipt_cleared(SandboxState::Suspended)); + let publication = SandboxQuarantineReasonCode::SuspendSnapshotPublicationFailed; + assert_accepts( + "suspend without an artifact", + &conn, + || { + assert_eq!( + complete_intent(&conn, &lost, CleanupCompletion::MarkSuspended, None), + CompleteRowIntentCleanupDisposition::Quarantined + ) + }, + ExpectedDiff::new() + .sandbox(&lost, cells) + .inserted( + "sandbox_quarantine_state", + quarantine_state_entered(&lost, "suspending", publication, lost_reason), + ) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&lost, publication, lost_reason), + ), + ); + + // Startup recovery's quarantine intent: `cleanup_required` with no + // completion recorded. + let held = insert_row(&conn, SandboxState::Started); + seed_runtime(&conn, &held); + mark_cleanup_required( + &conn, + &held, + true, + CleanupCompletion::Quarantine, + "probe failed", + ) + .unwrap(); + conn.execute( + "UPDATE sandboxes SET cleanup_completion = NULL, cleanup_keep_filesystem = NULL WHERE id = ?1", + params![held], + ) + .unwrap(); + seed_receipt(&conn, &held, SandboxState::Started); + backdate(&conn, &held); + let mut cells = vec![ + ("state", text("quarantined")), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(launched_runtime_cleared(false)); + cells.extend(receipt_cleared(SandboxState::Started)); + let probe = SandboxQuarantineReasonCode::RuntimeProbeInconsistent; + let mut state = quarantine_state_entered(&held, "cleanup_required", probe, "probe failed"); + for (name, cell) in &mut state { + if *name == "next_action_owner" { + *cell = text("operator_attention"); + } + } + assert_accepts( + "quarantine intent", + &conn, + || { + assert_eq!( + complete_intent(&conn, &held, CleanupCompletion::Quarantine, None), + CompleteRowIntentCleanupDisposition::Applied + ) + }, + ExpectedDiff::new() + .sandbox(&held, cells) + .inserted("sandbox_quarantine_state", state) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&held, probe, "probe failed"), + ), + ); + + let deleting = insert_row(&conn, SandboxState::Deleting); + assert!(set_cleanup_intent(&conn, &deleting, false, CleanupCompletion::DeleteRow).unwrap()); + assert_accepts( + "delete", + &conn, + || { + assert_eq!( + complete_intent(&conn, &deleting, CleanupCompletion::DeleteRow, None), + CompleteRowIntentCleanupDisposition::Applied + ) + }, + ExpectedDiff::new().deleted("sandboxes", ("id", &deleting)), + ); +} diff --git a/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_quarantine.rs b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_quarantine.rs new file mode 100644 index 0000000000..43eae0b933 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_quarantine.rs @@ -0,0 +1,490 @@ +//! The complete row diff each quarantine writer of the node-local SQLite +//! `sandboxes.state` makes when it accepts: entering quarantine, recovering a +//! pause failure, and the two terminal steps of automatic quarantine recovery, +//! orphan publication and the adopted cleanup episode's completion. +//! +//! Each case seeds the side columns and `sandbox_quarantine_*` rows the writer +//! might touch and asserts the whole-database diff, so moving a writer onto a +//! shared helper must stay column-identical. `thin_writer_coverage` covers the +//! adoption statement itself. + +use pplx_space_common_core::types::{ + ArtifactKind, ArtifactOrigin, LocalArtifactId, SandboxCleanupEpisodeId, SandboxId, + SandboxQuarantineNextAction, SandboxQuarantineReasonCode, SandboxQuarantineSource, + TransferScopeId, +}; +use rusqlite::{Connection, params}; + +use super::state_write_fixtures::{ + Cell, ExpectedDiff, NULL, as_synthetic_orphan, assert_accepts, backdate, decorate, + execution_metadata, generation, insert_row, int, pinned, quarantine_history_entered, + quarantine_state_entered, receipt_cleared, restore_source, seed_receipt, seed_runtime, + seed_teardown_job, set_suspend_pin, snapshot_path, suspend_artifact_inserted, + teardown_job_failed, text, with_sandbox_params, +}; +use crate::db::{ + ClaimedQuarantineRecoveryCandidate, PauseFailedRecoveryTarget, PlanRejectionRunEffect, + QuarantineAdoptionOutcome, QuarantineRecoveryBranch, adopt_quarantine_cleanup_episode, + claim_quarantine_recovery_candidate, complete_cleanup_episode_for_job, immediate_transaction, + insert_local_artifact, insert_quarantined_sandbox_in_transaction, + insert_restore_transfer_scope, mark_cross_node_restore_staging_failed_under_recovery_claim, + mark_quarantined_from_state, publish_orphan_quarantine_recovered_artifact, + recover_pause_failed_quarantine, restore_adopt_orphan_or_insert_sandbox_in_transaction, + scan_quarantine_recovery_candidates, set_orphan_quarantine_recovery_generation, test_conn, +}; +use crate::restore_operation::RemoteRestoreOperation; +use crate::state::SandboxState; + +fn quarantine( + conn: &Connection, + id: &str, + from: SandboxState, + reason: SandboxQuarantineReasonCode, + message: &str, +) { + assert!( + mark_quarantined_from_state( + conn, + id, + from, + SandboxQuarantineSource::SbxManager, + reason, + message, + SandboxQuarantineNextAction::RequiresOperatorAttention, + ) + .unwrap() + ); +} + +/// The quarantine episode the only recovery candidate on `conn` belongs to, +/// claimed under a live lease. +fn claim_only_candidate(conn: &Connection) -> ClaimedQuarantineRecoveryCandidate { + let [candidate] = scan_quarantine_recovery_candidates(conn) + .unwrap() + .candidates + .try_into() + .unwrap(); + claim_quarantine_recovery_candidate(conn, &candidate, PlanRejectionRunEffect::Leaves) + .unwrap() + .unwrap() +} + +fn quarantine_id(conn: &Connection, id: &str) -> String { + conn.query_row( + "SELECT quarantine_id FROM sandbox_quarantine_state WHERE sandbox_id = ?1", + params![id], + |row| row.get(0), + ) + .unwrap() +} + +#[test] +fn entering_quarantine_records_the_episode_and_clears_transition_evidence() { + let conn = test_conn(); + let reason = SandboxQuarantineReasonCode::RuntimeProbeInconsistent; + + let started = insert_row(&conn, SandboxState::Started); + seed_runtime(&conn, &started); + set_suspend_pin(&conn, &started); + conn.execute( + "UPDATE sandboxes + SET suspend_predecessor_artifact_id = current_local_artifact_id + WHERE id = ?1", + params![started], + ) + .unwrap(); + seed_teardown_job(&conn, &started); + decorate(&conn, &started, SandboxState::Started); + // The runtime, the route identity and the artifact pin stay: quarantine + // preserves uncertain resources for an operator. + let mut cells = vec![ + ("state", text("quarantined")), + ("failure_reason", text("probe failed")), + ("suspend_predecessor_artifact_id", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(receipt_cleared(SandboxState::Started)); + assert_accepts( + "from started", + &conn, + || { + quarantine( + &conn, + &started, + SandboxState::Started, + reason, + "probe failed", + ) + }, + teardown_job_failed(&started, "sandbox is quarantined") + .sandbox(&started, cells) + .inserted( + "sandbox_quarantine_state", + quarantine_state_entered(&started, "started", reason, "probe failed"), + ) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&started, reason, "probe failed"), + ), + ); + + let restoring = insert_row(&conn, SandboxState::Restoring); + conn.execute( + "UPDATE sandboxes SET restore_origin = 'local', restore_phase = 'activation' WHERE id = ?1", + params![restoring], + ) + .unwrap(); + decorate(&conn, &restoring, SandboxState::Stopped); + let launch = SandboxQuarantineReasonCode::RestoreFirecrackerLaunchFailed; + let mut cells = vec![ + ("state", text("quarantined")), + ("failure_reason", text("launch failed")), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(receipt_cleared(SandboxState::Stopped)); + assert_accepts( + "from restoring", + &conn, + || { + quarantine( + &conn, + &restoring, + SandboxState::Restoring, + launch, + "launch failed", + ) + }, + ExpectedDiff::new() + .sandbox(&restoring, cells) + .inserted( + "sandbox_quarantine_state", + quarantine_state_entered(&restoring, "restoring", launch, "launch failed"), + ) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&restoring, launch, "launch failed"), + ), + ); +} + +fn history_recorded( + id: &str, + quarantine_id: &str, + action: &str, + reason: SandboxQuarantineReasonCode, + message: &str, +) -> Vec<(&'static str, Cell)> { + vec![ + ("id", Cell::Fresh), + ("quarantine_id", text(quarantine_id)), + ("sandbox_id", text(id)), + ("occurred_at", Cell::NowText), + ("source", text("sbx_manager")), + ("action", text(action)), + ("reason_code", text(reason.as_str())), + ("message", text(message)), + ("next_action", NULL), + ("next_action_after", NULL), + ("attempt_count", NULL), + ("claim_token", NULL), + ("evidence_json", text("{}")), + ("error", NULL), + ] +} + +#[test] +fn pause_failure_recovery_settles_the_row_and_retires_its_episode() { + let reason = SandboxQuarantineReasonCode::PauseFailed; + for (target, action) in [ + (PauseFailedRecoveryTarget::Started, "recovered_to_started"), + (PauseFailedRecoveryTarget::Paused, "recovered_to_paused"), + ] { + let conn = test_conn(); + let id = insert_row(&conn, SandboxState::Pausing); + quarantine(&conn, &id, SandboxState::Pausing, reason, "pause failed"); + seed_receipt(&conn, &id, SandboxState::Started); + backdate(&conn, &id); + let quarantine_id = quarantine_id(&conn, &id); + assert_accepts( + action, + &conn, + || { + assert!( + recover_pause_failed_quarantine( + &conn, + &id.parse::().unwrap(), + target + ) + .unwrap() + ) + }, + ExpectedDiff::new() + .sandbox( + &id, + [ + ("state", text(target.sandbox_state().as_str())), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_quarantine_state", ("sandbox_id", &id)) + .inserted( + "sandbox_quarantine_history", + history_recorded( + &id, + &quarantine_id, + action, + reason, + "runtime state became definitive after pause failure", + ), + ), + ); + } +} + +/// An orphan's synthetic quarantined row, claimed by automatic recovery and +/// adopted by its restore at generation 7 with a transfer scope, as the +/// cross-node recovery path leaves it before publication. +fn insert_claimed_orphan_restore( + conn: &Connection, +) -> (String, ClaimedQuarantineRecoveryCandidate) { + let id = SandboxId::new().to_string(); + with_sandbox_params( + &id, + SandboxState::Quarantined, + as_synthetic_orphan, + |params| { + insert_quarantined_sandbox_in_transaction( + conn, + params, + SandboxQuarantineSource::SbxManager, + SandboxQuarantineReasonCode::OrphanedSandboxRowMissing, + "orphaned cache", + SandboxQuarantineNextAction::RequiresOperatorAttention, + ) + }, + ) + .unwrap(); + let claimed = claim_only_candidate(conn); + assert_eq!( + claimed.candidate.recovery, + QuarantineRecoveryBranch::RestoreOrphanedStaging + ); + let source = restore_source(); + let tx = immediate_transaction(conn).unwrap(); + with_sandbox_params( + &id, + SandboxState::Restoring, + as_synthetic_orphan, + |params| { + restore_adopt_orphan_or_insert_sandbox_in_transaction( + &tx, + params, + &RemoteRestoreOperation::artifact_staging(source), + ) + }, + ) + .unwrap(); + assert!(set_orphan_quarantine_recovery_generation(&tx, &claimed, generation(7)).unwrap()); + tx.commit().unwrap(); + assert!( + insert_restore_transfer_scope( + conn, + &id, + TransferScopeId::new(), + Some(generation(7)), + None, + &format!("/cache/restore-staging/{id}"), + ) + .unwrap() + ); + backdate(conn, &id); + (id, claimed) +} + +#[test] +fn orphan_recovery_publication_settles_suspended_with_a_recovery_receipt() { + for with_artifact in [true, false] { + let conn = test_conn(); + let (id, claimed) = insert_claimed_orphan_restore(&conn); + // A failed staging attempt the recovery retried leaves its reason. + assert!( + mark_cross_node_restore_staging_failed_under_recovery_claim( + &conn, + "cross-node restore artifact staging failed", + &claimed, + ) + .unwrap() + ); + backdate(&conn, &id); + let artifact = LocalArtifactId::new(); + let receipt = SandboxCleanupEpisodeId::new(); + let mut cells = vec![ + ("state", text("suspended")), + ("failure_reason", NULL), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("restore_storage_uri", NULL), + ("transition_receipt_id", text(receipt.to_string())), + ("transition_receipt_kind", text("quarantine_recovery")), + ("transition_receipt_target_state", text("suspended")), + ("transition_receipt_execution_generation", int(7)), + ( + "transition_receipt_reason_code", + text("orphaned_sandbox_row_missing"), + ), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + let mut expected = ExpectedDiff::new() + .deleted("sandbox_quarantine_state", ("sandbox_id", &id)) + .deleted("sandbox_restore_transfer_scopes", ("sandbox_id", &id)); + if with_artifact { + cells.extend(pinned(artifact, "full", "suspend")); + expected = + expected.inserted("local_artifacts", suspend_artifact_inserted(&id, artifact)); + } + assert_accepts( + &format!("with artifact {with_artifact}"), + &conn, + || { + assert!( + publish_orphan_quarantine_recovered_artifact( + &conn, + &claimed, + with_artifact.then_some(artifact), + &snapshot_path(&id), + generation(7), + receipt, + ) + .unwrap() + ) + }, + expected.sandbox(&id, cells), + ); + } +} + +/// A quarantined restore that pins its full suspend artifact at generation 7, +/// claimed by automatic recovery on the adoption branch. +pub(super) fn insert_claimed_adoptable( + conn: &Connection, +) -> (String, LocalArtifactId, ClaimedQuarantineRecoveryCandidate) { + let id = insert_row(conn, SandboxState::Restoring); + let artifact = LocalArtifactId::new(); + insert_local_artifact( + conn, + &id, + artifact, + ArtifactKind::Full, + ArtifactOrigin::Suspend, + None, + &snapshot_path(&id), + ) + .unwrap(); + conn.execute( + "UPDATE sandboxes + SET current_local_artifact_id = ?2, + current_local_artifact_kind = 'full', + current_local_artifact_origin = 'suspend', + space_api_route_execution_generation = 7, + restore_origin = 'local', + restore_phase = 'activation' + WHERE id = ?1", + params![id, artifact.to_string()], + ) + .unwrap(); + quarantine( + conn, + &id, + SandboxState::Restoring, + SandboxQuarantineReasonCode::RestoreFirecrackerLaunchFailed, + "restore launch failed", + ); + let claimed = claim_only_candidate(conn); + assert!(matches!( + claimed.candidate.recovery, + QuarantineRecoveryBranch::AdoptLocalArtifact(adoptable) + if adoptable.local_artifact_id() == artifact + )); + (id, artifact, claimed) +} + +pub(super) fn adoptable( + claimed: &ClaimedQuarantineRecoveryCandidate, +) -> crate::db::AdoptableArtifact { + match claimed.candidate.recovery { + QuarantineRecoveryBranch::AdoptLocalArtifact(adoptable) => adoptable, + QuarantineRecoveryBranch::RestoreOrphanedStaging => { + panic!("the fixture is on the adoption branch") + } + } +} + +#[test] +fn adopted_recovery_completion_settles_suspended_and_retires_the_quarantine() { + let conn = test_conn(); + let (id, _, claimed) = insert_claimed_adoptable(&conn); + let cleanup_id = SandboxCleanupEpisodeId::new(); + assert_eq!( + adopt_quarantine_cleanup_episode( + &conn, + &claimed, + adoptable(&claimed), + generation(7), + cleanup_id, + &execution_metadata(), + &snapshot_path(&id), + ) + .unwrap(), + QuarantineAdoptionOutcome::Adopted + ); + backdate(&conn, &id); + let quarantine_id = quarantine_id(&conn, &id); + let reason = SandboxQuarantineReasonCode::RestoreFirecrackerLaunchFailed; + let mut history = history_recorded( + &id, + &quarantine_id, + "recovered_to_suspended", + reason, + "canonical suspend artifact validated after runtime cleanup", + ); + history[0].1 = text(cleanup_id.to_string()); + // The full suspend pin keeps the generation as the restore's proof. + assert_accepts( + "adopted completion", + &conn, + || { + assert!( + complete_cleanup_episode_for_job(&conn, &id, cleanup_id, None) + .unwrap() + .is_some() + ) + }, + ExpectedDiff::new() + .sandbox( + &id, + [ + ("state", text("suspended")), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("failure_reason", NULL), + ("transition_receipt_id", text(cleanup_id.to_string())), + ("transition_receipt_kind", text("quarantine_recovery")), + ("transition_receipt_target_state", text("suspended")), + ("transition_receipt_execution_generation", int(7)), + ("transition_receipt_reason_code", text(reason.as_str())), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &id)) + .deleted("sandbox_quarantine_state", ("sandbox_id", &id)) + .inserted("sandbox_quarantine_history", history), + ); +} diff --git a/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_single_statement.rs b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_single_statement.rs new file mode 100644 index 0000000000..c7796eeecf --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/accepted_shapes_single_statement.rs @@ -0,0 +1,476 @@ +//! The complete row diff each single-statement compare-and-set writer of the +//! node-local SQLite `sandboxes.state` makes when it accepts. +//! +//! `state_writer_refusal` pins what these writers refuse. Here each accepting +//! case seeds the side columns the writer might touch (failure reason, +//! transition receipt, runtime, endpoint proof and route identity, artifact +//! pins, an active teardown-suspend job) and asserts the whole-database diff, +//! so moving a writer onto a shared helper must stay column-identical. Seeded +//! rows carry a disabled auto-lifecycle policy, so arming after activation +//! writes nothing here; `started_transition_arms_disarmed_lifecycle_for_rollback_writers` +//! and `start_arms_access_idle_only_after_guest_readiness` cover arming. + +use pplx_space_common_core::types::{ + ArtifactKind, ArtifactOrigin, LocalArtifactId, SandboxId, SandboxQuarantineNextAction, + SandboxQuarantineReasonCode, SandboxQuarantineSource, TransferScopeId, +}; +use rusqlite::Connection; + +use super::state_write_fixtures::{ + Cell, ExpectedDiff, NULL, assert_accepts, backdate, decorate, generation, insert_row, + insert_suspended_with_artifact, local_artifact_inserted, pinned, restore_source, + runtime_cleared, seed_receipt, seed_runtime, seed_teardown_job, set_failure_reason, + set_suspend_pin, snapshot_path, suspend_artifact_inserted, teardown_job_failed, text, unpinned, + with_sandbox_params, +}; + +/// The failure an eviction claim records on the teardown job it closes. +const DELETING: &str = "sandbox is deleting"; +use crate::db::{ + CleanupCompletion, PauseFailedRecoveryTarget, begin_suspend, begin_teardown_suspend, + claim_failed_restoring_sandbox_for_eviction, claim_quarantined_sandbox_for_eviction, + claim_stopped_sandbox_for_eviction, claim_suspended_sandbox_for_eviction, + claim_teardown_suspend_failed_sandbox_for_eviction, fail_teardown_suspend_backup_operation, + insert_local_artifact, insert_restore_transfer_scope, mark_cleanup_required, + mark_cross_node_restore_staging_failed, mark_quarantined_from_state, + pin_suspended_artifact_for_cleanup, publish_restored_disk_as_new, + publish_restored_suspended_artifact, publish_restored_suspended_without_artifact, + publish_stopped_artifact, recover_interrupted_pause_state, + restore_adopt_orphan_or_insert_sandbox, restore_checkpoint_state, + restore_resident_resume_state, rollback_suspending_to_paused, set_cleanup_intent, + stage_teardown_suspend_artifact, test_conn, +}; +use crate::restore_operation::RemoteRestoreOperation; +use crate::state::SandboxState; + +fn deleting_claimed() -> Vec<(&'static str, Cell)> { + vec![ + ("state", text("deleting")), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ] +} + +/// A cross-node restore staging its artifact, with a transfer scope and a +/// stale suspend-origin artifact row left by an earlier suspend on this node. +fn insert_remote_restore(conn: &Connection) -> (String, LocalArtifactId) { + let id = SandboxId::new().to_string(); + let source = restore_source(); + with_sandbox_params( + &id, + SandboxState::Restoring, + |_| {}, + |params| { + restore_adopt_orphan_or_insert_sandbox( + conn, + params, + &RemoteRestoreOperation::artifact_staging(source), + ) + }, + ) + .unwrap(); + assert!( + insert_restore_transfer_scope( + conn, + &id, + TransferScopeId::new(), + None, + None, + &format!("/cache/restore-staging/{id}"), + ) + .unwrap() + ); + let stale = LocalArtifactId::new(); + insert_local_artifact( + conn, + &id, + stale, + ArtifactKind::Full, + ArtifactOrigin::Suspend, + None, + &snapshot_path(&id), + ) + .unwrap(); + (id, stale) +} + +#[test] +fn eviction_claims_move_the_row_to_deleting_and_close_its_teardown_job() { + let conn = test_conn(); + + let stopped = insert_row(&conn, SandboxState::Stopped); + seed_teardown_job(&conn, &stopped); + decorate(&conn, &stopped, SandboxState::Started); + assert_accepts( + "stopped", + &conn, + || assert!(claim_stopped_sandbox_for_eviction(&conn, &stopped, "1").unwrap()), + teardown_job_failed(&stopped, DELETING).sandbox(&stopped, deleting_claimed()), + ); + + let quarantined = insert_row(&conn, SandboxState::Started); + assert!( + mark_quarantined_from_state( + &conn, + &quarantined, + SandboxState::Started, + SandboxQuarantineSource::SbxManager, + SandboxQuarantineReasonCode::RuntimeProbeInconsistent, + "runtime probe inconsistent", + SandboxQuarantineNextAction::RequiresOperatorAttention, + ) + .unwrap() + ); + seed_teardown_job(&conn, &quarantined); + decorate(&conn, &quarantined, SandboxState::Started); + assert_accepts( + "quarantined", + &conn, + || assert!(claim_quarantined_sandbox_for_eviction(&conn, &quarantined, "1").unwrap()), + teardown_job_failed(&quarantined, DELETING).sandbox(&quarantined, deleting_claimed()), + ); + + let (failed_restore, _) = insert_remote_restore(&conn); + assert!( + mark_cross_node_restore_staging_failed(&conn, &failed_restore, "staging failed").unwrap() + ); + seed_teardown_job(&conn, &failed_restore); + seed_receipt(&conn, &failed_restore, SandboxState::Started); + backdate(&conn, &failed_restore); + let mut cells = deleting_claimed(); + cells.extend([ + ("restore_origin", NULL), + ("restore_phase", NULL), + ("restore_storage_uri", NULL), + ]); + assert_accepts( + "failed restoring", + &conn, + || { + assert!( + claim_failed_restoring_sandbox_for_eviction(&conn, &failed_restore, "1").unwrap() + ) + }, + teardown_job_failed(&failed_restore, DELETING).sandbox(&failed_restore, cells), + ); + + // Each pin and version combination is its own statement. + for (pinned_claim, versioned) in [(true, true), (true, false), (false, true), (false, false)] { + let case = format!("suspended: pin {pinned_claim}, version {versioned}"); + let (id, artifact) = if pinned_claim { + insert_suspended_with_artifact(&conn) + } else { + let id = insert_row(&conn, SandboxState::Suspended); + (id, LocalArtifactId::new()) + }; + seed_teardown_job(&conn, &id); + decorate(&conn, &id, SandboxState::Started); + let pin = pinned_claim.then(|| artifact.to_string()); + assert_accepts( + &case, + &conn, + || { + assert!( + claim_suspended_sandbox_for_eviction( + &conn, + &id, + pin.as_deref(), + versioned.then_some("1"), + ) + .unwrap() + ) + }, + teardown_job_failed(&id, DELETING).sandbox(&id, deleting_claimed()), + ); + } + + // The terminal backup refusal already failed the row's teardown job. + let stuck = insert_row(&conn, SandboxState::Paused); + begin_teardown_suspend(&conn, &stuck, generation(7)).unwrap(); + let staged = LocalArtifactId::new(); + stage_teardown_suspend_artifact(&conn, &stuck, generation(7), staged, &snapshot_path(&stuck)) + .unwrap(); + assert!(fail_teardown_suspend_backup_operation(&conn, &stuck, generation(7), staged).unwrap()); + seed_receipt(&conn, &stuck, SandboxState::Started); + backdate(&conn, &stuck); + assert_accepts( + "terminal teardown suspend", + &conn, + || { + assert!( + claim_teardown_suspend_failed_sandbox_for_eviction(&conn, &stuck, staged, "1") + .unwrap() + ) + }, + ExpectedDiff::new().sandbox(&stuck, deleting_claimed()), + ); +} + +#[test] +fn suspend_begin_and_rollback_move_the_artifact_pin_through_the_predecessor() { + let conn = test_conn(); + let id = insert_row(&conn, SandboxState::Paused); + let pin = set_suspend_pin(&conn, &id); + decorate(&conn, &id, SandboxState::Started); + let mut cells = vec![ + ("state", text("suspending")), + ("suspend_predecessor_artifact_id", text(pin.to_string())), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(unpinned()); + assert_accepts( + "begin suspend", + &conn, + || assert!(begin_suspend(&conn, &id).unwrap()), + ExpectedDiff::new().sandbox(&id, cells), + ); + + assert!(set_cleanup_intent(&conn, &id, true, CleanupCompletion::MarkSuspended).unwrap()); + set_failure_reason(&conn, &id, "suspend failed"); + backdate(&conn, &id); + let mut cells = vec![ + ("state", text("paused")), + ("suspend_predecessor_artifact_id", NULL), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(pinned(pin, "full", "suspend")); + assert_accepts( + "suspend rollback", + &conn, + || assert!(rollback_suspending_to_paused(&conn, &id).unwrap()), + ExpectedDiff::new().sandbox(&id, cells), + ); +} + +#[test] +fn resident_settlements_write_their_target_and_clear_only_the_failure_reason() { + for target in [SandboxState::Started, SandboxState::Paused] { + let conn = test_conn(); + let settled = |state: SandboxState| { + vec![ + ("state", text(state.as_str())), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ] + }; + + let checkpointing = insert_row(&conn, SandboxState::Checkpointing); + decorate(&conn, &checkpointing, SandboxState::Stopped); + assert_accepts( + &format!("checkpoint rollback to {target}"), + &conn, + || assert!(restore_checkpoint_state(&conn, &checkpointing, target).unwrap()), + ExpectedDiff::new().sandbox(&checkpointing, settled(target)), + ); + + let pausing = insert_row(&conn, SandboxState::Pausing); + decorate(&conn, &pausing, SandboxState::Stopped); + let recovery = match target { + SandboxState::Started => PauseFailedRecoveryTarget::Started, + _ => PauseFailedRecoveryTarget::Paused, + }; + assert_accepts( + &format!("interrupted pause to {target}"), + &conn, + || assert!(recover_interrupted_pause_state(&conn, &pausing, recovery).unwrap()), + ExpectedDiff::new().sandbox(&pausing, settled(target)), + ); + + let resuming = insert_row(&conn, SandboxState::Resuming); + decorate(&conn, &resuming, SandboxState::Stopped); + assert_accepts( + &format!("resident resume rollback to {target}"), + &conn, + || assert!(restore_resident_resume_state(&conn, &resuming, target).unwrap()), + ExpectedDiff::new().sandbox(&resuming, settled(target)), + ); + } +} + +#[test] +fn restored_publications_settle_the_restore_and_retire_its_transfer_scope() { + let conn = test_conn(); + + let (with_artifact, stale) = insert_remote_restore(&conn); + decorate(&conn, &with_artifact, SandboxState::Started); + let artifact = LocalArtifactId::new(); + let mut cells = vec![ + ("state", text("suspended")), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("restore_storage_uri", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(pinned(artifact, "full", "suspend")); + assert_accepts( + "suspended with artifact", + &conn, + || { + publish_restored_suspended_artifact( + &conn, + &with_artifact, + artifact, + &snapshot_path(&with_artifact), + ) + .unwrap() + }, + ExpectedDiff::new() + .sandbox(&with_artifact, cells) + .deleted("local_artifacts", ("local_artifact_id", &stale.to_string())) + .inserted( + "local_artifacts", + suspend_artifact_inserted(&with_artifact, artifact), + ) + .deleted( + "sandbox_restore_transfer_scopes", + ("sandbox_id", &with_artifact), + ), + ); + + let (without_artifact, _) = insert_remote_restore(&conn); + decorate(&conn, &without_artifact, SandboxState::Started); + assert_accepts( + "suspended without artifact", + &conn, + || publish_restored_suspended_without_artifact(&conn, &without_artifact).unwrap(), + ExpectedDiff::new() + .sandbox( + &without_artifact, + [ + ("state", text("suspended")), + ("failure_reason", NULL), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("restore_storage_uri", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .deleted( + "sandbox_restore_transfer_scopes", + ("sandbox_id", &without_artifact), + ), + ); + + let (disk_as_new, _) = insert_remote_restore(&conn); + seed_runtime(&conn, &disk_as_new); + decorate(&conn, &disk_as_new, SandboxState::Started); + let mut cells = vec![ + ("state", text("stopped")), + ("failure_reason", NULL), + ("restore_origin", NULL), + ("restore_phase", NULL), + ("restore_storage_uri", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(runtime_cleared(&[])); + assert_accepts( + "disk as new", + &conn, + || publish_restored_disk_as_new(&conn, &disk_as_new).unwrap(), + ExpectedDiff::new().sandbox(&disk_as_new, cells).deleted( + "sandbox_restore_transfer_scopes", + ("sandbox_id", &disk_as_new), + ), + ); +} + +#[test] +fn stopped_publication_pins_the_disk_and_clears_the_runtime_from_each_accepted_shape() { + let conn = test_conn(); + let stopping = insert_row(&conn, SandboxState::Stopping); + assert!(set_cleanup_intent(&conn, &stopping, true, CleanupCompletion::MarkStopped).unwrap()); + let cleanup_required = insert_row(&conn, SandboxState::Stopping); + mark_cleanup_required( + &conn, + &cleanup_required, + true, + CleanupCompletion::MarkStopped, + "stop cleanup", + ) + .unwrap(); + let stopped = insert_row(&conn, SandboxState::Stopped); + for (case, id, intent) in [ + ("stopping", &stopping, true), + ("cleanup_required", &cleanup_required, true), + ("unpinned stopped", &stopped, false), + ] { + seed_runtime(&conn, id); + decorate(&conn, id, SandboxState::Started); + let artifact = LocalArtifactId::new(); + let path = format!("/cache/{id}/disk"); + let mut cells = vec![("failure_reason", NULL), ("updated_at", Cell::NowText)]; + if case != "unpinned stopped" { + cells.push(("state", text("stopped"))); + cells.push(("state_changed_at", Cell::NowText)); + } + if intent { + cells.push(("cleanup_completion", NULL)); + cells.push(("cleanup_keep_filesystem", NULL)); + } + cells.extend(pinned(artifact, "disk", "stop")); + cells.extend(runtime_cleared(&[])); + assert_accepts( + case, + &conn, + || publish_stopped_artifact(&conn, id, artifact, &path).unwrap(), + ExpectedDiff::new().sandbox(id, cells).inserted( + "local_artifacts", + local_artifact_inserted(id, artifact, "disk", "stop", &path), + ), + ); + } +} + +#[test] +fn suspended_cleanup_pin_hands_the_staged_artifact_to_suspend_cleanup() { + let conn = test_conn(); + let id = insert_row(&conn, SandboxState::Paused); + let predecessor = set_suspend_pin(&conn, &id); + assert!(begin_suspend(&conn, &id).unwrap()); + let stale = LocalArtifactId::new(); + insert_local_artifact( + &conn, + &id, + stale, + ArtifactKind::Full, + ArtifactOrigin::Suspend, + None, + &snapshot_path(&id), + ) + .unwrap(); + decorate(&conn, &id, SandboxState::Started); + let artifact = LocalArtifactId::new(); + let mut cells = vec![ + ("state", text("cleanup_required")), + ("suspend_predecessor_artifact_id", NULL), + ("cleanup_completion", text("mark_suspended")), + ( + "cleanup_keep_filesystem", + Cell::Is(rusqlite::types::Value::Integer(1)), + ), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + cells.extend(pinned(artifact, "full", "suspend")); + assert_ne!(predecessor, artifact); + assert_accepts( + "suspending", + &conn, + || pin_suspended_artifact_for_cleanup(&conn, &id, artifact, &snapshot_path(&id)).unwrap(), + ExpectedDiff::new() + .sandbox(&id, cells) + .deleted("local_artifacts", ("local_artifact_id", &stale.to_string())) + .inserted("local_artifacts", suspend_artifact_inserted(&id, artifact)), + ); +} 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 new file mode 100644 index 0000000000..4beb232dc6 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/state_write_fixtures.rs @@ -0,0 +1,839 @@ +//! Shared seeds and whole-database diffs for the node-local SQLite +//! `sandboxes.state` writer characterizations. +//! +//! A snapshot keys every row of every table by its primary key, so a diff can +//! tell an updated row from a replaced one, and an expectation can name exactly which +//! columns of which rows a writer may change. Anything the expectation does not +//! name must be identical before and after. + +use std::collections::BTreeMap; + +use pplx_space_common_core::types::{ + Arch, AutoLifecyclePolicy, ExecutionGeneration, LinuxPid, LinuxUid, LocalArtifactId, + RuntimeKind, SandboxActivationAttemptId, SandboxCleanupEpisodeId, SandboxId, SandboxOwner, + SandboxQuarantineReasonCode, SandboxRuntimeLabel, VsockCid, +}; +use rusqlite::types::Value; +use rusqlite::{Connection, params}; + +use super::{TestSandboxInsert, insert_test_sandbox}; +use crate::db::{ + CleanupCompletion, CleanupEpisodeSpec, InsertSandboxParams, ORPHANED_SANDBOX_OWNER, + ORPHANED_SANDBOX_TEMPLATE, PolicyAuthority, begin_suspend, complete_row_intent_cleanup_for_job, + ensure_teardown_suspend_operation, insert_sandbox_with_cleanup_episode_and_route_evidence, +}; +use crate::state::SandboxState; + +pub(super) type Row = BTreeMap; + +/// Every row of every table, keyed by table name and primary key (`rowid` +/// for a table without one), so a row deleted and re-inserted under a new key +/// reads as a deletion and an insertion even when SQLite reuses its `rowid`. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct Snapshot(BTreeMap>); + +pub(crate) fn snapshot(conn: &Connection) -> Snapshot { + let tables: Vec = conn + .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'") + .unwrap() + .query_map([], |row| row.get(0)) + .unwrap() + .collect::>() + .unwrap(); + Snapshot( + tables + .into_iter() + .map(|table| { + let mut key: Vec<(i64, String)> = conn + .prepare(&format!( + "SELECT pk, name FROM pragma_table_info('{table}') WHERE pk > 0" + )) + .unwrap() + .query_map([], |row| Ok((row.get(0)?, row.get(1)?))) + .unwrap() + .collect::>() + .unwrap(); + key.sort(); + let mut statement = conn + .prepare(&format!("SELECT rowid, * FROM \"{table}\"")) + .unwrap(); + let names: Vec = statement + .column_names() + .into_iter() + .skip(1) + .map(str::to_owned) + .collect(); + let rows = statement + .query_map([], |row| { + let rowid: i64 = row.get(0)?; + let values = names + .iter() + .enumerate() + .map(|(index, name)| { + Ok((name.clone(), row.get::<_, Value>(index + 1)?)) + }) + .collect::>()?; + let identity = if key.is_empty() { + format!("rowid {rowid}") + } else { + format!( + "{:?}", + key.iter() + .map(|(_, name)| &values[name]) + .collect::>() + ) + }; + Ok((identity, values)) + }) + .unwrap() + .collect::>>() + .unwrap(); + (table, rows) + }) + .collect(), + ) +} + +/// What one column of an expected row must hold after the write. +#[derive(Clone, Debug)] +pub(super) enum Cell { + Is(Value), + /// An epoch-seconds timestamp written by this write: text holding a value + /// no earlier than the test's start. Seeds backdate every timestamp, so a + /// fresh write is always visible as a change. + NowText, + /// As [`Cell::NowText`], stored as an integer. + NowInteger, + /// A value this write generated, such as a fresh UUID: any non-null value + /// that differs from the previous one. + Fresh, +} + +pub(super) fn text(value: impl Into) -> Cell { + Cell::Is(Value::Text(value.into())) +} + +pub(super) fn int(value: i64) -> Cell { + Cell::Is(Value::Integer(value)) +} + +pub(super) const NULL: Cell = Cell::Is(Value::Null); + +#[derive(Clone, Debug)] +enum Change { + Updated { + table: &'static str, + key: (&'static str, Value), + cells: Vec<(&'static str, Cell)>, + }, + Inserted { + table: &'static str, + cells: Vec<(&'static str, Cell)>, + }, + Deleted { + table: &'static str, + key: (&'static str, Value), + }, +} + +/// The exact set of row changes one write may make. +#[derive(Clone, Debug, Default)] +pub(super) struct ExpectedDiff { + changes: Vec, +} + +impl ExpectedDiff { + pub(super) fn new() -> Self { + Self::default() + } + + /// The row of `table` whose `key.0` column equals `key.1` changes exactly + /// the listed columns. + pub(super) fn updated( + mut self, + table: &'static str, + key: (&'static str, &str), + cells: impl IntoIterator, + ) -> Self { + self.changes.push(Change::Updated { + table, + key: (key.0, Value::Text(key.1.to_owned())), + cells: cells.into_iter().collect(), + }); + self + } + + /// The `sandboxes` row `id` changes exactly the listed columns. + pub(super) fn sandbox( + self, + id: &str, + cells: impl IntoIterator, + ) -> Self { + self.updated("sandboxes", ("id", id), cells) + } + + /// One new row of `table`; every column must be listed. + pub(super) fn inserted( + mut self, + table: &'static str, + cells: impl IntoIterator, + ) -> Self { + self.changes.push(Change::Inserted { + table, + cells: cells.into_iter().collect(), + }); + self + } + + pub(super) fn deleted(mut self, table: &'static str, key: (&'static str, &str)) -> Self { + self.changes.push(Change::Deleted { + table, + key: (key.0, Value::Text(key.1.to_owned())), + }); + self + } +} + +#[derive(Debug)] +struct ActualChange<'a> { + table: &'a str, + before: Option<&'a Row>, + after: Option<&'a Row>, +} + +fn actual_changes<'a>(before: &'a Snapshot, after: &'a Snapshot) -> Vec> { + static NO_ROWS: BTreeMap = BTreeMap::new(); + let mut changes = Vec::new(); + for (table, after_rows) in &after.0 { + let before_rows = before.0.get(table).unwrap_or(&NO_ROWS); + for (key, row) in after_rows { + match before_rows.get(key) { + Some(previous) if previous == row => {} + previous => changes.push(ActualChange { + table, + before: previous, + after: Some(row), + }), + } + } + for (key, row) in before_rows { + if !after_rows.contains_key(key) { + changes.push(ActualChange { + table, + before: Some(row), + after: None, + }); + } + } + } + changes +} + +fn cell_matches(cell: &Cell, previous: Option<&Value>, actual: &Value, start: i64) -> bool { + let fresh_timestamp = |value: Option| value.is_some_and(|value| value >= start); + match cell { + Cell::Is(expected) => expected == actual, + Cell::NowText => { + matches!(actual, Value::Text(value) if fresh_timestamp(value.parse().ok())) + } + Cell::NowInteger => { + matches!(actual, Value::Integer(value) if fresh_timestamp(Some(*value))) + } + Cell::Fresh => *actual != Value::Null && previous != Some(actual), + } +} + +/// Columns that differ between two versions of one row, with the cells the +/// expectation lists, as a readable mismatch report or `None` when they agree. +fn row_mismatch( + before: Option<&Row>, + after: &Row, + cells: &[(&'static str, Cell)], + start: i64, +) -> Option { + let mut problems = Vec::new(); + let listed: BTreeMap<&str, &Cell> = cells.iter().map(|(name, cell)| (*name, cell)).collect(); + for (name, value) in after { + let previous = before.and_then(|row| row.get(name)); + match listed.get(name.as_str()) { + Some(cell) => { + if !cell_matches(cell, previous, value, start) { + problems.push(format!("{name}: expected {cell:?}, found {value:?}")); + } else if before.is_some() && previous == Some(value) { + problems.push(format!("{name}: listed as changed but kept {value:?}")); + } + } + None if before.is_none() => problems.push(format!("{name}: unlisted {value:?}")), + None if previous != Some(value) => { + problems.push(format!( + "{name}: unexpected change {previous:?} -> {value:?}" + )); + } + None => {} + } + } + for name in listed.keys() { + if !after.contains_key(*name) { + problems.push(format!("{name}: no such column")); + } + } + (!problems.is_empty()).then(|| problems.join("; ")) +} + +/// Seconds since the epoch, captured before a write so [`Cell::NowText`] can +/// tell a fresh timestamp from a seeded one. +pub(super) fn test_start() -> i64 { + jiff::Timestamp::now().as_second() +} + +/// Assert that `after` differs from `before` by exactly `expected`. +pub(super) fn assert_diff( + case: &str, + before: &Snapshot, + after: &Snapshot, + start: i64, + expected: &ExpectedDiff, +) { + let actual = actual_changes(before, after); + let mut unmatched_actual: Vec = (0..actual.len()).collect(); + let mut problems = Vec::new(); + for change in &expected.changes { + let found = unmatched_actual.iter().position(|index| { + let candidate = &actual[*index]; + match change { + Change::Updated { table, key, .. } => { + candidate.table == *table + && candidate.after.is_some() + && candidate + .before + .is_some_and(|row| row.get(key.0) == Some(&key.1)) + } + Change::Inserted { table, .. } => { + candidate.table == *table && candidate.before.is_none() + } + Change::Deleted { table, key } => { + candidate.table == *table + && candidate.after.is_none() + && candidate + .before + .is_some_and(|row| row.get(key.0) == Some(&key.1)) + } + } + }); + let Some(position) = found else { + problems.push(format!("missing change {change:?}")); + continue; + }; + let candidate = &actual[unmatched_actual.remove(position)]; + let mismatch = match change { + Change::Updated { cells, .. } | Change::Inserted { cells, .. } => { + row_mismatch(candidate.before, candidate.after.unwrap(), cells, start) + } + Change::Deleted { .. } => None, + }; + if let Some(mismatch) = mismatch { + problems.push(format!("{}: {mismatch}", candidate.table)); + } + } + for index in unmatched_actual { + problems.push(format!("unexpected change {:?}", actual[index])); + } + assert!( + problems.is_empty(), + "{case}: the write's row diff differs from its accepted shape:\n {}", + problems.join("\n ") + ); +} + +/// Assert that a refused write left every row of every table as it was. +pub(crate) fn assert_unchanged(case: &str, before: &Snapshot, conn: &Connection) { + let after = snapshot(conn); + let changed = actual_changes(before, &after); + assert!( + changed.is_empty(), + "{case}: a refused write must leave every row unchanged; changed: {changed:?}" + ); +} + +/// Insert a plain sandbox row in `state`. +pub(super) fn insert_row(conn: &Connection, state: SandboxState) -> String { + let sandbox_id = SandboxId::new(); + let id = sandbox_id.to_string(); + insert_test_sandbox( + conn, + TestSandboxInsert::new( + &id, + SandboxRuntimeLabel::from_sandbox_id(sandbox_id).to_string(), + state, + ), + ); + id +} + +/// Move a seeded row to `state` without any writer's side effects. A +/// `restoring` row carries a remote restore's three restore columns, which the +/// table's `CHECK` ties to that state. +pub(super) fn force_state(conn: &Connection, id: &str, state: SandboxState) { + let restoring = state == SandboxState::Restoring; + conn.execute( + "UPDATE sandboxes + SET state = ?1, + restore_origin = CASE WHEN ?2 THEN 'remote' END, + restore_phase = CASE WHEN ?2 THEN 'metadata_staging' END, + restore_storage_uri = CASE WHEN ?2 THEN ?4 END + WHERE id = ?3", + params![state.as_str(), restoring, id, RESTORE_STORAGE_URI], + ) + .unwrap(); +} + +pub(super) const RESTORE_STORAGE_URI: &str = "s3://snapshots/suspend/restore?region=us-east-1"; + +/// The remote snapshot a seeded cross-node restore stages from. +pub(super) fn restore_source() -> pplx_space_common_core::types::ExactStorageUri { + RESTORE_STORAGE_URI.parse().unwrap() +} + +/// Backdate every timestamp a writer may stamp, so a fresh stamp is visible. +pub(super) fn backdate(conn: &Connection, id: &str) { + conn.execute( + "UPDATE sandboxes SET state_changed_at = '1', updated_at = '1' WHERE id = ?1", + params![id], + ) + .unwrap(); +} + +/// A live runtime with its spaced endpoint proof and complete route identity: +/// the columns a teardown writer clears and the route-retention trigger +/// observes. +pub(super) const RUNTIME_ROUTE_GENERATION: i64 = 3; +pub(super) const RUNTIME_EXECUTION_GENERATION: i64 = 7; +pub(super) const RUNTIME_ACTIVATION_ATTEMPT: &str = "019a0000-0000-7000-8000-000000000001"; +pub(super) const RUNTIME_POD_UID: &str = "pod-uid-1"; + +pub(super) fn seed_runtime(conn: &Connection, id: &str) { + conn.execute( + "UPDATE sandboxes + SET uid = 2001, slot = 1, guest_cid = 3, pid = 4242, + spaced_host_pid = 4243, netgtw_pid = 4244, + vsock_uds_path = 'vsock.sock', api_socket_path = 'api.sock', + guest_uptime_secs = 9, + space_api_route_generation = ?2, + space_api_route_node_id = '019a0000-0000-7000-8000-0000000000fe', + space_api_route_execution_generation = ?3, + space_api_route_pod_uid = ?4, + space_api_route_substrate_kind = 'k8s', + space_api_route_activation_attempt_id = ?5, + space_api_route_identity_provenance = 'cluster_fenced', + spaced_endpoint_proof_json = ?6 + WHERE id = ?1", + params![ + id, + RUNTIME_ROUTE_GENERATION, + RUNTIME_EXECUTION_GENERATION, + RUNTIME_POD_UID, + RUNTIME_ACTIVATION_ATTEMPT, + serde_json::to_string(&endpoint_proof(id.parse().unwrap())).unwrap(), + ], + ) + .unwrap(); +} + +/// The columns a write that clears [`seed_runtime`]'s runtime and route fields +/// changes, including the route-retention trigger's observation, with +/// `retained` naming the route columns the write keeps. +pub(super) fn runtime_cleared(retained: &[&str]) -> Vec<(&'static str, Cell)> { + [ + ("uid", NULL), + ("slot", NULL), + ("guest_cid", NULL), + ("pid", NULL), + ("spaced_host_pid", NULL), + ("netgtw_pid", NULL), + ("vsock_uds_path", NULL), + ("api_socket_path", NULL), + ("guest_uptime_secs", NULL), + ("space_api_route_generation", NULL), + ("space_api_route_node_id", NULL), + ("space_api_route_execution_generation", NULL), + ("space_api_route_pod_uid", NULL), + ("space_api_route_substrate_kind", NULL), + ("space_api_route_activation_attempt_id", NULL), + ("space_api_route_identity_provenance", NULL), + ("spaced_endpoint_proof_json", NULL), + ( + "last_observed_execution_generation", + int(RUNTIME_EXECUTION_GENERATION), + ), + ( + "last_observed_route_activation_id", + text(RUNTIME_ACTIVATION_ATTEMPT), + ), + ("last_observed_route_substrate_kind", text("k8s")), + ( + "last_observed_route_identity_provenance", + text("cluster_fenced"), + ), + ( + "last_observed_route_generation", + int(RUNTIME_ROUTE_GENERATION), + ), + ("last_observed_route_generation_substrate_kind", text("k8s")), + ] + .into_iter() + .filter(|(name, _)| !retained.contains(name)) + .collect() +} + +/// A transition receipt targeting `target`, in the kind the table's `CHECK` +/// admits for it: an activation completion for created, started and paused, +/// otherwise an activation rollback. +pub(super) fn seed_receipt(conn: &Connection, id: &str, target: SandboxState) { + let (kind, reason) = match target { + SandboxState::Created | SandboxState::Started | SandboxState::Paused => { + ("activation_completed", None) + } + SandboxState::Stopped | SandboxState::Suspended => { + ("activation_rollback", Some("cold_activation_interrupted")) + } + other => panic!("{other} cannot be a transition receipt target"), + }; + conn.execute( + "UPDATE sandboxes + SET transition_receipt_id = '019a0000-0000-7000-8000-0000000000aa', + transition_receipt_kind = ?2, + transition_receipt_target_state = ?3, + transition_receipt_execution_generation = 5, + transition_receipt_activation_attempt_id = '019a0000-0000-7000-8000-0000000000ab', + transition_receipt_reason_code = ?4 + WHERE id = ?1", + params![id, kind, target.as_str(), reason], + ) + .unwrap(); +} + +/// The receipt columns a write that clears [`seed_receipt`]'s receipt for +/// `target` changes: an activation completion carries no reason code. +pub(super) fn receipt_cleared(target: SandboxState) -> Vec<(&'static str, Cell)> { + let mut cells = vec![ + ("transition_receipt_id", NULL), + ("transition_receipt_kind", NULL), + ("transition_receipt_target_state", NULL), + ("transition_receipt_execution_generation", NULL), + ("transition_receipt_activation_attempt_id", NULL), + ]; + if matches!(target, SandboxState::Stopped | SandboxState::Suspended) { + cells.push(("transition_receipt_reason_code", NULL)); + } + cells +} + +pub(super) fn set_failure_reason(conn: &Connection, id: &str, reason: &str) { + conn.execute( + "UPDATE sandboxes SET failure_reason = ?2 WHERE id = ?1", + params![id, reason], + ) + .unwrap(); +} + +pub(super) fn generation(value: i64) -> ExecutionGeneration { + ExecutionGeneration::new(value).unwrap() +} + +/// Run `write` and assert it changed exactly `expected`. +pub(super) fn assert_accepts( + case: &str, + conn: &Connection, + write: impl FnOnce(), + expected: ExpectedDiff, +) { + let before = snapshot(conn); + let start = test_start(); + write(); + assert_diff(case, &before, &snapshot(conn), start, &expected); +} + +pub(super) fn snapshot_path(id: &str) -> String { + format!("/cache/{id}/snapshot") +} + +/// An active teardown-suspend job: a `deleting` claim must close it in the +/// same transaction. +pub(super) fn seed_teardown_job(conn: &Connection, id: &str) { + ensure_teardown_suspend_operation(conn, id, generation(7)).unwrap(); +} + +pub(super) fn teardown_job_failed(id: &str, error: &str) -> ExpectedDiff { + ExpectedDiff::new().updated( + "jobs", + ("subject_id", id), + [ + ("status", text("failed")), + ("error_kind", text("permanent")), + ("error", text(error)), + ("completed_at", Cell::Fresh), + ], + ) +} + +/// A suspend artifact row the writer inserts for `id`. +pub(super) fn suspend_artifact_inserted( + id: &str, + artifact: LocalArtifactId, +) -> Vec<(&'static str, Cell)> { + local_artifact_inserted(id, artifact, "full", "suspend", &snapshot_path(id)) +} + +pub(super) fn local_artifact_inserted( + id: &str, + artifact: LocalArtifactId, + kind: &str, + origin: &str, + path: &str, +) -> Vec<(&'static str, Cell)> { + vec![ + ("local_artifact_id", text(artifact.to_string())), + ("sandbox_id", text(id)), + ("artifact_kind", text(kind)), + ("artifact_origin", text(origin)), + ("snapshot_name", NULL), + ("local_path", text(path)), + ("created_at", Cell::NowText), + ("last_touched_at", Cell::NowText), + ("backup_completed_at", NULL), + ("state", text("ready")), + ] +} + +pub(super) fn pinned( + artifact: LocalArtifactId, + kind: &str, + origin: &str, +) -> Vec<(&'static str, Cell)> { + vec![ + ("current_local_artifact_id", text(artifact.to_string())), + ("current_local_artifact_kind", text(kind)), + ("current_local_artifact_origin", text(origin)), + ] +} + +pub(super) fn unpinned() -> Vec<(&'static str, Cell)> { + vec![ + ("current_local_artifact_id", NULL), + ("current_local_artifact_kind", NULL), + ("current_local_artifact_origin", NULL), + ] +} + +/// Pin a suspend artifact without its `local_artifacts` row, as a re-suspend's +/// predecessor carries it. +pub(super) fn set_suspend_pin(conn: &Connection, id: &str) -> LocalArtifactId { + let pin = LocalArtifactId::new(); + conn.execute( + "UPDATE sandboxes + SET current_local_artifact_id = ?1, + current_local_artifact_kind = 'full', + current_local_artifact_origin = 'suspend' + WHERE id = ?2", + params![pin.to_string(), id], + ) + .unwrap(); + pin +} + +/// A `suspended` row pinning its published suspend artifact. +pub(super) fn insert_suspended_with_artifact(conn: &Connection) -> (String, LocalArtifactId) { + let id = insert_row(conn, SandboxState::Paused); + assert!(begin_suspend(conn, &id).unwrap()); + let artifact = LocalArtifactId::new(); + complete_row_intent_cleanup_for_job( + conn, + &id, + CleanupCompletion::MarkSuspended, + Some((artifact, &snapshot_path(&id))), + None, + None, + ) + .unwrap(); + (id, artifact) +} + +pub(super) fn decorate(conn: &Connection, id: &str, receipt: SandboxState) { + set_failure_reason(conn, id, "seeded failure"); + seed_receipt(conn, id, receipt); + backdate(conn, id); +} + +/// Build the insert parameters of an ordinary sandbox `id` in `state`, let +/// `customize` adjust them, and pass them to `insert`. +pub(super) fn with_sandbox_params( + id: &str, + state: SandboxState, + customize: impl FnOnce(&mut InsertSandboxParams<'_>), + insert: impl FnOnce(InsertSandboxParams<'_>) -> T, +) -> T { + let uuid8 = SandboxRuntimeLabel::from_sandbox_id(id.parse().unwrap()); + let user = SandboxOwner::parse("user").unwrap(); + let env = BTreeMap::new(); + let lifecycle = AutoLifecyclePolicy::disabled(); + let mut params = InsertSandboxParams { + id, + uuid8: &uuid8, + state, + template: "seed", + template_hash: Some("hash"), + user: &user, + runtime: "fc", + snapshots_enabled: false, + rootfs_size_mb: None, + disk_io_limits: Default::default(), + pool_template: None, + volume_encryption: false, + process_env: &env, + auto_lifecycle: &lifecycle, + network_policy_override: None, + network_policy_generation: 0, + network_policy_authority: PolicyAuthority::Cluster, + auto_lifecycle_authority: PolicyAuthority::Cluster, + }; + customize(&mut params); + insert(params) +} + +/// The synthetic row the orphan reconciler materializes for a cache it cannot +/// match to a sandbox. +pub(super) fn as_synthetic_orphan(params: &mut InsertSandboxParams<'_>) { + params.template = ORPHANED_SANDBOX_TEMPLATE; + params.template_hash = None; + params.user = &ORPHANED_SANDBOX_OWNER; + params.runtime = RuntimeKind::Firecracker.alias(); +} + +/// A creating row owning an armed create episode with `completion`, bound to +/// `execution` when the create carries a route identity. +pub(super) fn insert_creating_with( + conn: &Connection, + completion: CleanupCompletion, + execution: Option<(ExecutionGeneration, SandboxActivationAttemptId)>, +) -> (String, SandboxCleanupEpisodeId) { + let id = SandboxId::new().to_string(); + let cleanup_id = SandboxCleanupEpisodeId::new(); + with_sandbox_params( + &id, + SandboxState::Creating, + |_| {}, + |params| { + insert_sandbox_with_cleanup_episode_and_route_evidence( + conn, + params, + CleanupEpisodeSpec { + id: cleanup_id, + completion, + keep_filesystem: false, + entered_from_state: None, + activation_state: SandboxState::Creating, + execution_generation: execution.map(|(generation, _)| generation), + activation_attempt_id: execution.map(|(_, attempt)| attempt), + }, + None, + None, + ) + }, + ) + .unwrap(); + (id, cleanup_id) +} + +pub(super) fn insert_creating( + conn: &Connection, + completion: CleanupCompletion, +) -> (String, SandboxCleanupEpisodeId) { + insert_creating_with(conn, completion, None) +} + +/// Every column of a fresh `sandbox_quarantine_state` row. +pub(super) fn quarantine_state_entered( + id: &str, + from: &str, + reason: SandboxQuarantineReasonCode, + message: &str, +) -> Vec<(&'static str, Cell)> { + vec![ + ("sandbox_id", text(id)), + ("quarantine_id", Cell::Fresh), + ("entered_at", Cell::NowText), + ("updated_at", Cell::NowText), + ("entered_from_state", text(from)), + ("source", text("sbx_manager")), + ("reason_code", text(reason.as_str())), + ("reason_message", text(message)), + ("evidence_json", text("{}")), + ("next_action", text("requires_operator_attention")), + ("next_action_after", NULL), + ("next_action_owner", NULL), + ("attempt_count", int(0)), + ("claim_token", NULL), + ("claim_expires_at", NULL), + ("last_error", NULL), + ] +} + +/// Every column of the history row a fresh quarantine episode records. +pub(super) fn quarantine_history_entered( + id: &str, + reason: SandboxQuarantineReasonCode, + message: &str, +) -> Vec<(&'static str, Cell)> { + vec![ + ("id", Cell::Fresh), + ("quarantine_id", Cell::Fresh), + ("sandbox_id", text(id)), + ("occurred_at", Cell::NowText), + ("source", text("sbx_manager")), + ("action", text("entered_quarantine")), + ("reason_code", text(reason.as_str())), + ("message", text(message)), + ("next_action", text("requires_operator_attention")), + ("next_action_after", NULL), + ("attempt_count", NULL), + ("claim_token", NULL), + ("evidence_json", text("{}")), + ("error", NULL), + ] +} + +/// A spaced endpoint proof for `sandbox_id`, matching [`seed_runtime`]'s +/// runtime and route identity. +pub(super) fn endpoint_proof(sandbox_id: SandboxId) -> crate::endpoint_proof::SpacedEndpointProof { + use crate::endpoint_proof::{ + ProcessActivationIdentity, SpacedEndpointProof, VsockActivationIdentity, + }; + let process = |pid| ProcessActivationIdentity { + pid: LinuxPid::try_from_u32(pid).unwrap(), + start_time_ticks: 11, + executable_device: 12, + executable_inode: 13, + }; + SpacedEndpointProof { + sandbox_id, + execution_generation: generation(RUNTIME_EXECUTION_GENERATION), + activation_attempt_id: RUNTIME_ACTIVATION_ATTEMPT.parse().unwrap(), + spaced_host: process(4243), + runtime_process: process(4242), + vsock: VsockActivationIdentity { + path: "vsock.sock".into(), + device: 31, + inode: 32, + owner_uid: LinuxUid::new(2001), + owner_gid: 2001, + guest_cid: VsockCid::try_from_u32(3).unwrap(), + runtime_fd: 8, + runtime_socket_inode: 33, + }, + } +} + +/// The execution metadata quarantine adoption records. +pub(super) fn execution_metadata() -> crate::db::ExecutionMetadata { + crate::db::ExecutionMetadata::initial_firecracker(Arch::Aarch64, vec![]) +} diff --git a/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs b/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs new file mode 100644 index 0000000000..340c2a29de --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/thin_writer_coverage.rs @@ -0,0 +1,993 @@ +//! Direct tests for node-local SQLite `sandboxes.state` writers that had no or +//! thin coverage: each one's accepting case and each refusing predicate in its +//! own case, every refusal starting from a row the write would otherwise change. + +use pplx_space_common_core::types::{ + LocalArtifactId, RuntimeKind, SandboxCleanupEpisodeId, SandboxId, SandboxQuarantineNextAction, + SandboxQuarantineReasonCode, SandboxQuarantineSource, TransferScopeId, +}; +use rusqlite::{Connection, params}; + +use super::accepted_shapes_quarantine::{adoptable, insert_claimed_adoptable}; +use super::state_write_fixtures::{ + Cell, ExpectedDiff, NULL, RESTORE_STORAGE_URI, Snapshot, as_synthetic_orphan, assert_accepts, + assert_unchanged, backdate, decorate, endpoint_proof, execution_metadata, generation, + insert_creating, insert_row, int, quarantine_history_entered, quarantine_state_entered, + receipt_cleared, restore_source, seed_receipt, set_failure_reason, snapshot, snapshot_path, + text, with_sandbox_params, +}; +use crate::db::{ + ActivationKind, ActivationRouteAdmission, CleanupCompletion, CleanupEpisodeSpec, + QuarantineAdoptionOutcome, RestoreAdoptOrphanInsertError, SandboxStateWriteOutcome, + StartupResidentSettlement, SupersedeActivationError, adopt_quarantine_cleanup_episode, + begin_activation_with_route_evidence, begin_teardown_suspend, get_sandbox, + immediate_transaction, insert_cleanup_episode, insert_quarantined_sandbox_in_transaction, + insert_restore_transfer_scope, mark_local_artifact_backup_completed, + pin_suspended_artifact_for_cleanup, request_cleanup_episode, + restore_adopt_orphan_or_insert_sandbox, restore_checkpoint_state, settle_startup_resident, + stage_teardown_suspend_artifact, test_conn, update_state_superseding_activation_in_transaction, +}; +use crate::restore_operation::RemoteRestoreOperation; +use crate::state::SandboxState; + +fn refuses( + case: &str, + conn: &Connection, + write: impl FnOnce() -> T, + refused: impl FnOnce(&T) -> bool, +) { + let before: Snapshot = snapshot(conn); + let result = write(); + assert!( + refused(&result), + "{case}: expected a refusal, got {result:?}" + ); + assert_unchanged(case, &before, conn); +} + +#[test] +fn checkpoint_rollback_requires_a_checkpointing_row_and_a_resident_target() { + let conn = test_conn(); + for target in SandboxState::all() { + let id = insert_row(&conn, SandboxState::Checkpointing); + let case = format!("checkpointing -> {target}"); + if matches!(target, SandboxState::Started | SandboxState::Paused) { + // `accepted_shapes_single_statement` pins the accepted row diff. + assert!( + restore_checkpoint_state(&conn, &id, target).unwrap(), + "{case}" + ); + assert_eq!( + get_sandbox(&conn, &id).unwrap().unwrap().state, + target, + "{case}" + ); + } else { + refuses( + &case, + &conn, + || restore_checkpoint_state(&conn, &id, target), + |result| matches!(result, Err(rusqlite::Error::InvalidQuery)), + ); + } + } + for state in SandboxState::all() { + if state == SandboxState::Checkpointing { + continue; + } + let id = insert_row(&conn, state); + refuses( + state.as_str(), + &conn, + || restore_checkpoint_state(&conn, &id, SandboxState::Started), + |result| matches!(result, Ok(false)), + ); + } + refuses( + "missing row", + &conn, + || restore_checkpoint_state(&conn, &SandboxId::new().to_string(), SandboxState::Paused), + |result| matches!(result, Ok(false)), + ); +} + +/// Cleanup admission's call: the writer in the caller's transaction, which a +/// refusal abandons. +fn supersede( + conn: &Connection, + id: &str, + target: SandboxState, +) -> Result<(), SupersedeActivationError> { + let tx = immediate_transaction(conn)?; + update_state_superseding_activation_in_transaction(&tx, id, target)?; + tx.commit()?; + Ok(()) +} + +fn superseded(target: SandboxState) -> Vec<(&'static str, Cell)> { + vec![ + ("state", text(target.as_str())), + ("cleanup_completion", NULL), + ("cleanup_keep_filesystem", NULL), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ] +} + +/// Per-source-state outcomes are `update_state`'s, pinned by +/// `update_state_matrix`; these cases pin what supersession adds around it. +#[test] +fn activation_supersession_retires_the_episode_before_the_generic_write() { + for target in [SandboxState::Stopping, SandboxState::Deleting] { + let conn = test_conn(); + + let starting = insert_row(&conn, SandboxState::Stopped); + begin_activation_with_route_evidence( + &conn, + &starting, + SandboxState::Stopped, + ActivationKind::ColdStart, + None, + None, + ActivationRouteAdmission::default(), + ) + .unwrap() + .unwrap(); + decorate(&conn, &starting, SandboxState::Started); + let mut cells = superseded(target); + cells.extend(receipt_cleared(SandboxState::Started)); + assert_accepts( + &format!("armed cold start -> {target}"), + &conn, + || supersede(&conn, &starting, target).unwrap(), + ExpectedDiff::new() + .sandbox(&starting, cells) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &starting)), + ); + + let (requested, cleanup_id) = insert_creating(&conn, CleanupCompletion::DeleteRow); + request_cleanup_episode( + &conn, + &requested, + cleanup_id, + SandboxQuarantineReasonCode::CreateInterrupted, + "create interrupted", + ) + .unwrap(); + backdate(&conn, &requested); + assert_accepts( + &format!("requested create cleanup -> {target}"), + &conn, + || supersede(&conn, &requested, target).unwrap(), + ExpectedDiff::new() + .sandbox(&requested, superseded(target)) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &requested)), + ); + + // A create quarantined while its episode was armed: the generic write + // refuses a stop of the quarantined row, and the episode it already + // deleted returns with the abandoned transaction. + let (held, _) = insert_creating(&conn, CleanupCompletion::DeleteRow); + assert!( + crate::db::mark_quarantined_from_state( + &conn, + &held, + SandboxState::Creating, + SandboxQuarantineSource::SbxManager, + SandboxQuarantineReasonCode::CreateInterrupted, + "create interrupted", + SandboxQuarantineNextAction::RequiresOperatorAttention, + ) + .unwrap() + ); + backdate(&conn, &held); + if target == SandboxState::Stopping { + refuses( + "quarantined create -> stopping", + &conn, + || supersede(&conn, &held, target), + |result| { + matches!( + result, + Err(SupersedeActivationError::StateWriteNotApplied(error)) + if error.outcome == SandboxStateWriteOutcome::BlockedByQuarantine + ) + }, + ); + } else { + assert_accepts( + "quarantined create -> deleting", + &conn, + || supersede(&conn, &held, target).unwrap(), + ExpectedDiff::new() + .sandbox(&held, superseded(target)) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &held)), + ); + } + + // Automatic quarantine recovery's adopted episode yields only to a + // deletion. + let (adopted, _, claimed) = insert_claimed_adoptable(&conn); + assert_eq!( + adopt_quarantine_cleanup_episode( + &conn, + &claimed, + adoptable(&claimed), + generation(7), + SandboxCleanupEpisodeId::new(), + &execution_metadata(), + &snapshot_path(&adopted), + ) + .unwrap(), + QuarantineAdoptionOutcome::Adopted + ); + backdate(&conn, &adopted); + if target == SandboxState::Stopping { + refuses( + "adopted recovery -> stopping", + &conn, + || supersede(&conn, &adopted, target), + |result| { + matches!( + result, + Err(SupersedeActivationError::QuarantineRecoveryInProgress) + ) + }, + ); + } else { + assert_accepts( + "adopted recovery -> deleting", + &conn, + || supersede(&conn, &adopted, target).unwrap(), + ExpectedDiff::new() + .sandbox(&adopted, superseded(target)) + .deleted("sandbox_cleanup_episodes", ("sandbox_id", &adopted)), + ); + } + + refuses( + &format!("missing row -> {target}"), + &conn, + || supersede(&conn, &SandboxId::new().to_string(), target), + |result| { + matches!( + result, + Err(SupersedeActivationError::StateWriteNotApplied(error)) + if error.outcome == SandboxStateWriteOutcome::Missing + ) + }, + ); + } +} + +/// A resident row as startup recovery observes it: a failure reason and an +/// activation receipt for `started`, with backdated timestamps. +fn insert_resident(conn: &Connection, state: SandboxState) -> String { + let id = insert_row(conn, state); + decorate(conn, &id, SandboxState::Started); + id +} + +fn settlement(target: SandboxState) -> StartupResidentSettlement<'static> { + match target { + SandboxState::Started => StartupResidentSettlement::Started, + SandboxState::Paused => StartupResidentSettlement::Paused, + _ => StartupResidentSettlement::Quarantined { + reason_code: SandboxQuarantineReasonCode::RuntimeProbeInconsistent, + message: "no live runtime", + }, + } +} + +#[test] +fn startup_resident_settlement_writes_the_observed_outcome() { + let probe = SandboxQuarantineReasonCode::RuntimeProbeInconsistent; + for observed_state in [ + SandboxState::Started, + SandboxState::Paused, + SandboxState::Resuming, + SandboxState::Checkpointing, + ] { + for target in [ + SandboxState::Started, + SandboxState::Paused, + SandboxState::Quarantined, + ] { + let case = format!("{observed_state} settled {target}"); + let conn = test_conn(); + let id = insert_resident(&conn, observed_state); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + // The change time is stamped even when the state is unchanged. + let mut cells = vec![ + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + if observed_state != target { + cells.push(("state", text(target.as_str()))); + } + if target != SandboxState::Started { + cells.extend(receipt_cleared(SandboxState::Started)); + } + let mut expected = ExpectedDiff::new(); + if target == SandboxState::Quarantined { + cells.push(("failure_reason", text("no live runtime"))); + expected = expected + .inserted( + "sandbox_quarantine_state", + quarantine_state_entered( + &id, + observed_state.as_str(), + probe, + "no live runtime", + ), + ) + .inserted( + "sandbox_quarantine_history", + quarantine_history_entered(&id, probe, "no live runtime"), + ); + } else { + cells.push(("failure_reason", NULL)); + } + assert_accepts( + &case, + &conn, + || assert!(settle_startup_resident(&conn, &observed, settlement(target)).unwrap()), + expected.sandbox(&id, cells), + ); + } + } +} + +#[test] +fn startup_resident_settlement_refuses_any_change_to_the_observed_row() { + let conn = test_conn(); + for state in SandboxState::all() { + if matches!( + state, + SandboxState::Started + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + ) { + continue; + } + let id = insert_resident(&conn, state); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + refuses( + &format!("observed {state}"), + &conn, + || settle_startup_resident(&conn, &observed, StartupResidentSettlement::Started), + |result| matches!(result, Err(rusqlite::Error::InvalidQuery)), + ); + } + + // Each change below supersedes the observation on its own. + let perturbations: [(&str, &str); 17] = [ + ( + "version", + "UPDATE sandboxes SET updated_at = '2' WHERE id = ?1", + ), + ( + "state", + "UPDATE sandboxes SET state = 'paused' WHERE id = ?1", + ), + ( + "failure reason", + "UPDATE sandboxes SET failure_reason = 'other' WHERE id = ?1", + ), + ( + "created_at", + "UPDATE sandboxes SET created_at = '2' WHERE id = ?1", + ), + ( + "uuid8", + "UPDATE sandboxes SET uuid8 = 'zzzzzzzz' WHERE id = ?1", + ), + ("uid", "UPDATE sandboxes SET uid = 2002 WHERE id = ?1"), + ("slot", "UPDATE sandboxes SET slot = 2 WHERE id = ?1"), + ( + "guest cid", + "UPDATE sandboxes SET guest_cid = 4 WHERE id = ?1", + ), + ("pid", "UPDATE sandboxes SET pid = 4300 WHERE id = ?1"), + ( + "spaced-host pid", + "UPDATE sandboxes SET spaced_host_pid = 4301 WHERE id = ?1", + ), + ( + "gateway pid", + "UPDATE sandboxes SET netgtw_pid = 4302 WHERE id = ?1", + ), + ( + "vsock path", + "UPDATE sandboxes SET vsock_uds_path = 'other.sock' WHERE id = ?1", + ), + ( + "api socket", + "UPDATE sandboxes SET api_socket_path = 'other.sock' WHERE id = ?1", + ), + ( + "route identity", + "UPDATE sandboxes SET space_api_route_execution_generation = 8 WHERE id = ?1", + ), + ( + "process environment", + "UPDATE sandboxes SET process_env_json = '{\"NEW\":\"1\"}' WHERE id = ?1", + ), + ( + "cleanup completion", + "UPDATE sandboxes SET cleanup_completion = 'mark_stopped' WHERE id = ?1", + ), + ( + "cleanup keep", + "UPDATE sandboxes SET cleanup_keep_filesystem = 1 WHERE id = ?1", + ), + ]; + for (case, perturbation) in perturbations { + let id = insert_resident(&conn, SandboxState::Started); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + conn.execute(perturbation, params![id]).unwrap(); + refuses( + case, + &conn, + || settle_startup_resident(&conn, &observed, settlement(SandboxState::Quarantined)), + |result| matches!(result, Ok(false)), + ); + } + + // A resident that pins an artifact, so each pin column can change alone. + for (case, perturbation) in [ + ( + "artifact id", + "UPDATE sandboxes SET current_local_artifact_id = '019a0000-0000-7000-8000-0000000000d3' WHERE id = ?1", + ), + ( + "artifact kind", + "UPDATE sandboxes SET current_local_artifact_kind = 'disk' WHERE id = ?1", + ), + ( + "artifact origin", + "UPDATE sandboxes SET current_local_artifact_origin = 'stop' WHERE id = ?1", + ), + ] { + let id = insert_resident(&conn, SandboxState::Started); + conn.execute( + "UPDATE sandboxes + SET current_local_artifact_id = '019a0000-0000-7000-8000-0000000000d1', + current_local_artifact_kind = 'full', + current_local_artifact_origin = 'suspend' + WHERE id = ?1", + params![id], + ) + .unwrap(); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + conn.execute(perturbation, params![id]).unwrap(); + refuses( + case, + &conn, + || settle_startup_resident(&conn, &observed, settlement(SandboxState::Quarantined)), + |result| matches!(result, Ok(false)), + ); + } + + let id = insert_resident(&conn, SandboxState::Started); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + insert_cleanup_episode( + &conn, + &id, + CleanupEpisodeSpec { + id: SandboxCleanupEpisodeId::new(), + completion: CleanupCompletion::MarkStopped, + keep_filesystem: true, + entered_from_state: Some(SandboxState::Stopped), + activation_state: SandboxState::Starting, + execution_generation: None, + activation_attempt_id: None, + }, + ) + .unwrap(); + // The episode insert also wrote its intent; restore the observed columns + // so only the episode differs. + conn.execute( + "UPDATE sandboxes SET cleanup_completion = NULL, cleanup_keep_filesystem = NULL WHERE id = ?1", + params![id], + ) + .unwrap(); + refuses( + "cleanup episode", + &conn, + || settle_startup_resident(&conn, &observed, settlement(SandboxState::Quarantined)), + |result| matches!(result, Ok(false)), + ); + + let id = insert_resident(&conn, SandboxState::Started); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + let proof = endpoint_proof(id.parse().unwrap()); + conn.execute( + "UPDATE sandboxes SET spaced_endpoint_proof_json = ?2 WHERE id = ?1", + params![id, serde_json::to_string(&proof).unwrap()], + ) + .unwrap(); + refuses( + "endpoint proof", + &conn, + || settle_startup_resident(&conn, &observed, settlement(SandboxState::Quarantined)), + |result| matches!(result, Ok(false)), + ); + + let id = insert_resident(&conn, SandboxState::Started); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + assert!( + insert_restore_transfer_scope( + &conn, + &id, + TransferScopeId::new(), + None, + None, + &format!("/cache/restore-staging/{id}"), + ) + .unwrap() + ); + refuses( + "restore transfer scope", + &conn, + || settle_startup_resident(&conn, &observed, settlement(SandboxState::Quarantined)), + |result| matches!(result, Ok(false)), + ); + + let id = insert_resident(&conn, SandboxState::Started); + let observed = get_sandbox(&conn, &id).unwrap().unwrap(); + conn.execute("DELETE FROM sandboxes WHERE id = ?1", params![id]) + .unwrap(); + refuses( + "missing row", + &conn, + || settle_startup_resident(&conn, &observed, settlement(SandboxState::Started)), + |result| matches!(result, Ok(false)), + ); +} + +/// Adopt or insert `id` as the restored sandbox's own row in `state`. +fn restore_into( + conn: &Connection, + id: &str, + state: SandboxState, +) -> Result<(), RestoreAdoptOrphanInsertError> { + let source = restore_source(); + with_sandbox_params( + id, + state, + |params| { + params.runtime = RuntimeKind::Firecracker.alias(); + params.snapshots_enabled = true; + params.rootfs_size_mb = Some(2048); + }, + |params| { + restore_adopt_orphan_or_insert_sandbox( + conn, + params, + &RemoteRestoreOperation::metadata_staging(source), + ) + }, + ) +} + +/// The synthetic quarantined row the orphan reconciler materializes, with the +/// timestamps adoption replaces backdated. +fn insert_synthetic_orphan(conn: &Connection) -> String { + let id = SandboxId::new().to_string(); + with_sandbox_params( + &id, + SandboxState::Quarantined, + as_synthetic_orphan, + |params| { + insert_quarantined_sandbox_in_transaction( + conn, + params, + SandboxQuarantineSource::SbxManager, + SandboxQuarantineReasonCode::OrphanedSandboxRowMissing, + "orphaned cache", + SandboxQuarantineNextAction::RequiresOperatorAttention, + ) + }, + ) + .unwrap(); + backdate(conn, &id); + conn.execute( + "UPDATE sandboxes SET created_at = '1' WHERE id = ?1", + params![id], + ) + .unwrap(); + id +} + +#[test] +fn orphan_restore_adoption_rewrites_only_a_synthetic_orphan() { + let conn = test_conn(); + let id = insert_synthetic_orphan(&conn); + // The quarantine episode stays for automatic recovery to retire, and the + // orphan's node execution authority is kept. + assert_accepts( + "adopt the quarantined orphan", + &conn, + || restore_into(&conn, &id, SandboxState::Restoring).unwrap(), + ExpectedDiff::new().sandbox( + &id, + [ + ("state", text("restoring")), + ("template", text("seed")), + ("template_hash", text("hash")), + ("user", text("user")), + ("snapshots_enabled", int(1)), + ("rootfs_size_mb", int(2048)), + ("failure_reason", NULL), + ("restore_origin", text("remote")), + ("restore_phase", text("metadata_staging")), + ("restore_storage_uri", text(RESTORE_STORAGE_URI)), + ("created_at", Cell::NowText), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ), + ); + + let conn = test_conn(); + for state in SandboxState::all() { + if state == SandboxState::Restoring { + continue; + } + let id = insert_synthetic_orphan(&conn); + refuses( + &format!("insert as {state}"), + &conn, + || restore_into(&conn, &id, state), + |result| { + matches!( + result, + Err(RestoreAdoptOrphanInsertError::NonRestoringState { .. }) + ) + }, + ); + } + + let mut perturbations: Vec<(String, String)> = SandboxState::all() + .into_iter() + .filter(|state| !matches!(state, SandboxState::Quarantined | SandboxState::Restoring)) + .map(|state| { + ( + format!("orphan in {state}"), + format!( + "UPDATE sandboxes SET state = '{}' WHERE id = ?1", + state.as_str() + ), + ) + }) + .collect(); + for (case, sql) in [ + ( + "no quarantine marker", + "DELETE FROM sandbox_quarantine_state WHERE sandbox_id = ?1", + ), + ( + "another quarantine source", + "UPDATE sandbox_quarantine_state SET source = 'operator' WHERE sandbox_id = ?1", + ), + ( + "another quarantine reason", + "UPDATE sandbox_quarantine_state SET reason_code = 'pause_failed' WHERE sandbox_id = ?1", + ), + ( + "real template", + "UPDATE sandboxes SET template = 'seed' WHERE id = ?1", + ), + ( + "template hash", + "UPDATE sandboxes SET template_hash = 'hash' WHERE id = ?1", + ), + ( + "real owner", + "UPDATE sandboxes SET user = 'user' WHERE id = ?1", + ), + ( + "another runtime", + "UPDATE sandboxes SET runtime = 'other' WHERE id = ?1", + ), + ("uid", "UPDATE sandboxes SET uid = 2001 WHERE id = ?1"), + ("slot", "UPDATE sandboxes SET slot = 1 WHERE id = ?1"), + ( + "guest cid", + "UPDATE sandboxes SET guest_cid = 3 WHERE id = ?1", + ), + ("pid", "UPDATE sandboxes SET pid = 4242 WHERE id = ?1"), + ( + "spaced-host pid", + "UPDATE sandboxes SET spaced_host_pid = 4243 WHERE id = ?1", + ), + ( + "gateway pid", + "UPDATE sandboxes SET netgtw_pid = 4244 WHERE id = ?1", + ), + ( + "vsock path", + "UPDATE sandboxes SET vsock_uds_path = 'vsock.sock' WHERE id = ?1", + ), + ( + "api socket", + "UPDATE sandboxes SET api_socket_path = 'api.sock' WHERE id = ?1", + ), + ( + "snapshots enabled", + "UPDATE sandboxes SET snapshots_enabled = 1 WHERE id = ?1", + ), + ( + "pool template", + "UPDATE sandboxes SET pool_template = 'pool' WHERE id = ?1", + ), + ( + "guest uptime", + "UPDATE sandboxes SET guest_uptime_secs = 9 WHERE id = ?1", + ), + ( + "process environment", + "UPDATE sandboxes SET process_env_json = '{\"A\":\"1\"}' WHERE id = ?1", + ), + ( + "auto-lifecycle policy", + "UPDATE sandboxes SET auto_lifecycle_json = '{\"wake\":{\"on\":[]}}' WHERE id = ?1", + ), + ( + "cleanup completion", + "UPDATE sandboxes SET cleanup_completion = 'delete_row' WHERE id = ?1", + ), + ( + "cleanup keep", + "UPDATE sandboxes SET cleanup_keep_filesystem = 0 WHERE id = ?1", + ), + ( + "applied network policy generation", + "UPDATE sandboxes SET applied_network_policy_generation = 1 WHERE id = ?1", + ), + ( + "applied network policy source", + "UPDATE sandboxes SET applied_network_policy_source = 'cluster' WHERE id = ?1", + ), + ( + "local network policy authority", + "UPDATE sandboxes SET network_policy_authority = 'local' WHERE id = ?1", + ), + ( + "local auto-lifecycle authority", + "UPDATE sandboxes SET auto_lifecycle_authority = 'local' WHERE id = ?1", + ), + ] { + perturbations.push((case.to_owned(), sql.to_owned())); + } + for (case, sql) in perturbations { + let id = insert_synthetic_orphan(&conn); + conn.execute(&sql, params![id]).unwrap(); + refuses( + &case, + &conn, + || restore_into(&conn, &id, SandboxState::Restoring), + |result| { + matches!( + result, + Err(RestoreAdoptOrphanInsertError::ExistingRowNotSyntheticOrphan { .. }) + ) + }, + ); + } +} + +fn adopt( + conn: &Connection, + claimed: &crate::db::ClaimedQuarantineRecoveryCandidate, + id: &str, +) -> QuarantineAdoptionOutcome { + adopt_quarantine_cleanup_episode( + conn, + claimed, + adoptable(claimed), + generation(7), + SandboxCleanupEpisodeId::new(), + &execution_metadata(), + &snapshot_path(id), + ) + .unwrap() +} + +#[test] +fn quarantine_adoption_binds_the_quarantined_full_suspend_pin_and_generation() { + let message = "automatic recovery adopted validated suspend evidence"; + for generation_seeded in [true, false] { + let conn = test_conn(); + let (id, artifact, claimed) = insert_claimed_adoptable(&conn); + if !generation_seeded { + conn.execute( + "UPDATE sandboxes SET space_api_route_execution_generation = NULL WHERE id = ?1", + params![id], + ) + .unwrap(); + } + backdate(&conn, &id); + conn.execute( + "UPDATE local_artifacts SET last_touched_at = '1' WHERE sandbox_id = ?1", + params![id], + ) + .unwrap(); + conn.execute( + "UPDATE sandbox_quarantine_state SET updated_at = '1' WHERE sandbox_id = ?1", + params![id], + ) + .unwrap(); + let mut cells = vec![ + ("state", text("cleanup_required")), + ("cleanup_completion", text("mark_suspended")), + ("cleanup_keep_filesystem", int(1)), + ("failure_reason", text(message)), + ("execution_metadata_json", Cell::Fresh), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ]; + if !generation_seeded { + cells.push(("space_api_route_execution_generation", int(7))); + } + let cleanup_id = SandboxCleanupEpisodeId::new(); + let reason = SandboxQuarantineReasonCode::RestoreFirecrackerLaunchFailed; + assert_accepts( + &format!("generation seeded {generation_seeded}"), + &conn, + || { + assert_eq!( + adopt_quarantine_cleanup_episode( + &conn, + &claimed, + adoptable(&claimed), + generation(7), + cleanup_id, + &execution_metadata(), + &snapshot_path(&id), + ) + .unwrap(), + QuarantineAdoptionOutcome::Adopted + ) + }, + ExpectedDiff::new() + .sandbox(&id, cells) + .updated( + "local_artifacts", + ("local_artifact_id", &artifact.to_string()), + [("last_touched_at", Cell::NowText)], + ) + .updated( + "sandbox_quarantine_state", + ("sandbox_id", &id), + [ + ("claim_token", NULL), + ("claim_expires_at", NULL), + ("updated_at", Cell::NowText), + ], + ) + .inserted( + "sandbox_cleanup_episodes", + vec![ + ("sandbox_id", text(&id)), + ("cleanup_id", text(cleanup_id.to_string())), + ("completion", text("mark_suspended")), + ("keep_filesystem", int(1)), + ("phase", text("requested")), + ("entered_from_state", text("quarantined")), + ("activation_state", text("restoring")), + ("requested_at", Cell::NowInteger), + ("quarantine_reason_code", text(reason.as_str())), + ("failure_message", text(message)), + ("last_cleanup_error", NULL), + ("execution_generation", NULL), + ("activation_attempt_id", NULL), + ("attempt_count", int(0)), + ("created_at", Cell::NowInteger), + ("recovery_local_artifact_id", text(artifact.to_string())), + ("recovery_execution_generation", int(7)), + ("prepared_clone_root", NULL), + ("operation_activation_attempt_id", NULL), + ], + ), + ); + } + + // The claim's fence conjuncts are swept by + // `every_claim_fenced_statement_pins_every_conjunct`; these are the + // statement's own sandbox predicates. Every other source state keeps the + // rest of the shape, and no restore columns, so only the state guard can + // refuse it. + let conn = test_conn(); + for state in SandboxState::all() { + if state == SandboxState::Quarantined { + continue; + } + let (id, _, claimed) = insert_claimed_adoptable(&conn); + conn.execute( + "UPDATE sandboxes SET state = ?2 WHERE id = ?1", + params![id, state.as_str()], + ) + .unwrap(); + refuses( + state.as_str(), + &conn, + || adopt(&conn, &claimed, &id), + |outcome| *outcome == QuarantineAdoptionOutcome::FenceChanged, + ); + } + for (case, sql) in [ + ( + "another pin", + "UPDATE sandboxes SET current_local_artifact_id = '019a0000-0000-7000-8000-0000000000d2' WHERE id = ?1", + ), + ( + "a disk pin", + "UPDATE sandboxes SET current_local_artifact_kind = 'disk' WHERE id = ?1", + ), + ( + "a stop pin", + "UPDATE sandboxes SET current_local_artifact_origin = 'stop' WHERE id = ?1", + ), + ( + "another generation", + "UPDATE sandboxes SET space_api_route_execution_generation = 8 WHERE id = ?1", + ), + ] { + let (id, _, claimed) = insert_claimed_adoptable(&conn); + conn.execute(sql, params![id]).unwrap(); + refuses( + case, + &conn, + || adopt(&conn, &claimed, &id), + |outcome| *outcome == QuarantineAdoptionOutcome::FenceChanged, + ); + } +} + +#[test] +fn suspended_cleanup_pin_reuses_a_staged_identity_and_keeps_its_backup_marker() { + let conn = test_conn(); + let id = insert_row(&conn, SandboxState::Paused); + begin_teardown_suspend(&conn, &id, generation(7)).unwrap(); + let staged = LocalArtifactId::new(); + stage_teardown_suspend_artifact(&conn, &id, generation(7), staged, "/cache/staged/snapshot") + .unwrap(); + mark_local_artifact_backup_completed(&conn, &staged.to_string()).unwrap(); + set_failure_reason(&conn, &id, "seeded failure"); + seed_receipt(&conn, &id, SandboxState::Started); + backdate(&conn, &id); + conn.execute( + "UPDATE local_artifacts SET last_touched_at = '1' WHERE local_artifact_id = ?1", + params![staged.to_string()], + ) + .unwrap(); + assert_accepts( + "staged identity", + &conn, + || pin_suspended_artifact_for_cleanup(&conn, &id, staged, &snapshot_path(&id)).unwrap(), + ExpectedDiff::new() + .sandbox( + &id, + [ + ("state", text("cleanup_required")), + ("cleanup_completion", text("mark_suspended")), + ("cleanup_keep_filesystem", int(1)), + ("failure_reason", NULL), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ], + ) + .updated( + "local_artifacts", + ("local_artifact_id", &staged.to_string()), + [ + ("local_path", text(snapshot_path(&id))), + ("last_touched_at", Cell::NowText), + ], + ), + ); +} 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 new file mode 100644 index 0000000000..a49d7940ad --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/db/tests/update_state_matrix.rs @@ -0,0 +1,440 @@ +//! Every outcome of the generic node-local SQLite `sandboxes.state` writer, +//! `update_state`, over all 18×18 `(from, to)` lifecycle pairs and each +//! cleanup-ownership shape that changes its answer, once with a transition +//! receipt for the target and once with a receipt for another state. +//! +//! The expected outcomes are written out below as one grid per shape, not +//! derived from `SandboxState::can_transition_to`: `update_state` deliberately +//! has no expected-state predicate, so the grids record which writes it applies +//! today, including edges the state machine forbids. + +use pplx_space_common_core::types::{ + SandboxCleanupEpisodeId, SandboxId, SandboxQuarantineReasonCode, +}; +use rusqlite::Connection; + +use super::state_write_fixtures::{ + Cell, ExpectedDiff, NULL, Snapshot, assert_diff, assert_unchanged, backdate, force_state, + generation, insert_row, receipt_cleared, seed_receipt, set_failure_reason, snapshot, + test_start, text, +}; +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, +}; +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 { + 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, + } +} + +/// `A` applied, `Q` blocked by quarantine, `S` superseded by cleanup +/// ownership, `X` rejected before reading the row (`InvalidQuery`). +type Grid = [&'static str; 18]; + +/// No cleanup intent and no cleanup episode. +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", + /* created */ "A A A A A A A A A A A A A A A A X X", + /* starting */ "A A A A A A A A A A A A A A A A X X", + /* started */ "A A A A A A A A A A A A A A A A X X", + /* stopping */ "A A A S A A A S S S S A A A A A X X", + /* stopped */ "A A A A A A A A A A A A A A A A X X", + /* deleting */ "S S S S S S A S S S S S S S S S X X", + /* pausing */ "A A A A A A A A A A A A A A A A X X", + /* paused */ "A A A A A A A A A A A A A A A A X X", + /* resuming */ "A A A A A A A A A A A A A A A A X X", + /* checkpt */ "A A A A A A A A A A A A A A A A X X", + /* susping */ "A A A A A A A A A A A A A A A A X X", + /* susped */ "A A A A A A A A A A A A A A A A X X", + /* restorng */ "A A A A A A A A A A A A A A A A X X", + /* clean_rq */ "A A A S A A A S S S S A A A A A X X", + /* clean_fl */ "A A A S A A A S S S S A A A A A X X", + /* quarant */ "Q Q Q Q Q Q A Q Q Q Q Q Q Q Q Q X X", + /* failed */ "A A A A A A A A A A A A A A A A X X", +]; + +/// A `delete_row` row intent with no episode. The armed create episode, which +/// also owns a `delete_row` intent, deliberately answers identically: create +/// completion relies on writing through it. +const DELETE_ROW_INTENT: 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", + /* created */ "A A A A A A A A A A A A A A A A X X", + /* starting */ "A A A A A A A A A A A A A A A A X X", + /* started */ "A A A A A A A A A A A A A A A A X X", + /* stopping */ "A A A S A A A S S S S A A A A A X X", + /* stopped */ "A A A A A A A A A A A A A A A A X X", + /* deleting */ "S S S S S S A S S S S S S S S S X X", + /* pausing */ "A A A A A A A A A A A A A A A A X X", + /* paused */ "A A A A A A A A A A A A A A A A X X", + /* resuming */ "A A A A A A A A A A A A A A A A X X", + /* checkpt */ "A A A A A A A A A A A A A A A A X X", + /* susping */ "A A A A A A A A A A A A A A A A X X", + /* susped */ "A A A A A A A A A A A A A A A A X X", + /* restorng */ "A A A A A A A A A A A A A A A A X X", + /* clean_rq */ "S S S S S S A S S S S S S S S S X X", + /* clean_fl */ "S S S S S S A S S S S S S S S S X X", + /* quarant */ "Q Q Q Q Q Q A Q Q Q Q Q Q Q Q Q X X", + /* failed */ "A A A A A A A A A A A A A A A A X X", +]; + +/// An episode that does not delete the row (an armed cold-start `mark_stopped` +/// episode) or that has been requested: cleanup owns every target, deletion +/// included, except that quarantine answers first. +const CLEANUP_OWNED: Grid = [ + // to: cre crd stg std stp spd del pau psd res chk ssg ssd rst crq crf qua fai + /* creating */ + "S S S S S S S S S S S S S S S S X X", + /* created */ "S S S S S S S S S S S S S S S S X X", + /* starting */ "S S S S S S S S S S S S S S S S X X", + /* started */ "S S S S S S S S S S S S S S S S X X", + /* stopping */ "S S S S S S S S S S S S S S S S X X", + /* stopped */ "S S S S S S S S S S S S S S S S X X", + /* deleting */ "S S S S S S S S S S S S S S S S X X", + /* pausing */ "S S S S S S S S S S S S S S S S X X", + /* paused */ "S S S S S S S S S S S S S S S S X X", + /* resuming */ "S S S S S S S S S S S S S S S S X X", + /* checkpt */ "S S S S S S S S S S S S S S S S X X", + /* susping */ "S S S S S S S S S S S S S S S S X X", + /* susped */ "S S S S S S S S S S S S S S S S X X", + /* restorng */ "S S S S S S S S S S S S S S S S X X", + /* clean_rq */ "S S S S S S S S S S S S S S S S X X", + /* clean_fl */ "S S S S S S S S S S S S S S S S X X", + /* quarant */ "Q Q Q Q Q Q S Q Q Q Q Q Q Q Q Q X X", + /* failed */ "S S S S S S S S S S S S S S S S X X", +]; + +/// A row the production quarantine writer moved to `quarantined`, with its +/// `sandbox_quarantine_*` episode: only deletion passes. +const QUARANTINE_EPISODE: &str = + "Q Q Q Q Q Q A Q Q Q Q Q Q Q Q Q X X"; + +/// No row at all. +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 { + Applied, + Quarantine, + Superseded, + Missing, + Rejected, +} + +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)] { + "A" => Expected::Applied, + "Q" => Expected::Quarantine, + "S" => Expected::Superseded, + "M" => Expected::Missing, + "X" => Expected::Rejected, + other => panic!("unknown grid cell {other:?}"), + } +} + +#[derive(Clone, Copy, Debug)] +enum Shape { + Plain, + DeleteRowIntent, + ArmedCreateEpisode, + ArmedActivationEpisode, + RequestedCreateEpisode, +} + +impl Shape { + const ALL: [Self; 5] = [ + Self::Plain, + Self::DeleteRowIntent, + Self::ArmedCreateEpisode, + Self::ArmedActivationEpisode, + Self::RequestedCreateEpisode, + ]; + + fn grid(self) -> &'static Grid { + match self { + Self::Plain => &PLAIN, + Self::DeleteRowIntent | Self::ArmedCreateEpisode => &DELETE_ROW_INTENT, + Self::ArmedActivationEpisode | Self::RequestedCreateEpisode => &CLEANUP_OWNED, + } + } + + /// The cleanup intent and episode, installed while the row is still + /// `suspending` so the production writers that create them accept it. + fn install(self, conn: &Connection, id: &str) { + let create_episode = CleanupEpisodeSpec { + id: SandboxCleanupEpisodeId::new(), + completion: CleanupCompletion::DeleteRow, + keep_filesystem: false, + entered_from_state: None, + activation_state: SandboxState::Creating, + execution_generation: None, + activation_attempt_id: None, + }; + match self { + Self::Plain => {} + Self::DeleteRowIntent => { + assert!(set_cleanup_intent(conn, id, false, CleanupCompletion::DeleteRow).unwrap()); + } + Self::ArmedCreateEpisode => insert_cleanup_episode(conn, id, create_episode).unwrap(), + Self::ArmedActivationEpisode => insert_cleanup_episode( + conn, + id, + CleanupEpisodeSpec { + id: SandboxCleanupEpisodeId::new(), + completion: CleanupCompletion::MarkStopped, + keep_filesystem: true, + entered_from_state: Some(SandboxState::Stopped), + activation_state: SandboxState::Starting, + execution_generation: None, + activation_attempt_id: None, + }, + ) + .unwrap(), + Self::RequestedCreateEpisode => { + let cleanup_id = create_episode.id; + insert_cleanup_episode(conn, id, create_episode).unwrap(); + assert_eq!( + request_cleanup_episode( + conn, + id, + cleanup_id, + SandboxQuarantineReasonCode::CreateInterrupted, + "create interrupted", + ) + .unwrap(), + CleanupEpisodeRequestOutcome::Requested + ); + } + } + } +} + +/// Which transition receipt a seeded row carries relative to the target. +#[derive(Clone, Copy, Debug)] +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, + /// A receipt for another receipt-capable state, which an applied write + /// clears. + ForAnotherState, +} + +impl Receipt { + const ALL: [Self; 2] = [Self::ForTarget, Self::ForAnotherState]; + + fn target(self, to: SandboxState) -> SandboxState { + let receipt_capable = match to { + SandboxState::Created + | SandboxState::Started + | SandboxState::Paused + | SandboxState::Stopped + | SandboxState::Suspended => true, + SandboxState::Creating + | SandboxState::Starting + | SandboxState::Stopping + | SandboxState::Deleting + | SandboxState::Pausing + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Quarantined + | SandboxState::Failed => false, + }; + match self { + Self::ForTarget if receipt_capable => to, + Self::ForTarget | Self::ForAnotherState if to != SandboxState::Started => { + SandboxState::Started + } + Self::ForTarget | Self::ForAnotherState => SandboxState::Stopped, + } + } +} + +/// A row in `from` carrying `shape`, an active teardown-suspend job (which only +/// 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 { + let id = insert_row(conn, SandboxState::Paused); + begin_teardown_suspend(conn, &id, generation(7)).unwrap(); + shape.install(conn, &id); + force_state(conn, &id, from); + set_failure_reason(conn, &id, "seeded failure"); + seed_receipt(conn, &id, receipt); + backdate(conn, &id); + id +} + +/// The row diff an applied `update_state(to)` makes to a [`seed`] row whose +/// receipt targets `receipt`. +fn applied_diff( + id: &str, + from: SandboxState, + to: SandboxState, + receipt: SandboxState, +) -> ExpectedDiff { + let mut cells: Vec<(&'static str, Cell)> = + vec![("updated_at", Cell::NowText), ("failure_reason", NULL)]; + if from != to { + cells.push(("state", text(to.as_str()))); + cells.push(("state_changed_at", Cell::NowText)); + } + if from == SandboxState::Restoring && to != SandboxState::Restoring { + cells.push(("restore_origin", NULL)); + cells.push(("restore_phase", NULL)); + cells.push(("restore_storage_uri", NULL)); + } + if receipt != to { + cells.extend(receipt_cleared(receipt)); + } + ExpectedDiff::new().sandbox(id, cells) +} + +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:?}"), + } +} + +/// 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( + case: &str, + conn: &Connection, + id: &str, + to: SandboxState, + expected: Expected, + applied: ExpectedDiff, +) { + let before: Snapshot = snapshot(conn); + let start = test_start(); + let result = update_state(conn, id, to); + assert_outcome(case, result, expected); + if expected == Expected::Applied { + assert_diff(case, &before, &snapshot(conn), start, &applied); + } else { + assert_unchanged(case, &before, conn); + } +} + +#[test] +fn generic_state_write_outcomes_cover_every_lifecycle_pair_and_cleanup_shape() { + for (shape, receipt_kind) in Shape::ALL + .into_iter() + .flat_map(|shape| Receipt::ALL.map(|receipt| (shape, receipt))) + { + for from in SandboxState::all() { + for to in SandboxState::all() { + let receipt = receipt_kind.target(to); + let case = format!("{shape:?}, {receipt} receipt: {from} -> {to}"); + 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); + } + } + } +} + +#[test] +fn generic_state_write_outcomes_over_a_quarantine_episode_and_a_missing_row() { + for to in SandboxState::all() { + let case = format!("quarantine episode: quarantined -> {to}"); + let conn = test_conn(); + let id = insert_row(&conn, SandboxState::Paused); + begin_teardown_suspend(&conn, &id, generation(7)).unwrap(); + assert!(mark_suspending_quarantined(&conn, &id, "suspend publication failed").unwrap()); + backdate(&conn, &id); + // Entering quarantine already failed the teardown job, so a deletion + // changes only the row and keeps the quarantine episode. + let applied = ExpectedDiff::new().sandbox( + &id, + [ + ("state", text("deleting")), + ("state_changed_at", Cell::NowText), + ("updated_at", Cell::NowText), + ("failure_reason", NULL), + ], + ); + check(&case, &conn, &id, to, cell(QUARANTINE_EPISODE, to), applied); + + let case = format!("missing row -> {to}"); + let conn = test_conn(); + let id = SandboxId::new().to_string(); + check( + &case, + &conn, + &id, + to, + cell(MISSING, to), + ExpectedDiff::new(), + ); + } +} + +#[test] +fn every_lifecycle_state_has_one_grid_position() { + let mut seen = [false; 18]; + for state in SandboxState::all() { + assert!(!seen[index(state)], "{state} shares a grid position"); + seen[index(state)] = true; + } + assert!(seen.iter().all(|seen| *seen)); +} diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup.rs b/crates/libs/sbx/mgr/server/src/service/cleanup.rs index 4fa01ef3ee..5daa54ba7b 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup.rs @@ -2875,3 +2875,6 @@ impl BeforeStopFlush { } } } + +#[cfg(test)] +mod stop_admission_tests; diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs b/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs new file mode 100644 index 0000000000..ac71d11df7 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/cleanup/stop_admission_tests.rs @@ -0,0 +1,276 @@ +//! Stop's cleanup admission from every node-local SQLite `sandboxes.state` +//! source state, and the Stop RPC's runtime gate in front of it. +//! +//! Stop admission does not consult `SandboxState::transition`, unlike Kill: it +//! supersedes any activation and writes `stopping` through the generic writer, +//! so every state the generic writer does not refuse is admitted. The +//! expected outcomes are written out here, not derived from that writer. +//! +//! The Stop RPC admits only a sandbox whose runtime is live: it resolves its +//! target with `running_only`, which selects nothing unless the sandbox's +//! `fc.pid` resolves. A `creating`, `created`, `stopped` or `suspended` row has +//! no Firecracker process, so those admissions are reachable only by calling +//! admission directly. Only a non-blocking Stop persists its intent. +//! +//! The admission sweep calls admission directly, because an admitted Stop +//! RPC's watchdog removal stats the fixed sidecar pid path. Node cleanup +//! reaches only the executor's test dispatch, which holds until shutdown. + +use std::sync::Arc; +use std::time::Duration; + +use pplx_space_common_core::types::SandboxId; +use pplx_space_common_svc::PeerIdentity; +use tonic::Request; + +use super::{CleanupAdmission, SandboxManagerService, SandboxManagerServiceOptions, StopRequest}; +use crate::db::CleanupCompletion; +use crate::db::tests::state_write_fixtures::{assert_unchanged, snapshot}; +use crate::service::tests::service_tests::insert_test_sandbox; +use crate::state::SandboxState; + +const WAIT: Duration = Duration::from_secs(30); + +/// A test service whose host dependencies resolve under `scratch`, with node +/// cleanup held by a test dispatch. +fn isolated_service( + conn: rusqlite::Connection, + scratch: &std::path::Path, +) -> SandboxManagerService { + let service = SandboxManagerService::new_with_options( + conn, + SandboxManagerServiceOptions::for_test_in(scratch), + ); + // A dispatch that outlived shutdown would keep its executor task, and the + // tracker, from finishing. + let shutdown = service.shutdown_token.clone(); + *service.cleanup_executor.dispatch.lock().unwrap() = Some(Arc::new(move |_, _| { + let shutdown = shutdown.clone(); + Box::pin(async move { + shutdown.cancelled().await; + Err(tonic::Status::unavailable("test node cleanup shut down")) + }) + })); + service +} + +async fn shutdown(service: SandboxManagerService) { + service.shutdown_token.cancel(); + service.task_tracker.close(); + tokio::time::timeout(WAIT, service.task_tracker.wait()) + .await + .expect("cleanup tasks stop within 30 seconds of shutdown"); +} + +/// A row in `state`, with a local restore's columns when it is `restoring`. +fn insert_row(conn: &rusqlite::Connection, state: SandboxState) -> SandboxId { + let sandbox_id = SandboxId::new(); + let id = sandbox_id.to_string(); + insert_test_sandbox(conn, &id, state); + if state == SandboxState::Restoring { + conn.execute( + "UPDATE sandboxes SET restore_origin = 'local', restore_phase = 'activation' + WHERE id = ?1", + [&id], + ) + .unwrap(); + } + sandbox_id +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Admission { + /// Admitted, and the Stop RPC can reach it for a live runtime. + Admitted, + /// Admitted, but only by calling admission directly: the state has no + /// Firecracker process, so the Stop RPC's runtime gate selects nothing. + AdmittedOnlyDirectly, + /// Refused with `Unavailable`. + Refused, +} + +/// What Stop admission answers for a plain row in `state`. Wildcard-free, so +/// a new lifecycle state does not compile until it is placed. +fn admission(state: SandboxState) -> Admission { + match state { + SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopping + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Failed => Admission::Admitted, + SandboxState::Creating + | SandboxState::Created + | SandboxState::Stopped + | SandboxState::Suspended => Admission::AdmittedOnlyDirectly, + // The generic writer leaves a `deleting` row to its delete and a + // `quarantined` row to quarantine. + SandboxState::Deleting | SandboxState::Quarantined => Admission::Refused, + } +} + +/// Assert one admission's outcome: an admitted Stop leaves the row `stopping` +/// with its intent and cleanup job; a refused one changes nothing. +fn check_admission( + case: &str, + service: &SandboxManagerService, + id: &str, + state: SandboxState, + persist_intent: bool, + result: Result, + before: &crate::db::tests::state_write_fixtures::Snapshot, +) { + let db = service.db_conn().unwrap(); + if admission(state) == Admission::Refused { + let status = match result { + Err(error) => error.into_status(), + Ok(_) => panic!("{case}: expected a refusal"), + }; + assert_eq!( + status.code(), + tonic::Code::Unavailable, + "{case}: {status:?}" + ); + assert_unchanged(case, before, &db); + return; + } + assert!( + matches!(result, Ok(CleanupAdmission::Persisted { .. })), + "{case}: expected a persisted admission" + ); + let row = crate::db::get_sandbox(&db, id).unwrap().unwrap(); + assert_eq!(row.state, SandboxState::Stopping, "{case}"); + assert_eq!( + (row.cleanup_completion, row.cleanup_keep_filesystem), + if persist_intent { + (Some(CleanupCompletion::MarkStopped), Some(true)) + } else { + (None, None) + }, + "{case}" + ); + assert!( + pplx_space_common_jobs::db::find_active_job(&db, crate::db::cleanup_jobs::KIND, id) + .unwrap() + .is_some(), + "{case}: admission persists its cleanup job" + ); +} + +/// Admission never consults the shutdown token, but the cleanup executor it +/// starts does not start once the service is shut down. Each case cancels the +/// token and waits for the service's own tasks to exit before admitting, so +/// neither the executor nor its recovery scan runs: the refused cases' +/// whole-database comparison sees admission's writes alone, and an admitted +/// row stays as admission left it. The empty task tracker after admission +/// checks that. +#[tokio::test(flavor = "current_thread")] +async fn stop_admission_answers_every_source_state() { + if !crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) { + return; + } + // A non-blocking Stop persists its intent; a blocking one does not. + for persist_intent in [true, false] { + for state in SandboxState::all() { + let case = format!( + "{state} ({:?}), persist intent {persist_intent}", + admission(state) + ); + let conn = crate::db::test_conn(); + let sandbox_id = insert_row(&conn, state); + let scratch = tempfile::tempdir().unwrap(); + let service = isolated_service(conn, scratch.path()); + // The service's own background tasks exit on cancellation; once + // they have, any tracked task is one admission started. + service.shutdown_token.cancel(); + tokio::time::timeout(WAIT, async { + while !service.task_tracker.is_empty() { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .unwrap_or_else(|_| panic!("{case}: service tasks outlived shutdown")); + let before = snapshot(&service.db_conn().unwrap()); + + let result = tokio::time::timeout( + WAIT, + service.admit_selected_cleanup( + sandbox_id, + CleanupCompletion::MarkStopped, + false, + persist_intent, + None, + false, + ), + ) + .await + .unwrap_or_else(|_| panic!("{case}: admission waited longer than {WAIT:?}")); + assert!( + service.task_tracker.is_empty(), + "{case}: no cleanup executor started" + ); + + check_admission( + &case, + &service, + &sandbox_id.to_string(), + state, + persist_intent, + result, + &before, + ); + shutdown(service).await; + } + } +} + +/// The Stop RPC selects nothing without a live runtime, whatever the row's +/// state, and admits nothing. A live runtime is admitted, as +/// `stop_selects_only_a_live_runtime_and_both_treat_an_empty_template_filter_as_none` +/// pins. +#[tokio::test(flavor = "current_thread")] +async fn stop_rpc_admits_nothing_without_a_live_runtime() { + if !crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) { + return; + } + for blocking in [true, false] { + for state in SandboxState::all() { + let case = format!("{state}, blocking {blocking}"); + let conn = crate::db::test_conn(); + let sandbox_id = insert_row(&conn, state); + let scratch = tempfile::tempdir().unwrap(); + let service = isolated_service(conn, scratch.path()); + let before = snapshot(&service.db_conn().unwrap()); + let mut request = Request::new(StopRequest { + sandbox_id: sandbox_id.to_string(), + template_filter: None, + blocking, + immediate_kill: false, + }); + request + .extensions_mut() + .insert(PeerIdentity::new(0, 0, None)); + + let response = tokio::time::timeout(WAIT, service.stop(request)) + .await + .unwrap_or_else(|_| panic!("{case}: Stop waited longer than {WAIT:?}")) + .unwrap_or_else(|status| panic!("{case}: {status:?}")); + + assert!(response.into_inner().stopped_ids.is_empty(), "{case}"); + assert_unchanged(&case, &before, &service.db_conn().unwrap()); + shutdown(service).await; + } + } +}