From 4de62650dad111dc6c96c450a8a31fe4c6b1c4ac Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 6 Nov 2026 02:31:16 +0900 Subject: [PATCH] fix(standalone): retain exact consolidation admission through recovery Own accepted never-dispatched consolidation reservations across actual capacity and typed source storage failures. Subscribe before capacity checks, revalidate original binding/attempt/input/grant/retirement and all prior cleanup, join successful resource owners, preserve one deadline and do not redispatch admitted executions. Distinguish settled/admission-required/retained outcomes so Reserved storage recovery transfers the original admission. Persist pre-dispatch refusal using existing rejection evidence with a v8 guard-only migration, retaining immutable v7 text and old Job cutover semantics. Actual migration rollback/retry/reopen/fresh-schema and execution-negative controls; no operating migration or inferred cleanup. Validation: old-source D1 red2/D2 red2; focused11 and nonempty finalizer integration; parent normal standalone159+11+9(ignored1)+9/doc0; downstream connected7; root/doc/fmt/names/diff. Child long-temp SUN_LEN failures retained; same-source full parent package passes under prior short isolated temp root. 1230 actual source hashes unchanged during validation. Fresh whole-current/all-seven-AC formal review pending; no integration or Ticket conclusion. --- crates/standalone/src/job_store.rs | 2 +- .../src/job_store/admission_tests.rs | 130 ++++ crates/standalone/src/job_store/recovery.rs | 60 +- crates/standalone/src/jobs.rs | 49 +- crates/standalone/src/jobs/recovery.rs | 62 +- crates/standalone/src/jobs/tests.rs | 8 +- crates/standalone/src/subjektiv.rs | 158 ++-- crates/standalone/src/subjektiv/admission.rs | 203 +++++ .../src/subjektiv/admission_tests.rs | 701 ++++++++++++++++++ crates/standalone/src/subjektiv/tests.rs | 13 +- 10 files changed, 1297 insertions(+), 89 deletions(-) create mode 100644 crates/standalone/src/job_store/admission_tests.rs create mode 100644 crates/standalone/src/subjektiv/admission.rs create mode 100644 crates/standalone/src/subjektiv/admission_tests.rs diff --git a/crates/standalone/src/job_store.rs b/crates/standalone/src/job_store.rs index 6eae68e32..b30c1251e 100644 --- a/crates/standalone/src/job_store.rs +++ b/crates/standalone/src/job_store.rs @@ -33,7 +33,7 @@ mod recovery; #[path = "job_store/start_rejection_tests.rs"] mod start_rejection_tests; -const SCHEMA_VERSION: i64 = 7; +const SCHEMA_VERSION: i64 = 8; /// Availability of the immutable input body, independent of execution outcome. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/standalone/src/job_store/admission_tests.rs b/crates/standalone/src/job_store/admission_tests.rs new file mode 100644 index 000000000..7e3ce0790 --- /dev/null +++ b/crates/standalone/src/job_store/admission_tests.rs @@ -0,0 +1,130 @@ +//! Actual v7 guard upgrade: rollback/retry, no cutover, exact values and fresh +//! schema equality. Temporary SQLite only; this is not an operating DB migration. +use super::*; + +#[test] +fn granted_admission_guard_upgrade_rolls_back_retries_preserves_reservations_and_reopens() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("jobs.sqlite3"); + let mut store = JobStore::open(&path).unwrap(); + let mut request = super::super::tests::request("granted-refusal"); + request.serialization_key = None; + let granted = store + .reserve_granted(request.clone(), Some(serde_json::json!({"exact":true}))) + .unwrap(); + request.job_id = "prior-execution".into(); + let pending = store.reserve(request).unwrap(); + let dispatched = store + .dispatch(&pending.request.job_id, &pending.attempt.attempt_id, 8) + .unwrap(); + drop(store); + let mut conn = Connection::open(&path).unwrap(); + feature_storage::configure_connection(&conn).unwrap(); + // v8 has no table/row changes. Restore the actual immutable v7 guard text. + let old_guard = &START_REJECTION_SCHEMA[START_REJECTION_SCHEMA + .find("CREATE TRIGGER job_start_rejection_guard") + .unwrap() + ..START_REJECTION_SCHEMA + .find("CREATE TRIGGER job_start_rejection_immutable") + .unwrap()]; + conn.execute_batch("DROP TRIGGER job_start_rejection_guard") + .unwrap(); + conn.execute_batch(old_guard).unwrap(); + conn.pragma_update(None, "user_version", 7).unwrap(); + conn.execute_batch("CREATE TABLE migration_fault(value TEXT REFERENCES job_intents(job_id))") + .unwrap(); + conn.pragma_update(None, "foreign_keys", false).unwrap(); + conn.execute("INSERT INTO migration_fault VALUES('missing')", []) + .unwrap(); + conn.pragma_update(None, "foreign_keys", true).unwrap(); + assert!( + initialize(&mut conn).is_err(), + "actual FK check after DDL must abort the transaction" + ); + assert_eq!( + conn.pragma_query_value(None, "user_version", |row| row.get::<_, i64>(0)) + .unwrap(), + 7 + ); + let guard: String = conn + .query_row( + "SELECT sql FROM sqlite_schema WHERE name='job_start_rejection_guard'", + [], + |row| row.get(0), + ) + .unwrap(); + assert!( + guard.contains("result-only"), + "failed migration must restore old guard" + ); + assert_eq!( + require_snapshot(&conn, &granted.request.job_id).unwrap(), + granted + ); + assert_eq!( + require_snapshot(&conn, &dispatched.request.job_id).unwrap(), + dispatched + ); + conn.execute_batch("DROP TABLE migration_fault").unwrap(); + initialize(&mut conn).unwrap(); + assert_eq!( + require_snapshot(&conn, &granted.request.job_id).unwrap(), + granted + ); + assert_eq!( + require_snapshot(&conn, &dispatched.request.job_id).unwrap(), + dispatched + ); + assert_eq!( + conn.query_row("SELECT count(*) FROM job_version_stops", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + 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([], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, String>(2)?, + )) + }) + .unwrap() + .collect::, _>>() + .unwrap() + }; + let fresh = JobStore::open(&temp.path().join("fresh.sqlite3")).unwrap(); + assert_eq!(schema(&conn), schema(&fresh.conn)); + drop(conn); + let mut reopened = JobStore::open(&path).unwrap(); + reopened + .fail_reserved_granted(&granted, "permanent configuration refusal") + .unwrap(); + let failed = reopened.get(&granted.request.job_id).unwrap(); + assert_eq!( + failed.attempt.failure.as_deref(), + Some("permanent configuration refusal") + ); + assert_eq!(failed.request, granted.request); + assert_eq!(failed.domain_grant, granted.domain_grant); + assert!(failed.start_rejected_by.is_some()); + assert!(failed.executions.is_empty()); + assert_eq!( + reopened.get(&dispatched.request.job_id).unwrap(), + dispatched + ); + assert!(!dispatched.executions_released()); + assert!(reopened.conn.execute("INSERT INTO job_start_rejections(job_id,attempt_id,owner_id) VALUES(?1,?2,'not-an-owner')", + params![dispatched.request.job_id, dispatched.attempt.attempt_id]).is_err()); + drop(reopened); + let reopened = JobStore::open(&path).unwrap(); + assert_eq!(reopened.get(&granted.request.job_id).unwrap(), failed); + assert_eq!( + reopened.get(&dispatched.request.job_id).unwrap(), + dispatched + ); +} diff --git a/crates/standalone/src/job_store/recovery.rs b/crates/standalone/src/job_store/recovery.rs index e48e24cc1..d2faa1389 100644 --- a/crates/standalone/src/job_store/recovery.rs +++ b/crates/standalone/src/job_store/recovery.rs @@ -1,5 +1,9 @@ use super::*; +#[cfg(test)] +#[path = "admission_tests.rs"] +mod admission_tests; + pub(super) const UPDATE_INTERRUPTION: &str = "バージョン更新による実行打切り"; impl JobStore { @@ -52,6 +56,42 @@ impl JobStore { Ok(()) } + /// A pre-dispatch refusal is not execution cleanup. Bind it to the exact + /// available granted reservation and never clear historical execution owners. + pub(crate) fn fail_reserved_granted( + &mut self, + expected: &JobSnapshot, + cause: &str, + ) -> Result<(), JobStoreError> { + let tx = self + .conn + .transaction_with_behavior(TransactionBehavior::Immediate)?; + let saved = require_snapshot(&tx, &expected.request.job_id)?; + check_attempt(&saved, &expected.attempt.attempt_id)?; + saved.load_executable_request()?; + if saved.input_digest != expected.input_digest + || saved.domain_grant != expected.domain_grant + || saved.domain_grant.is_none() + || saved.version_stopped + || saved.state != JobState::Pending + || saved.attempt.state != JobAttemptState::Reserved + || !saved.executions_released() + { + return Err(JobStoreError::AttemptMismatch); + } + tx.execute( + "INSERT INTO job_start_rejections(job_id,attempt_id,owner_id) VALUES(?1,?2,?3)", + params![ + saved.request.job_id, + saved.attempt.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. @@ -205,7 +245,7 @@ pub(super) fn initialize(conn: &mut Connection) -> Result<(), JobStoreError> { } migrate(&tx)?; } - 6 => {} + 6 | 7 => {} SCHEMA_VERSION => {} other => return Err(JobStoreError::UnsupportedSchema(other)), } @@ -214,6 +254,11 @@ pub(super) fn initialize(conn: &mut Connection) -> Result<(), JobStoreError> { // reinterpret preexisting Failed rows as rejected-before-dispatch. tx.execute_batch(START_REJECTION_SCHEMA)?; } + if version <= 7 { + // Extend the existing refusal evidence to explicit domain admissions. + // No Job is cut over, reissued or retroactively declared unstarted. + tx.execute_batch(GRANTED_START_REJECTION_SCHEMA)?; + } if tx.prepare("PRAGMA foreign_key_check")?.exists([])? { return Err(JobStoreError::Corrupt( "Job migration foreign key check failed".into(), @@ -318,6 +363,19 @@ CREATE TRIGGER job_start_rejection_retained BEFORE DELETE ON job_start_rejection BEGIN SELECT RAISE(ABORT,'retained Job start rejection'); END; "; +// v8 changes only the guard. Keep the original v7 migration above intact. +const GRANTED_START_REJECTION_SCHEMA: &str = " +DROP TRIGGER job_start_rejection_guard; +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' + AND i.input_disposition='available') + OR EXISTS(SELECT 1 FROM job_execution_ownership WHERE attempt_id=NEW.attempt_id) + OR EXISTS(SELECT 1 FROM job_execution_ownership WHERE job_id=NEW.job_id AND (writes_revoked=0 OR cleanup_receipt IS NULL)) + OR EXISTS(SELECT 1 FROM job_version_stops WHERE job_id=NEW.job_id) +BEGIN SELECT RAISE(ABORT,'start rejection requires an available unstarted reservation and prior cleanup'); END; +"; + pub(super) const SCHEMA: &str = " CREATE TABLE job_version_history ( job_id TEXT NOT NULL, diff --git a/crates/standalone/src/jobs.rs b/crates/standalone/src/jobs.rs index 80c20e417..2d4139c30 100644 --- a/crates/standalone/src/jobs.rs +++ b/crates/standalone/src/jobs.rs @@ -41,6 +41,8 @@ pub enum JobServiceError { Store(#[from] JobStoreError), #[error("standalone Job execution requirements are unavailable: {0}")] Configuration(String), + #[error(transparent)] + DomainStorage(#[from] feature_storage::FeatureStorageError), #[error("the standalone Job Host is shutting down")] Closed, #[error("standalone Job execution capacity is exhausted")] @@ -82,7 +84,7 @@ pub(crate) trait JobDomainProvider: Send + Sync { &self, snapshot: &JobSnapshot, fence: Arc, - ) -> Result; + ) -> Result; } type OwnedExecutionReport = Arc< @@ -520,6 +522,15 @@ impl StandaloneJobs { &self, job_id: &str, client: Option>, + ) -> Result<(), JobServiceError> { + self.start_with_admission(job_id, client, None) + } + + fn start_with_admission( + &self, + job_id: &str, + client: Option>, + admission: Option<(&JobSnapshot, tokio::time::Instant)>, ) -> Result<(), JobServiceError> { // Preparation owns copies of the immutable input and domain adapter too. Exclude // owner shutdown before creating them, not only before publishing the task. @@ -535,6 +546,16 @@ impl StandaloneJobs { .map(|c| c.clone_boxed()) }); let snapshot = self.get(job_id)?; + if let Some((expected, _)) = admission { + if snapshot.attempt.attempt_id != expected.attempt.attempt_id + || snapshot.input_digest != expected.input_digest + || snapshot.domain_grant != expected.domain_grant + || snapshot.domain_grant.is_none() + || !snapshot.executions_released() + { + return Err(JobStoreError::AttemptMismatch.into()); + } + } snapshot.load_executable_request()?; if snapshot.version_stopped || snapshot.state != JobState::Pending @@ -547,8 +568,19 @@ impl StandaloneJobs { } let effect_gate = Arc::new(Mutex::new(())); let revoked = Arc::new(AtomicBool::new(false)); - let timeout = Duration::from_secs(u64::from(snapshot.request.limits.timeout_seconds)); - let deadline = tokio::time::Instant::now() + timeout; + let deadline = admission.map_or_else( + || { + tokio::time::Instant::now() + + Duration::from_secs(u64::from(snapshot.request.limits.timeout_seconds)) + }, + |(_, deadline)| deadline, + ); + let timeout = deadline.saturating_duration_since(tokio::time::Instant::now()); + if timeout.is_zero() { + return Err(JobServiceError::Configuration( + "consolidation_admission_deadline_exceeded".into(), + )); + } let fence: Arc = Arc::new(BoundAttemptFence { revoked: revoked.clone(), effect_gate: effect_gate.clone(), @@ -569,9 +601,7 @@ impl StandaloneJobs { &snapshot.domain_grant, self.domain.lock().expect("Job domain poisoned").as_ref(), ) { - (Some(_), Some(provider)) => provider - .grant(&snapshot, fence) - .map_err(JobServiceError::Configuration)?, + (Some(_), Some(provider)) => provider.grant(&snapshot, fence)?, (Some(_), None) => { return Err(JobServiceError::Configuration( "persisted Job requires an unavailable explicit domain capability".into(), @@ -625,6 +655,11 @@ impl StandaloneJobs { return Err(JobServiceError::Capacity); } } + if tokio::time::Instant::now() >= deadline { + return Err(JobServiceError::Configuration( + "consolidation_admission_deadline_exceeded during preparation".into(), + )); + } let snapshot = self.with_store_write(|store| { store.dispatch(job_id, &snapshot.attempt.attempt_id, self.capacity) })?; @@ -669,7 +704,7 @@ impl StandaloneJobs { result = &mut run => result, reason = async { tokio::select! { - _ = tokio::time::sleep(timeout) => "timeout", + _ = tokio::time::sleep_until(deadline) => "timeout", _ = cancellation.wait_for(|cancelled| *cancelled) => "Host cancellation/shutdown", } } => { diff --git a/crates/standalone/src/jobs/recovery.rs b/crates/standalone/src/jobs/recovery.rs index dd27fd2e0..f28b6dce9 100644 --- a/crates/standalone/src/jobs/recovery.rs +++ b/crates/standalone/src/jobs/recovery.rs @@ -66,7 +66,52 @@ impl StandaloneJobs { )) } - async fn recovery_closed(&self) { + /// Subscribe before attempting admission; callers must enable the returned + /// notification before checking capacity to retain releases racing that check. + pub(crate) fn admission_changed(&self) -> tokio::sync::futures::OwnedNotified { + self.capacity_changed.clone().notified_owned() + } + + /// A completion notification precedes task return. Join only successful + /// completions before releasing their slot; an error retains actual ownership. + pub(crate) async fn join_completed_admissions(&self) { + 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 id in completed { + let _ = self.wait(&id).await; + } + } + + /// Exact granted admission only, never a lookup of a future attempt by ID. + /// The start lock rechecks input, grant and ALL prior execution cleanup. + pub(crate) fn start_reserved_granted( + &self, + expected: &JobSnapshot, + deadline: tokio::time::Instant, + ) -> Result<(), JobServiceError> { + self.start_with_admission(&expected.request.job_id, None, Some((expected, deadline))) + } + + pub(crate) fn reject_reserved_granted( + &self, + expected: &JobSnapshot, + cause: &str, + ) -> Result<(), JobServiceError> { + let executions = self.executions.lock().expect("Job executions poisoned"); + if executions.closed { + return Err(JobServiceError::Closed); + } + self.with_store_write(|store| store.fail_reserved_granted(expected, cause)) + } + + pub(crate) async fn recovery_closed(&self) { loop { let changed = self.capacity_changed.notified(); tokio::pin!(changed); @@ -95,20 +140,7 @@ impl StandaloneJobs { 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; - } + self.join_completed_admissions().await; let started = recover_owned_settlement( &self.storage_recovered, issue, diff --git a/crates/standalone/src/jobs/tests.rs b/crates/standalone/src/jobs/tests.rs index 4f2ef085e..d30d95232 100644 --- a/crates/standalone/src/jobs/tests.rs +++ b/crates/standalone/src/jobs/tests.rs @@ -121,9 +121,11 @@ impl JobDomainProvider for RecordingDomainProvider { &self, snapshot: &JobSnapshot, fence: Arc, - ) -> Result { + ) -> Result { if snapshot.domain_grant != Some(domain_grant(&snapshot.request.job_id)) { - return Err("foreign or mismatched immutable domain grant".into()); + return Err(JobServiceError::Configuration( + "foreign or mismatched immutable domain grant".into(), + )); } self.0.lock().unwrap().push((snapshot.clone(), fence)); Ok(worker::job::JobFeatureGrant { @@ -201,7 +203,7 @@ impl JobDomainProvider for ControlledShutdown { &self, _: &JobSnapshot, _: Arc, - ) -> Result { + ) -> Result { Ok(worker::job::JobFeatureGrant { features: worker::feature::FeatureRegistryBuilder::new().with_module(self.clone()), satisfied_requirements: vec!["feature.subjektiv"], diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index 51146edc4..bf44521d9 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -18,6 +18,8 @@ //! delivery failures for accepted results do not trigger model reexecution. Memory application //! and surface repair use shared-domain CAS/receipts. +mod admission; +use admission::{ConsolidationAdmission, schedule_job}; mod journal; pub(crate) use journal::prepare_worker_deletion; @@ -277,6 +279,7 @@ impl StandaloneSubjects { self_weak: weak.clone(), journal_dispatch: Mutex::new(false), consolidation_recovery: Mutex::new(std::collections::HashMap::new()), + consolidation_admissions: Mutex::new(std::collections::HashMap::new()), #[cfg(test)] registration_before_attribution: Mutex::new(None), language: self.memory_language.clone(), @@ -386,6 +389,18 @@ enum ConsolidationRecoveryError { } impl ConsolidationRecoveryError { + fn into_service(self) -> crate::jobs::JobServiceError { + use crate::jobs::JobServiceError; + match self { + Self::Service(error) => error, + Self::Domain(subjektiv::api::OperationError::Storage(error)) + | Self::Writer(subjektiv::journal::JournalError::Storage(error)) + | Self::Writer(subjektiv::journal::JournalError::Subject( + subjektiv::SubjektivError::Storage(error), + )) => JobServiceError::DomainStorage(error), + error => JobServiceError::Configuration(error.to_string()), + } + } fn retryable_read_or_storage(&self) -> bool { use crate::jobs::{JobServiceError, JobStoreError}; fn sqlite(error: &rusqlite::Error) -> bool { @@ -413,7 +428,8 @@ impl ConsolidationRecoveryError { match self { Self::Service(JobServiceError::Store(JobStoreError::Sqlite(error))) => sqlite(error), Self::Service(JobServiceError::Store(JobStoreError::Io(error))) => io(error), - Self::Service(JobServiceError::Store(JobStoreError::Storage(error))) + Self::Service(JobServiceError::DomainStorage(error)) + | Self::Service(JobServiceError::Store(JobStoreError::Storage(error))) | Self::Domain(subjektiv::api::OperationError::Storage(error)) | Self::Writer(subjektiv::journal::JournalError::Storage(error)) | Self::Writer(subjektiv::journal::JournalError::Subject( @@ -433,10 +449,18 @@ enum ConsolidationRecoveryProgress { Blocked, } +#[derive(Debug, PartialEq, Eq)] +enum ConsolidationRecovery { + Settled, + AdmissionRequired, + Retained, +} + struct ConsolidationRecoveryOwner { progress: ConsolidationRecoveryProgress, passes: usize, error: ConsolidationRecoveryError, + admission_deadline: tokio::time::Instant, } impl ConsolidationRecoveryOwner { @@ -463,6 +487,8 @@ pub(crate) struct LocalSubjectHost { // grant and its unchanged input remain the durable obligation, not this map. consolidation_recovery: Mutex>, + consolidation_admissions: + Mutex>, #[cfg(test)] registration_before_attribution: Mutex>>, catalog: StandaloneSubjects, @@ -844,17 +870,26 @@ impl LocalSubjectHost { { continue; } + let admission_deadline = tokio::time::Instant::now() + + std::time::Duration::from_secs(u64::from( + job.request.limits.timeout_seconds, + )); match self.recover_consolidation_job(&jobs, &job) { - Ok(true) => {} - Ok(false) - if job.state == crate::jobs::JobState::Pending - && job.attempt.state == crate::jobs::JobAttemptState::Reserved - && job.executions_released() => - { - schedule_job(jobs.clone(), job.request.job_id, self.self_weak.clone()); + Ok(ConsolidationRecovery::AdmissionRequired) => { + schedule_job( + jobs.clone(), + job, + self.self_weak.clone(), + Some(admission_deadline), + ); } - Ok(false) => {} // Missing receipts never authorize replay of effects. - Err(error) => self.own_consolidation_recovery(jobs.clone(), job, error), + Ok(ConsolidationRecovery::Settled | ConsolidationRecovery::Retained) => {} + Err(error) => self.own_consolidation_recovery_until( + jobs.clone(), + job, + error, + admission_deadline, + ), } } } @@ -887,7 +922,7 @@ impl LocalSubjectHost { &self, jobs: &crate::jobs::StandaloneJobs, observed: &crate::jobs::JobSnapshot, - ) -> Result { + ) -> Result { self.ensure_consolidation_live()?; let current = jobs.get(&observed.request.job_id)?; if current.attempt.attempt_id != observed.attempt.attempt_id @@ -927,13 +962,22 @@ impl LocalSubjectHost { result, })?; jobs.acknowledge(&settled.request.job_id)?; - return Ok(true); + return Ok(ConsolidationRecovery::Settled); } if current.state != crate::jobs::JobState::Failed || current.attempt.state != crate::jobs::JobAttemptState::Failed || !current.executions_released() { - return Ok(false); + return Ok( + if current.state == crate::jobs::JobState::Pending + && current.attempt.state == crate::jobs::JobAttemptState::Reserved + && current.executions_released() + { + ConsolidationRecovery::AdmissionRequired + } else { + ConsolidationRecovery::Retained + }, + ); } // This inspection reads original candidate/disposition and preparation // records transactionally. Missing preparation is valid; stale preparation, @@ -974,7 +1018,7 @@ impl LocalSubjectHost { ¤t.attempt.attempt_id, &format!("consolidation-abandoned-v1:{reference}"), )?; - Ok(true) + Ok(ConsolidationRecovery::Settled) } /// Retain only typed temporary storage failures of this exact operation. @@ -986,6 +1030,18 @@ impl LocalSubjectHost { jobs: crate::jobs::StandaloneJobs, observed: crate::jobs::JobSnapshot, error: ConsolidationRecoveryError, + ) { + let admission_deadline = tokio::time::Instant::now() + + std::time::Duration::from_secs(u64::from(observed.request.limits.timeout_seconds)); + self.own_consolidation_recovery_until(jobs, observed, error, admission_deadline); + } + + fn own_consolidation_recovery_until( + &self, + jobs: crate::jobs::StandaloneJobs, + observed: crate::jobs::JobSnapshot, + error: ConsolidationRecoveryError, + admission_deadline: tokio::time::Instant, ) { let key = ( observed.request.job_id.clone(), @@ -1014,7 +1070,9 @@ impl LocalSubjectHost { }, passes: 1, error, + admission_deadline, }; + let admission_deadline = owner.admission_deadline; eprintln!( "standalone consolidation Job {} reconciliation {:?} after {} passes: {}; next action: {}", key.0, @@ -1039,14 +1097,24 @@ impl LocalSubjectHost { retained_delay = (retained_delay * 2).min(std::time::Duration::from_secs(30)); delay }; - tokio::time::sleep(delay).await; + tokio::select! { + biased; + _ = jobs.recovery_closed() => { + host.consolidation_recovery.lock().expect("consolidation recovery poisoned").remove(&key); + return; + } + _ = tokio::time::sleep(delay) => {} + } pass = pass.saturating_add(1); let result = host.recover_consolidation_job(&jobs, &observed); let mut owners = host.consolidation_recovery.lock().expect("consolidation recovery poisoned"); match result { - Ok(_) => { + Ok(outcome) => { owners.remove(&key); drop(owners); + if outcome == ConsolidationRecovery::AdmissionRequired { + schedule_job(jobs.clone(), observed.clone(), host.self_weak.clone(), Some(admission_deadline)); + } host.wake_journals(); return; } @@ -1496,7 +1564,7 @@ impl SubjektivHost for LocalSubjectHost { .is_some_and(|g| g.binding == self.binding) }) { - schedule_job(jobs, pending.request.job_id, self.self_weak.clone()); + schedule_job(jobs, pending, self.self_weak.clone(), None); let (count, bytes) = self .catalog .store @@ -1559,7 +1627,7 @@ impl SubjektivHost for LocalSubjectHost { .map_err(|e| SubjektivHostError::Unavailable(e.to_string()))?, ) .map_err(|e| SubjektivHostError::Unavailable(e.to_string()))?; - schedule_job(jobs, snapshot.request.job_id, self.self_weak.clone()); + schedule_job(jobs, snapshot, self.self_weak.clone(), None); Ok(memory::backend::MemoryConsolidationOutput { status:"started".into(),summary:"Requested a durable generic standalone consolidation Job; processing stops on Host shutdown".into(),candidate_count:count,total_bytes:bytes }) } } @@ -1582,15 +1650,21 @@ impl crate::jobs::JobDomainProvider for LocalDomainProvider { &self, snapshot: &crate::jobs::JobSnapshot, fence: Arc, - ) -> Result { - let host = self.0.upgrade().ok_or("Subject Host unavailable")?; - host.ensure_live().map_err(|e| e.to_string())?; + ) -> Result { + let host = self.0.upgrade().ok_or_else(|| { + crate::jobs::JobServiceError::Configuration("Subject Host unavailable".into()) + })?; if snapshot.request.purpose == journal::PURPOSE { - return journal::grant(host, snapshot, fence); + host.ensure_live() + .map_err(|error| crate::jobs::JobServiceError::Configuration(error.to_string()))?; + return journal::grant(host, snapshot, fence) + .map_err(crate::jobs::JobServiceError::Configuration); } + host.ensure_consolidation_live() + .map_err(ConsolidationRecoveryError::into_service)?; let grant = host .consolidation_grant(snapshot) - .map_err(|error| error.to_string())?; + .map_err(ConsolidationRecoveryError::into_service)?; let adapter = Arc::new(LocalConsolidationHost { host, grant, @@ -1605,11 +1679,11 @@ impl crate::jobs::JobDomainProvider for LocalDomainProvider { session_id: format!("job:{}", snapshot.attempt.attempt_id), }, ) - .map_err(|e| e.to_string())?; + .map_err(|error| crate::jobs::JobServiceError::Configuration(error.to_string()))?; let feature = worker::feature::builtin::memory::SubjektivConsolidationToolsFeature::new( connection.clone(), ) - .map_err(|e| e.to_string())?; + .map_err(|error| crate::jobs::JobServiceError::Configuration(error.to_string()))?; Ok(worker::job::JobFeatureGrant { features: worker::feature::FeatureRegistryBuilder::new().with_module(feature), satisfied_requirements: vec!["feature.subjektiv"], @@ -1815,40 +1889,6 @@ impl worker::job::JobResultFinalizer for LocalConsolidationFinalizer { } } -fn schedule_job( - jobs: crate::jobs::StandaloneJobs, - id: String, - host: std::sync::Weak, -) { - // Scheduling/consumer acknowledgement has no Internal Worker ownership. - // The generic service fences and retains every execution until cleanup. - tokio::spawn(async move { - if let Err(error) = jobs.start(&id).await { - eprintln!("standalone Subject Job {id} remains pending: {error}"); - return; - } - match jobs.wait(&id).await { - Ok(result) if result.state == crate::jobs::JobState::Completed => { - let _ = jobs.acknowledge(&id); - } - Ok(result) if result.state == crate::jobs::JobState::Failed => { - if let Some(host) = host.upgrade() { - if let Err(error) = host.recover_consolidation_job(&jobs, &result) { - host.own_consolidation_recovery(jobs.clone(), result, error); - } - } - } - Ok(_) => {} - Err(error) => eprintln!( - "standalone Subject Job {id} cleanup/acknowledgement is unconfirmed: {error}" - ), - } - if let Some(host) = host.upgrade() { - host.wake_journals(); - } - }); -} - #[cfg(test)] mod tests; diff --git a/crates/standalone/src/subjektiv/admission.rs b/crates/standalone/src/subjektiv/admission.rs new file mode 100644 index 000000000..2077bfc16 --- /dev/null +++ b/crates/standalone/src/subjektiv/admission.rs @@ -0,0 +1,203 @@ +//! Ownership of never-dispatched consolidation admissions. The durable obligation +//! is the original Job reservation, not a new queue or a retry execution attempt. +use super::*; +use crate::jobs::{JobServiceError, JobSnapshot, JobState, JobStoreError, StandaloneJobs}; +use std::time::Duration; +use tokio::time::Instant; + +pub(super) struct ConsolidationAdmission { + pub(super) deadline: Instant, + pub(super) issue: Option, + pub(super) blocked: bool, +} + +pub(super) fn schedule_job( + jobs: StandaloneJobs, + observed: JobSnapshot, + host: std::sync::Weak, + deadline: Option, +) { + let Some(host) = host.upgrade() else { + return; + }; + let key = ( + observed.request.job_id.clone(), + observed.attempt.attempt_id.clone(), + observed.input_digest.clone(), + ); + let deadline = deadline.unwrap_or_else(|| { + Instant::now() + Duration::from_secs(u64::from(observed.request.limits.timeout_seconds)) + }); + // An overlapping restore/request must not refill the earlier storage owner's + // admission budget while that exact operation is waiting to hand off. + let deadline = host + .consolidation_recovery + .lock() + .expect("consolidation recovery poisoned") + .get(&key) + .map_or(deadline, |owner| owner.admission_deadline.min(deadline)); + { + let mut owners = host + .consolidation_admissions + .lock() + .expect("consolidation admissions poisoned"); + if owners.contains_key(&key) { + return; + } + owners.insert( + key.clone(), + ConsolidationAdmission { + deadline, + issue: None, + blocked: false, + }, + ); + } + tokio::spawn(async move { + let outcome = tokio::select! { + biased; + _ = jobs.recovery_closed() => Ok(false), + outcome = host.admit_consolidation(&jobs, &observed, &key) => outcome, + }; + match outcome { + Err(error) => { + eprintln!( + "standalone consolidation admission {}/{} blocked: {error}; original input retained", + key.0, key.1 + ); + let mut owners = host + .consolidation_admissions + .lock() + .expect("consolidation admissions poisoned"); + let owner = owners.get_mut(&key).expect("owned consolidation admission"); + owner.issue = Some(error); + owner.blocked = true; + return; + } + Ok(started) => { + // Dispatch is never retried after this point. Execution/cleanup and + // settlement remain owned by the existing generic Job service. + if started { + match jobs.wait(&observed.request.job_id).await { + Ok(result) if result.state == JobState::Completed => { + if host.ensure_consolidation_live().is_ok() { + let _ = jobs.acknowledge(&result.request.job_id); + } + } + Ok(result) if result.state == JobState::Failed => { + if let Err(error) = host.recover_consolidation_job(&jobs, &result) { + host.own_consolidation_recovery(jobs.clone(), result, error); + } + } + Ok(_) => {} + Err(error) => eprintln!( + "standalone Subject Job {} cleanup/acknowledgement is unconfirmed: {error}", + key.0 + ), + } + } + host.consolidation_admissions + .lock() + .expect("consolidation admissions poisoned") + .remove(&key); + host.wake_journals(); + } + } + }); +} + +impl LocalSubjectHost { + async fn admit_consolidation( + &self, + jobs: &StandaloneJobs, + observed: &JobSnapshot, + key: &(String, String, String), + ) -> Result { + let deadline = self + .consolidation_admissions + .lock() + .expect("consolidation admissions poisoned") + .get(key) + .expect("owned consolidation admission") + .deadline; + let mut delay = Duration::from_millis(100); + // Once preparation genuinely refuses, retry ONLY persistence of that + // refusal, not preparation/model execution after an unrelated recovery. + let mut refusal: Option = None; + loop { + let changed = jobs.admission_changed(); + tokio::pin!(changed); + changed.as_mut().enable(); // BEFORE capacity checks, including task return. + jobs.join_completed_admissions().await; + let attempt = (|| { + self.ensure_consolidation_live()?; + match self.recover_consolidation_job(jobs, observed)? { + ConsolidationRecovery::AdmissionRequired => {} + ConsolidationRecovery::Settled | ConsolidationRecovery::Retained => { + return Ok(false); + } + } + if Instant::now() >= deadline { + refusal.get_or_insert_with(|| "consolidation_admission_deadline_exceeded: original reservation deadline elapsed before dispatch".into()); + } + if refusal.is_none() { + match jobs.start_reserved_granted(observed, deadline) { + Ok(()) => return Ok(true), + Err(JobServiceError::Configuration(cause)) => { + refusal = + Some(format!("consolidation_admission_cannot_start: {cause}")); + } + Err(error) => return Err(error.into()), + } + } + // Fresh exact snapshot, real Host binding, original input/grant and + // ALL cleanup were just checked. This write proves no runner cleanup. + self.ensure_consolidation_live()?; + jobs.reject_reserved_granted( + observed, + refusal.as_deref().expect("admission refusal"), + )?; + let failed = jobs.get(&observed.request.job_id)?; + if let Err(error) = self.recover_consolidation_job(jobs, &failed) { + self.own_consolidation_recovery(jobs.clone(), failed, error); + } + Ok(false) + })(); + match attempt { + Ok(started) => return Ok(started), + Err(error) => { + let capacity = matches!( + &error, + ConsolidationRecoveryError::Service( + JobServiceError::Capacity + | JobServiceError::Store( + JobStoreError::ConcurrentLimit + | JobStoreError::SerializationConflict(_) + ) + ) + ); + if !capacity && !error.retryable_read_or_storage() { + return Err(error); + } + self.consolidation_admissions + .lock() + .expect("consolidation admissions poisoned") + .get_mut(key) + .expect("owned consolidation admission") + .issue = Some(error); + if capacity { + tokio::select! { + _ = &mut changed => {} + _ = tokio::time::sleep_until(deadline) => {} + } + } else { + // Retry the SAME operation against its actual storage source. + // No healthy-other-store probe and no new dispatch attempt. + tokio::time::sleep(delay).await; + delay = (delay * 2).min(Duration::from_secs(30)); + } + } + } + } + } +} diff --git a/crates/standalone/src/subjektiv/admission_tests.rs b/crates/standalone/src/subjektiv/admission_tests.rs new file mode 100644 index 000000000..8e8b88dff --- /dev/null +++ b/crates/standalone/src/subjektiv/admission_tests.rs @@ -0,0 +1,701 @@ +//! Real Host admission with real Job/domain SQLite stores. Only model responses +//! and the held Job's release timing are substituted; no process-death proof. +use super::*; +use agen::llm_client::{ClientError, LlmClient, Request, event::Event, types::Item}; +use std::sync::atomic::AtomicUsize; +use std::time::Duration; + +#[derive(Clone, Default)] +struct AdmissionModel { + clones: Arc, + entered: Arc, + release: Arc, + submissions: Arc>>, +} +#[async_trait::async_trait] +impl LlmClient for AdmissionModel { + fn clone_boxed(&self) -> Box { + self.clones.fetch_add(1, Ordering::SeqCst); + Box::new(self.clone()) + } + async fn stream( + &self, + request: Request, + ) -> Result { + let envelope = request + .items + .iter() + .filter_map(Item::as_text) + .find(|text| text.contains("input_json: ")) + .expect("immutable Job input"); + let id = envelope + .lines() + .find_map(|line| line.strip_prefix("job_id: ")) + .unwrap(); + let input: serde_json::Value = serde_json::from_str( + envelope + .lines() + .find_map(|line| line.strip_prefix("input_json: ")) + .unwrap(), + ) + .unwrap(); + let events = if request.items.iter().any(|item| { + matches!(item, + Item::ToolResult { call_id, is_error: false, .. } if call_id == "admission-result") + }) { + vec![ + Event::text_block_start(0), + Event::text_delta(0, "done"), + Event::text_block_stop(0, None), + ] + } else { + { + let mut submissions = self.submissions.lock().unwrap(); + assert!( + !submissions.iter().any(|prior| prior == id), + "duplicate execution/model submission" + ); + submissions.push(id.to_owned()); + } + if id == "held-capacity" { + self.entered.notify_one(); + self.release.notified().await; + } + vec![ + Event::tool_use_start(0, "admission-result", "SubmitJobResult"), + Event::tool_input_delta(0, &serde_json::json!({"result": input}).to_string()), + Event::tool_use_stop(0), + ] + }; + Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok)))) + } +} + +struct AdmissionFixture { + body: SubjectConnection, + jobs: crate::jobs::StandaloneJobs, + model: AdmissionModel, + _temp: tempfile::TempDir, +} +impl Drop for AdmissionFixture { + fn drop(&mut self) { + self.body.host.live.store(false, Ordering::Release); + } +} +impl AdmissionFixture { + fn new() -> Self { + let temp = tempfile::tempdir().unwrap(); + let catalog = StandaloneSubjects::open(temp.path().join("state")).unwrap(); + let subject = catalog.create("admission", None).unwrap(); + let body = catalog + .connect(&subject.id, protocol::WorkerId::now_v7()) + .unwrap(); + let jobs = + crate::jobs::StandaloneJobs::open(&body.jobs_path, temp.path().to_path_buf()).unwrap(); + let model = AdmissionModel::default(); + jobs.bind_model_client(Box::new(model.clone())); + body.host.bind_jobs(jobs.clone()).unwrap(); + Self { + body, + jobs, + model, + _temp: temp, + } + } + fn reserve(&self) -> crate::jobs::JobSnapshot { + self.reserve_with_profile("builtin:standalone-subjektiv-consolidation") + } + fn reserve_with_profile(&self, profile: &str) -> crate::jobs::JobSnapshot { + self.jobs.request_granted(job::JobRequest { + job_id: "reserved-consolidation".into(), purpose: "subjektiv_consolidation".into(), + input_ref: format!("subjektiv://{}/consolidation", self.body.binding.subject_id), + input: serde_json::json!({"subject_id": self.body.binding.subject_id, "candidate_ids": []}), + instruction: "Return the exact immutable input with SubmitJobResult".into(), + profile: profile.into(), + serialization_key: Some(format!("subjektiv:{}", self.body.binding.subject_id)), + limits: Default::default(), + }, serde_json::to_value(LocalConsolidationGrant { + binding: self.body.binding.clone(), candidate_ids: vec![], + }).unwrap()).unwrap() + } + fn domain_path(&self) -> PathBuf { + self.body + .host + .catalog + .root + .join("features") + .join(&self.body.host.catalog.scope_id) + .join("features/subjektiv.sqlite") + } + async fn hold_capacity(&self) { + self.jobs + .request(job::JobRequest { + job_id: "held-capacity".into(), + purpose: "test".into(), + input_ref: "fixture:held".into(), + input: serde_json::json!({"held": true}), + instruction: "Return the input".into(), + profile: "builtin:job".into(), + serialization_key: None, + limits: job::JobLimits { + max_concurrent_jobs: 1, + ..Default::default() + }, + }) + .unwrap(); + self.jobs.start("held-capacity").await.unwrap(); + tokio::time::timeout(Duration::from_secs(10), self.model.entered.notified()) + .await + .unwrap(); + let held = self.jobs.get("held-capacity").unwrap(); + assert_eq!(held.attempt.state, crate::jobs::JobAttemptState::Dispatched); + assert!(!held.executions_released()); + } + async fn assert_completed_once(&self, before: &crate::jobs::JobSnapshot) { + let id = &before.request.job_id; + until( + "same reserved consolidation must progress after ONLY fault/capacity release", + || { + self.jobs.get(id).is_ok_and(|saved| { + saved.state == crate::jobs::JobState::Completed + && saved.acknowledged + && saved.executions_released() + }) + }, + ) + .await; + let after = self.jobs.wait(id).await.unwrap(); + assert_eq!(after.request, before.request); + assert_eq!(after.input_digest, before.input_digest); + assert_eq!(after.domain_grant, before.domain_grant); + assert_eq!(after.attempt.attempt_id, before.attempt.attempt_id); + assert_eq!(after.attempt.number, before.attempt.number); + assert_eq!(after.executions.len(), 1); + assert!(after.result.as_ref().unwrap().get("surface").is_some()); + assert_eq!( + self.model + .submissions + .lock() + .unwrap() + .iter() + .filter(|prior| *prior == id) + .count(), + 1 + ); + let job_db = rusqlite::Connection::open(&self.body.jobs_path).unwrap(); + let attempts: i64 = job_db + .query_row( + "SELECT count(*) FROM job_attempts WHERE job_id=?1", + [id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(attempts, 1); + let domain = rusqlite::Connection::open(self.domain_path()).unwrap(); + let generations: i64 = domain + .query_row( + "SELECT count(*) FROM surface_generation_runs WHERE job_id=?1 AND attempt_id=?2", + rusqlite::params![id, before.attempt.attempt_id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(generations, 1, "finalizer must prepare exactly once"); + assert_eq!( + subjektiv::job::lookup_result( + &self.body.host.catalog.store, + &self.body.binding.subject_id, + &[], + id, + &before.attempt.attempt_id + ) + .unwrap(), + after.result + ); + } +} +async fn until(message: &str, mut check: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(10), async { + while !check() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect(message); +} + +#[tokio::test] +async fn consolidation_admission_capacity_only_release_starts_original_request() { + capacity_only_release(false).await; +} +#[tokio::test] +async fn consolidation_admission_capacity_only_release_starts_restored_reservation() { + capacity_only_release(true).await; +} +async fn capacity_only_release(restore: bool) { + let fixture = AdmissionFixture::new(); + fixture.hold_capacity().await; + let clones = fixture.model.clones.load(Ordering::SeqCst); + let before = if restore { + let before = fixture.reserve(); + fixture.body.host.recover_jobs(); + before + } else { + let result = fixture + .body + .host + .request_consolidation( + &SubjektivHostContext { + worker_id: fixture.body.host.worker_key.clone(), + session_id: session_store::new_session_id().to_string(), + }, + memory::backend::MemoryConsolidateStagingOperation { force: true }, + ) + .unwrap(); + assert_eq!(result.status, "started"); + fixture + .jobs + .pending() + .unwrap() + .into_iter() + .find(|saved| saved.domain_grant.is_some()) + .unwrap() + }; + // Preparation clones the configured transport before checking actual capacity. + // This barrier proves the production scheduler attempted admission while held. + until("scheduler must reach capacity check", || { + fixture.model.clones.load(Ordering::SeqCst) > clones + }) + .await; + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert!(before.executions.is_empty()); + assert!(matches!( + fixture.jobs.start(&before.request.job_id).await, + Err(crate::jobs::JobServiceError::Capacity) + )); + fixture.model.release.notify_one(); + let held = tokio::time::timeout(Duration::from_secs(10), fixture.jobs.wait("held-capacity")) + .await + .unwrap() + .unwrap(); + assert!(held.executions_released()); + fixture.assert_completed_once(&before).await; +} + +#[tokio::test] +async fn consolidation_admission_reserved_domain_busy_release_before_fast_budget() { + reserved_domain_busy_release(false).await; +} +#[tokio::test] +async fn consolidation_admission_reserved_domain_busy_release_beyond_fast_budget() { + reserved_domain_busy_release(true).await; +} +async fn reserved_domain_busy_release(beyond: bool) { + let fixture = AdmissionFixture::new(); + let before = fixture.reserve(); + let connection = rusqlite::Connection::open(fixture.domain_path()).unwrap(); + connection.execute_batch("BEGIN IMMEDIATE").unwrap(); + // Production restore entry, not a manually armed retry or failed-attempt fixture. + fixture.body.host.recover_jobs(); + let required_passes = if beyond { 8 } else { 1 }; + // Real SQLite busy_timeout makes eight attempts longer than the short wait. + tokio::time::timeout(Duration::from_secs(60), async { + loop { + let owners = fixture.body.host.consolidation_recovery.lock().unwrap(); + if let Some(owner) = owners + .values() + .find(|owner| owner.passes >= required_passes) + { + super::assert_real_busy(&owner.error, true); + if beyond { + assert_eq!( + owner.progress, + ConsolidationRecoveryProgress::RetainedStorage + ); + } + break; + } + drop(owners); + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("restore must own the actual reserved-domain Busy operation"); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert!(fixture.model.submissions.lock().unwrap().is_empty()); + eprintln!( + "D2 actual reserved Feature Busy retained; beyond_fast_budget={beyond}; rollback ONLY" + ); + connection.execute_batch("ROLLBACK").unwrap(); + // No restore/recover_jobs/reclick/new request after releasing the SQLite fault. + fixture.assert_completed_once(&before).await; +} + +async fn waiting_capacity(fixture: &AdmissionFixture) { + until( + "exact admission must own the actual capacity refusal", + || { + fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .values() + .any(|owner| { + matches!( + owner.issue, + Some(ConsolidationRecoveryError::Service( + crate::jobs::JobServiceError::Capacity + )) + ) + }) + }, + ) + .await; +} + +#[tokio::test] +async fn consolidation_admission_duplicate_notifications_preserve_deadline_and_never_forge_cleanup() +{ + let fixture = AdmissionFixture::new(); + fixture.hold_capacity().await; + let before = fixture.reserve(); + let deadline = tokio::time::Instant::now() + Duration::from_millis(300); + schedule_job( + fixture.jobs.clone(), + before.clone(), + fixture.body.host.self_weak.clone(), + Some(deadline), + ); + waiting_capacity(&fixture).await; + // A late notification carrying a fresh timeout must not reset the original. + schedule_job( + fixture.jobs.clone(), + before.clone(), + fixture.body.host.self_weak.clone(), + Some(deadline + Duration::from_secs(60)), + ); + { + let owners = fixture.body.host.consolidation_admissions.lock().unwrap(); + assert_eq!(owners.len(), 1); + assert_eq!(owners.values().next().unwrap().deadline, deadline); + } + until( + "the original admission deadline must persist a specific pre-dispatch failure", + || { + fixture.jobs.get(&before.request.job_id).unwrap().state == crate::jobs::JobState::Failed + && fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .is_empty() + }, + ) + .await; + let failed = fixture.jobs.get(&before.request.job_id).unwrap(); + assert!( + failed + .attempt + .failure + .as_deref() + .unwrap() + .contains("consolidation_admission_deadline_exceeded") + ); + assert_eq!(failed.attempt.attempt_id, before.attempt.attempt_id); + assert_eq!(failed.domain_grant, before.domain_grant); + assert_eq!(failed.input_digest, before.input_digest); + assert!(failed.executions.is_empty()); + assert!(failed.start_rejected_by.is_some()); + assert!( + !fixture + .jobs + .get("held-capacity") + .unwrap() + .executions_released(), + "refusal must not invent prior cleanup" + ); + assert_eq!( + *fixture.model.submissions.lock().unwrap(), + ["held-capacity"] + ); + fixture.model.release.notify_one(); + assert!( + fixture + .jobs + .wait("held-capacity") + .await + .unwrap() + .executions_released() + ); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), failed); +} + +#[tokio::test] +async fn consolidation_admission_capacity_wait_rechecks_retirement_binding_and_version_stop() { + for case in ["erased", "version_stopped", "inactive", "released_writer"] { + let fixture = AdmissionFixture::new(); + fixture.hold_capacity().await; + let before = fixture.reserve(); + fixture.body.host.recover_jobs(); + waiting_capacity(&fixture).await; + match case { + "erased" => { + fixture + .jobs + .reclaim_input( + &before.request.job_id, + &before.input_digest, + crate::jobs::JobInputRetirementReason::ExplicitErasure, + ) + .unwrap(); + } + "version_stopped" => { + rusqlite::Connection::open(&fixture.body.jobs_path) + .unwrap() + .execute( + "INSERT INTO job_version_stops(job_id) VALUES(?1)", + [&before.request.job_id], + ) + .unwrap(); + } + "inactive" => fixture.body.host.live.store(false, Ordering::Release), + "released_writer" => fixture + .body + .host + .catalog + .store + .release_journal_writer(&fixture.body.host.writer) + .unwrap(), + _ => unreachable!(), + } + let expected = fixture.jobs.get(&before.request.job_id).unwrap(); + fixture.model.release.notify_one(); + fixture.jobs.wait("held-capacity").await.unwrap(); + until( + "capacity wake must retain the specific refusal without dispatch", + || { + fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .values() + .any(|owner| owner.blocked) + }, + ) + .await; + assert_eq!( + fixture.jobs.get(&before.request.job_id).unwrap(), + expected, + "{case}" + ); + assert!(expected.executions.is_empty(), "{case}"); + assert_eq!( + *fixture.model.submissions.lock().unwrap(), + ["held-capacity"], + "{case}" + ); + } +} + +#[tokio::test] +async fn consolidation_admission_stale_attempt_input_or_grant_never_starts_current_reservation() { + for case in ["attempt", "digest", "grant"] { + let fixture = AdmissionFixture::new(); + let before = fixture.reserve(); + let mut stale = before.clone(); + match case { + "attempt" => stale.attempt.attempt_id = "different-attempt".into(), + "digest" => stale.input_digest = "different-digest".into(), + "grant" => stale.domain_grant = Some(serde_json::json!({"foreign":true})), + _ => unreachable!(), + } + schedule_job( + fixture.jobs.clone(), + stale, + fixture.body.host.self_weak.clone(), + None, + ); + until( + "stale exact admission must be diagnosed, never followed by ID", + || { + fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .values() + .any(|owner| owner.blocked) + }, + ) + .await; + assert_eq!( + fixture.jobs.get(&before.request.job_id).unwrap(), + before, + "{case}" + ); + assert!( + fixture.model.submissions.lock().unwrap().is_empty(), + "{case}" + ); + } +} + +#[tokio::test] +async fn consolidation_admission_shutdown_releases_waiter_without_dispatch_or_cleanup_claim() { + let fixture = AdmissionFixture::new(); + fixture.hold_capacity().await; + let before = fixture.reserve(); + fixture.body.host.recover_jobs(); + waiting_capacity(&fixture).await; + // Real service shutdown closes admission and signals the actual held runner. + fixture.jobs.close(); + until("shutdown must release its volatile admission owner", || { + fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .is_empty() + }) + .await; + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert_eq!( + *fixture.model.submissions.lock().unwrap(), + ["held-capacity"] + ); + fixture.model.release.notify_one(); + tokio::time::timeout(Duration::from_secs(10), fixture.jobs.wait("held-capacity")) + .await + .unwrap() + .unwrap(); +} + +#[tokio::test] +async fn consolidation_admission_permanent_preparation_refusal_is_saved_without_execution() { + let fixture = AdmissionFixture::new(); + // A real existing Profile that does not satisfy consolidation's explicit + // policy requirement; no fake transport/storage error stands in for refusal. + let before = fixture.reserve_with_profile("builtin:job"); + fixture.body.host.recover_jobs(); + until( + "permanent preparation refusal must become a specific durable failure", + || { + fixture.jobs.get(&before.request.job_id).unwrap().state == crate::jobs::JobState::Failed + && fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .is_empty() + }, + ) + .await; + let failed = fixture.jobs.get(&before.request.job_id).unwrap(); + assert!( + failed + .attempt + .failure + .as_deref() + .unwrap() + .contains("consolidation_admission_cannot_start") + ); + assert!(failed.start_rejected_by.is_some()); + assert!(failed.executions.is_empty()); + assert_eq!(failed.request, before.request); + assert_eq!(failed.domain_grant, before.domain_grant); + assert_eq!(failed.input_digest, before.input_digest); + assert_eq!(failed.attempt.attempt_id, before.attempt.attempt_id); + assert!(fixture.model.submissions.lock().unwrap().is_empty()); + assert!( + crate::job_store::JobStore::open(&fixture.body.jobs_path) + .unwrap() + .execution_obligations() + .unwrap() + .is_empty() + ); +} + +#[tokio::test] +async fn consolidation_admission_reserved_storage_handoff_preserves_deadline_while_capacity_held() { + let fixture = AdmissionFixture::new(); + fixture.hold_capacity().await; + let before = fixture.reserve(); + let connection = rusqlite::Connection::open(fixture.domain_path()).unwrap(); + connection.execute_batch("BEGIN IMMEDIATE").unwrap(); + fixture.body.host.recover_jobs(); + let deadline = { + let owners = fixture.body.host.consolidation_recovery.lock().unwrap(); + assert_eq!(owners.len(), 1); + let owner = owners.values().next().unwrap(); + super::assert_real_busy(&owner.error, true); + owner.admission_deadline + }; + connection.execute_batch("ROLLBACK").unwrap(); + waiting_capacity(&fixture).await; + { + let owners = fixture.body.host.consolidation_admissions.lock().unwrap(); + assert_eq!(owners.len(), 1); + assert_eq!(owners.values().next().unwrap().deadline, deadline); + } + // Duplicate restoration while capacity remains held cannot reset the budget. + fixture.body.host.recover_jobs(); + assert_eq!( + fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .values() + .next() + .unwrap() + .deadline, + deadline + ); + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + fixture.model.release.notify_one(); + fixture.jobs.wait("held-capacity").await.unwrap(); + fixture.assert_completed_once(&before).await; +} + +#[tokio::test] +async fn consolidation_admission_missing_prior_cleanup_never_dispatches_or_creates_release_proof() { + let fixture = super::ConsolidationRecoveryFixture::new(); + let before = fixture.saved.clone(); + assert!(!before.executions_released()); + schedule_job( + fixture.jobs.clone(), + before.clone(), + fixture.body.host.self_weak.clone(), + None, + ); + until( + "non-reserved work must leave admission without replay", + || { + fixture + .body + .host + .consolidation_admissions + .lock() + .unwrap() + .is_empty() + }, + ) + .await; + assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before); + assert!(fixture.domain_closure().is_none()); + assert!( + !fixture + .jobs + .get(&before.request.job_id) + .unwrap() + .executions_released() + ); +} diff --git a/crates/standalone/src/subjektiv/tests.rs b/crates/standalone/src/subjektiv/tests.rs index ecbca1fab..d5bde4d5c 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -4,6 +4,9 @@ //! cleanup receipt, not the domain lookup. No test here proves an OS crash/drain. use super::*; +#[path = "admission_tests.rs"] +mod admission_tests; + struct ConsolidationRecoveryFixture { body: SubjectConnection, jobs: crate::jobs::StandaloneJobs, @@ -279,12 +282,13 @@ async fn failed_consolidation_closes_only_attempt_and_preserves_unresolved_input memory ); // Stable exact replay; a reference does not mean all candidate work is done. - assert!( + assert_eq!( fixture .body .host .recover_consolidation_job(&fixture.jobs, &before) - .unwrap() + .unwrap(), + ConsolidationRecovery::Settled ); assert_eq!(fixture.domain_closure(), Some(receipt)); let reopened = crate::job_store::JobStore::open(&fixture.body.jobs_path).unwrap(); @@ -356,7 +360,10 @@ async fn failed_consolidation_retains_unreleased_stale_inactive_migrated_and_era .body .host .recover_consolidation_job(&fixture.jobs, &observed); - assert!(!matches!(outcome, Ok(true)), "{case}"); + assert!( + !matches!(outcome, Ok(ConsolidationRecovery::Settled)), + "{case}" + ); assert!(fixture.domain_closure().is_none(), "{case}"); assert_eq!( fixture.jobs.get(&fixture.saved.request.job_id).unwrap(),