From 27df76c733c240a6604946ca176f600950139ba1 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Tue, 13 Oct 2026 18:15:24 +0000 Subject: [PATCH] sbx-mgr: let Stop and Kill share one admission rule A fatal Kill (SIGKILL/SIGTERM) now admits exactly the node-local SQLite sandboxes.state values that SandboxState::can_transition_to(Stopping) accepts. That rule is widened from started/paused to every state whose VM is live or launching, mid-operation included: starting, started, stopping, pausing, paused, resuming, checkpointing, suspending and restoring. Kill still refuses creating, created, stopped, suspended, cleanup_required, cleanup_failed, failed, deleting and quarantined with FailedPrecondition, as before. Kill reads only the row, so a live runtime does not change its answer. A non-fatal signal starts no cleanup and keeps its started/paused guard. Stop is unchanged. Its RPC selects only a live runtime and then admits every state but deleting and quarantined, including the settled and cleanup states the shared rule excludes. Making Stop consult the rule would refuse a live runtime under creating, created, stopped, suspended, cleanup_required, cleanup_failed or failed, so Stop keeps its runtime-gated admission. The rule excludes those states because Kill has no runtime gate: superseding a failed or cleanup row would replace its pending cleanup or erase-safety evidence with a stop. The newly admitted Kill states race their in-flight operation exactly as Stop already does for a live runtime. Activation, resume and the scheduled checkpoint use compare-and-set writers and lose cleanly. Tests sweep all 18 states for Stop (live and not live, blocking and not), fatal Kill (both signals, live and not live, blocking and not) and non-fatal Kill, plus can_transition_to(Stopping) itself. --- .../grpc_lifecycle/backup_policy_tests.rs | 3 +- .../server/src/service/grpc_lifecycle/kill.rs | 28 +- .../libs/sbx/mgr/server/src/service/tests.rs | 3 + .../tests/stop_kill_admission_tests.rs | 444 ++++++++++++++++++ crates/libs/sbx/mgr/server/src/state.rs | 30 +- crates/libs/sbx/mgr/server/src/state/tests.rs | 16 +- 6 files changed, 506 insertions(+), 18 deletions(-) create mode 100644 crates/libs/sbx/mgr/server/src/service/tests/stop_kill_admission_tests.rs diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs index a22d15935f..834a46c9f2 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs @@ -1089,7 +1089,8 @@ async fn kill_refuses_signals_it_cannot_deliver_before_cleanup_admission() { }; for state in SandboxState::all() { - let deliverable = matches!(state, SandboxState::Started | SandboxState::Paused); + let deliverable = + crate::service::tests::stop_kill_admission_tests::fatal_kill_admits(state); for signal in [9, 15] { let (service, id) = service_with(Some(state)); let case = format!("{state} signal {signal}"); diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/kill.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/kill.rs index 51c2662c2e..8b5eeac32a 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/kill.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/kill.rs @@ -24,18 +24,18 @@ impl SandboxManagerService { return Err(Status::invalid_argument("signal must be non-zero")); } - // Must be running or paused to receive a signal - if !matches!(sandbox.state, SandboxState::Started | SandboxState::Paused) { - return Err(Status::failed_precondition(format!( - "cannot kill: sandbox is {}, not started or paused", - sandbox.state - ))); - } - // Only fatal signals (SIGKILL=9, SIGTERM=15) trigger state transition and cleanup. // Non-fatal signals (e.g. SIGHUP, SIGUSR1) are just forwarded to the process. let is_fatal = signal == 9 || signal == 15; if !is_fatal { + // A forwarded signal starts no cleanup, so it never supersedes an + // in-flight operation. + if !matches!(sandbox.state, SandboxState::Started | SandboxState::Paused) { + return Err(Status::failed_precondition(format!( + "cannot kill: sandbox is {}, not started or paused", + sandbox.state + ))); + } let mut node_client = connect_node(&self.node_manager) .await .map_err(|e| Status::internal(format!("cannot connect to node-manager: {e}")))?; @@ -51,10 +51,14 @@ impl SandboxManagerService { return Ok(Response::new(KillResponse {})); } - sandbox - .state - .transition(SandboxState::Stopping) - .map_err(Status::from)?; + // A fatal signal stops the sandbox, so it is admitted exactly where a + // stop may supersede the current state. + if !sandbox.state.can_transition_to(SandboxState::Stopping) { + return Err(Status::failed_precondition(format!( + "cannot kill: sandbox is {}, which cannot stop", + sandbox.state + ))); + } let (sandbox_id, uuid8) = Self::cleanup_identity(id, Some(&sandbox.uuid8))?; let completion = CleanupCompletion::MarkStopped; diff --git a/crates/libs/sbx/mgr/server/src/service/tests.rs b/crates/libs/sbx/mgr/server/src/service/tests.rs index 746f911618..056a1242be 100644 --- a/crates/libs/sbx/mgr/server/src/service/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/tests.rs @@ -244,6 +244,9 @@ mod local_backup_inventory_tests; #[cfg(test)] pub(crate) mod node_manager_transport_tests; +#[cfg(test)] +pub(crate) mod stop_kill_admission_tests; + #[cfg(test)] pub(crate) mod service_tests { use super::start_support::*; diff --git a/crates/libs/sbx/mgr/server/src/service/tests/stop_kill_admission_tests.rs b/crates/libs/sbx/mgr/server/src/service/tests/stop_kill_admission_tests.rs new file mode 100644 index 0000000000..e00cf009ea --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/tests/stop_kill_admission_tests.rs @@ -0,0 +1,444 @@ +//! Stop and Kill from every node-local SQLite `sandboxes.state` source state, +//! with and without a live runtime, blocking and not. +//! +//! Both RPCs stop through lifecycle cleanup, which claims `stopping` and ends +//! `stopped`. A fatal Kill admits exactly the states a stop may supersede +//! (`SandboxState::can_transition_to(Stopping)`): every state whose VM is live +//! or launching, mid-operation included. It reads only the row, so a live +//! runtime does not change its answer. A non-fatal signal starts no cleanup +//! and reaches only a started or paused sandbox. Stop selects only a live +//! runtime, then admits every state its cleanup admission does not leave to a +//! delete or quarantine. The expected outcomes are written out here, not +//! derived from the code under test. +//! +//! Node cleanup reaches only the executor's test dispatch, which completes one +//! attempt per granted permit, and node-manager is an in-process fake that +//! answers `StopProcess`. The runtime probe reads a test-owned run directory. +//! An admitted stop or kill removes watchdog monitoring, which stats the fixed +//! sidecar pid path of a fresh sandbox id; nothing exists there. + +use std::collections::BTreeMap; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use pplx_space_common_core::types::SandboxId; +use pplx_space_node_mgr_client::proto::{StopProcessRequest, StopProcessResponse}; +use tonic::Status; + +use super::service_tests::insert_test_sandbox; +use crate::db::CleanupCompletion; +use crate::node_manager_transport::fake::{FakeNodeManager, ServedNodeManager}; +use crate::proto::{KillRequest, StopRequest}; +use crate::service::activation_support::internal_request; +use crate::service::{SandboxManagerService, SandboxManagerServiceOptions}; +use crate::state::SandboxState; + +const WAIT: Duration = Duration::from_secs(30); + +/// Whether a fatal Kill admits a row in `state`. Wildcard-free, so a new +/// lifecycle state does not compile until it is placed. +pub(crate) fn fatal_kill_admits(state: SandboxState) -> bool { + match state { + SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopping + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Restoring => true, + // No VM to stop, a cleanup or failure that owns the row's remaining + // resources, or a delete or quarantine. + SandboxState::Creating + | SandboxState::Created + | SandboxState::Stopped + | SandboxState::Deleting + | SandboxState::Suspended + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Quarantined + | SandboxState::Failed => false, + } +} + +/// Whether a non-fatal Kill forwards its signal to a row in `state`. +fn signal_delivered(state: SandboxState) -> bool { + match state { + SandboxState::Started | SandboxState::Paused => true, + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Stopping + | SandboxState::Stopped + | SandboxState::Deleting + | SandboxState::Pausing + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Quarantined + | SandboxState::Failed => false, + } +} + +/// Whether Stop admits a row in `state` whose runtime is live. Without a live +/// runtime Stop selects nothing from any state. +fn stop_admits_live(state: SandboxState) -> bool { + match state { + SandboxState::Creating + | SandboxState::Created + | SandboxState::Starting + | SandboxState::Started + | SandboxState::Stopping + | SandboxState::Stopped + | SandboxState::Pausing + | SandboxState::Paused + | SandboxState::Resuming + | SandboxState::Checkpointing + | SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::Restoring + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + | SandboxState::Failed => true, + SandboxState::Deleting | SandboxState::Quarantined => false, + } +} + +/// Every row of every table, so a refused call can be shown to change nothing. +fn snapshot(conn: &rusqlite::Connection) -> BTreeMap> { + 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(); + tables + .into_iter() + .map(|table| { + let mut statement = conn.prepare(&format!("SELECT * FROM \"{table}\"")).unwrap(); + let columns = statement.column_count(); + let mut rows: Vec = statement + .query_map([], |row| { + (0..columns) + .map(|index| row.get::<_, rusqlite::types::Value>(index)) + .collect::>>() + .map(|values| format!("{values:?}")) + }) + .unwrap() + .collect::>() + .unwrap(); + rows.sort(); + (table, rows) + }) + .collect() +} + +struct Fixture { + service: SandboxManagerService, + id: String, + signals: Arc>>, + dispatch_permits: Arc, + _node_manager: ServedNodeManager, + _scratch: tempfile::TempDir, +} + +impl Fixture { + /// A row in `state`, with a local restore's columns when it is + /// `restoring`. Node cleanup completes `dispatch_permits` attempts and + /// then holds. + fn new(state: SandboxState, live: bool, dispatch_permits: usize) -> Self { + let conn = crate::db::test_conn(); + let id = SandboxId::new().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(); + } + let signals = Arc::new(Mutex::new(Vec::new())); + let node_manager = FakeNodeManager::default() + .unary("StopProcess", { + let signals = Arc::clone(&signals); + move |request: StopProcessRequest| { + signals.lock().unwrap().push(request); + Ok::<_, Status>(StopProcessResponse::default()) + } + }) + .serve(); + let scratch = tempfile::tempdir().unwrap(); + if live { + let sandbox_dir = scratch.path().join(&id); + std::fs::create_dir(&sandbox_dir).unwrap(); + std::os::unix::fs::symlink(scratch.path(), sandbox_dir.join("fc.pid")).unwrap(); + } + let service = SandboxManagerService::new_with_options( + conn, + SandboxManagerServiceOptions { + node_manager: node_manager.connector(), + ..SandboxManagerServiceOptions::for_test_in(scratch.path()) + }, + ); + let dispatch_permits = Arc::new(tokio::sync::Semaphore::new(dispatch_permits)); + let permits = Arc::clone(&dispatch_permits); + *service.cleanup_executor.dispatch.lock().unwrap() = Some(Arc::new(move |_, _| { + let permits = Arc::clone(&permits); + Box::pin(async move { + match permits.acquire().await { + Ok(permit) => { + permit.forget(); + Ok(()) + } + Err(_) => Err(Status::unavailable("test node cleanup closed")), + } + }) + })); + Self { + service, + id, + signals, + dispatch_permits, + _node_manager: node_manager, + _scratch: scratch, + } + } + + fn snapshot(&self) -> BTreeMap> { + snapshot(&self.service.db_conn().unwrap()) + } + + fn signals(&self) -> Vec { + let signals = self.signals.lock().unwrap(); + for request in signals.iter() { + assert_eq!(request.sandbox_id, self.id); + } + signals.iter().map(|request| request.signal).collect() + } + + /// Assert that cleanup claimed the row: `stopping`, with a cleanup job and, + /// when the caller did not wait, its persisted stop intent. + fn assert_claimed(&self, persisted_intent: bool, case: &str) { + let db = self.service.db_conn().unwrap(); + let row = crate::db::get_sandbox(&db, &self.id).unwrap().unwrap(); + assert_eq!(row.state, SandboxState::Stopping, "{case}"); + assert_eq!( + (row.cleanup_completion, row.cleanup_keep_filesystem), + if persisted_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, + &self.id + ) + .unwrap() + .is_some(), + "{case}: the claim persists its cleanup job" + ); + } + + /// Assert that cleanup completed: `stopped`, with no cleanup work left. + fn assert_stopped(&self, case: &str) { + let db = self.service.db_conn().unwrap(); + let row = crate::db::get_sandbox(&db, &self.id).unwrap().unwrap(); + assert_eq!(row.state, SandboxState::Stopped, "{case}"); + assert_eq!( + (row.cleanup_completion, row.cleanup_keep_filesystem), + (None, None), + "{case}" + ); + assert!( + pplx_space_common_jobs::db::find_active_job( + &db, + crate::db::cleanup_jobs::KIND, + &self.id + ) + .unwrap() + .is_none(), + "{case}" + ); + } + + async fn kill(&self, signal: u32, blocking: bool) -> Result<(), Status> { + tokio::time::timeout( + WAIT, + self.service.kill(internal_request(KillRequest { + sandbox_id: self.id.clone(), + signal, + blocking, + })), + ) + .await + .expect("Kill answers within the wait") + .map(|_| ()) + } + + async fn stop(&self, blocking: bool) -> Result, Status> { + tokio::time::timeout( + WAIT, + self.service.stop(internal_request(StopRequest { + sandbox_id: self.id.clone(), + template_filter: None, + blocking, + immediate_kill: false, + })), + ) + .await + .expect("Stop answers within the wait") + .map(|response| response.into_inner().stopped_ids) + } + + async fn shutdown(self) { + self.dispatch_permits.close(); + self.service.shutdown_token.cancel(); + self.service.task_tracker.close(); + tokio::time::timeout(WAIT, self.service.task_tracker.wait()) + .await + .expect("service tasks stop within the wait after shutdown"); + } +} + +fn enter_isolated_test_process() -> bool { + crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) +} + +/// A fatal Kill stops every live state, mid-operation included, whether or not +/// the runtime probe finds the VM, and refuses every other state without a +/// write or a signal. +#[tokio::test] +async fn fatal_kill_admits_every_state_a_stop_may_supersede() { + if !enter_isolated_test_process() { + return; + } + for signal in [9, 15] { + for live in [true, false] { + for blocking in [true, false] { + for state in SandboxState::all() { + let case = + format!("{state}, signal {signal}, live {live}, blocking {blocking}"); + // A blocking kill waits for one node cleanup attempt; a + // non-blocking one returns while cleanup holds. + let fixture = Fixture::new(state, live, usize::from(blocking)); + let before = fixture.snapshot(); + + let result = fixture.kill(signal, blocking).await; + + if fatal_kill_admits(state) { + result.unwrap_or_else(|status| panic!("{case}: {status:?}")); + assert_eq!(fixture.signals(), [signal], "{case}"); + if blocking { + fixture.assert_stopped(&case); + } else { + fixture.assert_claimed(true, &case); + } + } else { + let status = result.expect_err(&case); + assert_eq!( + status.code(), + tonic::Code::FailedPrecondition, + "{case}: {status:?}" + ); + assert!(fixture.signals().is_empty(), "{case}"); + assert_eq!(fixture.snapshot(), before, "{case}"); + } + fixture.shutdown().await; + } + } + } + } +} + +/// A non-fatal Kill only forwards its signal, and only to a started or paused +/// sandbox; it never writes. +#[tokio::test] +async fn non_fatal_kill_signals_only_a_started_or_paused_sandbox() { + if !enter_isolated_test_process() { + return; + } + for live in [true, false] { + for state in SandboxState::all() { + let case = format!("{state}, live {live}"); + let fixture = Fixture::new(state, live, 0); + let before = fixture.snapshot(); + + let result = fixture.kill(1, true).await; + + if signal_delivered(state) { + result.unwrap_or_else(|status| panic!("{case}: {status:?}")); + assert_eq!(fixture.signals(), [1], "{case}"); + } else { + let status = result.expect_err(&case); + assert_eq!( + status.code(), + tonic::Code::FailedPrecondition, + "{case}: {status:?}" + ); + assert!(fixture.signals().is_empty(), "{case}"); + } + assert_eq!(fixture.snapshot(), before, "{case}"); + fixture.shutdown().await; + } + } +} + +/// Stop stops a live runtime under every state but a delete or quarantine, +/// which it refuses as retryable without a write. Without a live runtime it +/// selects nothing and writes nothing, whatever the state. +#[tokio::test] +async fn stop_admits_a_live_runtime_from_every_state_but_deleting_or_quarantined() { + if !enter_isolated_test_process() { + return; + } + for live in [true, false] { + for blocking in [true, false] { + for state in SandboxState::all() { + let case = format!("{state}, live {live}, blocking {blocking}"); + let fixture = Fixture::new(state, live, usize::from(blocking)); + let before = fixture.snapshot(); + + let result = fixture.stop(blocking).await; + + if !live { + let stopped = result.unwrap_or_else(|status| panic!("{case}: {status:?}")); + assert!(stopped.is_empty(), "{case}"); + assert_eq!(fixture.snapshot(), before, "{case}"); + } else if stop_admits_live(state) { + let stopped = result.unwrap_or_else(|status| panic!("{case}: {status:?}")); + assert_eq!(stopped, std::slice::from_ref(&fixture.id), "{case}"); + if blocking { + fixture.assert_stopped(&case); + } else { + fixture.assert_claimed(true, &case); + } + } else { + let status = result.expect_err(&case); + assert_eq!( + status.code(), + tonic::Code::Unavailable, + "{case}: {status:?}" + ); + assert_eq!(fixture.snapshot(), before, "{case}"); + } + // Stop never signals the process itself. + assert!(fixture.signals().is_empty(), "{case}"); + fixture.shutdown().await; + } + } + } +} diff --git a/crates/libs/sbx/mgr/server/src/state.rs b/crates/libs/sbx/mgr/server/src/state.rs index 9fa44ec00e..ef57cdd638 100644 --- a/crates/libs/sbx/mgr/server/src/state.rs +++ b/crates/libs/sbx/mgr/server/src/state.rs @@ -40,6 +40,9 @@ impl SandboxState { if target == SandboxState::Quarantined { return *self != SandboxState::Failed; } + if target == SandboxState::Stopping { + return self.can_stop(); + } use SandboxState::*; matches!( @@ -50,7 +53,6 @@ impl SandboxState { | (Created, Deleting) // Start/Stop lifecycle | (Starting, Started) - | (Started, Stopping) | (Stopping, Stopped) | (Stopped, Starting) | (Stopped, Deleting) @@ -59,7 +61,6 @@ impl SandboxState { | (Pausing, Paused) | (Paused, Resuming) | (Resuming, Started) - | (Paused, Stopping) // stop/kill from paused // Checkpoint lifecycle (VM keeps running, or from paused for migration) | (Started, Checkpointing) | (Paused, Checkpointing) @@ -96,6 +97,26 @@ impl SandboxState { ) } + /// Whether a stop or fatal kill may supersede this state: the VM is live + /// or launching, including mid-operation, and no delete or quarantine owns + /// the row. The superseded operation's own completion write then races the + /// stop's cleanup. + /// + /// Settled states without a VM, and cleanup and failure states whose VM + /// may already be gone, are excluded: superseding them would replace their + /// pending cleanup or erase-safety evidence with a stop. The Stop RPC also + /// cleans up a live runtime found under those rows; that runtime-gated + /// admission is Stop's own and does not go through this rule. + const fn can_stop(self) -> bool { + use SandboxState::*; + match self { + Starting | Started | Stopping | Pausing | Paused | Resuming | Checkpointing + | Suspending | Restoring => true, + Creating | Created | Stopped | Deleting | Suspended | CleanupRequired + | CleanupFailed | Quarantined | Failed => false, + } + } + pub fn transition(self, target: SandboxState) -> Result { if self.can_transition_to(target) { Ok(target) @@ -286,7 +307,10 @@ impl SandboxState { use SandboxState::*; match self { Starting | Started => "created, stopped, or suspended", - Stopping | Stopped => "started or paused", + Stopping => { + "started, paused, or starting, stopping, pausing, resuming, checkpointing, suspending, or restoring" + } + Stopped => "started or paused", Pausing | Paused => "started", Resuming => "paused", Checkpointing => "started or paused", diff --git a/crates/libs/sbx/mgr/server/src/state/tests.rs b/crates/libs/sbx/mgr/server/src/state/tests.rs index ad4419f047..4ba0f92803 100644 --- a/crates/libs/sbx/mgr/server/src/state/tests.rs +++ b/crates/libs/sbx/mgr/server/src/state/tests.rs @@ -168,9 +168,21 @@ fn starting_to_started() { assert!(Starting.can_transition_to(Started)); } +/// A stop or fatal kill supersedes any state whose VM is live or launching, +/// mid-operation included, and nothing a delete or quarantine owns. Settled +/// states without a VM, and cleanup and failure states, are not stoppable. #[test] -fn started_to_stopping() { - assert!(Started.can_transition_to(Stopping)); +fn stopping_is_reachable_exactly_from_live_states() { + for state in super::SandboxState::all() { + let stoppable = match state { + Starting | Started | Stopping | Pausing | Paused | Resuming | Checkpointing + | Suspending | Restoring => true, + Creating | Created | Stopped | Deleting | Suspended | CleanupRequired + | CleanupFailed | Quarantined | Failed => false, + }; + assert_eq!(state.can_transition_to(Stopping), stoppable, "{state}"); + assert_eq!(state.transition(Stopping).is_ok(), stoppable, "{state}"); + } } #[test]