diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 0c4fc8656..fe3097cf4 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2212,6 +2212,97 @@ impl ServerApiContractService { } } +#[derive(Debug, thiserror::Error)] +enum WorkerRemovalFailure { + #[error("{0}")] + Internal(String), + #[error("{error}")] + Storage { + error: Error, + response: Option, + }, +} + +impl From for WorkerRemovalFailure { + fn from(message: String) -> Self { + Self::Internal(message) + } +} + +impl From<&str> for WorkerRemovalFailure { + fn from(message: &str) -> Self { + Self::Internal(message.to_owned()) + } +} + +impl From for WorkerRemovalFailure { + fn from(error: Error) -> Self { + Self::Storage { + error, + response: None, + } + } +} + +impl WorkerRemovalFailure { + fn retention_response( + error: crate::retention::WorkerRetentionError, + ) -> std::result::Result { + match error { + crate::retention::WorkerRetentionError::Store(error) => Err(Self::Storage { + error, + response: Some(worker_remove_error_response( + StatusCode::SERVICE_UNAVAILABLE, + "worker_removal_authority_unavailable", + "Worker removal authority is unavailable; removal can be retried", + )), + }), + error => Ok(worker_retention_error_response(error)), + } + } + + fn storage_response( + response: worker::WorkspaceResponse, + error: Error, + ) -> std::result::Result { + match error { + Error::Io(_) + | Error::Sqlite(_) + | Error::SubjektivStorage(_) + | Error::DbCapacityExceeded + | Error::DbExecutorUnavailable + | Error::DbOperationOutcomeUnknown + | Error::Store(_) => Err(Self::Storage { + error, + response: Some(response), + }), + _ => Ok(response), + } + } + + fn into_error(self, runtime_id: &str) -> Error { + match self { + Self::Internal(message) => Error::RuntimeOperationFailed { + runtime_id: runtime_id.to_owned(), + code: "backend_job_cleanup_removal_failed".to_string(), + message, + }, + Self::Storage { error, .. } => error, + } + } + + fn into_response(self) -> std::result::Result { + match self { + Self::Internal(message) => Err(message), + Self::Storage { + response: Some(response), + .. + } => Ok(response), + Self::Storage { error, .. } => Err(error.to_string()), + } + } +} + #[derive(Clone)] struct WorkerRemovalService { workspace_id: String, @@ -2376,7 +2467,7 @@ impl WorkerRemovalService { target: &RuntimeWorkerRef, mut prepared: crate::retention::PreparedWorkerRemoval, parent_workspace_deletion_operation_id: Option<&str>, - ) -> std::result::Result { + ) -> std::result::Result { if prepared.plan.session_disposition == worker_runtime::retention::SessionDisposition::Archive { @@ -2385,11 +2476,14 @@ impl WorkerRemovalService { ) { Ok(ack) => ack, Err(error) => { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "erased_removal_reconciliation_unconfirmed", - &error.to_string(), - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::CONFLICT, + "erased_removal_reconciliation_unconfirmed", + &error.to_string(), + ), + error, + ); } }; if let Some(ack) = erasure_ack { @@ -2401,7 +2495,7 @@ impl WorkerRemovalService { .store .audit_erased_worker_removal_receipt(&prepared, &ack, &result) { - return Ok(worker_retention_error_response(error)); + return WorkerRemovalFailure::retention_response(error); } return match self.worker_projection.publish_ordered(|| { self.store @@ -2414,7 +2508,7 @@ impl WorkerRemovalService { .map(|commit| ((), commit.catalog)) }) { Ok(()) => Ok(worker_remove_success_response(target)), - Err(error) => Ok(worker_retention_error_response(error)), + Err(error) => WorkerRemovalFailure::retention_response(error), }; } Err(_) => { @@ -2434,7 +2528,7 @@ impl WorkerRemovalService { .store .supersede_erased_worker_removal(&prepared, &ack, &inventory) { - return Ok(worker_retention_error_response(error)); + return WorkerRemovalFailure::retention_response(error); } let request = crate::retention::WorkerRemovalPlanRequest { workspace_id: self.workspace_id.clone(), @@ -2444,7 +2538,7 @@ impl WorkerRemovalService { }; let plan = match self.store.plan_worker_removal(&request, &inventory) { Ok(plan) => plan, - Err(error) => return Ok(worker_retention_error_response(error)), + Err(error) => return WorkerRemovalFailure::retention_response(error), }; prepared = match self.store.prepare_worker_removal_execution( &self.workspace_id, @@ -2452,7 +2546,7 @@ impl WorkerRemovalService { &plan.input_fingerprint, ) { Ok(prepared) => prepared, - Err(error) => return Ok(worker_retention_error_response(error)), + Err(error) => return WorkerRemovalFailure::retention_response(error), }; if let Some(parent) = parent_workspace_deletion_operation_id { self.store @@ -2460,18 +2554,21 @@ impl WorkerRemovalService { parent, &prepared.plan.operation_id, ) - .map_err(|error| error.to_string())?; + .map_err(WorkerRemovalFailure::from)?; } } } } } if let Err(error) = subjektiv_journal::validate_removal_retention(self, target, &prepared) { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "journal_retention_unconfirmed", - &error.to_string(), - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::CONFLICT, + "journal_retention_unconfirmed", + &error.to_string(), + ), + error, + ); } let result = match runtime.execute_worker_retention(target, prepared.runtime_request.clone()) { @@ -2500,11 +2597,14 @@ impl WorkerRemovalService { }) { Ok(result) => result, Err(error) => { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "subject_removal_completion_unconfirmed", - &error.to_string(), - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::CONFLICT, + "subject_removal_completion_unconfirmed", + &error.to_string(), + ), + error, + ); } }; match commit_result { @@ -2512,7 +2612,7 @@ impl WorkerRemovalService { subjektiv_journal::after_worker_removal(self, &result); Ok(worker_remove_success_response(target)) } - Err(error) => Ok(worker_retention_error_response(error)), + Err(error) => WorkerRemovalFailure::retention_response(error), } } @@ -2653,6 +2753,7 @@ impl WorkerRemovalService { None, ) .await + .or_else(WorkerRemovalFailure::into_response) } /// Restarted Job cleanup may resume only its already-observed canonical @@ -2665,7 +2766,7 @@ impl WorkerRemovalService { reason: &str, parent_workspace_deletion_operation_id: Option<&str>, job_cleanup_owner: Option<&Mutex>>, - ) -> std::result::Result { + ) -> std::result::Result { let remove_lock = { let mut locks = self .worker_remove_locks @@ -2692,7 +2793,14 @@ impl WorkerRemovalService { let prepared = self .store .recover_worker_removal_execution(&self.workspace_id, &target) - .map_err(|_| "Worker removal recovery authority is unavailable".to_string())?; + .map_err(|error| match error { + crate::retention::WorkerRetentionError::Store(error) => { + WorkerRemovalFailure::from(error) + } + _ => WorkerRemovalFailure::Internal( + "Worker removal recovery authority is unavailable".to_string(), + ), + })?; let expected_removal = job_cleanup_owner .map(|owner| { owner @@ -2746,7 +2854,7 @@ impl WorkerRemovalService { .recover_succeeded_worker_removal_catalog(&self.workspace_id, target) .map(|commit| ((), commit)) }) - .map_err(|error| error.to_string())?; + .map_err(WorkerRemovalFailure::from)?; return Ok(worker_remove_success_response(target)); } let prepared = if matches!( @@ -2760,7 +2868,7 @@ impl WorkerRemovalService { &prepared.plan.input_fingerprint, ) { Ok(prepared) => prepared, - Err(error) => return Ok(worker_retention_error_response(error)), + Err(error) => return WorkerRemovalFailure::retention_response(error), } } else { prepared @@ -2771,7 +2879,7 @@ impl WorkerRemovalService { parent_operation_id, &prepared.plan.operation_id, ) - .map_err(|error| error.to_string())?; + .map_err(WorkerRemovalFailure::from)?; } if close_worker_workdir_sessions(&self.workdir_sessions, &target) .await @@ -2789,27 +2897,26 @@ impl WorkerRemovalService { "Worker Workdir sessions could not be closed; removal can be retried", )); } - if self - .store - .detach_worker_workdir( - &self.workspace_id, - &target, - None, - &Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), - ) - .is_err() - { + if let Err(error) = self.store.detach_worker_workdir( + &self.workspace_id, + &target, + None, + &Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), + ) { let _ = self.store.fail_worker_removal( &self.workspace_id, &prepared.plan.operation_id, &prepared.plan.input_fingerprint, "workdir_attachment_release_failed", ); - return Ok(worker_remove_error_response( - StatusCode::SERVICE_UNAVAILABLE, - "attachment_release_failed", - "Worker Workdir attachment could not be released; removal can be retried", - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::SERVICE_UNAVAILABLE, + "attachment_release_failed", + "Worker Workdir attachment could not be released; removal can be retried", + ), + error, + ); } return self .resume_worker_retention( @@ -2843,11 +2950,14 @@ impl WorkerRemovalService { { Ok(proof) => proof, Err(error) => { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "journal_handoff_incomplete", - &error.to_string(), - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::CONFLICT, + "journal_handoff_incomplete", + &error.to_string(), + ), + error, + ); } }; let inventory = match runtime.worker_retention_inventory(&target) { @@ -2873,7 +2983,7 @@ impl WorkerRemovalService { self.store.plan_worker_removal(&request, &inventory) } { Ok(plan) => plan, - Err(error) => return Ok(worker_retention_error_response(error)), + Err(error) => return WorkerRemovalFailure::retention_response(error), }; if let Some(owner) = job_cleanup_owner { *owner @@ -2887,7 +2997,7 @@ impl WorkerRemovalService { &plan.input_fingerprint, ) { Ok(prepared) => prepared, - Err(error) => return Ok(worker_retention_error_response(error)), + Err(error) => return WorkerRemovalFailure::retention_response(error), }; if let Some(parent_operation_id) = parent_workspace_deletion_operation_id { @@ -2896,7 +3006,7 @@ impl WorkerRemovalService { parent_operation_id, &prepared.plan.operation_id, ) - .map_err(|error| error.to_string())?; + .map_err(WorkerRemovalFailure::from)?; } if close_worker_workdir_sessions(&self.workdir_sessions, &target) .await @@ -2915,7 +3025,7 @@ impl WorkerRemovalService { )); } - if let Err(_) = self.store.detach_worker_workdir( + if let Err(error) = self.store.detach_worker_workdir( &self.workspace_id, &target, None, @@ -2927,19 +3037,25 @@ impl WorkerRemovalService { &plan.input_fingerprint, "workdir_attachment_release_failed", ); - return Ok(worker_remove_error_response( - StatusCode::SERVICE_UNAVAILABLE, - "attachment_release_failed", - "Worker Workdir attachment could not be released; removal can be retried", - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::SERVICE_UNAVAILABLE, + "attachment_release_failed", + "Worker Workdir attachment could not be released; removal can be retried", + ), + error, + ); } if let Err(error) = subjektiv_journal::validate_removal_retention(self, target, &prepared) { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "journal_retention_unconfirmed", - &error.to_string(), - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::CONFLICT, + "journal_retention_unconfirmed", + &error.to_string(), + ), + error, + ); } let retention_result = match runtime.execute_worker_retention(&target, prepared.runtime_request.clone()) { @@ -2979,11 +3095,14 @@ impl WorkerRemovalService { ) { Ok(result) => result, Err(error) => { - return Ok(worker_remove_error_response( - StatusCode::CONFLICT, - "subject_removal_completion_unconfirmed", - &error.to_string(), - )); + return WorkerRemovalFailure::storage_response( + worker_remove_error_response( + StatusCode::CONFLICT, + "subject_removal_completion_unconfirmed", + &error.to_string(), + ), + error, + ); } }; match commit_result { @@ -2995,7 +3114,7 @@ impl WorkerRemovalService { &plan.input_fingerprint, "metadata_commit_failed", ); - return Ok(worker_retention_error_response(error)); + return WorkerRemovalFailure::retention_response(error); } }; subjektiv_journal::after_worker_removal(self, &retention_result); @@ -5469,22 +5588,39 @@ impl WorkspaceApi { Error::Store("claimed Backend Job Worker cleanup lost its binding".to_string()) })?; - let fail = |category: &str, detail: String| -> Result<()> { + let record_failure = |category: &str, detail: &str| -> Result<()> { self.store.finish_backend_job_worker_cleanup_owned( &self.config.workspace_id, &history, &attempt, BackendJobWorkerCleanupState::Failed, Some(category), - Some(&detail), + Some(detail), &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), )?; + Ok(()) + }; + let fail = |category: &str, detail: String| -> Result<()> { + record_failure(category, &detail)?; Err(Error::RuntimeOperationFailed { runtime_id: worker.runtime_id.clone(), code: category.to_string(), message: detail, }) }; + let fail_with_error = |category: &str, error: Error| -> Result<()> { + if let Err(diagnostic) = record_failure(category, &error.to_string()) { + tracing::warn!(%error, %diagnostic, "Cleanup failure diagnosis was not confirmed"); + // Keep transient storage ownership even when a definitive operation + // failure could not be diagnosed durably. Neither error proves cleanup. + if !backend_job_recovery::transient_storage(&error) + && backend_job_recovery::transient_storage(&diagnostic) + { + return Err(diagnostic); + } + } + Err(error) + }; let complete = || -> Result<()> { self.require_backend_job_attempt_effects_ended(&attempt)?; let journal_job = self @@ -5495,7 +5631,10 @@ impl WorkspaceApi { let removal = self .config_store .recover_worker_removal_execution(self.workspace_id(), &worker) - .map_err(|e| Error::Store(e.to_string()))?; + .map_err(|error| match error { + crate::retention::WorkerRetentionError::Store(error) => error, + error => Error::Store(error.to_string()), + })?; if !removal.is_some_and(|p| { p.plan.state == crate::retention::WorkerRemovalPlanState::Succeeded && p.runtime_request.session_disposition @@ -5552,7 +5691,7 @@ impl WorkspaceApi { Err(error) => return fail("backend_job_cleanup_stop_unconfirmed", error.message()), } if let Err(error) = self.fence_backend_job_effects(&attempt, drain).await { - return fail("backend_job_cleanup_effects_unconfirmed", error.to_string()); + return fail_with_error("backend_job_cleanup_effects_unconfirmed", error); } } @@ -5564,15 +5703,15 @@ impl WorkspaceApi { None, Some(removal_owner), ) - .await - .map_err(|message| Error::RuntimeOperationFailed { - runtime_id: worker.runtime_id.clone(), - code: "backend_job_cleanup_removal_failed".to_string(), - message, - }); + .await; let response = match response { Ok(response) => response, - Err(error) => return fail("backend_job_cleanup_removal_failed", error.to_string()), + Err(error) => { + return fail_with_error( + "backend_job_cleanup_removal_failed", + error.into_error(&worker.runtime_id), + ); + } }; if response.status == StatusCode::OK.as_u16() { return complete(); diff --git a/crates/workspace-server/src/server/backend_job_recovery.rs b/crates/workspace-server/src/server/backend_job_recovery.rs index 9b2d3422b..2ebcf0315 100644 --- a/crates/workspace-server/src/server/backend_job_recovery.rs +++ b/crates/workspace-server/src/server/backend_job_recovery.rs @@ -33,6 +33,10 @@ pub(super) struct RecoveryOwners { #[cfg(test)] pub(super) nonjournal_lookup_pending: tokio::sync::Notify, #[cfg(test)] + pub(super) cleanup_storage_settlement: tokio::sync::Notify, + #[cfg(test)] + pub(super) cleanup_stopped: tokio::sync::Notify, + #[cfg(test)] pub(super) cleanup_before_claim: Mutex>>, #[cfg(test)] pub(super) cleanup_before_writer_fence: tokio::sync::Notify, @@ -205,7 +209,7 @@ fn transient_subjektiv_storage(error: &Error) -> bool { } } -fn transient_storage(error: &Error) -> bool { +pub(super) fn transient_storage(error: &Error) -> bool { transient_subjektiv_storage(error) || workspace_storage_failure(error) } @@ -483,6 +487,8 @@ impl WorkspaceApi { if let Err(error) = api.run_backend_job_cleanup(expected.clone()).await { tracing::warn!(job_id=%expected.job_id, attempt_id=%expected.attempt_id, %error, "Job cleanup stopped; execution/input/resource fences remain retained"); + #[cfg(test)] + api.backend_job_recovery.cleanup_stopped.notify_one(); } }); true @@ -494,7 +500,7 @@ impl WorkspaceApi { ) -> Result<()> { // Pin the observed canonical operation across this owner's storage retries. let removal_owner = Arc::new(Mutex::new(None::<(String, String)>)); - let mut failures = 0; + let mut failures = 0usize; loop { if self.backend_job_recovery.stopping.load(Ordering::Acquire) { return Ok(()); @@ -611,7 +617,7 @@ impl WorkspaceApi { .lock() .unwrap() .clone(); - runtime.block_on(async { + let result = runtime.block_on(async { #[cfg(test)] if let Some(gate) = after_claim { api.backend_job_recovery.cleanup_claimed_writer.notify_one(); @@ -625,9 +631,40 @@ impl WorkspaceApi { } api.cleanup_backend_job_attempt_worker(attempt, &removal_owner, &drain) .await - }) + }); + if let Err(error) = &result + && transient_subjektiv_storage(error) + { + let history = api.store.list_backend_job_attempts(api.workspace_id(), &owner.job_id)?; + let current = history.iter().find(|attempt| attempt.attempt_id == owner.attempt_id) + .ok_or_else(|| Error::Store("Feature cleanup owner disappeared".into()))?; + if JobHostOwner::try_from(current)? != JobHostOwner::try_from(&owner)? + || !current.state.terminal() + { + return Err(Error::RepositoryConflict("Feature cleanup diagnostic owner changed".into())); + } + api.store.finish_backend_job_worker_cleanup_owned( + api.workspace_id(), &history, current, + BackendJobWorkerCleanupState::Failed, + Some("backend_job_cleanup_feature_storage_unavailable"), + Some(&format!("{error}; original cleanup owner retains input and canonical child; reconcile the same Feature operation after backoff, never infer termination or resubmit input")), + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + } + result }) .await; + #[cfg(test)] + if failures >= 5 + && result.as_ref().is_err_and(|error| { + transient_storage(error) + || matches!(error, Error::RuntimeOperationFailed { .. }) + }) + { + self.backend_job_recovery + .cleanup_storage_settlement + .notify_one(); + } match result { Ok(()) => return Ok(()), Err(error) @@ -640,6 +677,16 @@ impl WorkspaceApi { .await; failures += 1; } + Err(error) if transient_subjektiv_storage(&error) => { + // This is the original Feature operation, not a healthy Workspace + // probe. Re-enter its owner/history, retained-writer and canonical + // child checks; no new execution or replacement removal is allowed. + // Terminal Job deadlines do not authorize discarding failed cleanup. + let delay = + Duration::from_secs((1u64 << failures.saturating_sub(5).min(5)).min(30)); + failures = failures.saturating_add(1); + self.backend_job_recovery.pause(delay).await; + } Err(error) if workspace_storage_failure(&error) => { tracing::warn!(attempt_id=%expected.attempt_id, %error, "Cleanup settlement budget exhausted; exact-owner storage probe only"); let mut delay = Duration::from_secs(1); diff --git a/crates/workspace-server/src/server/tests/backend_job_tests.rs b/crates/workspace-server/src/server/tests/backend_job_tests.rs index f9d0531bb..78016fade 100644 --- a/crates/workspace-server/src/server/tests/backend_job_tests.rs +++ b/crates/workspace-server/src/server/tests/backend_job_tests.rs @@ -1111,15 +1111,15 @@ async fn backend_job_cleanup_commit_failure_retries_old_attempt_without_touching 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") + Some("backend_job_cleanup_removal_failed") ); assert!( blocked .worker_cleanup_failure_detail .as_deref() .unwrap() - .contains("worker_removal_authority_unavailable"), - "canonical catalog commit failure must remain a removal error: {blocked:?}" + .contains("injected cleanup catalog failure"), + "cleanup must retain the original catalog cause, not a flattened HTTP rejection: {blocked:?}" ); let blocked_removal = api .config_store diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index 5e2817620..33b5549f2 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -1339,6 +1339,342 @@ fn lock_journal_domain(f: &RuntimeFixture) -> JournalCompletionSaveFault { JournalCompletionSaveFault(connection) } +#[tokio::test] +async fn cleanup_feature_storage_outage_beyond_budget_recovers_without_sweep() { + let f = RuntimeFixture::new().await; + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + *f.api + .backend_job_recovery + .cleanup_after_claim + .lock() + .unwrap() = Some(gate.clone()); + let dispatched = f.dispatch_input(); + let worker = dispatched.attempt.worker.as_ref().unwrap(); + f.execution.take_inputs(); + f.api + .finish_backend_job_attempt( + &f.request.job_id, + &dispatched.attempt.attempt_id, + BackendJobAttemptState::Failed, + "cancelled", + "retain the uncheckpointed original input", + ) + .unwrap(); + tokio::time::timeout( + std::time::Duration::from_secs(10), + f.api.backend_job_recovery.cleanup_claimed_writer.notified(), + ) + .await + .unwrap(); + let fault = lock_journal_domain(&f); + *f.api + .backend_job_recovery + .cleanup_after_claim + .lock() + .unwrap() = None; + gate.add_permits(1); + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api + .backend_job_recovery + .cleanup_storage_settlement + .notified(), + ) + .await + .expect("real Feature storage fault did not exhaust initial cleanup settlement"); + let blocked = current_journal_attempt(&f, &dispatched.attempt.attempt_id); + assert_eq!(blocked.state, BackendJobAttemptState::Failed); + assert_eq!(blocked.failure_category.as_deref(), Some("cancelled")); + assert_ne!( + blocked.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Completed) + ); + assert!( + f.api + .backend_job_recovery + .cleanup_owner_ids() + .contains(&blocked.attempt_id), + "Feature outage exhausted cleanup retries and lost its autonomous owner" + ); + assert_eq!( + blocked.worker_cleanup_failure_category.as_deref(), + Some("backend_job_cleanup_feature_storage_unavailable") + ); + // The Workspace is healthy, but cannot certify the locked Feature operation. + assert!( + f.api + .store + .probe_backend_job_recovery(TEST_WORKSPACE_ID, &blocked) + .unwrap() + ); + tokio::time::timeout( + std::time::Duration::from_secs(10), + f.api + .backend_job_recovery + .cleanup_storage_settlement + .notified(), + ) + .await + .expect("source-aware cleanup reconciliation lost its owner after settlement"); + assert_ne!( + current_journal_attempt(&f, &blocked.attempt_id).worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Completed) + ); + assert!( + f.api + .backend_job_recovery + .cleanup_owner_ids() + .contains(&blocked.attempt_id) + ); + let original = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker) + .unwrap() + .unwrap(); + drop(fault); + // Only release the original Feature writer lock; do not sweep, reopen or replay. + let cleaned = wait_for_backend_job_worker_cleanup( + &f.api, + &f.request.job_id, + &blocked.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert_eq!(cleaned.worker, blocked.worker); + assert_eq!(cleaned.runtime_run_id, blocked.runtime_run_id); + assert_eq!(cleaned.failure_category, blocked.failure_category); + let after = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker) + .unwrap() + .unwrap(); + assert_eq!(after.plan.operation_id, original.plan.operation_id); + assert_eq!( + after.plan.input_fingerprint, + original.plan.input_fingerprint + ); + assert!(f.execution.take_inputs().is_empty()); + let saved = f + .api + .store + .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id) + .unwrap() + .unwrap(); + assert_eq!( + saved.input_disposition, + crate::backend_job::JobInputDisposition::Available + ); + assert!(f.store.journal_retention_required(&f.host).unwrap()); + f.api.stop_backend_job_recovery().await; +} + +struct CleanupCatalogCapacityFault { + store: Arc, + previous_limit: i64, +} + +impl CleanupCatalogCapacityFault { + fn install(f: &RuntimeFixture) -> Self { + let previous_limit = f + .job_store + .with_conn(|connection| { + connection.execute_batch( + "CREATE TABLE cleanup_capacity_fault(payload BLOB); + CREATE TRIGGER fill_cleanup_catalog_capacity + BEFORE INSERT ON worker_registry_projection_removals + BEGIN INSERT INTO cleanup_capacity_fault VALUES(zeroblob(4194304)); END;", + )?; + let previous: i64 = + connection.query_row("PRAGMA max_page_count", [], |row| row.get(0))?; + let pages: i64 = connection.query_row("PRAGMA page_count", [], |row| row.get(0))?; + // Small owner/diagnostic writes fit; the catalog commit's trigger does not. + connection.pragma_update(None, "max_page_count", pages + 32)?; + let error = connection + .execute( + "INSERT INTO cleanup_capacity_fault VALUES(zeroblob(4194304))", + [], + ) + .unwrap_err(); + assert!(matches!(error, rusqlite::Error::SqliteFailure(error, _) + if error.code == rusqlite::ErrorCode::DiskFull)); + Ok(previous) + }) + .unwrap(); + Self { + store: f.job_store.clone(), + previous_limit, + } + } +} + +impl Drop for CleanupCatalogCapacityFault { + fn drop(&mut self) { + self.store.with_conn(|connection| { + connection.pragma_update(None, "max_page_count", self.previous_limit)?; + connection.execute_batch("DROP TRIGGER fill_cleanup_catalog_capacity; DROP TABLE cleanup_capacity_fault;")?; + Ok(()) + }).unwrap(); + } +} + +#[tokio::test] +async fn cleanup_catalog_storage_outage_beyond_budget_recovers_original_child() { + cleanup_catalog_storage_reconciliation(false).await; +} + +#[tokio::test] +async fn cleanup_catalog_storage_recovery_rejects_changed_canonical_authority() { + cleanup_catalog_storage_reconciliation(true).await; +} + +async fn cleanup_catalog_storage_reconciliation(change_authority: bool) { + let f = RuntimeFixture::new().await; + let dispatched = f.dispatch_input(); + let worker = dispatched.attempt.worker.as_ref().unwrap(); + f.execution.take_inputs(); + let fault = CleanupCatalogCapacityFault::install(&f); + f.api + .finish_backend_job_attempt( + &f.request.job_id, + &dispatched.attempt.attempt_id, + BackendJobAttemptState::Failed, + "cancelled", + "retain the uncheckpointed original input", + ) + .unwrap(); + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api + .backend_job_recovery + .cleanup_storage_settlement + .notified(), + ) + .await + .expect("real catalog capacity fault did not exhaust initial cleanup settlement"); + let blocked = current_journal_attempt(&f, &dispatched.attempt.attempt_id); + assert_eq!(blocked.failure_category.as_deref(), Some("cancelled")); + assert_ne!( + blocked.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Completed) + ); + assert!( + f.api + .backend_job_recovery + .cleanup_owner_ids() + .contains(&blocked.attempt_id), + "catalog capacity outage lost its autonomous cleanup owner" + ); + let original = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker) + .unwrap() + .unwrap(); + assert_eq!( + original.prior_failure_category.as_deref(), + Some("metadata_commit_failed") + ); + assert!( + f.api + .require_backend_job_attempt_effects_ended(&blocked) + .is_ok() + ); + if change_authority { + // Simulate a canonical child's invalidation while storage is unavailable. + // Normal pin updates are fenced during an admitted removal and cannot do this. + f.api.config_store.with_conn(|connection| { + assert_eq!(connection.execute("UPDATE worker_removal_operations SET state='stale', failure_category='fixture_child_invalidated' WHERE workspace_id=?1 AND operation_id=?2 AND input_fingerprint=?3", + rusqlite::params![TEST_WORKSPACE_ID, original.plan.operation_id, original.plan.input_fingerprint])?, 1); + Ok(()) + }).unwrap(); + } + drop(fault); + // Restore only the failed catalog write's capacity; no sweep/reopen/result replay. + if change_authority { + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api.backend_job_recovery.cleanup_stopped.notified(), + ) + .await + .unwrap_or_else(|error| panic!("changed removal authority did not reach bounded rejection: {error}; current={:?}; owners={:?}; removal={:?}", + current_journal_attempt(&f, &blocked.attempt_id), f.api.backend_job_recovery.cleanup_owner_ids(), + f.api.config_store.recover_worker_removal_execution(TEST_WORKSPACE_ID, worker))); + assert!( + !f.api + .backend_job_recovery + .cleanup_owner_ids() + .contains(&blocked.attempt_id) + ); + let rejected = current_journal_attempt(&f, &blocked.attempt_id); + assert_eq!( + rejected.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Failed) + ); + assert_eq!(rejected.failure_category, blocked.failure_category); + f.api.config_store.with_conn(|connection| { + let count: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![TEST_WORKSPACE_ID, worker.runtime_id, worker.worker_id], |row| row.get(0))?; + assert_eq!(count, 1); + Ok(()) + }).unwrap(); + assert!(f.execution.take_inputs().is_empty()); + assert_eq!( + f.api + .store + .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id) + .unwrap() + .unwrap() + .input_disposition, + crate::backend_job::JobInputDisposition::Available + ); + assert!(f.store.journal_retention_required(&f.host).unwrap()); + f.api.stop_backend_job_recovery().await; + return; + } + let cleaned = wait_for_backend_job_worker_cleanup( + &f.api, + &f.request.job_id, + &blocked.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + let after = f + .api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker) + .unwrap() + .unwrap(); + assert_eq!(after.plan.operation_id, original.plan.operation_id); + assert_eq!( + after.plan.input_fingerprint, + original.plan.input_fingerprint + ); + assert_eq!(cleaned.worker, blocked.worker); + assert_eq!(cleaned.runtime_run_id, blocked.runtime_run_id); + assert_eq!(cleaned.failure_category, blocked.failure_category); + assert!(f.execution.take_inputs().is_empty()); + assert_eq!( + f.api + .store + .list_backend_job_attempts(TEST_WORKSPACE_ID, &f.request.job_id) + .unwrap() + .len(), + 1 + ); + assert_eq!( + f.api + .store + .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id) + .unwrap() + .unwrap() + .input_disposition, + crate::backend_job::JobInputDisposition::Available + ); + assert!(f.store.journal_retention_required(&f.host).unwrap()); + f.api.stop_backend_job_recovery().await; +} + #[tokio::test] async fn domain_storage_lock_beyond_initial_settlement_recovers_original_journal_attempt() { for already_prepared in [false, true] { @@ -2377,7 +2713,8 @@ async fn saved_checkpoint_recovery_completes_without_proving_failed_cleanup_ende let dispatched = f.dispatch_input(); let worker = dispatched.attempt.worker.as_ref().unwrap(); - // Inject a recoverable cleanup commit failure before checkpoint publication. + // A deterministic SQL constraint failure needs explicit fault correction, + // not autonomous transient-storage probing. Preserve the original child. // Pin/unpin changes Worker authority and deliberately makes the original child stale; // that negative contract is covered by the canonical-cleanup staleness tests. f.api @@ -2388,6 +2725,14 @@ async fn saved_checkpoint_recovery_completes_without_proving_failed_cleanup_ende BEFORE INSERT ON worker_registry_projection_removals BEGIN SELECT RAISE(ABORT, 'injected checkpoint cleanup failure'); END;", )?; + let error = conn + .execute( + "INSERT INTO worker_registry_projection_removals DEFAULT VALUES", + [], + ) + .unwrap_err(); + assert!(matches!(error, rusqlite::Error::SqliteFailure(error, _) + if error.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_TRIGGER)); Ok(()) }) .unwrap(); @@ -2443,6 +2788,19 @@ async fn saved_checkpoint_recovery_completes_without_proving_failed_cleanup_ende f.execution.take_inputs().is_empty(), "saved checkpoint must not cause model rerun" ); + tokio::time::timeout( + std::time::Duration::from_secs(15), + f.api.backend_job_recovery.cleanup_stopped.notified(), + ) + .await + .expect("deterministic commit failure did not stop with its retained cause"); + assert!( + !f.api + .backend_job_recovery + .cleanup_owner_ids() + .contains(&blocked.attempt_id), + "permanent constraint failure must not become an endless transient probe" + ); f.api .config_store .with_conn(|conn| {