From ceb01f6eaa4a467a74274b1df2564ac092b38816 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 26 Oct 2026 10:03:56 +0900 Subject: [PATCH] feat: runtime own restore continuation beyond caller timeout --- crates/worker-runtime/src/execution.rs | 84 ++- crates/worker-runtime/src/fs_store.rs | 93 +++- crates/worker-runtime/src/runtime.rs | 138 ++++- .../src/runtime/restore_continuation.rs | 498 ++++++++++++++++++ .../worker-runtime/src/test_support_tests.rs | 1 + crates/worker-runtime/src/worker_backend.rs | 94 +++- .../src/worker_backend/restore_recovery.rs | 4 + .../worker_backend_restore_failure_tests.rs | 2 + ...orker_backend_restore_publication_tests.rs | 138 +++++ 9 files changed, 1030 insertions(+), 22 deletions(-) create mode 100644 crates/worker-runtime/src/runtime/restore_continuation.rs diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index b95347102..39faadab5 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -242,6 +242,9 @@ enum WorkerExecutionPublicationState { #[derive(Clone)] pub struct WorkerExecutionContext { worker_ref: WorkerRef, + restore_progress: Option>, + restore_repair_authorizer: + Option Result<(), RuntimeError> + Send + Sync>>, #[cfg(feature = "ws-server")] observation_publisher: WorkerObservationPublisher, #[cfg(feature = "ws-server")] @@ -256,6 +259,8 @@ impl WorkerExecutionContext { ) -> Self { Self { worker_ref, + restore_progress: None, + restore_repair_authorizer: None, observation_publisher, publication_state: Arc::new(Mutex::new(WorkerExecutionPublicationState::Active)), } @@ -268,6 +273,8 @@ impl WorkerExecutionContext { ) -> Self { Self { worker_ref, + restore_progress: None, + restore_repair_authorizer: None, observation_publisher, publication_state: Arc::new(Mutex::new(WorkerExecutionPublicationState::Pending( VecDeque::new(), @@ -277,12 +284,56 @@ impl WorkerExecutionContext { #[cfg(not(feature = "ws-server"))] pub(crate) fn new(worker_ref: WorkerRef) -> Self { - Self { worker_ref } + Self { + worker_ref, + restore_progress: None, + restore_repair_authorizer: None, + } } #[cfg(not(feature = "ws-server"))] pub(crate) fn candidate(worker_ref: WorkerRef) -> Self { - Self { worker_ref } + Self { + worker_ref, + restore_progress: None, + restore_repair_authorizer: None, + } + } + + pub(crate) fn with_restore_repair_authorizer( + mut self, + authorizer: Arc< + dyn Fn(WorkerLifecycleOperationId) -> Result<(), RuntimeError> + Send + Sync, + >, + ) -> Self { + self.restore_repair_authorizer = Some(authorizer); + self + } + + pub(crate) fn authorize_restore_source_repair( + &self, + operation: WorkerLifecycleOperationId, + ) -> Result<(), RuntimeError> { + match &self.restore_repair_authorizer { + Some(authorizer) => authorizer(operation), + // Direct factory/component users own their source repair separately; + // only Runtime contexts carry a Runtime aggregate authority callback. + None => Ok(()), + } + } + + pub(crate) fn with_restore_progress( + mut self, + progress: Arc, + ) -> Self { + self.restore_progress = Some(progress); + self + } + + pub(crate) fn restore_progressed(&self, operation_id: WorkerLifecycleOperationId) { + if let Some(progress) = &self.restore_progress { + progress(operation_id); + } } pub fn worker_ref(&self) -> &WorkerRef { @@ -424,6 +475,7 @@ pub struct WorkerExecutionSpawnRequest { /// Request passed to a [`WorkerExecutionBackend`] when restoring a persisted Worker. #[derive(Clone, Debug)] pub struct WorkerExecutionRestoreRequest { + pub source_repair_possible: bool, pub operation_id: WorkerLifecycleOperationId, pub worker_ref: WorkerRef, pub request: crate::catalog::CreateWorkerRequest, @@ -565,6 +617,15 @@ impl fmt::Debug for WorkerProtocolTransport { } } +/// Derived resource disposition for an existing restore owner. Not durable +/// authority: Runtime must reread its decision before choosing any action. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RestoreRecoveryDisposition { + InFlight, + Continue, + Discard, +} + /// The connection selects this backend; WorkerRef identifies an execution, /// never a capability or authorization token. pub trait WorkerExecutionBackend: Send + Sync + 'static { @@ -643,6 +704,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { None } + /// Opt in to Runtime-owned continuation. Waiting is signaled through the + /// original execution context; no callback may wait for Runtime locks. + fn restore_recovery_disposition( + &self, + _operation_id: WorkerLifecycleOperationId, + _worker_ref: &WorkerRef, + ) -> Option { + None + } + fn reconcile_restore( &self, request: WorkerExecutionRestoreRequest, @@ -938,6 +1009,15 @@ impl WorkerExecutionBackendRef { .completed_restore_failure(operation_id, worker_ref) } + pub(crate) fn restore_recovery_disposition( + &self, + operation_id: WorkerLifecycleOperationId, + worker_ref: &WorkerRef, + ) -> Option { + self.backend + .restore_recovery_disposition(operation_id, worker_ref) + } + pub(crate) fn reconcile_restore( &self, request: WorkerExecutionRestoreRequest, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 5418550d1..d2b9c86de 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -43,7 +43,8 @@ use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; #[path = "test_support.rs"] pub mod test_support; -pub(crate) const SCHEMA_VERSION: u32 = 13; +pub(crate) const SCHEMA_VERSION: u32 = 14; +const RESTORE_DECISION_SCHEMA_VERSION: u32 = 13; const RESTORE_ADMISSION_SCHEMA_VERSION: u32 = 12; // Frozen source schemas: do not move their meaning when adding a migration. const SOURCE_SCHEMA_VERSION: u32 = 11; @@ -653,6 +654,7 @@ pub(crate) struct PersistedWorkerRestoreOperation { pub(crate) operation_id: WorkerLifecycleOperationId, pub(crate) mode: PersistedWorkerRestoreMode, pub(crate) commit: crate::runtime::RestoreCommitFact, + pub(crate) source_repair_possible: bool, pub(crate) request: CreateWorkerRequest, pub(crate) binding: Option, pub(crate) last_settled_status: WorkerStatus, @@ -829,11 +831,12 @@ fn plan_runtime_store_migration( | EXECUTION_SCHEMA_VERSION | SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION + | RESTORE_DECISION_SCHEMA_VERSION ) { return Err(runtime_store_corrupt( &runtime_path, format!( - "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, or {SCHEMA_VERSION}" + "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, {AGGREGATE_SCHEMA_VERSION}, {EXECUTION_SCHEMA_VERSION}, {SOURCE_SCHEMA_VERSION}, {RESTORE_ADMISSION_SCHEMA_VERSION}, {RESTORE_DECISION_SCHEMA_VERSION}, or {SCHEMA_VERSION}" ), )); } @@ -1165,7 +1168,10 @@ fn migrate_worker_record( ) -> Result { if matches!( source_schema_version, - EXECUTION_SCHEMA_VERSION | SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION + EXECUTION_SCHEMA_VERSION + | SOURCE_SCHEMA_VERSION + | RESTORE_ADMISSION_SCHEMA_VERSION + | RESTORE_DECISION_SCHEMA_VERSION ) { // Admission identity survives migration. Older pending records did not // distinguish a committed decision from preparation or failed cleanup. @@ -1179,7 +1185,10 @@ fn migrate_worker_record( "Worker schema does not match Runtime migration source".into(), )); } - migrate_restore_commit_fact(&mut identity_document, identity_path)?; + if source_schema_version != RESTORE_DECISION_SCHEMA_VERSION { + migrate_restore_commit_fact(&mut identity_document, identity_path)?; + } + migrate_restore_repair_fact(&mut identity_document, identity_path)?; identity_document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION); let aggregate: WorkerAggregateRecord = serde_json::from_value(identity_document) .map_err(|error| runtime_store_corrupt(identity_path, error.to_string()))?; @@ -1294,6 +1303,42 @@ fn migrate_restore_commit_fact( Ok(()) } +// Earlier writers could repair another canonical store before commit without +// retaining that authority in this aggregate. Missing evidence is never false. +fn migrate_restore_repair_fact( + document: &mut serde_json::Value, + path: &Path, +) -> Result<(), RuntimeError> { + let execution = &mut document["execution_state"]; + if execution["state"] != "reconciliation_required" { + return Ok(()); + } + let operation = &mut execution["execution"]; + let restore = match operation["operation"].as_str() { + Some("restore") => Some(&mut operation["intent"]), + Some("stop") => operation["intent"] + .as_object_mut() + .and_then(|stop| stop.get_mut("pending_restore")) + .filter(|restore| !restore.is_null()), + _ => None, + }; + if let Some(restore) = restore { + let restore = restore.as_object_mut().ok_or_else(|| { + runtime_store_corrupt(path, "legacy pending Restore must be an object".into()) + })?; + if restore + .insert("source_repair_possible".into(), serde_json::json!(true)) + .is_some() + { + return Err(runtime_store_corrupt( + path, + "legacy Restore contains unexpected source repair evidence".into(), + )); + } + } + Ok(()) +} + // Frozen schema-v9 conversion. Preserve the exact token and all Restore owners // and receipts; generating a fresh token here would lose retry/admission authority. fn migrate_v9_restore_guard( @@ -1608,7 +1653,7 @@ fn migrate_runtime_document( .or_insert_with(|| serde_json::json!({})); if !matches!( source_schema_version, - SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION + SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION | RESTORE_DECISION_SCHEMA_VERSION ) { object.remove("workers"); // This legacy cache is not Runtime lifecycle or restore authority. Bundles @@ -2148,7 +2193,9 @@ fn migrate_runtime_store_in_place( if let Some(metadata) = metadata && !matches!( plan.current_schema_version, - SOURCE_SCHEMA_VERSION | RESTORE_ADMISSION_SCHEMA_VERSION + SOURCE_SCHEMA_VERSION + | RESTORE_ADMISSION_SCHEMA_VERSION + | RESTORE_DECISION_SCHEMA_VERSION ) { atomic_write_json( @@ -4001,7 +4048,13 @@ mod tests { #[test] fn legacy_pending_restore_migrates_as_unknown_and_preserves_stop_ownership() { use crate::runtime::RestoreCommitFact; - for under_stop in [false, true] { + for (source_schema, under_stop) in [ + RESTORE_ADMISSION_SCHEMA_VERSION, + RESTORE_DECISION_SCHEMA_VERSION, + ] + .into_iter() + .flat_map(|schema| [false, true].map(move |stop| (schema, stop))) + { let (_temp, store, mut worker, path) = canonical_worker_store(); let root = store.root.clone(); let PersistedWorkerExecutionState::Available(execution) = @@ -4013,6 +4066,7 @@ mod tests { operation_id: WorkerLifecycleOperationId::new(), mode: PersistedWorkerRestoreMode::Explicit, commit: RestoreCommitFact::Committed, + source_repair_possible: false, request: execution.request.clone(), binding: execution.binding.clone(), last_settled_status: worker.status, @@ -4033,21 +4087,27 @@ mod tests { PersistedWorkerExecutionState::ReconciliationRequired(operation); let mut legacy = serde_json::to_value(WorkerAggregateRecord::from_persisted(&worker)).unwrap(); - legacy["schema_version"] = serde_json::json!(RESTORE_ADMISSION_SCHEMA_VERSION); + legacy["schema_version"] = serde_json::json!(source_schema); let intent = &mut legacy["execution_state"]["execution"]["intent"]; let restore = if under_stop { &mut intent["pending_restore"] } else { intent }; - restore.as_object_mut().unwrap().remove("commit"); + if source_schema == RESTORE_ADMISSION_SCHEMA_VERSION { + restore.as_object_mut().unwrap().remove("commit"); + } + restore + .as_object_mut() + .unwrap() + .remove("source_repair_possible"); let expected_request = restore["request"].clone(); let before_guard = legacy["restore_guard"].clone(); fs::write(&path, serde_json::to_vec(&legacy).unwrap()).unwrap(); write_empty_schema_v8_store(&root); let runtime_path = root.join(RUNTIME_FILE); let mut runtime: serde_json::Value = read_json(&runtime_path, "read fixture").unwrap(); - runtime["schema_version"] = serde_json::json!(RESTORE_ADMISSION_SCHEMA_VERSION); + runtime["schema_version"] = serde_json::json!(source_schema); fs::write(&runtime_path, serde_json::to_vec(&runtime).unwrap()).unwrap(); drop(store); @@ -4062,7 +4122,18 @@ mod tests { PersistedWorkerLifecycleOperation::Restore(restore) => restore, PersistedWorkerLifecycleOperation::Stop(stop) => stop.pending_restore.unwrap(), }; - assert_eq!(restored.commit, RestoreCommitFact::LegacyUnknown); + assert_eq!( + restored.commit, + if source_schema == RESTORE_DECISION_SCHEMA_VERSION { + RestoreCommitFact::Committed + } else { + RestoreCommitFact::LegacyUnknown + } + ); + assert!( + restored.source_repair_possible, + "older source-repair effects must remain possible" + ); assert_eq!( serde_json::to_value(restored.request).unwrap(), expected_request diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 251e423c2..6c396e6b4 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1,3 +1,4 @@ +mod restore_continuation; use crate::catalog::{ ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, ProfileSelector, ProfileSourceArchiveSource, RepositoryRefObservation, RepositoryRefObservationRequest, @@ -323,6 +324,7 @@ impl Drop for RuntimeEventSelectorSubscription { pub struct Runtime { inner: Arc>, worker_operations: Arc>>>>, + restore_continuations: Arc, } struct WorkerOperationLease { @@ -586,6 +588,7 @@ impl Runtime { Self { inner: Arc::new(Mutex::new(state)), worker_operations: Arc::new(Mutex::new(BTreeMap::new())), + restore_continuations: Arc::default(), } } @@ -662,6 +665,7 @@ impl Runtime { let runtime = Self { inner: Arc::new(Mutex::new(state)), worker_operations: Arc::new(Mutex::new(BTreeMap::new())), + restore_continuations: Arc::default(), }; runtime.restore_persisted_worker_executions()?; Ok(runtime) @@ -2230,6 +2234,7 @@ impl Runtime { operation_id: WorkerLifecycleOperationId::new(), mode: WorkerRestoreMode::Explicit, commit: RestoreCommitFact::NotCommitted, + source_repair_possible: false, last_settled_status: candidate.status, }); let operation_id = pending.operation_id; @@ -2256,6 +2261,10 @@ impl Runtime { match state.persist_restore_worker_record(&candidate) { Ok(()) => {} Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => { + drop(state); + if !coordinate { + self.start_restore_continuation(worker_ref.clone(), operation_id)?; + } return Ok(( Some(RuntimeWorkerRestoreResult::failed( WorkerRestoreState::ReconciliationRequired, @@ -2287,8 +2296,10 @@ impl Runtime { return Ok((None, owner.request.preparation.clone())); } } - self.restore_admitted_worker_under_lock(worker_ref, WorkerRestoreMode::Explicit) - .map(|result| (Some(result), None)) + let result = + self.restore_admitted_worker_under_lock(worker_ref, WorkerRestoreMode::Explicit); + self.ensure_restore_continuation(worker_ref)?; + result.map(|result| (Some(result), None)) } #[cfg(all(test, feature = "fs-store"))] @@ -2529,6 +2540,8 @@ impl Runtime { .cloned() }); let request = WorkerExecutionRestoreRequest { + source_repair_possible: pending_restore + .is_some_and(|pending| pending.source_repair_possible), operation_id, worker_ref: worker_ref.clone(), request: worker_request, @@ -4075,12 +4088,26 @@ impl Runtime { crate::execution::WorkerExecutionContext::new(worker_ref) } + fn restore_progress_notifier( + &self, + worker_ref: WorkerRef, + ) -> Arc { + let continuations = Arc::downgrade(&self.restore_continuations); + Arc::new(move |operation| { + if let Some(continuations) = continuations.upgrade() { + continuations.signal(&worker_ref, operation); + } + }) + } + #[cfg(feature = "ws-server")] fn restore_candidate_execution_context( &self, worker_ref: WorkerRef, ) -> crate::execution::WorkerExecutionContext { let runtime = Arc::downgrade(&self.inner); + let progress = self.restore_progress_notifier(worker_ref.clone()); + let repair = self.restore_repair_authorizer(worker_ref.clone()); crate::execution::WorkerExecutionContext::candidate( worker_ref, Arc::new(move |worker_ref, payload| { @@ -4101,6 +4128,8 @@ impl Runtime { Ok(state.push_worker_observation_event(worker_ref, payload)) }), ) + .with_restore_progress(progress) + .with_restore_repair_authorizer(repair) } #[cfg(not(feature = "ws-server"))] @@ -4108,7 +4137,11 @@ impl Runtime { &self, worker_ref: WorkerRef, ) -> crate::execution::WorkerExecutionContext { + let progress = self.restore_progress_notifier(worker_ref.clone()); + let repair = self.restore_repair_authorizer(worker_ref.clone()); crate::execution::WorkerExecutionContext::candidate(worker_ref) + .with_restore_progress(progress) + .with_restore_repair_authorizer(repair) } #[cfg(feature = "fs-store")] @@ -4410,6 +4443,9 @@ impl Runtime { commit: candidate .pending_restore .map_or(RestoreCommitFact::NotCommitted, |pending| pending.commit), + source_repair_possible: candidate + .pending_restore + .is_some_and(|pending| pending.source_repair_possible), last_settled_status: candidate.status, }); candidate.restore_candidate_context = Some(candidate_context); @@ -6327,6 +6363,9 @@ struct PendingWorkerRestore { operation_id: WorkerLifecycleOperationId, mode: WorkerRestoreMode, commit: RestoreCommitFact, + // False proves Runtime never granted this operation authority to repair + // canonical source before its decision. True includes legacy uncertainty. + source_repair_possible: bool, last_settled_status: WorkerStatus, } @@ -6460,10 +6499,9 @@ struct WorkerRecord { impl WorkerRecord { fn ensure_restore_rollback_allowed(&self) -> Result<(), RuntimeError> { - if self - .pending_restore - .is_some_and(|pending| pending.commit != RestoreCommitFact::NotCommitted) - { + if self.pending_restore.is_some_and(|pending| { + pending.commit != RestoreCommitFact::NotCommitted || pending.source_repair_possible + }) { return Err(RuntimeError::InvalidRequest( "A committed or legacy-unknown Restore cannot be settled as a rollback".into(), )); @@ -6682,6 +6720,7 @@ impl WorkerRecord { operation_id: operation.operation_id, mode: operation.mode.into(), commit: operation.commit, + source_repair_possible: operation.source_repair_possible, last_settled_status: operation.last_settled_status, }), None, @@ -6698,6 +6737,7 @@ impl WorkerRecord { operation_id: restore.operation_id, mode: restore.mode.into(), commit: restore.commit, + source_repair_possible: restore.source_repair_possible, last_settled_status: restore.last_settled_status, }), Some(PendingWorkerStop { @@ -6762,6 +6802,7 @@ impl WorkerRecord { operation_id: restore.operation_id, mode: restore.mode.into(), commit: restore.commit, + source_repair_possible: restore.source_repair_possible, request: self .request .clone() @@ -6786,6 +6827,7 @@ impl WorkerRecord { operation_id: pending.operation_id, mode: pending.mode.into(), commit: pending.commit, + source_repair_possible: pending.source_repair_possible, request, binding: self .execution_bound @@ -9156,6 +9198,7 @@ mod tests { operation_id: WorkerLifecycleOperationId::new(), mode: WorkerRestoreMode::Explicit, commit: RestoreCommitFact::NotCommitted, + source_repair_possible: false, last_settled_status: WorkerStatus::Stopped, }); let id = "workdir-irrelevant-to-recovery"; @@ -13588,7 +13631,7 @@ mod tests { }) .unwrap_err(); assert!(error.to_string().contains( - "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, or 13" + "unsupported Runtime store schema version 2; expected 6, 7, 8, 9, 10, 11, 12, 13, or 14" )); let _ = std::fs::remove_dir_all(root); @@ -15014,6 +15057,87 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[cfg(feature = "fs-store")] + #[test] + fn restore_source_repair_authority_is_durable_before_effects_without_reentering_lifecycle_lock() + { + for fault in [ + None, + Some(crate::fs_store::AtomicWriteFault::BeforeRename), + Some(crate::fs_store::AtomicWriteFault::AfterRename), + ] { + let dir = tempfile::tempdir().unwrap(); + let options = FsRuntimeStoreOptions::new(dir.path().join("runtime")) + .with_runtime_id("repair-authority"); + let runtime = Runtime::with_fs_store_and_execution_backend( + options, + Arc::new(TestExecutionBackend::default()), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("repair authority")) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + let lease = runtime.worker_operation_lock(worker.worker_id).unwrap(); + let _guard = lease.lock().unwrap(); + let operation = WorkerLifecycleOperationId::new(); + runtime + .begin_restore_operation( + &worker.worker_ref, + operation, + WorkerRestoreMode::Explicit, + runtime.restore_candidate_execution_context(worker.worker_ref.clone()), + ) + .unwrap(); + if let Some(fault) = fault { + runtime_store(&runtime).fail_next_worker_write(fault); + if matches!(fault, crate::fs_store::AtomicWriteFault::AfterRename) { + runtime_store(&runtime).fail_next_worker_reconcile_sync(); + } + } + let authorizer = runtime.restore_repair_authorizer(worker.worker_ref.clone()); + let callback = authorizer.clone(); + let (sent, received) = std::sync::mpsc::channel(); + let task = std::thread::spawn(move || sent.send(callback(operation)).unwrap()); + // The lifecycle lock above remains held while this exact factory + // callback persists its grant. Waiting on that lock would deadlock. + let granted = received + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("repair callback reentered lifecycle lock"); + task.join().unwrap(); + assert_eq!(granted.is_ok(), fault.is_none()); + let saved = runtime_store(&runtime) + .read_worker_record(&worker.worker_ref) + .unwrap(); + let PersistedWorkerExecutionState::ReconciliationRequired( + PersistedWorkerLifecycleOperation::Restore(saved), + ) = saved.execution_state + else { + panic!("repair grant lost operation") + }; + assert_eq!(saved.operation_id, operation); + assert_eq!(saved.commit, RestoreCommitFact::NotCommitted); + assert_eq!( + saved.source_repair_possible, + !matches!(fault, Some(crate::fs_store::AtomicWriteFault::BeforeRename)) + ); + // A denied/unknown grant must not authorize external repair. Once + // storage is healthy the same operation reconciles, never a new ID. + authorizer(operation).unwrap(); + let rollback = RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::RolledBack, + "unsafe", + "repair effects may exist", + ); + assert!( + runtime + .settle_restore_rollback(&worker.worker_ref, &rollback) + .is_err() + ); + } + } + #[cfg(feature = "fs-store")] #[test] fn restore_decision_reread_preserves_commit_across_uncertainty_stop_and_restart() { diff --git a/crates/worker-runtime/src/runtime/restore_continuation.rs b/crates/worker-runtime/src/runtime/restore_continuation.rs new file mode 100644 index 000000000..78cda5fdc --- /dev/null +++ b/crates/worker-runtime/src/runtime/restore_continuation.rs @@ -0,0 +1,498 @@ +//! Runtime-owned continuation of an admitted restore, independent of its caller. +//! +//! Completion callbacks only signal a condition variable. They never acquire a +//! Runtime/lifecycle lock and are never joined by this driver while holding one. +//! Each step revalidates the canonical owner under the existing Worker lock; +//! waiting/backoff and driver teardown happen outside it. A repeated obstruction +//! parks this owner with its reason, rather than spinning or inventing rollback. + +use super::*; +use crate::execution::RestoreRecoveryDisposition; +use std::collections::HashMap; +use std::sync::{Condvar, Weak}; +use std::thread::JoinHandle; +use std::time::Duration; + +type Key = (WorkerRef, WorkerLifecycleOperationId); + +#[derive(Debug, Default)] +pub(super) struct RestoreContinuations { + owners: Mutex>>, + #[cfg(test)] + paused: std::sync::atomic::AtomicBool, +} + +#[derive(Debug, Default)] +struct Schedule { + signaled: bool, + cancelled: bool, + finished: bool, + blocked_reason: Option, + parked: bool, +} + +#[derive(Debug, Default)] +struct Continuation { + schedule: Mutex, + wake: Condvar, + task: Mutex>>, +} + +impl RestoreContinuations { + pub(super) fn signal(&self, worker_ref: &WorkerRef, operation: WorkerLifecycleOperationId) { + if let Some(owner) = self + .owners + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&(worker_ref.clone(), operation)) + { + let mut schedule = owner + .schedule + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + schedule.signaled = true; + owner.wake.notify_all(); + } + } +} + +impl Drop for RestoreContinuations { + fn drop(&mut self) { + let owners = self + .owners + .get_mut() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for owner in owners.values() { + let mut schedule = owner + .schedule + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + schedule.cancelled = true; + owner.wake.notify_all(); + } + for owner in owners.values() { + if let Some(task) = owner + .task + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .take() + { + if task.thread().id() != std::thread::current().id() { + let _ = task.join(); + } + } + } + } +} + +#[cfg(test)] +pub(crate) struct ContinuationPause(Arc); + +#[cfg(test)] +impl Drop for ContinuationPause { + fn drop(&mut self) { + self.0 + .paused + .store(false, std::sync::atomic::Ordering::Release); + for owner in self.0.owners.lock().unwrap().values() { + owner.schedule.lock().unwrap().signaled = true; + owner.wake.notify_all(); + } + } +} + +enum Step { + Finished, + Waiting, + Contended, + Retry(String), +} + +impl Runtime { + pub(super) fn restore_repair_authorizer( + &self, + worker_ref: WorkerRef, + ) -> Arc Result<(), RuntimeError> + Send + Sync> { + let inner = Arc::downgrade(&self.inner); + Arc::new(move |operation| { + let inner = inner.upgrade().ok_or(RuntimeError::RuntimeStopped)?; + // The factory caller owns the lifecycle lock. Never reacquire it + // here: it may be awaiting this exact factory task. + let mut state = inner.lock().map_err(|_| RuntimeError::StatePoisoned)?; + state.ensure_running()?; + state.refresh_uncertain_worker_write(&worker_ref)?; + let worker = state.worker(&worker_ref)?; + state.ensure_subject_not_reserved(&worker_ref, false)?; + if worker.pending_stop.is_some() + || worker.pending_restore.is_none_or(|pending| { + pending.operation_id != operation + || pending.commit == RestoreCommitFact::LegacyUnknown + }) + { + return Err(RuntimeError::RestoreObservationConflict { + worker_id: worker_ref.worker_id, + }); + } + if let Some(owner) = worker + .restore_guard + .active_request_id + .as_ref() + .and_then(|id| worker.restore_guard.owners.get(id)) + { + owner.validate_workdir_preparation_authority(worker)?; + } + if worker + .pending_restore + .is_some_and(|pending| pending.source_repair_possible) + { + return Ok(()); + } + let mut candidate = worker.clone(); + candidate + .pending_restore + .as_mut() + .expect("validated restore owner") + .source_repair_possible = true; + // Grant before touching another canonical store. Unknown durability + // denies repair until authoritative reconciliation; no old-memory write. + state.persist_restore_worker_record(&candidate)?; + state.workers.insert(worker_ref.worker_id, candidate); + Ok(()) + }) + } + + #[cfg(test)] + pub(crate) fn test_wait_restore_continuation_parked( + &self, + worker_ref: &WorkerRef, + operation: WorkerLifecycleOperationId, + timeout: Duration, + ) -> String { + let owner = self + .restore_continuations + .owners + .lock() + .unwrap() + .get(&(worker_ref.clone(), operation)) + .cloned() + .unwrap(); + let (schedule, waited) = owner + .wake + .wait_timeout_while(owner.schedule.lock().unwrap(), timeout, |s| { + !s.parked && !s.finished + }) + .unwrap(); + assert!( + !waited.timed_out() && schedule.parked, + "continuation did not park: {:?}", + schedule.blocked_reason + ); + schedule.blocked_reason.clone().expect("parked reason") + } + + /// Gate autonomous steps when a component test observes a single fault + /// boundary. Dedicated continuation tests run without this gate. + #[cfg(test)] + pub(crate) fn test_pause_restore_continuation(&self) -> ContinuationPause { + assert!( + !self + .restore_continuations + .paused + .swap(true, std::sync::atomic::Ordering::AcqRel) + ); + ContinuationPause(self.restore_continuations.clone()) + } + + #[cfg(test)] + pub(crate) fn test_wait_restore_continuation( + &self, + worker_ref: &WorkerRef, + operation: WorkerLifecycleOperationId, + timeout: Duration, + ) { + let owner = self + .restore_continuations + .owners + .lock() + .unwrap() + .get(&(worker_ref.clone(), operation)) + .cloned() + .expect("continuation owner"); + let schedule = owner.schedule.lock().unwrap(); + let (schedule, waited) = owner + .wake + .wait_timeout_while(schedule, timeout, |s| !s.finished && !s.cancelled) + .unwrap(); + assert!( + !waited.timed_out() && schedule.finished, + "restore continuation did not settle: {:?}", + schedule.blocked_reason + ); + } + + pub(super) fn ensure_restore_continuation( + &self, + worker_ref: &WorkerRef, + ) -> Result<(), RuntimeError> { + let operation = { + let state = self.lock()?; + let worker = state.worker(worker_ref)?; + worker + .pending_restore + .map(|pending| pending.operation_id) + .or_else(|| { + worker + .restore_guard + .active_request_id + .as_ref() + .and_then(|id| worker.restore_guard.owners.get(id)) + .filter(|owner| owner.receipt.is_none()) + .map(|owner| owner.operation_id) + }) + }; + if let Some(operation) = operation { + self.start_restore_continuation(worker_ref.clone(), operation)?; + } + Ok(()) + } + + pub(super) fn start_restore_continuation( + &self, + worker_ref: WorkerRef, + operation: WorkerLifecycleOperationId, + ) -> Result<(), RuntimeError> { + { + let state = self.lock()?; + let Some(backend) = &state.execution_backend else { + return Ok(()); + }; + if backend + .restore_recovery_disposition(operation, &worker_ref) + .is_none() + { + return Ok(()); + } + } + let key = (worker_ref.clone(), operation); + let mut owners = self + .restore_continuations + .owners + .lock() + .map_err(|_| RuntimeError::StatePoisoned)?; + if owners.contains_key(&key) { + return Ok(()); + } + let owner = Arc::new(Continuation::default()); + let retained = owner.clone(); + let state = Arc::downgrade(&self.inner); + let operations = Arc::downgrade(&self.worker_operations); + let continuations = Arc::downgrade(&self.restore_continuations); + let task = std::thread::Builder::new() + .name("restore-continuation".into()) + .spawn(move || { + run( + state, + operations, + continuations, + retained, + worker_ref, + operation, + ); + }) + .map_err(|error| { + RuntimeError::InvalidRequest(format!( + "Restore continuation could not start: {error}" + )) + })?; + *owner.task.lock().map_err(|_| RuntimeError::StatePoisoned)? = Some(task); + owners.insert(key, owner); + Ok(()) + } + + fn continue_restore_operation( + &self, + worker_ref: &WorkerRef, + operation: WorkerLifecycleOperationId, + ) -> Result { + #[cfg(test)] + if self + .restore_continuations + .paused + .load(std::sync::atomic::Ordering::Acquire) + { + return Ok(Step::Contended); + } + let lease = self.worker_operation_lock(worker_ref.worker_id)?; + // Another caller may be waiting for a backend task. Do not queue a join + // of that caller/driver while retaining the lock it needs to complete. + let _guard = match lease.try_lock() { + Ok(guard) => guard, + Err(std::sync::TryLockError::WouldBlock) => return Ok(Step::Contended), + Err(_) => return Err(RuntimeError::StatePoisoned), + }; + let (backend, pending) = { + let mut state = self.lock()?; + state.ensure_running()?; + state.refresh_uncertain_worker_write(worker_ref)?; + let worker = state.worker(worker_ref)?; + if worker.pending_stop.is_some() { + return Ok(Step::Finished); + } + let owner = worker + .restore_guard + .active_request_id + .as_ref() + .and_then(|id| worker.restore_guard.owners.get(id)); + if owner.is_some_and(|owner| owner.operation_id != operation || owner.receipt.is_some()) + { + return Ok(Step::Finished); + } + let Some(pending) = worker.pending_restore else { + return Ok(Step::Finished); + }; + if pending.operation_id != operation { + return Ok(Step::Finished); + } + if pending.commit == RestoreCommitFact::LegacyUnknown { + return Ok(Step::Retry( + "Saved Restore has no authoritative commit evidence".into(), + )); + } + state.ensure_subject_not_reserved(worker_ref, false)?; + if let Some(owner) = owner { + owner.validate_workdir_preparation_authority(worker)?; + } + let Some(backend) = state.execution_backend.clone() else { + return Ok(Step::Finished); + }; + (backend, pending) + }; + match backend.restore_recovery_disposition(operation, worker_ref) { + None => return Ok(Step::Finished), + Some(RestoreRecoveryDisposition::InFlight) => return Ok(Step::Waiting), + Some(RestoreRecoveryDisposition::Discard) + if pending.commit == RestoreCommitFact::NotCommitted => + { + match backend.discard_uncommitted_restore(operation, worker_ref) { + WorkerExecutionSpawnResult::RolledBack(result) => { + let receipt = RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::RolledBack, + "worker_restore_rolled_back", + result.message_or_default(), + ); + self.settle_restore_rollback(worker_ref, &receipt)?; + return Ok(Step::Finished); + } + result => { + return Ok(Step::Retry(format!( + "Restore cleanup remains unresolved: {result:?}" + ))); + } + } + } + Some(_) => {} + } + let result = self.restore_admitted_worker_under_lock(worker_ref, pending.mode)?; + if backend.restore_recovery_disposition(operation, worker_ref) + == Some(RestoreRecoveryDisposition::InFlight) + { + return Ok(Step::Waiting); + } + if result.state == WorkerRestoreState::ReconciliationRequired { + Ok(Step::Retry(result.message.unwrap_or_else(|| { + "Restore reconciliation remains pending".into() + }))) + } else { + Ok(Step::Finished) + } + } +} + +fn run( + inner: Weak>, + operations: Weak>>>>, + continuations: Weak, + owner: Arc, + worker_ref: WorkerRef, + operation: WorkerLifecycleOperationId, +) { + let mut repeated = 0u32; + let mut last_reason = None; + loop { + let (Some(inner), Some(worker_operations), Some(restore_continuations)) = ( + inner.upgrade(), + operations.upgrade(), + continuations.upgrade(), + ) else { + break; + }; + owner + .schedule + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .parked = false; + let runtime = Runtime { + inner, + worker_operations, + restore_continuations, + }; + let step = runtime.continue_restore_operation(&worker_ref, operation); + // No lifecycle lock or strong Runtime/registry reference survives waiting. + drop(runtime); + let mut schedule = owner + .schedule + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if schedule.cancelled { + break; + } + let contended = matches!(&step, Ok(Step::Contended)); + let reason = match step { + Ok(Step::Finished) => { + schedule.finished = true; + owner.wake.notify_all(); + break; + } + Ok(Step::Waiting) | Ok(Step::Contended) => None, + Ok(Step::Retry(reason)) => Some(reason), + Err(error) => Some(sanitize_worker_create_failure_message(&error.to_string())), + }; + if let Some(reason) = &reason { + repeated = repeated.saturating_add(1); + last_reason = Some(reason.clone()); + schedule.blocked_reason = Some(reason.clone()); + } + // Completed failed attempts can have signaled while this step joined + // them. Do not let our own unchanged failure reset the retry budget. + if reason.is_some() { + schedule.signaled = false; + } + if (!contended && reason.is_none()) || repeated >= 4 { + if repeated >= 4 { + schedule.parked = true; + owner.wake.notify_all(); + tracing::warn!(worker_id = %worker_ref.worker_id, operation_id = %operation, reason = ?last_reason, + "Restore continuation parked after bounded attempts; operation remains unresolved"); + } + schedule = owner + .wake + .wait_while(schedule, |s| !s.signaled && !s.cancelled) + .unwrap_or_else(std::sync::PoisonError::into_inner); + } else { + let delay = Duration::from_millis(100 * (1u64 << repeated.min(4))); + schedule = owner + .wake + .wait_timeout_while(schedule, delay, |s| !s.signaled && !s.cancelled) + .unwrap_or_else(std::sync::PoisonError::into_inner) + .0; + } + if schedule.cancelled { + break; + } + if schedule.signaled { + schedule.signaled = false; + // A completion wakes this same owner, not a new request. Its next + // observed result, not the signal itself, determines progress. + } + } +} diff --git a/crates/worker-runtime/src/test_support_tests.rs b/crates/worker-runtime/src/test_support_tests.rs index a4931def8..562ec2e8e 100644 --- a/crates/worker-runtime/src/test_support_tests.rs +++ b/crates/worker-runtime/src/test_support_tests.rs @@ -564,6 +564,7 @@ fn extraction_rejects_nonstopped_pending_schema_and_metadata_mismatch() { operation_id: WorkerLifecycleOperationId::new(), mode: PersistedWorkerRestoreMode::Explicit, commit: crate::runtime::RestoreCommitFact::NotCommitted, + source_repair_possible: false, request: request(), binding: None, last_settled_status: WorkerStatus::Stopped, diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index cf6bfe5d5..3d95f99ff 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -370,10 +370,23 @@ async fn prepare_restore_operation( factory: &F, request: WorkerExecutionRestoreRequest, ) -> Result, RuntimeRestoreFailure> { + let prior_effects = request.source_repair_possible; factory.preflight_restore(&request).await.map_err(|_| { - RuntimeRestoreFailure::refused(RuntimeRestoreStage::Preflight, "restore preflight refused") + let mut failure = RuntimeRestoreFailure::refused( + RuntimeRestoreStage::Preflight, + "restore preflight refused", + ); + failure.durable_effects_possible = prior_effects; + failure })?; - let repaired = factory.recover_restore_source(&request).await?; + let repaired = factory + .recover_restore_source(&request) + .await + .map_err(|mut failure| { + failure.durable_effects_possible |= prior_effects; + failure + })? + || prior_effects; let candidate = factory .prepare_restore(request) .await @@ -804,6 +817,7 @@ struct RestoreFaults { cleanup_fail: AtomicBool, socket_failure: AtomicBool, publication_failure: AtomicBool, + publication_attempts: std::sync::atomic::AtomicUsize, factory_gate: Mutex>>, factory_entered: Mutex>>, prepared_gate: Mutex>>, @@ -2626,6 +2640,11 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate { } let started = async { #[cfg(test)] + self.factory + .restore_faults + .publication_attempts + .fetch_add(1, Ordering::SeqCst); + #[cfg(test)] if self .factory .restore_faults @@ -3120,6 +3139,7 @@ struct RuntimeWorkerExecution { struct PendingRuntimeRestore { request: WorkerExecutionRestoreRequest, task: tokio::sync::Mutex>>, + task_in_flight: AtomicBool, result: Mutex, RuntimeRestoreFailure>>>, activated: Mutex>>, // Monotonic for this operation, not a claim that its current candidate has @@ -3131,7 +3151,24 @@ struct PendingRuntimeRestore { failure: Mutex>, } +struct RestoreTaskCompletion(Arc); + +impl Drop for RestoreTaskCompletion { + fn drop(&mut self) { + self.0.task_in_flight.store(false, Ordering::Release); + self.0 + .request + .context + .restore_progressed(self.0.request.operation_id); + } +} + impl PendingRuntimeRestore { + fn task_completion(self: &Arc) -> RestoreTaskCompletion { + self.task_in_flight.store(true, Ordering::Release); + RestoreTaskCompletion(self.clone()) + } + fn owns_prepared_candidate(&self) -> bool { matches!( self.result @@ -3455,7 +3492,9 @@ where if task.is_some() { return Err("restore task has not been joined".into()); } + let completion = pending.task_completion(); *task = Some(self.spawn_on_adapter_runtime(async move { + let _completion = completion; if resume_activation && !failure.needs_cleanup() && (failure.durable_effects_possible @@ -5340,6 +5379,7 @@ where let pending = Arc::new(PendingRuntimeRestore { request: request.clone(), task: tokio::sync::Mutex::new(None), + task_in_flight: AtomicBool::new(false), result: Mutex::new(None), activated: Mutex::new(None), activation_started: AtomicBool::new(false), @@ -5361,7 +5401,9 @@ where resources.insert(worker_ref.clone(), Arc::clone(&pending)); let factory = Arc::clone(&self.factory); let result_owner = Arc::clone(&pending); + let completion = pending.task_completion(); let task = self.spawn_on_adapter_runtime(async move { + let _completion = completion; let result = match prepare_restore_operation(factory.as_ref(), request.clone()) .await { @@ -5551,6 +5593,42 @@ where } } + fn restore_recovery_disposition( + &self, + operation_id: crate::execution::WorkerLifecycleOperationId, + worker_ref: &WorkerRef, + ) -> Option { + use crate::execution::RestoreRecoveryDisposition as D; + let pending = self.pending_restore(worker_ref).ok().flatten(); + Some(match pending { + Some(pending) if pending.request.operation_id == operation_id => { + if pending.task_in_flight.load(Ordering::Acquire) { + D::InFlight + } else if pending + .discard_result + .lock() + .ok() + .is_some_and(|r| matches!(*r, Some(Ok(())))) + { + // Cleanup is proven; prior repair effects may now require + // reconstruction, not repeated discard of a closed candidate. + D::Continue + } else if pending.discard_started.load(Ordering::Acquire) + || pending + .discard_result + .lock() + .ok() + .is_some_and(|r| r.is_some()) + { + D::Discard + } else { + D::Continue + } + } + _ => D::Continue, + }) + } + fn discard_uncommitted_restore( &self, operation_id: crate::execution::WorkerLifecycleOperationId, @@ -5582,8 +5660,10 @@ where self.wait_pending_restore(&pending, self.spawn_restore_timeout)?; if !pending.discard_started.swap(true, Ordering::AcqRel) { let owner = pending.clone(); + let completion = pending.task_completion(); let task = self.spawn_on_adapter_runtime(async move { + let _completion = completion; let retained = owner .result .lock() @@ -5709,8 +5789,10 @@ where pending.activation_started.store(true, Ordering::Release); let owner = pending.clone(); let factory = self.factory.clone(); + let completion = pending.task_completion(); let task = self.spawn_on_adapter_runtime(async move { + let _completion = completion; let candidate = owner .result .lock() @@ -6683,6 +6765,7 @@ mod tests { }); store.write(&metadata).unwrap(); WorkerExecutionRestoreRequest { + source_repair_possible: false, operation_id: crate::execution::WorkerLifecycleOperationId::new(), worker_ref: worker_ref.clone(), request, @@ -7614,6 +7697,7 @@ mod tests { Arc::new(PendingRuntimeRestore { request: direct_restore_request(worker_ref.clone()), task: tokio::sync::Mutex::new(None), + task_in_flight: AtomicBool::new(false), result: Mutex::new(None), activated: Mutex::new(None), activation_started: AtomicBool::new(false), @@ -7657,6 +7741,7 @@ mod tests { let pending = Arc::new(PendingRuntimeRestore { request: request.clone(), task: tokio::sync::Mutex::new(Some(failed)), + task_in_flight: AtomicBool::new(false), result: Mutex::new(None), activated: Mutex::new(None), activation_started: AtomicBool::new(false), @@ -10156,6 +10241,7 @@ mod tests { fn direct_restore_request(worker_ref: WorkerRef) -> WorkerExecutionRestoreRequest { WorkerExecutionRestoreRequest { + source_repair_possible: false, operation_id: crate::execution::WorkerLifecycleOperationId::new(), request: create_request("pending operation"), workspace_scope: None, @@ -10729,6 +10815,7 @@ mod tests { .with_runtime_store_dir(&runtime_store_dir) .with_remote_worker_mutation_identity(identity) .restore_controller(WorkerExecutionRestoreRequest { + source_repair_possible: false, operation_id: crate::execution::WorkerLifecycleOperationId::new(), worker_ref: worker_ref.clone(), request, @@ -10826,6 +10913,7 @@ mod tests { .with_runtime_store_dir(&runtime_store_dir) .with_controller_transport(WorkerControllerTransport::InProcess) .restore_controller(WorkerExecutionRestoreRequest { + source_repair_possible: false, operation_id: crate::execution::WorkerLifecycleOperationId::new(), worker_ref: worker_ref.clone(), request: create_request("embedded restore"), @@ -10990,6 +11078,7 @@ mod tests { runtime.stop_worker(&worker.worker_ref, None).unwrap(); let operation_id = crate::execution::WorkerLifecycleOperationId::new(); let restore_request = WorkerExecutionRestoreRequest { + source_repair_possible: false, operation_id, worker_ref: worker.worker_ref.clone(), request, @@ -11383,6 +11472,7 @@ mod tests { let detail = runtime.create_worker(request.clone()).unwrap(); runtime.stop_worker(&detail.worker_ref, None).unwrap(); let restore = WorkerExecutionRestoreRequest { + source_repair_possible: false, operation_id: crate::execution::WorkerLifecycleOperationId::new(), worker_ref: detail.worker_ref.clone(), request, diff --git a/crates/worker-runtime/src/worker_backend/restore_recovery.rs b/crates/worker-runtime/src/worker_backend/restore_recovery.rs index 54feb63bd..4c751e872 100644 --- a/crates/worker-runtime/src/worker_backend/restore_recovery.rs +++ b/crates/worker-runtime/src/worker_backend/restore_recovery.rs @@ -33,6 +33,10 @@ impl ProfileRuntimeWorkerFactory { if !pending && !incomplete { return Ok(false); } + request + .context + .authorize_restore_source_repair(request.operation_id) + .map_err(|_| cause("canonical source repair authority is not durably confirmed"))?; // The accepted Runtime operation already owns this Session. Its stable // Session identity and sole committed seed, not a new restore request, // are used to reconcile an interrupted first-head publication. diff --git a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs index 620fd140d..d622a7462 100644 --- a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs @@ -688,6 +688,7 @@ fn profile_restore_prior_effects_survive_read_only_reprepare_failure() { for read_stage in [RuntimeRestoreStage::Metadata, RuntimeRestoreStage::Reconstruction] { let activation_stage = RuntimeRestoreStage::ActivationEffects; let fixture = ProfileFailureFixture::new(Duration::from_secs(5)); + let _pause = fixture.runtime.test_pause_restore_continuation(); let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload(); fixture.fail(activation_stage); let request = fixture @@ -882,6 +883,7 @@ fn profile_restore_prior_effects_from_pending_head_repair_survive_read_only_repr #[serial_test::serial(worker_allocation)] fn profile_restore_prior_effects_survive_rebuilt_candidate_pre_effect_activation_failure() { let fixture = ProfileFailureFixture::new(Duration::from_secs(5)); + let _pause = fixture.runtime.test_pause_restore_continuation(); let (artifacts, pinned) = fixture.pin_unclaimed_restore_upload(); fixture.fail(RuntimeRestoreStage::ActivationEffects); let request = fixture diff --git a/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs b/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs index 394fc64d9..3cc482aab 100644 --- a/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_publication_tests.rs @@ -94,6 +94,141 @@ mod publication { (release, received) } + #[test] + #[serial_test::serial(worker_allocation)] + fn autonomous_late_prepare_and_publication_settle_without_another_request() { + for publication_failure in [false, true] { + let mut fixture = Fixture::no_model(); + Arc::get_mut(&mut fixture.backend).unwrap().spawn_restore_timeout = Duration::from_millis(250); + let runtime = stopped_runtime(&fixture); + let worker_ref = fixture.request.worker_ref.clone(); + let request = runtime.test_restore_request(&worker_ref); + let (release, prepared) = prepared_gate(&fixture); + fixture.backend.factory.restore_faults.publication_failure.store(publication_failure, Ordering::SeqCst); + let first = runtime.restore_worker_operation_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, request.clone()).unwrap(); + prepared.recv_timeout(WAIT).unwrap(); + assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired); + let operation = durable_operation(&runtime, &worker_ref).unwrap(); + assert_eq!(operation.commit, crate::runtime::RestoreCommitFact::NotCommitted); + fixture.assert_no_allocation(); + release.notify_one(); + runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT); + let loaded = runtime.test_runtime_store().load_runtime_state().unwrap(); + let receipt = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap(); + assert_eq!(receipt["owners"][&request.request_id]["receipt"]["state"], "accepted"); + assert!(durable_operation(&runtime, &worker_ref).is_none()); + let transport = fixture.backend.clone().attach_worker_protocol(&worker_ref).unwrap(); + transport.validate().unwrap(); + drop(transport); + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1); + fixture.assert_quiet(); + runtime.stop_worker_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, None).unwrap(); + fixture.assert_released(); + } + } + + #[test] + #[serial_test::serial(worker_allocation)] + fn autonomous_cleanup_dispatch_settles_uncommitted_candidate_without_repost_or_stop() { + let fixture = Fixture::no_model(); + let runtime = Arc::new(stopped_runtime(&fixture)); + let pause = runtime.test_pause_restore_continuation(); + let worker_ref = fixture.request.worker_ref.clone(); + let scope = fixture.request.workspace_scope.clone().unwrap(); + let request = runtime.test_restore_request(&worker_ref); + let before = fixture.canonical_files(); + let (release, prepared) = prepared_gate(&fixture); + let restoring = { + let runtime = runtime.clone(); + let worker_ref = worker_ref.clone(); + let scope = scope.clone(); + let request = request.clone(); + std::thread::spawn(move || runtime.restore_worker_operation_scoped(&scope, &worker_ref, request)) + }; + prepared.recv_timeout(WAIT).unwrap(); + runtime.test_runtime_store().fail_next_worker_write(AtomicWriteFault::BeforeRename); + fixture.backend.factory.restore_faults.cleanup_fail.store(true, Ordering::SeqCst); + release.notify_one(); + let first = restoring.join().unwrap().unwrap(); + assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired); + let operation = durable_operation(&runtime, &worker_ref).unwrap(); + assert_eq!(operation.commit, crate::runtime::RestoreCommitFact::NotCommitted); + let pending = fixture.backend.pending_restore(&worker_ref).unwrap().unwrap(); + assert!(matches!(pending.discard_result.lock().unwrap().as_ref(), Some(Err(_)))); + fixture.backend.factory.restore_faults.cleanup_fail.store(false, Ordering::SeqCst); + drop(pending); + drop(pause); + runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT); + let loaded = runtime.test_runtime_store().load_runtime_state().unwrap(); + let receipt = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap(); + assert_eq!(receipt["owners"][&request.request_id]["receipt"]["state"], "rolled_back"); + assert_eq!(fixture.canonical_files(), before); + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1); + fixture.assert_released(); + fixture.assert_quiet(); + } + + #[test] + #[serial_test::serial(worker_allocation)] + fn autonomous_unchanged_publication_obstruction_parks_without_false_terminal_outcome() { + let fixture = Fixture::no_model(); + let runtime = stopped_runtime(&fixture); + let worker_ref = fixture.request.worker_ref.clone(); + fixture.backend.factory.restore_faults.socket_failure.store(true, Ordering::SeqCst); + let request = runtime.test_restore_request(&worker_ref); + let first = runtime.restore_worker_operation_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, request.clone()).unwrap(); + assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired); + let pending = fixture.backend.pending_restore(&worker_ref).unwrap().unwrap(); + let reason = runtime.test_wait_restore_continuation_parked(&worker_ref, pending.request.operation_id, WAIT); + assert!(reason.contains("publication failed"), "{reason}"); + let durable = durable_operation(&runtime, &worker_ref).unwrap(); + assert_eq!(durable.commit, crate::runtime::RestoreCommitFact::Committed); + let loaded = runtime.test_runtime_store().load_runtime_state().unwrap(); + let guard = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap(); + assert!(guard["owners"][&request.request_id]["receipt"].is_null()); + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1); + // Initial call plus at most four backed-off continuation attempts. The + // parked condition is observed directly, not guessed using a sleep. + assert!(fixture.backend.factory.restore_faults.publication_attempts.load(Ordering::SeqCst) <= 5); + assert!(fixture.backend.workers.lock().unwrap().is_empty()); + assert!(fixture.backend.factory.observation_hub.get(&worker_ref).is_none()); + drop(pending); + // Permanent path-length obstruction: explicit fixture teardown only, + // not the successful completion path asserted by autonomous tests. + runtime.stop_worker_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, None).unwrap(); + fixture.assert_released(); + fixture.assert_quiet(); + } + + #[test] + #[serial_test::serial(worker_allocation)] + fn autonomous_late_failure_settles_without_another_request() { + let mut fixture = Fixture::no_model(); + Arc::get_mut(&mut fixture.backend).unwrap().spawn_restore_timeout = Duration::from_millis(250); + let runtime = stopped_runtime(&fixture); + let worker_ref = fixture.request.worker_ref.clone(); + let request = runtime.test_restore_request(&worker_ref); + let release = Arc::new(tokio::sync::Notify::new()); + let (entered, prepared) = mpsc::channel(); + *fixture.backend.factory.restore_faults.factory_gate.lock().unwrap() = Some(release.clone()); + *fixture.backend.factory.restore_faults.factory_entered.lock().unwrap() = Some(entered); + *fixture.backend.factory.restore_faults.fail_at.lock().unwrap() = Some(RuntimeRestoreStage::Feature); + let before = fixture.canonical_files(); + let first = runtime.restore_worker_operation_scoped(fixture.request.workspace_scope.as_ref().unwrap(), &worker_ref, request.clone()).unwrap(); + prepared.recv_timeout(WAIT).unwrap(); + assert_eq!(first.state, server_api::WorkerRestoreState::ReconciliationRequired); + let operation = durable_operation(&runtime, &worker_ref).unwrap(); + release.notify_one(); + runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT); + let loaded = runtime.test_runtime_store().load_runtime_state().unwrap(); + let receipt = serde_json::to_value(&loaded.workers[&worker_ref.worker_id].restore_guard).unwrap(); + assert_eq!(receipt["owners"][&request.request_id]["receipt"]["state"], "rolled_back"); + assert_eq!(fixture.canonical_files(), before); + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 1); + fixture.assert_released(); + fixture.assert_quiet(); + } + #[test] #[serial_test::serial(worker_allocation)] fn runtime_accepted_follows_real_prepared_snapshot_commit_attach_and_input() { @@ -288,6 +423,7 @@ mod publication { ] { let fixture = Fixture::no_model(); let runtime = Arc::new(stopped_runtime(&fixture)); + let _pause = runtime.test_pause_restore_continuation(); let worker_ref = fixture.request.worker_ref.clone(); let scope = fixture.request.workspace_scope.clone().unwrap(); let request = runtime.test_restore_request(&worker_ref); @@ -396,6 +532,7 @@ mod publication { fn uncommitted_cleanup_failure_fences_discarding_candidate_from_activation() { let fixture = Fixture::no_model(); let runtime = Arc::new(stopped_runtime(&fixture)); + let _pause = runtime.test_pause_restore_continuation(); let worker_ref = fixture.request.worker_ref.clone(); let scope = fixture.request.workspace_scope.clone().unwrap(); let request = runtime.test_restore_request(&worker_ref); @@ -443,6 +580,7 @@ mod publication { fn publication_failure_retains_same_committed_candidate_without_reconstruction() { let fixture = Fixture::no_model(); let runtime = stopped_runtime(&fixture); + let _pause = runtime.test_pause_restore_continuation(); let worker_ref = fixture.request.worker_ref.clone(); let scope = fixture.request.workspace_scope.clone().unwrap(); let request = runtime.test_restore_request(&worker_ref);