Recover teardown of restored snapshots with read-only metadata

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;

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<Option<EnqueueArtifactBackupResponse>, Status> {

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<GetArtifactBackupLocationRequest> for ReadyBackup {
+    type Response = GetArtifactBackupLocationResponse;
+    type Future = std::future::Ready<Result<Response<Self::Response>, Status>>;
+
+    fn call(&mut self, request: Request<GetArtifactBackupLocationRequest>) -> 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<B> tower::Service<http::Request<B>> for ReadyBackup
+where
+    B: http_body::Body + Send + 'static,
+    B::Error: Into<tonic::codegen::StdError> + Send + 'static,
+{
+    type Response = http::Response<tonic::body::Body>;
+    type Error = std::convert::Infallible;
+    type Future = futures_util::future::BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(
+        &mut self,
+        _: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), Self::Error>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, request: http::Request<B>) -> 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();
+}

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)

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))?;

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();