From bf43fd5b1b773a1e2730687c9dd181a3a2932a5d Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Tue, 13 Oct 2026 23:57:57 +0000 Subject: [PATCH] sbx-mgr: one refusal-to-status mapping for update_state_from Each update_state_from conversion wrote its own StateWriteOutcome-to-Status helper, with different names, message text and policies. Add one shared mapping and best-effort writer in service/cleanup_support.rs, with no production callers; the conversions adopt them and delete their copies. - require_state_write_from(result, context) -> Result<(), Status>. A from-state refusal is Aborted "{context}: sandbox state write was superseded: sandbox is {found}". A quarantine or cleanup-ownership refusal and a missing row keep the exact Aborted status require_state_write returns for update_state (blocked_by_quarantine, superseded_by_cleanup_episode, missing). A database error is Internal "{context}: {error}", as before. - state_write_result(outcome) -> Result is the typed form. It returns the overwritten state, or one of FromState { found }, Guard { refusal, found } and Missing, with into_status(context). A caller that answers one case differently, such as warm start with FailedPrecondition, matches it and passes the rest on. Guard carries found for callers that decide by edge.from().contains(found). - update_state_from_best_effort(db, id, edge) -> Option is update_state_best_effort for update_state_from, with a wildcard-free match. It logs refusals and errors with that function's messages and field names, but logs outcome with Debug where the legacy used Display. Only an applied write marks the sandbox dirty. None means a poisoned connection or a database error. The legacy require_state_write is unchanged. Outside tests the new items expect dead_code. Adopting require_state_write_from leaves several of these expectations unfulfilled at once: the function, StateWriteRejected, StateWriteGuard, state_write_result and the impl. unfulfilled_lint_expectations is warn-by-default and an error under CI's clippy -D warnings, so the adopting conversion must drop those attributes. Tests: - Wildcard-free matches cover every StateWriteOutcome and StateWriteRefusal and assert each one's typed rejection and exact code and message. The refusal list is built from such a match. - The same row gets the same status from the new mapping and from the update_state caller path, and every row stays unchanged, for: a quarantined row, a stopping row revived to started, a cleanup row with a delete_row intent, and a missing row. The row's own state is expected, and the messages match literals copied from 8f12bd6. - The best-effort writer applies a write and marks only that sandbox dirty. A refusal and a database error return their result and change nothing. A poisoned connection is None, changes nothing, and wakes the dirty-outbox worker, which exits on the poisoned connection and fails its critical-task guard. The test pauses the clock so the worker's poll timer cannot fire. The sbx-mgr AGENTS.md writer rule names both helpers and the override. --- crates/libs/sbx/mgr/server/AGENTS.md | 6 +- .../mgr/server/src/service/cleanup_support.rs | 160 +++++++ .../src/service/cleanup_support/tests.rs | 418 ++++++++++++++++++ 3 files changed, 583 insertions(+), 1 deletion(-) create mode 100644 crates/libs/sbx/mgr/server/src/service/cleanup_support/tests.rs 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); +}