From 38e9d2e221ce0a080242f31957f441b9b556f429 Mon Sep 17 00:00:00 2001 From: Mostafa Abdelraouf Date: Tue, 13 Oct 2026 03:07:56 +0000 Subject: [PATCH 1/2] fix(sbx): re-drive failed suspend backups behind admission gate Co-authored-by: pplx-teammatebot --- crates/libs/common/jobs/src/worker.rs | 38 +++ crates/libs/common/jobs/src/worker/tests.rs | 30 +++ crates/libs/metrics/src/snapshot.rs | 5 + crates/libs/sbx/mgr/server/src/db.rs | 6 +- .../sbx/mgr/server/src/db/teardown_jobs.rs | 87 ++++++- .../mgr/server/src/service/grpc_lifecycle.rs | 1 + .../service/grpc_lifecycle/artifact_backup.rs | 2 + .../grpc_lifecycle/backup_policy_tests.rs | 220 ++++++++++++++++++ .../grpc_lifecycle/backup_submission.rs | 26 ++- .../grpc_lifecycle/suspend_snapshot.rs | 1 + .../grpc_lifecycle/teardown_suspend.rs | 169 ++++++++++++-- crates/libs/sbx/mgr/server/src/service/mod.rs | 3 + .../sbx/mgr/server/src/service/recovery.rs | 16 ++ .../mgr/server/src/snapshot_upload_jobs.rs | 129 ++++++++-- .../server/src/snapshot_upload_jobs/tests.rs | 110 ++++++++- docs/snapshot-upload-streaming.md | 32 +++ 16 files changed, 835 insertions(+), 40 deletions(-) diff --git a/crates/libs/common/jobs/src/worker.rs b/crates/libs/common/jobs/src/worker.rs index 61575fe911..5e6e085558 100644 --- a/crates/libs/common/jobs/src/worker.rs +++ b/crates/libs/common/jobs/src/worker.rs @@ -635,6 +635,31 @@ impl BackgroundWorker { ) } + /// Check caller-owned state under the same database lock as durable job + /// admission. Refusal performs no insert or deduplication side effects. + pub fn submit_json_guarded( + &self, + kind: &str, + payload: &T, + subject_id: Option<&str>, + dedup_key: Option<&str>, + admit: impl FnOnce(&rusqlite::Connection) -> Result<(), Status>, + ) -> Result { + let json = serde_json::to_string(payload) + .map_err(|error| Status::invalid_argument(format!("encode job payload: {error}")))?; + self.submit_with_options_guarded( + kind, + &json, + subject_id, + dedup_key, + SubmitOptions { + priority: JobPriority::default(), + duplicate_policy: DuplicatePolicy::ReuseExisting, + }, + admit, + ) + } + pub fn submit_json_with_priority( &self, kind: &str, @@ -725,12 +750,25 @@ impl BackgroundWorker { subject_id: Option<&str>, dedup_key: Option<&str>, options: SubmitOptions, + ) -> Result { + self.submit_with_options_guarded(kind, payload, subject_id, dedup_key, options, |_| Ok(())) + } + + fn submit_with_options_guarded( + &self, + kind: &str, + payload: &str, + subject_id: Option<&str>, + dedup_key: Option<&str>, + options: SubmitOptions, + admit: impl FnOnce(&rusqlite::Connection) -> Result<(), Status>, ) -> Result { let SubmitOptions { priority, duplicate_policy, } = options; let conn = self.lock_db_for_status()?; + admit(&conn)?; let existing = match (dedup_key, duplicate_policy) { (Some(dk), DuplicatePolicy::ReuseExistingOrCompleted) => { diff --git a/crates/libs/common/jobs/src/worker/tests.rs b/crates/libs/common/jobs/src/worker/tests.rs index 8767361ddb..bf8b54bb29 100644 --- a/crates/libs/common/jobs/src/worker/tests.rs +++ b/crates/libs/common/jobs/src/worker/tests.rs @@ -14,6 +14,36 @@ struct TestExecutor { should_block: Arc, } +#[tokio::test] +async fn guarded_submission_refusal_neither_inserts_nor_reuses_work() { + let conn = Connection::open_in_memory().unwrap(); + init_test_schema(&conn); + let worker = BackgroundWorker::new( + Arc::new(Mutex::new(conn)), + TestExecutor::blocking(), + WorkerConfig::new(1, Duration::from_secs(30), 1), + ); + let existing = worker + .submit("backup", "{}", Some("sandbox"), Some("artifact")) + .unwrap(); + for key in ["artifact", "new-artifact"] { + let error = worker + .submit_json_guarded( + "backup", + &serde_json::json!({}), + Some("sandbox"), + Some(key), + |_| Err(Status::aborted("generation retired")), + ) + .unwrap_err(); + assert_eq!(error.code(), tonic::Code::Aborted); + } + let conn = worker.lock_db_for_status().unwrap(); + let jobs = db::list_jobs(&conn, Some("sandbox"), None, 0).unwrap(); + assert_eq!(jobs.len(), 1); + assert_eq!(jobs[0].job_id, existing); +} + impl TestExecutor { fn blocking() -> Self { Self { diff --git a/crates/libs/metrics/src/snapshot.rs b/crates/libs/metrics/src/snapshot.rs index cf688a4016..fa236fd490 100644 --- a/crates/libs/metrics/src/snapshot.rs +++ b/crates/libs/metrics/src/snapshot.rs @@ -38,6 +38,11 @@ use std::time::Instant; use crate::emit::{counter, elapsed_ms, gauge, tag, timing_ms}; +/// A fresh submission attempt, not an observation or an individual vol-mgr retry. +pub fn backup_redrive() { + counter("snapshot.backup.redrive", 1, &[]); +} + pub fn delivery_duration_ms( stage: &str, origin: &str, diff --git a/crates/libs/sbx/mgr/server/src/db.rs b/crates/libs/sbx/mgr/server/src/db.rs index 5f73d8b8e5..23428edfaf 100644 --- a/crates/libs/sbx/mgr/server/src/db.rs +++ b/crates/libs/sbx/mgr/server/src/db.rs @@ -52,9 +52,11 @@ mod quarantine_scan; mod sandbox_row; mod teardown_jobs; pub(crate) use teardown_jobs::{ - TeardownSuspendFailure, defer_teardown_suspend_retry, fail_teardown_suspend_backup_operation, + TeardownSuspendFailure, defer_teardown_suspend_retry, due_teardown_suspend_redrives, + fail_teardown_suspend_artifact_unavailable, fail_teardown_suspend_backup_operation, fail_teardown_suspend_operation, reconcile_terminal_teardown_suspend_jobs, - teardown_suspend_retry_admissible, teardown_suspend_retry_due, + teardown_suspend_owns_backup_source, teardown_suspend_retry_admissible, + teardown_suspend_retry_due, }; #[cfg(test)] mod test_schema_image_cache; diff --git a/crates/libs/sbx/mgr/server/src/db/teardown_jobs.rs b/crates/libs/sbx/mgr/server/src/db/teardown_jobs.rs index 35458d27c9..5841245ece 100644 --- a/crates/libs/sbx/mgr/server/src/db/teardown_jobs.rs +++ b/crates/libs/sbx/mgr/server/src/db/teardown_jobs.rs @@ -20,6 +20,7 @@ pub(crate) enum TeardownSuspendFailure { Quarantined, Failed, BackupUnavailable, + ArtifactUnavailable, Reconciled, } @@ -32,6 +33,9 @@ impl TeardownSuspendFailure { Self::Quarantined => "sandbox is quarantined", Self::Failed => "generation suspend permanently failed", Self::BackupUnavailable => "backup cannot complete: cluster-api sandbox not found", + Self::ArtifactUnavailable => { + "backup cannot complete: local suspend artifact unavailable" + } Self::Reconciled => { "teardown suspend reconciled: sandbox missing, deleting or quarantined" } @@ -45,6 +49,39 @@ impl std::fmt::Display for TeardownSuspendFailure { } } +/// Caller holds the local database lock through job admission. Checking the +/// captured artifact as well as the operation prevents a late retry from +/// publishing the next execution's bytes. +pub(crate) fn teardown_suspend_owns_backup_source( + conn: &Connection, + id: &str, + generation: ExecutionGeneration, + artifact_id: LocalArtifactId, + local_path: &str, +) -> Result { + use crate::state::SandboxState; + if super::runtime_execution_generation(conn, id)?.is_some_and(|current| current > generation) { + return Ok(false); + } + Ok( + super::teardown_suspend_status(conn, id)?.is_some_and(|status| { + matches!( + status.state, + SandboxState::Suspending + | SandboxState::Suspended + | SandboxState::CleanupRequired + | SandboxState::CleanupFailed + ) && status.operation.is_some_and(|operation| { + operation.status.is_active() && operation.execution_generation == generation + }) && status.artifact.is_some_and(|artifact| { + artifact.local_artifact_id == artifact_id + && artifact.sandbox_id == id + && artifact.local_path == local_path + }) + }), + ) +} + /// A recovered resident may start a fresh attempt; retained artifacts and /// in-flight cleanup cannot supply that authority, even for the same generation. pub(crate) fn teardown_suspend_retry_admissible( @@ -93,6 +130,21 @@ pub(crate) fn fail_teardown_suspend_backup_operation( ) } +pub(crate) fn fail_teardown_suspend_artifact_unavailable( + conn: &Connection, + id: &str, + generation: ExecutionGeneration, + failed_artifact: LocalArtifactId, +) -> Result { + fail_operation( + conn, + id, + generation, + TeardownSuspendFailure::ArtifactUnavailable, + Some(failed_artifact), + ) +} + fn fail_operation( conn: &Connection, id: &str, @@ -195,10 +247,10 @@ pub(crate) fn reconcile_terminal_teardown_suspend_jobs(conn: &Connection) -> Res Ok(jobs.len()) } -const TEARDOWN_SUSPEND_RETRY_DELAY_SECS: i64 = 30; - /// Persist retry pacing on the generation's existing job, so daemon restarts /// and frequent shim polls cannot turn a blocked backup into a tight loop. +/// These jobs have no BackgroundWorker; after the initial running claim, +/// attempts counts failed driver runs. pub(crate) fn defer_teardown_suspend_retry( conn: &Connection, id: &str, @@ -210,10 +262,12 @@ pub(crate) fn defer_teardown_suspend_retry( { conn.execute( sql!( - "UPDATE jobs SET next_attempt_at = datetime('now', '+' || ?3 || ' seconds'), error = ?2 + "UPDATE jobs SET + next_attempt_at = datetime('now', '+' || min(900, 30 * (1 << min(max(attempts - 1, 0), 5))) || ' seconds'), + attempts = min(attempts + 1, 30), error = ?2 WHERE job_id = ?1 AND status IN ('pending', 'running')" ), - params![operation.job_id, error, TEARDOWN_SUSPEND_RETRY_DELAY_SECS], + params![operation.job_id, error], )?; } Ok(()) @@ -239,3 +293,28 @@ pub(crate) fn teardown_suspend_retry_due( |row| row.get(0), ) } + +/// Bounded discovery for retries whose original RPC/shim has disappeared. +/// Running drivers are deduplicated by the service's existing driver slots. +pub(crate) fn due_teardown_suspend_redrives( + conn: &Connection, +) -> Result> { + let mut statement = conn.prepare(sql!( + "SELECT subject_id FROM jobs + WHERE kind = ?1 AND status IN ('pending', 'running') + AND (next_attempt_at IS NULL OR next_attempt_at <= datetime('now')) + ORDER BY next_attempt_at, created_at LIMIT 128" + ))?; + let ids = statement + .query_map([super::TEARDOWN_SUSPEND_JOB_KIND], |row| { + row.get::<_, String>(0) + })? + .collect::>>()?; + ids.into_iter() + .filter_map(|id| match active_teardown_suspend_operation(conn, &id) { + Ok(Some(operation)) => Some(Ok((id, operation.execution_generation))), + Ok(None) => None, + Err(error) => Some(Err(error)), + }) + .collect() +} 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 d68a35ae68..0622069f0b 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle.rs @@ -317,6 +317,7 @@ struct ArtifactBackupSubmission<'a> { format: pplx_space_sbx_mgr_client::proto::SuspendSnapshotFormat, dedup_key: Option<&'a str>, handoff: ArtifactBackupHandoff, + execution_generation: Option, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] 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 b2c59a5ee1..aef2b9d65b 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 @@ -921,6 +921,7 @@ impl SandboxManagerService { .submit_artifact_backup( &sandbox, ArtifactBackupSubmission { + execution_generation: None, local_artifact_id: artifact.local_artifact_id, artifact_kind: artifact.artifact_kind, artifact_origin: artifact.artifact_origin, @@ -1129,6 +1130,7 @@ impl SandboxManagerService { .submit_artifact_backup( &sandbox, ArtifactBackupSubmission { + execution_generation: None, local_artifact_id, artifact_kind: pplx_space_common_core::types::ArtifactKind::Full, artifact_origin: pplx_space_common_core::types::ArtifactOrigin::Suspend, diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs index 1406d5f6d0..86f8cea3de 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_policy_tests.rs @@ -349,6 +349,225 @@ fn write_artifact(dir: &std::path::Path, sandbox_id: &str, files: &[&str]) { .unwrap(); } +#[tokio::test] +async fn exhausted_suspend_backup_admits_only_the_owned_valid_artifact() { + use crate::service::tests::service_tests::insert_stranded_teardown_suspend_sandbox; + use pplx_space_common_core::types::{ExecutionGeneration, JobId}; + use pplx_space_common_jobs::db as jobs; + let root = tempfile::tempdir().unwrap(); + let token_path = root.path().join("absent-token"); + if !pplx_space_test_env::enter_test_process(&cluster_api_env( + LOOPBACK_ENDPOINT, + STATIC_TOKEN, + &token_path, + )) { + return; + } + for case in [ + "valid", + "missing", + "invalid_metadata", + "stale_generation", + "newer_runtime", + "retired", + "wrong_artifact", + ] { + let conn = crate::db::test_conn(); + let id = SandboxId::new(); + let generation = ExecutionGeneration::new(7).unwrap(); + let artifact_id = insert_stranded_teardown_suspend_sandbox(&conn, id, generation); + let path = root.path().join(case); + write_artifact( + &path, + &id.to_string(), + &["vmstate", "memory", "rootfs.ext4"], + ); + conn.execute( + "UPDATE local_artifacts SET local_path = ?1 WHERE local_artifact_id = ?2", + rusqlite::params![path.to_str().unwrap(), artifact_id.to_string()], + ) + .unwrap(); + let mut artifact = crate::db::get_local_artifact(&conn, artifact_id) + .unwrap() + .unwrap(); + let sandbox = crate::db::get_sandbox(&conn, &id.to_string()) + .unwrap() + .unwrap(); + let old_job = JobId::new(); + jobs::insert_job( + &conn, + jobs::InsertJob { + job_id: &old_job, + kind: crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND, + payload: "{}", + subject_id: Some(&id.to_string()), + dedup_key: Some(&artifact_id.to_string()), + max_attempts: 5, + traceparent: None, + priority: Default::default(), + }, + ) + .unwrap(); + jobs::mark_job_running(&conn, &old_job).unwrap(); + jobs::fail_running_job_retriable_exhausted(&conn, &old_job, "failed transfer").unwrap(); + if case == "missing" { + std::fs::remove_file(path.join("memory")).unwrap(); + } + if case == "invalid_metadata" { + std::fs::write( + path.join(crate::snapshot_upload::SNAPSHOT_METADATA_FILENAME), + b"{", + ) + .unwrap(); + } + if case == "retired" { + crate::db::fail_teardown_suspend_operation( + &conn, + &id.to_string(), + generation, + crate::db::TeardownSuspendFailure::Deleted, + ) + .unwrap(); + } + if case == "newer_runtime" { + conn.execute( + "UPDATE sandboxes SET space_api_route_execution_generation = 8 WHERE id = ?1", + [id.to_string()], + ) + .unwrap(); + } + if case == "wrong_artifact" { + artifact.local_artifact_id = LocalArtifactId::new(); + } + let mut service = SandboxManagerService::new(conn); + service.suspend_backup_redrive_enabled = true; + service.snapshot_upload_config = Some(Arc::new(upload_config("synthetic-backups"))); + let error = service + .ensure_teardown_suspend_backup( + &sandbox, + super::TeardownSuspendBackup { + execution_generation: if case == "stale_generation" { + ExecutionGeneration::new(6).unwrap() + } else { + generation + }, + artifact: &artifact, + format: SuspendSnapshotFormat::Full, + }, + ) + .await + .unwrap_err(); + // The injected missing token refuses registration after admission. No + // node-manager, volume-manager or ambient cluster connection is used. + assert_eq!( + snapshot_upload_jobs(&service, &id.to_string()), + if case == "valid" { 2 } else { 1 }, + "{case}: {error}" + ); + if case == "valid" { + let job = service + .snapshot_upload_worker + .find_job_by_dedup_key( + crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND, + &artifact_id.to_string(), + ) + .unwrap() + .unwrap(); + assert_ne!(job.job_id, old_job); + let payload = job + .payload_as::() + .unwrap(); + assert_eq!(payload.meta.local_artifact_id, Some(artifact_id)); + assert_eq!(payload.snapshot_dir, path); + } else if matches!(case, "missing" | "invalid_metadata") { + let db = service.db_conn().unwrap(); + assert_eq!( + crate::db::latest_teardown_suspend_operation(&db, &id.to_string()) + .unwrap() + .unwrap() + .status, + crate::db::TeardownSuspendOperationStatus::Failed, + "{case}: unavailable artifact must remain terminal", + ); + } else { + assert_eq!(error.code(), tonic::Code::Aborted, "{case}"); + } + } +} + +#[test] +fn suspend_redrive_backoff_is_durable_capped_and_generation_fenced() { + use crate::service::tests::service_tests::insert_stranded_teardown_suspend_sandbox; + use pplx_space_common_core::types::ExecutionGeneration; + let conn = crate::db::test_conn(); + let id = SandboxId::new().to_string(); + let generation = ExecutionGeneration::new(7).unwrap(); + insert_stranded_teardown_suspend_sandbox(&conn, id.parse().unwrap(), generation); + let operation = crate::db::latest_teardown_suspend_operation(&conn, &id) + .unwrap() + .unwrap(); + assert_eq!( + crate::db::due_teardown_suspend_redrives(&conn).unwrap(), + vec![(id.clone(), generation)] + ); + for delay in [30, 60, 120, 240, 480, 900, 900] { + crate::db::defer_teardown_suspend_retry(&conn, &id, generation, "unavailable").unwrap(); + let remaining: i64 = conn + .query_row( + "SELECT unixepoch(next_attempt_at) - unixepoch('now') FROM jobs WHERE job_id = ?1", + [operation.job_id], + |row| row.get(0), + ) + .unwrap(); + assert!( + (delay - 1..=delay).contains(&remaining), + "{remaining} != {delay}" + ); + assert!(!crate::db::teardown_suspend_retry_due(&conn, &id, generation).unwrap()); + assert!( + crate::db::due_teardown_suspend_redrives(&conn) + .unwrap() + .is_empty() + ); + } + let before = pplx_space_common_jobs::db::get_job(&conn, &operation.job_id) + .unwrap() + .unwrap(); + crate::db::defer_teardown_suspend_retry( + &conn, + &id, + ExecutionGeneration::new(6).unwrap(), + "stale", + ) + .unwrap(); + let after = pplx_space_common_jobs::db::get_job(&conn, &operation.job_id) + .unwrap() + .unwrap(); + assert_eq!(before.attempts, after.attempts); + assert_eq!(before.error, after.error); + conn.execute( + "UPDATE jobs SET next_attempt_at = datetime('now', '-1 second') WHERE job_id = ?1", + [operation.job_id], + ) + .unwrap(); + assert_eq!( + crate::db::due_teardown_suspend_redrives(&conn).unwrap(), + vec![(id.clone(), generation)] + ); + crate::db::fail_teardown_suspend_operation( + &conn, + &id, + generation, + crate::db::TeardownSuspendFailure::Deleted, + ) + .unwrap(); + assert!( + crate::db::due_teardown_suspend_redrives(&conn) + .unwrap() + .is_empty() + ); +} + #[tokio::test] async fn durable_artifact_backup_persists_the_submitted_artifact_identity() { // A regressed handoff that began the backup remotely would stop at the @@ -447,6 +666,7 @@ async fn durable_artifact_backup_persists_the_submitted_artifact_identity() { .submit_artifact_backup( &sandbox, ArtifactBackupSubmission { + execution_generation: None, local_artifact_id, artifact_kind: case.kind, artifact_origin: case.origin, diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_submission.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_submission.rs index 0053efe075..e7fe7bb710 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_submission.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/backup_submission.rs @@ -74,6 +74,7 @@ impl SandboxManagerService { self.submit_artifact_backup( sandbox, ArtifactBackupSubmission { + execution_generation: None, local_artifact_id, artifact_kind: pplx_space_common_core::types::ArtifactKind::Full, artifact_origin: pplx_space_common_core::types::ArtifactOrigin::Suspend, @@ -223,6 +224,7 @@ impl SandboxManagerService { format, dedup_key, handoff, + execution_generation, } = submission; let config: &crate::snapshot_upload::SnapshotUploadConfig = match request_override { Some(c) => c, @@ -297,13 +299,33 @@ impl SandboxManagerService { ) .map_err(|error| format!("prepare snapshot upload intent failed: {error}"))?; let mut storage_uri = jobs.storage_uri.clone(); + let source_artifact_id = local_artifact_id; let local_artifact_id = local_artifact_id.to_string(); - let job_id = match self.snapshot_upload_worker.submit_json( + let job_id = match self.snapshot_upload_worker.submit_json_guarded( crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND, &jobs, Some(&sandbox.id), dedup_key, + |db| { + if let Some(generation) = execution_generation + && !crate::db::teardown_suspend_owns_backup_source( + db, + &sandbox.id, + generation, + source_artifact_id, + snapshot_dir, + ) + .map_err(|error| { + Status::internal(format!("read suspend backup ownership: {error}")) + })? + { + return Err(Status::aborted( + "generation suspend no longer owns backup source", + )); + } + Ok(()) + }, ) { Ok(id) => id, Err(e) => { @@ -476,6 +498,7 @@ impl SandboxManagerService { snapshot_dir: &str, format: pplx_space_sbx_mgr_client::proto::SuspendSnapshotFormat, dedup_key: Option<&str>, + execution_generation: Option, ) -> Result { let request_override = match self.resolve_org_snapshot_upload_config(&sandbox.id).await { OrgSnapshotConfigResolution::Resolved(config) => Some(config), @@ -503,6 +526,7 @@ impl SandboxManagerService { format, dedup_key, handoff: ArtifactBackupHandoff::TeardownFence, + execution_generation, }, ) .await diff --git a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend_snapshot.rs b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend_snapshot.rs index 1460d63154..0c25589686 100644 --- a/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend_snapshot.rs +++ b/crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/suspend_snapshot.rs @@ -965,6 +965,7 @@ impl SandboxManagerService { &snapshot_dir_string, published_format, None, + None, ) .await .map_err(|error| artifact_backup_fence_status(error.to_string())); 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 6cc6b2e5fe..4dcff63e6e 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 @@ -897,6 +897,7 @@ impl SandboxManagerService { sandbox: &crate::db::SandboxRow, backup: TeardownSuspendBackup<'_>, ) -> Result<(), Status> { + self.ensure_teardown_suspend_backup_source(sandbox, &backup)?; let artifact_dedup_key = backup.artifact.local_artifact_id.to_string(); let persisted_job = self.snapshot_upload_worker.find_job_by_dedup_key( crate::snapshot_upload_jobs::SNAPSHOT_UPLOAD_JOB_KIND, @@ -915,32 +916,139 @@ impl SandboxManagerService { { return self.verify_teardown_suspend_backup_ownership(sandbox, &backup); } + let persisted_job_was_failed = persisted_job + .as_ref() + .is_some_and(|job| job.failure.is_some()); + let mut failed_transfer = false; + if self.suspend_backup_redrive_enabled + && let Some(job) = persisted_job.as_ref() + && job.is_terminal() + && matches!( + job.failure, + Some(pplx_space_common_jobs::JobFailure::Permanent { .. }) + ) + { + let payload = job + .payload_as::() + .map_err(|error| { + Status::failed_precondition(format!("cannot inspect failed backup: {error}")) + })?; + if payload.meta.local_artifact_id == Some(backup.artifact.local_artifact_id) + && payload.meta.sandbox_id == sandbox.id + { + failed_transfer = crate::snapshot_upload_jobs::has_failed_transfer(&payload) + .await + .map_err(|error| Status::unavailable(error.to_string()))?; + } + } let job_id = match persisted_job { Some(job) if !job.is_terminal() - || !teardown_backup_failure_requires_new_attempt(job.failure) => + || (!failed_transfer + && !teardown_backup_failure_requires_new_attempt(job.failure)) => { job.job_id } - _ => match self - .enqueue_suspend_artifact_backup( - sandbox, - backup.artifact.local_artifact_id, - &backup.artifact.local_path, - backup.format, - Some(&artifact_dedup_key), - ) - .await - .map_err(|error| { - teardown_artifact_backup_status(error, backup.artifact.local_artifact_id) - })? { - SubmitSuspendSnapshotOutcome::Enqueued { job_id, .. } => job_id, - SubmitSuspendSnapshotOutcome::NoUploadConfig => { - return Err(Status::failed_precondition( - "generation suspend requires a configured snapshot upload destination", - )); + _ => { + if self.suspend_backup_redrive_enabled { + // The immutable published artifact, not vol-mgr's disposable + // per-attempt reflink, is the source of a new attempt. Use + // exactly the operator enqueue preflight; failed inspection + // is not permission to publish a partial capture. + let validation = async { + let metadata_path = std::path::Path::new(&backup.artifact.local_path) + .join(crate::snapshot_upload::SNAPSHOT_METADATA_FILENAME); + let bytes = tokio::fs::read(&metadata_path).await.map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + Status::failed_precondition( + "local suspend artifact metadata is missing", + ) + } else { + Status::internal(format!( + "cannot verify local suspend metadata: {error}" + )) + } + })?; + let meta: crate::snapshot_upload::SnapshotMeta = + serde_json::from_slice(&bytes).map_err(|error| { + Status::failed_precondition(format!( + "local suspend metadata is invalid: {error}" + )) + })?; + if meta.sandbox_id != sandbox.id + || meta + .local_artifact_id + .is_some_and(|id| id != backup.artifact.local_artifact_id) + { + return Err(Status::failed_precondition( + "local suspend metadata does not identify the owned artifact", + )); + } + super::artifact_backup::require_artifact_backup_files( + self.retained_source_inspector.as_ref(), + &sandbox.id, + &backup.artifact.local_path, + backup.artifact.artifact_kind, + super::artifact_backup::artifact_backup_format(backup.artifact)?, + super::artifact_backup::BackupFilesPolicy::Backup { + full_current_instance: false, + }, + ) + .await + } + .await; + if let Err(error) = validation { + tracing::error!( + sandbox_id = %sandbox.id, + execution_generation = %backup.execution_generation, + local_artifact_id = %backup.artifact.local_artifact_id, + %error, + "suspend backup cannot be re-driven: local artifact unavailable or unverified" + ); + if error.code() == tonic::Code::FailedPrecondition { + let db = self.db_conn()?; + crate::db::fail_teardown_suspend_artifact_unavailable( + &db, + &sandbox.id, + backup.execution_generation, + backup.artifact.local_artifact_id, + ) + .map_err(|error| { + Status::internal(format!( + "persist unavailable suspend artifact: {error}" + )) + })?; + } + return Err(error); + } + // Inspection may await node-manager. Never submit based on + // the generation or artifact observed before that await. + self.ensure_teardown_suspend_backup_source(sandbox, &backup)?; + } + if persisted_job_was_failed { + pplx_space_metrics::snapshot::backup_redrive(); + } + match self + .enqueue_suspend_artifact_backup( + sandbox, + backup.artifact.local_artifact_id, + &backup.artifact.local_path, + backup.format, + Some(&artifact_dedup_key), + Some(backup.execution_generation), + ) + .await + .map_err(|error| { + teardown_artifact_backup_status(error, backup.artifact.local_artifact_id) + })? { + SubmitSuspendSnapshotOutcome::Enqueued { job_id, .. } => job_id, + SubmitSuspendSnapshotOutcome::NoUploadConfig => { + return Err(Status::failed_precondition( + "generation suspend requires a configured snapshot upload destination", + )); + } } - }, + } }; let job = self .snapshot_upload_worker @@ -981,6 +1089,29 @@ impl SandboxManagerService { self.verify_teardown_suspend_backup_ownership(sandbox, &backup) } + fn ensure_teardown_suspend_backup_source( + &self, + sandbox: &crate::db::SandboxRow, + backup: &TeardownSuspendBackup<'_>, + ) -> Result<(), Status> { + let db = self.db_conn()?; + let owned = crate::db::teardown_suspend_owns_backup_source( + &db, + &sandbox.id, + backup.execution_generation, + backup.artifact.local_artifact_id, + &backup.artifact.local_path, + ) + .map_err(|error| Status::internal(format!("read suspend ownership: {error}")))?; + if owned { + Ok(()) + } else { + Err(Status::aborted( + "generation suspend no longer owns backup source", + )) + } + } + fn verify_teardown_suspend_backup_ownership( &self, sandbox: &crate::db::SandboxRow, diff --git a/crates/libs/sbx/mgr/server/src/service/mod.rs b/crates/libs/sbx/mgr/server/src/service/mod.rs index f42685dc99..da7bdfb08f 100644 --- a/crates/libs/sbx/mgr/server/src/service/mod.rs +++ b/crates/libs/sbx/mgr/server/src/service/mod.rs @@ -408,6 +408,7 @@ pub struct SandboxManagerService { gateway_authority: gateway_authority::GatewayAuthority, /// S3 upload config for suspend snapshots. `None` means uploads are disabled. snapshot_upload_config: Option>, + suspend_backup_redrive_enabled: bool, /// Durable worker that drives the poll → metadata → register tail of the /// suspend-snapshot pipeline. Persists in sbx-mgr's local SQLite so a /// mid-flight process restart resumes the upload instead of orphaning it. @@ -1732,6 +1733,8 @@ impl SandboxManagerService { ), gateway_caller_client: options.node_manager.lazy(), gateway_authority, + suspend_backup_redrive_enabled: + crate::snapshot_upload_jobs::suspend_backup_redrive_enabled(), snapshot_upload_config: { let cfg = crate::snapshot_upload::SnapshotUploadConfig::from_env(); if cfg.is_none() { diff --git a/crates/libs/sbx/mgr/server/src/service/recovery.rs b/crates/libs/sbx/mgr/server/src/service/recovery.rs index 3bdf4a16a5..e3068a5f01 100644 --- a/crates/libs/sbx/mgr/server/src/service/recovery.rs +++ b/crates/libs/sbx/mgr/server/src/service/recovery.rs @@ -2271,6 +2271,22 @@ impl SandboxManagerService { Ok(_) => {}, Err(error) => tracing::warn!(%error, "teardown suspend job reconciliation failed; will retry"), } + if service.suspend_backup_redrive_enabled { + match service.storage(|db| { + crate::db::due_teardown_suspend_redrives(db) + .map_err(|error| Status::internal(format!("discover suspend backup retries: {error}"))) + }).await { + Ok(candidates) => { + for (id, generation) in candidates { + match id.parse() { + Ok(id) => { service.spawn_teardown_suspend_driver(id, generation, SandboxEventTrigger::System); } + Err(error) => tracing::error!(sandbox_id = id, %error, "invalid sandbox identity in suspend retry job"), + } + } + } + Err(error) => tracing::warn!(%error, "suspend backup retry discovery failed; will retry"), + } + } let mut cursor = std::mem::take(&mut recovery_cursor); let candidates = match service.storage(move |db| { let rows = crate::db::periodic_lifecycle_recovery_batch(db, &mut cursor) diff --git a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs index 31141ff26f..39f4df9341 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs.rs @@ -20,7 +20,10 @@ //! skip this step. //! 2. Concurrently polls all vol-mgr jobs — every VM artifact and the //! `snapshot.json` metadata document — to a terminal -//! state. Any permanent vol-mgr failure → [`JobError::Permanent`]; any +//! state. With generation re-drive enabled, a failed full suspend transfer +//! exhausts this attempt as retriable so the generation owner can validate +//! its local source and start a fresh attempt. Other permanent vol-mgr +//! failures → [`JobError::Permanent`]; any //! transient transport/RPC failure → [`JobError::Retriable`]. //! 3. Calls `ClusterApi.FinalizeArtifactBackup` with `target_state=READY` //! only once *all* of those uploads succeed, so the row never becomes @@ -60,6 +63,16 @@ const JOB_TIMEOUT: Duration = Duration::from_secs(7200); const MAX_ATTEMPTS: i32 = 5; const TRANSFER_PROBE_TIMEOUT: Duration = Duration::from_secs(10); +/// Keep disabled until the cluster's generation-fenced backup admission is deployed. +pub(crate) fn suspend_backup_redrive_enabled() -> bool { + pplx_space_common_core::env::env_flag( + std::env::var("SPACE_SUSPEND_BACKUP_REDRIVE_ENABLED") + .ok() + .as_deref(), + false, + ) +} + /// Build a `BackgroundWorker` configured for the snapshot-upload pipeline. /// Caller owns the `rusqlite::Connection` (it lives alongside the rest of /// sbx-mgr's SQLite state) and is responsible for calling @@ -78,6 +91,7 @@ pub(crate) fn build_worker( .with_kind_concurrency(SNAPSHOT_UPLOAD_JOB_KIND, DEFAULT_CONCURRENCY) .with_recovery_job_kind(SNAPSHOT_UPLOAD_JOB_KIND); let executor = SnapshotUploadExecutor { + redrive_failed_transfers: suspend_backup_redrive_enabled(), db: std::sync::Arc::clone(&db), drain_state, retained_source_inspector: std::sync::Arc::new( @@ -106,6 +120,7 @@ pub fn refund_interrupted_running_attempts(conn: &rusqlite::Connection) -> rusql /// same lock hold. #[derive(Clone)] pub struct SnapshotUploadExecutor { + redrive_failed_transfers: bool, db: std::sync::Arc>, drain_state: std::sync::Arc, /// Sizes a retained source before its stream is reclaimed, so the local @@ -142,6 +157,14 @@ impl JobExecutor for SnapshotUploadExecutor { .artifact_origin .map(|origin| origin.as_str()) .unwrap_or("suspend"); + let redrive_failed_transfers = self.redrive_failed_transfers + && payload + .meta + .artifact_origin + .unwrap_or(pplx_space_common_core::types::ArtifactOrigin::Suspend) + == pplx_space_common_core::types::ArtifactOrigin::Suspend + && artifact_kind_from_meta(&payload.meta) + == pplx_space_common_core::types::ArtifactKind::Full; if job.attempts == 1 && let Some(duration_ms) = job @@ -188,12 +211,14 @@ impl JobExecutor for SnapshotUploadExecutor { // a failure here is logged and never gates the upload. touch_with_cluster_api(&snapshot_id).await; - let results = join_all( - payload - .job_ids - .iter() - .map(|job_id| wait_for_artifact_job(&sandbox_id, job_id, cancellation.clone())), - ) + let results = join_all(payload.job_ids.iter().map(|job_id| { + wait_for_artifact_job( + &sandbox_id, + job_id, + cancellation.clone(), + redrive_failed_transfers, + ) + })) .await; // Per-job upload bytes in transfer-submission order: the @@ -235,7 +260,14 @@ impl JobExecutor for SnapshotUploadExecutor { .map_err(JobError::retriable)?; payload.job_ids.push(metadata_job); persist_submitted_transfer_ids(&self.db, &job.job_id, &payload)?; - match wait_for_artifact_job(&sandbox_id, &metadata_job, cancellation.clone()).await { + match wait_for_artifact_job( + &sandbox_id, + &metadata_job, + cancellation.clone(), + redrive_failed_transfers, + ) + .await + { Ok(observation) => artifact_results.push(observation), Err(error) => { if matches!(error, JobError::Permanent(_)) { @@ -859,6 +891,7 @@ async fn wait_for_artifact_job( sandbox_id: &str, job_id: &JobId, cancellation: tokio_util::sync::CancellationToken, + redrive_failed_transfers: bool, ) -> Result { // One reconnecting channel per artifact attempt, not a fresh connection // on every poll. These are observation-only requests: timing out does not @@ -871,7 +904,14 @@ async fn wait_for_artifact_job( .map_err(|e| JobError::retriable(format!("vol-mgr connect ({job_id}): {e}")))? } }; - poll_artifact_job(sandbox_id, *job_id, client, cancellation).await + poll_artifact_job( + sandbox_id, + *job_id, + client, + cancellation, + redrive_failed_transfers, + ) + .await } /// Probe interval after the first pending answer. A teardown suspend holds its @@ -882,6 +922,61 @@ async fn wait_for_artifact_job( const ARTIFACT_POLL_INITIAL: Duration = Duration::from_millis(100); const ARTIFACT_POLL_MAX: Duration = Duration::from_millis(500); +/// Old binaries recorded transfer failure as a permanent snapshot-job error. +/// Recover only from live typed transfer evidence, never historical error text +/// (which could instead describe a cluster admission/finalization refusal). +pub(crate) async fn has_failed_transfer(payload: &SnapshotUploadJobs) -> Result { + if payload.job_ids.is_empty() { + return Ok(false); + } + let client = tokio::time::timeout( + TRANSFER_PROBE_TIMEOUT, + pplx_space_volmgr_client::connect(VOLMGR_SOCKET), + ) + .await + .map_err(|_| JobError::retriable("vol-mgr failed-transfer probe connect timed out"))? + .map_err(|error| JobError::retriable(format!("vol-mgr failed-transfer probe: {error}")))?; + has_failed_transfer_with_client(&payload.job_ids, client).await +} + +async fn has_failed_transfer_with_client( + job_ids: &[JobId], + client: pplx_space_volmgr_client::ConnectedVolumeManagerClient, +) -> Result { + let probes = job_ids.iter().map(|id| { + let mut client = client.clone(); + async move { + let response = tokio::time::timeout( + TRANSFER_PROBE_TIMEOUT, + client.get_transfer_job(GetTransferJobRequest { + job_id: id.to_string(), + }), + ) + .await + .map_err(|_| JobError::retriable("vol-mgr failed-transfer probe timed out"))? + .map_err(|error| { + JobError::retriable(format!("vol-mgr failed-transfer probe: {error}")) + })?; + Ok::<_, JobError>( + response + .into_inner() + .job + .is_some_and(|job| job.status == TransferJobStatus::Failed as i32), + ) + } + }); + let results = join_all(probes).await; + // One proven failed member makes the old attempt unrecoverable even when + // a sibling's status is temporarily unavailable. + if results.iter().any(|result| matches!(result, Ok(true))) { + return Ok(true); + } + for result in results { + result?; + } + Ok(false) +} + fn next_artifact_poll_interval(interval: Duration) -> Duration { (interval * 2).min(ARTIFACT_POLL_MAX) } @@ -891,6 +986,7 @@ async fn poll_artifact_job( job_id: JobId, mut client: pplx_space_volmgr_client::ConnectedVolumeManagerClient, cancellation: tokio_util::sync::CancellationToken, + redrive_failed_transfers: bool, ) -> Result { let mut interval = ARTIFACT_POLL_INITIAL; loop { @@ -952,10 +1048,17 @@ async fn poll_artifact_job( .transfer_job_id(job_id.to_string()) .error(error.clone()), ); - return Err(JobError::permanent(format!( - "vol-mgr job {job_id} failed permanently: {}", - error - ))); + let message = format!("vol-mgr job {job_id} failed: {error}"); + // A failed transfer is terminal for this immutable attempt, + // not necessarily for the captured generation. Exhaust the + // bounded observation budget as retriable so its generation + // driver can validate the retained artifact and submit a new + // attempt through the normal, fenced admission path. + return Err(if redrive_failed_transfers { + JobError::retriable(message) + } else { + JobError::permanent(message) + }); } _ => { tokio::select! { diff --git a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs index c164544270..8305f0b951 100644 --- a/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs +++ b/crates/libs/sbx/mgr/server/src/snapshot_upload_jobs/tests.rs @@ -3,6 +3,108 @@ use pplx_space_common_core::types::LocalArtifactId; use super::*; +#[derive(Clone)] +struct ExhaustedTransfer; + +impl tonic::server::NamedService for ExhaustedTransfer { + const NAME: &'static str = "space.volume.manager.v1.VolumeManager"; +} + +impl tonic::server::UnaryService for ExhaustedTransfer { + type Response = pplx_space_volmgr_client::proto::GetTransferJobResponse; + type Future = std::future::Ready, tonic::Status>>; + + fn call(&mut self, request: tonic::Request) -> Self::Future { + std::future::ready(Ok(tonic::Response::new(Self::Response { + job: Some(pplx_space_volmgr_client::proto::TransferJob { + job_id: request.into_inner().job_id, + status: TransferJobStatus::Failed as i32, + error: Some("credential exchange failed: storage role unavailable".into()), + ..Default::default() + }), + }))) + } +} + +impl tower::Service> for ExhaustedTransfer +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("/GetTransferJob")); + Box::pin(async move { + let mut grpc = tonic::server::Grpc::new(tonic_prost::ProstCodec::default()); + Ok(grpc.unary(ExhaustedTransfer, request).await) + }) + } +} + +/// Exercise the actual transfer observation RPC, not a classifier fed an error +/// string. Exhausted credential attempts must leave the generation driver a +/// retriable outcome; the gate preserves the old outcome until safe rollout. +#[tokio::test] +async fn exhausted_credential_transfer_can_be_redriven() { + 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(ExhaustedTransfer) + .serve_with_incoming_shutdown( + tokio_stream::wrappers::TcpListenerStream::new(listener), + stop.clone().cancelled_owned(), + ), + ); + for enabled in [false, true] { + let channel = tonic::transport::Endpoint::from_shared(endpoint.clone()) + .unwrap() + .connect() + .await + .unwrap(); + let client = pplx_space_volmgr_client::VolumeManagerClient::new( + pplx_space_common_svc::bootstrap::wrap_grpc_client_channel(channel), + ); + assert!( + has_failed_transfer_with_client(&[JobId::new()], client.clone()) + .await + .unwrap() + ); + assert!( + !has_failed_transfer_with_client(&[], client.clone()) + .await + .unwrap() + ); + let error = poll_artifact_job( + "synthetic-sandbox", + JobId::new(), + client, + tokio_util::sync::CancellationToken::new(), + enabled, + ) + .await + .unwrap_err(); + assert_eq!( + error.kind() == pplx_space_common_jobs::JobFailureKind::Retriable, + enabled, + "an exhausted credential upload must not permanently strand its suspend: {error}" + ); + } + stop.cancel(); + server.await.unwrap().unwrap(); +} + fn job_fixture(kind: &str, payload: String) -> Job { Job { job_id: JobId::new(), @@ -35,6 +137,7 @@ async fn upload_executor_converges_unstaged_metadata_failure() { payload.includes_metadata_job = false; payload.metadata_staging_dir = Some(metadata_dir.clone()); let executor = SnapshotUploadExecutor { + redrive_failed_transfers: false, db: std::sync::Arc::new(std::sync::Mutex::new(open_backfill_test_db())), drain_state: crate::service::tests::closed_drain_for_test(), node_manager: crate::node_manager_transport::NodeManagerConnector::host(), @@ -92,7 +195,7 @@ async fn cancelled_transfer_poll_yields_before_connecting() { let cancellation = tokio_util::sync::CancellationToken::new(); cancellation.cancel(); - let error = wait_for_artifact_job("sandbox", &JobId::new(), cancellation) + let error = wait_for_artifact_job("sandbox", &JobId::new(), cancellation, false) .await .unwrap_err(); @@ -117,6 +220,7 @@ async fn stalled_transfer_probe_is_bounded_without_cancelling_remote_work() { JobId::new(), client, cancellation.clone(), + false, )); // Accept the connection but never answer its HTTP/2 handshake or request. let (_connection, _) = listener.accept().await.unwrap(); @@ -181,6 +285,7 @@ fn refund_interrupted_running_attempts_only_touches_running_snapshot_upload_jobs #[tokio::test] async fn executor_rejects_non_snapshot_upload_job_kind() { let executor = SnapshotUploadExecutor { + redrive_failed_transfers: false, db: std::sync::Arc::new(std::sync::Mutex::new( rusqlite::Connection::open_in_memory().unwrap(), )), @@ -214,6 +319,7 @@ async fn executor_requires_begin_before_volume_staging() { return; } let executor = SnapshotUploadExecutor { + redrive_failed_transfers: false, db: std::sync::Arc::new(std::sync::Mutex::new( rusqlite::Connection::open_in_memory().unwrap(), )), @@ -362,6 +468,7 @@ fn ready_backup_keeps_completed_evidence_when_bookkeeping_misses_execution_deadl std::sync::Arc::clone(&db), ReadyBackupExecutor { executor: SnapshotUploadExecutor { + redrive_failed_transfers: false, db: std::sync::Arc::clone(&db), drain_state: std::sync::Arc::new(crate::service::drain::DrainState::default()), node_manager: crate::node_manager_transport::NodeManagerConnector::host(), @@ -1135,6 +1242,7 @@ fn ready_backup_returns_before_the_retained_source_is_inspected() { let db = std::sync::Arc::new(std::sync::Mutex::new(conn)); let gate = std::sync::Arc::new(tokio::sync::Notify::new()); let executor = SnapshotUploadExecutor { + redrive_failed_transfers: false, db: std::sync::Arc::clone(&db), drain_state: std::sync::Arc::new(crate::service::drain::DrainState::default()), node_manager: crate::node_manager_transport::NodeManagerConnector::host(), diff --git a/docs/snapshot-upload-streaming.md b/docs/snapshot-upload-streaming.md index acdbb72394..49542ae96d 100644 --- a/docs/snapshot-upload-streaming.md +++ b/docs/snapshot-upload-streaming.md @@ -1,5 +1,37 @@ # Snapshot upload streaming +## Retrying failed generation-suspend backups + +`SPACE_SUSPEND_BACKUP_REDRIVE_ENABLED` on sandbox-manager defaults to `false`. +Do not enable it until the cluster API enforces execution-generation admission +for new backup attempts. Local ownership cannot detect an execution retired +only in the control plane while its worker still holds the old suspend job. + +When enabled, a terminal failed volume-manager transfer exhausts the +snapshot-upload worker's bounded observation retries as a **retriable** failure, +not a permanent generation failure. The generation driver then uses the normal +artifact-backup submission path to allocate a fresh backup attempt. It verifies +the published local artifact and checks the generation and artifact again +atomically with local job admission. Missing or invalid local captures remain +terminal; failed I/O or retained-source inspection is not evidence of data loss. +Cluster registration and finalization refusals are not reclassified. + +The generation driver's persisted retry delay doubles from 30 seconds to a +15-minute cap, surviving daemon restarts and caller polling. Periodic lifecycle +recovery discovers due generation jobs even when the original caller is gone, +using the existing bounded, deduplicated driver pool. The +`snapshot.backup.redrive` counter counts re-drive submission attempts, not polls or +individual file-transfer retries. A non-durable-generation age gauge is not +provided by this change. + +Volume-manager's per-attempt staged copies are disposable. Re-drive uses the +separate immutable artifact registered with sandbox-manager, not a failed +transfer's staging path. If that artifact has been removed, this feature does +not reconstruct it or mark it durable. Historical permanent snapshot-upload +failures are eligible only when volume-manager still reports a failed member +transfer for the exact artifact. Missing transfer history is not recovery +authority; such cases still require an explicit operator recovery decision. + Volume-manager can feed its zstd encoder directly into the object store's bounded multipart uploader. For a compressed artifact of size C, this avoids writing a C-byte temporary file and reading it again for MD5 and upload. It From 88a3af1925ef08cca734ea1c0b63d33f4982d379 Mon Sep 17 00:00:00 2001 From: Mostafa Abdelraouf Date: Tue, 13 Oct 2026 03:16:43 +0000 Subject: [PATCH 2/2] test(sbx-mgr): require stale suspend callers to leave durability unchanged Co-authored-by: pplx-teammatebot --- .../src/service/grpc_lifecycle/teardown_backup_tests.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 index ec457aac13..f9b61c19f7 100644 --- 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 @@ -157,7 +157,9 @@ async fn teardown_reuses_ready_backup_but_waits_for_active_local_upload() { let status = crate::db::teardown_suspend_status(&db, &sandbox_id.to_string()) .unwrap() .unwrap(); - assert!(status.backup_completed); + // A stale caller is now rejected before remote lookup and must not + // mutate the current generation's durability, even if S3 is ready. + assert_eq!(status.backup_completed, !superseded); assert!( pplx_space_common_jobs::db::list_jobs(&db, Some(&sandbox_id.to_string()), None, 0) .unwrap()