From 0d1840249beea0402c7b3ae38fb3a49651835083 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 5 Nov 2026 00:57:32 +0900 Subject: [PATCH] fix: fence original host writers and retain canonical job cleanup children --- .../src/backend_job_host_effects.rs | 154 +++- .../src/backend_job_recovery_store.rs | 35 +- crates/workspace-server/src/retention.rs | 116 +++ crates/workspace-server/src/server.rs | 56 +- .../src/server/backend_job_recovery.rs | 145 ++-- .../src/server/tests/backend_job_tests.rs | 543 +++++++++--- .../src/server/tests/subjektiv_jobs_tests.rs | 9 + .../subjektiv_journal_connected_tests.rs | 11 +- .../server/tests/subjektiv_journal_tests.rs | 804 ++++++++++++++---- crates/workspace-server/src/store.rs | 19 + 10 files changed, 1537 insertions(+), 355 deletions(-) diff --git a/crates/workspace-server/src/backend_job_host_effects.rs b/crates/workspace-server/src/backend_job_host_effects.rs index 77bd25cd5..22b5e2933 100644 --- a/crates/workspace-server/src/backend_job_host_effects.rs +++ b/crates/workspace-server/src/backend_job_host_effects.rs @@ -1,12 +1,16 @@ -//! Live ownership of Host effects admitted for one exact Job execution. +//! Exact Backend Job Host admission, independent of Controller and domain completion. //! -//! This supplements durable allocation/submission/domain receipts; it is not a Job state machine. -//! Create it only when initially admitting an execution, before its first Host callback. Recovery -//! must not manufacture a drain receipt by reconstructing an empty owner after a restart. A lost -//! owner requires reconciliation of the retained domain operations and writer fences instead. -//! Closing admission does not cancel accepted operations. Their existing DbExecutor lifetimes -//! retain the request lease through actual work and discarded-result cleanup, even if the HTTP -//! waiter disappears. Detached work must explicitly retain this same lease or is not covered. +//! Each accepted callback retains a shared admission-file lease through actual DB/Host work, +//! including discarded-response cleanup. After the canonical attempt closes admission, an +//! exclusive lease on that same retained resource fences those callbacks across Server owners; +//! neither an empty counter nor process exit issues a receipt. The original identity file must +//! exist and match. This does not certify detached external work: Backend Job grants permit +//! only result and bounded synchronous Subject operations, not arbitrary external mutations. +//! Canonical notification delivery and Worker removal retain their separate operation owners. +use sha2::{Digest, Sha256}; +use std::fs::{File, OpenOptions}; +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use tokio::sync::Notify; @@ -58,19 +62,69 @@ struct Admission { pub(crate) struct JobHostEffects { owner: JobHostOwner, + admission_path: PathBuf, admission: Mutex, finished: Notify, } impl JobHostEffects { - pub(crate) fn initially_admit(owner: JobHostOwner) -> Arc { - Arc::new(Self { + pub(crate) fn initially_admit(owner: JobHostOwner, database: &Path) -> Result> { + let admission_path = admission_path(database, &owner)?; + let parent = admission_path.parent().unwrap(); + std::fs::create_dir_all(parent)?; + // Publish a complete identity before any callback can acquire its lease. + // The temporary name is not an execution, release or recovery identity. + let temporary = parent.join(format!(".{}.tmp", uuid::Uuid::new_v4())); + let publish = (|| -> Result<()> { + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary)?; + file.write_all( + &serde_json::to_vec(&owner).map_err(|error| Error::Store(error.to_string()))?, + )?; + file.sync_all()?; + match std::fs::hard_link(&temporary, &admission_path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + open_admission(&admission_path, &owner)?; + } + Err(error) => return Err(error.into()), + } + // Resolve a prior publication whose directory sync acknowledgement failed. + File::open(parent)?.sync_all()?; + Ok(()) + })(); + let _ = std::fs::remove_file(temporary); + publish?; + Ok(Arc::new(Self { owner, + admission_path, admission: Mutex::new(Admission { open: true, requests: 0, }), finished: Notify::new(), + })) + } + + /// The caller must recheck the canonical current terminal attempt while holding + /// this guard, then persist its fence before releasing it. No resource is created. + pub(crate) fn fence_retained( + database: &Path, + attempt: &BackendJobAttemptRecord, + ) -> Result { + if !attempt.state.terminal() { + return Err(Error::RepositoryConflict( + "Job Host admission is not canonically closed".into(), + )); + } + let owner = JobHostOwner::try_from(attempt)?; + let file = open_admission(&admission_path(database, &owner)?, &owner)?; + file.lock()?; + Ok(JobHostDrain { + owner, + _admission: file, }) } @@ -89,6 +143,12 @@ impl JobHostEffects { "Job Host execution admission is closed".into(), )); } + let file = open_admission(&self.admission_path, expected)?; + // Never block a DB lane behind cleanup's exclusive guard: cleanup needs + // that lane to commit its evidence. The canonical terminal row rejects replay. + file.try_lock_shared().map_err(|error| { + Error::RepositoryConflict(format!("Job Host admission is fenced: {error}")) + })?; admission.requests = admission .requests .checked_add(1) @@ -96,16 +156,24 @@ impl JobHostEffects { Ok(JobHostRequestLease { request: Arc::new(AdmittedRequest { effects: self.clone(), + _admission: file, }), }) } - /// The receipt proves only this live Host owner's accepted operations ended. Its constructor - /// is private so a missing map entry, Job terminal state, or timeout cannot create evidence. - pub(crate) async fn close_and_drain( - self: &Arc, - expected: &JobHostOwner, - ) -> Result { + pub(crate) fn close_admission(&self, expected: &JobHostOwner) -> Result<()> { + if &self.owner != expected { + return Err(Error::InvalidInput("Job Host close owner mismatch".into())); + } + self.admission + .lock() + .map_err(|_| Error::Store("Job Host admission poisoned".into()))? + .open = false; + Ok(()) + } + + #[cfg(test)] + async fn close_and_drain(self: &Arc, expected: &JobHostOwner) -> Result { if &self.owner != expected { return Err(Error::InvalidInput("Job Host close owner mismatch".into())); } @@ -124,8 +192,11 @@ impl JobHostEffects { .requests == 0 { + let file = open_admission(&self.admission_path, expected)?; + file.lock()?; return Ok(JobHostDrain { owner: self.owner.clone(), + _admission: file, }); } notified.await; @@ -148,6 +219,7 @@ impl DbOperationLifetime for JobHostRequestLease { struct AdmittedRequest { effects: Arc, + _admission: File, } impl Drop for AdmittedRequest { @@ -165,6 +237,7 @@ impl Drop for AdmittedRequest { pub(crate) struct JobHostDrain { owner: JobHostOwner, + _admission: File, } impl JobHostDrain { @@ -173,6 +246,41 @@ impl JobHostDrain { } } +fn admission_path(database: &Path, owner: &JobHostOwner) -> Result { + let identity = serde_json::to_vec(owner).map_err(|error| Error::Store(error.to_string()))?; + let key: String = Sha256::digest(identity) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + let mut root = database.as_os_str().to_owned(); + root.push(".job-host-admission"); + Ok(PathBuf::from(root).join(key)) +} + +fn open_admission(path: &Path, expected: &JobHostOwner) -> Result { + // Recovery must never create a missing predecessor's admission resource. + let mut file = OpenOptions::new() + .read(true) + .write(true) + .open(path) + .map_err(|error| { + Error::RepositoryConflict(format!( + "Original Job Host admission resource unavailable: {error}" + )) + })?; + let mut bytes = Vec::new(); + (&mut file).take(8193).read_to_end(&mut bytes)?; + let saved: JobHostOwner = serde_json::from_slice(&bytes).map_err(|_| { + Error::RepositoryConflict("Original Job Host admission identity invalid".into()) + })?; + if bytes.len() > 8192 || &saved != expected { + return Err(Error::RepositoryConflict( + "Original Job Host admission identity changed".into(), + )); + } + Ok(file) +} + #[cfg(test)] mod tests { use super::*; @@ -193,7 +301,12 @@ mod tests { #[tokio::test] async fn cancelled_waiter_cannot_release_accepted_host_effect_or_admit_after_close() { let expected = owner("attempt-1"); - let effects = JobHostEffects::initially_admit(expected.clone()); + let directory = tempfile::tempdir().unwrap(); + let effects = JobHostEffects::initially_admit( + expected.clone(), + &directory.path().join("store.sqlite"), + ) + .unwrap(); let lease = effects.admit(&expected).unwrap(); let executor = DbExecutor::new(DbExecutorLimits { workers: NonZeroUsize::new(1).unwrap(), @@ -234,7 +347,12 @@ mod tests { #[tokio::test] async fn stale_close_cannot_revoke_a_successor_owner_or_create_its_drain_receipt() { let current = owner("attempt-2"); - let effects = JobHostEffects::initially_admit(current.clone()); + let directory = tempfile::tempdir().unwrap(); + let effects = JobHostEffects::initially_admit( + current.clone(), + &directory.path().join("store.sqlite"), + ) + .unwrap(); assert!(effects.close_and_drain(&owner("attempt-1")).await.is_err()); let request = effects.admit(¤t).unwrap(); let child = request.clone(); diff --git a/crates/workspace-server/src/backend_job_recovery_store.rs b/crates/workspace-server/src/backend_job_recovery_store.rs index cf33c8967..d5f3b9ee6 100644 --- a/crates/workspace-server/src/backend_job_recovery_store.rs +++ b/crates/workspace-server/src/backend_job_recovery_store.rs @@ -301,6 +301,10 @@ fn validate_fence( || ended.execution_id.is_nil() || ended.shutdown_command_id.is_none() || ended.worker_ref.worker_id.to_string() != owner.worker.worker_id + || !ended + .submissions + .iter() + .any(|submission| submission.submission_request_id == owner.runtime_run_id) { return Err(Error::InvalidInput("Controller/Feature/background and Host drain must confirm the exact owner independently".into())); } @@ -1070,7 +1074,11 @@ mod tests { "scope":"controller", "worker_ref":{"worker_id":owner.worker.as_ref().unwrap().worker_id}, "stop_operation_id":"019a4ea8-0000-7000-8000-000000000001", "execution_id":"019a4ea8-0000-7000-8000-000000000002", - "shutdown_command_id":1,"submissions":[],"feature_failures":[] + "shutdown_command_id":1,"submissions":[{ + "submission_request_id":owner.runtime_run_id, + "submission_id":"accepted-submission", + "disposition":"started" + }],"feature_failures":[] }); ( controller.to_string(), @@ -1103,6 +1111,31 @@ mod tests { .record_backend_job_execution_fence("workspace", &owner, &controller, &wrong, NOW) .is_err() ); + let mut wrong_run: serde_json::Value = serde_json::from_str(&controller).unwrap(); + wrong_run["submissions"][0]["submission_request_id"] = "successor-run".into(); + assert!( + store + .record_backend_job_execution_fence( + "workspace", + &owner, + &wrong_run.to_string(), + &host, + NOW + ) + .is_err() + ); + wrong_run["submissions"] = serde_json::json!([]); + assert!( + store + .record_backend_job_execution_fence( + "workspace", + &owner, + &wrong_run.to_string(), + &host, + NOW + ) + .is_err() + ); let bad = controller.replace("controller", "unpublished_startup_candidate"); assert!( store diff --git a/crates/workspace-server/src/retention.rs b/crates/workspace-server/src/retention.rs index 1d808bba5..16f4f908c 100644 --- a/crates/workspace-server/src/retention.rs +++ b/crates/workspace-server/src/retention.rs @@ -310,6 +310,26 @@ impl SqliteWorkspaceStore { &self, req: &WorkerRemovalPlanRequest, inv: &WorkerRetentionInventory, + ) -> Result { + self.plan_worker_removal_inner(req, inv, false) + } + + /// An automatic Job cleanup may create its first canonical child, but cannot + /// replace one hidden by authority/staleness filters after an uncertain effect. + /// The predecessor check and first-plan save share the canonical writer Tx. + pub(crate) fn plan_job_cleanup_worker_removal( + &self, + req: &WorkerRemovalPlanRequest, + inv: &WorkerRetentionInventory, + ) -> Result { + self.plan_worker_removal_inner(req, inv, true) + } + + fn plan_worker_removal_inner( + &self, + req: &WorkerRemovalPlanRequest, + inv: &WorkerRetentionInventory, + first_job_cleanup_child: bool, ) -> Result { validate_plan(req, inv)?; if req @@ -325,6 +345,15 @@ impl SqliteWorkspaceStore { let now = Utc::now().to_rfc3339(); self.with_conn_mut(|conn| { let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + if first_job_cleanup_child { + let prior: Option = tx.query_row( + "SELECT plan_id FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 ORDER BY created_at,operation_id LIMIT 1", + params![req.workspace_id,req.worker.runtime_id,req.worker.worker_id], |row| row.get(0), + ).optional()?; + if let Some(plan_id) = prior { + return Err(StoreError::InvalidInput(format!("stale:{plan_id}:Job cleanup must reconcile its existing canonical removal; a filtered child is not absence"))); + } + } let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal,true)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?; let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? { Some(v)=>v, @@ -2785,6 +2814,93 @@ mod tests { assert_ne!(replacement.plan_id, plan.plan_id); } + #[test] + fn automatic_job_cleanup_cannot_replace_a_hidden_or_stale_canonical_child() { + for changed in ["policy", "worker"] { + let store = setup(); + let request = req(); + let original = store + .plan_job_cleanup_worker_removal(&request, &inv()) + .unwrap(); + store + .prepare_worker_removal_execution( + "w", + &original.plan_id, + &original.input_fingerprint, + ) + .unwrap(); + store + .fail_worker_removal( + "w", + &original.operation_id, + &original.input_fingerprint, + "runtime_outcome_unconfirmed", + ) + .unwrap(); + if changed == "policy" { + let policy = store.worker_retention_policy("w").unwrap().unwrap(); + store + .update_worker_retention_policy( + "w", + &policy.policy_digest, + &WorkerRetentionPolicyUpdate { + policy_id: "changed-after-uncertain-removal".into(), + session_disposition: policy.session_disposition, + metadata_disposition: policy.metadata_disposition, + archive_retention: policy.archive_retention, + diagnostics_disposition: policy.diagnostics_disposition, + diagnostics_retention_seconds: policy.diagnostics_retention_seconds, + }, + ) + .unwrap(); + assert!( + store + .prepare_worker_removal_execution( + "w", + &original.plan_id, + &original.input_fingerprint + ) + .is_err() + ); + } else { + store.with_conn(|connection| { + connection.execute("UPDATE worker_registry SET updated_at='2026-08-24T00:00:01Z' WHERE workspace_id='w' AND runtime_id='r' AND worker_id=?1", [worker_id().to_string()])?; + Ok(()) + }).unwrap(); + } + assert!( + store + .recover_worker_removal_execution("w", &request.worker) + .unwrap() + .is_none() + ); + assert!( + store + .plan_job_cleanup_worker_removal(&request, &inv()) + .is_err(), + "{changed}" + ); + store + .with_conn(|connection| { + let count: i64 = connection.query_row( + "SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id='w'", + [], + |row| row.get(0), + )?; + assert_eq!( + count, 1, + "filtered canonical ownership must not become a new operation" + ); + let saved = load_plan(connection, &original.plan_id)?.unwrap(); + assert_eq!(saved.operation_id, original.operation_id); + assert_eq!(saved.input_fingerprint, original.input_fingerprint); + assert_ne!(saved.state, WorkerRemovalPlanState::Succeeded); + Ok(()) + }) + .unwrap(); + } + } + #[test] fn succeeded_worker_removal_recovers_after_registry_purge() { let s = setup(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 6e296276e..a913b11f0 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -2664,7 +2664,7 @@ impl WorkerRemovalService { target: &RuntimeWorkerRef, reason: &str, parent_workspace_deletion_operation_id: Option<&str>, - expected_removal: Option<(&str, &str)>, + job_cleanup_owner: Option<&Mutex>>, ) -> std::result::Result { let remove_lock = { let mut locks = self @@ -2693,6 +2693,15 @@ impl WorkerRemovalService { .store .recover_worker_removal_execution(&self.workspace_id, &target) .map_err(|_| "Worker removal recovery authority is unavailable".to_string())?; + let expected_removal = job_cleanup_owner + .map(|owner| { + owner + .lock() + .map(|saved| saved.clone()) + .map_err(|_| "Job cleanup child owner poisoned".to_string()) + }) + .transpose()? + .flatten(); if let Some((operation_id, fingerprint)) = expected_removal { if !prepared.as_ref().is_some_and(|saved| { saved.plan.workspace_id == self.workspace_id @@ -2708,6 +2717,28 @@ impl WorkerRemovalService { } } if let Some(prepared) = prepared { + if let Some(owner) = job_cleanup_owner { + if prepared.plan.workspace_id != self.workspace_id + || prepared.plan.worker != *target + || prepared.plan.operation_id.is_empty() + || prepared.plan.input_fingerprint.is_empty() + || prepared.runtime_request.workspace_id != self.workspace_id + || prepared.runtime_request.source_runtime_id != target.runtime_id + || prepared.runtime_request.worker_id.to_string() != target.worker_id + || prepared.runtime_request.operation_id != prepared.plan.operation_id + || prepared.runtime_request.input_fingerprint != prepared.plan.input_fingerprint + { + return Err("Job cleanup canonical removal binding mismatch".into()); + } + // Retain the first observed canonical child before any external + // effect or retry, including children created by this live owner. + *owner + .lock() + .map_err(|_| "Job cleanup child owner poisoned".to_string())? = Some(( + prepared.plan.operation_id.clone(), + prepared.plan.input_fingerprint.clone(), + )); + } if prepared.plan.state == crate::retention::WorkerRemovalPlanState::Succeeded { self.worker_projection .publish_ordered(|| { @@ -2835,10 +2866,21 @@ impl WorkerRemovalService { reason: reason.to_string(), journal_erasure, }; - let plan = match self.store.plan_worker_removal(&request, &inventory) { + let plan = match if job_cleanup_owner.is_some() { + self.store + .plan_job_cleanup_worker_removal(&request, &inventory) + } else { + self.store.plan_worker_removal(&request, &inventory) + } { Ok(plan) => plan, Err(error) => return Ok(worker_retention_error_response(error)), }; + if let Some(owner) = job_cleanup_owner { + *owner + .lock() + .map_err(|_| "Job cleanup child owner poisoned".to_string())? = + Some((plan.operation_id.clone(), plan.input_fingerprint.clone())); + } let prepared = match self.store.prepare_worker_removal_execution( &self.workspace_id, &plan.plan_id, @@ -5129,6 +5171,7 @@ impl WorkspaceApi { worker: worker.worker.clone(), runtime_run_id: runtime_run_id.clone(), }, + &self.config.database_path, )?; let newly_bound = self.worker_projection.publish_ordered(|| { self.store @@ -5414,7 +5457,8 @@ impl WorkspaceApi { async fn cleanup_backend_job_attempt_worker( &self, attempt: BackendJobAttemptRecord, - retained_removal: Option<(String, String)>, + removal_owner: &Mutex>, + drain: &crate::backend_job_host_effects::JobHostDrain, ) -> Result<()> { let history = self .store @@ -5507,7 +5551,7 @@ impl WorkspaceApi { } Err(error) => return fail("backend_job_cleanup_stop_unconfirmed", error.message()), } - if let Err(error) = self.fence_backend_job_effects(&attempt).await { + if let Err(error) = self.fence_backend_job_effects(&attempt, drain).await { return fail("backend_job_cleanup_effects_unconfirmed", error.to_string()); } } @@ -5518,9 +5562,7 @@ impl WorkspaceApi { &worker, &format!("Terminal Backend Job attempt {attempt_id}"), None, - retained_removal - .as_ref() - .map(|(id, fingerprint)| (id.as_str(), fingerprint.as_str())), + Some(removal_owner), ) .await .map_err(|message| Error::RuntimeOperationFailed { diff --git a/crates/workspace-server/src/server/backend_job_recovery.rs b/crates/workspace-server/src/server/backend_job_recovery.rs index 83ef2fc1e..dbc3a0158 100644 --- a/crates/workspace-server/src/server/backend_job_recovery.rs +++ b/crates/workspace-server/src/server/backend_job_recovery.rs @@ -3,7 +3,9 @@ //! Only initial dispatch opens Host admission; a reconstructed empty map proves nothing. use super::*; use crate::backend_job::BackendJobRecord; -use crate::backend_job_host_effects::{JobHostEffects, JobHostOwner, JobHostRequestLease}; +use crate::backend_job_host_effects::{ + JobHostDrain, JobHostEffects, JobHostOwner, JobHostRequestLease, +}; use feature_storage::FeatureStorageError; use std::collections::HashSet; use std::time::Duration; @@ -28,6 +30,12 @@ pub(super) struct RecoveryOwners { pub(super) nonjournal_lookup_pending: tokio::sync::Notify, #[cfg(test)] pub(super) cleanup_before_claim: Mutex>>, + #[cfg(test)] + pub(super) cleanup_before_writer_fence: tokio::sync::Notify, + #[cfg(test)] + pub(super) cleanup_after_claim: Mutex>>, + #[cfg(test)] + pub(super) cleanup_claimed_writer: tokio::sync::Notify, } impl RecoveryOwners { @@ -60,7 +68,11 @@ impl RecoveryOwners { .clone() } - pub(super) fn initially_admit(&self, owner: JobHostOwner) -> Result<()> { + pub(super) fn initially_admit( + &self, + owner: JobHostOwner, + database: &std::path::Path, + ) -> Result<()> { let mut hosts = self .hosts .lock() @@ -71,7 +83,10 @@ impl RecoveryOwners { None => { hosts.insert( owner.attempt_id.clone(), - (owner.clone(), JobHostEffects::initially_admit(owner)), + ( + owner.clone(), + JobHostEffects::initially_admit(owner, database)?, + ), ); Ok(()) } @@ -98,6 +113,19 @@ impl RecoveryOwners { .insert(owner.attempt_id.clone())) } + fn close_host_admission(&self, expected: &JobHostOwner) -> Result<()> { + let hosts = self + .hosts + .lock() + .map_err(|_| Error::Store("Job Host owners poisoned".into()))?; + if let Some((_, effects)) = hosts.get(&expected.attempt_id) { + effects.close_admission(expected)?; + } + // A missing live owner is not evidence. The original retained resource + // below must independently fence every accepted Host and cleanup writer. + Ok(()) + } + pub(super) fn effects(&self, expected: &JobHostOwner) -> Result> { self.hosts.lock().map_err(|_| Error::Store("Job Host owners poisoned".into()))? .get(&expected.attempt_id).filter(|(owner, _)| owner == expected) @@ -455,7 +483,6 @@ impl WorkspaceApi { &self, expected: BackendJobAttemptRecord, ) -> Result<()> { - let claimed = Arc::new(std::sync::atomic::AtomicBool::new(false)); // Pin the observed canonical operation across this owner's storage retries. let removal_owner = Arc::new(Mutex::new(None::<(String, String)>)); let mut failures = 0; @@ -465,13 +492,23 @@ impl WorkspaceApi { } let api = self.clone(); let owner = expected.clone(); - let mine = claimed.clone(); let removal_owner = removal_owner.clone(); let runtime = tokio::runtime::Handle::current(); let result = self .host_executors .cleanup()? .run(move || { + // 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 current = api .store .get_backend_job_attempt( @@ -480,7 +517,8 @@ impl WorkspaceApi { &owner.attempt_id, )? .ok_or_else(|| Error::Store("Cleanup owner disappeared".into()))?; - if current.worker != owner.worker + if !current.state.terminal() + || current.worker != owner.worker || current.runtime_run_id != owner.runtime_run_id || current.input_digest != owner.input_digest { @@ -498,40 +536,9 @@ impl WorkspaceApi { let attempt = if current.worker_cleanup_state == Some(BackendJobWorkerCleanupState::Executing) { - if !mine.load(Ordering::Acquire) { - // Executing is not a lease we can steal after restart. - // First require the exact durable execution/Host fence, - // then recover (never plan) its canonical removal owner. - api.require_backend_job_attempt_effects_ended(¤t)?; - let worker = current.worker.as_ref().ok_or_else(|| - Error::RepositoryConflict("Executing cleanup Worker missing".into()))?; - let saved = api.config_store.recover_worker_removal_execution( - api.workspace_id(), worker, - ).map_err(|error| Error::Store(error.to_string()))? - .ok_or_else(|| Error::RepositoryConflict( - "Executing cleanup has no retained canonical removal operation; ownership cannot be reconstructed".into()))?; - if saved.plan.workspace_id != api.workspace_id() - || saved.plan.worker != *worker - || saved.plan.operation_id.is_empty() - || saved.plan.input_fingerprint.is_empty() - || saved.runtime_request.workspace_id != api.workspace_id() - || saved.runtime_request.source_runtime_id != worker.runtime_id - || saved.runtime_request.worker_id.to_string() != worker.worker_id - || saved.runtime_request.operation_id != saved.plan.operation_id - || saved.runtime_request.input_fingerprint != saved.plan.input_fingerprint - { - return Err(Error::RepositoryConflict( - "Executing cleanup canonical removal owner mismatch".into())); - } - let expected = (saved.plan.operation_id, saved.plan.input_fingerprint); - let mut pinned = removal_owner.lock().map_err(|_| Error::Store( - "Cleanup removal owner poisoned".into()))?; - if pinned.as_ref().is_some_and(|prior| prior != &expected) { - return Err(Error::RepositoryConflict( - "Executing cleanup canonical removal owner changed".into())); - } - *pinned = Some(expected); - } + // Exclusive original-writer fencing authorizes recovery, + // not the Executing marker. Canonical child continuity is + // checked under the removal service's own operation lock. current } else { let history = api @@ -549,19 +556,29 @@ impl WorkspaceApi { "Cleanup belongs to another admitted owner".into(), )); } - mine.store(true, Ordering::Release); attempt }; - let retained_removal = removal_owner.lock().map_err(|_| Error::Store( - "Cleanup removal owner poisoned".into()))?.clone(); + #[cfg(test)] + let after_claim = api + .backend_job_recovery + .cleanup_after_claim + .lock() + .unwrap() + .clone(); runtime.block_on(async { + #[cfg(test)] + if let Some(gate) = after_claim { + api.backend_job_recovery.cleanup_claimed_writer.notify_one(); + gate.acquire().await.unwrap().forget(); + } if attempt.state == BackendJobAttemptState::Completed || attempt.failure_category.as_deref() == Some("invalid_result") { api.wait_for_backend_job_worker_run_teardown(&attempt) .await?; } - api.cleanup_backend_job_attempt_worker(attempt, retained_removal).await + api.cleanup_backend_job_attempt_worker(attempt, &removal_owner, &drain) + .await }) }) .await; @@ -1231,7 +1248,13 @@ impl WorkspaceApi { pub(super) async fn fence_backend_job_effects( &self, attempt: &BackendJobAttemptRecord, + drain: &JobHostDrain, ) -> Result<()> { + if drain.owner() != &JobHostOwner::try_from(attempt)? { + return Err(Error::RepositoryConflict( + "Job cleanup admission owner changed".into(), + )); + } if self .store .has_backend_job_execution_fence(self.workspace_id(), attempt)? @@ -1252,15 +1275,15 @@ impl WorkspaceApi { || termination.scope != worker_runtime::execution::WorkerExecutionTerminationScope::Controller || termination.worker_ref.worker_id.to_string() != worker.worker_id + || !termination.submissions.iter().any(|submission| { + Some(&submission.submission_request_id) == attempt.runtime_run_id.as_ref() + }) { return Err(Error::RepositoryConflict( "Runtime receipt does not prove exact Controller/Feature/background termination" .into(), )); } - let owner = JobHostOwner::try_from(attempt)?; - let effects = self.backend_job_recovery.effects(&owner)?; - let drain = effects.close_and_drain(&owner).await?; self.store.record_backend_job_execution_fence( self.workspace_id(), attempt, @@ -1321,9 +1344,31 @@ pub(super) async fn admit_host_request( .get_backend_job_attempt(lookup.workspace_id(), &job.request.job_id, &id)? .ok_or_else(|| Error::Store("Job Host attempt missing".into()))?; let owner = JobHostOwner::try_from(&attempt)?; - Ok(Some( - lookup.backend_job_recovery.effects(&owner)?.admit(&owner)?, - )) + let lease = lookup.backend_job_recovery.effects(&owner)?.admit(&owner)?; + // Close the stale-read window: cleanup's exclusive admission fence and + // terminal row must preclude every later callback, including another Host. + let current_job = lookup + .store + .get_active_backend_job_for_worker(lookup.workspace_id(), &worker)? + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "Job admission closed before callback acquisition".into(), + ) + })?; + let current = lookup + .store + .get_backend_job_attempt(lookup.workspace_id(), &job.request.job_id, &id)? + .ok_or_else(|| Error::Store("Job Host attempt disappeared".into()))?; + if current_job.request.job_id != owner.job_id + || current_job.current_attempt != current.attempt + || current.state != BackendJobAttemptState::Dispatched + || JobHostOwner::try_from(¤t)? != owner + { + return Err(Error::WorkspacePermissionDenied( + "Job callback owner changed during admission".into(), + )); + } + Ok(Some(lease)) }) .await; match admission { 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 2d11d5c70..828b8b816 100644 --- a/crates/workspace-server/src/server/tests/backend_job_tests.rs +++ b/crates/workspace-server/src/server/tests/backend_job_tests.rs @@ -194,18 +194,232 @@ async fn backend_job_lost_allocation_response_binds_original_receipt_without_hos } #[tokio::test] -async fn backend_job_executing_cleanup_restart_requires_fence_and_original_removal() { +async fn backend_job_executing_cleanup_restart_fences_original_writers_at_each_cut() { + for cut in [ + "claimed", + "fenced", + "removed", + "missing-resource", + "wrong-resource", + ] { + let temp = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let request = recovery_request(&format!("executing-cleanup-{cut}")); + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let worker = dispatched.attempt.worker.clone().unwrap(); + assert_eq!(execution.take_inputs().len(), 1); + seed_worker_session_for_cleanup(temp.path(), &worker); + // Quiesce only fixture recovery tasks. The substituted execution boundary + // and real Host admission resource remain; this is not a process-exit test. + api.stop_backend_job_recovery().await; + let current = api + .store + .finish_backend_job_attempt( + api.workspace_id(), + &request.job_id, + &dispatched.attempt.attempt_id, + BackendJobAttemptState::Failed, + "fixture_failure", + "retained original failure", + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + let history = api + .store + .list_backend_job_attempts(api.workspace_id(), &request.job_id) + .unwrap(); + let (claimed, acquired) = api + .store + .claim_backend_job_worker_cleanup_owned( + api.workspace_id(), + &history, + ¤t, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + assert!(acquired); + assert!( + api.require_backend_job_attempt_effects_ended(&claimed) + .is_err() + ); + assert!( + api.config_store + .recover_worker_removal_execution(api.workspace_id(), &worker) + .unwrap() + .is_none() + ); + // A claimed marker cannot fabricate a missing canonical child. + assert!( + WorkerRemovalService::new(&api) + .execute_target_removal_on_host_checked( + api.runtime.as_ref(), + &worker, + "must not replace missing child", + None, + Some(&Mutex::new(Some(( + "missing-operation".into(), + "missing-fingerprint".into() + )))), + ) + .await + .is_err() + ); + if matches!(cut, "missing-resource" | "wrong-resource") { + let mut path = api.config.database_path.as_os_str().to_owned(); + path.push(".job-host-admission"); + let path = std::path::PathBuf::from(path); + if cut == "missing-resource" { + std::fs::remove_dir_all(&path).unwrap(); + } else { + let file = std::fs::read_dir(&path) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + let mut owner = serde_json::to_value( + crate::backend_job_host_effects::JobHostOwner::try_from(&claimed).unwrap(), + ) + .unwrap(); + owner["runtime_run_id"] = "successor-run".into(); + std::fs::write(file, serde_json::to_vec(&owner).unwrap()).unwrap(); + } + let mut restarted = api.clone(); + restarted.backend_job_recovery = Arc::default(); + let error = restarted + .run_backend_job_cleanup(claimed.clone()) + .await + .unwrap_err(); + assert!( + error.to_string().contains("Original Job Host admission"), + "{cut}: {error}" + ); + assert!( + api.require_backend_job_attempt_effects_ended(&claimed) + .is_err() + ); + assert!( + api.config_store + .recover_worker_removal_execution(api.workspace_id(), &worker) + .unwrap() + .is_none() + ); + assert!(execution.take_inputs().is_empty()); + if cut == "missing-resource" { + assert!(!path.exists()); + } + continue; + } + let original = if cut != "claimed" { + let drain = crate::backend_job_host_effects::JobHostEffects::fence_retained( + &api.config.database_path, + &claimed, + ) + .unwrap(); + api.runtime + .stop_worker( + &worker, + WorkerLifecycleRequest { + stop_intent: None, + reason: Some("fixture exact execution termination".into()), + ticket_assignment: None, + }, + ) + .unwrap(); + api.fence_backend_job_effects(&claimed, &drain) + .await + .unwrap(); + assert!( + api.require_backend_job_attempt_effects_ended(&claimed) + .is_ok() + ); + if cut == "removed" { + let response = WorkerRemovalService::new(&api) + .execute_target_removal_on_host( + api.runtime.as_ref(), + &worker, + &format!("Terminal Backend Job attempt {}", claimed.attempt_id), + None, + ) + .await + .unwrap(); + assert_eq!(response.status, StatusCode::OK.as_u16()); + } + api.config_store + .recover_worker_removal_execution(api.workspace_id(), &worker) + .unwrap() + } else { + None + }; + let mut restarted = api.clone(); + restarted.backend_job_recovery = Arc::default(); + let owner = crate::backend_job_host_effects::JobHostOwner::try_from(&claimed).unwrap(); + assert!(restarted.backend_job_recovery.effects(&owner).is_err()); + restarted.recover_backend_job_worker_cleanups().unwrap(); + let cleaned = wait_for_backend_job_worker_cleanup( + &restarted, + &request.job_id, + &claimed.attempt_id, + BackendJobWorkerCleanupState::Completed, + ) + .await; + assert_eq!( + cleaned.failure_category.as_deref(), + Some("fixture_failure"), + "{cut}" + ); + assert_eq!( + cleaned.failure_detail.as_deref(), + Some("retained original failure") + ); + assert_eq!(cleaned.worker, claimed.worker); + assert_eq!(cleaned.runtime_run_id, claimed.runtime_run_id); + assert_eq!(cleaned.input_digest, claimed.input_digest); + restarted + .require_backend_job_attempt_effects_ended(&cleaned) + .unwrap(); + let retained = api + .config_store + .recover_worker_removal_execution(api.workspace_id(), &worker) + .unwrap() + .unwrap(); + assert_eq!( + retained.plan.state, + crate::retention::WorkerRemovalPlanState::Succeeded + ); + if let Some(original) = original { + assert_eq!(retained.plan.operation_id, original.plan.operation_id); + assert_eq!( + retained.plan.input_fingerprint, + original.plan.input_fingerprint + ); + } + assert_eq!( + api.store + .list_backend_job_attempts(api.workspace_id(), &request.job_id) + .unwrap() + .len(), + 1 + ); + assert!(execution.take_inputs().is_empty()); + assert!(api.runtime.worker(&worker).is_err()); + restarted.stop_backend_job_recovery().await; + } +} + +#[tokio::test] +async fn job_cleanup_policy_staleness_retains_the_first_child_across_retry_and_new_owner() { + use crate::retention::{WorkerRemovalPlanRequest, WorkerRetentionPolicyUpdate}; let temp = tempfile::tempdir().unwrap(); let (api, execution) = test_api_with_recording_backend(temp.path()).await; - let request = recovery_request("executing-cleanup-restart"); + let request = recovery_request("stale-cleanup-child"); let dispatched = api.dispatch_backend_job(&request).unwrap(); let worker = dispatched.attempt.worker.clone().unwrap(); execution.take_inputs(); seed_worker_session_for_cleanup(temp.path(), &worker); - // Stop only the recovery tasks, not execution. Inject a crash boundary after - // cleanup claim without a live cleanup task. This is not a process-exit test. api.stop_backend_job_recovery().await; - api.store + let failed = api + .store .finish_backend_job_attempt( api.workspace_id(), &request.job_id, @@ -216,146 +430,250 @@ async fn backend_job_executing_cleanup_restart_requires_fence_and_original_remov &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), ) .unwrap(); - let current = api - .store - .get_backend_job_attempt( - api.workspace_id(), - &request.job_id, - &dispatched.attempt.attempt_id, - ) - .unwrap() - .unwrap(); - let history = api - .store - .list_backend_job_attempts(api.workspace_id(), &request.job_id) - .unwrap(); - let (claimed, acquired) = api - .store - .claim_backend_job_worker_cleanup_owned( - api.workspace_id(), - &history, - ¤t, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ) - .unwrap(); - assert!(acquired); - let mut restarted = api.clone(); - restarted.backend_job_recovery = Arc::default(); - let error = restarted - .run_backend_job_cleanup(claimed.clone()) - .await - .unwrap_err(); - assert!( - error - .to_string() - .contains("Host writer/drain evidence missing") - ); - assert!( - api.config_store - .recover_worker_removal_execution(api.workspace_id(), &worker) - .unwrap() - .is_none() - ); - + let drain = crate::backend_job_host_effects::JobHostEffects::fence_retained( + &api.config.database_path, + &failed, + ) + .unwrap(); api.runtime .stop_worker( &worker, WorkerLifecycleRequest { stop_intent: None, - reason: Some("fixture exact execution termination".into()), + reason: Some("fixture original scope join".into()), ticket_assignment: None, }, ) .unwrap(); - api.fence_backend_job_effects(&claimed).await.unwrap(); - let error = restarted - .run_backend_job_cleanup(claimed.clone()) + api.fence_backend_job_effects(&failed, &drain) .await - .unwrap_err(); - assert!( - error - .to_string() - .contains("no retained canonical removal operation") - ); - assert!( - api.config_store - .recover_worker_removal_execution(api.workspace_id(), &worker) - .unwrap() - .is_none(), - "a restarted Executing marker cannot authorize a new removal" - ); + .unwrap(); + let reason = format!("Terminal Backend Job attempt {}", failed.attempt_id); + let plan = api + .config_store + .plan_job_cleanup_worker_removal( + &WorkerRemovalPlanRequest { + workspace_id: api.workspace_id().into(), + worker: worker.clone(), + reason: reason.clone(), + journal_erasure: None, + }, + &api.runtime.worker_retention_inventory(&worker).unwrap(), + ) + .unwrap(); + api.config_store + .prepare_worker_removal_execution( + api.workspace_id(), + &plan.plan_id, + &plan.input_fingerprint, + ) + .unwrap(); + // Seed an uncertain canonical child, not a successful provider response. + api.config_store + .fail_worker_removal( + api.workspace_id(), + &plan.operation_id, + &plan.input_fingerprint, + "fixture_runtime_outcome_unconfirmed", + ) + .unwrap(); + let policy = api + .config_store + .worker_retention_policy(api.workspace_id()) + .unwrap() + .unwrap(); + api.config_store + .update_worker_retention_policy( + api.workspace_id(), + &policy.policy_digest, + &WorkerRetentionPolicyUpdate { + policy_id: "changed-after-uncertain-job-child".into(), + session_disposition: policy.session_disposition, + metadata_disposition: policy.metadata_disposition, + archive_retention: policy.archive_retention, + diagnostics_disposition: policy.diagnostics_disposition, + diagnostics_retention_seconds: policy.diagnostics_retention_seconds, + }, + ) + .unwrap(); + let child = Mutex::new(None); + let service = WorkerRemovalService::new(&api); + let response = service + .execute_target_removal_on_host_checked( + api.runtime.as_ref(), + &worker, + &reason, + None, + Some(&child), + ) + .await + .unwrap(); + assert_ne!(response.status, StatusCode::OK.as_u16()); assert_eq!( - api.store - .get_backend_job_attempt(api.workspace_id(), &request.job_id, &claimed.attempt_id) - .unwrap() - .unwrap() - .worker_cleanup_state, - Some(BackendJobWorkerCleanupState::Executing) + *child.lock().unwrap(), + Some((plan.operation_id.clone(), plan.input_fingerprint.clone())) ); - - // Checked restart entry cannot fall through to planning without the pinned - // original operation, even though execution/Host fences now exist. assert!( - WorkerRemovalService::new(&api) + service .execute_target_removal_on_host_checked( api.runtime.as_ref(), &worker, - "must not create a replacement", + &reason, None, - Some(("missing-operation", "missing-fingerprint")) + Some(&child) ) .await .is_err() ); + // A fresh recovery owner's empty memory still cannot treat filtered Stale as + // proof that no canonical child was ever created. + let cold = Mutex::new(None); + let response = service + .execute_target_removal_on_host_checked( + api.runtime.as_ref(), + &worker, + &reason, + None, + Some(&cold), + ) + .await + .unwrap(); + assert_ne!(response.status, StatusCode::OK.as_u16()); + assert!(cold.lock().unwrap().is_none()); + api.config_store.with_conn(|connection| { + let count: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![api.workspace_id(), worker.runtime_id, worker.worker_id], |row| row.get(0))?; + assert_eq!(count, 1); + Ok(()) + }).unwrap(); + assert!(api.runtime.worker(&worker).is_ok()); + assert!(execution.take_inputs().is_empty()); + assert_eq!( + api.store + .get_backend_job(api.workspace_id(), &request.job_id) + .unwrap() + .unwrap() + .failure_category + .as_deref(), + Some("fixture_failure") + ); +} + +#[tokio::test] +async fn cancelled_cleanup_waiter_keeps_its_actual_writer_exclusive_across_recovery_owners() { + let temp = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let request = recovery_request("cancelled-cleanup-writer"); + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let worker = dispatched.attempt.worker.clone().unwrap(); + assert_eq!(execution.take_inputs().len(), 1); + seed_worker_session_for_cleanup(temp.path(), &worker); + api.stop_backend_job_recovery().await; + let failed = api + .store + .finish_backend_job_attempt( + api.workspace_id(), + &request.job_id, + &dispatched.attempt.attempt_id, + BackendJobAttemptState::Failed, + "fixture_failure", + "retain original failure", + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + ) + .unwrap(); + let mut original = api.clone(); + original.backend_job_recovery = Arc::default(); + let hold = Arc::new(tokio::sync::Semaphore::new(0)); + *original + .backend_job_recovery + .cleanup_after_claim + .lock() + .unwrap() = Some(hold.clone()); + let original_api = original.clone(); + let expected = failed.clone(); + let waiter = tokio::spawn(async move { original_api.run_backend_job_cleanup(expected).await }); + tokio::time::timeout( + std::time::Duration::from_secs(5), + original + .backend_job_recovery + .cleanup_claimed_writer + .notified(), + ) + .await + .unwrap(); + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + // The real cleanup-lane closure is still gated after claim. Its exclusive + // resource must survive the cancelled async response waiter independently. + let mut directory = api.config.database_path.as_os_str().to_owned(); + directory.push(".job-host-admission"); + let path = std::fs::read_dir(std::path::PathBuf::from(directory)) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + let competing_writer = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + .unwrap(); + assert!( + competing_writer.try_lock().is_err(), + "actual predecessor cleanup lost its writer lease" + ); + let mut recovered = api.clone(); + recovered.backend_job_recovery = Arc::default(); + recovered.recover_backend_job_worker_cleanups().unwrap(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + recovered + .backend_job_recovery + .cleanup_before_writer_fence + .notified(), + ) + .await + .unwrap(); + let pending = api + .store + .get_backend_job_attempt(api.workspace_id(), &request.job_id, &failed.attempt_id) + .unwrap() + .unwrap(); + assert_eq!( + pending.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Executing) + ); + assert!( + api.require_backend_job_attempt_effects_ended(&pending) + .is_err() + ); assert!( api.config_store .recover_worker_removal_execution(api.workspace_id(), &worker) .unwrap() .is_none() ); - - // The original owner performs canonical removal, then loses its cleanup save. - let response = WorkerRemovalService::new(&api) - .execute_target_removal_on_host( - api.runtime.as_ref(), - &worker, - &format!("Terminal Backend Job attempt {}", claimed.attempt_id), - None, - ) - .await - .unwrap(); - assert_eq!(response.status, StatusCode::OK.as_u16()); - let original = api - .config_store - .recover_worker_removal_execution(api.workspace_id(), &worker) - .unwrap() - .unwrap(); - assert_eq!( - original.plan.state, - crate::retention::WorkerRemovalPlanState::Succeeded - ); - restarted.recover_backend_job_worker_cleanups().unwrap(); + hold.add_permits(1); let cleaned = wait_for_backend_job_worker_cleanup( - &restarted, + &recovered, &request.job_id, - &claimed.attempt_id, + &failed.attempt_id, BackendJobWorkerCleanupState::Completed, ) .await; - assert_eq!(cleaned.failure_category.as_deref(), Some("fixture_failure")); - let retained = api - .config_store - .recover_worker_removal_execution(api.workspace_id(), &worker) - .unwrap() + assert_eq!(cleaned.worker, failed.worker); + assert_eq!(cleaned.runtime_run_id, failed.runtime_run_id); + assert_eq!(cleaned.failure_category, failed.failure_category); + recovered + .require_backend_job_attempt_effects_ended(&cleaned) .unwrap(); - assert_eq!(retained.plan.operation_id, original.plan.operation_id); - assert_eq!( - retained.plan.input_fingerprint, - original.plan.input_fingerprint - ); + api.config_store.with_conn(|connection| { + let count: i64 = connection.query_row("SELECT COUNT(*) FROM worker_removal_operations WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3", rusqlite::params![api.workspace_id(), worker.runtime_id, worker.worker_id], |row| row.get(0))?; + assert_eq!(count, 1, "both admitted cleanup callers must converge on one canonical child"); + Ok(()) + }).unwrap(); assert!(execution.take_inputs().is_empty()); assert!(api.runtime.worker(&worker).is_err()); - restarted.stop_backend_job_recovery().await; + recovered.stop_backend_job_recovery().await; } #[tokio::test] @@ -417,7 +735,7 @@ async fn backend_job_definitive_input_rejection_is_not_receipt_ambiguity() { #[tokio::test] async fn backend_job_cancelled_callback_keeps_actual_host_db_effect_owned_until_drain() { let temp = tempfile::tempdir().unwrap(); - let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let (api, _execution) = test_api_with_recording_backend(temp.path()).await; let request = recovery_request("host-drain"); let reservation = api.dispatch_backend_job(&request).unwrap(); let owner = @@ -489,8 +807,7 @@ async fn backend_job_cancelled_callback_keeps_actual_host_db_effect_owned_until_ // closure is real and still blocked despite its cancelled async waiter. tokio::time::timeout(std::time::Duration::from_secs(5), async { loop { - if !execution.terminations.lock().unwrap().is_empty() && effects.admit(&owner).is_err() - { + if effects.admit(&owner).is_err() { break; } tokio::task::yield_now().await; diff --git a/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs index 9fad3eef8..f6fa143ed 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_jobs_tests.rs @@ -476,6 +476,14 @@ async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_o ); let job = resource(&api, &subject.id); let attempt = current_attempt(&api, &job); + // This test exercises live-grant operations before result reconciliation. + // The owner may otherwise complete the ready domain surface during an await. + let recovery_gate = api.backend_job_recovery.dispatch_gate(&attempt.attempt_id); + let recovery_guard = recovery_gate.lock().unwrap(); + assert_eq!( + current_attempt(&api, &job).state, + BackendJobAttemptState::Dispatched + ); let worker = attempt.worker.as_ref().unwrap(); assert!( store @@ -541,6 +549,7 @@ async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_o result: serde_json::json!({"subject_id":subject.id,"candidate_ids":[],"surface":{"availability":"ready","generation_id":generation.id,"memory_fingerprint":generation.memory_fingerprint,"snapshot_id":snapshot.id}}), }; api.accept_backend_job_result(worker, &submission).unwrap(); + drop(recovery_guard); wait_for_backend_job_worker_cleanup( &api, &job.request.job_id, 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 d008b6667..48ce064b8 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 @@ -451,9 +451,10 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() worker_id: old.worker_id.clone(), binding_id: old_host.writer_binding_id().unwrap().into(), }; - // Inject a durable result-acceptance gap after real Journal completion. This must - // reject before writer revocation and confirm cancellation of the actual Controller - // guard; merely returning Err would strand the body and its execution-admission fence. + // Corrupt only the saved Job projection after real completion, deliberately + // leaving its attempt/checkpoint completed. This is a fail-closed consumer + // contrast, not a simulated transaction interruption or a recovery mechanism. + // The pre-revoke refusal must cancel the actual Controller guard before retry. let recorded = api .store .list_backend_journal_jobs(TEST_WORKSPACE_ID, &old_attr.session_id, None, 100) @@ -488,7 +489,7 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() rusqlite::params![TEST_WORKSPACE_ID, uncertain_job], |r| Ok((r.get(0)?, r.get(1)?)), ).unwrap(); conn.execute( - "UPDATE backend_jobs SET state='unknown',result_json=NULL,result_digest=NULL WHERE workspace_id=?1 AND job_id=?2", + "UPDATE backend_jobs SET state='pending',result_json=NULL,result_digest=NULL WHERE workspace_id=?1 AND job_id=?2", rusqlite::params![TEST_WORKSPACE_ID, uncertain_job], ) .unwrap(); @@ -500,7 +501,7 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() serde_json::from_value(incomplete.clone()).unwrap(), ) .unwrap_err(); - assert!(error.to_string().contains("Unknown"), "{error}"); + assert!(error.to_string().contains("still pending"), "{error}"); assert!(store.journal_writer_active(&old_binding).unwrap()); assert!( !api.config_store diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index bb373ac61..16b379175 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -1790,8 +1790,26 @@ async fn migration_stopped_checkpoint_is_excluded_from_recovery_result_acceptanc async fn saved_checkpoint_never_promotes_erased_released_retired_or_missing_source_to_success() { for unavailable in ["erased", "body_released", "retired", "deleted_worker"] { let f = RuntimeFixture::new().await; + // Join only this fixture's background recovery before constructing the cut. + // This orders checkpoint/unavailability; it is not execution-ended evidence. + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api.stop_backend_job_recovery(), + ) + .await + .expect("fixture recovery must quiesce before the unavailable-source cut"); let dispatched = f.dispatch_input(); + let gate = f + .api + .backend_job_recovery + .dispatch_gate(&dispatched.attempt.attempt_id); + let dispatch_finished = gate.lock().unwrap(); let worker = dispatched.attempt.worker.as_ref().unwrap(); + let generation = f + .store + .active_journal_generation(&f.host, &f.id) + .unwrap() + .unwrap(); f.checkpoint(worker); let completion = subjektiv_journal::finish(&f.api, worker).unwrap(); let checkpoint = f @@ -1830,9 +1848,53 @@ async fn saved_checkpoint_never_promotes_erased_released_retired_or_missing_sour } let job = current_journal_job(&f); let attempt = current_journal_attempt(&f, &dispatched.attempt.attempt_id); + let registry = f + .api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.source) + .unwrap(); + // Read-only evidence remains observable after writer release; the public + // active-generation API correctly refuses that released writer's authority. + let generation_evidence = || { + f.api + .feature_storage + .open(&f.api.subjektiv_registration) + .unwrap() + .with_connection(|connection| { + use rusqlite::OptionalExtension as _; + Ok(connection.query_row( + "SELECT json_object('snapshot',json(snapshot_json),'active',active,'completion',json(completion_json)) FROM journal_generations WHERE id=?1", + [&generation.id], |row| row.get::<_, String>(0), + ).optional()?) + }) + .unwrap() + }; + let retained = generation_evidence(); + if unavailable == "erased" { + assert!( + retained.is_none(), + "explicit erasure removes generation evidence" + ); + } else { + let evidence: Value = serde_json::from_str(retained.as_ref().unwrap()).unwrap(); + assert_eq!( + evidence["snapshot"], + serde_json::to_value(&generation).unwrap() + ); + assert_eq!( + evidence["active"], 0, + "saved domain completion closes only the generation" + ); + assert_eq!( + evidence["completion"], + serde_json::to_value(&completion).unwrap() + ); + } f.execution.take_inputs(); subjektiv_journal::reconcile_attempt(&f.api, &job, &attempt).unwrap(); + drop(dispatch_finished); let failed = current_journal_job(&f); + let failed_attempt = current_journal_attempt(&f, &attempt.attempt_id); assert_eq!(failed.state, BackendJobState::Failed, "{unavailable}"); assert_eq!( failed.failure_category.as_deref(), @@ -1847,6 +1909,11 @@ async fn saved_checkpoint_never_promotes_erased_released_retired_or_missing_sour assert!(failed.result.is_none()); assert_eq!(failed.current_attempt, attempt.attempt); assert_eq!(failed.input_digest, attempt.input_digest); + assert_eq!(failed.input_digest, f.request.input_digest().unwrap()); + assert_eq!(failed.intent_fingerprint, f.request.fingerprint().unwrap()); + if unavailable != "retired" { + assert_eq!(failed.load_executable_request().unwrap(), &f.request); + } assert!( f.api .accept_backend_job_result( @@ -1860,10 +1927,56 @@ async fn saved_checkpoint_never_promotes_erased_released_retired_or_missing_sour ) .is_err() ); - assert!( + // The stale caller cannot execute. Available terminal input returns only + // its saved Failed observation; explicitly retired input remains rejected. + let replay = f.api.dispatch_backend_job_reservation(dispatched.clone()); + if unavailable == "retired" { + assert!(matches!(replay, Err(Error::WorkspacePermissionDenied(_)))); + } else { + let replay = replay.unwrap(); + assert!(replay.replayed, "{unavailable}"); + assert_eq!(replay.job, failed, "{unavailable}"); + assert_eq!(replay.attempt, failed_attempt, "{unavailable}"); + } + assert_eq!( + current_journal_job(&f), + failed, + "failure, input and retirement must survive replay: {unavailable}" + ); + assert_eq!( + current_journal_attempt(&f, &attempt.attempt_id), + failed_attempt + ); + assert_eq!(failed.request, job.request); + assert_eq!(failed.intent_fingerprint, job.intent_fingerprint); + assert_eq!(failed.input_disposition, job.input_disposition); + assert_eq!(failed_attempt.state, BackendJobAttemptState::Failed); + assert_eq!(failed_attempt.worker, attempt.worker); + assert_eq!(failed_attempt.runtime_run_id, attempt.runtime_run_id); + assert_eq!(failed_attempt.failure_category, failed.failure_category); + assert!(failed_attempt.result.is_none()); + assert_eq!(generation_evidence(), retained); + assert_eq!( + f.store + .journal_worker_was_released( + &f.request + .grants + .subjektiv_journal + .as_ref() + .unwrap() + .subject_id, + &f.source.runtime_id, + &f.source.worker_id, + ) + .unwrap(), + unavailable == "body_released" + ); + assert_eq!( f.api - .dispatch_backend_job_reservation(dispatched.clone()) - .is_err() + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.source) + .unwrap(), + registry ); assert!( f.execution.take_inputs().is_empty(), @@ -1881,6 +1994,7 @@ async fn saved_checkpoint_never_promotes_erased_released_retired_or_missing_sour } if unavailable == "retired" { assert!(failed.load_executable_request().is_err()); + assert!(failed.request.input.is_null()); assert_eq!( failed.input_disposition, crate::backend_job::JobInputDisposition::Invalidated @@ -2451,6 +2565,7 @@ async fn explicit_erasure_interrupts_completed_journal_cleanup_waiting_for_run_t struct SubjectRemovalFixture { dir: tempfile::TempDir, api: WorkspaceApi, + execution: Arc, dispatcher: Arc, controller: RuntimeWorkerRef, target: RuntimeWorkerRef, @@ -2475,10 +2590,11 @@ impl SubjectRemovalFixture { control.clone(), ), ); + let execution = Arc::new(DeterministicExecutionBackend::default()); let api = WorkspaceApi::new_with_execution_backend_and_broker( config, control, - Arc::new(DeterministicExecutionBackend::default()), + execution.clone(), BackendResourceBroker::default(), HostExecutors::default(), Some(dispatcher.clone()), @@ -2595,6 +2711,7 @@ impl SubjectRemovalFixture { Self { dir, api, + execution, dispatcher, controller, target, @@ -3636,7 +3753,7 @@ async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a } #[tokio::test] -async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { +async fn disabled_subject_removal_defers_unreflected_prefix_without_admitting_new_history() { let f = SubjectRemovalFixture::new().await; let runtime_api::SessionMaterialCatalogAvailability::Page { page } = f .api @@ -3684,20 +3801,41 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { .store .accept_journal_sources(&f.host, "accepted before disablement", &ranges[..1]) .unwrap(); + let admitted = f + .store + .list_journal_sources(&f.host, &id, None, 32) + .unwrap(); + assert_eq!(admitted.items.len(), 1); + let before = f.store.read_journal(&f.host, &id).unwrap(); + let source_before = subjektiv_journal::capture_input(&f.api, &f.attr, 0, 0, 48 * 1024).unwrap(); f.disable(); + let registry = f + .api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap() + .unwrap(); let response = f.remove(); assert_eq!( response.status, - StatusCode::OK.as_u16(), + StatusCode::CONFLICT.as_u16(), "{}", response.body ); + let body: Value = serde_json::from_str(&response.body).unwrap(); + assert_eq!(body["code"], "journal_handoff_incomplete"); assert!( + body["message"] + .as_str() + .unwrap() + .contains("unreflected input") + ); + assert_eq!( f.api .store .get_worker_registry(TEST_WORKSPACE_ID, &f.target) - .unwrap() - .is_none() + .unwrap(), + Some(registry) ); assert!( f.api @@ -3710,9 +3848,66 @@ async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() { f.store.journal_for_session(&f.host).unwrap().as_deref(), Some(id.as_str()) ); + assert_eq!(f.store.read_journal(&f.host, &id).unwrap(), before); + assert_eq!( + f.store + .list_journal_sources(&f.host, &id, None, 32) + .unwrap(), + admitted + ); + assert!( + f.store + .read_journal_source(&f.host, &id, ranges[1].source.order) + .is_err() + ); assert!(f.store.journal_retention_required(&f.host).unwrap()); - assert!(!session_root(f.dir.path(), &f.target).exists()); - f.assert_removed_with_history(); + assert!(!f.store.journal_session_erased(&f.host).unwrap()); + assert!(session_root(f.dir.path(), &f.target).exists()); + let runtime_api::SessionRecordingCaptureAvailability::Captured { slice } = f + .api + .runtime + .session_recording_capture( + &f.target.runtime_id, + runtime_api::SessionRecordingCaptureRequest { + workspace_id: TEST_WORKSPACE_ID.into(), + source: runtime_api::SessionPublicSource::Retained { + worker_id: worker_runtime::identity::WorkerId::parse(&f.target.worker_id) + .unwrap(), + }, + expected_session_id: f.attr.session_id.clone(), + start: session_store::capture::RecordingPosition { + canonical_order: 0, + byte_offset: 0, + }, + max_entries: MAX_JOURNAL_RANGES, + max_serialized_bytes: 48 * 1024, + }, + ) + .unwrap() + else { + panic!("ordinary removal must retain the readable original history") + }; + assert_eq!( + slice.capture, source_before, + "ordinary removal must retain both the admitted and unadmitted original history" + ); + assert!( + f.api + .config_store + .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id) + .unwrap() + .is_none() + ); + assert!( + !f.store + .journal_worker_was_released( + &f.attr.subject_id, + &f.target.runtime_id, + &f.target.worker_id, + ) + .unwrap() + ); + f.assert_occupied(); } #[tokio::test] @@ -4006,6 +4201,7 @@ impl SubjectRemovalFixture { let Self { dir, api, + execution, dispatcher, controller, target, @@ -4015,6 +4211,7 @@ impl SubjectRemovalFixture { } = self; let config = api.config.clone(); drop(api); + drop(execution); drop(dispatcher); drop(store); let control = Arc::new(test_control_store(&config)); @@ -4024,10 +4221,11 @@ impl SubjectRemovalFixture { control.clone(), ), ); + let execution = Arc::new(DeterministicExecutionBackend::default()); let api = WorkspaceApi::new_with_execution_backend_and_broker( config, control, - Arc::new(DeterministicExecutionBackend::default()), + execution.clone(), BackendResourceBroker::default(), HostExecutors::default(), Some(dispatcher.clone()), @@ -4038,6 +4236,7 @@ impl SubjectRemovalFixture { Self { dir, api, + execution, dispatcher, controller, target, @@ -4527,7 +4726,7 @@ journal_recovery_cases! { } #[tokio::test] -async fn legacy_journal_grant_preserves_old_json_and_only_recovers_original_current_body() { +async fn legacy_journal_grant_preserves_wire_digest_but_authorizes_neither_writer() { let f = SubjectRemovalFixture::new().await; let old = json!({"subject_id":f.attr.subject_id,"session_id":f.attr.session_id,"journal_id":"legacy-journal","selected_ranges":[]}); let wire = serde_json::to_string(&old).unwrap(); @@ -4538,11 +4737,22 @@ async fn legacy_journal_grant_preserves_old_json_and_only_recovers_original_curr old, "old canonical request/digest must not acquire a new empty binding_id field" ); + let roundtrip = serde_json::to_string(&serde_json::to_value(&grant).unwrap()).unwrap(); + assert_eq!(roundtrip, wire); assert_eq!( - serde_json::to_string(&serde_json::to_value(&grant).unwrap()).unwrap(), - wire + job::sha256(roundtrip.as_bytes()), + job::sha256(wire.as_bytes()) + ); + let mut bound = grant.clone(); + bound.binding_id = f.host.writer_binding_id().unwrap().into(); + assert!(subjektiv_journal::host(&f.api, &f.store, &bound).is_ok()); + assert!( + matches!( + subjektiv_journal::host(&f.api, &f.store, &grant), + Err(Error::WorkspacePermissionDenied(_)) + ), + "wire compatibility is not authority even for the original writer" ); - assert!(subjektiv_journal::host(&f.api, &f.store, &grant).is_ok()); let writer = JournalWriterBinding { subject_id: f.attr.subject_id.clone(), runtime_id: f.attr.runtime_id.clone(), @@ -4550,13 +4760,21 @@ async fn legacy_journal_grant_preserves_old_json_and_only_recovers_original_curr binding_id: f.host.writer_binding_id().unwrap().into(), }; f.store.release_journal_writer(&writer).unwrap(); - f.store + let successor = f + .store .claim_journal_writer(&f.attr.subject_id, &f.attr.runtime_id, "different-body") .unwrap(); + assert!(f.store.journal_writer_active(&successor).unwrap()); + assert!(!f.store.journal_writer_active(&writer).unwrap()); + assert_ne!(successor.binding_id, writer.binding_id); assert!( - subjektiv_journal::host(&f.api, &f.store, &grant).is_err(), + matches!( + subjektiv_journal::host(&f.api, &f.store, &grant), + Err(Error::WorkspacePermissionDenied(_)) + ), "legacy grant must never map to a later writer" ); + assert_eq!(serde_json::to_value(&grant).unwrap(), old); } /// Real embedded Runtime capture + SQLite + production dispatch; synthetic ordinary @@ -4662,6 +4880,85 @@ mod immutable { .unwrap(); } + fn frozen_lifetime_evidence( + f: &RuntimeFixture, + generation: &subjektiv::journal::JournalGeneration, + phase: &str, + ) -> Value { + let job = current_journal_job(f); + let attempt = current_journal_attempt(f, &generation.job.attempt_id); + let fence = f.job_store.with_conn(|connection| { + use rusqlite::OptionalExtension as _; + Ok(connection.query_row( + "SELECT json_object('owner', json(owner_json), 'history', json(history_json), + 'controller', json(controller_json), 'host', json(host_json), 'recorded_at', recorded_at) + FROM backend_job_execution_fences WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3", + rusqlite::params![TEST_WORKSPACE_ID, generation.job.job_id, generation.job.attempt_id], + |row| row.get::<_, String>(0), + ).optional()?) + }).unwrap(); + let removal = attempt.worker.as_ref().and_then(|worker| { + f.api + .config_store + .recover_worker_removal_execution(TEST_WORKSPACE_ID, worker) + .unwrap() + .map(|saved| { + json!({ + "operation_id": saved.plan.operation_id, + "input_fingerprint": saved.plan.input_fingerprint, + "state": saved.plan.state, + "worker": saved.plan.worker, + "session_disposition": saved.runtime_request.session_disposition, + }) + }) + }); + let domain = f + .api + .feature_storage + .open(&f.api.subjektiv_registration) + .unwrap() + .with_connection(|connection| { + let domain: String = connection.query_row( + "SELECT json_object('generation_id', g.id, 'job_id', g.job_id, 'attempt_id', g.attempt_id, 'active', g.active, + 'completion', json(g.completion_json), + 'process_order', j.process_order, 'process_byte', j.process_byte, + 'checkpoints', (SELECT COUNT(*) FROM journal_operations o + WHERE o.generation_id=g.id + AND json_extract(o.request_json,'$.action.action')='checkpoint')) + FROM journal_generations g JOIN journals j ON j.id=g.journal_id + WHERE g.id=?1 AND g.job_id=?2 AND g.attempt_id=?3", + rusqlite::params![ + generation.id, + generation.job.job_id, + generation.job.attempt_id + ], + |row| row.get(0), + )?; + Ok(domain) + }) + .unwrap(); + let evidence = json!({ + "phase": phase, + "job": { + "job_id": job.request.job_id, "state": job.state, + "current_attempt": job.current_attempt, "input_digest": job.input_digest, + "input_disposition": job.input_disposition, "result": job.result, + "failure_category": job.failure_category, "failure_detail": job.failure_detail, + }, + "attempt": attempt, + "retained": f.store.retained_journal_generation(&f.host, &f.id, &generation.job).unwrap(), + "domain": serde_json::from_str::(&domain).unwrap(), + "execution_fence": f.api.store.has_backend_job_execution_fence(TEST_WORKSPACE_ID, &attempt).unwrap(), + "fence_receipt": fence.map(|raw| serde_json::from_str::(&raw).unwrap()), + "canonical_removal": removal, + "source_writer_binding_id": f.host.writer_binding_id(), + "source_erased": f.store.journal_session_erased(&f.host).unwrap(), + "source_registered": f.api.store.get_worker_registry(TEST_WORKSPACE_ID, &f.source).unwrap().is_some(), + }); + eprintln!("frozen-generation-lifetime: {evidence}"); + evidence + } + #[tokio::test] async fn backend_immutable_input_survives_missing_source_and_predispatch_host_reopen() { let RuntimeFixture { @@ -4719,24 +5016,27 @@ mod immutable { range, id, }; - let saved = f + // The production constructor itself restores the accepted owner and may + // dispatch. Join that test owner's work, then use its real dispatch gate + // to order the two receipt-reconciliation cuts below. No live Worker is stopped. + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api.stop_backend_job_recovery(), + ) + .await + .expect("startup recovery did not quiesce for the frozen-generation cut"); + let gate = f .api - .store - .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id) - .unwrap() - .unwrap(); + .backend_job_recovery + .dispatch_gate(&generation.job.attempt_id); + let saved = current_journal_job(&f); + let reopened = frozen_lifetime_evidence(&f, &generation, "reopen_before_checkpoint"); assert_eq!( serde_json::to_vec(&saved.load_executable_request().unwrap().input).unwrap(), frozen ); assert_eq!(saved.input_digest, digest); - assert_eq!( - saved.state, - BackendJobState::Pending, - "pre-dispatch reopen failure: {:?}: {:?}", - saved.failure_category, - saved.failure_detail - ); + assert_eq!(saved.state, BackendJobState::Pending, "{reopened}"); let dispatched = f .api .dispatch_backend_job_reservation(f.reserved.clone()) @@ -4744,50 +5044,150 @@ mod immutable { assert_eq!( dispatched.attempt.state, BackendJobAttemptState::Dispatched, - "dispatch failure: {:?}: {:?}", - dispatched.attempt.failure_category, - dispatched.attempt.failure_detail - ); - assert_eq!( - f.store - .active_journal_generation(&f.host, &f.id) - .unwrap() - .unwrap_or_else(|| { - let current = current_journal_job(&f); - let attempt = current_journal_attempt(&f, &dispatched.attempt.attempt_id); - panic!("generation disappeared after dispatch: job={:?} {:?}: {:?}; attempt={:?} {:?}: {:?}", - current.state, current.failure_category, current.failure_detail, - attempt.state, attempt.failure_category, attempt.failure_detail); - }), - generation + "{reopened}" ); let worker = dispatched.attempt.worker.as_ref().unwrap(); - assert!(!root.exists()); + // Background recovery was joined; initial dispatch has returned. Hold the + // existing producer gate through the unreflected recovery and checkpoint. + let dispatch_finished = gate.lock().unwrap(); + let before = frozen_lifetime_evidence(&f, &generation, "dispatch_finished"); + assert_eq!(before["domain"]["active"], 1, "{before}"); + assert!(before["domain"]["completion"].is_null(), "{before}"); + assert_eq!(before["domain"]["checkpoints"], 0, "{before}"); + assert_eq!(before["execution_fence"], false, "{before}"); assert_eq!( - f.execution.take_inputs().last().unwrap().1, + f.store + .retained_journal_generation(&f.host, &f.id, &generation.job) + .unwrap(), + Some(generation.clone()), + "{before}" + ); + assert!(!root.exists()); + let inputs = f.execution.take_inputs(); + assert_eq!( + inputs.len(), + 1, + "one exact submission, including startup dispatch" + ); + assert_eq!( + inputs[0].1, f.request .worker_input(&dispatched.attempt.attempt_id) .unwrap() ); + + // Same callee as recover_backend_job_step, at an exact current owner cut. + // There is neither a checkpoint nor an execution/Host fence. Incomplete + // finish MUST NOT close; missing fences MUST NOT authorize continuation. + let error = subjektiv_journal::reconcile_attempt( + &f.api, + ¤t_journal_job(&f), + ¤t_journal_attempt(&f, &generation.job.attempt_id), + ) + .expect_err("unreflected generation without execution fences cannot complete or retry"); + let unreflected = frozen_lifetime_evidence(&f, &generation, "owner_before_checkpoint"); + eprintln!("frozen-generation-lifetime: incomplete_owner_error={error}"); + assert!( + matches!(error, Error::RepositoryConflict(_)), + "{error}; {unreflected}" + ); + assert_eq!(unreflected["domain"], before["domain"], "{unreflected}"); + assert_eq!(current_journal_job(&f).state, BackendJobState::Pending); + assert_eq!( + current_journal_attempt(&f, &generation.job.attempt_id).state, + BackendJobAttemptState::Dispatched + ); + + // Keep canonical cleanup before claim: a terminal Job cannot hide the + // generation's close cause behind purge, retirement or a later owner. + let cleanup_gate = Arc::new(tokio::sync::Semaphore::new(0)); + *f.api + .backend_job_recovery + .cleanup_before_claim + .lock() + .unwrap() = Some(cleanup_gate.clone()); f.checkpoint(worker); - let completion = subjektiv_journal::finish(&f.api, worker).unwrap(); + let checkpoint = f + .store + .query_journal_operation(&f.host, &f.id, "checkpoint") + .unwrap() + .unwrap(); + let checkpointed = frozen_lifetime_evidence(&f, &generation, "checkpoint_before_owner"); + assert_eq!(checkpointed["domain"]["active"], 1, "{checkpointed}"); + assert!( + checkpointed["domain"]["completion"].is_null(), + "{checkpointed}" + ); + assert_eq!(checkpointed["domain"]["checkpoints"], 1, "{checkpointed}"); + assert_eq!(checkpointed["domain"]["process_order"], 1, "{checkpointed}"); + f.api.backend_job_recovery.resume_after_test_pause(); + f.api.own_backend_job_recovery(&dispatched.attempt); + drop(dispatch_finished); + + // Release only the producer gate. No explicit finish/result/second input: + // the actual root owner must recover the saved checkpoint on the same Job. + let completed = observe_journal_completion(&f).await; + let settled = frozen_lifetime_evidence(&f, &generation, "owner_completed_cleanup_blocked"); + assert_eq!(completed.state, BackendJobState::Completed, "{settled}"); + assert_eq!(completed.current_attempt, dispatched.attempt.attempt); + assert_eq!(completed.input_digest, digest); + let completion: JournalJobCompletion = + serde_json::from_value(completed.result.clone().unwrap()).unwrap(); + assert_eq!(completion.generation_id, generation.id, "{settled}"); assert_eq!(completion.process_order, 1); + assert_eq!(completion.process_byte, 0); + assert_eq!(settled["domain"]["active"], 0, "{settled}"); + assert_eq!( + settled["domain"]["completion"], + json!(completion), + "{settled}" + ); + // The retained completion is written atomically with the normal close. + // Cleanup is still gated, so neither a cleanup close nor an execution + // fence can be the cause of this active -> completed transition. + assert_eq!(settled["domain"]["job_id"], json!(generation.job.job_id)); + assert_eq!( + settled["domain"]["attempt_id"], + json!(generation.job.attempt_id) + ); + assert_eq!(settled["domain"]["generation_id"], json!(generation.id)); + assert_eq!(settled["domain"]["checkpoints"], 1, "{settled}"); + assert_eq!(settled["execution_fence"], false, "{settled}"); + let terminal = current_journal_attempt(&f, &generation.job.attempt_id); + assert_eq!(terminal.worker, dispatched.attempt.worker); + assert_eq!(terminal.runtime_run_id, dispatched.attempt.runtime_run_id); + assert_eq!( + terminal.worker_cleanup_state, + Some(BackendJobWorkerCleanupState::Pending) + ); + assert_eq!( + f.store + .retained_journal_generation(&f.host, &f.id, &generation.job) + .unwrap(), + Some(generation.clone()), + "{settled}" + ); + assert_eq!( + f.store + .query_journal_operation(&f.host, &f.id, "checkpoint") + .unwrap(), + Some(checkpoint) + ); + assert_eq!( + completed.load_executable_request().unwrap().input, + f.request.input + ); + assert!( + f.execution.take_inputs().is_empty(), + "owner must not resubmit input" + ); assert!( !root.exists(), - "checkpoint completed with original capture/catalog/read unavailable" + "completion must not reacquire the original source" ); + std::fs::rename(&unavailable, &root).unwrap(); - f.api - .accept_backend_job_result( - worker, - &BackendJobResultSubmission { - job_id: f.request.job_id.clone(), - attempt_id: dispatched.attempt.attempt_id.clone(), - input_digest: digest, - result: serde_json::to_value(completion).unwrap(), - }, - ) - .unwrap(); + cleanup_gate.add_permits(1); wait_for_backend_job_worker_cleanup( &f.api, &f.request.job_id, @@ -4795,6 +5195,27 @@ mod immutable { BackendJobWorkerCleanupState::Completed, ) .await; + let cleaned = frozen_lifetime_evidence(&f, &generation, "cleanup_completed"); + assert_eq!(cleaned["execution_fence"], true, "{cleaned}"); + assert_eq!( + cleaned["domain"], settled["domain"], + "cleanup cannot reclose a different owner" + ); + assert_eq!( + f.api + .store + .list_backend_job_attempts(TEST_WORKSPACE_ID, &f.request.job_id) + .unwrap() + .len(), + 1, + "checkpoint recovery must not create a continuation attempt" + ); + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api.stop_backend_job_recovery(), + ) + .await + .expect("frozen-generation recovery did not finish"); } #[tokio::test] @@ -5640,8 +6061,8 @@ mod journal_migration { use subjektiv::journal::recording::JournalInput; #[tokio::test(flavor = "multi_thread", worker_threads = 4)] - async fn reserved_legacy_descriptor_invalidates_exact_identity_and_migrates_only_after_saved_checkpoint() - { + async fn reserved_legacy_descriptor_fails_without_reselection_and_preserves_partial_checkpoint() + { let f = SubjectRemovalFixture::new().await; let capture = subjektiv_journal::capture_input(&f.api, &f.attr, 0, 0, 48 * 1024).unwrap(); let ranges = subjektiv::journal::recording::capture_ranges(&capture).unwrap(); @@ -5693,7 +6114,7 @@ mod journal_migration { &ctx, JournalOperation { request_id: "legacy-saved-record".into(), - expected_state: generation.initial_journal.state_token, + expected_state: generation.initial_journal.state_token.clone(), action: JournalAction::Create { title: "Saved before migration".into(), body: "First original reflected; second still pending".into(), @@ -5726,7 +6147,7 @@ mod journal_migration { &ctx, JournalOperation { request_id: "legacy-saved-checkpoint".into(), - expected_state: position.state_token, + expected_state: position.state_token.clone(), action: JournalAction::Checkpoint { record_operation_ids: vec!["legacy-saved-record".into()], }, @@ -5737,19 +6158,56 @@ mod journal_migration { f.store .close_journal_generation(&f.host, &generation.id, &binding) .unwrap(); - assert_eq!(f.store.read_journal(&f.host, &id).unwrap().process_order, 1); + let before = f.store.read_journal(&f.host, &id).unwrap(); + assert_eq!(before.process_order, 1); + assert_eq!(before.process_byte, 0); + let sources = f + .store + .list_journal_sources(&f.host, &id, None, 32) + .unwrap(); + let entries = f + .store + .list_journal_entries(&f.host, &id, None, 16) + .unwrap(); + assert_eq!(entries.items.len(), 1); + assert!(f.store.journal_retention_required(&f.host).unwrap()); + let registry = f + .api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap(); + // This is the descriptor-only admission rejection boundary, NOT schema + // migration. Make the live source unavailable so dispatch/recovery cannot + // silently materialize a new body-bearing Job from the original Session. + let root = session_root(f.dir.path(), &f.target); + let unavailable = f.dir.path().join("legacy-source-unavailable"); + std::fs::rename(&root, &unavailable).unwrap(); + assert!(subjektiv_journal::capture_input(&f.api, &f.attr, 0, 0, 48 * 1024).is_err()); + f.execution.take_inputs(); + let original = reserved.clone(); let api = f.api.clone(); - let migrated = joined(spawn(move || { + let rejected = joined(spawn(move || { api.dispatch_backend_job_reservation(reserved) })) .unwrap(); - assert_eq!(migrated.job.request.job_id, legacy.job_id); let old = f .api .store .get_backend_job(TEST_WORKSPACE_ID, &legacy.job_id) .unwrap() .unwrap(); + let old_attempt = f + .api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &legacy.job_id, + &original.attempt.attempt_id, + ) + .unwrap() + .unwrap(); + assert_eq!(rejected.job, old); + assert_eq!(rejected.attempt, old_attempt); assert_eq!(old.state, BackendJobState::Failed); assert_eq!( old.failure_category.as_deref(), @@ -5757,122 +6215,146 @@ mod journal_migration { ); assert_eq!( old.input_disposition, - crate::backend_job::JobInputDisposition::Invalidated + crate::backend_job::JobInputDisposition::Available ); + assert_eq!(old.request, legacy); + assert_eq!(old.load_executable_request().unwrap(), &legacy); assert_eq!(old.input_digest, digest); assert_eq!(old.intent_fingerprint, fingerprint); - assert_eq!(old.request.job_id, legacy.job_id); - assert!(old.request.input.is_null()); - assert!(old.load_executable_request().is_err()); assert_eq!(old.current_attempt, 1); - let old_attempt = f - .api - .store - .get_backend_job_attempt( - TEST_WORKSPACE_ID, - &legacy.job_id, - &crate::backend_job::attempt_id(&legacy.job_id, 1), - ) - .unwrap() - .unwrap(); - assert_eq!(old_attempt.input_digest, digest); + assert!(old.result.is_none()); assert!( - old_attempt.worker.is_none(), - "descriptor-only input must never reach a Worker" + !f.api + .store + .backend_job_was_migrated(TEST_WORKSPACE_ID, &legacy.job_id) + .unwrap(), + "descriptor rejection must not impersonate schema migration" ); + assert_eq!(old_attempt.state, BackendJobAttemptState::Failed); + assert_eq!(old_attempt.attempt_id, original.attempt.attempt_id); + assert_eq!(old_attempt.attempt, 1); + assert_eq!(old_attempt.input_digest, digest); + assert_eq!(old_attempt.failure_category, old.failure_category); + assert_eq!(old_attempt.failure_detail, old.failure_detail); + assert!(old_attempt.worker.is_none()); + assert!(old_attempt.runtime_run_id.is_none()); + assert!(old_attempt.result.is_none()); - let (new, attempt) = tokio::time::timeout(std::time::Duration::from_secs(20), async { - loop { - for job in f - .api - .store - .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) - .unwrap() - { - if job.request.job_id == legacy.job_id { - continue; - } - let attempt = f - .api - .store - .get_backend_job_attempt( - TEST_WORKSPACE_ID, - &job.request.job_id, - &crate::backend_job::attempt_id( - &job.request.job_id, - job.current_attempt, - ), - ) - .unwrap() - .unwrap(); - if attempt.state == BackendJobAttemptState::Dispatched { - return (job, attempt); - } - } - tokio::task::yield_now().await; - } - }) - .await - .expect("migration dispatch progresses without a lifecycle gate inversion"); - assert_ne!(new.request.job_id, legacy.job_id); - let frozen = JournalInput::decode(&new.request.input).unwrap(); - assert_eq!(frozen.selected_ranges, vec![ranges[1].clone()]); - assert_eq!(frozen.initial_state_token, checkpoint.state_token); + subjektiv_journal::reconcile_attempt(&f.api, &old, &old_attempt).unwrap(); + subjektiv_journal::recover_attempts(&f.api).unwrap(); + f.api.recover_backend_jobs().unwrap(); + let api = f.api.clone(); + let replay = joined(spawn(move || { + api.dispatch_backend_job_reservation(original) + })) + .unwrap(); + assert!( + replay.replayed, + "terminal rejection is observation, not execution" + ); + assert_eq!(replay.job, old); + assert_eq!(replay.attempt, old_attempt); assert_eq!( - f.store - .query_journal_operation(&f.host, &id, "legacy-saved-record") + f.api + .store + .get_backend_job(TEST_WORKSPACE_ID, &legacy.job_id) .unwrap(), - Some(record) + Some(old.clone()) ); assert_eq!( - f.store - .query_journal_operation(&f.host, &id, "legacy-saved-checkpoint") + f.api + .store + .get_backend_job_attempt(TEST_WORKSPACE_ID, &legacy.job_id, &old_attempt.attempt_id) .unwrap(), - Some(checkpoint) + Some(old_attempt) + ); + assert_eq!( + f.api + .store + .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100) + .unwrap(), + vec![old], + "descriptor failure must not automatically select a new Job" + ); + assert!( + f.api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &legacy.job_id, + &crate::backend_job::attempt_id(&legacy.job_id, 2), + ) + .unwrap() + .is_none(), + "no automatic continuation attempt" + ); + assert!( + f.execution.take_inputs().is_empty(), + "neither the old nor a new Job may submit" + ); + assert!( + !root.exists(), + "dispatch must not recreate the unavailable source" + ); + assert_eq!(f.store.read_journal(&f.host, &id).unwrap(), before); + assert_eq!( + f.store + .list_journal_sources(&f.host, &id, None, 32) + .unwrap(), + sources ); assert_eq!( f.store .list_journal_entries(&f.host, &id, None, 16) + .unwrap(), + entries + ); + for (request_id, receipt) in [ + ("legacy-saved-record", record), + ("legacy-saved-position", position), + ("legacy-saved-checkpoint", checkpoint), + ] { + assert_eq!( + f.store + .query_journal_operation(&f.host, &id, request_id) + .unwrap(), + Some(receipt) + ); + } + assert_eq!( + f.store + .retained_journal_generation(&f.host, &id, &binding) + .unwrap(), + Some(generation) + ); + assert!( + f.store + .active_journal_generation(&f.host, &id) .unwrap() - .items - .len(), - 1 + .is_none() ); - let api = f.api.clone(); - let worker = attempt.worker.clone().unwrap(); assert!( - joined(spawn(move || subjektiv_journal::execute( - &api, - &worker, - JournalJobRequest::ReadState - ))) - .is_ok() + f.store.journal_retention_required(&f.host).unwrap(), + "Failed is not reflection of the second original" ); - let api = f.api.clone(); - let stale = migrated.clone(); + assert_eq!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.target) + .unwrap(), + registry + ); + assert!(!f.store.journal_session_erased(&f.host).unwrap()); assert!( - joined(spawn(move || api.dispatch_backend_job_reservation(stale))).is_err(), - "invalidated internal reservations cannot dispatch or silently replay" + !f.store + .journal_worker_was_released( + &f.attr.subject_id, + &f.target.runtime_id, + &f.target.worker_id, + ) + .unwrap() ); - // Stop this intentionally unfinished new body Job without pretending its - // remaining original was reflected; normal cleanup still needs its receipt. - f.api - .finish_backend_job_attempt( - &new.request.job_id, - &attempt.attempt_id, - BackendJobAttemptState::Failed, - "test_finished", - "remaining original deliberately unprocessed", - ) - .unwrap(); - wait_for_backend_job_worker_cleanup( - &f.api, - &new.request.job_id, - &attempt.attempt_id, - BackendJobWorkerCleanupState::Completed, - ) - .await; - assert_eq!(f.store.read_journal(&f.host, &id).unwrap().process_order, 1); + std::fs::rename(&unavailable, &root).unwrap(); } #[tokio::test(flavor = "multi_thread", worker_threads = 4)] diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index fbe60c3d2..e9e437f82 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -1616,6 +1616,14 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { now_seconds: u64, consumed_at: &str, ) -> Result; + fn plan_job_cleanup_worker_removal( + &self, + request: &crate::retention::WorkerRemovalPlanRequest, + inventory: &worker_runtime::retention::WorkerRetentionInventory, + ) -> std::result::Result< + crate::retention::WorkerRemovalPlan, + crate::retention::WorkerRetentionError, + >; fn plan_worker_removal( &self, request: &crate::retention::WorkerRemovalPlanRequest, @@ -6415,6 +6423,17 @@ impl ControlPlaneStore for SqliteWorkspaceStore { .await } + fn plan_job_cleanup_worker_removal( + &self, + request: &crate::retention::WorkerRemovalPlanRequest, + inventory: &worker_runtime::retention::WorkerRetentionInventory, + ) -> std::result::Result< + crate::retention::WorkerRemovalPlan, + crate::retention::WorkerRetentionError, + > { + SqliteWorkspaceStore::plan_job_cleanup_worker_removal(self, request, inventory) + } + fn plan_worker_removal( &self, request: &crate::retention::WorkerRemovalPlanRequest,