Pace failed teardown drivers without holding worker capacity

crates/libs/sbx/mgr/server/src/db.rs

@@ -52,9 +52,9 @@ mod quarantine_scan;
 mod sandbox_row;
 mod teardown_jobs;
 pub(crate) use teardown_jobs::{
-    TeardownSuspendFailure, fail_teardown_suspend_backup_operation,
+    TeardownSuspendFailure, defer_teardown_suspend_retry, fail_teardown_suspend_backup_operation,
     fail_teardown_suspend_operation, reconcile_terminal_teardown_suspend_jobs,
-    teardown_suspend_retry_admissible,
+    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

@@ -194,3 +194,48 @@ pub(crate) fn reconcile_terminal_teardown_suspend_jobs(conn: &Connection) -> Res
     tx.commit()?;
     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.
+pub(crate) fn defer_teardown_suspend_retry(
+    conn: &Connection,
+    id: &str,
+    generation: ExecutionGeneration,
+    error: &str,
+) -> Result<()> {
+    if let Some(operation) = active_teardown_suspend_operation(conn, id)?
+        && operation.execution_generation == generation
+    {
+        conn.execute(
+            sql!(
+                "UPDATE jobs SET next_attempt_at = datetime('now', '+' || ?3 || ' seconds'), error = ?2
+              WHERE job_id = ?1 AND status IN ('pending', 'running')"
+            ),
+            params![operation.job_id, error, TEARDOWN_SUSPEND_RETRY_DELAY_SECS],
+        )?;
+    }
+    Ok(())
+}
+
+pub(crate) fn teardown_suspend_retry_due(
+    conn: &Connection,
+    id: &str,
+    generation: ExecutionGeneration,
+) -> Result<bool> {
+    let Some(operation) = active_teardown_suspend_operation(conn, id)? else {
+        return Ok(true);
+    };
+    if operation.execution_generation != generation {
+        return Ok(true);
+    }
+    conn.query_row(
+        sql!(
+            "SELECT next_attempt_at IS NULL OR next_attempt_at <= datetime('now')
+           FROM jobs WHERE job_id = ?1"
+        ),
+        params![operation.job_id],
+        |row| row.get(0),
+    )
+}

crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_backup_tests.rs

@@ -169,3 +169,43 @@ async fn teardown_reuses_ready_backup_but_waits_for_active_local_upload() {
     stop.cancel();
     server.await.unwrap().unwrap();
 }
+
+#[tokio::test]
+async fn failed_teardown_defers_repeated_polls_without_holding_driver_capacity() {
+    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 sandbox = SandboxId::new();
+    let generation = ExecutionGeneration::new(7).unwrap();
+    insert_stranded_teardown_suspend_sandbox(&conn, sandbox, generation);
+    let service = SandboxManagerService::new(conn);
+    service
+        .drive_teardown_suspend(sandbox, generation, SandboxEventTrigger::System)
+        .await
+        .unwrap_err();
+    let capacity = service.teardown_suspend_driver_sem.available_permits();
+    for _ in 0..4 {
+        service.spawn_teardown_suspend_driver(sandbox, generation, SandboxEventTrigger::System);
+        assert!(
+            service.teardown_suspend_drivers.is_empty(),
+            "a failed operation must defer repeated polls"
+        );
+        assert_eq!(
+            service.teardown_suspend_driver_sem.available_permits(),
+            capacity
+        );
+    }
+    {
+        let db = service.db.lock().unwrap();
+        db.execute("UPDATE jobs SET next_attempt_at=datetime('now','-1 second') WHERE subject_id=?1 AND kind='teardown_suspend'", [sandbox.to_string()]).unwrap();
+    }
+    service.spawn_teardown_suspend_driver(sandbox, generation, SandboxEventTrigger::System);
+    assert_eq!(
+        service.teardown_suspend_drivers.len(),
+        1,
+        "expired backoff must allow another attempt"
+    );
+    service.shutdown_token.cancel();
+    service.task_tracker.close();
+    service.task_tracker.wait().await;
+}

crates/libs/sbx/mgr/server/src/service/grpc_lifecycle/teardown_suspend.rs

@@ -476,6 +476,22 @@ impl SandboxManagerService {
         execution_generation: pplx_space_common_core::types::ExecutionGeneration,
         event_trigger: SandboxEventTrigger,
     ) -> TeardownSuspendDriverAdmission {
+        let due = self.db_conn().and_then(|db| {
+            crate::db::teardown_suspend_retry_due(
+                &db,
+                &sandbox_id.to_string(),
+                execution_generation,
+            )
+            .map_err(|error| Status::internal(format!("read teardown retry deadline: {error}")))
+        });
+        match due {
+            Ok(true) => {}
+            Ok(false) => return TeardownSuspendDriverAdmission::Deferred,
+            Err(error) => {
+                tracing::warn!(%sandbox_id, %error, "cannot read teardown retry deadline");
+                return TeardownSuspendDriverAdmission::Deferred;
+            }
+        }
         let dashmap::mapref::entry::Entry::Vacant(entry) =
             self.teardown_suspend_drivers.entry(sandbox_id)
         else {
@@ -560,6 +576,19 @@ impl SandboxManagerService {
                 tracing::warn!(%sandbox_id, %execution_generation, %record_error,
                     "failed to persist teardown suspend driver failure; caller polling will retry");
             }
+            if let Err(error) = self.db_conn().and_then(|db| {
+                crate::db::defer_teardown_suspend_retry(
+                    &db,
+                    &sandbox_id.to_string(),
+                    execution_generation,
+                    error.message(),
+                )
+                .map_err(|error| {
+                    Status::internal(format!("persist teardown retry deadline: {error}"))
+                })
+            }) {
+                tracing::warn!(%sandbox_id, %error, "cannot persist teardown retry deadline");
+            }
         }
         result
     }

crates/libs/sbx/mgr/server/src/service/mod.rs

@@ -310,6 +310,7 @@ type SbxExportSandboxStream =
 type TeardownSuspendDrivers = Arc<dashmap::DashMap<SandboxId, tokio::sync::OwnedSemaphorePermit>>;
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum TeardownSuspendDriverAdmission {
+    Deferred,
     Started,
     Existing,
     Saturated,
@@ -318,6 +319,7 @@ enum TeardownSuspendDriverAdmission {
 impl TeardownSuspendDriverAdmission {
     const fn as_str(self) -> &'static str {
         match self {
+            Self::Deferred => "deferred",
             Self::Started => "started",
             Self::Existing => "existing",
             Self::Saturated => "saturated",