diff --git a/crates/subjektiv/src/api.rs b/crates/subjektiv/src/api.rs index 4ce172e3d..56c01d58a 100644 --- a/crates/subjektiv/src/api.rs +++ b/crates/subjektiv/src/api.rs @@ -1293,6 +1293,8 @@ pub enum OperationError { SubjectNotFound(String), #[error("{0}")] Store(String), + #[error(transparent)] + Storage(#[from] feature_storage::FeatureStorageError), } pub type OperationResult = std::result::Result; @@ -1323,7 +1325,7 @@ pub fn subjektiv_store_error(error: crate::SubjektivError) -> OperationError { crate::SubjektivError::MemoryNotFound(id) => { OperationError::InvalidInput(format!("memory_not_found: Memory `{id}` was not found")) } - crate::SubjektivError::Storage(_) => OperationError::Store(error.to_string()), + crate::SubjektivError::Storage(error) => OperationError::Storage(error), _ => OperationError::InvalidInput(error.to_string()), } } diff --git a/crates/subjektiv/src/job.rs b/crates/subjektiv/src/job.rs index 9ad62c011..f586b43cb 100644 --- a/crates/subjektiv/src/job.rs +++ b/crates/subjektiv/src/job.rs @@ -94,7 +94,7 @@ pub fn validate_result( for candidate_id in batch { if store .staging_resolution(subject_id, candidate_id) - .map_err(|e| OperationError::Store(e.to_string()))? + .map_err(crate::api::subjektiv_store_error)? .is_none() { return Err(OperationError::InvalidInput(format!( @@ -113,8 +113,11 @@ pub fn validate_result( job_id, attempt_id, ) - .map_err(|e| { - OperationError::InvalidInput(format!("consolidation surface result mismatch: {e}")) + .map_err(|error| match error { + crate::SubjektivError::Storage(error) => OperationError::Storage(error), + other => OperationError::InvalidInput(format!( + "consolidation surface result mismatch: {other}" + )), }) } @@ -204,7 +207,7 @@ pub fn lookup_result( if let Some(id) = snapshot_id { surface["snapshot_id"] = id.into(); } if let Some(reason) = reason_code { surface["reason_code"] = reason.into(); } Ok(Some(serde_json::json!({"subject_id":subject_id, "candidate_ids":batch,"surface":surface}))) - }).map_err(|error: SubjektivError| OperationError::Store(error.to_string()))?; + }).map_err(crate::api::subjektiv_store_error)?; let Some(result) = result else { return Ok(None); }; @@ -225,7 +228,7 @@ pub fn with_dispositions( for id in batch { let resolution = store .staging_resolution(subject_id, id) - .map_err(|e| OperationError::Store(e.to_string()))? + .map_err(crate::api::subjektiv_store_error)? .ok_or_else(|| { OperationError::InvalidInput(format!("candidate `{id}` has no durable disposition")) })?; diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index c7a028ec6..324569a6f 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -98,6 +98,8 @@ pub enum Error { Io(#[from] std::io::Error), #[error("sqlite error: {0}")] Sqlite(#[from] rusqlite::Error), + #[error("Subjektiv storage error: {0}")] + SubjektivStorage(#[source] FeatureStorageError), #[error("ticket error: {0}")] Ticket(#[from] ticket::TicketError), #[error("merge request error: {0}")] diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index bead2e4c8..6e296276e 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -23919,7 +23919,7 @@ where fn open_legacy_subjektiv_store(api: &WorkspaceApi) -> ApiResult { crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) - .map_err(|error| ApiError::from(Error::Store(error.to_string()))) + .map_err(|error| ApiError::from(subjektiv_store_error(error))) } fn open_subjektiv_store(api: &WorkspaceApi) -> ApiResult { @@ -26253,7 +26253,7 @@ fn subjektiv_store_error(error: crate::subjektiv::SubjektivError) -> Error { crate::subjektiv::SubjektivError::MemoryNotFound(memory_id) => Error::InvalidInput( format!("memory_not_found: Memory `{memory_id}` was not found"), ), - crate::subjektiv::SubjektivError::Storage(_) => Error::Store(error.to_string()), + crate::subjektiv::SubjektivError::Storage(error) => Error::SubjektivStorage(error), _ => Error::InvalidInput(error.to_string()), } } @@ -71200,6 +71200,7 @@ impl From for Error { Common::PermissionDenied(message) => Error::WorkspacePermissionDenied(message), Common::SubjectNotFound(id) => Error::SubjektivSubjectNotFound(id), Common::Store(message) => Error::Store(message), + Common::Storage(error) => Error::SubjektivStorage(error), }; error } diff --git a/crates/workspace-server/src/server/backend_job_recovery.rs b/crates/workspace-server/src/server/backend_job_recovery.rs index b6cc9e42e..83ef2fc1e 100644 --- a/crates/workspace-server/src/server/backend_job_recovery.rs +++ b/crates/workspace-server/src/server/backend_job_recovery.rs @@ -4,6 +4,7 @@ use super::*; use crate::backend_job::BackendJobRecord; use crate::backend_job_host_effects::{JobHostEffects, JobHostOwner, JobHostRequestLease}; +use feature_storage::FeatureStorageError; use std::collections::HashSet; use std::time::Duration; @@ -22,6 +23,8 @@ pub(super) struct RecoveryOwners { #[cfg(test)] pub(super) storage_probe_started: tokio::sync::Notify, #[cfg(test)] + pub(super) domain_storage_settlement: tokio::sync::Notify, + #[cfg(test)] pub(super) nonjournal_lookup_pending: tokio::sync::Notify, #[cfg(test)] pub(super) cleanup_before_claim: Mutex>>, @@ -145,7 +148,31 @@ fn receipt_matches( && result.runtime_run_id == attempt.runtime_run_id } +fn transient_subjektiv_storage(error: &Error) -> bool { + match error { + Error::SubjektivStorage(FeatureStorageError::Sqlite(rusqlite::Error::SqliteFailure( + error, + _, + ))) => matches!( + error.code, + rusqlite::ErrorCode::DatabaseBusy + | rusqlite::ErrorCode::DatabaseLocked + | rusqlite::ErrorCode::SystemIoFailure + | rusqlite::ErrorCode::DiskFull + | rusqlite::ErrorCode::CannotOpen + ), + // Feature I/O is bounded by the same operation's deadline, not a health flag. + Error::SubjektivStorage(FeatureStorageError::Io(_)) => true, + Error::SubjektivStorage(FeatureStorageError::OutcomeUnknown { .. }) => true, + _ => false, + } +} + fn transient_storage(error: &Error) -> bool { + transient_subjektiv_storage(error) || workspace_storage_failure(error) +} + +fn workspace_storage_failure(error: &Error) -> bool { matches!(error, Error::Sqlite(rusqlite::Error::SqliteFailure(error, _)) if matches!(error.code, rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked | rusqlite::ErrorCode::SystemIoFailure | rusqlite::ErrorCode::DiskFull @@ -550,7 +577,7 @@ impl WorkspaceApi { .await; failures += 1; } - Err(error) if transient_storage(&error) => { + Err(error) if workspace_storage_failure(&error) => { tracing::warn!(attempt_id=%expected.attempt_id, %error, "Cleanup settlement budget exhausted; exact-owner storage probe only"); let mut delay = Duration::from_secs(1); loop { @@ -601,14 +628,25 @@ impl WorkspaceApi { } async fn run_backend_job_recovery(&self, expected: BackendJobAttemptRecord) -> Result<()> { - // Only transient storage failure enters probe mode. Timers, unrelated writes and - // receipt failures never refill this settlement budget or authorize execution. + // Workspace write probes never refill a Feature operation's settlement budget. + // Feature failures reconcile the original operation until its original deadline. let mut storage_failures = 0; + let mut domain_failures = 0usize; + let deadline = chrono::DateTime::parse_from_rfc3339(&expected.deadline_at) + .map_err(|error| Error::Store(error.to_string()))?; let mut definitive_failure: Option = None; + let mut domain_failure: Option = None; loop { if self.backend_job_recovery.stopping.load(Ordering::Acquire) { return Ok(()); } + if Utc::now() >= deadline + && let Some(error) = domain_failure.take() + { + definitive_failure = Some(format!( + "Original attempt deadline exceeded during exact Feature operation reconciliation: {error}; execution and Host termination remain unconfirmed" + )); + } let api = self.clone(); let owner = expected.clone(); let failure = definitive_failure.clone(); @@ -619,18 +657,56 @@ impl WorkspaceApi { if let Some(detail) = failure { api.fail_backend_job_recovery(&owner, &detail) } else { - api.recover_backend_job_step(&owner) + let result = api.recover_backend_job_step(&owner); + if let Err(error) = &result + && transient_subjektiv_storage(error) + { + let current = api.store.get_backend_job_attempt(api.workspace_id(), + &owner.job_id, &owner.attempt_id)? + .ok_or_else(|| Error::Store("Domain reconciliation owner disappeared".into()))?; + api.retain_backend_job_recovery(¤t, "domain_storage_unavailable", + &error.to_string(), + "Reconcile the same Feature operation and receipt until the original attempt deadline; do not resubmit or probe another database")?; + } + result } }) .await; match step { Ok(true) => return Ok(()), Ok(false) => { + domain_failure = None; self.backend_job_recovery .pause(Duration::from_millis(250)) .await; } - Err(error) if transient_storage(&error) => { + Err(error) if transient_subjektiv_storage(&error) => { + domain_failure = Some(error.to_string()); + if Utc::now() >= deadline { + continue; + } + if domain_failures >= SETTLEMENT_BACKOFF_MS.len() { + #[cfg(test)] + self.backend_job_recovery + .domain_storage_settlement + .notify_one(); + } + let delay = SETTLEMENT_BACKOFF_MS + .get(domain_failures) + .copied() + .map(Duration::from_millis) + .unwrap_or_else(|| { + Duration::from_secs( + (1u64 << domain_failures.saturating_sub(5).min(5)).min(30), + ) + }); + domain_failures = domain_failures.saturating_add(1); + let remaining = (deadline.with_timezone(&Utc) - Utc::now()) + .to_std() + .unwrap_or_default(); + self.backend_job_recovery.pause(delay.min(remaining)).await; + } + Err(error) if workspace_storage_failure(&error) => { if let Some(delay) = SETTLEMENT_BACKOFF_MS.get(storage_failures) { storage_failures += 1; self.backend_job_recovery @@ -790,6 +866,14 @@ impl WorkspaceApi { return Ok(true); } if attempt.state == BackendJobAttemptState::Reserved { + let deadline = chrono::DateTime::parse_from_rfc3339(&attempt.deadline_at) + .map_err(|error| Error::Store(error.to_string()))?; + if Utc::now() >= deadline { + self.finish_backend_job_attempt(&attempt.job_id, &attempt.attempt_id, + BackendJobAttemptState::Failed, "timeout", + "Original attempt deadline exceeded before dispatch; no execution termination is inferred")?; + return Ok(false); + } self.dispatch_backend_job_reservation(BackendJobReservation { job, attempt, diff --git a/crates/workspace-server/src/server/subject_authorization.rs b/crates/workspace-server/src/server/subject_authorization.rs index 141d48098..762746d70 100644 --- a/crates/workspace-server/src/server/subject_authorization.rs +++ b/crates/workspace-server/src/server/subject_authorization.rs @@ -51,7 +51,7 @@ impl From<&WorkspaceApi> for WorkerAuthorityView { impl WorkerAuthorityView { fn subject_store(&self) -> ApiResult { crate::subjektiv::SubjektivStore::open(&self.feature_storage, &self.subjektiv_registration) - .map_err(|error| Error::Store(error.to_string()).into()) + .map_err(|error| subjektiv_store_error(error).into()) } fn require_not_deleted(&self, worker: &RuntimeWorkerRef) -> ApiResult<()> { diff --git a/crates/workspace-server/src/server/subjektiv_job.rs b/crates/workspace-server/src/server/subjektiv_job.rs index d778962ab..e0cd4b5ca 100644 --- a/crates/workspace-server/src/server/subjektiv_job.rs +++ b/crates/workspace-server/src/server/subjektiv_job.rs @@ -293,7 +293,7 @@ pub(super) fn validate_result( }; let store = crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) - .map_err(|e| Error::Store(e.to_string()))?; + .map_err(subjektiv_store_error)?; subjektiv::job::validate_result( &store, &grant.subject_id, @@ -320,7 +320,7 @@ pub(super) fn with_dispositions( .ok_or_else(|| Error::InvalidInput("missing consolidation grant".into()))?; let store = crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) - .map_err(|e| Error::Store(e.to_string()))?; + .map_err(subjektiv_store_error)?; subjektiv::job::with_dispositions(&store, &grant.subject_id, &grant.candidate_ids, value) .map_err(Error::from) } diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index 1d310fb51..da0db6cd1 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -12,13 +12,15 @@ pub(super) fn domain(error: JournalError) -> Error { match error { JournalError::PermissionDenied(s) => Error::WorkspacePermissionDenied(s), JournalError::Conflict(s) => Error::RepositoryConflict(s), + JournalError::Storage(error) => Error::SubjektivStorage(error), + JournalError::Subject(error) => subjektiv_store_error(error), other => Error::Store(other.to_string()), } } fn store(api: &WorkspaceApi) -> Result { crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) - .map_err(|e| Error::Store(e.to_string())) + .map_err(subjektiv_store_error) } pub(super) fn host( @@ -28,7 +30,7 @@ pub(super) fn host( ) -> Result { let attribution = store .session_attribution(&grant.session_id) - .map_err(|e| Error::Store(e.to_string()))? + .map_err(subjektiv_store_error)? .filter(|a| a.subject_id == grant.subject_id) .ok_or_else(|| { Error::WorkspacePermissionDenied("Journal Session attribution mismatch".into()) @@ -53,7 +55,7 @@ pub(super) fn prepare(api: &WorkspaceApi, reservation: &BackendJobReservation) - let store = store(api)?; let attribution = store .session_attribution(&grant.session_id) - .map_err(|e| Error::Store(e.to_string()))? + .map_err(subjektiv_store_error)? .filter(|attr| attr.subject_id == grant.subject_id) .ok_or_else(|| { Error::WorkspacePermissionDenied("Journal source attribution missing".into()) @@ -63,7 +65,7 @@ pub(super) fn prepare(api: &WorkspaceApi, reservation: &BackendJobReservation) - let host = host(api, &store, grant)?; let attr = store .session_attribution(&grant.session_id) - .map_err(|e| Error::Store(e.to_string()))? + .map_err(subjektiv_store_error)? .ok_or_else(|| { Error::WorkspacePermissionDenied("Journal Session attribution missing".into()) })?; diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index f1c91ccf1..bb373ac61 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -4,6 +4,7 @@ //! Tool-result persistence is a fixture boundary; no signed remote HTTP or model evaluation. use super::*; use crate::backend_job::{BackendJobGrants, SubjektivJournalGrant, journal_serialization_key}; +use feature_storage::FeatureStorageError; use sha2::{Digest, Sha256}; use subjektiv::journal::transport::*; use subjektiv::journal::*; @@ -1154,6 +1155,201 @@ impl Drop for JournalCompletionSaveFault { } } +fn lock_journal_domain(f: &RuntimeFixture) -> JournalCompletionSaveFault { + let database = f + .api + .feature_storage + .open(&f.api.subjektiv_registration) + .unwrap(); + let path: String = database + .with_connection(|connection| { + connection.busy_timeout(std::time::Duration::ZERO)?; + Ok(connection.query_row("PRAGMA database_list", [], |row| row.get(2))?) + }) + .unwrap(); + let connection = rusqlite::Connection::open(path).unwrap(); + connection.execute_batch("BEGIN IMMEDIATE").unwrap(); + JournalCompletionSaveFault(connection) +} + +#[tokio::test] +async fn domain_storage_lock_beyond_initial_settlement_recovers_original_journal_attempt() { + for already_prepared in [false, true] { + let f = RuntimeFixture::new().await; + if already_prepared { + subjektiv_journal::prepare(&f.api, &f.reserved).unwrap(); + } + let binding = subjektiv::api::JobAttemptBinding { + job_id: f.request.job_id.clone(), + attempt_id: f.reserved.attempt.attempt_id.clone(), + }; + let saved_generation = f + .store + .retained_journal_generation(&f.host, &f.id, &binding) + .unwrap(); + assert_eq!(saved_generation.is_some(), already_prepared); + let fault = lock_journal_domain(&f); + // Real domain storage stays locked. The healthy Workspace DB must not be + // used as proof that Journal preparation can proceed. + let error = subjektiv_journal::prepare(&f.api, &f.reserved).unwrap_err(); + assert!( + matches!(error, Error::SubjektivStorage( + FeatureStorageError::Sqlite(rusqlite::Error::SqliteFailure(code, _))) + if code.code == rusqlite::ErrorCode::DatabaseBusy), + "{error}" + ); + f.api.own_backend_job_recovery(&f.reserved.attempt); + tokio::time::timeout( + std::time::Duration::from_secs(20), + f.api + .backend_job_recovery + .domain_storage_settlement + .notified(), + ) + .await + .expect("domain operation did not retain ownership beyond six failed settlements"); + let pending = current_journal_attempt(&f, &binding.attempt_id); + assert_eq!(pending.state, BackendJobAttemptState::Reserved); + assert_eq!( + pending.failure_category.as_deref(), + Some("domain_storage_unavailable") + ); + assert_eq!(pending.deadline_at, f.reserved.attempt.deadline_at); + assert!(pending.worker.is_none()); + assert!(f.execution.inputs.lock().unwrap().is_empty()); + f.api + .store + .probe_backend_job_recovery(TEST_WORKSPACE_ID, &pending) + .unwrap(); + assert_eq!( + current_journal_attempt(&f, &binding.attempt_id).state, + BackendJobAttemptState::Reserved + ); + drop(fault); + // Only release the independent DB lock. No second dispatch or recovery call. + let dispatched = tokio::time::timeout(std::time::Duration::from_secs(20), async { + loop { + let attempt = current_journal_attempt(&f, &binding.attempt_id); + if attempt.state != BackendJobAttemptState::Reserved + && attempt.state != BackendJobAttemptState::Dispatching + { + break attempt; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .expect("same Journal attempt did not autonomously dispatch"); + // Observe the existing producer boundary before issuing fixture model operations. + // The saved Dispatched row alone can precede its first input acceptance. + { + let gate = f + .api + .backend_job_recovery + .dispatch_gate(&binding.attempt_id); + let _dispatch_finished = gate.lock().unwrap(); + } + assert_eq!( + dispatched.state, + BackendJobAttemptState::Dispatched, + "{dispatched:?}" + ); + assert_eq!(dispatched.attempt_id, binding.attempt_id); + assert_eq!(dispatched.input_digest, f.reserved.attempt.input_digest); + assert_eq!(dispatched.deadline_at, f.reserved.attempt.deadline_at); + let worker = dispatched.worker.as_ref().unwrap(); + f.api + .store + .update_worker_retention( + TEST_WORKSPACE_ID, + worker, + "pinned", + &Utc::now().to_rfc3339(), + ) + .unwrap(); + f.checkpoint(worker); + let completion = subjektiv_journal::finish(&f.api, worker).unwrap(); + let completed = observe_journal_completion(&f).await; + assert_eq!(completed.state, BackendJobState::Completed); + assert_eq!(completed.current_attempt, f.reserved.attempt.attempt); + assert_eq!(completed.input_digest, f.reserved.attempt.input_digest); + assert_eq!( + completed.result, + Some(serde_json::to_value(completion).unwrap()) + ); + let terminal = current_journal_attempt(&f, &binding.attempt_id); + assert_eq!(terminal.worker, dispatched.worker); + assert_eq!(terminal.runtime_run_id, dispatched.runtime_run_id); + let generation = f + .store + .retained_journal_generation(&f.host, &f.id, &binding) + .unwrap() + .unwrap(); + if let Some(saved) = saved_generation { + assert_eq!(generation.id, saved.id); + } + assert_eq!(f.execution.inputs.lock().unwrap().len(), 1); + assert_eq!( + completed.load_executable_request().unwrap().input, + f.request.input + ); + assert!( + f.api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &binding.job_id, + &crate::backend_job::attempt_id(&binding.job_id, 2) + ) + .unwrap() + .is_none() + ); + f.api.stop_backend_job_recovery().await; + } +} + +#[tokio::test] +async fn domain_storage_lock_stops_at_original_attempt_deadline_without_dispatch() { + let mut f = RuntimeFixture::new().await; + let deadline = (Utc::now() + chrono::Duration::seconds(1)).to_rfc3339(); + f.job_store.with_conn(|connection| { + connection.execute("UPDATE backend_job_attempts SET deadline_at=?1 WHERE workspace_id=?2 AND attempt_id=?3", + rusqlite::params![deadline, TEST_WORKSPACE_ID, f.reserved.attempt.attempt_id])?; + Ok(()) + }).unwrap(); + f.reserved.attempt = current_journal_attempt(&f, &f.reserved.attempt.attempt_id); + let fault = lock_journal_domain(&f); + f.api.own_backend_job_recovery(&f.reserved.attempt); + let failed = observe_journal_completion(&f).await; + assert_eq!(failed.state, BackendJobState::Failed); + assert_eq!( + failed.failure_category.as_deref(), + Some("recovery_cannot_continue") + ); + let detail = failed.failure_detail.as_deref().unwrap(); + assert!( + detail.contains("Original attempt deadline exceeded"), + "{detail}" + ); + assert!(detail.contains("Subjektiv storage"), "{detail}"); + let attempt = current_journal_attempt(&f, &f.reserved.attempt.attempt_id); + assert_eq!(attempt.deadline_at, deadline); + assert!(attempt.worker.is_none()); + assert!(f.execution.inputs.lock().unwrap().is_empty()); + drop(fault); + assert!( + f.store + .active_journal_generation(&f.host, &f.id) + .unwrap() + .is_none() + ); + assert_eq!( + failed.load_executable_request().unwrap().input, + f.request.input + ); + f.api.stop_backend_job_recovery().await; +} + fn current_journal_job(f: &RuntimeFixture) -> crate::backend_job::BackendJobRecord { f.api .store diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index a6be930d1..fbe60c3d2 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -8368,7 +8368,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore { tx.commit()?; return Ok((BackendJobReservation { job, attempt, replayed: true, resource_reused: false }, false)); } - let deadline = backend_job_deadline(now, job.request.limits.timeout_seconds)?; + // Admission already fixed this attempt's deadline. Recovering prepare or + // waiting for a slot must not extend the same accepted operation's budget. + let deadline = &attempt.deadline_at; let changed = tx.execute( "UPDATE backend_job_attempts SET state = 'dispatching', deadline_at = ?4, updated_at = ?5 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'reserved' AND worker_id IS NULL", params![workspace_id, job_id, attempt_id, deadline, now],