diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs index b23739c2f6..d68a35ae68 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs @@ -394,3 +394,6 @@ pub(super) fn sidecar_rootfs_send_source_id( mod backup_policy_tests; #[cfg(test)] mod tests; + +#[cfg(test)] +mod teardown_backup_tests; diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/artifact_backup.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/artifact_backup.rs index 1c2b7ff6fc..c255036780 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/artifact_backup.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/artifact_backup.rs @@ -490,7 +490,7 @@ impl SandboxManagerService { Ok(current_id) } - async fn existing_ready_artifact_backup_response( + pub(super) async fn existing_ready_artifact_backup_response( &self, artifact: &crate::db::LocalArtifactRow, ) -> Result, Status> { diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_backup_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_backup_tests.rs new file mode 100644 index 0000000000..5b1d6bd5c7 --- /dev/null +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_backup_tests.rs @@ -0,0 +1,171 @@ +use super::*; +use pplx_space_sbx_mgr_client::proto::SuspendSnapshotFormat; + +use pplx_space_node_api_client::proto::{ + GetArtifactBackupLocationRequest, GetArtifactBackupLocationResponse, +}; + +#[derive(Clone)] +struct ReadyBackup; + +impl tonic::server::NamedService for ReadyBackup { + const NAME: &'static str = "space.node.api.v1.ClusterApi"; +} + +impl tonic::server::UnaryService for ReadyBackup { + type Response = GetArtifactBackupLocationResponse; + type Future = std::future::Ready, Status>>; + + fn call(&mut self, request: Request) -> Self::Future { + let request = request.into_inner(); + std::future::ready(Ok(Response::new(GetArtifactBackupLocationResponse { + local_artifact_id: request.local_artifact_id, + artifact_kind: request.artifact_kind, + backup_attempt_id: uuid::Uuid::now_v7().to_string(), + ..Default::default() + }))) + } +} + +impl tower::Service> for ReadyBackup +where + B: http_body::Body + Send + 'static, + B::Error: Into + Send + 'static, +{ + type Response = http::Response; + type Error = std::convert::Infallible; + type Future = futures_util::future::BoxFuture<'static, Result>; + + fn poll_ready( + &mut self, + _: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(Ok(())) + } + + fn call(&mut self, request: http::Request) -> Self::Future { + assert!(request.uri().path().ends_with("/GetArtifactBackupLocation")); + Box::pin(async move { + let mut grpc = tonic::server::Grpc::new(tonic_prost::ProstCodec::default()); + Ok(grpc.unary(ReadyBackup, request).await) + }) + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn teardown_reuses_ready_backup_but_waits_for_active_local_upload() { + use crate::service::tests::service_tests::insert_stranded_teardown_suspend_sandbox; + use pplx_space_common_core::types::ExecutionGeneration; + use std::ffi::OsStr; + + // The parent serves while the isolated child exercises environment-based + // client wiring. A second runtime thread keeps the server making progress. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let stop = tokio_util::sync::CancellationToken::new(); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(ReadyBackup) + .serve_with_incoming_shutdown( + tokio_stream::wrappers::TcpListenerStream::new(listener), + stop.clone().cancelled_owned(), + ), + ); + let token = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(token.path(), "synthetic-backup-test-token").unwrap(); + let child = pplx_space_test_env::enter_test_process(&[ + ("SPACE_INTERNAL_RPC_INSECURE_DEV", Some(OsStr::new("1"))), + ( + pplx_space_node_api_client::cluster::CLUSTER_API_ENDPOINT_ENV, + Some(OsStr::new(&endpoint)), + ), + ( + pplx_space_common_svc::bearer_token::SPACE_CLUSTER_API_TOKEN_PATH_ENV, + Some(token.path().as_os_str()), + ), + ]); + if child { + for (superseded, active_upload) in [(false, false), (true, false), (false, true)] { + let conn = crate::db::test_conn(); + let sandbox_id = SandboxId::new(); + let generation = ExecutionGeneration::new(7).unwrap(); + let artifact_id = + insert_stranded_teardown_suspend_sandbox(&conn, sandbox_id, generation); + let row = crate::db::get_sandbox(&conn, &sandbox_id.to_string()) + .unwrap() + .unwrap(); + let artifact = crate::db::get_local_artifact(&conn, artifact_id) + .unwrap() + .unwrap(); + if active_upload { + pplx_space_common_jobs::db::insert_job( + &conn, + pplx_space_common_jobs::db::InsertJob { + job_id: &JobId::new(), + kind: crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND, + payload: "{}", + subject_id: Some(&sandbox_id.to_string()), + dedup_key: Some(&artifact_id.to_string()), + max_attempts: 1, + traceparent: None, + priority: Default::default(), + }, + ) + .unwrap(); + } + let service = SandboxManagerService::new(conn); + let pending = service.ensure_teardown_suspend_backup( + &row, + super::TeardownSuspendBackup { + execution_generation: if superseded { + ExecutionGeneration::new(6).unwrap() + } else { + generation + }, + artifact: &artifact, + format: SuspendSnapshotFormat::Full, + }, + ); + if active_upload { + // A real loopback RPC returns Ready promptly, but that must not + // complete teardown while this locally persisted upload is active. + tokio::time::timeout(std::time::Duration::from_millis(250), pending) + .await + .expect_err("teardown must wait for the active local upload"); + let db = service.db.lock().unwrap(); + assert!( + !crate::db::teardown_suspend_status(&db, &sandbox_id.to_string()) + .unwrap() + .unwrap() + .backup_completed + ); + continue; + } + let result = pending.await; + if superseded { + assert_eq!(result.unwrap_err().code(), tonic::Code::Aborted); + } else { + assert!( + result.is_ok(), + "{result:?}; lookup: {:?}", + service + .existing_ready_artifact_backup_response(&artifact) + .await + ); + } + let db = service.db.lock().unwrap(); + let status = crate::db::teardown_suspend_status(&db, &sandbox_id.to_string()) + .unwrap() + .unwrap(); + assert!(status.backup_completed); + assert!( + pplx_space_common_jobs::db::list_jobs(&db, Some(&sandbox_id.to_string()), None, 0) + .unwrap() + .iter() + .all(|job| job.kind != crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND) + ); + } + } + stop.cancel(); + server.await.unwrap().unwrap(); +} diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_suspend.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_suspend.rs index 88393deedc..feeed29fb0 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_suspend.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_suspend.rs @@ -873,6 +873,19 @@ impl SandboxManagerService { crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND, &artifact_dedup_key, )?; + // An active local upload still owns its completion. Only probe before + // enqueueing or joining a terminal job: a restored artifact can already + // be durable without a local completion marker. A failed probe leaves + // the ordinary upload path responsible for durability. + if persisted_job.as_ref().is_none_or(|job| job.is_terminal()) + && matches!( + self.existing_ready_artifact_backup_response(backup.artifact) + .await, + Ok(Some(_)) + ) + { + return self.verify_teardown_suspend_backup_ownership(sandbox, &backup); + } let job_id = match persisted_job { Some(job) if !job.is_terminal() @@ -936,7 +949,14 @@ impl SandboxManagerService { Status::internal(format!("persist durable suspend completion: {error}")) })?; } + self.verify_teardown_suspend_backup_ownership(sandbox, &backup) + } + fn verify_teardown_suspend_backup_ownership( + &self, + sandbox: &crate::db::SandboxRow, + backup: &TeardownSuspendBackup<'_>, + ) -> Result<(), Status> { let still_owned = { let db = self.db_conn()?; crate::db::teardown_suspend_status(&db, &sandbox.id) diff --git a/crates/libs/sbx/mgr/server/src/snapshot_upload.rs b/crates/libs/sbx/mgr/server/src/snapshot_upload.rs index 70d943dde3..057c2113e0 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_upload.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_upload.rs @@ -721,8 +721,16 @@ pub(crate) fn write_snapshot_metadata_to( }; let meta_json = serde_json::to_string_pretty(&written)?; std::fs::create_dir_all(metadata_dir).map_err(SnapshotMetadataWriteError::CreateMetadataDir)?; - std::fs::write(metadata_dir.join(SNAPSHOT_METADATA_FILENAME), &meta_json) - .map_err(SnapshotMetadataWriteError::WriteMetadata)?; + // Restores may leave a sidecar owned by volume-manager. Directory write + // permission lets us replace it without writing another service's inode; + // readers keep a complete old or new document even across a crash. + replace_file_durably( + metadata_dir, + ".snapshot.json.upload", + SNAPSHOT_METADATA_FILENAME, + meta_json.as_bytes(), + ) + .map_err(SnapshotMetadataWriteError::WriteMetadata)?; Ok(meta_final) } @@ -736,9 +744,12 @@ pub(crate) fn replace_file_durably( bytes: &[u8], ) -> std::io::Result<()> { use std::io::Write as _; + use std::os::unix::fs::PermissionsExt; let temporary = dir.join(temporary_name); let mut file = std::fs::File::create(&temporary)?; file.write_all(bytes)?; + // Volume-manager reads metadata published by sandbox-manager. + file.set_permissions(std::fs::Permissions::from_mode(0o644))?; file.sync_all()?; drop(file); std::fs::rename(&temporary, dir.join(final_name))?; diff --git a/crates/libs/sbx/mgr/server/src/snapshot_upload/tests.rs b/crates/libs/sbx/mgr/server/src/snapshot_upload/tests.rs index 1fe33ddb31..b7465b13d4 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_upload/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_upload/tests.rs @@ -381,7 +381,18 @@ fn write_snapshot_metadata_sizes_from_local_artifacts_at_submit_time() { std::fs::write(dir.join("rootfs.ext4"), vec![0u8; 30]).unwrap(); let meta = fixture_meta(None); + let metadata_path = dir.join(SNAPSHOT_METADATA_FILENAME); + std::fs::write(&metadata_path, b"previous metadata").unwrap(); + // Restored metadata can be readable but not writable by this service. + // A reader that already opened it must retain the previous complete bytes. + use std::io::Read; + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&metadata_path, std::fs::Permissions::from_mode(0o444)).unwrap(); + let mut previous = std::fs::File::open(&metadata_path).unwrap(); let out = write_snapshot_metadata(&meta, dir, &vm, None).unwrap(); + let mut previous_bytes = Vec::new(); + previous.read_to_end(&mut previous_bytes).unwrap(); + assert_eq!(previous_bytes, b"previous metadata"); assert_eq!(out.size_bytes, 60, "size is the sum of local VM artifacts"); let written = std::fs::read_to_string(dir.join(SNAPSHOT_METADATA_FILENAME)).unwrap();