diff --git a/crates/workspace-server/src/backend_job.rs b/crates/workspace-server/src/backend_job.rs index e7f7bce77..08effd749 100644 --- a/crates/workspace-server/src/backend_job.rs +++ b/crates/workspace-server/src/backend_job.rs @@ -284,6 +284,38 @@ pub struct BackendJobRecord { pub completed_at: Option, } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum BackendJobWorkerCleanupState { + Pending, + Executing, + Completed, + Failed, +} + +impl BackendJobWorkerCleanupState { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Pending => "pending", + Self::Executing => "executing", + Self::Completed => "completed", + Self::Failed => "failed", + } + } + + pub(crate) fn parse(value: &str) -> Result { + match value { + "pending" => Ok(Self::Pending), + "executing" => Ok(Self::Executing), + "completed" => Ok(Self::Completed), + "failed" => Ok(Self::Failed), + other => Err(Error::Store(format!( + "unknown Backend Job Worker cleanup state `{other}`" + ))), + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct BackendJobAttemptRecord { pub workspace_id: String, @@ -300,6 +332,11 @@ pub struct BackendJobAttemptRecord { pub result_digest: Option, pub failure_category: Option, pub failure_detail: Option, + pub worker_cleanup_state: Option, + pub worker_cleanup_failure_category: Option, + pub worker_cleanup_failure_detail: Option, + pub worker_cleanup_updated_at: Option, + pub worker_cleanup_completed_at: Option, pub created_at: String, pub updated_at: String, pub completed_at: Option, diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index 3a24911fc..16d98f1f4 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -875,6 +875,13 @@ CREATE TABLE backend_job_attempts ( result_digest TEXT, failure_category TEXT, failure_detail TEXT, + worker_cleanup_state TEXT CHECK ( + worker_cleanup_state IN ('pending', 'executing', 'completed', 'failed') + ), + worker_cleanup_failure_category TEXT, + worker_cleanup_failure_detail TEXT, + worker_cleanup_updated_at TEXT, + worker_cleanup_completed_at TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, completed_at TEXT, @@ -889,6 +896,8 @@ CREATE TABLE backend_job_attempts ( ); CREATE INDEX backend_job_attempts_recovery ON backend_job_attempts(workspace_id, state, deadline_at); +CREATE INDEX backend_job_attempts_worker_cleanup + ON backend_job_attempts(workspace_id, worker_cleanup_state, updated_at); CREATE TABLE backend_job_deliveries ( workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ebc92c143..e76d72e73 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -144,9 +144,10 @@ use crate::authority::{ SqliteWorkspaceAuthority, TicketAuthority, TicketMergeRevisionSource, merge_request_summary, }; use crate::backend_job::{ - ABSOLUTE_MAX_CONCURRENT_JOBS, BackendJobAttemptState, BackendJobDeliveryState, - BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance, - BackendJobResultSubmission, BackendJobState, MAX_JOB_DELIVERY_BYTES, allocation_key, + ABSOLUTE_MAX_CONCURRENT_JOBS, BackendJobAttemptRecord, BackendJobAttemptState, + BackendJobDeliveryState, BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance, + BackendJobResultSubmission, BackendJobState, BackendJobWorkerCleanupState, + MAX_JOB_DELIVERY_BYTES, allocation_key, }; use crate::companion::{ CompanionCancelRequest, CompanionConsole, CompanionMessageRequest, CompanionMessageResponse, @@ -3417,14 +3418,12 @@ impl WorkspaceApi { Ok(spawned) => spawned, Err(error) => { let detail = error.error.to_string(); - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &request.job_id, &reservation.attempt.attempt_id, BackendJobAttemptState::Failed, "dispatch_error", &detail, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(error.error); } @@ -3437,14 +3436,12 @@ impl WorkspaceApi { .first() .map(|diagnostic| diagnostic.message.as_str()) .unwrap_or("embedded Runtime rejected Backend Job Worker dispatch"); - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &request.job_id, &reservation.attempt.attempt_id, BackendJobAttemptState::Failed, "dispatch_rejected", detail, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(Error::RuntimeOperationFailed { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), @@ -3505,14 +3502,12 @@ impl WorkspaceApi { Ok(submitted) => submitted, Err(error) => { let detail = error.message(); - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &request.job_id, &reservation.attempt.attempt_id, BackendJobAttemptState::Unknown, "input_outcome_unknown", &detail, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(error.into_error()); } @@ -3531,14 +3526,12 @@ impl WorkspaceApi { (BackendJobAttemptState::Failed, "input_rejected") } }; - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &request.job_id, &reservation.attempt.attempt_id, state, category, detail, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(Error::RuntimeOperationFailed { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), @@ -3547,14 +3540,12 @@ impl WorkspaceApi { }); } if submitted.runtime_run_id.as_deref() != Some(runtime_run_id.as_str()) { - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &request.job_id, &reservation.attempt.attempt_id, BackendJobAttemptState::Unknown, "runtime_run_mismatch", "embedded Runtime accepted Backend Job input without matching tracked run evidence", - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; return Err(Error::RuntimeOperationFailed { runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), @@ -3672,6 +3663,267 @@ impl WorkspaceApi { } } + fn finish_backend_job_attempt( + &self, + job_id: &str, + attempt_id: &str, + state: BackendJobAttemptState, + failure_category: &str, + failure_detail: &str, + ) -> Result { + let attempt = self.store.finish_backend_job_attempt( + &self.config.workspace_id, + job_id, + attempt_id, + state, + failure_category, + failure_detail, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + let _ = self.schedule_backend_job_worker_cleanup(&attempt); + Ok(attempt) + } + + fn schedule_backend_job_worker_cleanup(&self, attempt: &BackendJobAttemptRecord) -> bool { + if !attempt.state.terminal() + || attempt.worker.is_none() + || !matches!( + attempt.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Pending | BackendJobWorkerCleanupState::Failed) + ) + { + return false; + } + let Ok(handle) = tokio::runtime::Handle::try_current() else { + tracing::warn!( + job_id = %attempt.job_id, + attempt_id = %attempt.attempt_id, + "Backend Job Worker cleanup remains durable because no Tokio runtime is available" + ); + return false; + }; + let (attempt, claimed) = match self.store.claim_backend_job_worker_cleanup( + &self.config.workspace_id, + &attempt.job_id, + &attempt.attempt_id, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) { + Ok(claim) => claim, + Err(error) => { + tracing::warn!( + %error, + job_id = %attempt.job_id, + attempt_id = %attempt.attempt_id, + "failed to claim durable Backend Job Worker cleanup" + ); + return false; + } + }; + if !claimed { + return false; + } + let api = self.clone(); + let job_id = attempt.job_id.clone(); + let attempt_id = attempt.attempt_id.clone(); + handle.spawn_blocking(move || { + let cleanup = || -> Result<()> { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|error| Error::Store(error.to_string()))? + .block_on(async { + // Never run stop/removal inline with structured result + // acceptance. A separate scheduler thread lets the HTTP + // response complete while canonical stop waits for the + // Worker Run teardown. + tokio::task::yield_now().await; + api.cleanup_backend_job_attempt_worker(attempt).await + }) + }; + if let Err(error) = cleanup() { + tracing::warn!(%error, %job_id, %attempt_id, "Backend Job Worker cleanup failed"); + } + }); + true + } + + fn recover_backend_job_worker_cleanups(&self) -> Result<()> { + let page_limit = usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS); + let mut attempts = self.store.recover_backend_job_worker_cleanups( + &self.config.workspace_id, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + page_limit, + )?; + loop { + if attempts.is_empty() { + return Ok(()); + } + let scheduled = attempts + .iter() + .filter(|attempt| self.schedule_backend_job_worker_cleanup(attempt)) + .count(); + if attempts.len() < page_limit || scheduled == 0 { + return Ok(()); + } + attempts = self + .store + .list_backend_job_worker_cleanups(&self.config.workspace_id, page_limit)?; + } + } + + async fn cleanup_backend_job_attempt_worker( + &self, + attempt: BackendJobAttemptRecord, + ) -> Result<()> { + let job_id = attempt.job_id.as_str(); + let attempt_id = attempt.attempt_id.as_str(); + let worker = attempt.worker.clone().ok_or_else(|| { + Error::Store("claimed Backend Job Worker cleanup lost its binding".to_string()) + })?; + + let fail = |category: &str, detail: String| -> Result<()> { + self.store.finish_backend_job_worker_cleanup( + &self.config.workspace_id, + job_id, + attempt_id, + BackendJobWorkerCleanupState::Failed, + Some(category), + Some(&detail), + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + Err(Error::RuntimeOperationFailed { + runtime_id: worker.runtime_id.clone(), + code: category.to_string(), + message: detail, + }) + }; + let complete = || -> Result<()> { + self.store.finish_backend_job_worker_cleanup( + &self.config.workspace_id, + job_id, + attempt_id, + BackendJobWorkerCleanupState::Completed, + None, + None, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + Ok(()) + }; + + let detail = match self.runtime.worker(&worker) { + Ok(detail) => detail, + Err(RuntimeRegistryError::UnknownWorker { .. }) => { + if self + .store + .get_worker_registry(&self.config.workspace_id, &worker)? + .is_none() + { + return complete(); + } + // A canonical removal may have deleted the Runtime object before + // its catalog commit. Let the existing durable removal operation + // recover that midpoint; without such an operation this remains + // a diagnosable retryable failure rather than a direct DB delete. + let response = WorkerRemovalService::new(self) + .execute_target_removal( + self.runtime.as_ref(), + &worker, + &format!("Terminal Backend Job attempt {attempt_id}"), + None, + ) + .await + .map_err(|message| Error::RuntimeOperationFailed { + runtime_id: worker.runtime_id.clone(), + code: "backend_job_cleanup_removal_failed".to_string(), + message, + }); + return match response { + Ok(response) if response.status == StatusCode::OK.as_u16() => complete(), + Ok(response) => { + fail("backend_job_cleanup_runtime_worker_missing", response.body) + } + Err(error) => fail("backend_job_cleanup_removal_failed", error.to_string()), + }; + } + Err(error) => { + return fail("backend_job_cleanup_runtime_unavailable", error.message()); + } + }; + if !detail.state.eq_ignore_ascii_case("stopped") { + match self.runtime.stop_worker( + &worker, + WorkerLifecycleRequest { + reason: Some(format!( + "Backend Job attempt {attempt_id} reached terminal state" + )), + ticket_assignment: None, + }, + ) { + Ok(result) if result.state == InternalWorkerOperationState::Accepted => {} + Ok(result) => { + return fail( + "backend_job_cleanup_stop_rejected", + runtime_diagnostics_message(&result.diagnostics), + ); + } + Err(error) => { + return fail("backend_job_cleanup_stop_failed", error.message()); + } + } + } + match self.runtime.worker(&worker) { + Ok(detail) if detail.state.eq_ignore_ascii_case("stopped") => {} + Ok(detail) => { + return fail( + "backend_job_cleanup_stop_unconfirmed", + format!( + "Runtime reported Worker state `{}` after accepting stop", + detail.state + ), + ); + } + Err(error) => { + return fail( + "backend_job_cleanup_stop_confirmation_failed", + error.message(), + ); + } + } + + let response = WorkerRemovalService::new(self) + .execute_target_removal( + self.runtime.as_ref(), + &worker, + &format!("Terminal Backend Job attempt {attempt_id}"), + None, + ) + .await + .map_err(|message| Error::RuntimeOperationFailed { + runtime_id: worker.runtime_id.clone(), + code: "backend_job_cleanup_removal_failed".to_string(), + message, + }); + let response = match response { + Ok(response) => response, + Err(error) => return fail("backend_job_cleanup_removal_failed", error.to_string()), + }; + if response.status == StatusCode::OK.as_u16() { + return complete(); + } + let runtime_absent = matches!( + self.runtime.worker(&worker), + Err(RuntimeRegistryError::UnknownWorker { .. }) + ); + let catalog_absent = self + .store + .get_worker_registry(&self.config.workspace_id, &worker)? + .is_none(); + if runtime_absent && catalog_absent { + return complete(); + } + fail("backend_job_cleanup_removal_rejected", response.body) + } + /// Reconcile durable non-terminal attempts after Backend restart. Reserved /// attempts are replay-dispatched through their deterministic allocation; /// dispatched attempts keep running only while their bound Worker remains @@ -3701,26 +3953,22 @@ impl WorkspaceApi { } } BackendJobAttemptState::Dispatching => { - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &attempt.job_id, &attempt.attempt_id, BackendJobAttemptState::Unknown, "recovery_dispatch_claim_unknown", "a Backend Job dispatch claim could not be safely replayed after restart", - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; } BackendJobAttemptState::Dispatched => { if attempt.runtime_run_id.is_none() { - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &attempt.job_id, &attempt.attempt_id, BackendJobAttemptState::Unknown, "recovery_runtime_run_unknown", "the Runtime accepted Worker dispatch without durable run evidence", - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; continue; } @@ -3729,14 +3977,12 @@ impl WorkspaceApi { .as_ref() .is_some_and(|worker| self.runtime.worker(worker).is_ok()); if !worker_known { - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &attempt.job_id, &attempt.attempt_id, BackendJobAttemptState::Unknown, "recovery_worker_unknown", "the bound Runtime Worker was not recoverable after Backend restart", - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; } else { self.schedule_backend_job_timeout( @@ -3751,6 +3997,7 @@ impl WorkspaceApi { | BackendJobAttemptState::Unknown => {} } } + self.recover_backend_job_worker_cleanups()?; self.drain_reserved_backend_jobs(); Ok(()) } @@ -3766,15 +4013,13 @@ impl WorkspaceApi { return; }; let api = self.clone(); - let store = api.store.clone(); - let runtime = api.runtime.clone(); - let workspace_id = api.config.workspace_id.clone(); let job_id = job_id.to_string(); let attempt_id = attempt_id.to_string(); handle.spawn(async move { tokio::time::sleep(delay).await; let Ok(Some(attempt)) = - store.get_backend_job_attempt(&workspace_id, &job_id, &attempt_id) + api.store + .get_backend_job_attempt(&api.config.workspace_id, &job_id, &attempt_id) else { return; }; @@ -3783,27 +4028,16 @@ impl WorkspaceApi { } // Fence result acceptance before stopping the Worker so a late // completion cannot turn a timed-out attempt into success. - if store + if api .finish_backend_job_attempt( - &workspace_id, &job_id, &attempt_id, BackendJobAttemptState::Failed, "timeout", "Backend Job attempt exceeded its durable deadline", - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), ) .is_ok() { - if let Some(worker) = attempt.worker { - let _ = runtime.stop_worker( - &worker, - WorkerLifecycleRequest { - reason: Some("Backend Job attempt timed out".to_string()), - ticket_assignment: None, - }, - ); - } api.drain_reserved_backend_jobs(); } }); @@ -3848,14 +4082,12 @@ impl WorkspaceApi { { let detail = error.to_string(); if attempt.state == BackendJobAttemptState::Dispatched { - self.store.finish_backend_job_attempt( - &self.config.workspace_id, + self.finish_backend_job_attempt( &submission.job_id, &submission.attempt_id, BackendJobAttemptState::Failed, "invalid_result", &detail, - &accepted_at, )?; self.drain_reserved_backend_jobs(); } @@ -3871,6 +4103,7 @@ impl WorkspaceApi { if let Some(target) = acceptance.job.request.notification_target.as_ref() { self.deliver_backend_job_result(&acceptance, target); } + let _ = self.schedule_backend_job_worker_cleanup(&acceptance.attempt); self.drain_reserved_backend_jobs(); Ok(acceptance) } @@ -41627,6 +41860,27 @@ mod tests { config } + fn seed_worker_session_for_cleanup(root: &Path, worker: &RuntimeWorkerRef) { + let worker_root = root + .join(".test-embedded-runtime-store/workers") + .join(&worker.worker_id); + fs::create_dir_all(worker_root.join("session/segments")).unwrap(); + fs::write( + worker_root.join("session/session.json"), + serde_json::to_vec_pretty(&json!({ + "schema_version": 1, + "session_id": format!("backend-job-cleanup-{}", worker.worker_id) + })) + .unwrap(), + ) + .unwrap(); + fs::write( + worker_root.join("session/segments/segment-a.jsonl"), + b"backend job retained session evidence\n", + ) + .unwrap(); + } + fn spawn_ticket_check_source(api: &WorkspaceApi, name: &str) -> RuntimeWorkerRef { api.spawn_workspace_worker( EMBEDDED_WORKER_RUNTIME_ID, @@ -41711,6 +41965,30 @@ mod tests { panic!("Ticket item checker attempt remained reserved") } + async fn wait_for_backend_job_worker_cleanup( + api: &WorkspaceApi, + job_id: &str, + attempt_id: &str, + expected: BackendJobWorkerCleanupState, + ) -> crate::backend_job::BackendJobAttemptRecord { + let mut last = None; + for _ in 0..500 { + let attempt = api + .store + .get_backend_job_attempt(TEST_WORKSPACE_ID, job_id, attempt_id) + .unwrap() + .unwrap(); + if attempt.worker_cleanup_state == Some(expected) { + return attempt; + } + last = Some(attempt); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + panic!( + "Backend Job Worker cleanup for {job_id}/{attempt_id} did not reach {expected:?}: {last:?}" + ) + } + #[tokio::test] async fn worker_ticket_create_returns_before_checker_and_delivers_grounded_advisory() { let temp = tempfile::tempdir().unwrap(); @@ -42472,6 +42750,286 @@ mod tests { assert_eq!(accepted.job.state, BackendJobState::Completed); } + #[tokio::test] + async fn terminal_backend_job_cleanup_removes_only_bound_worker_and_retains_result_delivery() { + let temp = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let notification_target = spawn_ticket_check_source(&api, "job-cleanup-target"); + let request = BackendJobRequest { + job_id: "runner-cleanup-success-1".to_string(), + purpose: "cleanup_contract_check".to_string(), + input_revision: "revision-1".to_string(), + input_ref: "test://runner/cleanup/1".to_string(), + input: serde_json::json!({"value": "immutable"}), + instruction: "Return a structured check result.".to_string(), + profile: "builtin:backend-job".to_string(), + source_worker: Some(notification_target.clone()), + notification_target: Some(notification_target.clone()), + limits: crate::backend_job::BackendJobLimits::default(), + }; + let dispatched = api.dispatch_backend_job(&request).unwrap(); + execution.take_inputs(); + execution.reject_inputs("notification unavailable"); + let worker = dispatched.attempt.worker.clone().unwrap(); + seed_worker_session_for_cleanup(temp.path(), &worker); + let mut subscriber = api.worker_projection.subscribe_ordered(10).unwrap(); + let _ = subscriber.take_snapshot(); + let submission = BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: dispatched.attempt.attempt_id.clone(), + input_revision: request.input_revision.clone(), + result: serde_json::json!({"valid": true}), + }; + + let accepted = api.accept_backend_job_result(&worker, &submission).unwrap(); + assert_eq!(accepted.job.state, BackendJobState::Completed); + let cleaned = wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &dispatched.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert_eq!(cleaned.result, Some(serde_json::json!({"valid": true}))); + assert!(matches!( + api.runtime.worker(&worker), + Err(RuntimeRegistryError::UnknownWorker { .. }) + )); + assert!( + !temp + .path() + .join(".test-embedded-runtime-store/workers") + .join(&worker.worker_id) + .exists(), + "Runtime Worker and Session storage must be removed" + ); + assert!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_none(), + "canonical removal must delete the Backend catalog projection" + ); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + loop { + let event = subscriber.recv().await.expect("projection event"); + if event.changes.iter().any(|change| { + matches!(change, crate::store::WorkerCatalogChange::Removed(removed) if removed == &worker) + }) { + return; + } + } + }) + .await + .expect("Backend Job cleanup must publish canonical Worker removal"); + assert!( + api.runtime.worker(¬ification_target).is_ok(), + "origin/notification Workers are not cleanup targets" + ); + assert!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, ¬ification_target) + .unwrap() + .is_some() + ); + let replay = api.accept_backend_job_result(&worker, &submission).unwrap(); + assert!( + replay.replayed, + "duplicate result remains idempotent after cleanup" + ); + let delivery_id = + crate::backend_job::delivery_id(&request.job_id, &dispatched.attempt.attempt_id); + let (delivery, claimed) = api + .store + .reserve_backend_job_delivery( + TEST_WORKSPACE_ID, + &delivery_id, + &request.job_id, + ¬ification_target, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + assert!(!claimed); + assert_eq!(delivery.state, BackendJobDeliveryState::Unknown); + assert_eq!( + delivery.failure_category.as_deref(), + Some("notification_outcome_unknown") + ); + } + + #[tokio::test] + async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_new_worker() { + let temp = tempfile::tempdir().unwrap(); + let (api, _execution) = test_api_with_recording_backend(temp.path()).await; + let request = BackendJobRequest { + job_id: "runner-cleanup-retry-1".to_string(), + purpose: "cleanup_retry_contract_check".to_string(), + input_revision: "revision-1".to_string(), + input_ref: "test://runner/cleanup-retry/1".to_string(), + input: serde_json::json!({"value": "immutable"}), + instruction: "Return a structured check result.".to_string(), + profile: "builtin:backend-job".to_string(), + source_worker: None, + notification_target: None, + limits: crate::backend_job::BackendJobLimits::default(), + }; + let first = api.dispatch_backend_job(&request).unwrap(); + let first_worker = first.attempt.worker.clone().unwrap(); + seed_worker_session_for_cleanup(temp.path(), &first_worker); + let _ = set_worker_retention( + api.clone(), + first_worker.runtime_id.clone(), + first_worker.worker_id.clone(), + true, + ) + .await + .unwrap(); + api.finish_backend_job_attempt( + &request.job_id, + &first.attempt.attempt_id, + BackendJobAttemptState::Failed, + "model_failure", + "model rejected the request", + ) + .unwrap(); + let blocked = wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &first.attempt.attempt_id, + BackendJobWorkerCleanupState::Failed, + ) + .await; + assert_eq!(blocked.failure_category.as_deref(), Some("model_failure")); + assert_eq!( + blocked.worker_cleanup_failure_category.as_deref(), + Some("backend_job_cleanup_removal_rejected") + ); + + let second = api.retry_backend_job(&request.job_id).unwrap(); + let second_worker = second.attempt.worker.clone().unwrap(); + seed_worker_session_for_cleanup(temp.path(), &second_worker); + assert_ne!(first_worker, second_worker); + let _ = set_worker_retention( + api.clone(), + first_worker.runtime_id.clone(), + first_worker.worker_id.clone(), + false, + ) + .await + .unwrap(); + api.recover_backend_jobs().unwrap(); + wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &first.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert!(api.runtime.worker(&second_worker).is_ok()); + + api.accept_backend_job_result( + &second_worker, + &BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: second.attempt.attempt_id.clone(), + input_revision: request.input_revision.clone(), + result: serde_json::json!({"retried": true}), + }, + ) + .unwrap(); + wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &second.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert_eq!( + api.store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &request.job_id, + &first.attempt.attempt_id, + ) + .unwrap() + .unwrap() + .failure_detail + .as_deref(), + Some("model rejected the request") + ); + } + + #[tokio::test] + async fn backend_job_cleanup_restart_recovery_converges_interrupted_and_removed_workers() { + let temp = tempfile::tempdir().unwrap(); + let (api, _execution) = test_api_with_recording_backend(temp.path()).await; + let request = BackendJobRequest { + job_id: "runner-cleanup-restart-1".to_string(), + purpose: "cleanup_restart_contract_check".to_string(), + input_revision: "revision-1".to_string(), + input_ref: "test://runner/cleanup-restart/1".to_string(), + input: serde_json::json!({"value": "immutable"}), + instruction: "Return a structured check result.".to_string(), + profile: "builtin:backend-job".to_string(), + source_worker: None, + notification_target: None, + limits: crate::backend_job::BackendJobLimits::default(), + }; + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let worker = dispatched.attempt.worker.clone().unwrap(); + seed_worker_session_for_cleanup(temp.path(), &worker); + api.store + .accept_backend_job_result( + TEST_WORKSPACE_ID, + &worker, + &BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: dispatched.attempt.attempt_id.clone(), + input_revision: request.input_revision.clone(), + result: serde_json::json!({"persisted": true}), + }, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + let (claimed, did_claim) = api + .store + .claim_backend_job_worker_cleanup( + TEST_WORKSPACE_ID, + &request.job_id, + &dispatched.attempt.attempt_id, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + assert!(did_claim); + assert_eq!( + claimed.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Executing) + ); + + api.recover_backend_jobs().unwrap(); + let cleaned = wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &dispatched.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert_eq!(cleaned.result, Some(serde_json::json!({"persisted": true}))); + api.recover_backend_jobs().unwrap(); + assert_eq!( + api.store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &request.job_id, + &dispatched.attempt.attempt_id, + ) + .unwrap() + .unwrap() + .worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Completed) + ); + } + #[tokio::test] async fn backend_job_fast_result_is_accepted_after_prebound_tracked_run() { let temp = tempfile::tempdir().unwrap(); @@ -42492,9 +43050,11 @@ mod tests { let hook_api = api.clone(); let hook_request = request.clone(); let hook_observed = acceptance_observed.clone(); + let hook_root = temp.path().to_path_buf(); execution.before_input_returns(move |worker| { let source = RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, worker.worker_id.to_string()); + seed_worker_session_for_cleanup(&hook_root, &source); let accepted = hook_api .accept_backend_job_result( &source, @@ -42518,6 +43078,18 @@ mod tests { dispatched.attempt.runtime_run_id.as_deref(), Some(expected_run_id.as_str()) ); + let worker = dispatched.attempt.worker.clone().unwrap(); + wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &dispatched.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert!(matches!( + api.runtime.worker(&worker), + Err(RuntimeRegistryError::UnknownWorker { .. }) + )); } #[tokio::test] @@ -42675,6 +43247,12 @@ mod tests { async fn backend_job_input_error_is_durable_unknown_and_not_automatically_retried() { let temp = tempfile::tempdir().unwrap(); let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let cleanup_root = temp.path().to_path_buf(); + execution.before_input_returns(move |worker| { + let worker = + RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, worker.worker_id.to_string()); + seed_worker_session_for_cleanup(&cleanup_root, &worker); + }); execution.reject_inputs("model transport unavailable"); let request = BackendJobRequest { job_id: "runner-input-failure-1".to_string(), @@ -42704,6 +43282,18 @@ mod tests { failed.failure_category.as_deref(), Some("input_outcome_unknown") ); + let worker = failed.worker.clone().unwrap(); + wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &failed.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert!(matches!( + api.runtime.worker(&worker), + Err(RuntimeRegistryError::UnknownWorker { .. }) + )); assert!( api.retry_backend_job(&request.job_id).is_err(), "an ambiguous input outcome must not be retried automatically" @@ -42924,6 +43514,7 @@ mod tests { }; let dispatched = api.dispatch_backend_job(&request).unwrap(); let worker = dispatched.attempt.worker.clone().unwrap(); + seed_worker_session_for_cleanup(temp.path(), &worker); tokio::time::sleep(std::time::Duration::from_millis(1_200)).await; let attempt = api .store @@ -42936,6 +43527,17 @@ mod tests { .unwrap(); assert_eq!(attempt.state, BackendJobAttemptState::Failed); assert_eq!(attempt.failure_category.as_deref(), Some("timeout")); + wait_for_backend_job_worker_cleanup( + &api, + &request.job_id, + &dispatched.attempt.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert!(matches!( + api.runtime.worker(&worker), + Err(RuntimeRegistryError::UnknownWorker { .. }) + )); assert!( api.accept_backend_job_result( &worker, diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 30ecf1aa7..ecfebdec8 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -21,13 +21,14 @@ use crate::backend_job::{ BackendJobAttemptRecord, BackendJobAttemptState, BackendJobDeliveryRecord, BackendJobDeliveryState, BackendJobRecord, BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance, BackendJobResultSubmission, BackendJobState, - BackendJobWorkerBinding, attempt_id, bounded_failure_detail, result_digest, + BackendJobWorkerBinding, BackendJobWorkerCleanupState, attempt_id, bounded_failure_detail, + result_digest, }; use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 76; +const LATEST_SCHEMA_VERSION: i64 = 77; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -74,6 +75,8 @@ const RUNTIME_SCOPED_WORKER_IDENTITY_MIGRATION_NAME: &str = "Runtime-scoped durable Worker catalog identity"; const ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME: &str = "preserve explicit observe grants for committed Worker Session archives"; +const BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME: &str = + "durable terminal Backend Job Worker cleanup"; const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7; const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline"; const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#" @@ -251,6 +254,11 @@ const MIGRATIONS: &[Migration] = &[ name: ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME, apply: migrate_archive_observe_grants_v75_to_v76, }, + Migration { + version: 77, + name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME, + apply: migrate_backend_job_worker_cleanup_v76_to_v77, + }, ]; #[derive(Clone, Copy)] @@ -1592,6 +1600,34 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { failure_detail: &str, now: &str, ) -> Result; + fn recover_backend_job_worker_cleanups( + &self, + workspace_id: &str, + now: &str, + limit: usize, + ) -> Result>; + fn list_backend_job_worker_cleanups( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; + fn claim_backend_job_worker_cleanup( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + now: &str, + ) -> Result<(BackendJobAttemptRecord, bool)>; + fn finish_backend_job_worker_cleanup( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + state: BackendJobWorkerCleanupState, + failure_category: Option<&str>, + failure_detail: Option<&str>, + now: &str, + ) -> Result; fn reserve_backend_job_delivery( &self, workspace_id: &str, @@ -6648,7 +6684,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )); } let attempt_changed = tx.execute( - "UPDATE backend_job_attempts SET state = 'completed', result_json = ?4, result_digest = ?5, failure_category = NULL, failure_detail = NULL, completed_at = ?6, updated_at = ?6 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'dispatched'", + "UPDATE backend_job_attempts SET state = 'completed', result_json = ?4, result_digest = ?5, failure_category = NULL, failure_detail = NULL, worker_cleanup_state = 'pending', worker_cleanup_failure_category = NULL, worker_cleanup_failure_detail = NULL, worker_cleanup_updated_at = ?6, worker_cleanup_completed_at = NULL, completed_at = ?6, updated_at = ?6 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'dispatched'", params![workspace_id, submission.job_id, submission.attempt_id, result_json, digest, now], )?; let job_changed = tx.execute( @@ -6707,7 +6743,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { return Ok(attempt); } tx.execute( - "UPDATE backend_job_attempts SET state = ?4, failure_category = ?5, failure_detail = ?6, completed_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state IN ('reserved', 'dispatching', 'dispatched')", + "UPDATE backend_job_attempts SET state = ?4, failure_category = ?5, failure_detail = ?6, worker_cleanup_state = CASE WHEN worker_id IS NULL THEN 'completed' ELSE 'pending' END, worker_cleanup_failure_category = NULL, worker_cleanup_failure_detail = NULL, worker_cleanup_updated_at = ?7, worker_cleanup_completed_at = CASE WHEN worker_id IS NULL THEN ?7 ELSE NULL END, completed_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state IN ('reserved', 'dispatching', 'dispatched')", params![workspace_id, job_id, attempt_id, state.as_str(), failure_category, detail, now], )?; if attempt.attempt == job.current_attempt && job.state == BackendJobState::Pending { @@ -6728,6 +6764,154 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn recover_backend_job_worker_cleanups( + &self, + workspace_id: &str, + now: &str, + limit: usize, + ) -> Result> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + tx.execute( + "UPDATE backend_job_attempts SET worker_cleanup_state = 'pending', worker_cleanup_failure_category = 'cleanup_interrupted', worker_cleanup_failure_detail = 'Backend restart interrupted Worker cleanup; retrying canonical removal', worker_cleanup_updated_at = ?2 WHERE workspace_id = ?1 AND worker_cleanup_state = 'executing'", + params![workspace_id, now], + )?; + let mut statement = tx.prepare( + "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('completed', 'failed', 'unknown') AND runtime_id IS NOT NULL AND worker_cleanup_state IN ('pending', 'failed') ORDER BY worker_cleanup_updated_at, created_at LIMIT ?2", + )?; + let identities = statement + .query_map(params![workspace_id, limit as i64], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })? + .collect::>>()?; + drop(statement); + let attempts = identities + .into_iter() + .map(|(job_id, attempt_id)| { + read_backend_job_attempt(&tx, workspace_id, &job_id, &attempt_id)?.ok_or_else( + || Error::Store("recoverable Backend Job Worker cleanup disappeared".to_string()), + ) + }) + .collect::>>()?; + tx.commit()?; + Ok(attempts) + }) + } + + fn list_backend_job_worker_cleanups( + &self, + workspace_id: &str, + limit: usize, + ) -> Result> { + self.with_conn(|conn| { + let mut statement = conn.prepare( + "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('completed', 'failed', 'unknown') AND runtime_id IS NOT NULL AND worker_cleanup_state IN ('pending', 'failed') ORDER BY worker_cleanup_updated_at, created_at LIMIT ?2", + )?; + let identities = statement + .query_map(params![workspace_id, limit as i64], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })? + .collect::>>()?; + identities + .into_iter() + .map(|(job_id, attempt_id)| { + read_backend_job_attempt(conn, workspace_id, &job_id, &attempt_id)?.ok_or_else( + || Error::Store("recoverable Backend Job Worker cleanup disappeared".to_string()), + ) + }) + .collect() + }) + } + + fn claim_backend_job_worker_cleanup( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + now: &str, + ) -> Result<(BackendJobAttemptRecord, bool)> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?; + if !attempt.state.terminal() || attempt.worker.is_none() { + return Err(Error::InvalidInput( + "Backend Job Worker cleanup requires a terminal bound attempt".to_string(), + )); + } + if !matches!( + attempt.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Pending | BackendJobWorkerCleanupState::Failed) + ) { + tx.commit()?; + return Ok((attempt, false)); + } + let changed = tx.execute( + "UPDATE backend_job_attempts SET worker_cleanup_state = 'executing', worker_cleanup_failure_category = NULL, worker_cleanup_failure_detail = NULL, worker_cleanup_updated_at = ?4 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND worker_cleanup_state IN ('pending', 'failed')", + params![workspace_id, job_id, attempt_id, now], + )?; + if changed != 1 { + return Err(Error::Store( + "Backend Job Worker cleanup claim compare-and-set failed".to_string(), + )); + } + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| Error::Store("claimed Backend Job Worker cleanup disappeared".to_string()))?; + tx.commit()?; + Ok((attempt, true)) + }) + } + + fn finish_backend_job_worker_cleanup( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + state: BackendJobWorkerCleanupState, + failure_category: Option<&str>, + failure_detail: Option<&str>, + now: &str, + ) -> Result { + if !matches!( + state, + BackendJobWorkerCleanupState::Completed | BackendJobWorkerCleanupState::Failed + ) { + return Err(Error::InvalidInput( + "Backend Job Worker cleanup completion requires completed or failed state" + .to_string(), + )); + } + let failure_detail = failure_detail.map(bounded_failure_detail); + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let current = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?; + if current.worker_cleanup_state == Some(state) { + tx.commit()?; + return Ok(current); + } + if current.worker_cleanup_state != Some(BackendJobWorkerCleanupState::Executing) { + return Err(Error::InvalidInput( + "Backend Job Worker cleanup is not executing".to_string(), + )); + } + let completed_at = (state == BackendJobWorkerCleanupState::Completed).then_some(now); + let changed = tx.execute( + "UPDATE backend_job_attempts SET worker_cleanup_state = ?4, worker_cleanup_failure_category = ?5, worker_cleanup_failure_detail = ?6, worker_cleanup_updated_at = ?7, worker_cleanup_completed_at = ?8 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND worker_cleanup_state = 'executing'", + params![workspace_id, job_id, attempt_id, state.as_str(), failure_category, failure_detail, now, completed_at], + )?; + if changed != 1 { + return Err(Error::Store( + "Backend Job Worker cleanup completion compare-and-set failed".to_string(), + )); + } + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| Error::Store("finished Backend Job Worker cleanup disappeared".to_string()))?; + tx.commit()?; + Ok(attempt) + }) + } + fn reserve_backend_job_delivery( &self, workspace_id: &str, @@ -10457,14 +10641,16 @@ fn read_backend_job_attempt( attempt_id: &str, ) -> Result> { let row = conn.query_row( - "SELECT attempt, input_revision, state, runtime_id, worker_id, runtime_run_id, dispatched_at, deadline_at, result_json, result_digest, failure_category, failure_detail, created_at, updated_at, completed_at FROM backend_job_attempts WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3", + "SELECT attempt, input_revision, state, runtime_id, worker_id, runtime_run_id, dispatched_at, deadline_at, result_json, result_digest, failure_category, failure_detail, worker_cleanup_state, worker_cleanup_failure_category, worker_cleanup_failure_detail, worker_cleanup_updated_at, worker_cleanup_completed_at, created_at, updated_at, completed_at FROM backend_job_attempts WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3", params![workspace_id, job_id, attempt_id], |row| Ok(( row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, Option>(3)?, row.get::<_, Option>(4)?, row.get::<_, Option>(5)?, row.get::<_, Option>(6)?, row.get::<_, String>(7)?, row.get::<_, Option>(8)?, row.get::<_, Option>(9)?, row.get::<_, Option>(10)?, row.get::<_, Option>(11)?, - row.get::<_, String>(12)?, row.get::<_, String>(13)?, row.get::<_, Option>(14)?, + row.get::<_, Option>(12)?, row.get::<_, Option>(13)?, row.get::<_, Option>(14)?, + row.get::<_, Option>(15)?, row.get::<_, Option>(16)?, row.get::<_, String>(17)?, + row.get::<_, String>(18)?, row.get::<_, Option>(19)?, )), ).optional()?; let Some(( @@ -10480,6 +10666,11 @@ fn read_backend_job_attempt( result_digest, failure_category, failure_detail, + worker_cleanup_state, + worker_cleanup_failure_category, + worker_cleanup_failure_detail, + worker_cleanup_updated_at, + worker_cleanup_completed_at, created_at, updated_at, completed_at, @@ -10521,6 +10712,14 @@ fn read_backend_job_attempt( result_digest, failure_category, failure_detail, + worker_cleanup_state: worker_cleanup_state + .as_deref() + .map(BackendJobWorkerCleanupState::parse) + .transpose()?, + worker_cleanup_failure_category, + worker_cleanup_failure_detail, + worker_cleanup_updated_at, + worker_cleanup_completed_at, created_at, updated_at, completed_at, @@ -14248,6 +14447,45 @@ fn migrate_archive_observe_grants_v75_to_v76(conn: &Connection) -> Result<()> { Ok(()) } +fn migrate_backend_job_worker_cleanup_v76_to_v77(conn: &Connection) -> Result<()> { + let current = current_schema_version(conn)?; + if current != 76 { + return Err(Error::Store(format!( + "expected schema version 76 before {BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME} migration, found {current}" + ))); + } + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + tx.execute_batch( + r#" + ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_state TEXT + CHECK (worker_cleanup_state IN ('pending', 'executing', 'completed', 'failed')); + ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_failure_category TEXT; + ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_failure_detail TEXT; + ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_updated_at TEXT; + ALTER TABLE backend_job_attempts ADD COLUMN worker_cleanup_completed_at TEXT; + UPDATE backend_job_attempts + SET worker_cleanup_state = CASE + WHEN runtime_id IS NULL THEN 'completed' + ELSE 'pending' + END, + worker_cleanup_updated_at = updated_at, + worker_cleanup_completed_at = CASE + WHEN runtime_id IS NULL THEN updated_at + ELSE NULL + END + WHERE state IN ('completed', 'failed', 'unknown'); + CREATE INDEX backend_job_attempts_worker_cleanup + ON backend_job_attempts(workspace_id, worker_cleanup_state, updated_at); + "#, + )?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![77_i64, BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?; conn.execute_batch(include_str!("latest_schema.sql"))?; @@ -16810,6 +17048,10 @@ mod tests { version: 76, name: ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 77, + name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(), + }, ] ); @@ -16893,6 +17135,10 @@ mod tests { RUNTIME_SCOPED_WORKER_IDENTITY_MIGRATION_NAME.to_string(), ), (76, ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME.to_string()), + ( + 77, + BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(), + ), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -17160,6 +17406,71 @@ mod tests { ); } + #[test] + fn backend_job_worker_cleanup_migration_queues_legacy_terminal_workers() { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch( + r#" + CREATE TABLE __yoi_schema_migrations ( + version INTEGER PRIMARY KEY, + name TEXT NOT NULL + ); + INSERT INTO __yoi_schema_migrations(version, name) + VALUES (76, 'workspace schema baseline'); + CREATE TABLE backend_job_attempts ( + workspace_id TEXT NOT NULL, + job_id TEXT NOT NULL, + attempt_id TEXT NOT NULL, + state TEXT NOT NULL, + runtime_id TEXT, + worker_id TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + INSERT INTO backend_job_attempts VALUES + ('w', 'terminal-bound', 'a1', 'completed', 'runtime', 'worker', '1', '2'), + ('w', 'terminal-unbound', 'a2', 'failed', NULL, NULL, '1', '3'), + ('w', 'running', 'a3', 'dispatched', 'runtime', 'running-worker', '1', '4'); + "#, + ) + .unwrap(); + + migrate_backend_job_worker_cleanup_v76_to_v77(&conn).unwrap(); + + let states = conn + .prepare( + "SELECT job_id, worker_cleanup_state, worker_cleanup_completed_at FROM backend_job_attempts ORDER BY job_id", + ) + .unwrap() + .query_map([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, Option>(1)?, + row.get::<_, Option>(2)?, + )) + }) + .unwrap() + .collect::>>() + .unwrap(); + assert_eq!( + states, + vec![ + ("running".to_string(), None, None,), + ( + "terminal-bound".to_string(), + Some("pending".to_string()), + None, + ), + ( + "terminal-unbound".to_string(), + Some("completed".to_string()), + Some("3".to_string()), + ), + ] + ); + assert_eq!(current_schema_version(&conn).unwrap(), 77); + } + #[test] fn migration_resumes_from_a_valid_partially_applied_chain() { let temp = tempfile::tempdir().unwrap(); @@ -17179,7 +17490,7 @@ mod tests { .collect::>(), vec![ 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, - 73, 74, 75, 76 + 73, 74, 75, 76, 77 ] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); @@ -17188,7 +17499,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 27); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 28); assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap()); assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap()); assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap());