From 27b401e0254a41166b735295945ab55dea3c990b Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 22:37:09 +0900 Subject: [PATCH] fix: reconcile typed domain storage faults within the original job deadline Keep Feature storage causes through Journal, Subject authorization and consolidation boundaries. Reconcile the exact operation and receipt with capped backoff until the original accepted attempt deadline, without treating another database write as domain recovery. Retain the original deadline when claiming dispatch; do not fabricate execution termination. Validated: real independent Journal DB writer lock beyond initial settlement, both absent and existing generation, same Job/attempt/input and one submit after release; deadline failure retains cause and input. Two new tests pass, adjacent Server 13, domain 4, standalone 20+13, root check/fmt/names/diff pass and 608 source hashes match. Initial 300s build timeout is retained; second build passes. Unfinished/unapproved checkpoint: R1 lost original Host/Controller evidence and R6/final suites/review remain open. --- crates/subjektiv/src/api.rs | 4 +- crates/subjektiv/src/job.rs | 13 +- crates/workspace-server/src/lib.rs | 2 + crates/workspace-server/src/server.rs | 5 +- .../src/server/backend_job_recovery.rs | 94 ++++++++- .../src/server/subject_authorization.rs | 2 +- .../src/server/subjektiv_job.rs | 4 +- .../src/server/subjektiv_journal.rs | 10 +- .../server/tests/subjektiv_journal_tests.rs | 196 ++++++++++++++++++ crates/workspace-server/src/store.rs | 4 +- 10 files changed, 313 insertions(+), 21 deletions(-) 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],