diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core.rs index bb018178bf..017f58d8c6 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_core.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core.rs @@ -1178,7 +1178,11 @@ fn project_template_catalog( .collect() } +#[cfg(test)] +mod create_after_lookup_tests; #[cfg(test)] mod create_template_lookup_tests; #[cfg(test)] mod tests; +#[cfg(test)] +mod unpersisted_cleanup_tests; diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs new file mode 100644 index 0000000000..a1c5f05c9e --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs @@ -0,0 +1,672 @@ +//! Create's paths after template lookup, against a scripted template resolver +//! and an in-process node-manager. +//! +//! A create that inserts a new sandbox row goes on to create the sandbox's +//! cache root under the fixed host cache directory. Every service here +//! refuses new sandbox rows with a SQLite trigger, so a create that reaches +//! the insert stops there. The refusal carries the row create tried to +//! write, which is what the cases that reach `create_with_id` assert on. + +use super::*; +use crate::node_manager_transport::fake::{FakeNodeManager, ServedNodeManager}; +use crate::template_resolver::fake::FakeTemplateResolver; +use pplx_space_common_svc::PeerIdentity; +use pplx_space_node_mgr_client::proto::{NetworkPoolStatusRequest, NetworkPoolStatusResponse}; +use pplx_space_template_mgr_client::proto::{GetTemplateRequest, GetTemplateResponse, TemplateRef}; +use std::collections::HashMap; + +const TEMPLATE: &str = "seed"; +const BUILD: &str = "sha256:create-after-lookup"; +const OWNER: &str = "root@local"; +/// Prefix of the insert fence's refusal; the rest is the refused row as JSON. +const FENCE: &str = "create fence refused row: "; +/// Bounds every create on real time, including the node-manager calls it +/// makes, so a regression that waits on node-manager fails instead of hanging. +const CREATE_WAIT: Duration = Duration::from_secs(30); + +/// Run the test in a child process without the feature-flag provider +/// settings, so service construction cannot start a provider. Call before +/// constructing any service; return when false. +fn enter_isolated_test_process() -> bool { + crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) +} + +/// The environment rows are seeded with, distinct from every request's. +fn row_env() -> BTreeMap { + BTreeMap::from([("ROW_ENV".to_owned(), "from-row".to_owned())]) +} + +/// The fields of a refused `sandboxes` insert this module asserts on. +#[derive(Debug, serde::Deserialize, PartialEq)] +struct FencedRow { + id: String, + state: String, + template: String, + template_hash: Option, + user: String, + auto_lifecycle: AutoLifecyclePolicy, + auto_lifecycle_authority: String, + network_policy_authority: String, + process_env: BTreeMap, +} + +/// Refuse every new `sandboxes` row, reporting the row in the refusal. +fn install_insert_fence(conn: &rusqlite::Connection) { + conn.execute_batch(&format!( + "CREATE TEMP TRIGGER create_after_lookup_insert_fence + BEFORE INSERT ON main.sandboxes + BEGIN + SELECT RAISE(ABORT, '{FENCE}' || json_object( + 'id', NEW.id, + 'state', NEW.state, + 'template', NEW.template, + 'template_hash', NEW.template_hash, + 'user', NEW.user, + 'auto_lifecycle', json(NEW.auto_lifecycle_json), + 'auto_lifecycle_authority', NEW.auto_lifecycle_authority, + 'network_policy_authority', NEW.network_policy_authority, + 'process_env', json(NEW.process_env_json) + )); + END;" + )) + .unwrap(); +} + +/// The row a create tried to insert, from the fence's refusal. +fn fenced_row(error: &Status) -> FencedRow { + let (_, row) = error + .message() + .split_once(FENCE) + .unwrap_or_else(|| panic!("create did not reach the row insert: {error}")); + serde_json::from_str(row).unwrap() +} + +/// A template build in a test-owned directory, built for this node with a +/// supported spaced guest. +struct TemplateBuild { + dir: tempfile::TempDir, +} + +impl TemplateBuild { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + std::fs::write( + dir.path().join("metadata.json"), + serde_json::json!({ + "arch": std::env::consts::ARCH, + "spaced_version": pplx_space_spaced_common::spaced_version(), + "spaced_compatibility_family": pplx_space_spaced_common::spaced_compatibility_family(), + }) + .to_string(), + ) + .unwrap(); + Self { dir } + } + + fn response(&self, build: &str) -> GetTemplateResponse { + GetTemplateResponse { + template: Some(TemplateRef { + name: TEMPLATE.into(), + content_hash: Some(build.into()), + tag: None, + }), + arch: std::env::consts::ARCH.into(), + template_dir: self.dir.path().to_str().unwrap().into(), + default_auto_lifecycle: Some((&access_idle_suspend_policy()).into()), + network_policy_authority: crate::db::PolicyAuthority::Local.as_proto_i32(), + auto_lifecycle_authority: crate::db::PolicyAuthority::Local.as_proto_i32(), + ..Default::default() + } + } + + /// A resolver answering every lookup with this template at build `build`. + fn resolver(&self, build: &str) -> Arc { + let response = self.response(build); + Arc::new(FakeTemplateResolver::new(move || { + Ok(tonic::Response::new(response.clone())) + })) + } +} + +/// The template's default auto-lifecycle policy, distinct from the disabled +/// policy test rows start with. +fn access_idle_suspend_policy() -> AutoLifecyclePolicy { + use pplx_space_common_core::types::{ + AutoLifecycleConditionSet, AutoLifecycleConditionSetId, AutoLifecycleStage, + }; + AutoLifecyclePolicy { + wake: Default::default(), + condition_sets: BTreeMap::from([( + AutoLifecycleConditionSetId::AccessIdle, + AutoLifecycleConditionSet::access_idle(), + )]), + stages: vec![AutoLifecycleStage { + condition_set: AutoLifecycleConditionSetId::AccessIdle, + at: 600, + action: AutoLifecycleIdleAction::Suspend, + }], + } + .normalized() +} + +struct Fixture { + service: SandboxManagerService, + templates: Arc, + node_manager: ServedNodeManager, +} + +impl Fixture { + /// A service over `conn`, with the insert fence installed, whose template + /// lookups go to `templates` and whose node-manager is `node_manager`. + fn new( + conn: rusqlite::Connection, + templates: Arc, + node_manager: FakeNodeManager, + ) -> Self { + install_insert_fence(&conn); + let node_manager = node_manager.serve(); + let service = SandboxManagerService::new_with_options( + conn, + SandboxManagerServiceOptions { + node_manager: node_manager.connector(), + template_resolver: Arc::clone(&templates) as _, + ..SandboxManagerServiceOptions::for_test() + }, + ); + Self { + service, + templates, + node_manager, + } + } + + async fn create(&self, request: CreateRequest) -> Result { + let mut request = Request::new(request); + request + .extensions_mut() + .insert(PeerIdentity::new(0, 0, None)); + tokio::time::timeout(CREATE_WAIT, self.service.create(request)) + .await + .unwrap_or_else(|_| panic!("create waited longer than {CREATE_WAIT:?}")) + .map(Response::into_inner) + } + + fn row(&self, id: &str) -> Option { + crate::db::get_sandbox(&self.service.db_conn().unwrap(), id).unwrap() + } + + fn row_count(&self) -> i64 { + self.service + .db_conn() + .unwrap() + .query_row("SELECT COUNT(*) FROM sandboxes", [], |row| row.get(0)) + .unwrap() + } + + /// Create looked the template up once, by name. + fn assert_one_lookup(&self, context: &str) { + assert_eq!( + self.templates.requests(), + vec![GetTemplateRequest { + caller_org_id: None, + template: Some(TemplateRef { + name: TEMPLATE.into(), + content_hash: None, + tag: None, + }), + vcpu: None, + memory_mib: None, + }], + "{context}" + ); + } +} + +fn create_request() -> CreateRequest { + CreateRequest { + template: TEMPLATE.into(), + owner: OWNER.into(), + runtime: pplx_space_common_core::types::RUNTIME_FIRECRACKER_ALIAS.into(), + env: HashMap::from([("CREATE_ENV".to_owned(), "from-request".to_owned())]), + ..Default::default() + } +} + +/// Insert a row the way create writes one, with [`row_env`], before the fence +/// is installed. +fn insert_created_row( + conn: &rusqlite::Connection, + id: &str, + state: SandboxState, + template_hash: &str, + owner: &SandboxOwner, + pool_template: Option<&str>, +) { + let sandbox_id = SandboxId::from_str(id).unwrap(); + crate::db::insert_sandbox( + conn, + crate::db::InsertSandboxParams { + id, + uuid8: &SandboxRuntimeLabel::from_sandbox_id(sandbox_id), + state, + template: TEMPLATE, + template_hash: Some(template_hash), + user: owner, + runtime: pplx_space_common_core::types::RUNTIME_FIRECRACKER_ALIAS, + snapshots_enabled: false, + rootfs_size_mb: None, + disk_io_limits: Default::default(), + pool_template, + volume_encryption: false, + process_env: &row_env(), + auto_lifecycle: &AutoLifecyclePolicy::disabled(), + network_policy_override: None, + network_policy_generation: 0, + network_policy_authority: crate::db::PolicyAuthority::Cluster, + auto_lifecycle_authority: crate::db::PolicyAuthority::Cluster, + }, + ) + .unwrap(); +} + +#[tokio::test] +async fn create_reaches_the_row_insert_with_the_templates_build_and_defaults() { + if !enter_isolated_test_process() { + return; + } + struct Case { + name: &'static str, + request: CreateRequest, + /// The id create must insert, when the request names one. + id: Option, + auto_lifecycle: AutoLifecyclePolicy, + auto_lifecycle_authority: &'static str, + } + let requested_id = SandboxId::new().to_string(); + let cases = [ + Case { + name: "template defaults", + request: create_request(), + id: None, + auto_lifecycle: access_idle_suspend_policy(), + auto_lifecycle_authority: "local", + }, + Case { + name: "requested policy", + request: CreateRequest { + auto_lifecycle: Some((&AutoLifecyclePolicy::disabled()).into()), + auto_lifecycle_authority: crate::db::PolicyAuthority::Cluster.as_proto_i32(), + ..create_request() + }, + id: None, + auto_lifecycle: AutoLifecyclePolicy::disabled(), + auto_lifecycle_authority: "cluster", + }, + Case { + name: "requested id with no row", + request: CreateRequest { + sandbox_id: Some(requested_id.clone()), + ..create_request() + }, + id: Some(requested_id), + auto_lifecycle: access_idle_suspend_policy(), + auto_lifecycle_authority: "local", + }, + // An empty pool and no base snapshot fall back to a cold create. + Case { + name: "empty pool", + request: CreateRequest { + boot_mode: BootMode::Pool as i32, + ..create_request() + }, + id: None, + auto_lifecycle: access_idle_suspend_policy(), + auto_lifecycle_authority: "local", + }, + ]; + for case in cases { + let build = TemplateBuild::new(); + let fixture = Fixture::new( + crate::db::test_conn(), + build.resolver(BUILD), + FakeNodeManager::default(), + ); + + let error = fixture.create(case.request).await.unwrap_err(); + + assert_eq!( + error.code(), + tonic::Code::Internal, + "{}: {error}", + case.name + ); + let row = fenced_row(&error); + if let Some(id) = &case.id { + assert_eq!(&row.id, id, "{}", case.name); + } + assert_eq!( + row, + FencedRow { + id: row.id.clone(), + state: "creating".into(), + template: TEMPLATE.into(), + template_hash: Some(BUILD.into()), + user: OWNER.into(), + auto_lifecycle: case.auto_lifecycle, + auto_lifecycle_authority: case.auto_lifecycle_authority.into(), + network_policy_authority: "local".into(), + process_env: BTreeMap::from([("CREATE_ENV".into(), "from-request".into())]), + }, + "{}", + case.name + ); + fixture.assert_one_lookup(case.name); + assert!(fixture.node_manager.calls().is_empty(), "{}", case.name); + assert_eq!(fixture.row_count(), 0, "{}", case.name); + } +} + +#[tokio::test] +async fn autostart_create_checks_ready_network_slots_before_the_row_insert() { + if !enter_isolated_test_process() { + return; + } + struct Case { + name: &'static str, + pool_status: fn() -> Result, + code: tonic::Code, + /// `None` when create gets past the check to the row insert. + refusal: Option<&'static str>, + } + let cases = [ + Case { + name: "no ready slot", + pool_status: || Ok(NetworkPoolStatusResponse::default()), + code: tonic::Code::ResourceExhausted, + refusal: Some("no ready network slots"), + }, + Case { + name: "status refused", + pool_status: || Err(Status::unavailable("scripted pool status refusal")), + code: tonic::Code::Unavailable, + refusal: Some("network pool status failed: scripted pool status refusal"), + }, + Case { + name: "ready slot", + pool_status: || { + Ok(NetworkPoolStatusResponse { + network_slots_ready: 1, + ..Default::default() + }) + }, + code: tonic::Code::Internal, + refusal: None, + }, + ]; + for case in cases { + let build = TemplateBuild::new(); + let pool_status = case.pool_status; + let fixture = Fixture::new( + crate::db::test_conn(), + build.resolver(BUILD), + FakeNodeManager::default().unary( + "GetNetworkPoolStatus", + move |_: NetworkPoolStatusRequest| pool_status(), + ), + ); + + let error = fixture + .create(CreateRequest { + autostart: true, + ..create_request() + }) + .await + .unwrap_err(); + + assert_eq!(error.code(), case.code, "{}: {error}", case.name); + match case.refusal { + Some(refusal) => assert_eq!(error.message(), refusal, "{}", case.name), + None => assert_eq!(fenced_row(&error).template_hash.as_deref(), Some(BUILD)), + } + assert_eq!( + fixture.node_manager.calls(), + vec!["GetNetworkPoolStatus".to_owned()], + "{}", + case.name + ); + fixture.assert_one_lookup(case.name); + assert_eq!(fixture.row_count(), 0, "{}", case.name); + } +} + +/// A paused pool donor at `build`, as the pool leaves it for adoption. +fn insert_pool_donor(conn: &rusqlite::Connection, build: &str) -> String { + let donor = SandboxId::new(); + let id = donor.to_string(); + insert_created_row( + conn, + &id, + SandboxState::Paused, + build, + &TEMPLATE_MANAGER_SANDBOX_OWNER, + Some(TEMPLATE), + ); + crate::db::update_space_api_route_identity( + conn, + donor, + &pplx_space_common_core::types::SpaceApiRouteIdentity { + sandbox_id: Some(donor), + route_generation: Some(pplx_space_common_core::types::RouteGeneration::new(1).unwrap()), + current_node_id: Some(pplx_space_common_core::types::NodeId::new()), + substrate_kind: Some(NodeManagedBy::Space), + execution_generation: Some(ExecutionGeneration::new(1).unwrap()), + activation_attempt_id: Some( + pplx_space_common_core::types::SandboxActivationAttemptId::new(), + ), + ..Default::default() + }, + crate::auto_lifecycle_delete_receipts::RouteIdentityProvenance::Unfenced, + ) + .unwrap(); + id +} + +#[tokio::test] +async fn pool_create_adopts_a_donor_of_the_resolved_build_with_the_templates_defaults() { + if !enter_isolated_test_process() { + return; + } + let conn = crate::db::test_conn(); + let donor = insert_pool_donor(&conn, BUILD); + let build = TemplateBuild::new(); + let fixture = Fixture::new(conn, build.resolver(BUILD), FakeNodeManager::default()); + + let response = fixture + .create(CreateRequest { + boot_mode: BootMode::Pool as i32, + ..create_request() + }) + .await + .unwrap(); + + assert_eq!(response.sandbox_id, donor); + assert_eq!(response.boot_mode_used, BootMode::Pool as i32); + assert!(!response.reused_existing); + let adopted = fixture.row(&donor).unwrap(); + assert_eq!(adopted.state, SandboxState::Paused); + assert_eq!(adopted.user.as_str(), OWNER); + assert_eq!( + adopted.process_env, + BTreeMap::from([("CREATE_ENV".into(), "from-request".into())]) + ); + assert_eq!(adopted.auto_lifecycle, access_idle_suspend_policy()); + assert_eq!( + adopted.auto_lifecycle_authority, + crate::db::PolicyAuthority::Local + ); + fixture.assert_one_lookup("pool adoption"); + assert!(fixture.node_manager.calls().is_empty()); + assert_eq!(fixture.row_count(), 1); +} + +#[tokio::test] +async fn pool_create_never_adopts_a_donor_of_another_build() { + if !enter_isolated_test_process() { + return; + } + let conn = crate::db::test_conn(); + let donor = insert_pool_donor(&conn, "sha256:previous-build"); + let build = TemplateBuild::new(); + let fixture = Fixture::new(conn, build.resolver(BUILD), FakeNodeManager::default()); + let before = fixture.row(&donor).unwrap(); + + // Adoption selects donors by the resolved build. What create does next + // when the pool holds only stale donors is not pinned here: it refuses + // with ResourceExhausted, while the documented contract falls through to + // a base-snapshot or cold create, which the insert fence would refuse. + let result = fixture + .create(CreateRequest { + boot_mode: BootMode::Pool as i32, + ..create_request() + }) + .await; + + assert!(result.is_err(), "{result:?}"); + let after = fixture.row(&donor).unwrap(); + assert_eq!(after.state, before.state); + assert_eq!(after.user, before.user); + assert_eq!(after.process_env, row_env()); + assert_eq!(after.auto_lifecycle, before.auto_lifecycle); + fixture.assert_one_lookup("stale donor"); + assert!(fixture.node_manager.calls().is_empty()); + assert_eq!(fixture.row_count(), 1); +} + +/// What a requested-id create does with a compatible existing row in +/// `state`, written out per state. +fn requested_create_outcome(state: SandboxState) -> Result<(), (tonic::Code, String)> { + use SandboxState::*; + match state { + Created | Started | Paused | Suspended | Stopped => Ok(()), + CleanupRequired | CleanupFailed => Err(( + tonic::Code::Unavailable, + "is still completing activation cleanup; retry".to_owned(), + )), + Creating | Starting | Stopping | Deleting | Pausing | Resuming | Checkpointing + | Suspending | Restoring | Quarantined | Failed => Err(( + tonic::Code::AlreadyExists, + format!("already exists in state {state}"), + )), + } +} + +#[tokio::test] +async fn requested_id_create_reuses_only_a_compatible_settled_row() { + if !enter_isolated_test_process() { + return; + } + struct Case { + name: String, + state: SandboxState, + template_hash: &'static str, + request_policy: AutoLifecyclePolicy, + outcome: Result<(), (tonic::Code, String)>, + } + // Every state with a request whose policy differs from the row's, so a + // reuse must update the policy and a refusal must leave it. + let mut cases: Vec = SandboxState::all() + .into_iter() + .map(|state| Case { + name: format!("compatible {state} row"), + state, + template_hash: BUILD, + request_policy: access_idle_suspend_policy(), + outcome: requested_create_outcome(state), + }) + .collect(); + cases.push(Case { + name: "compatible started row with the same policy".into(), + state: SandboxState::Started, + template_hash: BUILD, + request_policy: AutoLifecyclePolicy::disabled(), + outcome: Ok(()), + }); + // Create checks the build before and, when the policy changes, again + // under the wake lock; the same-policy case reaches only the first check. + for (policy, request_policy) in [ + ("the same policy", AutoLifecyclePolicy::disabled()), + ("a changed policy", access_idle_suspend_policy()), + ] { + cases.push(Case { + name: format!("started row of another build with {policy}"), + state: SandboxState::Started, + template_hash: "sha256:previous-build", + request_policy, + outcome: Err(( + tonic::Code::AlreadyExists, + "exists with incompatible create parameters: template_hash".to_owned(), + )), + }); + } + for case in cases { + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + crate::db::seed_historical_lifecycle_rows(&conn, |conn| { + insert_created_row( + conn, + &id, + SandboxState::Created, + case.template_hash, + &SandboxOwner::parse(OWNER.to_owned()).unwrap(), + None, + ); + conn.execute( + "UPDATE sandboxes SET state = ?1 WHERE id = ?2", + rusqlite::params![case.state.as_str(), &id], + ) + .unwrap(); + }); + let build = TemplateBuild::new(); + let fixture = Fixture::new(conn, build.resolver(BUILD), FakeNodeManager::default()); + let before = fixture.row(&id).unwrap(); + assert_eq!(before.state, case.state, "{}", case.name); + + let result = fixture + .create(CreateRequest { + sandbox_id: Some(id.clone()), + auto_lifecycle: Some((&case.request_policy).into()), + auto_lifecycle_authority: crate::db::PolicyAuthority::Cluster.as_proto_i32(), + ..create_request() + }) + .await; + + let after = fixture.row(&id).unwrap(); + match &case.outcome { + Err((code, message)) => { + let error = result.unwrap_err(); + assert_eq!(error.code(), *code, "{}: {error}", case.name); + assert!( + error.message().contains(message.as_str()), + "{}: {error}", + case.name + ); + assert_eq!(after.auto_lifecycle, before.auto_lifecycle, "{}", case.name); + } + Ok(()) => { + let response = result.unwrap_or_else(|error| panic!("{}: {error}", case.name)); + assert_eq!(response.sandbox_id, id, "{}", case.name); + assert!(response.reused_existing, "{}", case.name); + assert_eq!(after.auto_lifecycle, case.request_policy, "{}", case.name); + } + } + // Reuse keeps the row's own environment over the request's. + assert_eq!(after.state, before.state, "{}", case.name); + assert_eq!(after.process_env, row_env(), "{}", case.name); + fixture.assert_one_lookup(&case.name); + assert!(fixture.node_manager.calls().is_empty(), "{}", case.name); + assert_eq!(fixture.row_count(), 1, "{}", case.name); + } +} diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_core/unpersisted_cleanup_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_core/unpersisted_cleanup_tests.rs new file mode 100644 index 0000000000..d5fe6c0d85 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/grpc_core/unpersisted_cleanup_tests.rs @@ -0,0 +1,210 @@ +//! Stop and delete when the lifecycle claim cannot be persisted. +//! +//! An ordinary stop or delete whose admission cannot write the `stopping` or +//! `deleting` claim still cleans the runtime up, through node-manager directly +//! rather than the durable cleanup executor, and then reports the persistence +//! failure. A SQLite trigger refuses the claim write here, and an in-process +//! node-manager answers the cleanup. Stops pass `immediate_kill`, which skips +//! the guest filesystem flush: that flush dials the guest's fixed spaced-host +//! socket. + +use super::*; +use crate::node_manager_transport::fake::FakeNodeManager; +use crate::service::tests::service_tests::insert_test_sandbox; +use pplx_space_common_svc::PeerIdentity; +use pplx_space_node_mgr_client::proto::{CleanupSandboxRequest, CleanupSandboxResponse}; + +const CLAIM_REFUSAL: &str = "scripted lifecycle claim refusal"; + +/// Refuse every write of a sandbox's `stopping` or `deleting` claim. +fn refuse_lifecycle_claims(conn: &rusqlite::Connection) { + conn.execute_batch(&format!( + "CREATE TEMP TRIGGER unpersisted_cleanup_claim_refusal + BEFORE UPDATE OF state ON main.sandboxes + WHEN NEW.state IN ('{}', '{}') + BEGIN + SELECT RAISE(ABORT, '{CLAIM_REFUSAL}'); + END;", + SandboxState::Stopping.as_str(), + SandboxState::Deleting.as_str(), + )) + .unwrap(); +} + +#[derive(Clone, Copy, Debug)] +enum Rpc { + Stop, + Delete, +} + +impl Rpc { + async fn call(self, service: &SandboxManagerService, id: &str, blocking: bool) -> Status { + let peer = PeerIdentity::new(0, 0, None); + let call = async { + match self { + Self::Stop => { + let mut request = Request::new(StopRequest { + sandbox_id: id.to_owned(), + template_filter: None, + blocking, + immediate_kill: true, + }); + request.extensions_mut().insert(peer); + service.stop(request).await.map(|_| ()) + } + Self::Delete => { + let mut request = Request::new(DeleteRequest { + sandbox_id: id.to_owned(), + template_filter: None, + blocking, + }); + request.extensions_mut().insert(peer); + service.delete(request).await.map(|_| ()) + } + } + }; + // Bounds a regression that waits on cleanup instead of failing. + tokio::time::timeout(Duration::from_secs(30), call) + .await + .unwrap_or_else(|_| panic!("{self:?} waited longer than 30 s")) + .expect_err("the lifecycle claim was refused") + } + + /// The node cleanup an unpersisted cleanup of `id` asks for. + fn node_cleanup( + self, + id: &str, + execution_fence: Option, + ) -> CleanupSandboxRequest { + let stop = matches!(self, Self::Stop); + CleanupSandboxRequest { + execution_fence, + sandbox_id: id.to_owned(), + force: true, + // A stop keeps the filesystem and returns once the runtime's slot + // is released; a delete removes it. + keep_filesystem: stop, + return_after_slot_release: stop, + immediate_kill: stop, + } + } + + fn transition(self) -> &'static str { + match self { + Self::Stop => "stopping", + Self::Delete => "deleting", + } + } + + fn context(self) -> &'static str { + match self { + Self::Stop => "cleanup after unpersisted stop", + Self::Delete => "cleanup after unpersisted delete", + } + } +} + +#[tokio::test] +async fn unpersisted_stop_and_delete_clean_up_through_node_manager_and_report_the_claim_failure() { + if !crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) { + return; + } + for rpc in [Rpc::Stop, Rpc::Delete] { + for blocking in [false, true] { + for node_refuses in [false, true] { + let context = format!("{rpc:?} blocking={blocking} node_refuses={node_refuses}"); + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + insert_test_sandbox(&conn, &id, SandboxState::Started); + let execution_fence = crate::db::node_execution_fence(&conn, &id) + .unwrap() + .map(Into::into); + refuse_lifecycle_claims(&conn); + let cleanups = Arc::new(Mutex::new(Vec::new())); + let node_manager = FakeNodeManager::default() + .unary("CleanupSandbox", { + let cleanups = Arc::clone(&cleanups); + move |request: CleanupSandboxRequest| { + cleanups.lock().unwrap().push(request); + if node_refuses { + Err(Status::unavailable("scripted cleanup refusal")) + } else { + Ok(CleanupSandboxResponse::default()) + } + } + }) + .serve(); + let run_dir = tempfile::tempdir().unwrap(); + // Stop selects only a live runtime: link its `fc.pid`. + std::fs::create_dir(run_dir.path().join(&id)).unwrap(); + std::os::unix::fs::symlink(run_dir.path(), run_dir.path().join(&id).join("fc.pid")) + .unwrap(); + let service = SandboxManagerService::new_with_options( + conn, + SandboxManagerServiceOptions { + node_manager: node_manager.connector(), + ..SandboxManagerServiceOptions::for_test_in(run_dir.path()) + }, + ); + + let error = rpc.call(&service, &id, blocking).await; + + if node_refuses { + assert_eq!(error.code(), tonic::Code::Unavailable, "{context}: {error}"); + assert_eq!( + error.message(), + format!("{}: scripted cleanup refusal", rpc.context()), + "{context}" + ); + } else { + assert_eq!(error.code(), tonic::Code::Internal, "{context}: {error}"); + let message = error.message(); + assert!( + message.starts_with(&format!( + "runtime cleanup completed but {} transition could not be persisted for sandbox {id}: claim cleanup lifecycle failed: ", + rpc.transition() + )) && message.contains(CLAIM_REFUSAL), + "{context}: {error}" + ); + } + assert_eq!( + *cleanups.lock().unwrap(), + vec![rpc.node_cleanup(&id, execution_fence)], + "{context}" + ); + assert_eq!( + node_manager.calls(), + vec!["CleanupSandbox".to_owned()], + "{context}" + ); + let db = service.db_conn().unwrap(); + let row = crate::db::get_sandbox(&db, &id).unwrap().unwrap(); + assert_eq!(row.state, SandboxState::Started, "{context}"); + assert_eq!(row.cleanup_completion, None, "{context}"); + assert!( + crate::db::cleanup_episode(&db, &id).unwrap().is_none(), + "{context}" + ); + assert!( + pplx_space_common_jobs::db::find_active_job( + &db, + crate::db::cleanup_jobs::KIND, + &id + ) + .unwrap() + .is_none(), + "{context}" + ); + assert!( + crate::db::list_node_dirty_refs(&db, 100) + .unwrap() + .is_empty(), + "{context}" + ); + } + } + } +} diff --git a/crates/libs/sbx/mgr/server/src/service/tests.rs b/crates/libs/sbx/mgr/server/src/service/tests.rs index 0dfbc9dfb0..746f911618 100644 --- a/crates/libs/sbx/mgr/server/src/service/tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/tests.rs @@ -236,6 +236,8 @@ fn encrypted_restore_snapshot_error_preserves_invalid_key_status() { #[cfg(test)] mod cancelled_cleanup_rpc_tests; +#[cfg(test)] +mod endpoint_capture_tests; #[cfg(test)] mod local_backup_inventory_tests; @@ -517,7 +519,7 @@ pub(crate) mod service_tests { request } - fn node_api_trusted_lifecycle_request(message: T) -> Request { + pub(crate) fn node_api_trusted_lifecycle_request(message: T) -> Request { let mut request = Request::new(message); request.extensions_mut().insert(PeerIdentity::new( users::SPACE_NODEAPI.uid, @@ -4656,55 +4658,6 @@ pub(crate) mod service_tests { ); } - // Exposure re-apply during a wake in flight: after a suspend, the - // cluster marks open port exposures pending and the node-ctl reconciler - // re-arms the in-guest loopback bridge once the sandbox is back. The - // guest is reachable as soon as the VM resumes — long before the wake's - // readiness contract and network-policy gates finish — so the bridge - // RPCs must not gate on `Started`. Gating on `Started` left exposed - // ports answering 502 for the full duration of a stalled wake even - // though the guest was already serving spaced RPCs. - #[tokio::test] - async fn spaced_open_loopback_bridge_passes_state_gate_while_restoring() { - if !crate::service::tests::enter_route_test_process(&[]) { - return; - } - - let conn = crate::db::test_conn(); - let sandbox_id = SandboxId::new().to_string(); - insert_test_sandbox_with_runtime(&conn, &sandbox_id, SandboxState::Restoring); - complete_runtime_endpoint_identity(&conn, &sandbox_id); - let service = SandboxManagerService::for_test(conn); - - let error = service - .spaced_open_loopback_bridge(node_ctl_request(SpacedOpenLoopbackBridgeRequest { - sandbox_id: sandbox_id.clone(), - operation: Some( - pplx_space_sbx_mgr_client::operations::OpenLoopbackBridgeRequest { - port: 9012, - wait_timeout: None, - }, - ), - })) - .await - .expect_err("no spaced-host socket exists in unit tests"); - - // The state gate must let a restoring sandbox through; the endpoint - // attestation dependency is intentionally absent from this unit test. - assert_eq!( - error.code(), - tonic::Code::Unavailable, - "expected the endpoint connect failure, got: {error:?}" - ); - assert!( - error - .message() - .contains("spaced endpoint proof is temporarily unavailable"), - "unexpected error message: {}", - error.message() - ); - } - #[tokio::test] async fn spaced_open_loopback_bridge_marks_non_reachable_states_terminal() { if !crate::service::tests::enter_route_test_process(&[]) { @@ -4870,83 +4823,6 @@ pub(crate) mod service_tests { ); } - #[tokio::test] - async fn spaced_open_loopback_bridge_exempts_node_ctl_without_route_token() { - if !crate::service::tests::enter_route_test_process(&[]) { - return; - } - - let conn = crate::db::test_conn(); - let sandbox_id = SandboxId::new().to_string(); - insert_test_sandbox_with_runtime(&conn, &sandbox_id, SandboxState::Restoring); - complete_runtime_endpoint_identity(&conn, &sandbox_id); - let service = SandboxManagerService::for_test(conn); - - let error = service - .spaced_open_loopback_bridge(node_ctl_request(SpacedOpenLoopbackBridgeRequest { - sandbox_id: sandbox_id.clone(), - operation: Some( - pplx_space_sbx_mgr_client::operations::OpenLoopbackBridgeRequest { - port: 9012, - wait_timeout: None, - }, - ), - })) - .await - .expect_err("no spaced-host socket exists in unit tests"); - - assert_eq!( - error.code(), - tonic::Code::Unavailable, - "expected the endpoint connect failure, got: {error:?}" - ); - assert!( - error - .message() - .contains("spaced endpoint proof is temporarily unavailable"), - "unexpected error message: {}", - error.message() - ); - } - - #[tokio::test] - async fn spaced_close_loopback_bridge_passes_state_gate_while_restoring() { - if !crate::service::tests::enter_route_test_process(&[]) { - return; - } - - let conn = crate::db::test_conn(); - let sandbox_id = SandboxId::new().to_string(); - insert_test_sandbox_with_runtime(&conn, &sandbox_id, SandboxState::Restoring); - complete_runtime_endpoint_identity(&conn, &sandbox_id); - let service = SandboxManagerService::for_test(conn); - - let error = service - .spaced_close_loopback_bridge(node_ctl_request(SpacedCloseLoopbackBridgeRequest { - sandbox_id: sandbox_id.clone(), - operation: Some( - pplx_space_sbx_mgr_client::operations::CloseLoopbackBridgeRequest { - port: 9012, - }, - ), - })) - .await - .expect_err("no spaced-host socket exists in unit tests"); - - assert_eq!( - error.code(), - tonic::Code::Unavailable, - "expected the endpoint connect failure, got: {error:?}" - ); - assert!( - error - .message() - .contains("spaced endpoint proof is temporarily unavailable"), - "unexpected error message: {}", - error.message() - ); - } - #[tokio::test] async fn spaced_close_loopback_bridge_rejects_invalid_route_token() { if !crate::service::tests::enter_route_test_process(&[]) { @@ -5011,78 +4887,6 @@ pub(crate) mod service_tests { ); } - #[tokio::test] - async fn spaced_close_loopback_bridge_exempts_node_ctl_without_route_token() { - if !crate::service::tests::enter_route_test_process(&[]) { - return; - } - - let conn = crate::db::test_conn(); - let sandbox_id = SandboxId::new().to_string(); - insert_test_sandbox_with_runtime(&conn, &sandbox_id, SandboxState::Restoring); - complete_runtime_endpoint_identity(&conn, &sandbox_id); - let service = SandboxManagerService::for_test(conn); - - let error = service - .spaced_close_loopback_bridge(node_ctl_request(SpacedCloseLoopbackBridgeRequest { - sandbox_id: sandbox_id.clone(), - operation: Some( - pplx_space_sbx_mgr_client::operations::CloseLoopbackBridgeRequest { - port: 9012, - }, - ), - })) - .await - .expect_err("no spaced-host socket exists in unit tests"); - - assert_eq!( - error.code(), - tonic::Code::Unavailable, - "expected the endpoint connect failure, got: {error:?}" - ); - assert!( - error - .message() - .contains("spaced endpoint proof is temporarily unavailable"), - "unexpected error message: {}", - error.message() - ); - } - - #[tokio::test] - async fn spaced_wait_ready_trusted_node_api_reaches_spaced_host_without_route_token() { - if !crate::service::tests::enter_route_test_process(&[]) { - return; - } - - let conn = crate::db::test_conn(); - let sandbox_id = SandboxId::new().to_string(); - insert_test_sandbox_with_runtime(&conn, &sandbox_id, SandboxState::Started); - complete_runtime_endpoint_identity(&conn, &sandbox_id); - let service = SandboxManagerService::for_test(conn); - - let error = service - .spaced_wait_ready(node_api_trusted_lifecycle_request(SpacedWaitReadyRequest { - sandbox_id, - timeout: None, - })) - .await - .expect_err("no spaced-host socket exists in unit tests"); - - assert_eq!( - error.code(), - tonic::Code::Unavailable, - "expected the endpoint connect failure, got: {error:?}" - ); - assert!( - error - .message() - .contains("spaced endpoint proof is temporarily unavailable"), - "unexpected error message: {}", - error.message() - ); - } - #[tokio::test] async fn spaced_wait_ready_allows_identity_free_pool_donor_to_reach_spaced_host() { if !crate::service::tests::enter_route_test_process(&[]) { @@ -6852,7 +6656,7 @@ pub(crate) mod service_tests { assert!(crate::db::record_remote_restore_artifact_staging(conn, id).unwrap()); } - fn insert_test_sandbox_with_runtime( + pub(crate) fn insert_test_sandbox_with_runtime( conn: &rusqlite::Connection, id: &str, state: SandboxState, @@ -6891,7 +6695,7 @@ pub(crate) mod service_tests { insert_test_sandbox_with_runtime(conn, id, SandboxState::Started); } - fn complete_runtime_endpoint_identity(conn: &rusqlite::Connection, id: &str) { + pub(crate) fn complete_runtime_endpoint_identity(conn: &rusqlite::Connection, id: &str) { conn.execute( "UPDATE sandboxes SET space_api_route_execution_generation = 1, diff --git a/crates/libs/sbx/mgr/server/src/service/tests/endpoint_capture_tests.rs b/crates/libs/sbx/mgr/server/src/service/tests/endpoint_capture_tests.rs new file mode 100644 index 0000000000..46df4d4b20 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/tests/endpoint_capture_tests.rs @@ -0,0 +1,150 @@ +//! Spaced bridge and readiness calls that pass their state and caller gates +//! ask node-manager for the sandbox's endpoint proof before any spaced-host +//! connection. +//! +//! Either an in-process node-manager refuses the capture, or node-manager is +//! down and the call fails retryably. Neither reaches the guest's fixed +//! spaced-host socket. + +use super::node_manager_transport_tests::service_with_node_manager; +use super::service_tests::{ + complete_runtime_endpoint_identity, insert_test_sandbox_with_runtime, + node_api_trusted_lifecycle_request, node_ctl_request, +}; +use super::*; +use crate::node_manager_transport::fake::FakeNodeManager; +use pplx_space_node_mgr_client::proto::{CaptureSandboxEndpointProofRequest, SandboxEndpointProof}; + +#[derive(Clone, Copy, Debug)] +enum Call { + OpenLoopbackBridge, + CloseLoopbackBridge, + WaitReady, +} + +impl Call { + async fn send(self, service: &SandboxManagerService, sandbox_id: &str) -> Status { + let sandbox_id = sandbox_id.to_owned(); + match self { + Self::OpenLoopbackBridge => service + .spaced_open_loopback_bridge(node_ctl_request(SpacedOpenLoopbackBridgeRequest { + sandbox_id, + operation: Some( + pplx_space_sbx_mgr_client::operations::OpenLoopbackBridgeRequest { + port: 9012, + wait_timeout: None, + }, + ), + })) + .await + .map(|_| ()), + Self::CloseLoopbackBridge => service + .spaced_close_loopback_bridge(node_ctl_request(SpacedCloseLoopbackBridgeRequest { + sandbox_id, + operation: Some( + pplx_space_sbx_mgr_client::operations::CloseLoopbackBridgeRequest { + port: 9012, + }, + ), + })) + .await + .map(|_| ()), + Self::WaitReady => service + .spaced_wait_ready(node_api_trusted_lifecycle_request(SpacedWaitReadyRequest { + sandbox_id, + timeout: None, + })) + .await + .map(|_| ()), + } + .expect_err("node-manager refuses the endpoint proof") + } +} + +#[tokio::test] +async fn gated_spaced_calls_capture_the_endpoint_proof_from_node_manager() { + if !crate::service::tests::enter_route_test_process(&[ + ("EPPO_SDK_KEY", None), + ("SPACE_FLAGD_ENABLED", None), + ]) { + return; + } + // Exposure re-apply during a wake in flight: after a suspend, the + // cluster marks open port exposures pending and the node-ctl reconciler + // re-arms the in-guest loopback bridge once the sandbox is back. The + // guest is reachable as soon as the VM resumes — long before the wake's + // readiness contract and network-policy gates finish — so the bridge + // RPCs must not gate on `Started`. Gating on `Started` left exposed + // ports answering 502 for the full duration of a stalled wake even + // though the guest was already serving spaced RPCs. node-ctl needs no + // route token, and a trusted node-api lifecycle caller needs none to + // wait for readiness. + for (call, state) in [ + (Call::OpenLoopbackBridge, SandboxState::Restoring), + (Call::CloseLoopbackBridge, SandboxState::Restoring), + (Call::WaitReady, SandboxState::Started), + ] { + for node_manager_up in [true, false] { + let context = + format!("{call:?} on a {state} sandbox, node-manager up: {node_manager_up}"); + let conn = crate::db::test_conn(); + let sandbox_id = SandboxId::new().to_string(); + insert_test_sandbox_with_runtime(&conn, &sandbox_id, state); + complete_runtime_endpoint_identity(&conn, &sandbox_id); + let captures = Arc::new(Mutex::new(Vec::new())); + let node_manager = FakeNodeManager::default() + .unary("CaptureSandboxEndpointProof", { + let captures = Arc::clone(&captures); + move |request: CaptureSandboxEndpointProofRequest| { + captures.lock().unwrap().push(request); + Err::(Status::failed_precondition( + "scripted endpoint proof refusal", + )) + } + }) + .serve(); + let service = if node_manager_up { + service_with_node_manager(conn, node_manager.connector()) + } else { + // The test default: node-manager's socket is unbound. + SandboxManagerService::for_test(conn) + }; + + let error = call.send(&service, &sandbox_id).await; + + if node_manager_up { + // node-manager's refusal of the local binding. + assert_eq!( + error.code(), + tonic::Code::PermissionDenied, + "{context}: {error}" + ); + assert_eq!( + error.message(), + "spaced endpoint attestation failed", + "{context}" + ); + assert_eq!( + *captures.lock().unwrap(), + vec![CaptureSandboxEndpointProofRequest { + sandbox_id: sandbox_id.clone() + }], + "{context}" + ); + assert_eq!( + node_manager.calls(), + vec!["CaptureSandboxEndpointProof".to_owned()], + "{context}" + ); + } else { + // A down node-manager is retryable. + assert_eq!(error.code(), tonic::Code::Unavailable, "{context}: {error}"); + assert_eq!( + error.message(), + "spaced endpoint proof is temporarily unavailable", + "{context}" + ); + } + } + } +} diff --git a/crates/libs/sbx/mgr/server/src/snapshot_gc/tests.rs b/crates/libs/sbx/mgr/server/src/snapshot_gc/tests.rs index d6f4855a90..20ccf1ae59 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_gc/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_gc/tests.rs @@ -168,33 +168,83 @@ async fn malformed_artifact_times_are_retained_without_blocking_valid_siblings() assert!(!Path::new(&valid).exists()); } -/// Backup-completed + idle + not pinned → reaped. Disk directory -/// is gone after the tick; row is gone too. +/// Bounds the reaper's served node-manager call on real time. The test's +/// clock is paused and auto-advances while the call waits on its socket, so a +/// tokio timeout would fire early. +const NODE_MANAGER_WAIT: std::time::Duration = std::time::Duration::from_secs(30); + +/// `future`'s output, or a panic once [`NODE_MANAGER_WAIT`] of real time has +/// passed. The watchdog thread exits as soon as `future` completes. +async fn within_node_manager_wait(future: F) -> F::Output { + let (done, finished) = std::sync::mpsc::channel::<()>(); + let (expired, timed_out) = tokio::sync::oneshot::channel::<()>(); + std::thread::spawn(move || { + if let Err(std::sync::mpsc::RecvTimeoutError::Timeout) = + finished.recv_timeout(NODE_MANAGER_WAIT) + { + let _ = expired.send(()); + } + }); + let output = tokio::select! { + output = future => output, + _ = timed_out => panic!("waited longer than {NODE_MANAGER_WAIT:?} on node-manager"), + }; + drop(done); + output +} + +/// Backup-completed + idle + not pinned → reaped. The row is gone, and the +/// artifact directory moves to its `.gc-pending-` staging sibling, which +/// node-manager is asked to delete. #[tokio::test(start_paused = true)] async fn reaps_idle_uploaded_unpinned_artifact() { + use pplx_space_node_mgr_client::proto::{ + DeleteCacheArtifactRequest, DeleteCacheArtifactResponse, + }; + let conn = open_test_db(); let tmp = TempDir::new().unwrap(); let path = make_artifact_dir(&tmp, "art1", 4096); insert_artifact_row(&conn, TEST_ART_ID, "sbx1", &path, 0, 0, Some(0)); insert_sandbox_row(&conn, "sbx1", None); + let deletions = Arc::new(Mutex::new(Vec::new())); + let node_manager = crate::node_manager_transport::fake::FakeNodeManager::default() + .unary("DeleteCacheArtifact", { + let deletions = Arc::clone(&deletions); + move |request: DeleteCacheArtifactRequest| { + deletions.lock().unwrap().push(request.local_path); + Ok(DeleteCacheArtifactResponse::default()) + } + }) + .serve(); let db = Arc::new(Mutex::new(conn)); let artifacts = LocalArtifacts::new(Arc::clone(&db)); - let report = gc_tick( + let ownership = GcOwnership { + node_manager: node_manager.connector(), + ..test_ownership() + }; + let mut scan = ArtifactScan::default(); + let report = within_node_manager_wait(gc_tick( artifacts, Arc::clone(&db), test_cfg(), now_secs(), full_disk(), - &test_ownership(), - &mut ArtifactScan::default(), - ) + &ownership, + &mut scan, + )) .await .unwrap(); assert!(report.engaged); assert_eq!(report.deleted, 1); assert!(report.bytes_freed >= 4096); assert!(!std::path::Path::new(&path).exists()); + let staging = format!("{path}.gc-pending-{TEST_ART_ID}"); + // The fake acknowledges without deleting, so the staging directory stays. + assert!(std::path::Path::new(&staging).is_dir()); + assert_eq!(*deletions.lock().unwrap(), vec![staging]); + assert_eq!(node_manager.calls(), vec!["DeleteCacheArtifact".to_owned()]); let conn = db.lock().expect("mutex poisoned"); let count: i64 = conn .query_row("SELECT COUNT(*) FROM local_artifacts", [], |r| r.get(0))