From 7a7589ae6e64720ed57748605222c349151a6c8d Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 5 Nov 2026 06:14:08 +0900 Subject: [PATCH] fix(job): reconcile retained owners and diagnose cleanup admission failures Own all startup recovery pages fairly, consume original scoped removal history, and distinguish ordinary Worker failure from retained cleanup obligations. Persist exact-owner Host admission preflight diagnostics without inventing execution fences, and retain original Stop identity across uncertain acknowledgements. Checkpoint only: Runtime normal package passes; standalone selected target closure passes after fixture corrections; Server owner closure is 43/44 with a retained real-Journal convergence failure. Not approved or integrated. --- crates/standalone/src/job_store.rs | 56 +++ crates/standalone/src/jobs.rs | 48 +- crates/standalone/src/jobs/recovery_tests.rs | 12 +- crates/standalone/src/jobs/tests.rs | 55 +- crates/standalone/src/subjektiv.rs | 216 ++++++-- crates/standalone/src/subjektiv/tests.rs | 307 +++++++++++- crates/standalone/tests/jobs.rs | 53 +- crates/standalone/tests/subjektiv.rs | 54 ++ crates/subjektiv/src/job.rs | 87 ++++ crates/subjektiv/src/job/tests.rs | 75 +++ .../tests/session_material_tests.rs | 12 +- ...ubject_job_removal_purge_recovery_tests.rs | 70 ++- .../tests/subject_job_removal_purge_tests.rs | 9 + crates/worker-runtime/src/retention.rs | 147 ++++++ .../src/retention/runtime_v5_tests.rs | 1 + crates/worker-runtime/src/runtime.rs | 54 +- .../src/runtime/subject_delete.rs | 133 +++-- .../src/subject_delete_backend_tests.rs | 310 +++++++++++- crates/worker-runtime/src/worker_backend.rs | 15 +- .../src/backend_job_recovery_store.rs | 403 +++++++++++++++ crates/workspace-server/src/server.rs | 29 +- .../src/server/backend_job_recovery.rs | 76 ++- .../src/server/tests/backend_job_tests.rs | 474 +++++++++++++++++- .../subjektiv_journal_connected_tests.rs | 49 +- crates/workspace-server/src/store.rs | 86 +++- 25 files changed, 2628 insertions(+), 203 deletions(-) diff --git a/crates/standalone/src/job_store.rs b/crates/standalone/src/job_store.rs index 744858028..6eae68e32 100644 --- a/crates/standalone/src/job_store.rs +++ b/crates/standalone/src/job_store.rs @@ -366,6 +366,13 @@ impl JobStore { .transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot = require_snapshot(&tx, job_id)?; check_attempt(&snapshot, attempt_id)?; + if snapshot.version_stopped || snapshot.input_disposition != JobInputDisposition::Available + { + return Err(JobStoreError::InvalidTransition( + "stopped or retired input forbids domain reconciliation", + )); + } + snapshot.load_executable_request()?; if snapshot.domain_grant.is_none() || snapshot.attempt.state != JobAttemptState::Failed || !matches!(snapshot.state, JobState::Failed | JobState::Cancelled) @@ -1263,6 +1270,54 @@ mod tests { assert!(reopened.retry("first").is_err()); } + #[test] + fn granted_reconciliation_rechecks_retired_input_in_its_own_transaction() { + let (dir, mut store) = fixture(); + for (id, erase) in [("live-input", false), ("erased-input", true)] { + let reserved = store + .reserve_granted(request(id), Some(json!({"domain":"consolidation"}))) + .unwrap(); + store.dispatch(id, &reserved.attempt.attempt_id, 8).unwrap(); + store + .finish( + id, + &reserved.attempt.attempt_id, + JobAttemptState::Failed, + Some("provider failure"), + ) + .unwrap(); + record_runner_cleanup(&mut store, id); + let observed = store.get(id).unwrap(); + assert!(observed.executions_released()); + if erase { + let mut other = JobStore::open(&dir.path().join("jobs.sqlite3")).unwrap(); + other + .reclaim_input( + id, + &observed.input_digest, + JobInputRetirementReason::ExplicitErasure, + ) + .unwrap(); + } + let result = store.reconcile_granted_attempt( + id, + &observed.attempt.attempt_id, + "inspected-original-effects", + ); + assert_eq!(result.is_ok(), !erase); + let receipts: i64 = store + .conn + .query_row( + "SELECT COUNT(*) FROM job_domain_reconciliations WHERE job_id=?1", + [id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(receipts, i64::from(!erase)); + assert_eq!(store.get(id).unwrap().state, JobState::Failed); + } + } + #[test] fn schema_three_cutover_preserves_grant_history_and_execution_fence() { let dir = tempfile::tempdir().unwrap(); @@ -2612,6 +2667,7 @@ mod tests { Some("known failure"), ) .unwrap(); + assert!(!failed.executions_released()); record_runner_cleanup(&mut store, "atomic"); let failed = store.get("atomic").unwrap(); store diff --git a/crates/standalone/src/jobs.rs b/crates/standalone/src/jobs.rs index caba25d51..80c20e417 100644 --- a/crates/standalone/src/jobs.rs +++ b/crates/standalone/src/jobs.rs @@ -85,6 +85,10 @@ pub(crate) trait JobDomainProvider: Send + Sync { ) -> Result; } +type OwnedExecutionReport = Arc< + Mutex>>, +>; + struct Execution { cancel: watch::Sender, complete: watch::Receiver>>, @@ -95,11 +99,7 @@ struct Execution { recovery_issue: Arc>>, task: tokio::task::JoinHandle>, // Retain typed Worker failures and their cleanup resources until the owner can resolve them. - _owned: Arc< - Mutex< - Option>, - >, - >, + _owned: OwnedExecutionReport, } struct Executions { closed: bool, @@ -133,9 +133,30 @@ pub struct StandaloneJobs { // Exact restore admissions and typed causes remain Host-owned, even if a // dispatcher fails. These are identities, never copies of immutable bodies. result_recovery: Arc>>>>>, + // Process-local typed reports survive successful wait/capacity release. + // They are diagnostics, never durable shutdown or replay authority. + diagnostics: Arc>>, } impl StandaloneJobs { + /// Inspect a retained ordinary Worker failure for one exact attempt in this Host. + /// The report survives `wait` and shutdown, but is not restart or cleanup evidence. + pub fn with_execution_diagnostic( + &self, + job_id: &str, + attempt_id: &str, + inspect: impl FnOnce(&worker::job::OwnedInternalJobExecution) -> T, + ) -> Option { + let report = self + .diagnostics + .lock() + .expect("Job diagnostics poisoned") + .get(&(job_id.to_owned(), attempt_id.to_owned())) + .cloned()?; + let owned = report.lock().expect("Job owned execution poisoned"); + owned.as_ref()?.as_ref().ok().map(inspect) + } + /// Excludes request/start/result acknowledgement while a Subject Host checks its final /// saved obligations and releases ownership. Accepted results alone are insufficient: /// every spawned execution must have actually finished successful cleanup. @@ -178,6 +199,7 @@ impl StandaloneJobs { storage_recovered: Arc::default(), capacity_changed: Arc::default(), result_recovery: Arc::default(), + diagnostics: Arc::default(), }) } @@ -635,6 +657,7 @@ impl StandaloneJobs { let task_issue = recovery_issue.clone(); let storage_recovered = self.storage_recovered.clone(); let capacity_changed = self.capacity_changed.clone(); + let diagnostics = self.diagnostics.clone(); let task = tokio::spawn(async move { let (sender_tx, mut sender_rx) = tokio::sync::oneshot::channel(); let run = prepared.run_owned(attempt_id.clone(), sink, move |sender| { @@ -708,6 +731,19 @@ impl StandaloneJobs { }, ) .await; + if outcome.is_ok() + && retained + .lock() + .expect("Job owned execution poisoned") + .as_ref() + .and_then(|result| result.as_ref().ok()) + .is_some_and(|execution| execution.worker_failure().is_some()) + { + diagnostics + .lock() + .expect("Job diagnostics poisoned") + .insert((id.clone(), attempt_id.clone()), retained.clone()); + } complete_tx.send_replace(Some( outcome.as_ref().map(|_| ()).map_err(ToString::to_string), )); @@ -1018,7 +1054,7 @@ fn settle_owned_execution( .map_err(|error| JobServiceError::Cleanup(error.to_string()))?; if !domain_fenced || !execution.feature_shutdown_complete() - || execution.worker_failure().is_some() + || execution.requires_worker_cleanup_reconciliation() { return Err(JobServiceError::Cleanup(format!( "owned Worker shutdown unconfirmed ({} Feature shutdown failures); typed execution retained", diff --git a/crates/standalone/src/jobs/recovery_tests.rs b/crates/standalone/src/jobs/recovery_tests.rs index 37892ec61..627b4728a 100644 --- a/crates/standalone/src/jobs/recovery_tests.rs +++ b/crates/standalone/src/jobs/recovery_tests.rs @@ -190,6 +190,13 @@ async fn restored_results_continue_after_wait_error_without_forging_cleanup() { let jobs = StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().into()).unwrap(); let failed = jobs.request(request("wait-error")).unwrap(); let later = jobs.request(request("later-result")).unwrap(); + let db = rusqlite::Connection::open(temp.path().join("jobs.sqlite3")).unwrap(); + db.execute_batch( + "CREATE TRIGGER reject_cleanup_receipt BEFORE UPDATE OF cleanup_receipt ON job_execution_ownership + WHEN NEW.job_id='wait-error' AND NEW.cleanup_receipt IS NOT NULL + BEGIN SELECT RAISE(FAIL, 'scripted cleanup receipt rejection'); END;", + ) + .unwrap(); let ids = jobs.reserved_result_attempts().unwrap(); let release = Arc::new(Notify::new()); let _release_on_failure = ReleaseShutdownOnDrop(release.clone()); @@ -206,7 +213,8 @@ async fn restored_results_continue_after_wait_error_without_forging_cleanup() { let owner_jobs = jobs.clone(); let owner = tokio::spawn(async move { owner_jobs.resume_reserved_result_jobs(ids).await }); // Wait for the later model to enter, then observe its real cleanup. The first - // owner still cannot finish; it is held before a known provider failure. + // owner is held before provider return and its exact cleanup receipt write + // will be rejected independently of that ordinary model failure. await_model(&model).await; assert_eq!( await_job(&jobs, "later-result").await.state, @@ -233,7 +241,7 @@ async fn restored_results_continue_after_wait_error_without_forging_cleanup() { assert!( jobs.result_recovery_diagnostic("wait-error") .unwrap() - .contains("cleanup") + .contains("scripted cleanup receipt rejection") ); assert_eq!(model.requests.lock().unwrap().len(), 2); assert_eq!(bad_model.requests.lock().unwrap().len(), 2); diff --git a/crates/standalone/src/jobs/tests.rs b/crates/standalone/src/jobs/tests.rs index 351b7072b..4f2ef085e 100644 --- a/crates/standalone/src/jobs/tests.rs +++ b/crates/standalone/src/jobs/tests.rs @@ -366,7 +366,7 @@ async fn accepted_result_with_feature_shutdown_failure_retains_typed_owner_and_b } #[tokio::test] -async fn accepted_result_with_worker_failure_does_not_confirm_execution_cleanup() { +async fn accepted_result_with_ordinary_worker_failure_confirms_independent_cleanup() { let temp = tempfile::tempdir().unwrap(); let jobs = StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().into()).unwrap(); jobs.request(request("post-result-error")).unwrap(); @@ -380,31 +380,52 @@ async fn accepted_result_with_worker_failure_does_not_confirm_execution_cleanup( jobs.start_with_model_client("post-result-error", model) .await .unwrap(); - assert!(matches!( - tokio::time::timeout(Duration::from_secs(5), jobs.wait("post-result-error")) - .await - .unwrap(), - Err(JobServiceError::Cleanup(_)) - )); - let accepted = jobs.get("post-result-error").unwrap(); - assert_eq!(accepted.state, JobState::Completed); - assert_eq!(accepted.result, Some(json!({"valid":true}))); - assert!(!accepted.executions_released()); + let mut completion = jobs.executions.lock().unwrap().jobs["post-result-error"] + .complete + .clone(); + completion.wait_for(Option::is_some).await.unwrap(); { let executions = jobs.executions.lock().unwrap(); let owned = executions.jobs["post-result-error"]._owned.lock().unwrap(); let owned = owned.as_ref().unwrap().as_ref().unwrap(); assert!(owned.feature_shutdown_complete()); assert!(owned.worker_failure().is_some()); + assert!(!owned.requires_worker_cleanup_reconciliation()); assert!(owned.result.is_ok()); } + let accepted = tokio::time::timeout(Duration::from_secs(5), jobs.wait("post-result-error")) + .await + .unwrap() + .unwrap(); + assert_eq!(accepted.state, JobState::Completed); + assert_eq!(accepted.result, Some(json!({"valid":true}))); + assert!(accepted.executions_released()); + assert_eq!( + jobs.with_execution_diagnostic( + "post-result-error", + &accepted.attempt.attempt_id, + |report| { + assert_eq!(report.run_id.to_string(), accepted.executions[0].run_id); + assert!(report.feature_shutdown_complete()); + assert!(!report.requires_worker_cleanup_reconciliation()); + assert!(report.result.is_ok()); + matches!( + report.worker_failure(), + Some(worker::WorkerError::Engine(_)) + ) + } + ), + Some(true), + "typed ordinary failure survives execution-slot release" + ); + assert!( + jobs.with_execution_diagnostic("post-result-error", "foreign-attempt", |_| ()) + .is_none() + ); assert!(jobs.retry("post-result-error").is_err()); - assert!(jobs.reserve_quiescent_closure().is_err()); - assert!(matches!( - jobs.shutdown().await, - Err(JobServiceError::Cleanup(_)) - )); - assert!(jobs.store.lock().unwrap().is_some()); + assert!(jobs.reserve_quiescent_closure().is_ok()); + jobs.shutdown().await.unwrap(); + assert!(jobs.store.lock().unwrap().is_none()); } #[tokio::test] diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index daa1bdc49..6d90e2651 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -276,6 +276,7 @@ impl StandaloneSubjects { writer, self_weak: weak.clone(), journal_dispatch: Mutex::new(false), + consolidation_recovery: Mutex::new(std::collections::HashMap::new()), #[cfg(test)] registration_before_attribution: Mutex::new(None), language: self.memory_language.clone(), @@ -372,10 +373,46 @@ impl Drop for SubjectConnection { } } +#[derive(Debug, thiserror::Error)] +enum ConsolidationRecoveryError { + #[error(transparent)] + Host(#[from] SubjektivHostError), + #[error(transparent)] + Service(#[from] crate::jobs::JobServiceError), + #[error(transparent)] + Domain(#[from] subjektiv::api::OperationError), +} + +impl ConsolidationRecoveryError { + fn retryable_read_or_storage(&self) -> bool { + use crate::jobs::{JobServiceError, JobStoreError}; + matches!( + self, + Self::Service(JobServiceError::Store( + JobStoreError::Sqlite(_) | JobStoreError::Storage(_) | JobStoreError::Io(_) + )) | Self::Domain(subjektiv::api::OperationError::Storage(_)) + // The Host's writer/Subject liveness reads wrap storage errors in + // Unavailable. Recheck them within this same finite read-only budget; + // their failure NEVER authorizes a receipt, release, or execution. + | Self::Host(SubjektivHostError::Unavailable(_)) + ) + } +} + +struct ConsolidationRecoveryOwner { + running: bool, + passes: usize, + error: ConsolidationRecoveryError, +} + pub(crate) struct LocalSubjectHost { writer: subjektiv::journal::JournalWriterBinding, self_weak: std::sync::Weak, journal_dispatch: Mutex, + // Exact-attempt, process-local bounded recovery owners/diagnostics. The Job + // grant and its unchanged input remain the durable obligation, not this map. + consolidation_recovery: + Mutex>, #[cfg(test)] registration_before_attribution: Mutex>>, catalog: StandaloneSubjects, @@ -767,10 +804,7 @@ impl LocalSubjectHost { schedule_job(jobs.clone(), job.request.job_id, self.self_weak.clone()); } Ok(false) => {} // Missing receipts never authorize replay of effects. - Err(error) => eprintln!( - "standalone consolidation Job {} reconciliation blocked: {error}", - job.request.job_id - ), + Err(error) => self.own_consolidation_recovery(jobs.clone(), job, error), } } } @@ -779,24 +813,24 @@ impl LocalSubjectHost { self.recover_journals(); } - /// Reconstructs a result from domain receipts, never by reentering the finalizer. - /// Acceptance does not assert prior owner shutdown or discharge cleanup ownership. + /// Prefer a complete saved result. Otherwise close only this failed attempt's + /// domain reservation, retaining unresolved work and the original failure. + /// Neither path reruns the finalizer or proves prior execution shutdown. fn recover_consolidation_job( &self, jobs: &crate::jobs::StandaloneJobs, observed: &crate::jobs::JobSnapshot, - ) -> Result { - let service_error = |error: crate::jobs::JobServiceError| { - SubjektivHostError::Unavailable(error.to_string()) - }; + ) -> Result { self.ensure_live()?; - let current = jobs.get(&observed.request.job_id).map_err(service_error)?; + let current = jobs.get(&observed.request.job_id)?; if current.attempt.attempt_id != observed.attempt.attempt_id || current.input_digest != observed.input_digest + || current.domain_grant != observed.domain_grant { return Err(SubjektivHostError::InvalidArgument( - "consolidation recovery attempt/input changed".into(), - )); + "consolidation recovery attempt/input/grant changed".into(), + ) + .into()); } if current.version_stopped || current.input_disposition != crate::jobs::JobInputDisposition::Available @@ -804,37 +838,156 @@ impl LocalSubjectHost { { return Err(SubjektivHostError::InvalidArgument( "stopped or retired consolidation cannot be recovered".into(), - )); + ) + .into()); } current .load_executable_request() - .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?; + .map_err(crate::jobs::JobServiceError::from)?; let grant = self.consolidation_grant(¤t)?; - let Some(result) = subjektiv::job::lookup_result( + if let Some(result) = subjektiv::job::lookup_result( &self.catalog.store, &grant.binding.subject_id, &grant.candidate_ids, ¤t.request.job_id, ¤t.attempt.attempt_id, - ) - .map_err(operation_error)? - else { - return Ok(false); - }; - self.ensure_live()?; - let settled = jobs - .accept_reconciled(&job::JobResultSubmission { + )? { + self.ensure_live()?; + let settled = jobs.accept_reconciled(&job::JobResultSubmission { job_id: current.request.job_id, attempt_id: current.attempt.attempt_id, input_digest: current.input_digest, result, - }) - .map_err(service_error)?; - jobs.acknowledge(&settled.request.job_id) - .map_err(service_error)?; + })?; + jobs.acknowledge(&settled.request.job_id)?; + return Ok(true); + } + if current.state != crate::jobs::JobState::Failed + || current.attempt.state != crate::jobs::JobAttemptState::Failed + || !current.executions_released() + { + return Ok(false); + } + // This inspection reads original candidate/disposition and preparation + // records transactionally. Missing preparation is valid; stale preparation, + // missing candidates, retired Subjects and malformed receipts are not. + // Do NOT turn unresolved IDs into dispositions or fail a newer/ready surface. + let _retained = subjektiv::job::inspect_abandoned_attempt( + &self.catalog.store, + &grant.binding.subject_id, + &grant.candidate_ids, + ¤t.request.job_id, + ¤t.attempt.attempt_id, + )?; + // Stable reference to the original immutable grant/input and failed + // attempt, not a hash of mutable pending-work projections. Later explicit + // work can resolve retained candidates without invalidating exact replay. + // The existing Job receipt is the closure; no parallel domain ledger. + let reference = job::fingerprint(&( + "consolidation-abandoned-v1", + &grant.binding, + ¤t.request.job_id, + ¤t.attempt.attempt_id, + ¤t.input_digest, + )) + .map_err(|error| SubjektivHostError::InvalidArgument(error.to_string()))?; + self.ensure_live()?; + // Recheck the exact Host/grant before persistence. The service/store also + // fence the current attempt and independently require ALL executions released. + let checked = jobs.get(¤t.request.job_id)?; + if checked != current { + return Err(SubjektivHostError::InvalidArgument( + "consolidation changed during abandoned-attempt inspection".into(), + ) + .into()); + } + self.consolidation_grant(&checked)?; + jobs.reconcile_granted_attempt( + ¤t.request.job_id, + ¤t.attempt.attempt_id, + &format!("consolidation-abandoned-v1:{reference}"), + )?; Ok(true) } + /// Keep a finite recovery owner for read/storage failure of this exact + /// snapshot. No startup-only repair, disconnect polling, new attempt or model + /// call. Exhaustion retains the typed cause and original durable obligation; + /// a fresh Host can reconcile it, never infer success from budget exhaustion. + fn own_consolidation_recovery( + &self, + jobs: crate::jobs::StandaloneJobs, + observed: crate::jobs::JobSnapshot, + error: ConsolidationRecoveryError, + ) { + eprintln!( + "standalone consolidation Job {} reconciliation blocked: {error}", + observed.request.job_id + ); + if !error.retryable_read_or_storage() { + return; + } + let key = ( + observed.request.job_id.clone(), + observed.attempt.attempt_id.clone(), + observed.input_digest.clone(), + ); + let mut owners = self + .consolidation_recovery + .lock() + .expect("consolidation recovery poisoned"); + if let Some(owner) = owners.get(&key) { + if !owner.running { + eprintln!( + "standalone consolidation Job {} retains exhausted reconciliation after {} passes: {}; next action: resolve cause and restore the same Host binding", + key.0, owner.passes, owner.error + ); + } + return; // Repeated disconnect cannot refill a running or exhausted budget. + } + let Some(host) = self.self_weak.upgrade() else { + return; + }; + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + return; + }; + owners.insert( + key.clone(), + ConsolidationRecoveryOwner { + running: true, + passes: 1, + error, + }, + ); + runtime.spawn(async move { + const PASSES: usize = 8; + for pass in 2..=PASSES { + tokio::time::sleep(std::time::Duration::from_millis(100 << (pass - 2).min(5))).await; + let result = host.recover_consolidation_job(&jobs, &observed); + let mut owners = host.consolidation_recovery.lock().expect("consolidation recovery poisoned"); + match result { + Ok(_) => { + owners.remove(&key); + drop(owners); + host.wake_journals(); + return; + } + Err(error) => { + let retry = error.retryable_read_or_storage() && pass < PASSES; + let owner = owners.get_mut(&key).expect("retained consolidation owner"); + owner.passes = pass; + owner.running = retry; + owner.error = error; + if !retry { + eprintln!("standalone consolidation Job {} reconciliation stopped after {} passes: {}; original Failed/input/receipts retained; next action: resolve storage/authority failure and restore the same Host binding", key.0, owner.passes, owner.error); + return; + } + } + } + } + }); + } + fn consolidation_grant( &self, snapshot: &crate::jobs::JobSnapshot, @@ -1596,6 +1749,13 @@ fn schedule_job( Ok(result) if result.state == crate::jobs::JobState::Completed => { let _ = jobs.acknowledge(&id); } + Ok(result) if result.state == crate::jobs::JobState::Failed => { + if let Some(host) = host.upgrade() { + if let Err(error) = host.recover_consolidation_job(&jobs, &result) { + host.own_consolidation_recovery(jobs.clone(), result, error); + } + } + } Ok(_) => {} Err(error) => eprintln!( "standalone Subject Job {id} cleanup/acknowledgement is unconfirmed: {error}" diff --git a/crates/standalone/src/subjektiv/tests.rs b/crates/standalone/src/subjektiv/tests.rs index 420b36aee..9a1280930 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -22,39 +22,45 @@ impl Drop for ConsolidationRecoveryFixture { impl ConsolidationRecoveryFixture { fn new() -> Self { + Self::with_candidates(&["candidate"]) + } + + fn with_candidates(candidate_ids: &[&str]) -> Self { let temp = tempfile::tempdir().unwrap(); let catalog = StandaloneSubjects::open(temp.path().join("state")).unwrap(); let subject = catalog.create("recovery-owner", None).unwrap(); let body = catalog .connect(&subject.id, protocol::WorkerId::now_v7()) .unwrap(); - body.host - .catalog - .store - .stage_candidate(subjektiv::SubjectStagingRecord::attach( - &subject.id, - memory::extract::StagingRecord::from_candidate( - "candidate", - "dedup", - memory::schema::SourceRef { - segment_id: "segment".into(), - range: [1, 1], - }, - memory::extract::ExtractedCandidate { - kind: memory::extract::CandidateKind::Lesson, - claim: "claim".into(), - why_useful: "reason".into(), - staleness: None, - evidence_ids: vec![], - }, - vec![], - vec![], - ), - )) - .unwrap(); + for candidate_id in candidate_ids { + body.host + .catalog + .store + .stage_candidate(subjektiv::SubjectStagingRecord::attach( + &subject.id, + memory::extract::StagingRecord::from_candidate( + *candidate_id, + *candidate_id, + memory::schema::SourceRef { + segment_id: "segment".into(), + range: [1, 1], + }, + memory::extract::ExtractedCandidate { + kind: memory::extract::CandidateKind::Lesson, + claim: "claim".into(), + why_useful: "reason".into(), + staleness: None, + evidence_ids: vec![], + }, + vec![], + vec![], + ), + )) + .unwrap(); + } let grant = LocalConsolidationGrant { binding: body.binding.clone(), - candidate_ids: vec!["candidate".into()], + candidate_ids: candidate_ids.iter().map(|id| (*id).to_owned()).collect(), }; let request = job::JobRequest { job_id: "consolidation-recovery".into(), @@ -93,6 +99,19 @@ impl ConsolidationRecoveryFixture { } } + fn domain_closure(&self) -> Option { + use rusqlite::OptionalExtension; + rusqlite::Connection::open(&self.body.jobs_path) + .unwrap() + .query_row( + "SELECT domain_receipt FROM job_domain_reconciliations WHERE job_id=?1 AND attempt_id=?2", + rusqlite::params![self.saved.request.job_id, self.saved.attempt.attempt_id], + |row| row.get(0), + ) + .optional() + .unwrap() + } + fn confirm_execution_joined(&mut self) { // The execution boundary is substituted in this domain-consumer fixture; // use its exact owner/run to commit the receipt through the real store. @@ -148,6 +167,244 @@ impl ConsolidationRecoveryFixture { } } +#[tokio::test] +async fn failed_consolidation_closes_only_attempt_and_preserves_unresolved_input_and_applied_memory() + { + for partial_apply in [false, true] { + let mut fixture = + ConsolidationRecoveryFixture::with_candidates(&["candidate", "unresolved"]); + let subject_id = fixture.body.binding.subject_id.clone(); + let subject = &subject_id; + let store = &fixture.body.host.catalog.store; + let applied = partial_apply.then(|| { + store + .decide_candidate( + subject, + subjektiv::CandidateDecisionRequest { + request_id: "applied-before-failure".into(), + candidate_id: "candidate".into(), + reason: "retained applied lesson".into(), + decision: subjektiv::CandidateDecision::Apply { + target: subjektiv::MemoryChangeTarget::Create, + draft: subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Lesson, + "claim", + "body", + "reason", + "change", + ), + }, + }, + ) + .unwrap() + }); + // This fixture substitutes only positive execution termination, not the + // domain lookup. No model is configured; any accidental execution fails. + fixture.confirm_execution_joined(); + let before = fixture.jobs.get(&fixture.saved.request.job_id).unwrap(); + let memory = serde_json::to_value( + fixture + .body + .host + .catalog + .store + .list_memories(subject) + .unwrap(), + ) + .unwrap(); + fixture.body.host.recover_jobs(); + let receipt = fixture + .domain_closure() + .expect("failed attempt closure persisted"); + assert!(receipt.starts_with("consolidation-abandoned-v1:")); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert_eq!(before.state, crate::jobs::JobState::Failed); + assert!(before.result.is_none()); + assert!(fixture.jobs.unreleased_granted().unwrap().is_empty()); + let retained = subjektiv::job::inspect_abandoned_attempt( + &fixture.body.host.catalog.store, + subject, + &["candidate".into(), "unresolved".into()], + &before.request.job_id, + &before.attempt.attempt_id, + ) + .unwrap(); + assert!( + retained.prepared_generation_id.is_none(), + "no new generation" + ); + assert_eq!( + retained.unresolved_candidate_ids, + if partial_apply { + vec!["unresolved"] + } else { + vec!["candidate", "unresolved"] + } + ); + if let Some(applied) = applied { + assert_eq!(retained.disposition_ids, vec![applied.resolution.id]); + let saved = fixture + .body + .host + .catalog + .store + .staging_resolution(subject, "candidate") + .unwrap() + .unwrap(); + assert_eq!(saved.affected_memory, applied.resolution.affected_memory); + } else { + assert!(retained.disposition_ids.is_empty()); + } + assert_eq!( + serde_json::to_value( + fixture + .body + .host + .catalog + .store + .list_memories(subject) + .unwrap() + ) + .unwrap(), + memory + ); + // Stable exact replay; a reference does not mean all candidate work is done. + assert!( + fixture + .body + .host + .recover_consolidation_job(&fixture.jobs, &before) + .unwrap() + ); + assert_eq!(fixture.domain_closure(), Some(receipt)); + let reopened = crate::job_store::JobStore::open(&fixture.body.jobs_path).unwrap(); + assert_eq!(reopened.get(&before.request.job_id).unwrap(), before); + } +} + +#[tokio::test] +async fn failed_consolidation_retains_unreleased_stale_inactive_migrated_and_erased_controls() { + for case in [ + "unreleased", + "stale", + "inactive", + "migrated", + "erased", + "changed_grant", + ] { + let mut fixture = ConsolidationRecoveryFixture::new(); + if case != "unreleased" { + fixture.confirm_execution_joined(); + } + if case == "stale" { + fixture.generation(&fixture.saved.attempt.attempt_id); + fixture + .body + .host + .catalog + .store + .create_memory( + &fixture.body.binding.subject_id, + subjektiv::MemoryDraft::active( + memory::extract::CandidateKind::Lesson, + "new", + "body", + "reason", + "change", + ), + ) + .unwrap(); + } + if case == "inactive" { + fixture.body.host.live.store(false, Ordering::Release); + } + if case == "migrated" { + rusqlite::Connection::open(&fixture.body.jobs_path) + .unwrap() + .execute( + "INSERT INTO job_version_stops(job_id) VALUES (?1)", + [&fixture.saved.request.job_id], + ) + .unwrap(); + } + if case == "erased" { + fixture + .jobs + .reclaim_input( + &fixture.saved.request.job_id, + &fixture.saved.input_digest, + crate::jobs::JobInputRetirementReason::ExplicitErasure, + ) + .unwrap(); + } + let mut observed = fixture.saved.clone(); + if case == "changed_grant" { + observed.domain_grant = None; + } + let before = fixture.jobs.get(&fixture.saved.request.job_id).unwrap(); + let outcome = fixture + .body + .host + .recover_consolidation_job(&fixture.jobs, &observed); + assert!(!matches!(outcome, Ok(true)), "{case}"); + assert!(fixture.domain_closure().is_none(), "{case}"); + assert_eq!( + fixture.jobs.get(&fixture.saved.request.job_id).unwrap(), + before, + "{case}" + ); + } +} + +#[tokio::test] +async fn failed_consolidation_receipt_storage_error_has_autonomous_exact_attempt_recovery() { + let mut fixture = ConsolidationRecoveryFixture::new(); + fixture.confirm_execution_joined(); + let before = fixture.saved.clone(); + let connection = rusqlite::Connection::open(&fixture.body.jobs_path).unwrap(); + // Fail only the closure write; the read-only domain check and all independent + // cleanup proofs are real. A storage error is never a successful closure. + connection + .execute_batch( + "CREATE TRIGGER fail_domain_closure BEFORE INSERT ON job_domain_reconciliations + BEGIN SELECT RAISE(FAIL, 'fixture domain receipt storage failure'); END;", + ) + .unwrap(); + let error = fixture + .body + .host + .recover_consolidation_job(&fixture.jobs, &before) + .unwrap_err(); + assert!(error.retryable_read_or_storage()); + assert!(fixture.domain_closure().is_none()); + fixture + .body + .host + .own_consolidation_recovery(fixture.jobs.clone(), before.clone(), error); + connection + .execute_batch("DROP TRIGGER fail_domain_closure;") + .unwrap(); + // No disconnect, recover_jobs call, restart, new Job, model or finalizer. + tokio::time::timeout(std::time::Duration::from_secs(5), async { + while fixture.domain_closure().is_none() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + assert!(fixture.jobs.unreleased_granted().unwrap().is_empty()); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert!( + fixture + .body + .host + .consolidation_recovery + .lock() + .unwrap() + .is_empty() + ); +} + #[tokio::test] async fn consolidation_recovery_accepts_same_attempt_receipts_without_claiming_owner_drain() { for availability in ["ready", "failed"] { diff --git a/crates/standalone/tests/jobs.rs b/crates/standalone/tests/jobs.rs index 5f549275f..548be660d 100644 --- a/crates/standalone/tests/jobs.rs +++ b/crates/standalone/tests/jobs.rs @@ -580,7 +580,7 @@ async fn process_loss_fails_same_attempt_without_claiming_execution_cleanup() { } #[tokio::test] -async fn model_failure_retains_unconfirmed_execution_and_forbids_retry() { +async fn ordinary_model_failure_preserves_cause_and_confirms_independent_cleanup() { let temp = tempfile::tempdir().unwrap(); let (host, _) = host(&temp).await; let jobs = host.jobs(); @@ -589,14 +589,14 @@ async fn model_failure_retains_unconfirmed_execution_and_forbids_retry() { fail: true, ..ScriptedClient::new(Vec::new()) }; - jobs.start_with_model_client("model-failure", client) + jobs.start_with_model_client("model-failure", client.clone()) .await .unwrap(); - assert!(matches!( - jobs.wait("model-failure").await, - Err(JobServiceError::Cleanup(_)) - )); - let failure = jobs.get("model-failure").unwrap(); + let failure = tokio::time::timeout(Duration::from_secs(10), jobs.wait("model-failure")) + .await + .unwrap() + .unwrap(); + assert_eq!(jobs.get("model-failure").unwrap(), failure); assert_eq!(failure.state, JobState::Failed); assert!(failure.result.is_none()); assert!( @@ -607,11 +607,40 @@ async fn model_failure_retains_unconfirmed_execution_and_forbids_retry() { .unwrap() .contains("scripted model failure") ); - assert!(!failure.executions_released()); - assert!(jobs.retry("model-failure").is_err()); - assert_eq!(jobs.get("model-failure").unwrap(), failure); + assert!(failure.executions_released()); + assert_eq!(failure.executions.len(), 1); + let execution = &failure.executions[0]; + assert!(execution.writes_revoked); assert_eq!( - host.shutdown().await, - Err(standalone::StandaloneShutdownError::JobCleanup) + execution.cleanup_receipt.as_deref(), + Some(format!("owned-run:{}", execution.run_id).as_str()) ); + assert_eq!(client.requests().len(), 1, "no automatic model retry"); + assert_eq!( + jobs.with_execution_diagnostic("model-failure", &failure.attempt.attempt_id, |report| { + assert_eq!(report.run_id.to_string(), execution.run_id); + assert!(report.feature_shutdown_complete()); + assert!(report.worker_failure().is_some()); + assert!(report.result.is_err()); + !report.requires_worker_cleanup_reconciliation() + }), + Some(true) + ); + assert_eq!(jobs.request(request("model-failure")).unwrap(), failure); + let retried = jobs.retry("model-failure").unwrap(); + assert_eq!(retried.attempt.number, 2); + assert_ne!(retried.attempt.attempt_id, failure.attempt.attempt_id); + assert_eq!(retried.executions, failure.executions); + let recovery = success(); + jobs.start_with_model_client("model-failure", recovery.clone()) + .await + .unwrap(); + let completed = jobs.wait("model-failure").await.unwrap(); + assert_eq!(completed.state, JobState::Completed); + assert_eq!(completed.executions.len(), 2); + assert_eq!(completed.executions[0], failure.executions[0]); + assert!(completed.executions_released()); + assert_eq!(recovery.requests().len(), 2); + assert_eq!(client.requests().len(), 1); + host.shutdown().await.unwrap(); } diff --git a/crates/standalone/tests/subjektiv.rs b/crates/standalone/tests/subjektiv.rs index 28edfb5be..5bc909f3b 100644 --- a/crates/standalone/tests/subjektiv.rs +++ b/crates/standalone/tests/subjektiv.rs @@ -1383,7 +1383,61 @@ async fn transport_loss_after_candidate_apply_preserves_memory_and_later_job_onl failed.result.is_none(), "partial apply is not an accepted Job result" ); + assert!( + failed.executions_released(), + "real cleanup is independent of failure" + ); + let requests_for_failed_job = || { + losing + .requests() + .iter() + .filter(|request| { + request.items.iter().filter_map(Item::as_text).any(|text| { + text.lines() + .any(|line| line == format!("job_id: {first_id}")) + }) + }) + .count() + }; + let failed_job_requests = requests_for_failed_job(); + let job_db = rusqlite::Connection::open( + state + .join("subjektiv/jobs") + .join(&subject.id) + .join("jobs.sqlite3"), + ) + .unwrap(); + // Wait for the production post-terminal consumer, NOT disconnect retries or + // a manual startup/recovery call. This row is explicitly non-success and the + // original Failed snapshot/cause/input remain unchanged. + tokio::time::timeout(Duration::from_secs(5), async { + loop { + use rusqlite::OptionalExtension; + let receipt: Option = job_db.query_row( + "SELECT domain_receipt FROM job_domain_reconciliations WHERE job_id=?1 AND attempt_id=?2", + rusqlite::params![first_id, failed.attempt.attempt_id], |row| row.get(0), + ).optional().unwrap(); + if let Some(receipt) = receipt { + assert!(receipt.starts_with("consolidation-abandoned-v1:")); + break; + } + tokio::task::yield_now().await; + } + }).await.expect("failed domain reconciliation is owned without disconnect/startup"); + assert_eq!(host.jobs().get(&first_id).unwrap(), failed); + assert_eq!( + requests_for_failed_job(), + failed_job_requests, + "no extra model request" + ); disconnect_recorded(&mut host).await; + // Disconnect closes this service; compare the durable Failed snapshot after + // the next Host reopens the same Subject ledger below. + assert_eq!( + requests_for_failed_job(), + failed_job_requests, + "no replay during handoff" + ); host.shutdown().await.unwrap(); let original = inspect_store(&state, |store| { assert!( diff --git a/crates/subjektiv/src/job.rs b/crates/subjektiv/src/job.rs index f586b43cb..9559aa09a 100644 --- a/crates/subjektiv/src/job.rs +++ b/crates/subjektiv/src/job.rs @@ -215,6 +215,93 @@ pub fn lookup_result( with_dispositions(store, subject_id, batch, &result).map(Some) } +/// A read-only view of work retained by an abandoned attempt, NOT a successful +/// consolidation result. Receipt IDs point to immutable dispositions; unresolved +/// IDs remain in the original input/staging store for later explicit work. +#[derive(Debug, PartialEq, Eq)] +pub struct AbandonedAttempt { + pub disposition_ids: Vec, + pub unresolved_candidate_ids: Vec, + pub prepared_generation_id: Option, +} + +/// Inspect the original batch and preparation without deciding candidates, +/// preparing/failing a surface, or calling a model/finalizer. The Host must first +/// prove the exact failed attempt's execution release and current grant/input. +/// No preparation is a valid pre-effects cut; an existing but stale preparation +/// is not. A ready surface (possibly shared by another preparation) is untouched. +pub fn inspect_abandoned_attempt( + store: &SubjektivStore, + subject_id: &str, + batch: &[String], + job_id: &str, + attempt_id: &str, +) -> OperationResult { + use crate::SubjektivError; + use rusqlite::{OptionalExtension, params}; + + if batch.len() > CANDIDATE_BATCH_LIMIT + || batch + .iter() + .collect::>() + .len() + != batch.len() + { + return Err(OperationError::InvalidInput( + "invalid consolidation batch".into(), + )); + } + store.database.try_transaction(|tx| { + crate::require_active_subject(tx, subject_id)?; + crate::validate_label("Job id", job_id)?; + crate::validate_label("attempt id", attempt_id)?; + let mut retained = AbandonedAttempt { + disposition_ids: Vec::new(), + unresolved_candidate_ids: Vec::new(), + prepared_generation_id: None, + }; + for id in batch { + let raw: Option = tx.query_row( + "SELECT record_json FROM staging_records WHERE subject_id=?1 AND candidate_id=?2", + params![subject_id, id], |row| row.get(0) + ).optional()?; + let raw_candidate = raw.ok_or_else(|| SubjektivError::CandidateNotFound(id.clone()))?; + let candidate = crate::parse_staging(store.workspace_id(), &raw_candidate)?; + if candidate.subject_id != subject_id || candidate.id != *id { + return Err(SubjektivError::InvalidRecord("consolidation candidate identity mismatch".into())); + } + let resolution: Option = tx.query_row( + "SELECT resolution_json FROM staging_resolutions WHERE subject_id=?1 AND candidate_id=?2", + params![subject_id, id], |row| row.get(0) + ).optional()?; + if let Some(raw) = resolution { + let saved = crate::parse_resolution(&raw)?; + if saved.subject_id != subject_id || saved.candidate_id != *id + || !crate::equivalent_staging_payload(&serde_json::to_string(&saved.candidate)?, &raw_candidate)? + { + return Err(SubjektivError::InvalidRecord("consolidation disposition identity mismatch".into())); + } + retained.disposition_ids.push(saved.id); + } else { + retained.unresolved_candidate_ids.push(id.clone()); + } + } + let fingerprint = crate::memory_fingerprint_in_connection(tx, subject_id)?; + retained.prepared_generation_id = tx.query_row( + "SELECT generation_id FROM surface_generation_runs + WHERE subject_id=?1 AND job_id=?2 AND attempt_id=?3 AND memory_fingerprint=?4 + ORDER BY generation_id LIMIT 1", + params![subject_id, job_id, attempt_id, fingerprint], |row| row.get(0) + ).optional()?; + if retained.prepared_generation_id.is_none() && tx.prepare( + "SELECT 1 FROM surface_generation_runs WHERE subject_id=?1 AND job_id=?2 AND attempt_id=?3" + )?.exists(params![subject_id, job_id, attempt_id])? { + return Err(SubjektivError::InvalidRecord("abandoned consolidation generation input is stale".into())); + } + Ok(retained) + }).map_err(crate::api::subjektiv_store_error) +} + /// Adds a compact projection of stored dispositions to a validated result. /// Reasons, change references, and provenance stay in the receipt store, not duplicated in bounded Job results. /// Neither reapplies candidates nor creates a new ledger. diff --git a/crates/subjektiv/src/job/tests.rs b/crates/subjektiv/src/job/tests.rs index 99ab21e88..3f5118e12 100644 --- a/crates/subjektiv/src/job/tests.rs +++ b/crates/subjektiv/src/job/tests.rs @@ -6,6 +6,81 @@ use crate::{ use memory::extract::{CandidateKind, ExtractedCandidate, StagingRecord}; use memory::schema::SourceRef; +#[test] +fn abandoned_attempt_inspection_preserves_original_preparation_and_rejects_stale_input() { + let temp = tempfile::tempdir().unwrap(); + let manager = feature_storage::FeatureStorage::new(temp.path()); + let scope = manager.scope("abandoned-consolidation").unwrap(); + let registration = SubjektivStore::register(&scope).unwrap(); + let store = SubjektivStore::open(&scope, ®istration).unwrap(); + let subject = store + .create_subject(SubjectRole::new("companion").unwrap()) + .unwrap(); + assert_eq!( + inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").unwrap(), + AbandonedAttempt { + disposition_ids: vec![], + unresolved_candidate_ids: vec![], + prepared_generation_id: None, + } + ); + let generation = store + .prepare_job_surface_generation(&subject.id, Some(("job", "attempt"))) + .unwrap(); + let surface = store.resident_surface(&subject.id).unwrap(); + for _ in 0..2 { + let abandoned = + inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").unwrap(); + assert_eq!( + abandoned.prepared_generation_id.as_deref(), + Some(generation.id.as_str()) + ); + // Inspection never fabricates a terminal surface failure or finalizes it. + assert!( + lookup_result(&store, &subject.id, &[], "job", "attempt") + .unwrap() + .is_none() + ); + assert_eq!(store.resident_surface(&subject.id).unwrap(), surface); + } + let count: i64 = store + .database + .try_with_connection(|connection| { + Ok::<_, crate::SubjektivError>(connection.query_row( + "SELECT COUNT(*) FROM surface_generation_runs", + [], + |row| row.get(0), + )?) + }) + .unwrap(); + assert_eq!(count, 1, "no new generation on abandoned reconciliation"); + let ready = store + .publish_surface_generation(&subject.id, &generation.id, vec![]) + .unwrap(); + inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").unwrap(); + assert_eq!( + store.resident_surface(&subject.id).unwrap().snapshot, + Some(ready) + ); + assert!( + inspect_abandoned_attempt(&store, &subject.id, &["missing".into()], "job", "attempt") + .is_err() + ); + store + .create_memory( + &subject.id, + crate::MemoryDraft::active(CandidateKind::Lesson, "new", "body", "reason", "change"), + ) + .unwrap(); + assert!(inspect_abandoned_attempt(&store, &subject.id, &[], "job", "attempt").is_err()); + store + .begin_subject_deletion(&subject.id, "delete-abandoned-subject") + .unwrap(); + assert!( + inspect_abandoned_attempt(&store, &subject.id, &[], "fresh-job", "fresh-attempt").is_err() + ); +} + #[test] fn prepared_dirty_surface_waits_for_terminal_receipt_and_rejects_stale_input() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs index c0bebd3f9..493d4e707 100644 --- a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs @@ -58,6 +58,10 @@ impl Fixture { let mut request = task_request("canonical Session fixture"); request.config_bundle = None; request.memory_settings.as_mut().unwrap().workspace_id = scope.workspace_id.clone(); + request.workspace_api = Some(crate::catalog::WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://backend.test".into(), + }); let worker = runtime .create_worker_scoped(&scope, request) .unwrap() @@ -1265,11 +1269,11 @@ async fn subject_delete_worker_signed_http_requires_delete_scope_without_inventi assert_eq!(response.status(), StatusCode::BAD_REQUEST); let error: RuntimeHttpErrorResponse = read_json(response).await; assert_eq!(error.error.code, "invalid_request"); + let diagnostic = error.error.message.to_ascii_lowercase(); assert!( - error - .error - .message - .contains("Missing original confirmed Subject shutdown") + diagnostic.contains("original") && diagnostic.contains("shutdown"), + "expected rejection for absent original shutdown proof, got: {}", + error.error.message ); assert!(f.runtime.worker_detail(&f.worker).is_ok()); assert!( diff --git a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs index 6d5b0e01f..ccebd1129 100644 --- a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_recovery_tests.rs @@ -66,28 +66,43 @@ fn subject_job_removal_purge_partial_physical_failure_keeps_fence_and_retry_eras fn subject_job_removal_purge_unconfirmed_fence_never_invents_removed_execution_proof() { let f = Fixture::new(); let request = f.removal_purge_request(SessionDisposition::Archive); - f.complete_removal(&request); + let original_result = f.complete_removal(&request); let copies = f.copies(); let receipt_path = f.receipt_path(); - let receipt = std::fs::read(&receipt_path).unwrap(); + let mut legacy_receipt: serde_json::Value = + serde_json::from_slice(&std::fs::read(&receipt_path).unwrap()).unwrap(); let Fixture { temp, runtime, scope, store, + session, .. } = f; drop(store); drop(runtime); - // Retained incomplete-proof fixture: deliberately remove the confirmation - // after removal. Correct current ordering requires that confirmation before - // removal; this is not evidence that such a crash window can occur now. - // The old removal result cannot manufacture the missing Subject proof. + // Model proofless retained history, not a current crash window. Removing + // only the Subject response is insufficient: the original canonical Stop + // now survives removal and legitimately permits historical reconciliation. + // Remove that proof too, leaving the completed removal request/result intact. + assert!( + legacy_receipt + .as_object_mut() + .unwrap() + .remove("shutdown") + .is_some_and(|shutdown| shutdown.is_object()) + ); + let receipt = serde_json::to_vec(&legacy_receipt).unwrap(); + std::fs::write(&receipt_path, &receipt).unwrap(); let snapshot_path = temp.path().join("runtime.json"); let mut snapshot: serde_json::Value = serde_json::from_slice(&std::fs::read(&snapshot_path).unwrap()).unwrap(); - snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()]["response"] = - serde_json::Value::Null; + let fence = + &mut snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()]; + assert!(fence["response"].is_object()); + assert!(fence.get("canonical_removal").is_none()); + fence["response"] = serde_json::Value::Null; + let unconfirmed_fence = fence.clone(); std::fs::write(&snapshot_path, serde_json::to_vec(&snapshot).unwrap()).unwrap(); let runtime = Runtime::with_fs_store_and_execution_backend( crate::FsRuntimeStoreOptions::new(temp.path()).with_runtime_id("runtime-test"), @@ -102,16 +117,47 @@ fn subject_job_removal_purge_unconfirmed_fence_never_invents_removed_execution_p ); assert_eq!(tree(&temp.path().join("archives")), copies); assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt); + let removal = serde_json::from_value(serde_json::to_value(&request.removal).unwrap()).unwrap(); + assert_eq!( + runtime.execute_worker_retention(&removal).unwrap(), + original_result + ); + assert_eq!(std::fs::read(&receipt_path).unwrap(), receipt); + let error = runtime + .subject_delete_worker_scoped(&scope, request.deletion.clone()) + .unwrap_err(); + let diagnostic = error.to_string().to_ascii_lowercase(); assert!( - runtime - .subject_delete_worker_scoped(&scope, request.deletion.clone()) - .is_err() + diagnostic.contains("original") && diagnostic.contains("shutdown"), + "expected rejection for absent original shutdown proof, got: {error}" ); assert!( runtime - .purge_subject_job_removal_scoped(&scope, request) + .purge_subject_job_removal_scoped(&scope, request.clone()) .is_err() ); assert_eq!(tree(&temp.path().join("archives")), copies); assert_eq!(std::fs::read(receipt_path).unwrap(), receipt); + let snapshot: serde_json::Value = + serde_json::from_slice(&std::fs::read(snapshot_path).unwrap()).unwrap(); + assert_eq!( + snapshot["subject_delete_worker_fences"][request.deletion.worker_id.to_string()], + unconfirmed_fence, + "proofless history must not acquire a Subject response or a fabricated Stop child" + ); + assert!( + runtime + .worker_detail(&WorkerRef::new(request.deletion.worker_id)) + .is_err() + ); + assert!( + !temp + .path() + .join("workers") + .join(request.deletion.worker_id.to_string()) + .exists() + ); + runtime_session_retention(temp.path()) + .ensure_live(session) + .unwrap(); } diff --git a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs index e90337ce1..be7275892 100644 --- a/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/subject_job_removal_purge_tests.rs @@ -80,6 +80,10 @@ impl Fixture { let mut request = task_request("independent Worker"); request.config_bundle = None; request.memory_settings.as_mut().unwrap().workspace_id = self.scope.workspace_id.clone(); + request.workspace_api = Some(crate::catalog::WorkspaceApiRef { + workspace_id: self.scope.workspace_id.clone(), + base_url: "https://backend.test".into(), + }); self.runtime .create_worker_scoped(&self.scope, request) .unwrap() @@ -145,6 +149,11 @@ fn subject_job_removal_purge_erases_copies_preserves_receipt_and_prior_success_a let prior = tree(&f.temp.path().join("archives/workers/prior-archive")); assert!(!prior.is_empty()); let unrelated = f.independent_worker(); + // Compare settled retained material, not an active Worker whose cold + // automatic restore legitimately changes worker.json independently of purge. + f.runtime + .stop_worker_scoped(&f.scope, &unrelated, None) + .unwrap(); let unrelated_root = f .temp .path() diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index 29296d3f4..d5a80985a 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -764,6 +764,112 @@ impl FsWorkerRetentionProvider { )) } + /// Read, never create or finish, the original canonical removal. The Worker + /// operation and Runtime owner locks are held by the caller; the retention + /// lock also excludes receipt publication while inspecting history. + pub(crate) fn subject_shutdown_removal( + &self, + deletion: &runtime_api::SubjectDeleteWorkerRequest, + runtime_id: &str, + expected: Option<(&WorkerRetentionExecutionRequest, &str)>, + ) -> Result< + ( + WorkerRetentionExecutionRequest, + crate::execution::WorkerExecutionTermination, + String, + ), + RuntimeError, + > { + let invalid = || { + RuntimeError::InvalidRequest( + "Missing or mismatched original canonical removal shutdown".into(), + ) + }; + let lock_path = self.runtime_root.join(RETENTION_LOCK); + let lock = OpenOptions::new() + .create(true) + .read(true) + .write(true) + .open(&lock_path) + .map_err(|source| RuntimeError::StoreIo { + operation: "lock Subject removal history", + path: lock_path.clone(), + source, + })?; + lock.lock().map_err(|source| RuntimeError::StoreIo { + operation: "lock Subject removal history", + path: lock_path, + source, + })?; + let operations = self.runtime_root.join("retention").join("operations"); + let entries = fs::read_dir(&operations).map_err(|_| invalid())?; + let mut found = None; + for entry in entries { + let path = entry.map_err(|_| invalid())?.path(); + if path.extension().and_then(|value| value.to_str()) != Some("json") { + continue; + } + let receipt = read_operation_receipt(&path)?.ok_or_else(&invalid)?; + if receipt.request.worker_id != deletion.worker_id { + continue; + } + // More than one removal of this ID is successor history, not proof + // that the original scope covers whichever generation is now gone. + if found.is_some() { + return Err(invalid()); + } + let request = &receipt.request; + // Identity of the immutable child receipt, not another copy of its + // shutdown facts in the parent. Detect changed proof on cold retry. + let receipt_sha256 = + Sha256::digest(serde_json::to_vec(&receipt).map_err(|_| invalid())?) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + validate_request(request)?; + validate_receipt_request(&receipt, request)?; + let shutdown = receipt.shutdown.as_ref().ok_or_else(&invalid)?; + let result = &receipt.result; + let worker = crate::identity::WorkerRef::new(deletion.worker_id); + if path != self.operation_path(&request.operation_id)? + || request.workspace_id != deletion.workspace_id + || request.source_runtime_id != runtime_id + || expected.is_some_and(|(saved, digest)| { + saved != request || digest != receipt_sha256.as_str() + }) + || shutdown.create_fingerprint != deletion.create_fingerprint + || !shutdown + .termination + .confirms_scope_shutdown(&worker, shutdown.stop_operation_id) + || result.worker_id != request.worker_id + || result.session_disposition != request.session_disposition + || result.diagnostics_disposition != request.diagnostics_disposition + || result.diagnostics_retained + != (request.diagnostics_disposition == DiagnosticsDisposition::Retain) + || !result.source_removed + { + return Err(invalid()); + } + match (&result.archive, request.session_disposition) { + (None, SessionDisposition::Purge) => {} + (Some(archive), SessionDisposition::Archive) + if archive.workspace_id == request.workspace_id + && archive.source_runtime_id == request.source_runtime_id + && archive.source_worker_id == request.worker_id + && archive.operation_id == request.operation_id + && archive.input_fingerprint == request.input_fingerprint + && Some(&archive.archive_id) == request.archive_id.as_ref() => {} + _ => return Err(invalid()), + } + found = Some(( + receipt.request.clone(), + shutdown.termination.clone(), + receipt_sha256, + )); + } + found.ok_or_else(invalid) + } + pub(crate) fn recover_after_source_removal( &self, request: &WorkerRetentionExecutionRequest, @@ -908,6 +1014,18 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { fn execute( &self, request: &WorkerRetentionExecutionRequest, + ) -> Result { + self.execute_with_shutdown(request, None) + } +} + +impl FsWorkerRetentionProvider { + // Only Runtime's canonical removal caller supplies this evidence, before + // deleting its source aggregate. Replays never upgrade a legacy receipt. + pub(crate) fn execute_with_shutdown( + &self, + request: &WorkerRetentionExecutionRequest, + shutdown: Option, ) -> Result { validate_request(request)?; fs::create_dir_all(&self.runtime_root).map_err(|source| RuntimeError::StoreIo { @@ -938,6 +1056,14 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { if receipt.result.source_removed { return Ok(receipt.result.clone()); } + if receipt.shutdown.is_some() + && self.worker_dir(request.worker_id).is_dir() + && receipt.shutdown != shutdown + { + return Err(RuntimeError::InvalidRequest( + "Canonical removal original shutdown changed".into(), + )); + } } let worker_dir = self.worker_dir(request.worker_id); @@ -1026,6 +1152,12 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider { schema_version: OPERATION_SCHEMA_VERSION, request: request.clone(), result: result.clone(), + // The original operation owns this fact. Do not retag a pending + // removal with a later Stop, or enrich an older proofless receipt. + shutdown: match pending { + Some(receipt) => receipt.shutdown, + None => shutdown, + }, }; // Pending receipt makes the delete/final-receipt crash window // recoverable without treating an uncommitted archive as completion. @@ -1079,11 +1211,25 @@ struct CanonicalSessionManifest { session_id: String, } +/// Evidence destroyed with the canonical aggregate, retained only by its +/// existing removal operation. This is not a new Stop or an execution ledger. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct RetainedWorkerShutdown { + pub(crate) create_fingerprint: String, + pub(crate) stop_operation_id: crate::execution::WorkerLifecycleOperationId, + pub(crate) termination: crate::execution::WorkerExecutionTermination, +} + #[derive(Serialize, Deserialize)] struct RetentionOperationReceipt { schema_version: u32, request: WorkerRetentionExecutionRequest, result: WorkerRetentionExecutionResult, + // Additive read compatibility: an old receipt remains valid retention + // history, but cannot establish previously unrecorded Subject shutdown. + #[serde(default, skip_serializing_if = "Option::is_none")] + shutdown: Option, } fn read_operation_receipt(path: &Path) -> Result, RuntimeError> { @@ -2320,6 +2466,7 @@ mod tests { schema_version: OPERATION_SCHEMA_VERSION, request: request.clone(), result: completed.clone(), + shutdown: None, }, ); fs::remove_dir_all(provider.worker_dir(worker)).unwrap(); diff --git a/crates/worker-runtime/src/retention/runtime_v5_tests.rs b/crates/worker-runtime/src/retention/runtime_v5_tests.rs index 064caecc9..168011156 100644 --- a/crates/worker-runtime/src/retention/runtime_v5_tests.rs +++ b/crates/worker-runtime/src/retention/runtime_v5_tests.rs @@ -218,6 +218,7 @@ fn completed_v4_archive( schema_version: OPERATION_SCHEMA_VERSION, request: request.clone(), result: completed.clone(), + shutdown: None, }, ); fs::remove_dir_all(provider.worker_dir(worker)).unwrap(); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 3b0cc0498..d172db44a 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -6224,7 +6224,59 @@ impl Runtime { "Worker retention cannot remove pending {operation} reconciliation" ))); } - let result = provider.execute(request)?; + // Preserve only the original canonical Stop, not stopped status, map + // absence or a backend receipt for a different execution. The aggregate + // is about to disappear; the existing retention operation owns the + // historical copy. Missing legacy evidence does not block removal. + let shutdown = worker.request.as_ref().and_then(|create| { + worker + .restore_guard + .execution_termination + .as_ref() + .and_then(|termination| { + termination + .confirms_scope_shutdown(&worker.worker_ref, termination.stop_operation_id) + .then(|| crate::retention::RetainedWorkerShutdown { + create_fingerprint: create.create_fingerprint.clone(), + stop_operation_id: termination.stop_operation_id, + termination: termination.clone(), + }) + }) + }); + let expected = worker.persisted_record(); + let backend = state.execution_backend.clone(); + drop(state); + if let Some(saved) = shutdown.as_ref() { + let current = backend + .ok_or_else(|| { + RuntimeError::InvalidRequest( + "Canonical removal shutdown has no execution owner".into(), + ) + })? + .reconcile_termination_receipt(&expected.worker_ref, &saved.termination) + .map_err(RuntimeError::InvalidRequest)?; + if current.as_ref() != Some(&saved.termination) { + return Err(RuntimeError::InvalidRequest( + "Canonical removal shutdown no longer matches execution owner".into(), + )); + } + } + let mut state = self.lock()?; + state.ensure_running()?; + if shutdown.is_some() { + let store = state.fs_store().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Canonical removal shutdown requires filesystem storage".into(), + ) + })?; + let (_, matches) = store.reconcile_worker_record(&expected)?; + if !matches { + return Err(RuntimeError::InvalidRequest( + "Canonical removal shutdown source changed".into(), + )); + } + } + let result = provider.execute_with_shutdown(request, shutdown)?; state.workers.remove(&request.worker_id); state.persist_runtime_snapshot()?; Ok(result) diff --git a/crates/worker-runtime/src/runtime/subject_delete.rs b/crates/worker-runtime/src/runtime/subject_delete.rs index a14df1e3e..115f069ac 100644 --- a/crates/worker-runtime/src/runtime/subject_delete.rs +++ b/crates/worker-runtime/src/runtime/subject_delete.rs @@ -1,49 +1,98 @@ use super::*; +struct SubjectOriginalAbsence { + // None only for the existing confirmation-before-removal path. + removal_shutdown: Option, +} + impl RuntimeState { - // Absence is usable only after this exact Subject operation confirmed the - // original scope and cleanup. A new missing catalog ID is never proof. + // Absence needs original scope shutdown evidence, independently of current + // resource cleanup. A new missing catalog ID is never proof. fn confirmed_subject_delete_originals_absent( - &self, + &mut self, request: &runtime_api::SubjectDeleteWorkerRequest, - ) -> Result { + ) -> Result, RuntimeError> { if self.workers.contains_key(&request.worker_id) { - return Ok(false); + return Ok(None); } if self.unreconciled_worker_writes.contains(&request.worker_id) { return Err(RuntimeError::InvalidRequest( "Subject Worker write remains uncertain".into(), )); } - self.subject_delete_worker_fences - .get(&request.worker_id) - .filter(|fence| fence.request == *request) - .and_then(|fence| fence.response.as_ref()) - .filter(|saved| { - saved.workspace_id == request.workspace_id - && saved.worker_id == request.worker_id - && saved.create_fingerprint == request.create_fingerprint - && saved.operation_id == request.operation_id - }) - .ok_or_else(|| { - RuntimeError::InvalidRequest("Missing original confirmed Subject shutdown".into()) - })?; #[cfg(feature = "fs-store")] - if self - .fs_store() - .ok_or_else(|| { - RuntimeError::InvalidRequest("Subject original absence requires fs storage".into()) - })? - .worker_originals_absent(request.worker_id)? { - return Ok(true); + let fence = self + .subject_delete_worker_fences + .get(&request.worker_id) + .filter(|fence| fence.request == *request) + .ok_or_else(|| { + RuntimeError::InvalidRequest("Missing exact Subject fence".into()) + })?; + let store = self.fs_store().ok_or_else(|| { + RuntimeError::InvalidRequest("Subject original absence requires fs storage".into()) + })?; + if !store.worker_originals_absent(request.worker_id)? { + return Err(RuntimeError::InvalidRequest( + "Subject Worker original material remains".into(), + )); + } + // Preserve confirmation-before-removal replay. A fence confirmed + // through retention history must instead reread its original child + // every time; its response is not a substitute for that receipt. + if fence.canonical_removal.is_none() + && fence.response.as_ref().is_some_and(|saved| { + saved.workspace_id == request.workspace_id + && saved.worker_id == request.worker_id + && saved.create_fingerprint == request.create_fingerprint + && saved.operation_id == request.operation_id + }) + { + return Ok(Some(SubjectOriginalAbsence { + removal_shutdown: None, + })); + } + let runtime_id = self.runtime_identity.as_deref().ok_or_else(|| { + RuntimeError::InvalidRequest("Subject removal Runtime identity missing".into()) + })?; + let (removal, termination, receipt_sha256) = + FsWorkerRetentionProvider::new(store.runtime_dir()).subject_shutdown_removal( + request, + runtime_id, + fence + .canonical_removal + .as_ref() + .map(|child| (&child.request, child.receipt_sha256.as_str())), + )?; + self.subject_delete_worker_fences + .get_mut(&request.worker_id) + .expect("exact Subject fence held under Worker lock") + .canonical_removal = Some(SubjectCanonicalRemoval { + request: removal, + receipt_sha256, + }); + // Bind the original child before external resource cleanup. Retry + // also synchronizes a prior uncertain snapshot acknowledgement. + self.persist_runtime_snapshot()?; + return Ok(Some(SubjectOriginalAbsence { + removal_shutdown: Some(termination), + })); } + #[cfg(not(feature = "fs-store"))] Err(RuntimeError::InvalidRequest( "Subject Worker original material remains or cannot be inspected".into(), )) } } +#[cfg(feature = "fs-store")] +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct SubjectCanonicalRemoval { + request: WorkerRetentionExecutionRequest, + receipt_sha256: String, +} + /// Permanent exclusion survives WorkerRemoval. A receipt never substitutes for a /// fresh execution reconciliation on exact retry, including after Runtime reopen. #[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] @@ -51,6 +100,10 @@ impl RuntimeState { pub(crate) struct SubjectDeleteWorkerFence { pub(crate) request: runtime_api::SubjectDeleteWorkerRequest, pub(crate) response: Option, + // Child identity only; shutdown remains authoritative in the removal receipt. + #[cfg(feature = "fs-store")] + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(crate) canonical_removal: Option, } impl Runtime { @@ -218,6 +271,8 @@ impl Runtime { SubjectDeleteWorkerFence { request: request.clone(), response: None, + #[cfg(feature = "fs-store")] + canonical_removal: None, }, ); } @@ -233,17 +288,31 @@ impl Runtime { backend .fence_subject_delete_worker(&request) .map_err(RuntimeError::InvalidRequest)?; - if self - .lock()? - .confirmed_subject_delete_originals_absent(&request)? - { - // The durable confirmed fence owns the historical shutdown proof; - // this separately checks current adapter resources after removal. + let absent = { + let mut state = self.lock()?; + state.confirmed_subject_delete_originals_absent(&request)? + }; + if let Some(absent) = absent { + if let Some(saved) = absent.removal_shutdown.as_ref() { + let current = backend + .reconcile_termination_receipt(&worker_ref, saved) + .map_err(RuntimeError::InvalidRequest)?; + if current.as_ref() != Some(saved) { + return Err(RuntimeError::InvalidRequest( + "Original removal Stop no longer matches execution owner".into(), + )); + } + } + // The durable confirmed fence or original removal owns historical + // shutdown proof; separately check current adapter/factory resources. backend .confirm_subject_delete_worker_absent(&request) .map_err(RuntimeError::InvalidRequest)?; let mut state = self.lock()?; - if !state.confirmed_subject_delete_originals_absent(&request)? { + if state + .confirmed_subject_delete_originals_absent(&request)? + .is_none() + { return Err(RuntimeError::InvalidRequest( "Subject originals changed during absence reconciliation".into(), )); diff --git a/crates/worker-runtime/src/subject_delete_backend_tests.rs b/crates/worker-runtime/src/subject_delete_backend_tests.rs index dd2fd3f0c..3b8f17680 100644 --- a/crates/worker-runtime/src/subject_delete_backend_tests.rs +++ b/crates/worker-runtime/src/subject_delete_backend_tests.rs @@ -343,7 +343,9 @@ fn subject_delete_without_original_owner_retains_child_identity_but_never_issues #[test] #[cfg(feature = "fs-store")] -fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() { +fn subject_delete_terminal_release_history_consumes_prior_canonical_stop() { + use runtime_api::{SubjectReleaseAction as Action, SubjectReleaseState as State}; + for (action, terminal) in [(Action::Cancel, State::Cancelled), (Action::Disconnect, State::Disconnected)] { let temp = tempfile::tempdir_in(".").unwrap(); let root = fs::canonicalize(temp.path()).unwrap(); let backend = Arc::new( @@ -397,20 +399,23 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() { .stop_worker_scoped(&scope, &worker.worker_ref, None) .is_err() ); - release.action = runtime_api::SubjectReleaseAction::Cancel; + release.action = action; assert_eq!( runtime .subject_release_scoped(&scope, release.clone()) .unwrap() .state, - runtime_api::SubjectReleaseState::Cancelled + terminal ); { let reservations = backend.subject_reservations.lock().unwrap(); let saved = &reservations[&(worker.worker_ref.clone(), release.operation_id.clone())]; - assert_eq!(saved.state, runtime_api::SubjectReleaseState::Cancelled); + assert_eq!(saved.state, terminal); assert!(saved.guard.is_none()); } + if terminal == State::Disconnected { + assert!(runtime.send_input(&worker.worker_ref, WorkerInput::user("must not resume a disconnected body")).is_err()); + } runtime .stop_worker_scoped(&scope, &worker.worker_ref, None) .unwrap(); @@ -424,7 +429,7 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() { create_fingerprint: create.create_fingerprint, operation_id: "delete-prior-stop".into(), }; - // Retained non-Cancelled state still blocks canonical consumption even if + // Retained non-terminal state still blocks canonical consumption even if // its local guard is unavailable; only the terminal no-guard row is harmless. let key = (worker.worker_ref.clone(), release.operation_id); backend @@ -458,7 +463,7 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() { .unwrap() .get_mut(&key) .unwrap() - .state = runtime_api::SubjectReleaseState::Cancelled; + .state = terminal; runtime .subject_delete_worker_scoped(&scope, deletion.clone()) .unwrap(); @@ -478,6 +483,7 @@ fn subject_delete_cancelled_release_history_consumes_prior_canonical_stop() { assert!(backend.workers.lock().unwrap().is_empty()); assert!(backend.pending_spawns.lock().unwrap().is_empty()); assert!(backend.pending_restores.lock().unwrap().is_empty()); + } } // Only the extra factory-owned cleanup is faulted. Controller Stop, scope join, @@ -1217,3 +1223,295 @@ fn subject_delete_confirmed_removal_retries_warm_and_cold_without_new_stop() { assert!(cold.pending_restores.lock().unwrap().is_empty()); assert!(reopened.create_worker_scoped(&scope, create).is_err()); } + +#[test] +#[cfg(feature = "fs-store")] +#[serial_test::serial(worker_allocation)] +fn subject_delete_removal_first_retains_original_stop_and_rechecks_cold_history() { + let temp = tempfile::tempdir_in(".").unwrap(); + let root = fs::canonicalize(temp.path()).unwrap(); + let spawns = Arc::new(AtomicUsize::new(0)); + let cleanup_allowed = Arc::new(AtomicBool::new(true)); + let cleanup_calls = Arc::new(AtomicUsize::new(0)); + let new_backend = || { + Arc::new( + WorkerRuntimeExecutionBackend::new(SubjectCleanupFaultFactory { + inner: MockFactory { + client: MockClient::new(vec![]), + runtime_base: PathBuf::from(temp.path().file_name().unwrap()).join("c"), + cwd: root.clone(), + store_dir: root.join("sessions"), + worker_metadata_dir: root.join("metadata"), + observed_cwds: Arc::default(), + observed_workspace_clients: Arc::default(), + }, + spawns: spawns.clone(), + cleanup_allowed: cleanup_allowed.clone(), + cleanup_calls: cleanup_calls.clone(), + }) + .unwrap(), + ) + }; + let backend = new_backend(); + let options = crate::fs_store::FsRuntimeStoreOptions::new(root.join("catalog")) + .with_runtime_id("removal-first"); + let runtime = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.bind_runtime_identity("removal-first").unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let scope = crate::runtime::RuntimeWorkspaceScope::new("ws-test", "server-test"); + let mut create = create_request("ordinary removal before first Subject confirmation"); + create.workspace_api = Some(WorkspaceApiRef { + workspace_id: scope.workspace_id.clone(), + base_url: "https://workspace.invalid".into(), + }); + create.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: scope.workspace_id.clone(), + language: "English".into(), + }); + let worker = runtime + .create_worker_scoped(&scope, create.clone()) + .unwrap(); + let execution = backend.workers.lock().unwrap()[&worker.worker_ref].clone(); + runtime + .stop_worker_scoped(&scope, &worker.worker_ref, None) + .unwrap(); + let original = runtime + .execution_termination(&worker.worker_ref) + .unwrap() + .unwrap(); + assert_eq!(original.execution_id, execution.tasks.execution_id); + assert!(execution.handle.protocol_is_closed()); + assert!(execution.tasks.tasks.lock().unwrap().is_empty()); + let removal = crate::retention::WorkerRetentionExecutionRequest { + operation_id: "ordinary-removal".into(), + input_fingerprint: "ordinary-input".into(), + archive_id: None, + workspace_id: scope.workspace_id.clone(), + source_runtime_id: "removal-first".into(), + worker_id: create.worker_id, + source_created_at: "2026-11-04T00:00:00Z".into(), + removed_at: "2026-11-04T01:00:00Z".into(), + effective_profile: None, + retention_class: None, + policy_id: "ordinary-policy".into(), + session_disposition: crate::retention::SessionDisposition::Purge, + diagnostics_disposition: crate::retention::DiagnosticsDisposition::Purge, + }; + assert!( + runtime + .execute_worker_retention(&removal) + .unwrap() + .source_removed + ); + assert!(runtime.worker_detail(&worker.worker_ref).is_err()); + let receipt_path = options + .root + .join("retention/operations/ordinary-removal.json"); + let receipt_bytes = fs::read(&receipt_path).unwrap(); + let receipt: serde_json::Value = serde_json::from_slice(&receipt_bytes).unwrap(); + assert_eq!( + receipt["shutdown"]["termination"], + serde_json::to_value(&original).unwrap() + ); + let deletion = runtime_api::SubjectDeleteWorkerRequest { + workspace_id: scope.workspace_id.clone(), + worker_id: create.worker_id, + create_fingerprint: create.create_fingerprint.clone(), + operation_id: "first-Subject".into(), + }; + // Every failure happens after a real ordinary removal, before any successful + // Subject response; an empty adapter/catalog cannot provide the missing fact. + let mut bad = receipt.clone(); + bad.as_object_mut().unwrap().remove("shutdown"); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .execute_worker_retention(&removal) + .unwrap() + .source_removed + ); + assert!( + serde_json::from_slice::(&fs::read(&receipt_path).unwrap()) + .unwrap() + .get("shutdown") + .is_none(), + "legacy replay must not synthesize shutdown" + ); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["shutdown"]["stop_operation_id"] = + serde_json::to_value(crate::execution::WorkerLifecycleOperationId::new()).unwrap(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["shutdown"]["termination"]["worker_ref"] = + serde_json::to_value(WorkerRef::new(WorkerId::now_v7())).unwrap(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["request"]["source_runtime_id"] = "other-runtime".into(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["request"]["operation_id"] = "other-removal".into(); + bad["result"]["operation_id"] = "other-removal".into(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["request"]["workspace_id"] = "other-workspace".into(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["shutdown"]["create_fingerprint"] = "other-create".into(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["result"]["source_removed"] = false.into(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + fs::write(&receipt_path, &receipt_bytes).unwrap(); + cleanup_allowed.store(false, Ordering::SeqCst); + assert!( + runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + cleanup_allowed.store(true, Ordering::SeqCst); + let confirmed = runtime + .subject_delete_worker_scoped(&scope, deletion.clone()) + .unwrap(); + assert!(confirmed.originals_absent); + assert_eq!( + backend.completed_stops.lock().unwrap()[&worker.worker_ref], + original + ); + assert_eq!(fs::read(&receipt_path).unwrap(), receipt_bytes); + drop(execution); + drop(runtime); + drop(backend); + + let cold = new_backend(); + let reopened = + EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), cold.clone()) + .unwrap(); + reopened.bind_runtime_identity("removal-first").unwrap(); + // A saved Subject response cannot hide a lost or changed original child. + fs::remove_file(&receipt_path).unwrap(); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + let mut bad = receipt.clone(); + bad["request"]["input_fingerprint"] = "changed-input".into(); + bad["result"]["input_fingerprint"] = "changed-input".into(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + fs::write(&receipt_path, &receipt_bytes).unwrap(); + let mut bad = receipt.clone(); + bad["shutdown"]["termination"]["execution_id"] = + serde_json::to_value(uuid::Uuid::now_v7()).unwrap(); + fs::write(&receipt_path, serde_json::to_vec(&bad).unwrap()).unwrap(); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err(), + "cold retry must reread the original whole child receipt, not just request fields" + ); + fs::write(&receipt_path, &receipt_bytes).unwrap(); + // A later completed Controller is successor evidence even with empty maps. + let mut successor = original.clone(); + successor.execution_id = uuid::Uuid::now_v7(); + cold.completed_stops + .lock() + .unwrap() + .insert(worker.worker_ref.clone(), successor); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + cold.completed_stops.lock().unwrap().clear(); + let mut duplicate = receipt.clone(); + duplicate["request"]["operation_id"] = "later-removal".into(); + duplicate["result"]["operation_id"] = "later-removal".into(); + let duplicate_path = options.root.join("retention/operations/later-removal.json"); + fs::write(&duplicate_path, serde_json::to_vec(&duplicate).unwrap()).unwrap(); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + fs::remove_file(duplicate_path).unwrap(); + cleanup_allowed.store(false, Ordering::SeqCst); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + cleanup_allowed.store(true, Ordering::SeqCst); + let material = options + .root + .join("workers") + .join(create.worker_id.to_string()); + fs::create_dir_all(&material).unwrap(); + fs::write(material.join("residual"), "still owned original").unwrap(); + assert!( + reopened + .subject_delete_worker_scoped(&scope, deletion.clone()) + .is_err() + ); + fs::remove_dir_all(material).unwrap(); + assert_eq!( + reopened + .subject_delete_worker_scoped(&scope, deletion) + .unwrap(), + confirmed + ); + assert_eq!(spawns.load(Ordering::SeqCst), 1); + assert!(cleanup_calls.load(Ordering::SeqCst) > 0); + assert!(cold.completed_stops.lock().unwrap().is_empty()); + assert!(cold.workers.lock().unwrap().is_empty()); + assert!(cold.pending_spawns.lock().unwrap().is_empty()); + assert!(cold.pending_restores.lock().unwrap().is_empty()); + assert_eq!(fs::read(receipt_path).unwrap(), receipt_bytes); + assert!(reopened.create_worker_scoped(&scope, create).is_err()); +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index dc80488e1..d4f58f6c7 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -5442,7 +5442,11 @@ where .any(|((worker, _), reservation)| { worker == worker_ref && (reservation.guard.is_some() - || reservation.state != runtime_api::SubjectReleaseState::Cancelled) + || !matches!( + reservation.state, + runtime_api::SubjectReleaseState::Cancelled + | runtime_api::SubjectReleaseState::Disconnected + )) }) { return Ok(None); @@ -6056,7 +6060,11 @@ where .any(|((saved, _), reservation)| { *saved == worker && (reservation.guard.is_some() - || reservation.state != runtime_api::SubjectReleaseState::Cancelled) + || !matches!( + reservation.state, + runtime_api::SubjectReleaseState::Cancelled + | runtime_api::SubjectReleaseState::Disconnected + )) }) { return Err("Subject Worker still has an execution resource owner".into()); @@ -6213,6 +6221,9 @@ where .as_mut() .ok_or("Controller reservation missing")? .commit_disconnect()?; + // The Controller's permanent mutation fence is now committed; + // the temporary quiescence lease no longer owns a live handoff. + saved.guard.take(); saved.state = State::Disconnected; } (Action::Disconnect, State::Disconnected) | (Action::Cancel, State::Cancelled) => {} diff --git a/crates/workspace-server/src/backend_job_recovery_store.rs b/crates/workspace-server/src/backend_job_recovery_store.rs index d5f3b9ee6..75d24c9c3 100644 --- a/crates/workspace-server/src/backend_job_recovery_store.rs +++ b/crates/workspace-server/src/backend_job_recovery_store.rs @@ -189,6 +189,66 @@ pub(super) fn claim_cleanup_owned( }) } +/// A failed Host preflight is diagnostic evidence, never cleanup ownership or +/// termination proof. In particular, a competing unfenced Executing writer must +/// remain Executing; only its original admission resource can authorize recovery. +pub(super) fn record_cleanup_preflight_failure( + store: &SqliteWorkspaceStore, + workspace: &str, + expected_history: &[BackendJobAttemptRecord], + expected: &BackendJobAttemptRecord, + detail: &str, + now: &str, +) -> Result { + if detail.trim().is_empty() { + return Err(Error::InvalidInput( + "cleanup preflight requires a cause detail".into(), + )); + } + store.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + require_current_contract(&tx, workspace, &expected.job_id)?; + require_history(&tx, workspace, expected_history, expected)?; + let job = require_retained_owner(&tx, workspace, expected)?; + if job.current_attempt != expected.attempt + || job.state == BackendJobState::Pending + || !expected.state.terminal() + || expected.worker.is_none() + || expected + .runtime_run_id + .as_ref() + .is_none_or(|run| run.is_empty()) + || expected.worker_cleanup_completed_at.is_some() + || !matches!( + expected.worker_cleanup_state, + Some( + BackendJobWorkerCleanupState::Pending + | BackendJobWorkerCleanupState::Failed + | BackendJobWorkerCleanupState::Executing + ) + ) + { + return Err(Error::RepositoryConflict( + "Cleanup preflight diagnostic owner is no longer eligible".into(), + )); + } + let fenced: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM backend_job_execution_fences WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3)", + params![workspace, expected.job_id, expected.attempt_id], |row| row.get(0), + )?; + if fenced { + return Err(Error::RepositoryConflict( + "Cleanup execution fence already recorded; preflight cannot overwrite progress".into(), + )); + } + tx.execute("UPDATE backend_job_attempts SET worker_cleanup_state=CASE WHEN worker_cleanup_state='executing' THEN 'executing' ELSE 'failed' END,worker_cleanup_failure_category='backend_job_cleanup_host_admission_unconfirmed',worker_cleanup_failure_detail=?4,worker_cleanup_updated_at=?5 WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3", params![workspace, expected.job_id, expected.attempt_id, bounded_failure_detail(detail), now])?; + let result = read_backend_job_attempt(&tx, workspace, &expected.job_id, &expected.attempt_id)? + .ok_or_else(|| Error::Store("Cleanup diagnostic owner disappeared".into()))?; + tx.commit()?; + Ok(result) + }) +} + pub(super) fn finish_cleanup_owned( store: &SqliteWorkspaceStore, workspace: &str, @@ -1286,6 +1346,349 @@ mod tests { .unwrap() } + #[test] + fn cleanup_preflight_failure_full_cas_preserves_result_input_and_rolls_back_storage_error() { + let (dir, store, reserved) = fixture(); + let owner = bind_fixture_owner(&store, &reserved); + let completed = store + .reconcile_backend_job_domain_result( + "workspace", + &owner, + &BackendJobResultSubmission { + job_id: owner.job_id.clone(), + attempt_id: owner.attempt_id.clone(), + input_digest: owner.input_digest.clone(), + result: serde_json::json!({"saved": true}), + }, + NOW, + ) + .unwrap(); + let history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + let job = store + .get_backend_job("workspace", "accepted") + .unwrap() + .unwrap(); + let detail = + "Original Host admission unavailable; termination unconfirmed; retain result/input"; + let later = "2026-11-04T00:00:01Z"; + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &[], + &completed, + detail, + later, + ) + .is_err() + ); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "other-workspace", + &history, + &completed, + detail, + later, + ) + .is_err() + ); + let mut foreign = completed.clone(); + foreign.runtime_run_id = Some("successor-run".into()); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &[foreign.clone()], + &foreign, + detail, + later, + ) + .is_err() + ); + // Real SQLite failure at the diagnostic write must propagate, leaving + // the exact observation available for its existing retry/probe owner. + store.with_conn(|conn| { + conn.execute_batch("CREATE TEMP TRIGGER preflight_fault BEFORE UPDATE OF worker_cleanup_failure_category ON main.backend_job_attempts BEGIN SELECT RAISE(ABORT,'diagnostic storage fault'); END;")?; + Ok(()) + }).unwrap(); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &completed, + detail, + later, + ) + .is_err() + ); + assert_eq!( + store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(), + history + ); + store + .with_conn(|conn| { + conn.execute_batch("DROP TRIGGER preflight_fault")?; + Ok(()) + }) + .unwrap(); + let failed = store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &completed, + detail, + later, + ) + .unwrap(); + let mut expected = completed.clone(); + expected.worker_cleanup_state = Some(BackendJobWorkerCleanupState::Failed); + expected.worker_cleanup_failure_category = + Some("backend_job_cleanup_host_admission_unconfirmed".into()); + expected.worker_cleanup_failure_detail = Some(detail.into()); + expected.worker_cleanup_updated_at = Some(later.into()); + assert_eq!( + failed, expected, + "only cleanup diagnostic columns may change" + ); + assert_eq!( + store + .get_backend_job("workspace", "accepted") + .unwrap() + .unwrap(), + job + ); + assert!( + !store + .has_backend_job_execution_fence("workspace", &failed) + .unwrap() + ); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &completed, + "late stale failure", + later, + ) + .is_err() + ); + let fresh_history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + assert!( + store + .finish_backend_job_worker_cleanup_owned( + "workspace", + &fresh_history, + &failed, + BackendJobWorkerCleanupState::Completed, + None, + None, + later, + ) + .is_err(), + "diagnosis must not claim Executing ownership" + ); + assert!( + store + .reserve_backend_job_retry("workspace", "accepted", later) + .is_err() + ); + drop(store); + let store = + SqliteWorkspaceStore::open_for_test(&dir.path().join("recovery.sqlite")).unwrap(); + assert_eq!( + store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(), + vec![failed] + ); + assert_eq!( + store + .get_backend_job("workspace", "accepted") + .unwrap() + .unwrap(), + job + ); + } + + #[test] + fn cleanup_preflight_diagnosis_keeps_executing_and_rejects_fenced_completed_or_successor_owner() + { + let (_dir, store, reserved) = fixture(); + let bound = bind_fixture_owner(&store, &reserved); + // As in the fence component tests, only the external termination receipt + // is fixture evidence; this test does not claim real execution ended. + store.with_conn(|conn| { + conn.execute("UPDATE backend_job_attempts SET worker_id='019a4ea8-0000-7000-8000-000000000003' WHERE attempt_id=?1", params![bound.attempt_id])?; + Ok(()) + }).unwrap(); + let failed = store + .finish_backend_job_attempt( + "workspace", + "accepted", + &bound.attempt_id, + BackendJobAttemptState::Failed, + "execution_failed", + "saved original failure", + NOW, + ) + .unwrap(); + let history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + let (executing, acquired) = store + .claim_backend_job_worker_cleanup_owned("workspace", &history, &failed, NOW) + .unwrap(); + assert!(acquired); + let history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + let diagnosed = store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &executing, + "Host close/fence unavailable; no termination proof", + NOW, + ) + .unwrap(); + assert_eq!( + diagnosed.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Executing) + ); + assert_eq!( + diagnosed.failure_detail.as_deref(), + Some("saved original failure") + ); + assert_eq!( + diagnosed.worker_cleanup_failure_category.as_deref(), + Some("backend_job_cleanup_host_admission_unconfirmed") + ); + assert!(diagnosed.worker_cleanup_completed_at.is_none()); + assert!( + !store + .has_backend_job_execution_fence("workspace", &diagnosed) + .unwrap() + ); + assert!( + store + .finish_backend_job_worker_cleanup_owned( + "workspace", + &history, + &executing, + BackendJobWorkerCleanupState::Completed, + None, + None, + NOW, + ) + .is_err(), + "intervening diagnosis must invalidate stale completion CAS" + ); + let history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + let (controller, host) = evidence(&diagnosed); + store + .record_backend_job_execution_fence("workspace", &diagnosed, &controller, &host, NOW) + .unwrap(); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &diagnosed, + "late unfenced observation", + NOW, + ) + .is_err() + ); + assert_eq!( + store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(), + history + ); + let completed = store + .finish_backend_job_worker_cleanup_owned( + "workspace", + &history, + &diagnosed, + BackendJobWorkerCleanupState::Completed, + None, + None, + NOW, + ) + .unwrap(); + let history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &completed, + "late failure after completion", + NOW, + ) + .is_err() + ); + let next = store + .reserve_backend_job_retry("workspace", "accepted", NOW) + .unwrap(); + let successor = bind_fixture_owner(&store, &next); + let next_job = store + .get_backend_job("workspace", "accepted") + .unwrap() + .unwrap(); + let history = store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &completed, + "late historical failure", + NOW, + ) + .is_err() + ); + assert!( + store + .record_backend_job_cleanup_preflight_failure( + "workspace", + &history, + &successor, + "nonterminal owner cannot diagnose cleanup", + NOW, + ) + .is_err() + ); + assert_eq!( + store + .list_backend_job_attempts("workspace", "accepted") + .unwrap(), + history + ); + assert_eq!( + store + .get_backend_job("workspace", "accepted") + .unwrap() + .unwrap(), + next_job + ); + } + #[test] fn cleanup_cas_rejects_changed_history_and_wrong_worker_run_without_releasing_owner() { let (_dir, store, reserved) = fixture(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 8e6588480..0c4fc8656 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -5582,11 +5582,30 @@ impl WorkspaceApi { /// Restore owners for current-contract attempts without replaying uncertain effects. pub fn recover_backend_jobs(&self) -> Result<()> { - for attempt in self.store.list_recoverable_backend_job_attempts( - self.workspace_id(), - usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS), - )? { - self.own_backend_job_recovery(&attempt); + let page_limit = usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS); + let mut after = None; + loop { + let attempts = self.store.list_recoverable_backend_job_attempts_page( + self.workspace_id(), + after.as_ref(), + page_limit, + )?; + let Some(last) = attempts.last() else { + break; + }; + let next = crate::store::BackendJobRecoveryCursor { + job_id: last.job_id.clone(), + attempt_id: last.attempt_id.clone(), + }; + for attempt in &attempts { + self.own_backend_job_recovery(attempt); + } + if attempts.len() < page_limit { + break; + } + // Reservations are not bounded by active concurrency. Traverse past + // admitted owners even if they remain pending or change timestamps. + after = Some(next); } subjektiv_journal::recover_attempts(self)?; self.recover_backend_job_worker_cleanups()?; diff --git a/crates/workspace-server/src/server/backend_job_recovery.rs b/crates/workspace-server/src/server/backend_job_recovery.rs index dbc3a0158..9b2d3422b 100644 --- a/crates/workspace-server/src/server/backend_job_recovery.rs +++ b/crates/workspace-server/src/server/backend_job_recovery.rs @@ -22,6 +22,10 @@ pub(super) struct RecoveryOwners { submitted: Mutex>, stopping: std::sync::atomic::AtomicBool, changed: tokio::sync::Notify, + // Already accepted owners take fair turns before entering the finite Host + // lane. Immediate rejection plus independent retry timers otherwise lets + // earlier reservations monopolize every slot after a large reopen. + step_admission: tokio::sync::Mutex<()>, #[cfg(test)] pub(super) storage_probe_started: tokio::sync::Notify, #[cfg(test)] @@ -44,6 +48,11 @@ impl RecoveryOwners { self.stopping.store(false, Ordering::Release); } + #[cfg(test)] + pub(super) fn recovery_owner_ids(&self) -> HashSet { + self.running.lock().unwrap().clone() + } + #[cfg(test)] pub(super) fn cleanup_owner_ids(&self) -> HashSet { self.cleanup_running.lock().unwrap().clone() @@ -498,17 +507,54 @@ impl WorkspaceApi { .host_executors .cleanup()? .run(move || { + // Observe before preflight: a failure may diagnose only this + // exact row/history, never a successor or intervening progress. + let history = api + .store + .list_backend_job_attempts(api.workspace_id(), &owner.job_id)?; + let observed = history + .iter() + .find(|attempt| attempt.attempt_id == owner.attempt_id) + .ok_or_else(|| Error::Store("Cleanup owner disappeared".into()))?; + if !owner.state.terminal() + || !observed.state.terminal() + || observed.attempt != owner.attempt + || JobHostOwner::try_from(observed)? != JobHostOwner::try_from(&owner)? + { + return Err(Error::RepositoryConflict( + "Cleanup execution owner changed before preflight".into(), + )); + } // Every cleanup caller, including its blocking descendants, holds // the original admission resource exclusively through actual return. // Recovery therefore fences the predecessor's writers rather than // treating Executing, an empty map, or process exit as ownership. - api.backend_job_recovery - .close_host_admission(&JobHostOwner::try_from(&owner)?)?; - #[cfg(test)] - api.backend_job_recovery - .cleanup_before_writer_fence - .notify_one(); - let drain = JobHostEffects::fence_retained(&api.config.database_path, &owner)?; + let preflight = (|| { + api.backend_job_recovery + .close_host_admission(&JobHostOwner::try_from(observed)?)?; + #[cfg(test)] + api.backend_job_recovery + .cleanup_before_writer_fence + .notify_one(); + JobHostEffects::fence_retained(&api.config.database_path, observed) + })(); + let drain = match preflight { + Ok(drain) => drain, + Err(error) => { + api.store.record_backend_job_cleanup_preflight_failure( + api.workspace_id(), + &history, + observed, + &format!( + "Host admission close/fence failed: {error}; execution and Host termination remain unconfirmed. Retain result/input and reconcile the original admission resource; never create replacement evidence." + ), + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + // Neither a saved diagnosis nor failure to save it is + // success. Storage errors still reach retry/probe ownership. + return Err(error); + } + }; let current = api .store .get_backend_job_attempt( @@ -667,6 +713,10 @@ impl WorkspaceApi { let api = self.clone(); let owner = expected.clone(); let failure = definitive_failure.clone(); + let admission = self.backend_job_recovery.step_admission.lock().await; + if self.backend_job_recovery.stopping.load(Ordering::Acquire) { + return Ok(()); + } let step = self .host_executors .host()? @@ -689,6 +739,7 @@ impl WorkspaceApi { } }) .await; + drop(admission); match step { Ok(true) => return Ok(()), Ok(false) => { @@ -741,12 +792,17 @@ impl WorkspaceApi { self.backend_job_recovery.pause(delay).await; let api = self.clone(); let owner = expected.clone(); - match self + let admission = self.backend_job_recovery.step_admission.lock().await; + if self.backend_job_recovery.stopping.load(Ordering::Acquire) { + return Ok(()); + } + let probe = self .host_executors .host()? .run(move || api.probe_backend_job_storage(&owner)) - .await - { + .await; + drop(admission); + match probe { Ok(()) => { storage_failures = 0; break; 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 828b8b816..f9d0531bb 100644 --- a/crates/workspace-server/src/server/tests/backend_job_tests.rs +++ b/crates/workspace-server/src/server/tests/backend_job_tests.rs @@ -123,7 +123,14 @@ async fn wait_for_attempt( } }) .await - .expect("same accepted attempt did not reach its expected durable state") + .unwrap_or_else(|error| { + panic!( + "same accepted attempt did not reach its expected durable state: {error}; current={:?}; recovery_owners={:?}", + api.store + .get_backend_job_attempt(api.workspace_id(), job, attempt), + api.backend_job_recovery.recovery_owner_ids() + ) + }) } #[tokio::test] @@ -1055,7 +1062,7 @@ async fn terminal_backend_job_cleanup_removes_only_bound_worker_and_retains_resu } #[tokio::test] -async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_new_worker() { +async fn backend_job_cleanup_commit_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 { @@ -1074,14 +1081,18 @@ async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_ 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(); + // Fail the canonical catalog commit after Runtime retention, without changing + // Worker/policy authority and thereby invalidating the original removal child. + api.config_store + .with_conn(|conn| { + conn.execute_batch( + "CREATE TRIGGER fail_job_cleanup_catalog_commit + BEFORE INSERT ON worker_registry_projection_removals + BEGIN SELECT RAISE(ABORT, 'injected cleanup catalog failure'); END;", + )?; + Ok(()) + }) + .unwrap(); api.finish_backend_job_attempt( &request.job_id, &first.attempt.attempt_id, @@ -1102,33 +1113,87 @@ async fn backend_job_cleanup_guard_failure_retries_old_attempt_without_touching_ blocked.worker_cleanup_failure_category.as_deref(), Some("backend_job_cleanup_removal_rejected") ); + assert!( + blocked + .worker_cleanup_failure_detail + .as_deref() + .unwrap() + .contains("worker_removal_authority_unavailable"), + "canonical catalog commit failure must remain a removal error: {blocked:?}" + ); + let blocked_removal = api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &first_worker) + .unwrap() + .unwrap(); + assert_eq!( + blocked_removal.prior_failure_category.as_deref(), + Some("metadata_commit_failed") + ); + let first_removal_children = || { + api.config_store + .with_conn(|conn| { + let mut statement = conn.prepare( + "SELECT operation_id, input_fingerprint FROM worker_removal_operations + WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 + ORDER BY operation_id", + )?; + let children = statement + .query_map( + rusqlite::params![ + api.workspace_id(), + first_worker.runtime_id, + first_worker.worker_id + ], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)), + )? + .collect::>>()?; + Ok(children) + }) + .unwrap() + }; + let original_children = first_removal_children(); + assert_eq!(original_children.len(), 1); + assert_eq!(blocked.worker, Some(first_worker.clone())); + assert_eq!(blocked.runtime_run_id, first.attempt.runtime_run_id); assert!( api.retry_backend_job(&request.job_id).is_err(), "retry must not overlap an incompletely cleaned attempt" ); - let _ = set_worker_retention( - api.clone(), - first_worker.runtime_id.clone(), - first_worker.worker_id.clone(), - false, - ) - .await - .unwrap(); + api.config_store + .with_conn(|conn| { + conn.execute_batch("DROP TRIGGER fail_job_cleanup_catalog_commit;")?; + Ok(()) + }) + .unwrap(); api.recover_backend_jobs().unwrap(); - wait_for_backend_job_worker_cleanup( + let cleaned = wait_for_backend_job_worker_cleanup( &api, &request.job_id, &first.attempt.attempt_id, BackendJobWorkerCleanupState::Completed, ) .await; + assert_eq!(first_removal_children(), original_children); + assert_eq!(cleaned.worker, blocked.worker); + assert_eq!(cleaned.runtime_run_id, blocked.runtime_run_id); + api.require_backend_job_attempt_effects_ended(&cleaned) + .unwrap(); + assert!(api.runtime.worker(&first_worker).is_err()); + assert!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &first_worker) + .unwrap() + .is_none() + ); 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); api.recover_backend_jobs().unwrap(); assert!(api.runtime.worker(&second_worker).is_ok()); + assert_eq!(first_removal_children(), original_children); api.accept_backend_job_result( &second_worker, @@ -1312,6 +1377,12 @@ async fn backend_job_restart_without_live_host_owner_keeps_cleanup_unconfirmed() .worker_cleanup_state, Some(BackendJobWorkerCleanupState::Pending) ); + // This fixture deliberately bypasses producer Host admission. Missing the + // original durable resource is not merely losing the live owner map. + let mut admission_root = config.database_path.as_os_str().to_owned(); + admission_root.push(".job-host-admission"); + let admission_root = std::path::PathBuf::from(admission_root); + assert!(!admission_root.exists()); let attempt_id = reservation.attempt.attempt_id; drop(runtime); drop(embedded); @@ -1325,7 +1396,7 @@ async fn backend_job_restart_without_live_host_owner_keeps_cleanup_unconfirmed() let restarted = WorkspaceApi::new_with_execution_backend( config, Arc::new(restarted_store), - restarted_execution, + restarted_execution.clone(), ) .await .unwrap(); @@ -1337,6 +1408,53 @@ async fn backend_job_restart_without_live_host_owner_keeps_cleanup_unconfirmed() ) .await; assert_eq!(cleaned.result, Some(serde_json::json!({"persisted": true}))); + assert_eq!(cleaned.state, BackendJobAttemptState::Completed); + assert_eq!(cleaned.worker.as_ref(), Some(&worker)); + assert_eq!( + cleaned.runtime_run_id.as_deref(), + Some(runtime_run_id.as_str()) + ); + assert_eq!( + cleaned.worker_cleanup_failure_category.as_deref(), + Some("backend_job_cleanup_host_admission_unconfirmed") + ); + let cause = cleaned.worker_cleanup_failure_detail.as_deref().unwrap(); + assert!(cause.contains("Original Job Host admission resource unavailable")); + assert!(cause.contains("termination remain unconfirmed")); + assert!(cleaned.worker_cleanup_completed_at.is_none()); + assert!(cleaned.allocation_precluded_at.is_none()); + assert!( + !admission_root.exists(), + "recovery must not create Host evidence" + ); + let saved = restarted + .store + .get_backend_job(TEST_WORKSPACE_ID, &request.job_id) + .unwrap() + .unwrap(); + assert_eq!(saved.state, BackendJobState::Completed); + assert_eq!(saved.result, Some(serde_json::json!({"persisted": true}))); + assert_eq!(saved.request, request); + assert_eq!( + saved.input_disposition, + crate::backend_job::JobInputDisposition::Available + ); + assert!( + !restarted + .store + .has_backend_job_execution_fence(TEST_WORKSPACE_ID, &cleaned) + .unwrap() + ); + restarted.config_store.with_conn(|connection| { + let fences: i64 = connection.query_row("SELECT COUNT(*) FROM backend_job_execution_fences WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3", rusqlite::params![TEST_WORKSPACE_ID, request.job_id, attempt_id], |row| row.get(0))?; + assert_eq!(fences, 0, "preflight failure must not fabricate execution evidence"); + let removals: 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!(removals, 0, "preflight failure must not call canonical removal"); + Ok(()) + }).unwrap(); + assert!(restarted_execution.stops.lock().unwrap().is_empty()); + assert!(restarted_execution.terminations.lock().unwrap().is_empty()); + assert!(restarted_execution.take_inputs().is_empty()); assert!( restarted .require_backend_job_attempt_effects_ended(&cleaned) @@ -1710,22 +1828,82 @@ async fn backend_job_input_response_loss_is_owned_until_deadline_without_resubmi Some("submission_receipt_unconfirmed") ); let worker = failed.worker.clone().unwrap(); - wait_for_backend_job_worker_cleanup( + // The substituted callee returned Errored before saving an acceptance receipt. + // Stop can join its fixture execution, but cannot attest this exact run ended. + let cleaned = wait_for_backend_job_worker_cleanup( &api, &request.job_id, &failed.attempt_id, - BackendJobWorkerCleanupState::Completed, + BackendJobWorkerCleanupState::Failed, ) .await; - assert!(matches!( - api.runtime.worker(&worker), - Err(RuntimeRegistryError::UnknownWorker { .. }) - )); + assert_eq!(cleaned.state, BackendJobAttemptState::Failed); + assert_eq!(cleaned.failure_category.as_deref(), Some("timeout")); + assert!( + cleaned + .failure_detail + .as_deref() + .unwrap() + .contains("Attempt deadline exceeded") + ); + assert_eq!(cleaned.worker, failed.worker); + assert_eq!(cleaned.runtime_run_id, failed.runtime_run_id); + assert_eq!(cleaned.deadline_at, failed.deadline_at); + assert_eq!( + cleaned.worker_cleanup_failure_category.as_deref(), + Some("backend_job_cleanup_effects_unconfirmed") + ); + assert!( + cleaned + .worker_cleanup_failure_detail + .as_deref() + .unwrap() + .contains( + "Runtime receipt does not prove exact Controller/Feature/background termination" + ) + ); + assert!(cleaned.worker_cleanup_completed_at.is_none()); + assert!( + api.require_backend_job_attempt_effects_ended(&cleaned) + .is_err() + ); + assert!(api.retry_backend_job(&request.job_id).is_err()); + let retained = api + .store + .get_backend_job(TEST_WORKSPACE_ID, &request.job_id) + .unwrap() + .unwrap(); + assert_eq!(retained.state, BackendJobState::Failed); + assert_eq!(retained.failure_category.as_deref(), Some("timeout")); + assert_eq!(retained.request, request); + assert_eq!( + retained.input_disposition, + crate::backend_job::JobInputDisposition::Available + ); + assert!(api.runtime.worker(&worker).is_ok()); + assert!( + api.store + .get_worker_registry(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_some() + ); + assert!( + api.config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, &worker) + .unwrap() + .is_none(), + "unconfirmed exact run termination must not authorize canonical removal" + ); assert_eq!( execution.take_inputs().len(), 1, "ambiguous input was never blindly resubmitted" ); + assert_eq!( + execution.take_input_request_ids(), + vec![failed.runtime_run_id.clone()], + "only the original tracked Submit may reach the execution callee" + ); assert!( api.store .get_backend_job_attempt( @@ -1736,6 +1914,7 @@ async fn backend_job_input_response_loss_is_owned_until_deadline_without_resubmi .unwrap() .is_none() ); + api.stop_backend_job_recovery().await; } #[tokio::test] @@ -1962,7 +2141,10 @@ async fn backend_job_missing_worker_reaches_failure_without_fabricated_cleanup() serialization_key: None, source_worker: None, notification_target: None, - limits: crate::backend_job::BackendJobLimits::default(), + limits: crate::backend_job::BackendJobLimits { + timeout_seconds: 1, + ..Default::default() + }, }; let reserved = api .store @@ -2021,6 +2203,12 @@ async fn backend_job_missing_worker_reaches_failure_without_fabricated_cleanup() ) .await; assert_eq!(attempt.failure_category.as_deref(), Some("timeout")); + assert_eq!(attempt.deadline_at, reserved.attempt.deadline_at); + assert_eq!(attempt.worker, Some(missing_worker)); + assert_eq!( + attempt.runtime_run_id.as_deref(), + Some("missing-runtime-run") + ); assert_ne!( attempt.worker_cleanup_state, Some(BackendJobWorkerCleanupState::Completed) @@ -2093,3 +2281,235 @@ async fn backend_job_timeout_fences_late_structured_result() { .is_err() ); } +#[tokio::test] +async fn backend_job_reopen_owns_later_nonjournal_attempt_behind_queued_reservations() { + let temp = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let config = api.config.clone(); + // Pause only automatic recovery while creating the reachable accepted state. + // Dispatch, capacity checks, Runtime acceptance and persistence remain real. + api.stop_backend_job_recovery().await; + let mut first_request = recovery_request("a-startup-active"); + first_request.limits.max_concurrent_jobs = 1; + first_request.limits.timeout_seconds = crate::backend_job::MAX_JOB_TIMEOUT_SECONDS; + let first = api.dispatch_backend_job(&first_request).unwrap(); + assert_eq!(first.attempt.state, BackendJobAttemptState::Dispatched); + + let mut queued = Vec::new(); + for index in 0..usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS) { + let mut request = recovery_request(&format!("b-startup-queued-{index:03}")); + request.limits.max_concurrent_jobs = 1; + request.limits.timeout_seconds = crate::backend_job::MAX_JOB_TIMEOUT_SECONDS; + let reservation = api.dispatch_backend_job(&request).unwrap(); + assert_eq!(reservation.attempt.state, BackendJobAttemptState::Reserved); + assert!(reservation.attempt.worker.is_none()); + queued.push(reservation.attempt); + } + let mut later_request = recovery_request("z-startup-later-active"); + later_request.limits.max_concurrent_jobs = ABSOLUTE_MAX_CONCURRENT_JOBS; + // This attempt's own original deadline supplies the autonomous observation; + // older work stays pending. No fixed sleep or production timeout change. + later_request.limits.timeout_seconds = 1; + let later = api.dispatch_backend_job(&later_request).unwrap(); + assert_eq!(later.attempt.state, BackendJobAttemptState::Dispatched); + assert!(later.job.request.grants.subjektiv_journal.is_none()); + let original_worker = later.attempt.worker.clone().unwrap(); + assert_eq!(execution.take_inputs().len(), 2); + assert!( + execution + .take_input_request_ids() + .contains(&later.attempt.runtime_run_id) + ); + + let first_page = api + .store + .list_recoverable_backend_job_attempts( + api.workspace_id(), + usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS), + ) + .unwrap(); + assert_eq!(first_page.len(), usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS)); + assert!( + first_page + .iter() + .all(|attempt| attempt.job_id != later_request.job_id) + ); + assert!( + api.store + .list_reserved_backend_job_attempts(api.workspace_id(), queued.len() + 1) + .unwrap() + .iter() + .all(|attempt| attempt.job_id != later_request.job_id) + ); + + // Reopen the actual workspace DB and construct a fresh API/Runtime. Do not + // substitute an empty owner map on a live API or call full recovery manually. + shutdown_workspace_feature_storage(&api).await.unwrap(); + drop(api); + drop(execution); + let reopened_store = + Arc::new(SqliteWorkspaceStore::open_for_test(config.database_path.clone()).unwrap()); + let reopened_execution = Arc::new(DeterministicExecutionBackend::default()); + reopened_execution.accept_restores(); + let reopened = WorkspaceApi::new_with_execution_backend( + config, + reopened_store, + reopened_execution.clone(), + ) + .await + .unwrap(); + assert!( + reopened + .backend_job_recovery + .recovery_owner_ids() + .contains(&later.attempt.attempt_id) + || reopened + .store + .get_backend_job_attempt( + reopened.workspace_id(), + &later_request.job_id, + &later.attempt.attempt_id + ) + .unwrap() + .unwrap() + .state + .terminal(), + "startup must own later active attempt; owners={:?}; candidate={:?}; job={:?}", + reopened.backend_job_recovery.recovery_owner_ids(), + reopened.store.list_recoverable_backend_job_attempts_page( + reopened.workspace_id(), + Some(&crate::store::BackendJobRecoveryCursor { + job_id: "b-startup-queued-063".into(), + attempt_id: "".into() + }), + 64 + ), + reopened + .store + .get_backend_job(reopened.workspace_id(), &later_request.job_id) + ); + let recovered = wait_for_attempt( + &reopened, + &later_request.job_id, + &later.attempt.attempt_id, + |attempt| attempt.state == BackendJobAttemptState::Failed, + ) + .await; + assert_eq!(recovered.failure_category.as_deref(), Some("timeout")); + assert_eq!(recovered.worker, Some(original_worker)); + assert_eq!(recovered.attempt, later.attempt.attempt); + assert_eq!(recovered.runtime_run_id, later.attempt.runtime_run_id); + assert_eq!(recovered.input_digest, later.attempt.input_digest); + assert_eq!(recovered.deadline_at, later.attempt.deadline_at); + assert_eq!( + reopened + .store + .list_backend_job_attempts(reopened.workspace_id(), &later_request.job_id) + .unwrap() + .len(), + 1, + "startup recovery must retain the original accepted attempt" + ); + for attempt in &queued { + let saved = reopened + .store + .get_backend_job_attempt( + reopened.workspace_id(), + &attempt.job_id, + &attempt.attempt_id, + ) + .unwrap() + .unwrap(); + assert_eq!(saved.state, BackendJobAttemptState::Reserved); + assert!(saved.worker.is_none()); + } + assert!( + reopened_execution.take_inputs().is_empty(), + "restored active non-Journal work must not be submitted again" + ); + shutdown_workspace_feature_storage(&reopened).await.unwrap(); +} + +#[tokio::test] +async fn backend_job_recovery_pages_remain_read_only_when_earlier_attempts_terminalize() { + let temp = tempfile::tempdir().unwrap(); + let api = test_api(temp.path()).await; + api.stop_backend_job_recovery().await; + let page_limit = usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS); + let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); + let mut seeded = Vec::new(); + for index in 0..(page_limit + 2) { + let request = recovery_request(&format!("startup-keyset-{index:03}")); + seeded.push( + api.store + .reserve_backend_job(api.workspace_id(), &request, &now) + .unwrap() + .attempt, + ); + } + let first = api + .store + .list_recoverable_backend_job_attempts_page(api.workspace_id(), None, usize::MAX) + .unwrap(); + assert_eq!(first, seeded[..page_limit]); + for attempt in &seeded { + assert_eq!( + api.store + .get_backend_job_attempt(api.workspace_id(), &attempt.job_id, &attempt.attempt_id) + .unwrap() + .unwrap(), + *attempt, + "enumeration must not claim, rewrite timestamps, or reset dispatch" + ); + } + let last = first.last().unwrap(); + let cursor = crate::store::BackendJobRecoveryCursor { + job_id: last.job_id.clone(), + attempt_id: last.attempt_id.clone(), + }; + // An OFFSET or mutable-state cursor would skip the later accepted work after + // the first page's owners progress. All creation times are deliberately equal. + for attempt in &first { + api.store + .finish_backend_job_attempt( + api.workspace_id(), + &attempt.job_id, + &attempt.attempt_id, + BackendJobAttemptState::Failed, + "fixture_preallocation_failure", + "First page settled before the next bounded read", + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + } + let second = api + .store + .list_recoverable_backend_job_attempts_page(api.workspace_id(), Some(&cursor), page_limit) + .unwrap(); + assert_eq!(second, seeded[page_limit..]); + for attempt in &second { + assert_eq!( + api.store + .get_backend_job_attempt(api.workspace_id(), &attempt.job_id, &attempt.attempt_id) + .unwrap() + .unwrap(), + *attempt + ); + } + let last = second.last().unwrap(); + let cursor = crate::store::BackendJobRecoveryCursor { + job_id: last.job_id.clone(), + attempt_id: last.attempt_id.clone(), + }; + assert!( + api.store + .list_recoverable_backend_job_attempts_page( + api.workspace_id(), + Some(&cursor), + page_limit + ) + .unwrap() + .is_empty() + ); + shutdown_workspace_feature_storage(&api).await.unwrap(); +} diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs index b741a26b3..b346e233d 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs @@ -2759,7 +2759,40 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) { // independent Journal job, not a second execution of the restored body. let before_stop = fixture.execution.worker_snapshot(&worker_ref); let original_stop_count = probe.stops.lock().unwrap().len(); - let stopped = connected_stop_worker(api, worker).await; + let mut stopped = connected_stop_worker(api, worker).await; + let uncertain = { + let stops = probe.stops.lock().unwrap(); + stops[original_stop_count..] + .last() + .and_then(|(request, result)| { + result + .message + .as_deref() + .filter(|message| { + message.contains("admission is unknown and waiter is retained") + }) + .and_then(|message| message.strip_prefix("Shutdown command ")) + .and_then(|message| message.split_whitespace().next()?.parse::().ok()) + .map(|command| (request.clone(), command)) + }) + }; + if let Some((request, _)) = &uncertain { + // A bounded admission wait is not definitive rejection. Reconcile + // the retained operation/command once, without a readiness method, + // replacement Controller, new Stop identity, or a new input. + let pending = record(); + assert_eq!( + pending["execution_state"]["state"], + "reconciliation_required" + ); + assert_eq!(pending["execution_state"]["execution"]["operation"], "stop"); + assert_eq!( + pending["execution_state"]["execution"]["intent"]["operation_id"], + serde_json::to_value(request.operation_id).unwrap() + ); + assert!(pending["restore_guard"]["execution_termination"].is_null()); + stopped = connected_stop_worker(api, worker).await; + } let task_finished = probe .controller_tasks .lock() @@ -2788,6 +2821,20 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) { .expect("teardown must obtain a typed Stop from the real adapter"); assert_eq!(stop.worker_ref, worker_ref); assert!(result.is_accepted(), "real adapter teardown: {result:?}"); + assert!( + task_finished.iter().all(|finished| *finished), + "real Controller tasks must finish: {task_finished:?}" + ); + if let Some((original, command)) = uncertain { + assert!(stops[original_stop_count..].iter().all(|(request, _)| { + request.operation_id == original.operation_id + && request.worker_ref == original.worker_ref + })); + assert_eq!( + result.termination.as_ref().unwrap().shutdown_command_id, + Some(command) + ); + } assert!( result.termination.as_ref().is_some_and( |receipt| receipt.confirms_scope_shutdown(&worker_ref, stop.operation_id) diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index ccbbb0d00..4e20ec416 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -1396,6 +1396,13 @@ pub struct HostingAdminInitialBearer { pub expires_at: String, } +/// Immutable boundary for the all-state startup recovery scan, not an execution claim. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BackendJobRecoveryCursor { + pub job_id: String, + pub attempt_id: String, +} + #[async_trait] pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { fn plugin_upload( @@ -2054,6 +2061,15 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { expected: &BackendJobAttemptRecord, now: &str, ) -> Result<(BackendJobAttemptRecord, bool)>; + /// Diagnose failed Host preflight without claiming cleanup or proving termination. + fn record_backend_job_cleanup_preflight_failure( + &self, + workspace_id: &str, + expected_history: &[BackendJobAttemptRecord], + expected: &BackendJobAttemptRecord, + failure_detail: &str, + now: &str, + ) -> Result; fn finish_backend_job_worker_cleanup_owned( &self, workspace_id: &str, @@ -2064,10 +2080,21 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { failure_detail: Option<&str>, now: &str, ) -> Result; + /// Read the first bounded page; use the paged variant to enumerate all owners. fn list_recoverable_backend_job_attempts( &self, workspace_id: &str, limit: usize, + ) -> Result> { + self.list_recoverable_backend_job_attempts_page(workspace_id, None, limit) + } + /// Read one bounded page without claiming or changing any attempt. + /// Startup must traverse every page, independently of admitted owner progress. + fn list_recoverable_backend_job_attempts_page( + &self, + workspace_id: &str, + after: Option<&BackendJobRecoveryCursor>, + limit: usize, ) -> Result>; fn claim_backend_job_attempt_dispatch( &self, @@ -8315,6 +8342,23 @@ impl ControlPlaneStore for SqliteWorkspaceStore { now, ) } + fn record_backend_job_cleanup_preflight_failure( + &self, + workspace_id: &str, + expected_history: &[BackendJobAttemptRecord], + expected: &BackendJobAttemptRecord, + failure_detail: &str, + now: &str, + ) -> Result { + backend_job_recovery_store::record_cleanup_preflight_failure( + self, + workspace_id, + expected_history, + expected, + failure_detail, + now, + ) + } fn finish_backend_job_worker_cleanup_owned( &self, workspace_id: &str, @@ -8336,26 +8380,42 @@ impl ControlPlaneStore for SqliteWorkspaceStore { now, ) } - fn list_recoverable_backend_job_attempts( + fn list_recoverable_backend_job_attempts_page( &self, workspace_id: &str, + after: Option<&BackendJobRecoveryCursor>, limit: usize, ) -> Result> { - self.with_conn_mut(|conn| { - let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; - // Preserve the last confirmed dispatch fact. An unbound claim may - // already own an external allocation; the Server reconciles its ID. + self.with_conn(|conn| { + // A deferred read snapshot keeps identities and records consistent + // without reserving the writer or changing the last dispatch fact. + let tx = conn.unchecked_transaction()?; let mut statement = tx.prepare( - "SELECT a.job_id, a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j USING(workspace_id,job_id) WHERE a.workspace_id = ?1 AND a.state IN ('reserved', 'dispatching', 'dispatched') AND a.attempt=j.current_attempt AND j.state='pending' AND j.input_disposition='available' AND NOT EXISTS(SELECT 1 FROM backend_job_upgrade_history h WHERE h.workspace_id=a.workspace_id AND h.job_id=a.job_id) ORDER BY a.created_at LIMIT ?2", + "SELECT a.job_id, a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j USING(workspace_id,job_id) WHERE a.workspace_id = ?1 AND (?3 IS NULL OR (a.job_id,a.attempt_id) > (?3,?4)) AND a.state IN ('reserved', 'dispatching', 'dispatched') AND a.attempt=j.current_attempt AND j.state='pending' AND j.input_disposition='available' AND NOT EXISTS(SELECT 1 FROM backend_job_upgrade_history h WHERE h.workspace_id=a.workspace_id AND h.job_id=a.job_id) ORDER BY a.job_id, a.attempt_id 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 remain stable when earlier owners update state or + // timestamps. Reading a page never grants allocation/Submit authority. + let limit = limit.min(usize::from(crate::backend_job::ABSOLUTE_MAX_CONCURRENT_JOBS)); + let identities = statement + .query_map( + params![ + workspace_id, + limit as i64, + after.map(|cursor| cursor.job_id.as_str()), + after.map(|cursor| cursor.attempt_id.as_str()), + ], + |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 attempt disappeared".to_string())) - }).collect::>>()?; + 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 attempt disappeared".to_string()) + }) + }) + .collect::>>()?; tx.commit()?; Ok(attempts) })