From 50a828a0085d7d826fe465364f30008e3e36df0e Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 22:37:09 +0900 Subject: [PATCH] fix: retain exact standalone result recovery owners across capacity and job errors Recover captured Job/attempt identities independently; wait for actual task return and cleanup before freeing execution capacity. Preserve per-job rejection and wait causes without dropping subsequent reservations. Add one-time additive v7 admission-rejection evidence, never infer execution drain from Failed or an absent map. Validated: parent Jobs 20 tests and store/recovery/migration 13 tests on the current dependency snapshot; root check, format, names and diff pass. This is an unfinished, unapproved T-841 checkpoint; R1/R6 and final whole-contract validation/review remain outstanding. --- crates/standalone/src/host.rs | 17 +- crates/standalone/src/job_store.rs | 17 +- crates/standalone/src/job_store/recovery.rs | 76 ++++++ .../src/job_store/start_rejection_tests.rs | 195 ++++++++++++++ crates/standalone/src/jobs.rs | 37 +-- crates/standalone/src/jobs/recovery.rs | 191 ++++++++++++++ crates/standalone/src/jobs/recovery_tests.rs | 245 ++++++++++++++++++ crates/standalone/src/jobs/tests.rs | 5 +- 8 files changed, 753 insertions(+), 30 deletions(-) create mode 100644 crates/standalone/src/job_store/start_rejection_tests.rs create mode 100644 crates/standalone/src/jobs/recovery.rs create mode 100644 crates/standalone/src/jobs/recovery_tests.rs diff --git a/crates/standalone/src/host.rs b/crates/standalone/src/host.rs index b4f63236a..d46d32552 100644 --- a/crates/standalone/src/host.rs +++ b/crates/standalone/src/host.rs @@ -495,7 +495,7 @@ impl StandaloneHost { if let Some(connection) = &subject { connection.host.recover_jobs(); } - let reserved_result_jobs = jobs.reserved_result_job_ids(); + let reserved_result_jobs = jobs.reserved_result_attempts(); let result_jobs = jobs.clone(); let result_recovery = tokio::spawn(async move { result_jobs @@ -615,7 +615,7 @@ impl StandaloneHost { } pub async fn shutdown(mut self) -> Result<(), StandaloneShutdownError> { - self.stop_result_recovery().await; + let recovery_result = self.stop_result_recovery().await; let jobs_result = self.jobs.shutdown().await; let command = protocol::WorkerCommandEnvelope::new(u64::MAX); let _ = self.handle.send(Method::Shutdown { command }).await; @@ -627,7 +627,7 @@ impl StandaloneHost { self.retain_lease(); return Err(error); } - if jobs_result.is_err() { + if jobs_result.is_err() || recovery_result.is_err() { self.retain_lease(); return Err(StandaloneShutdownError::JobCleanup); } @@ -665,14 +665,15 @@ impl StandaloneHost { Ok(()) } - async fn stop_result_recovery(&mut self) { + async fn stop_result_recovery(&mut self) -> Result<(), crate::jobs::JobServiceError> { self.jobs.close(); if let Some(task) = self.result_recovery.take() { - // This joins only the dispatcher. Every admitted Worker/Host effect - // remains owned by jobs.shutdown; cancelling a waiter proves no cleanup. - task.abort(); - let _ = task.await; + // Close wakes capacity/storage waiters. Join the dispatcher and retain + // its error; only jobs.shutdown can confirm admitted execution cleanup. + task.await + .map_err(|error| crate::jobs::JobServiceError::Cleanup(error.to_string()))??; } + Ok(()) } fn retain_lease(&mut self) { diff --git a/crates/standalone/src/job_store.rs b/crates/standalone/src/job_store.rs index 30da5b2fe..06a1e3626 100644 --- a/crates/standalone/src/job_store.rs +++ b/crates/standalone/src/job_store.rs @@ -29,7 +29,11 @@ mod legacy_migrations; #[path = "job_store/recovery.rs"] mod recovery; -const SCHEMA_VERSION: i64 = 6; +#[cfg(test)] +#[path = "job_store/start_rejection_tests.rs"] +mod start_rejection_tests; + +const SCHEMA_VERSION: i64 = 7; /// Availability of the immutable input body, independent of execution outcome. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -120,6 +124,9 @@ pub struct JobSnapshot { pub acknowledged: bool, /// A version cutover is final even when input/checkpoint remains incomplete. pub version_stopped: bool, + /// The exact current reservation was rejected before dispatch, by this Host. + /// This is not a runner shutdown or cleanup receipt. + pub start_rejected_by: Option, /// All historical execution owners, independent of Job outcome/checkpoint. pub executions: Vec, } @@ -135,14 +142,16 @@ pub struct JobExecutionOwnership { impl JobSnapshot { /// Terminal failure/completion without the attempt's admission binding is not - /// an empty set of successfully joined executions. Keep that missing evidence - /// distinct from an unstarted reservation and from checkpoint coverage. + /// an empty set of successfully joined executions. Only an explicit rejection + /// saved while Reserved can preclude that binding; it proves no runner cleanup. + /// All historical admitted executions still require their own cleanup proof. pub fn executions_released(&self) -> bool { let needs_binding = matches!( self.attempt.state, JobAttemptState::Dispatched | JobAttemptState::Completed | JobAttemptState::Failed ); (!needs_binding + || (self.attempt.state == JobAttemptState::Failed && self.start_rejected_by.is_some()) || self .executions .iter() @@ -559,6 +568,7 @@ impl JobStore { WHERE a.state IN ('dispatched','failed','completed') AND NOT EXISTS(SELECT 1 FROM job_execution_ownership e WHERE e.attempt_id=a.attempt_id) AND NOT EXISTS(SELECT 1 FROM job_version_stops s WHERE s.job_id=i.job_id) + AND NOT (a.state='failed' AND EXISTS(SELECT 1 FROM job_start_rejections r WHERE r.job_id=i.job_id AND r.attempt_id=a.attempt_id)) LIMIT 1", [], |row|row.get(0)).optional()?; if let Some(attempt) = missing { return Err(JobStoreError::Corrupt(format!( @@ -1040,6 +1050,7 @@ fn read_snapshot(conn: &Connection, job_id: &str) -> Result, state, acknowledged, version_stopped: conn.query_row("SELECT EXISTS(SELECT 1 FROM job_version_stops WHERE job_id=?1)", [job_id], |r|r.get(0))?, + start_rejected_by: conn.query_row("SELECT owner_id FROM job_start_rejections WHERE job_id=?1 AND attempt_id=?2", params![job_id,attempt_id], |r|r.get(0)).optional()?, executions: { let mut stmt = conn.prepare("SELECT attempt_id,owner_id,run_id,writes_revoked,cleanup_receipt FROM job_execution_ownership WHERE job_id=?1 ORDER BY rowid")?; stmt.query_map([job_id], |r| Ok(JobExecutionOwnership { attempt_id:r.get(0)?,owner_id:r.get(1)?,run_id:r.get(2)?,writes_revoked:r.get(3)?,cleanup_receipt:r.get(4)? }))?.collect::,_>>()? diff --git a/crates/standalone/src/job_store/recovery.rs b/crates/standalone/src/job_store/recovery.rs index 43c2026a7..e48e24cc1 100644 --- a/crates/standalone/src/job_store/recovery.rs +++ b/crates/standalone/src/job_store/recovery.rs @@ -3,6 +3,55 @@ use super::*; pub(super) const UPDATE_INTERRUPTION: &str = "バージョン更新による実行打切り"; impl JobStore { + pub(crate) fn probe_reserved_result_write( + &mut self, + job_id: &str, + attempt_id: &str, + ) -> Result<(), JobStoreError> { + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + let saved = require_snapshot(&tx, job_id)?; + check_attempt(&saved, attempt_id)?; + if saved.version_stopped || saved.domain_grant.is_some() { + return Err(JobStoreError::AttemptMismatch); + } + // A write probe, not an execution completion or resource release receipt. + tx.execute( + "UPDATE job_attempts SET failure=failure WHERE job_id=?1 AND attempt_id=?2", + params![job_id, attempt_id], + )?; + tx.commit()?; + Ok(()) + } + + pub(crate) fn fail_reserved_result( + &mut self, + job_id: &str, + attempt_id: &str, + cause: &str, + ) -> Result<(), JobStoreError> { + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + let saved = require_snapshot(&tx, job_id)?; + check_attempt(&saved, attempt_id)?; + if saved.version_stopped + || saved.domain_grant.is_some() + || saved.state != JobState::Pending + || saved.attempt.state != JobAttemptState::Reserved + { + return Err(JobStoreError::AttemptMismatch); + } + tx.execute( + "INSERT INTO job_start_rejections(job_id,attempt_id,owner_id) VALUES(?1,?2,?3)", + params![job_id, attempt_id, self.owner_id], + )?; + set_terminal(&tx, &saved, JobAttemptState::Failed, Some(cause))?; + tx.commit()?; + Ok(()) + } + /// The Host supplies the identity from its revalidated exclusive Worker lease. /// This binds admission, results and restart reconciliation to the same owner, /// rather than a fresh connection UUID that could impersonate a restart. @@ -156,9 +205,15 @@ pub(super) fn initialize(conn: &mut Connection) -> Result<(), JobStoreError> { } migrate(&tx)?; } + 6 => {} SCHEMA_VERSION => {} other => return Err(JobStoreError::UnsupportedSchema(other)), } + if version <= 6 { + // Additive admission evidence only. Do not rerun the v6 cutover or + // reinterpret preexisting Failed rows as rejected-before-dispatch. + tx.execute_batch(START_REJECTION_SCHEMA)?; + } if tx.prepare("PRAGMA foreign_key_check")?.exists([])? { return Err(JobStoreError::Corrupt( "Job migration foreign key check failed".into(), @@ -242,6 +297,27 @@ fn migrate(conn: &Connection) -> Result<(), JobStoreError> { Ok(()) } +// The insertion guard binds the rejection to the still-Reserved exact attempt. +// Neither an absent execution row nor a Failed state can create this evidence. +const START_REJECTION_SCHEMA: &str = " +CREATE TABLE job_start_rejections ( + job_id TEXT NOT NULL REFERENCES job_intents(job_id), + attempt_id TEXT NOT NULL PRIMARY KEY REFERENCES job_attempts(attempt_id), + owner_id TEXT NOT NULL +); +CREATE TRIGGER job_start_rejection_guard BEFORE INSERT ON job_start_rejections +WHEN NOT EXISTS(SELECT 1 FROM job_intents i JOIN job_attempts a ON a.job_id=i.job_id AND a.number=i.current_attempt + WHERE i.job_id=NEW.job_id AND a.attempt_id=NEW.attempt_id AND i.state='pending' AND a.state='reserved') + OR EXISTS(SELECT 1 FROM job_execution_ownership WHERE attempt_id=NEW.attempt_id) + OR EXISTS(SELECT 1 FROM job_domain_grants WHERE job_id=NEW.job_id) + OR EXISTS(SELECT 1 FROM job_version_stops WHERE job_id=NEW.job_id) +BEGIN SELECT RAISE(ABORT,'start rejection requires unstarted result-only reservation'); END; +CREATE TRIGGER job_start_rejection_immutable BEFORE UPDATE ON job_start_rejections +BEGIN SELECT RAISE(ABORT,'immutable Job start rejection'); END; +CREATE TRIGGER job_start_rejection_retained BEFORE DELETE ON job_start_rejections +BEGIN SELECT RAISE(ABORT,'retained Job start rejection'); END; +"; + pub(super) const SCHEMA: &str = " CREATE TABLE job_version_history ( job_id TEXT NOT NULL, diff --git a/crates/standalone/src/job_store/start_rejection_tests.rs b/crates/standalone/src/job_store/start_rejection_tests.rs new file mode 100644 index 000000000..458a0ee10 --- /dev/null +++ b/crates/standalone/src/job_store/start_rejection_tests.rs @@ -0,0 +1,195 @@ +//! Admission evidence tests use the production store/initializer and real SQLite. +use super::*; + +fn request(id: &str) -> JobRequest { + JobRequest { + job_id: id.into(), + purpose: "rejection fixture".into(), + input_ref: "fixture:input".into(), + input: serde_json::json!({"body":"retained"}), + instruction: "Return result".into(), + profile: "builtin:job".into(), + serialization_key: None, + limits: job::JobLimits::default(), + } +} + +#[test] +fn start_rejection_is_atomic_and_failed_or_missing_execution_never_implies_it() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("jobs.sqlite3"); + let mut store = JobStore::open(&path).unwrap(); + let saved = store.reserve(request("reject")).unwrap(); + let attempt = &saved.attempt.attempt_id; + store + .conn + .execute_batch( + "CREATE TRIGGER refuse_failure BEFORE UPDATE OF state ON job_attempts + WHEN NEW.state='failed' BEGIN SELECT RAISE(ABORT,'injected terminal write failure'); END;", + ) + .unwrap(); + assert!( + store + .fail_reserved_result("reject", attempt, "configuration unavailable") + .is_err() + ); + assert_eq!(store.get("reject").unwrap(), saved); + assert_eq!( + store + .conn + .query_row("SELECT COUNT(*) FROM job_start_rejections", [], |r| r + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + store + .conn + .execute_batch("DROP TRIGGER refuse_failure") + .unwrap(); + store + .fail_reserved_result("reject", attempt, "configuration unavailable") + .unwrap(); + let rejected = store.get("reject").unwrap(); + assert_eq!( + rejected.attempt.failure.as_deref(), + Some("configuration unavailable") + ); + assert_eq!( + rejected.start_rejected_by.as_deref(), + Some(store.owner_id.as_str()) + ); + assert!(rejected.executions.is_empty()); + assert!(rejected.executions_released()); + assert!(store.execution_obligations().unwrap().is_empty()); + for sql in [ + "DELETE FROM job_start_rejections", + "UPDATE job_start_rejections SET owner_id='other'", + ] { + assert!(store.conn.execute_batch(sql).is_err()); + } + let bare = store.reserve(request("bare-failed")).unwrap(); + store + .finish( + "bare-failed", + &bare.attempt.attempt_id, + JobAttemptState::Failed, + Some("not admission evidence"), + ) + .unwrap(); + assert!(!store.get("bare-failed").unwrap().executions_released()); + assert!( + store + .conn + .execute( + "INSERT INTO job_start_rejections VALUES('bare-failed',?1,'forged')", + [&bare.attempt.attempt_id] + ) + .is_err() + ); + assert!(store.execution_obligations().is_err()); + let running = store.reserve(request("running")).unwrap(); + store + .dispatch("running", &running.attempt.attempt_id, 8) + .unwrap(); + assert!( + store + .fail_reserved_result("running", &running.attempt.attempt_id, "late failure") + .is_err() + ); + let running = store.get("running").unwrap(); + assert!(running.start_rejected_by.is_none()); + assert!(!running.executions_released()); + drop(store); + let reopened = JobStore::open(&path).unwrap(); + assert_eq!(reopened.get("reject").unwrap(), rejected); + assert_eq!(reopened.get("running").unwrap(), running); + assert!(reopened.execution_obligations().is_err()); +} + +#[test] +fn start_rejection_migration_adds_only_evidence_and_rolls_back_then_reopens_without_cutover() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("jobs.sqlite3"); + let mut conn = Connection::open(&path).unwrap(); + feature_storage::configure_connection(&conn).unwrap(); + conn.execute_batch(SCHEMA).unwrap(); + conn.execute_batch(GRANT_SCHEMA).unwrap(); + conn.execute_batch(RECONCILIATION_SCHEMA).unwrap(); + conn.execute_batch(recovery::SCHEMA).unwrap(); + conn.pragma_update(None, "user_version", 6).unwrap(); + let req = request("already-reserved"); + let tx = conn.transaction().unwrap(); + tx.execute("INSERT INTO job_intents(job_id,request_json,fingerprint,max_concurrent,state,current_attempt,acknowledged) + VALUES(?1,?2,?3,8,'pending',1,0)", params![req.job_id,serde_json::to_string(&req).unwrap(),req.fingerprint().unwrap()]).unwrap(); + insert_attempt(&tx, &req, 1).unwrap(); + tx.commit().unwrap(); + conn.execute_batch("CREATE TRIGGER job_start_rejection_guard BEFORE UPDATE ON job_intents BEGIN SELECT 1; END;").unwrap(); + assert!(recovery::initialize(&mut conn).is_err()); + assert_eq!( + conn.pragma_query_value(None, "user_version", |r| r.get::<_, i64>(0)) + .unwrap(), + 6 + ); + assert!( + !conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='job_start_rejections')", + [], + |r| r.get::<_, bool>(0) + ) + .unwrap() + ); + conn.execute_batch("DROP TRIGGER job_start_rejection_guard") + .unwrap(); + recovery::initialize(&mut conn).unwrap(); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM job_start_rejections", [], |r| r + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM job_version_stops", [], |r| r + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + drop(conn); + let mut store = JobStore::open(&path).unwrap(); + let saved = store.get(&req.job_id).unwrap(); + assert_eq!(saved.request, req); + assert_eq!(saved.state, JobState::Pending); + assert_eq!(saved.attempt.state, JobAttemptState::Reserved); + assert!(!saved.version_stopped); + assert!(saved.start_rejected_by.is_none()); + store + .fail_reserved_result( + &req.job_id, + &saved.attempt.attempt_id, + "unavailable profile", + ) + .unwrap(); + let rejected = store.get(&req.job_id).unwrap(); + drop(store); + let reopened = JobStore::open(&path).unwrap(); + assert_eq!(reopened.get(&req.job_id).unwrap(), rejected); + assert!(reopened.execution_obligations().unwrap().is_empty()); + let fresh = JobStore::open(&temp.path().join("fresh.sqlite3")).unwrap(); + let schema = |conn: &Connection| { + conn.prepare( + "SELECT type,name,sql FROM sqlite_schema WHERE sql IS NOT NULL ORDER BY type,name", + ) + .unwrap() + .query_map([], |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, String>(1)?, + r.get::<_, String>(2)?, + )) + }) + .unwrap() + .collect::>>() + .unwrap() + }; + assert_eq!(schema(&reopened.conn), schema(&fresh.conn)); +} diff --git a/crates/standalone/src/jobs.rs b/crates/standalone/src/jobs.rs index 37d654f5f..caba25d51 100644 --- a/crates/standalone/src/jobs.rs +++ b/crates/standalone/src/jobs.rs @@ -14,6 +14,7 @@ //! Timeout/shutdown fences unaccepted attempts before cancellation. Only owned runner shutdown //! plus joined/revoked Host callbacks discharge execution reservations. Clones cannot reopen //! the store after service closure or carry authority over to another Host. +mod recovery; #[cfg(test)] mod tests; use std::collections::HashMap; @@ -128,6 +129,10 @@ pub struct StandaloneJobs { domain: Arc>>>, model_client: Arc>>>, storage_recovered: Arc, + capacity_changed: Arc, + // 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>>>>>, } impl StandaloneJobs { @@ -171,6 +176,8 @@ impl StandaloneJobs { domain: Arc::default(), model_client: Arc::default(), storage_recovered: Arc::default(), + capacity_changed: Arc::default(), + result_recovery: Arc::default(), }) } @@ -199,7 +206,9 @@ impl StandaloneJobs { Ok(jobs) } - pub(crate) fn reserved_result_job_ids(&self) -> Result, JobServiceError> { + pub(crate) fn reserved_result_attempts( + &self, + ) -> Result, JobServiceError> { // Retain IDs, not a second queue of immutable bodies outside the owned // execution barrier while awaiting an earlier Job's cleanup. Ok(self @@ -210,23 +219,10 @@ impl StandaloneJobs { && !saved.version_stopped && saved.attempt.state == JobAttemptState::Reserved }) - .map(|saved| saved.request.job_id) + .map(|saved| (saved.request.job_id, saved.attempt.attempt_id)) .collect()) } - /// The restore caller captures IDs before returning the Host to clients, then - /// runs this once after binding adapters. Granted Jobs keep domain recovery. - pub(crate) async fn resume_reserved_result_jobs( - &self, - ids: Vec, - ) -> Result<(), JobServiceError> { - for id in ids { - self.start(&id).await?; - self.wait(&id).await?; - } - Ok(()) - } - /// Storage exhaustion is owned by the same execution Task, not a new pending /// Job state. The diagnostic retains its typed source until a successful Host /// store write wakes reconciliation; no polling or model retry is scheduled. @@ -481,7 +477,7 @@ impl StandaloneJobs { } pub async fn start(&self, job_id: &str) -> Result<(), JobServiceError> { - self.start_with_optional_model_client(job_id, None).await + self.start_with_optional_model_client(job_id, None) } /// Injectable model transport for embedders and scripted execution tests. @@ -496,10 +492,9 @@ impl StandaloneJobs { C: LlmClient + 'static, { self.start_with_optional_model_client(job_id, Some(Box::new(client))) - .await } - async fn start_with_optional_model_client( + fn start_with_optional_model_client( &self, job_id: &str, client: Option>, @@ -639,6 +634,7 @@ impl StandaloneJobs { let recovery_issue = Arc::new(Mutex::new(None)); let task_issue = recovery_issue.clone(); let storage_recovered = self.storage_recovered.clone(); + let capacity_changed = self.capacity_changed.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| { @@ -715,6 +711,9 @@ impl StandaloneJobs { complete_tx.send_replace(Some( outcome.as_ref().map(|_| ()).map_err(ToString::to_string), )); + // A wake is only an observation: recovery joins the successful task + // return before rechecking capacity. An error does not free its slot. + capacity_changed.notify_waiters(); outcome }); executions.jobs.insert( @@ -770,6 +769,7 @@ impl StandaloneJobs { .is_some_and(|run| run.attempt_id == attempt) { executions.jobs.remove(job_id); + self.capacity_changed.notify_waiters(); } } self.get(job_id) @@ -778,6 +778,7 @@ impl StandaloneJobs { pub(crate) fn close(&self) { let mut executions = self.executions.lock().expect("Job executions poisoned"); executions.closed = true; + self.capacity_changed.notify_waiters(); for execution in executions.jobs.values() { execution.cancel.send_replace(true); } diff --git a/crates/standalone/src/jobs/recovery.rs b/crates/standalone/src/jobs/recovery.rs new file mode 100644 index 000000000..dd27fd2e0 --- /dev/null +++ b/crates/standalone/src/jobs/recovery.rs @@ -0,0 +1,191 @@ +//! Restore ownership for result-only reservations. Domain recovery keeps its own contract. +use super::*; + +impl StandaloneJobs { + /// Capture exact Job/attempt identities before exposing the restored Host. + /// Each admission has its own waiter: capacity, storage or cleanup failure in + /// one Job must not discard the other accepted reservations. + pub(crate) async fn resume_reserved_result_jobs( + &self, + ids: Vec<(String, String)>, + ) -> Result<(), JobServiceError> { + let mut owners = tokio::task::JoinSet::new(); + for (id, attempt) in ids { + let issue = Arc::new(Mutex::new(None)); + { + let mut admissions = self.result_recovery.lock().expect("Job recovery poisoned"); + if admissions.contains_key(&id) { + continue; + } + admissions.insert(id.clone(), issue.clone()); + } + let jobs = self.clone(); + owners.spawn(async move { + let outcome = tokio::select! { + biased; + _ = jobs.recovery_closed() => return Ok(()), + outcome = jobs.resume_result_attempt(&id, &attempt, &issue) => outcome, + }; + match outcome { + Ok(()) => { + jobs.result_recovery + .lock() + .expect("Job recovery poisoned") + .remove(&id); + Ok(()) + } + Err(error) => { + let detail = error.to_string(); + *issue.lock().expect("Job recovery issue poisoned") = Some(error); + Err(JobServiceError::Cleanup(format!( + "result recovery {id}/{attempt}: {detail}" + ))) + } + } + }); + } + let mut first_error = None; + while let Some(result) = owners.join_next().await { + let result = + result.unwrap_or_else(|error| Err(JobServiceError::Cleanup(error.to_string()))); + if let Err(error) = result { + first_error.get_or_insert(error); + } + } + first_error.map_or(Ok(()), Err) + } + + /// Unresolved restore causes are independent of execution/cleanup evidence. + pub fn result_recovery_diagnostic(&self, job_id: &str) -> Option { + let admissions = self.result_recovery.lock().expect("Job recovery poisoned"); + let issue = admissions.get(job_id)?; + let issue = issue.lock().expect("Job recovery issue poisoned"); + Some(issue.as_ref().map_or_else( + || "accepted result-only reservation retained; awaiting execution capacity or owner settlement".into(), + ToString::to_string, + )) + } + + async fn recovery_closed(&self) { + loop { + let changed = self.capacity_changed.notified(); + tokio::pin!(changed); + changed.as_mut().enable(); + if self + .executions + .lock() + .expect("Job executions poisoned") + .closed + { + return; + } + changed.await; + } + } + + async fn resume_result_attempt( + &self, + id: &str, + attempt: &str, + issue: &Mutex>, + ) -> Result<(), JobServiceError> { + loop { + // Register before checking capacity, including completed-but-not-yet- + // returned tasks, so a real release cannot be lost between checks. + let changed = self.capacity_changed.notified(); + tokio::pin!(changed); + changed.as_mut().enable(); + let completed: Vec<_> = self + .executions + .lock() + .expect("Job executions poisoned") + .jobs + .iter() + .filter(|(_, run)| matches!(run.complete.borrow().as_ref(), Some(Ok(())))) + .map(|(id, _)| id.clone()) + .collect(); + for completed_id in completed { + // wait joins task return and only then removes a successful slot. + // A failed read here is not evidence against this reservation. + let _ = self.wait(&completed_id).await; + } + let started = recover_owned_settlement( + &self.storage_recovered, + issue, + || self.start_restored_result_attempt(id, attempt), + || self.with_store_write(|store| store.probe_reserved_result_write(id, attempt)), + ) + .await; + match started { + Err( + error @ (JobServiceError::Capacity + | JobServiceError::Store( + JobStoreError::ConcurrentLimit | JobStoreError::SerializationConflict(_), + )), + ) => { + *issue.lock().expect("Job recovery issue poisoned") = Some(error); + changed.await; + } + Err(error) => return Err(error), + Ok(()) => { + // Retry only a failed result read, never the already-admitted + // start. Cleanup errors retain their execution ownership. + let outcome = self.wait(id).await; + if let Err(error) = outcome { + if !transient_settlement_storage_error(&error) { + return Err(error); + } + recover_owned_settlement( + &self.storage_recovered, + issue, + || self.get(id).map(|_| ()), + || { + self.with_store_write(|store| { + store.probe_reserved_result_write(id, attempt) + }) + }, + ) + .await?; + } + return Ok(()); + } + } + } + } + + fn start_restored_result_attempt( + &self, + id: &str, + attempt: &str, + ) -> Result<(), JobServiceError> { + let saved = self.get(id)?; + if saved.attempt.attempt_id != attempt + || saved.domain_grant.is_some() + || saved.version_stopped + { + return Err(JobStoreError::AttemptMismatch.into()); + } + // A client may have cancelled or started this exact attempt since capture. + // Never dispatch it twice, and never follow a newly-issued retry attempt. + if saved.state != JobState::Pending || saved.attempt.state != JobAttemptState::Reserved { + return Ok(()); + } + match self.start_with_optional_model_client(id, None) { + Ok(()) => Ok(()), + Err( + error @ (JobServiceError::Capacity + | JobServiceError::Closed + | JobServiceError::Store( + JobStoreError::ConcurrentLimit | JobStoreError::SerializationConflict(_), + )), + ) => Err(error), + Err(error) if transient_settlement_storage_error(&error) => Err(error), + Err(error) => { + // This path has no admitted execution. Persist the individual + // rejection without forging cleanup for any prior ownership. + let detail = format!("result_recovery_cannot_start: {error}"); + self.with_store_write(|store| store.fail_reserved_result(id, attempt, &detail)) + } + } + } +} diff --git a/crates/standalone/src/jobs/recovery_tests.rs b/crates/standalone/src/jobs/recovery_tests.rs new file mode 100644 index 000000000..37892ec61 --- /dev/null +++ b/crates/standalone/src/jobs/recovery_tests.rs @@ -0,0 +1,245 @@ +//! Restore integration boundary: real SQLite, service admission, Internal Worker, +//! result sink and cleanup. Only model transport and the WorkerStopping barrier +//! are scripted. No process exit or absent map is used as shutdown evidence. +use super::*; + +async fn await_capacity_owner(jobs: &StandaloneJobs, id: &str) { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if jobs + .result_recovery_diagnostic(id) + .is_some_and(|cause| cause.contains("capacity is exhausted")) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} + +fn two_results() -> DomainTestClient { + let client = DomainTestClient::success(); + client.scripts.lock().unwrap().extend( + DomainTestClient::success() + .scripts + .lock() + .unwrap() + .drain(..), + ); + client +} + +#[tokio::test] +async fn restored_results_wait_for_domain_cleanup_then_resume_exact_attempts_without_clicks() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("jobs.sqlite3"); + let prior = StandaloneJobs::open(&path, temp.path().into()).unwrap(); + prior + .request_granted(domain_request("domain-first"), domain_grant("domain-first")) + .unwrap(); + let first = prior.request(request("result-first")).unwrap(); + let second = prior.request(request("result-second")).unwrap(); + drop(prior); + let mut jobs = StandaloneJobs::open(&path, temp.path().into()).unwrap(); + jobs.capacity = 1; + let hook = ControlledShutdown { + block: true, + ..Default::default() + }; + let _release_on_failure = ReleaseShutdownOnDrop(hook.release.clone()); + jobs.bind_domain(Arc::new(hook.clone())).unwrap(); + let model = two_results(); + jobs.bind_model_client(Box::new(model.clone())); + let ids = jobs.reserved_result_attempts().unwrap(); + assert_eq!(ids.len(), 2); + jobs.start_with_model_client("domain-first", DomainTestClient::success()) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), hook.entered.notified()) + .await + .unwrap(); + let domain = jobs.get("domain-first").unwrap(); + assert_eq!(domain.state, JobState::Completed); + assert!(!domain.executions_released()); + let recovery_jobs = jobs.clone(); + let owner = tokio::spawn(async move { recovery_jobs.resume_reserved_result_jobs(ids).await }); + await_capacity_owner(&jobs, "result-first").await; + await_capacity_owner(&jobs, "result-second").await; + assert_eq!(jobs.get("result-first").unwrap(), first); + assert_eq!(jobs.get("result-second").unwrap(), second); + assert!(model.requests.lock().unwrap().is_empty()); + assert!(!owner.is_finished()); + // A reservation accepted after restore capture is not silently swept in. + let outside = jobs.request(request("outside-capture")).unwrap(); + hook.release.notify_one(); + tokio::time::timeout(Duration::from_secs(10), owner) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(jobs.get("domain-first").unwrap().executions_released()); + for original in [first, second] { + let saved = jobs.get(&original.request.job_id).unwrap(); + assert_eq!(saved.state, JobState::Completed); + assert_eq!(saved.request, original.request); + assert_eq!(saved.input_digest, original.input_digest); + assert_eq!(saved.attempt.attempt_id, original.attempt.attempt_id); + assert_eq!(saved.attempt.number, 1); + assert_eq!(saved.executions.len(), 1); + assert!(saved.executions_released()); + assert_eq!(saved.result, Some(json!({"valid":true}))); + assert!( + jobs.result_recovery_diagnostic(&saved.request.job_id) + .is_none() + ); + } + assert_eq!(model.requests.lock().unwrap().len(), 4); + assert!(model.scripts.lock().unwrap().is_empty()); + assert_eq!(jobs.get("outside-capture").unwrap(), outside); + let db = rusqlite::Connection::open(&path).unwrap(); + assert_eq!( + db.query_row("SELECT COUNT(*) FROM job_intents", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 4 + ); + assert_eq!( + db.query_row("SELECT COUNT(*) FROM job_attempts", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 4 + ); + jobs.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn restored_results_save_individual_configuration_and_store_rejections_then_continue() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("jobs.sqlite3"); + let prior = StandaloneJobs::open(&path, temp.path().into()).unwrap(); + let mut invalid = request("bad-profile"); + invalid.profile = "builtin:no-such-job-profile".into(); + let bad_profile = prior.request(invalid).unwrap(); + let bad_store = prior.request(request("store-rejection")).unwrap(); + let good = prior.request(request("later-result")).unwrap(); + drop(prior); + let jobs = StandaloneJobs::open(&path, temp.path().into()).unwrap(); + let db = rusqlite::Connection::open(&path).unwrap(); + db.execute_batch( + "CREATE TRIGGER reject_one_dispatch BEFORE UPDATE OF state ON job_attempts + WHEN NEW.job_id='store-rejection' AND NEW.state='dispatched' + BEGIN SELECT RAISE(FAIL, 'scripted individual dispatch rejection'); END;", + ) + .unwrap(); + let model = DomainTestClient::success(); + jobs.bind_model_client(Box::new(model.clone())); + tokio::time::timeout( + Duration::from_secs(10), + jobs.resume_reserved_result_jobs(jobs.reserved_result_attempts().unwrap()), + ) + .await + .unwrap() + .unwrap(); + for (original, expected) in [ + (bad_profile, "no-such-job-profile"), + (bad_store, "scripted individual dispatch rejection"), + ] { + let saved = jobs.get(&original.request.job_id).unwrap(); + assert_eq!(saved.state, JobState::Failed); + assert_eq!(saved.attempt.attempt_id, original.attempt.attempt_id); + assert_eq!(saved.attempt.number, 1); + assert_eq!(saved.request, original.request); + assert!(saved.attempt.failure.as_deref().unwrap().contains(expected)); + assert!( + saved + .attempt + .failure + .as_deref() + .unwrap() + .starts_with("result_recovery_cannot_start:") + ); + assert!(saved.start_rejected_by.is_some()); + assert!( + saved.executions_released(), + "explicit pre-dispatch rejection, not Failed alone" + ); + assert!( + saved.executions.is_empty(), + "dispatch was rejected, not a cleanup receipt" + ); + } + let saved = jobs.get("later-result").unwrap(); + assert_eq!(saved.state, JobState::Completed); + assert_eq!(saved.attempt.attempt_id, good.attempt.attempt_id); + assert_eq!(saved.executions.len(), 1); + assert_eq!(model.requests.lock().unwrap().len(), 2); + assert_eq!( + db.query_row("SELECT COUNT(*) FROM job_attempts", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 3 + ); + jobs.shutdown().await.unwrap(); +} + +#[tokio::test] +async fn restored_results_continue_after_wait_error_without_forging_cleanup() { + let temp = tempfile::tempdir().unwrap(); + 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 ids = jobs.reserved_result_attempts().unwrap(); + let release = Arc::new(Notify::new()); + let _release_on_failure = ReleaseShutdownOnDrop(release.clone()); + let bad_model = DomainTestClient { + failure_release: Some(release.clone()), + ..DomainTestClient::success() + }; + bad_model.scripts.lock().unwrap().pop_back(); + jobs.start_with_model_client("wait-error", bad_model.clone()) + .await + .unwrap(); + let model = DomainTestClient::success(); + jobs.bind_model_client(Box::new(model.clone())); + 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. + await_model(&model).await; + assert_eq!( + await_job(&jobs, "later-result").await.state, + JobState::Completed + ); + assert_eq!( + jobs.get("later-result").unwrap().attempt.attempt_id, + later.attempt.attempt_id + ); + assert!(!owner.is_finished()); + release.notify_one(); + assert!( + tokio::time::timeout(Duration::from_secs(10), owner) + .await + .unwrap() + .unwrap() + .is_err() + ); + let saved = jobs.get("wait-error").unwrap(); + assert_eq!(saved.state, JobState::Completed); + assert_eq!(saved.attempt.attempt_id, failed.attempt.attempt_id); + assert_eq!(saved.executions.len(), 1); + assert!(!saved.executions_released()); + assert!( + jobs.result_recovery_diagnostic("wait-error") + .unwrap() + .contains("cleanup") + ); + assert_eq!(model.requests.lock().unwrap().len(), 2); + assert_eq!(bad_model.requests.lock().unwrap().len(), 2); + assert!(matches!( + jobs.shutdown().await, + Err(JobServiceError::Cleanup(_)) + )); + assert!(jobs.store.lock().unwrap().is_some()); +} diff --git a/crates/standalone/src/jobs/tests.rs b/crates/standalone/src/jobs/tests.rs index 80d554f0a..351b7072b 100644 --- a/crates/standalone/src/jobs/tests.rs +++ b/crates/standalone/src/jobs/tests.rs @@ -1,3 +1,6 @@ +#[path = "recovery_tests.rs"] +mod recovery_tests; + use super::*; use agen::llm_client::{ClientError, Request, event::Event}; use async_trait::async_trait; @@ -731,7 +734,7 @@ async fn host_restart_rejects_live_lease_then_fails_only_unconfirmed_dispatch_wi reopened.bind_model_client(Box::new(DomainTestClient::success())); tokio::time::timeout( Duration::from_secs(5), - reopened.resume_reserved_result_jobs(reopened.reserved_result_job_ids().unwrap()), + reopened.resume_reserved_result_jobs(reopened.reserved_result_attempts().unwrap()), ) .await .unwrap()