diff --git a/crates/libs/sbx/mgr/server/src/db.rs b/crates/libs/sbx/mgr/server/src/db.rs index 86b0944c6b..5f73d8b8e5 100644 --- a/crates/libs/sbx/mgr/server/src/db.rs +++ b/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; 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 8c8b81e3a9..35458d27c9 100644 --- a/crates/libs/sbx/mgr/server/src/db/teardown_jobs.rs +++ b/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 { + 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), + ) +} 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 5b1d6bd5c7..ec457aac13 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 @@ -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; +} 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 feeed29fb0..6cc6b2e5fe 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 @@ -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 } diff --git a/crates/libs/sbx/mgr/server/src/service/mod.rs b/crates/libs/sbx/mgr/server/src/service/mod.rs index 07b685bdbd..231fb56850 100644 --- a/crates/libs/sbx/mgr/server/src/service/mod.rs +++ b/crates/libs/sbx/mgr/server/src/service/mod.rs @@ -310,6 +310,7 @@ type SbxExportSandboxStream = type TeardownSuspendDrivers = Arc>; #[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",