From 8d7088c6700465471e792c3f3a23204745b3a6a8 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Mon, 12 Oct 2026 21:07:13 +0000 Subject: [PATCH] sbx-mgr: test create after template lookup, unpersisted cleanup and endpoint capture on the seams Tests only. These paths were deferred because they reached fixed template-manager or node-manager sockets. They now run against FakeTemplateResolver and FakeNodeManager. Every test that builds a service runs in a child process without the feature-flag provider settings, so construction cannot contact a provider. Create after template lookup (grpc_core/create_after_lookup_tests.rs). A create that inserts a new row goes on to create the sandbox's cache root under the fixed host cache directory. Each service here therefore carries a temporary SQLite trigger that refuses every new sandboxes row and reports the row in its refusal, so a create that reaches the insert stops there. The tests pin create's own delta over the lower-layer intent tests: - A fresh create inserts a Creating row with the resolved build hash, the request's owner and environment, the template's default network authority, and the template's default auto-lifecycle policy and authority unless the request carries a policy. It honours a requested id, and an empty pool falls back to a cold create. - An autostart create asks node-manager's GetNetworkPoolStatus before the insert. No ready slot refuses ResourceExhausted, and a refused status passes its code through. - A pool create adopts a paused donor of the resolved build. The donor takes the request's owner and environment and the template's defaults, and no row is inserted. A donor of another build is never adopted and is left untouched. - A requested-id create against an existing compatible row is swept over every sandbox state, with the expected outcome written out per state. Created, Started, Paused, Suspended and Stopped rows are reused and take the request's policy. CleanupRequired and CleanupFailed refuse with retryable Unavailable. Every other state, Quarantined included, refuses AlreadyExists without touching the row. A row of another build refuses AlreadyExists with or without a policy change. Reuse keeps the row's own environment. Each create is bounded on real time. Unpersisted stop and delete (grpc_core/unpersisted_cleanup_tests.rs). A trigger refuses the stopping or deleting claim write. Stop (with immediate_kill, which skips the guest flush) and delete, blocking or not, then clean the runtime up through node-manager's CleanupSandbox with the expected keep-filesystem, slot-release and kill options. They report Internal "runtime cleanup completed but ... could not be persisted", or node-manager's refusal, and leave the row, episode, jobs and dirty refs untouched. Endpoint capture (service/tests/endpoint_capture_tests.rs). Opening and closing a loopback bridge on a restoring sandbox as node-ctl, and a trusted node-api readiness wait, each ask node-manager to capture this sandbox's endpoint proof, and surface its refusal as PermissionDenied "spaced endpoint attestation failed". With node-manager down, each fails retryably with Unavailable "spaced endpoint proof is temporarily unavailable", before any spaced-host connection. This replaces five tests that inferred the same from an unreachable node-manager; two pairs of them had identical bodies. Snapshot GC. reaps_idle_uploaded_unpinned_artifact asserts that the reaped artifact's .gc-pending staging directory is handed to node-manager's DeleteCacheArtifact, bounded on real time because the test's clock is paused. AI assistance: written with Claude Code. --- .../sbx/mgr/server/src/service/grpc_core.rs | 4 + .../grpc_core/create_after_lookup_tests.rs | 672 ++++++++++++++++++ .../grpc_core/unpersisted_cleanup_tests.rs | 210 ++++++ .../libs/sbx/mgr/server/src/service/tests.rs | 206 +----- .../service/tests/endpoint_capture_tests.rs | 150 ++++ .../sbx/mgr/server/src/snapshot_gc/tests.rs | 62 +- 6 files changed, 1097 insertions(+), 207 deletions(-) create mode 100644 crates/libs/sbx/mgr/server/src/service/grpc_core/create_after_lookup_tests.rs create mode 100644 crates/libs/sbx/mgr/server/src/service/grpc_core/unpersisted_cleanup_tests.rs create mode 100644 crates/libs/sbx/mgr/server/src/service/tests/endpoint_capture_tests.rs 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))