Retry failed generation-suspend backups behind an admission gate

crates/libs/common/jobs/src/worker.rs

@@ -1,4 +1,4 @@
 use pplx_space_common_core::types::JobId;
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
@@ -635,6 +635,31 @@
         )
     }
 
+    /// 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<T: serde::Serialize>(
+        &self,
+        kind: &str,
+        payload: &T,
+        subject_id: Option<&str>,
+        dedup_key: Option<&str>,
+        admit: impl FnOnce(&rusqlite::Connection) -> Result<(), Status>,
+    ) -> Result<JobId, Status> {
+        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<T: serde::Serialize>(
         &self,
         kind: &str,
@@ -725,12 +750,25 @@
         subject_id: Option<&str>,
         dedup_key: Option<&str>,
         options: SubmitOptions,
+    ) -> Result<JobId, Status> {
+        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<JobId, Status> {
         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) => {

crates/libs/common/jobs/src/worker/tests.rs

@@ -14,6 +14,36 @@ struct TestExecutor {
     should_block: Arc<AtomicBool>,
 }
 
+#[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 {

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,

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;

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<bool> {
+    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<bool> {
+    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<Vec<(String, ExecutionGeneration)>> {
+    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::<Result<Vec<_>>>()?;
+    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()
+}

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<ExecutionGeneration>,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]

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,