diff --git a/crates/libs/sbx/mgr/server/AGENTS.md b/crates/libs/sbx/mgr/server/AGENTS.md index 7b373b449b..d070989387 100644 --- a/crates/libs/sbx/mgr/server/AGENTS.md +++ b/crates/libs/sbx/mgr/server/AGENTS.md @@ -10,7 +10,11 @@ `can_transition_to`, with a reason, to `ALLOWLIST` in `src/db/tests/state_write_table.rs`. Nothing checks the table against the writers; the `&'static Edge` argument does that at call sites. A refused - write must leave every row unchanged. + write must leave every row unchanged. Map an `update_state_from` outcome to + a status with `require_state_write_from`, or write it best effort with + `update_state_from_best_effort` (`src/service/cleanup_support.rs`); a caller + that answers one case differently matches `state_write_result` and passes + the rest to `into_status`. - Foreign keys are off: register a new sandbox child table in `delete_sandbox_child_rows` (`src/db.rs`). - `src/service/adapter.rs` implements the gRPC trait; most handler bodies live diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs index c7dd8eb663..9183a26e3e 100644 --- a/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs +++ b/crates/libs/sbx/mgr/server/src/service/cleanup_support.rs @@ -234,6 +234,163 @@ pub(super) fn require_state_write( .map_err(|error| Status::aborted(format!("{context}: {error}"))) } +/// [`require_state_write`] for [`crate::db::update_state_from`]. A database +/// error is `Internal`. Every refusal is `Aborted`: a from-state refusal +/// reports the state another transition left the row in, and a generic-guard +/// refusal or a missing row keeps `update_state`'s message. +#[cfg_attr( + not(test), + expect(dead_code, reason = "the update_state_from conversions adopt it") +)] +pub(super) fn require_state_write_from( + result: rusqlite::Result, + context: &'static str, +) -> Result<(), Status> { + let outcome = result.map_err(|error| Status::internal(format!("{context}: {error}")))?; + state_write_result(outcome) + .map(drop) + .map_err(|rejected| rejected.into_status(context)) +} + +/// Why an `update_state_from` write did not apply. A caller that answers one +/// case differently matches it and passes the rest to +/// [`StateWriteRejected::into_status`]. +#[cfg_attr( + not(test), + expect(dead_code, reason = "the update_state_from conversions adopt it") +)] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum StateWriteRejected { + /// Another transition moved the row outside the edge's source states. + FromState { + found: SandboxState, + }, + /// A generic guard refused, as it refuses `update_state`. The guards + /// answer first, so `found` may also be outside the edge's source states. + Guard { + refusal: StateWriteGuard, + found: SandboxState, + }, + Missing, +} + +/// The generic guards `update_state` and `update_state_from` share. +#[cfg_attr( + not(test), + expect(dead_code, reason = "the update_state_from conversions adopt it") +)] +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum StateWriteGuard { + Quarantined, + CleanupOwned, +} + +#[cfg_attr( + not(test), + expect(dead_code, reason = "the update_state_from conversions adopt it") +)] +pub(super) fn state_write_result( + outcome: crate::db::StateWriteOutcome, +) -> Result { + use crate::db::{StateWriteOutcome, StateWriteRefusal}; + match outcome { + StateWriteOutcome::Applied { from } => Ok(from), + StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::FromState, + } => Err(StateWriteRejected::FromState { found }), + StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::Quarantined, + } => Err(StateWriteRejected::Guard { + refusal: StateWriteGuard::Quarantined, + found, + }), + StateWriteOutcome::Refused { + found, + refusal: StateWriteRefusal::CleanupOwned, + } => Err(StateWriteRejected::Guard { + refusal: StateWriteGuard::CleanupOwned, + found, + }), + StateWriteOutcome::Missing => Err(StateWriteRejected::Missing), + } +} + +#[cfg_attr( + not(test), + expect(dead_code, reason = "the update_state_from conversions adopt it") +)] +impl StateWriteRejected { + pub(super) fn into_status(self, context: &str) -> Status { + use crate::db::{SandboxStateWriteNotApplied, SandboxStateWriteOutcome}; + let outcome = match self { + Self::FromState { found } => { + return Status::aborted(format!( + "{context}: sandbox state write was superseded: sandbox is {found}" + )); + } + Self::Guard { + refusal: StateWriteGuard::Quarantined, + .. + } => SandboxStateWriteOutcome::BlockedByQuarantine, + Self::Guard { + refusal: StateWriteGuard::CleanupOwned, + .. + } => SandboxStateWriteOutcome::SupersededByCleanupEpisode, + Self::Missing => SandboxStateWriteOutcome::Missing, + }; + Status::aborted(format!( + "{context}: {}", + SandboxStateWriteNotApplied { outcome } + )) + } +} + +/// `update_state_best_effort` for [`crate::db::update_state_from`]: a refusal +/// or an error is logged and leaves the row as it is, and only an applied +/// write marks the sandbox dirty. `None` is a poisoned connection or a +/// database error. +#[cfg_attr( + not(test), + expect(dead_code, reason = "the update_state_from conversions adopt it") +)] +pub(super) fn update_state_from_best_effort( + db: &Arc>, + sandbox_id: &str, + edge: &'static crate::db::Edge, +) -> Option { + use crate::db::StateWriteOutcome; + let outcome = { + let db = crate::node_state_sync::connection_or_signal_failure(db).ok()?; + crate::db::update_state_from(&db, sandbox_id, edge) + }; + match outcome { + Ok(outcome @ StateWriteOutcome::Applied { .. }) => { + mark_sandbox_dirty(db, sandbox_id); + Some(outcome) + } + Ok(outcome @ (StateWriteOutcome::Refused { .. } | StateWriteOutcome::Missing)) => { + tracing::warn!( + sandbox_id, + state = edge.to().as_str(), + ?outcome, + "sandbox state write was superseded" + ); + Some(outcome) + } + Err(error) => { + tracing::warn!( + sandbox_id, + state = edge.to().as_str(), + %error, + "failed to update sandbox state" + ); + None + } + } +} + #[derive(Clone, Copy, Debug)] pub(super) struct ActivationCleanupSpec { pub(super) cleanup_id: SandboxCleanupEpisodeId, @@ -489,3 +646,6 @@ pub(super) struct CleanupRecoveryTarget { pub(super) completion: CleanupCompletion, pub(super) cleanup_id: Option, } + +#[cfg(test)] +mod tests; diff --git a/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs b/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs new file mode 100644 index 0000000000..5b6db037eb --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs @@ -0,0 +1,418 @@ +//! The statuses [`require_state_write_from`] maps `update_state_from`'s +//! node-local SQLite `sandboxes.state` writes to, and what +//! [`update_state_from_best_effort`] returns and persists. Every outcome is +//! listed by a match without a wildcard, so a new outcome or refusal does not +//! compile until it has a status here. + +use std::sync::{Arc, Mutex}; + +use pplx_space_common_core::types::{ + SandboxId, SandboxQuarantineNextAction, SandboxQuarantineReasonCode, SandboxQuarantineSource, +}; +use tokio_util::sync::CancellationToken; +use tokio_util::task::TaskTracker; +use tonic::Code; + +use super::{ + StateWriteGuard, StateWriteRejected, require_state_write, require_state_write_from, + state_write_result, update_state_from_best_effort, +}; +use crate::db::tests::state_write_fixtures::{assert_unchanged, snapshot}; +use crate::db::{ + Edge, SandboxStateWriteOutcome, StateSet, StateWriteOutcome, StateWriteRefusal, edges, +}; +use crate::service::drain::{CriticalTaskGuard, DrainState}; +use crate::service::tests::service_tests::insert_test_sandbox; +use crate::state::SandboxState; + +const CONTEXT: &str = "complete synthetic transition"; + +/// Every refusal. Each names its successor, so a new refusal does not compile +/// until it joins the chain. +fn refusals() -> Vec { + let next = |refusal| match refusal { + StateWriteRefusal::FromState => Some(StateWriteRefusal::Quarantined), + StateWriteRefusal::Quarantined => Some(StateWriteRefusal::CleanupOwned), + StateWriteRefusal::CleanupOwned => None, + }; + std::iter::successors(Some(StateWriteRefusal::FromState), |refusal| next(*refusal)).collect() +} + +/// The messages `require_state_write` returned for `update_state` before +/// `update_state_from` existed, copied from 8f12bd6: it formats +/// `"{context}: {SandboxStateWriteNotApplied}"` (`service/cleanup_support.rs`) +/// over `SandboxStateWriteOutcome::as_str` (`db/state_transitions.rs`), and +/// `update_state`'s guards answer `BlockedByQuarantine` and +/// `SupersededByCleanupEpisode` there. +const BLOCKED_BY_QUARANTINE: &str = + "complete synthetic transition: sandbox state write was not applied: blocked_by_quarantine"; +const SUPERSEDED_BY_CLEANUP_EPISODE: &str = "complete synthetic transition: sandbox state write \ + was not applied: superseded_by_cleanup_episode"; +const MISSING: &str = "complete synthetic transition: sandbox state write was not applied: missing"; + +fn outcomes() -> Vec { + let mut outcomes = vec![StateWriteOutcome::Missing]; + for state in SandboxState::all() { + outcomes.push(StateWriteOutcome::Applied { from: state }); + for refusal in refusals() { + outcomes.push(StateWriteOutcome::Refused { + found: state, + refusal, + }); + } + } + outcomes +} + +fn expected_rejection(outcome: StateWriteOutcome) -> Result { + match outcome { + StateWriteOutcome::Applied { from } => Ok(from), + StateWriteOutcome::Refused { found, refusal } => Err(match refusal { + StateWriteRefusal::FromState => StateWriteRejected::FromState { found }, + StateWriteRefusal::Quarantined => StateWriteRejected::Guard { + refusal: StateWriteGuard::Quarantined, + found, + }, + StateWriteRefusal::CleanupOwned => StateWriteRejected::Guard { + refusal: StateWriteGuard::CleanupOwned, + found, + }, + }), + StateWriteOutcome::Missing => Err(StateWriteRejected::Missing), + } +} + +fn expected_status(outcome: StateWriteOutcome) -> Result<(), (Code, String)> { + match outcome { + StateWriteOutcome::Applied { .. } => Ok(()), + StateWriteOutcome::Refused { found, refusal } => Err(match refusal { + StateWriteRefusal::FromState => ( + Code::Aborted, + format!("{CONTEXT}: sandbox state write was superseded: sandbox is {found}"), + ), + StateWriteRefusal::Quarantined => (Code::Aborted, BLOCKED_BY_QUARANTINE.to_owned()), + StateWriteRefusal::CleanupOwned => { + (Code::Aborted, SUPERSEDED_BY_CLEANUP_EPISODE.to_owned()) + } + }), + StateWriteOutcome::Missing => Err((Code::Aborted, MISSING.to_owned())), + } +} + +fn code_and_message(result: Result) -> Result { + result.map_err(|status| (status.code(), status.message().to_owned())) +} + +#[test] +fn every_state_write_outcome_has_its_rejection_and_status() { + for outcome in outcomes() { + assert_eq!( + state_write_result(outcome), + expected_rejection(outcome), + "{outcome:?}" + ); + assert_eq!( + code_and_message(require_state_write_from(Ok(outcome), CONTEXT)), + expected_status(outcome), + "{outcome:?}" + ); + if let Err(rejected) = state_write_result(outcome) { + assert_eq!( + Err(( + rejected.into_status(CONTEXT).code(), + rejected.into_status(CONTEXT).message().to_owned() + )), + expected_status(outcome), + "{outcome:?}" + ); + } + } +} + +/// The literals above are still what the `update_state` caller path returns. +#[test] +fn update_state_outcomes_keep_their_statuses() { + for (outcome, expected) in [ + (SandboxStateWriteOutcome::Applied, Ok(())), + ( + SandboxStateWriteOutcome::BlockedByQuarantine, + Err(BLOCKED_BY_QUARANTINE), + ), + ( + SandboxStateWriteOutcome::SupersededByCleanupEpisode, + Err(SUPERSEDED_BY_CLEANUP_EPISODE), + ), + (SandboxStateWriteOutcome::Missing, Err(MISSING)), + ] { + assert_eq!( + code_and_message(require_state_write(Ok(outcome), CONTEXT)), + expected.map_err(|message| (Code::Aborted, message.to_owned())), + "{outcome}" + ); + } +} + +#[test] +fn a_database_error_is_internal_as_for_update_state() { + let from = code_and_message(require_state_write_from( + Err(rusqlite::Error::InvalidQuery), + CONTEXT, + )); + assert_eq!( + from, + Err(( + Code::Internal, + "complete synthetic transition: Query is not read-only".to_owned() + )) + ); + assert_eq!( + from, + code_and_message(require_state_write( + Err(rusqlite::Error::InvalidQuery), + CONTEXT + )) + ); +} + +/// A row each generic guard refuses `target` over, or no row. +struct GuardedRow { + name: &'static str, + seed: fn(&rusqlite::Connection, &str), + target: SandboxState, + rejected: fn(SandboxState) -> StateWriteRejected, + message: &'static str, +} + +const GUARDED_ROWS: [GuardedRow; 4] = [ + GuardedRow { + name: "quarantined row", + seed: |conn, id| { + insert_test_sandbox(conn, id, SandboxState::Started); + crate::db::mark_quarantined( + conn, + id, + SandboxQuarantineSource::SbxManager, + SandboxQuarantineReasonCode::SuspendSnapshotPublicationFailed, + "synthetic snapshot publication failure", + SandboxQuarantineNextAction::RequiresOperatorAttention, + ) + .unwrap(); + }, + target: SandboxState::Started, + rejected: |found| StateWriteRejected::Guard { + refusal: StateWriteGuard::Quarantined, + found, + }, + message: BLOCKED_BY_QUARANTINE, + }, + GuardedRow { + name: "stopping row revived", + seed: |conn, id| insert_test_sandbox(conn, id, SandboxState::Stopping), + target: SandboxState::Started, + rejected: |found| StateWriteRejected::Guard { + refusal: StateWriteGuard::CleanupOwned, + found, + }, + message: SUPERSEDED_BY_CLEANUP_EPISODE, + }, + GuardedRow { + name: "cleanup row with a delete_row intent", + seed: |conn, id| { + insert_test_sandbox(conn, id, SandboxState::Started); + crate::db::mark_cleanup_required( + conn, + id, + false, + crate::db::CleanupCompletion::DeleteRow, + "synthetic cleanup", + ) + .unwrap(); + }, + target: SandboxState::Stopped, + rejected: |found| StateWriteRejected::Guard { + refusal: StateWriteGuard::CleanupOwned, + found, + }, + message: SUPERSEDED_BY_CLEANUP_EPISODE, + }, + GuardedRow { + name: "missing row", + seed: |_, _| {}, + target: SandboxState::Started, + rejected: |_| StateWriteRejected::Missing, + message: MISSING, + }, +]; + +/// For the same row, a guard refusal or a missing row maps to the status the +/// `update_state` caller path returned, even with the row's state expected. +#[test] +fn guard_refusals_and_a_missing_row_keep_update_state_statuses() { + for row in GUARDED_ROWS { + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + (row.seed)(&conn, &id); + let found = crate::db::get_sandbox(&conn, &id) + .unwrap() + .map(|sandbox| sandbox.state); + let expected = found.unwrap_or(SandboxState::Started); + let before = snapshot(&conn); + + let outcome = + crate::db::update_state_from_states(&conn, &id, StateSet::of(&[expected]), row.target) + .unwrap(); + assert_eq!( + state_write_result(outcome), + Err((row.rejected)(expected)), + "{}", + row.name + ); + let from = code_and_message(require_state_write_from(Ok(outcome), CONTEXT)); + let generic = code_and_message(require_state_write( + crate::db::update_state(&conn, &id, row.target), + CONTEXT, + )); + assert_eq!( + from, + Err((Code::Aborted, row.message.to_owned())), + "{}", + row.name + ); + assert_eq!(from, generic, "{}", row.name); + assert_unchanged(row.name, &before, &conn); + } +} + +fn dirty_sandboxes(conn: &rusqlite::Connection) -> Vec { + crate::db::list_node_dirty_refs(conn, 100) + .unwrap() + .into_iter() + .filter(|dirty| dirty.element_kind == "sandbox_inventory") + .map(|dirty| dirty.element_value) + .collect() +} + +#[test] +fn best_effort_write_applies_and_marks_only_that_sandbox_dirty() { + let db = Arc::new(Mutex::new(crate::db::test_conn())); + let id = SandboxId::new().to_string(); + insert_test_sandbox(&db.lock().unwrap(), &id, SandboxState::Pausing); + let before = dirty_sandboxes(&db.lock().unwrap()); + + assert_eq!( + update_state_from_best_effort(&db, &id, &edges::PAUSE_ROLLBACK), + Some(StateWriteOutcome::Applied { + from: SandboxState::Pausing + }) + ); + let conn = db.lock().unwrap(); + assert_eq!( + crate::db::get_sandbox(&conn, &id).unwrap().unwrap().state, + SandboxState::Started + ); + assert!(!before.contains(&id)); + assert_eq!(dirty_sandboxes(&conn), [before, vec![id]].concat()); +} + +/// A refused write returns its outcome and changes nothing, so the sandbox is +/// not marked dirty. +#[test] +fn best_effort_refusals_change_nothing() { + for (name, state, outcome) in [ + ( + "row outside the source states", + Some(SandboxState::Stopped), + StateWriteOutcome::Refused { + found: SandboxState::Stopped, + refusal: StateWriteRefusal::FromState, + }, + ), + ( + "cleanup-owned row", + Some(SandboxState::Stopping), + StateWriteOutcome::Refused { + found: SandboxState::Stopping, + refusal: StateWriteRefusal::CleanupOwned, + }, + ), + ("missing row", None, StateWriteOutcome::Missing), + ] { + let db = Arc::new(Mutex::new(crate::db::test_conn())); + let id = SandboxId::new().to_string(); + if let Some(state) = state { + insert_test_sandbox(&db.lock().unwrap(), &id, state); + } + let before = snapshot(&db.lock().unwrap()); + + assert_eq!( + update_state_from_best_effort(&db, &id, &edges::PAUSE_ROLLBACK), + Some(outcome), + "{name}" + ); + assert_unchanged(name, &before, &db.lock().unwrap()); + } +} + +#[test] +fn best_effort_database_error_is_none_and_changes_nothing() { + let edge: &'static Edge = &edges::TEMPLATE_SNAPSHOT_COMPLETE_AFTER_STOP; + let db = Arc::new(Mutex::new(crate::db::test_conn())); + let id = SandboxId::new().to_string(); + insert_test_sandbox(&db.lock().unwrap(), &id, SandboxState::Stopped); + let before = snapshot(&db.lock().unwrap()); + + assert_eq!(update_state_from_best_effort(&db, &id, edge), None); + assert_unchanged("known-race edge", &before, &db.lock().unwrap()); +} + +/// A poisoned connection is `None`, changes nothing and signals the failure: +/// the dirty-outbox worker wakes, sees the poisoned connection and exits, +/// which fails its critical-task guard. The dirty-outbox and poison `Notify`s +/// are process-global statics, so this relies on nextest running each test in +/// its own process, as the poisoned-outbox test in `service/reconcile/tests.rs` +/// does. +#[tokio::test(start_paused = true)] +async fn best_effort_write_over_a_poisoned_connection_signals_the_failure() { + let db = Arc::new(Mutex::new(crate::db::test_conn())); + let id = SandboxId::new().to_string(); + insert_test_sandbox(&db.lock().unwrap(), &id, SandboxState::Pausing); + let before = snapshot(&db.lock().unwrap()); + let shutdown = CancellationToken::new(); + let worker_failed = CancellationToken::new(); + let tracker = TaskTracker::new(); + crate::node_state_sync::start_dirty_outbox_worker( + Arc::clone(&db), + shutdown.clone(), + &tracker, + CriticalTaskGuard::new( + "node dirty outbox", + shutdown.clone(), + worker_failed.clone(), + Arc::new(DrainState::default()), + ), + ); + tokio::task::yield_now().await; + let poisoner = Arc::clone(&db); + std::thread::spawn(move || { + let _conn = poisoner.lock().unwrap(); + panic!("synthetic panic while holding the database"); + }) + .join() + .unwrap_err(); + + assert_eq!( + update_state_from_best_effort(&db, &id, &edges::PAUSE_ROLLBACK), + None + ); + // The clock is paused and this task only yields, so the worker's poll + // timer cannot fire: only the failure signal wakes it. + for _ in 0..100 { + if worker_failed.is_cancelled() { + break; + } + tokio::task::yield_now().await; + } + assert!(worker_failed.is_cancelled()); + let conn = db.lock().unwrap_or_else(std::sync::PoisonError::into_inner); + assert_unchanged("poisoned connection", &before, &conn); +}