diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 39faadab5..ffa6e807f 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -621,9 +621,13 @@ impl fmt::Debug for WorkerProtocolTransport { /// authority: Runtime must reread its decision before choosing any action. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum RestoreRecoveryDisposition { + /// No live or pending owner for this exact operation. Not cleanup evidence. + Absent, InFlight, Continue, Discard, + /// A retained cleanup-only owner; never reconstruct or activate it. + Reclaim, } /// The connection selects this backend; WorkerRef identifies an execution, @@ -704,8 +708,15 @@ 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. + /// Opt in to Runtime-owned autonomous restore continuation. Pending tasks + /// signal completion through the original execution context; callbacks must + /// not wait for Runtime locks. Independent of resource disposition support. + fn supports_restore_continuation(&self) -> bool { + false + } + + /// Describe exact operation resource ownership for warm/cold recovery + /// dispatch. This does not opt in to callbacks or autonomous continuation. fn restore_recovery_disposition( &self, _operation_id: WorkerLifecycleOperationId, @@ -721,6 +732,24 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { self.restore_worker(request) } + /// Reclaim an uncommitted operation after restart without preparing, + /// repairing, or activating a replacement candidate. Runtime must establish + /// the durable uncommitted decision before calling this seam. Unsupported + /// cleanup is uncertainty, never a successful rollback. + fn reclaim_uncommitted_restore( + &self, + _request: WorkerExecutionRestoreRequest, + ) -> WorkerExecutionSpawnResult { + WorkerExecutionSpawnResult::ReconciliationRequired { + result: WorkerExecutionResult::unsupported( + WorkerExecutionOperation::Restore, + "backend cannot reclaim an uncommitted restore after restart", + ), + worker_state: None, + workdir_attachments: Vec::new(), + } + } + /// Discard only the exact unpublished operation after Runtime has proved /// its durable decision uncommitted. This is not Stop and must not affect a /// connected execution. RolledBack requires confirmed cleanup and no repair. @@ -1009,6 +1038,10 @@ impl WorkerExecutionBackendRef { .completed_restore_failure(operation_id, worker_ref) } + pub(crate) fn supports_restore_continuation(&self) -> bool { + self.backend.supports_restore_continuation() + } + pub(crate) fn restore_recovery_disposition( &self, operation_id: WorkerLifecycleOperationId, @@ -1025,6 +1058,13 @@ impl WorkerExecutionBackendRef { self.backend.reconcile_restore(request) } + pub(crate) fn reclaim_uncommitted_restore( + &self, + request: WorkerExecutionRestoreRequest, + ) -> WorkerExecutionSpawnResult { + self.backend.reclaim_uncommitted_restore(request) + } + pub(crate) fn discard_uncommitted_restore( &self, operation_id: WorkerLifecycleOperationId, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index ee9d0468e..f55d359f8 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -171,6 +171,12 @@ enum WorkerRestoreMode { Automatic, } +#[derive(Clone, Copy, Eq, PartialEq)] +enum RestoreRecoveryAction { + Continue, + ReclaimUncommitted, +} + #[cfg(feature = "fs-store")] impl From for PersistedWorkerRestoreMode { fn from(mode: WorkerRestoreMode) -> Self { @@ -2332,6 +2338,56 @@ impl Runtime { } } + // Seed durable crash checkpoints without launching a second Controller. + // Tests then reopen the real store with empty backend resource maps. + #[cfg(all(test, feature = "fs-store"))] + pub(crate) fn test_stage_restore_checkpoint( + &self, + worker_ref: &WorkerRef, + commit: RestoreCommitFact, + source_repair_possible: bool, + ) -> WorkerLifecycleOperationId { + let lease = self.worker_operation_lock(worker_ref.worker_id).unwrap(); + let _guard = lease.lock().unwrap(); + let operation = self + .lock() + .unwrap() + .worker(worker_ref) + .unwrap() + .pending_restore + .unwrap(); + self.begin_restore_operation( + worker_ref, + operation.operation_id, + operation.mode, + self.restore_candidate_execution_context(worker_ref.clone()), + ) + .unwrap(); + if source_repair_possible { + self.restore_repair_authorizer(worker_ref.clone())(operation.operation_id).unwrap(); + } + match commit { + RestoreCommitFact::Committed => { + self.commit_restored_worker_execution( + worker_ref, + protocol::WorkerStatus::Idle.into(), + WorkerStatus::Idle, + Vec::new(), + ) + .unwrap(); + } + RestoreCommitFact::LegacyUnknown => { + let mut state = self.lock().unwrap(); + let mut worker = state.worker(worker_ref).unwrap().clone(); + worker.pending_restore.as_mut().unwrap().commit = commit; + state.persist_restore_worker_record(&worker).unwrap(); + state.workers.insert(worker_ref.worker_id, worker); + } + RestoreCommitFact::NotCommitted => {} + } + operation.operation_id + } + #[cfg(test)] pub(crate) fn test_restore_request( &self, @@ -2357,6 +2413,19 @@ impl Runtime { &self, worker_ref: &WorkerRef, mode: WorkerRestoreMode, + ) -> Result { + self.restore_admitted_with_recovery_under_lock( + worker_ref, + mode, + RestoreRecoveryAction::Continue, + ) + } + + fn restore_admitted_with_recovery_under_lock( + &self, + worker_ref: &WorkerRef, + mode: WorkerRestoreMode, + recovery: RestoreRecoveryAction, ) -> Result { self.lock()? .ensure_subject_not_reserved(worker_ref, false)?; @@ -2407,14 +2476,28 @@ impl Runtime { } }; if let Some(detail) = committed_detail { - // Recover a receipt gap without launching execution again. + // A historical binding is not current readiness. The exact + // execution must still attest its operation and endpoint. + let backend = self.lock()?.execution_backend.clone(); + let readiness = backend + .ok_or_else(|| "Runtime has no Worker execution backend".to_owned()) + .and_then(|backend| { + backend.activate_restored_worker(owner.operation_id, worker_ref) + }); + if let Err(message) = readiness { + return Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_saved_binding_not_ready", + sanitize_worker_create_failure_message(&message), + )); + } let result = RuntimeWorkerRestoreResult::accepted(detail); self.finish_restore_receipt(worker_ref, &result)?; return Ok(result); } } } - let result = self.restore_worker_under_lock(worker_ref, mode)?; + let result = self.restore_worker_with_recovery_under_lock(worker_ref, mode, recovery)?; self.finish_restore_receipt(worker_ref, &result)?; Ok(result) } @@ -2451,6 +2534,19 @@ impl Runtime { &self, worker_ref: &WorkerRef, mode: WorkerRestoreMode, + ) -> Result { + self.restore_worker_with_recovery_under_lock( + worker_ref, + mode, + RestoreRecoveryAction::Continue, + ) + } + + fn restore_worker_with_recovery_under_lock( + &self, + worker_ref: &WorkerRef, + mode: WorkerRestoreMode, + recovery: RestoreRecoveryAction, ) -> Result { self.lock()? .ensure_subject_not_reserved(worker_ref, false)?; @@ -2562,8 +2658,12 @@ impl Runtime { .cloned() }); let request = WorkerExecutionRestoreRequest { - source_repair_possible: pending_restore - .is_some_and(|pending| pending.source_repair_possible), + // A committed decision also permits canonical activation effects. + // Cold reconstruction must retain that possibility through even a + // read-only failure, not cache it as an uncommitted rollback. + source_repair_possible: pending_restore.is_some_and(|pending| { + pending.source_repair_possible || pending.commit == RestoreCommitFact::Committed + }), operation_id, worker_ref: worker_ref.clone(), request: worker_request, @@ -2585,6 +2685,27 @@ impl Runtime { ) }; + let uncommitted = self + .lock()? + .worker(worker_ref)? + .pending_restore + .is_some_and(|pending| { + pending.operation_id == operation_id + && pending.commit == RestoreCommitFact::NotCommitted + }); + let reclaim = uncommitted + && (recovery == RestoreRecoveryAction::ReclaimUncommitted + || (reconcile + && match backend.restore_recovery_disposition(operation_id, worker_ref) { + Some(crate::execution::RestoreRecoveryDisposition::Reclaim) | None => true, + Some(crate::execution::RestoreRecoveryDisposition::Absent) => self + .lock()? + .worker(worker_ref)? + .restore_candidate_context + .is_none(), + Some(_) => false, + })); + let reconcile = reconcile || reclaim; if !reconcile && let Err(result) = backend.preflight_restore(&request) { // A preflight may attest an already-connected Controller snapshot. // Repeated restore is read-only; do not replace its operation owner. @@ -2643,6 +2764,8 @@ impl Runtime { let has_completed_failure = completed_failure.is_some(); let restore_result = if let Some(result) = completed_failure { result + } else if reclaim { + backend.reclaim_uncommitted_restore(request) } else if reconcile { backend.reconcile_restore(request) } else { @@ -4205,6 +4328,7 @@ impl Runtime { .lock() .map_err(|_| RuntimeError::StatePoisoned)?; if let Err(error) = self.reconcile_pending_stop_under_lock(&worker_ref) { + self.ensure_restore_continuation(&worker_ref)?; tracing::warn!( worker_id = %worker_ref.worker_id, error = %error, @@ -4242,8 +4366,34 @@ impl Runtime { let _operation_guard = operation_lock .lock() .map_err(|_| RuntimeError::StatePoisoned)?; - let result = self.restore_admitted_worker_under_lock(&worker_ref, mode)?; - self.record_startup_restore_result(&worker_ref, &result)?; + let recovery = { + let mut state = self.lock()?; + state.refresh_uncertain_worker_write(&worker_ref)?; + let pending = state.worker(&worker_ref)?.pending_restore; + let retained = pending.is_some_and(|pending| { + state.execution_backend.as_ref().is_some_and(|backend| { + matches!( + backend.restore_recovery_disposition(pending.operation_id, &worker_ref), + Some( + crate::execution::RestoreRecoveryDisposition::Continue + | crate::execution::RestoreRecoveryDisposition::InFlight + | crate::execution::RestoreRecoveryDisposition::Discard + ) + ) + }) + }); + if pending.is_some_and(|pending| pending.commit == RestoreCommitFact::NotCommitted) + && !retained + { + RestoreRecoveryAction::ReclaimUncommitted + } else { + RestoreRecoveryAction::Continue + } + }; + let result = + self.restore_admitted_with_recovery_under_lock(&worker_ref, mode, recovery); + self.ensure_restore_continuation(&worker_ref)?; + self.record_startup_restore_result(&worker_ref, &result?)?; } for worker_ref in candidates { @@ -4252,8 +4402,9 @@ impl Runtime { .lock() .map_err(|_| RuntimeError::StatePoisoned)?; let result = - self.restore_admitted_worker_under_lock(&worker_ref, WorkerRestoreMode::Automatic)?; - self.record_startup_restore_result(&worker_ref, &result)?; + self.restore_admitted_worker_under_lock(&worker_ref, WorkerRestoreMode::Automatic); + self.ensure_restore_continuation(&worker_ref)?; + self.record_startup_restore_result(&worker_ref, &result?)?; } Ok(()) } @@ -8411,6 +8562,7 @@ mod tests { restore_gate: Mutex>>, restore_count: Mutex, restore_operation_ids: Mutex>, + reclaim_operation_ids: Mutex>, activation_result: Mutex>>, activation_gate: Mutex>>, activation_operation_ids: Mutex>, @@ -8708,6 +8860,57 @@ mod tests { } } + fn restore_recovery_disposition( + &self, + operation_id: WorkerLifecycleOperationId, + _worker_ref: &WorkerRef, + ) -> Option { + Some( + if self + .restore_operation_ids + .lock() + .unwrap() + .contains(&operation_id) + { + crate::execution::RestoreRecoveryDisposition::Continue + } else { + crate::execution::RestoreRecoveryDisposition::Absent + }, + ) + } + + fn reclaim_uncommitted_restore( + &self, + request: WorkerExecutionRestoreRequest, + ) -> WorkerExecutionSpawnResult { + self.reclaim_operation_ids + .lock() + .unwrap() + .push(request.operation_id); + assert!( + !self + .contexts + .lock() + .unwrap() + .contains_key(&request.worker_ref.worker_id) + ); + if request.source_repair_possible { + WorkerExecutionSpawnResult::ReconciliationRequired { + result: WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + "prior repair may remain", + ), + worker_state: None, + workdir_attachments: Vec::new(), + } + } else { + WorkerExecutionSpawnResult::RolledBack(WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + "cold request resources confirmed absent", + )) + } + } + fn reconcile_restore( &self, request: WorkerExecutionRestoreRequest, @@ -9242,7 +9445,9 @@ mod tests { Some(&owner), ) .unwrap(); - assert_eq!(result.unwrap().state, WorkerRestoreState::Accepted); + assert_eq!(result.unwrap().state, WorkerRestoreState::RolledBack); + assert_eq!(*backend.restore_count.lock().unwrap(), 0); + assert_eq!(backend.reclaim_operation_ids.lock().unwrap().len(), 1); assert!(preparation.is_none()); assert!( runtime @@ -11212,7 +11417,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn restore_guard_interrupted_preparation_replays_original_pending_operation() { + fn restore_guard_interrupted_preparation_reclaims_original_pending_operation() { let dir = tempfile::tempdir().unwrap(); let options = FsRuntimeStoreOptions::new(dir.path().join("runtime")) .with_runtime_id("guard-partial-preparation"); @@ -11280,10 +11485,10 @@ mod tests { let receipt = resumed .restore_worker_operation(&worker.worker_ref, request) .unwrap(); - assert_eq!(receipt.state, WorkerRestoreState::Accepted); - assert_eq!(*backend.restore_count.lock().unwrap(), 1); + assert_eq!(receipt.state, WorkerRestoreState::RolledBack); + assert_eq!(*backend.restore_count.lock().unwrap(), 0); assert_eq!( - backend.restore_operation_ids.lock().unwrap().as_slice(), + backend.reclaim_operation_ids.lock().unwrap().as_slice(), &[operation_id] ); assert!( @@ -11422,7 +11627,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn restore_guard_unknown_preparation_write_retries_same_pending_operation() { + fn restore_guard_unknown_preparation_write_reclaims_same_pending_operation() { let dir = tempfile::tempdir().unwrap(); let options = FsRuntimeStoreOptions::new(dir.path().join("runtime")) .with_runtime_id("guard-unknown-preparation"); @@ -11484,12 +11689,16 @@ mod tests { .restore_worker_operation(&worker.worker_ref, request) .unwrap() .state, - WorkerRestoreState::Accepted + WorkerRestoreState::RolledBack + ); + assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 0); + assert_eq!( + resumed_backend.reclaim_operation_ids.lock().unwrap().len(), + 1 ); - assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 1); assert_eq!( resumed.worker_detail(&worker.worker_ref).unwrap().status, - WorkerStatus::Idle + WorkerStatus::Stopped ); } @@ -11667,7 +11876,7 @@ mod tests { Runtime::with_fs_store_and_execution_backend(options, resumed_backend.clone()).unwrap(); assert_eq!( resumed_backend - .restore_operation_ids + .reclaim_operation_ids .lock() .unwrap() .as_slice(), @@ -11676,8 +11885,8 @@ mod tests { let receipt = resumed .restore_worker_operation(&worker.worker_ref, request) .unwrap(); - assert_eq!(receipt.state, WorkerRestoreState::Accepted); - assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 1); + assert_eq!(receipt.state, WorkerRestoreState::RolledBack); + assert_eq!(*resumed_backend.restore_count.lock().unwrap(), 0); } #[test] @@ -15521,6 +15730,21 @@ mod tests { restarted_backend.clone(), ) .unwrap(); + assert_eq!( + restarted_backend + .reclaim_operation_ids + .lock() + .unwrap() + .as_slice(), + &[operation_id] + ); + assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 0); + assert_eq!( + restarted.worker_detail(&worker.worker_ref).unwrap().status, + WorkerStatus::Stopped + ); + // A deliberate fresh intent may restore after the old uncommitted + // request was reclaimed. It must allocate a different operation. let retry = restarted .restore_worker_operation( &worker.worker_ref, @@ -15529,13 +15753,9 @@ mod tests { .unwrap(); assert_eq!(retry.state, WorkerRestoreState::Accepted); assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 1); - assert_eq!( - restarted_backend - .restore_operation_ids - .lock() - .unwrap() - .as_slice(), - &[operation_id] + assert_ne!( + restarted_backend.restore_operation_ids.lock().unwrap()[0], + operation_id ); assert!( restarted diff --git a/crates/worker-runtime/src/runtime/restore_continuation.rs b/crates/worker-runtime/src/runtime/restore_continuation.rs index 3de663219..ad50b4d2d 100644 --- a/crates/worker-runtime/src/runtime/restore_continuation.rs +++ b/crates/worker-runtime/src/runtime/restore_continuation.rs @@ -266,9 +266,10 @@ impl Runtime { let Some(backend) = &state.execution_backend else { return Ok(()); }; - if backend - .restore_recovery_disposition(operation, &worker_ref) - .is_none() + if !backend.supports_restore_continuation() + || backend + .restore_recovery_disposition(operation, &worker_ref) + .is_none() { return Ok(()); } @@ -396,6 +397,10 @@ impl Runtime { "worker_restore_rolled_back", result.message_or_default(), ); + #[cfg(feature = "fs-store")] + self.lock()? + .record_restore_failure(worker_ref, &result, &receipt)?; + #[cfg(not(feature = "fs-store"))] self.settle_restore_rollback(worker_ref, &receipt)?; return Ok(Step::Finished); } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 3d95f99ff..7e1288701 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -2455,25 +2455,27 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .await .map_err(|_| RuntimeRestoreCleanupFailure::Workdir)?; } - if let Some(resources) = retained { - if let Some(allocation) = resources.allocation { - allocation - .release() - .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; - } else { - // Reconstruction may fail before returning its allocation guard. - // Absence is checked at the allocation authority, not inferred from Err. - let path = worker::runtime::worker_allocation::default_allocation_path() - .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; - let table = worker::runtime::worker_allocation::LockFileGuard::open(&path) - .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; - if table - .data() - .find(&Self::runtime_worker_name_for_ref(&request.worker_ref)) - .is_some() - { - return Err(RuntimeRestoreCleanupFailure::Allocation); - } + if let Some(allocation) = retained + .as_ref() + .and_then(|resources| resources.allocation.as_ref()) + { + allocation + .release() + .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; + } else { + // A cold restart loses the retained map as well as allocation guards. + // Prove absence at the allocation authority even with an empty map; + // never release a possibly different owner by Worker name. + let path = worker::runtime::worker_allocation::default_allocation_path() + .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; + let table = worker::runtime::worker_allocation::LockFileGuard::open(&path) + .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; + if table + .data() + .find(&Self::runtime_worker_name_for_ref(&request.worker_ref)) + .is_some() + { + return Err(RuntimeRestoreCleanupFailure::Allocation); } } // restore_controller starts the Controller only as its final fallible @@ -2486,10 +2488,18 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .map_err(|_| RuntimeRestoreCleanupFailure::Artifacts)?; Self::remove_restore_run_artifacts(&run_dir) .map_err(|_| RuntimeRestoreCleanupFailure::Artifacts)?; - self.failed_restore_resources + let mut resources = self + .failed_restore_resources .lock() - .map_err(|_| RuntimeRestoreCleanupFailure::Ownership)? - .remove(&request.worker_ref); + .map_err(|_| RuntimeRestoreCleanupFailure::Ownership)?; + if resources + .get(&request.worker_ref) + .is_some_and(|owner| owner.operation_id != request.operation_id) + { + return Err(RuntimeRestoreCleanupFailure::Ownership); + } + resources.remove(&request.worker_ref); + drop(resources); self.observation_hub.get(&request.worker_ref); Ok(()) } @@ -2763,6 +2773,29 @@ impl ProfileRuntimeWorkerFactory { let run_dir = self .worker_restore_run_dir(&request.worker_ref, request.operation_id) .map_err(|_| "restore run directory unavailable")?; + // A process restart loses the candidate, not its exact operation's + // disposable directory. Never infer writer absence from empty maps. + match std::fs::symlink_metadata(&run_dir) { + Ok(_) => { + self.stopped_restore_run_dirs(&request.worker_ref) + .map_err(|_| "restore run artifacts cannot be validated")?; + let path = worker::runtime::worker_allocation::default_allocation_path() + .map_err(|_| "restore allocation authority unavailable")?; + let table = worker::runtime::worker_allocation::LockFileGuard::open(&path) + .map_err(|_| "restore allocation authority unavailable")?; + if table + .data() + .find(&Self::runtime_worker_name_for_ref(&request.worker_ref)) + .is_some() + { + return Err("restore run directory still has an allocation owner"); + } + Self::remove_restore_run_artifacts(&run_dir) + .map_err(|_| "restore run artifacts cannot be reclaimed")?; + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(_) => return Err("restore run artifacts cannot be inspected"), + } let transport = self.controller_transport; #[cfg(test)] let transport = if self.restore_faults.socket_failure.load(Ordering::SeqCst) { @@ -3138,6 +3171,9 @@ struct RuntimeWorkerExecution { // marker. Timeout never drops the factory future or a late Controller result. struct PendingRuntimeRestore { request: WorkerExecutionRestoreRequest, + // Transient ownership mode, not a durable phase. Cold restart reclamation + // must never enter preparation, canonical repair, or activation. + cleanup_only: bool, task: tokio::sync::Mutex>>, task_in_flight: AtomicBool, result: Mutex, RuntimeRestoreFailure>>>, @@ -3469,6 +3505,7 @@ where pending: &Arc, resume_activation: bool, ) -> Result<(), String> { + let resume_activation = resume_activation && !pending.cleanup_only; let failure = pending .result .lock() @@ -5317,6 +5354,16 @@ where } }; if let Some(pending) = admitted_pending.as_ref() { + if pending.cleanup_only { + return WorkerExecutionSpawnResult::ReconciliationRequired { + result: WorkerExecutionResult::rejected( + WorkerExecutionOperation::Restore, + "cleanup-only restore cannot prepare or reconstruct execution", + ), + worker_state: None, + workdir_attachments: pending.request.previous_workdir_attachments.clone(), + }; + } if request.request != pending.request.request || request.workspace_scope != pending.request.workspace_scope || request.logical_workdir_attachments @@ -5378,6 +5425,7 @@ where .remove(&worker_ref); let pending = Arc::new(PendingRuntimeRestore { request: request.clone(), + cleanup_only: false, task: tokio::sync::Mutex::new(None), task_in_flight: AtomicBool::new(false), result: Mutex::new(None), @@ -5593,17 +5641,23 @@ where } } + fn supports_restore_continuation(&self) -> bool { + true + } + 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(); + let pending = self.pending_restore(worker_ref).ok()?; 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.cleanup_only { + D::Reclaim } else if pending .discard_result .lock() @@ -5625,10 +5679,132 @@ where D::Continue } } - _ => D::Continue, + _ => { + let workers = self.workers.lock().ok()?; + if workers + .get(worker_ref) + .is_some_and(|execution| execution.restore_operation_id == Some(operation_id)) + { + D::Continue + } else { + D::Absent + } + } }) } + fn reclaim_uncommitted_restore( + &self, + request: WorkerExecutionRestoreRequest, + ) -> WorkerExecutionSpawnResult { + let reclaimed = (|| -> Result { + let worker_ref = &request.worker_ref; + let operation_id = request.operation_id; + let operation_lock = self.worker_lock(worker_ref)?; + let _guard = operation_lock + .lock() + .map_err(|_| "worker operation lock is poisoned")?; + if request.context.worker_ref() != worker_ref { + return Err("restore cleanup context does not match Worker authority".into()); + } + if self + .workers + .lock() + .map_err(|_| "worker registry lock is poisoned")? + .contains_key(worker_ref) + { + return Err("uncommitted reclaim cannot target a connected execution".into()); + } + let pending = match self.pending_restore(worker_ref)? { + Some(pending) => { + if pending.request.operation_id != operation_id || !pending.cleanup_only { + return Err( + "uncommitted reclaim cannot replace an in-memory restore owner".into(), + ); + } + if request.request != pending.request.request + || request.workspace_scope != pending.request.workspace_scope + || request.logical_workdir_attachments + != pending.request.logical_workdir_attachments + || request.previous_workdir_attachments + != pending.request.previous_workdir_attachments + || request.source_repair_possible != pending.request.source_repair_possible + { + return Err( + "current Worker authority differs from the owned restore cleanup" + .into(), + ); + } + pending + } + None => { + if let Some(result) = self.completed_restore_failure(operation_id, worker_ref) { + return Ok(result); + } + let mut failure = RuntimeRestoreFailure::unproven( + RuntimeRestoreStage::Controller, + "restore decision was not committed before restart", + ); + failure.durable_effects_possible = request.source_repair_possible; + let pending = Arc::new(PendingRuntimeRestore { + request: request.clone(), + cleanup_only: true, + task: tokio::sync::Mutex::new(None), + task_in_flight: AtomicBool::new(false), + result: Mutex::new(Some(Err(failure))), + activated: Mutex::new(None), + activation_started: AtomicBool::new(false), + discard_started: AtomicBool::new(false), + discard_result: Mutex::new(None), + failure: Mutex::new(None), + }); + self.pending_restores + .lock() + .map_err(|_| "pending restore resources lock is poisoned")? + .insert(worker_ref.clone(), pending.clone()); + pending + } + }; + // Join the prior task before retrying. A timeout retains this owner, + // its original context notifier, and its exact task/JoinHandle. + self.wait_pending_restore(&pending, self.spawn_restore_timeout)?; + self.retry_pending_restore_cleanup(&pending, false)?; + self.wait_pending_restore(&pending, self.spawn_restore_timeout)?; + let failure = pending + .result + .lock() + .map_err(|_| "restore result lock is poisoned")? + .as_ref() + .and_then(|result| result.as_ref().err()) + .cloned() + .ok_or("restore cleanup has no retained failure result")?; + if failure.needs_cleanup() { + return Err(failure.to_string()); + } + if failure.durable_effects_possible { + return Err("Restore resource cleanup is confirmed; possible earlier canonical effects still require reconciliation".into()); + } + self.completed_restore_failures + .lock() + .map_err(|_| "restore failure receipts lock is poisoned")? + .insert(worker_ref.clone(), (operation_id, failure.clone())); + self.remove_pending_restore(worker_ref)?; + Ok(WorkerExecutionSpawnResult::RolledBack( + WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + failure.to_string(), + ), + )) + })(); + reclaimed.unwrap_or_else( + |message| WorkerExecutionSpawnResult::ReconciliationRequired { + result: WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message), + worker_state: None, + workdir_attachments: request.previous_workdir_attachments.clone(), + }, + ) + } + fn discard_uncommitted_restore( &self, operation_id: crate::execution::WorkerLifecycleOperationId, @@ -5643,6 +5819,7 @@ where .pending_restore(worker_ref)? .ok_or("unpublished restore owner is unavailable")?; if pending.request.operation_id != operation_id + || pending.cleanup_only || pending.activation_started.load(Ordering::Acquire) { return Err( @@ -5784,6 +5961,9 @@ where if pending.request.operation_id != operation_id { return Err("restored Worker operation identity does not match".into()); } + if pending.cleanup_only { + return Err("cleanup-only restore cannot activate execution".into()); + } self.wait_pending_restore(&pending, self.spawn_restore_timeout)?; if pending.owns_prepared_candidate() { pending.activation_started.store(true, Ordering::Release); @@ -7696,6 +7876,7 @@ mod tests { worker_ref.clone(), Arc::new(PendingRuntimeRestore { request: direct_restore_request(worker_ref.clone()), + cleanup_only: false, task: tokio::sync::Mutex::new(None), task_in_flight: AtomicBool::new(false), result: Mutex::new(None), @@ -7740,6 +7921,7 @@ mod tests { .unwrap(); let pending = Arc::new(PendingRuntimeRestore { request: request.clone(), + cleanup_only: false, task: tokio::sync::Mutex::new(Some(failed)), task_in_flight: AtomicBool::new(false), result: Mutex::new(None), @@ -10255,6 +10437,309 @@ mod tests { } } + // Backend component seam only: no saved Worker, model, or Controller is + // constructed. Any accidental preparation/recovery is an immediate failure. + struct CleanupOnlyFactory { + cleanup_calls: Arc, + cleanup_allowed: Arc, + release: Option>, + } + + #[async_trait] + impl RuntimeWorkerFactory for CleanupOnlyFactory { + async fn spawn_controller( + &self, + _: WorkerExecutionSpawnRequest, + ) -> Result { + panic!("reclaim must not spawn") + } + + async fn preflight_restore(&self, _: &WorkerExecutionRestoreRequest) -> Result<(), String> { + panic!("reclaim must not preflight preparation") + } + + async fn recover_restore_source( + &self, + _: &WorkerExecutionRestoreRequest, + ) -> Result { + panic!("reclaim must not repair canonical source") + } + + async fn prepare_restore( + &self, + _: WorkerExecutionRestoreRequest, + ) -> Result, RuntimeRestoreFailure> { + panic!("reclaim must not prepare") + } + + async fn cleanup_failed_restore( + &self, + _: &WorkerExecutionRestoreRequest, + ) -> Result<(), RuntimeRestoreCleanupFailure> { + if self.cleanup_calls.fetch_add(1, Ordering::SeqCst) == 0 { + if let Some(release) = &self.release { + release.notified().await; + } + } + if self.cleanup_allowed.load(Ordering::SeqCst) { + Ok(()) + } else { + Err(RuntimeRestoreCleanupFailure::Workdir) + } + } + + async fn reconcile_stopped_worker(&self, _: &WorkerRef) -> Result<(), String> { + panic!("reclaim must not clean all Worker operations through Stop") + } + } + + #[test] + fn cold_reclaim_retries_cleanup_only_and_keeps_cleaned_receipt_idempotent() { + use crate::execution::RestoreRecoveryDisposition as D; + let cleanup_calls = Arc::new(AtomicUsize::new(0)); + let cleanup_allowed = Arc::new(AtomicBool::new(false)); + let backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory { + cleanup_calls: cleanup_calls.clone(), + cleanup_allowed: cleanup_allowed.clone(), + release: None, + }) + .unwrap(); + let request = direct_restore_request(WorkerRef::new(WorkerId::now_v7())); + let worker_ref = &request.worker_ref; + assert_eq!( + backend.restore_recovery_disposition(request.operation_id, worker_ref), + Some(D::Absent) + ); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + let owner = backend.pending_restore(worker_ref).unwrap().unwrap(); + assert!(owner.cleanup_only); + assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1); + assert_eq!( + backend.restore_recovery_disposition(request.operation_id, worker_ref), + Some(D::Reclaim) + ); + // Wrong operation and changed authority must not replace or retry it. + let mut different = request.clone(); + different.operation_id = crate::execution::WorkerLifecycleOperationId::new(); + assert!(matches!( + backend.reclaim_uncommitted_restore(different), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + let mut mismatched = request.clone(); + mismatched.request = create_request("changed cleanup authority"); + assert!(matches!( + backend.reclaim_uncommitted_restore(mismatched), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + assert!(Arc::ptr_eq( + &owner, + &backend.pending_restore(worker_ref).unwrap().unwrap() + )); + // Erroneous generic callers cannot reconstruct, activate, or substitute + // in-process candidate discard for cold cleanup. + assert!(matches!( + backend.restore_worker(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + assert!( + backend + .activate_restored_worker(request.operation_id, worker_ref) + .is_err() + ); + assert!(matches!( + backend.discard_uncommitted_restore(request.operation_id, worker_ref), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1); + cleanup_allowed.store(true, Ordering::SeqCst); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::RolledBack(_) + )); + assert!(backend.pending_restore(worker_ref).unwrap().is_none()); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::RolledBack(_) + )); + assert!(matches!( + backend.restore_worker(request), + WorkerExecutionSpawnResult::RolledBack(_) + )); + assert_eq!(cleanup_calls.load(Ordering::SeqCst), 2); + assert!(backend.workers.lock().unwrap().is_empty()); + } + + #[test] + fn cold_reclaim_preserves_possible_effects_without_repreparing() { + use crate::execution::RestoreRecoveryDisposition as D; + let cleanup_calls = Arc::new(AtomicUsize::new(0)); + let backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory { + cleanup_calls: cleanup_calls.clone(), + cleanup_allowed: Arc::new(AtomicBool::new(true)), + release: None, + }) + .unwrap(); + let mut request = direct_restore_request(WorkerRef::new(WorkerId::now_v7())); + request.source_repair_possible = true; + for _ in 0..2 { + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + } + let owner = backend + .pending_restore(&request.worker_ref) + .unwrap() + .unwrap(); + assert!(owner.cleanup_only); + let retained = owner.result.lock().unwrap(); + let failure = retained.as_ref().unwrap().as_ref().err().unwrap(); + assert!(failure.durable_effects_possible); + assert_eq!(failure.cleanup, RuntimeRestoreCleanup::Confirmed); + drop(retained); + // Even an internal retry with the wrong resume flag cannot prepare. + backend.retry_pending_restore_cleanup(&owner, true).unwrap(); + assert!(owner.task.try_lock().unwrap().is_none()); + assert!( + backend + .completed_restore_failure(request.operation_id, &request.worker_ref) + .is_none() + ); + assert_eq!( + backend.restore_recovery_disposition(request.operation_id, &request.worker_ref), + Some(D::Reclaim) + ); + assert!(matches!( + backend.reconcile_restore(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + assert!( + backend + .activate_restored_worker(request.operation_id, &request.worker_ref) + .is_err() + ); + assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1); + } + + #[test] + fn cold_reclaim_timeout_retains_same_task_and_notifies_before_cleanup_retry() { + use crate::execution::RestoreRecoveryDisposition as D; + let cleanup_calls = Arc::new(AtomicUsize::new(0)); + let cleanup_allowed = Arc::new(AtomicBool::new(false)); + let release = Arc::new(tokio::sync::Notify::new()); + let mut backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory { + cleanup_calls: cleanup_calls.clone(), + cleanup_allowed: cleanup_allowed.clone(), + release: Some(release.clone()), + }) + .unwrap() + .with_spawn_restore_timeout(Duration::from_millis(1)); + let mut request = direct_restore_request(WorkerRef::new(WorkerId::now_v7())); + let (progress, progressed) = mpsc::channel(); + request.context = request + .context + .with_restore_progress(Arc::new(move |operation| { + let _ = progress.send(operation); + })); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + let owner = backend + .pending_restore(&request.worker_ref) + .unwrap() + .unwrap(); + let task_id = owner.task.try_lock().unwrap().as_ref().unwrap().id(); + assert_eq!( + backend.restore_recovery_disposition(request.operation_id, &request.worker_ref), + Some(D::InFlight) + ); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + assert!(Arc::ptr_eq( + &owner, + &backend + .pending_restore(&request.worker_ref) + .unwrap() + .unwrap() + )); + assert_eq!( + owner.task.try_lock().unwrap().as_ref().unwrap().id(), + task_id + ); + release.notify_one(); + assert_eq!( + progressed.recv_timeout(Duration::from_secs(5)).unwrap(), + request.operation_id + ); + backend + .wait_pending_restore(&owner, Duration::from_secs(5)) + .unwrap(); + assert!(owner.task.try_lock().unwrap().is_none()); + assert_eq!(cleanup_calls.load(Ordering::SeqCst), 1); + assert_eq!( + backend.restore_recovery_disposition(request.operation_id, &request.worker_ref), + Some(D::Reclaim) + ); + backend.spawn_restore_timeout = Duration::from_secs(5); + cleanup_allowed.store(true, Ordering::SeqCst); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::RolledBack(_) + )); + assert_eq!( + progressed.recv_timeout(Duration::from_secs(5)).unwrap(), + request.operation_id + ); + assert_eq!(cleanup_calls.load(Ordering::SeqCst), 2); + assert!(backend.workers.lock().unwrap().is_empty()); + } + + #[test] + fn cold_reclaim_refuses_existing_candidate_owner_even_for_same_operation() { + let backend = WorkerRuntimeExecutionBackend::new(CleanupOnlyFactory { + cleanup_calls: Arc::new(AtomicUsize::new(0)), + cleanup_allowed: Arc::new(AtomicBool::new(true)), + release: None, + }) + .unwrap(); + let request = direct_restore_request(WorkerRef::new(WorkerId::now_v7())); + let owner = Arc::new(PendingRuntimeRestore { + request: request.clone(), + cleanup_only: false, + 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), + discard_started: AtomicBool::new(false), + discard_result: Mutex::new(None), + failure: Mutex::new(None), + }); + backend + .pending_restores + .lock() + .unwrap() + .insert(request.worker_ref.clone(), owner.clone()); + assert!(matches!( + backend.reclaim_uncommitted_restore(request.clone()), + WorkerExecutionSpawnResult::ReconciliationRequired { .. } + )); + assert!(Arc::ptr_eq( + &owner, + &backend + .pending_restore(&request.worker_ref) + .unwrap() + .unwrap() + )); + assert_eq!(backend.factory.cleanup_calls.load(Ordering::SeqCst), 0); + } + #[test] fn absent_execution_stop_reconciles_pending_restore_cleanup_without_launching() { let restore_calls = Arc::new(AtomicUsize::new(0)); 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 d622a7462..99823169c 100644 --- a/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs +++ b/crates/worker-runtime/src/worker_backend_restore_failure_tests.rs @@ -193,6 +193,18 @@ max_tokens = 100 self.inner.restore_worker(request) } + fn restore_recovery_disposition( + &self, + operation_id: crate::execution::WorkerLifecycleOperationId, + worker_ref: &WorkerRef, + ) -> Option { + self.inner.restore_recovery_disposition(operation_id, worker_ref) + } + + fn reclaim_uncommitted_restore(&self, request: WorkerExecutionRestoreRequest) -> WorkerExecutionSpawnResult { + self.inner.reclaim_uncommitted_restore(request) + } + fn reconcile_restore( &self, request: WorkerExecutionRestoreRequest, 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 28323570f..d7518319b 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,133 @@ mod publication { (release, received) } + #[test] + #[serial_test::serial(worker_allocation)] + fn startup_fault_recovery_retains_cleanup_or_committed_reconstruction_without_repost() { + use crate::runtime::RestoreCommitFact as F; + for committed in [false, true] { + let fixture = Fixture::no_model(); + let runtime = stopped_runtime(&fixture); + let worker_ref = fixture.request.worker_ref.clone(); + let request = runtime.test_restore_request(&worker_ref); + runtime.coordinate_worker_restore_operation(&worker_ref, runtime_api::WorkerRestoreCoordinationRequest { + expected_observation_token: request.expected_observation_token.clone(), + request_id: request.request_id.clone(), preparation: request.preparation.clone(), + }, fixture.request.workspace_scope.as_ref()).unwrap(); + let operation = runtime.test_stage_restore_checkpoint(&worker_ref, if committed { F::Committed } else { F::NotCommitted }, false); + let before = fixture.canonical_files(); + drop(runtime); + if committed { + *fixture.backend.factory.restore_faults.fail_at.lock().unwrap() = Some(RuntimeRestoreStage::Metadata); + } else { + fixture.backend.factory.restore_faults.cleanup_fail.store(true, Ordering::SeqCst); + } + let options = FsRuntimeStoreOptions::new(fixture._root.path().join("catalog")).with_runtime_id("candidate-runtime"); + let restarted = EmbeddedRuntime::with_fs_store_and_execution_backend(options, fixture.backend.clone()).unwrap(); + let pause = restarted.test_pause_restore_continuation(); + assert_eq!(durable_operation(&restarted, &worker_ref).unwrap().operation_id, operation); + assert!(fixture.backend.completed_restore_failure(operation, &worker_ref).is_none()); + assert!(fixture.backend.workers.lock().unwrap().is_empty()); + *fixture.backend.factory.restore_faults.fail_at.lock().unwrap() = None; + fixture.backend.factory.restore_faults.cleanup_fail.store(false, Ordering::SeqCst); + drop(pause); + restarted.test_wait_restore_continuation(&worker_ref, operation, WAIT); + let receipt = restarted.restore_worker_operation(&worker_ref, request).unwrap(); + assert_eq!(receipt.state, if committed { server_api::WorkerRestoreState::Accepted } else { server_api::WorkerRestoreState::RolledBack }); + if committed { + let transport = fixture.backend.clone().attach_worker_protocol(&worker_ref).unwrap(); + transport.validate().unwrap(); + drop(transport); + restarted.stop_worker(&worker_ref, None).unwrap(); + } else { + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 0); + assert_eq!(fixture.canonical_files(), before); + } + fixture.assert_released(); + fixture.assert_quiet(); + } + } + + #[test] + #[serial_test::serial(worker_allocation)] + fn startup_dispatches_saved_decisions_without_replaying_uncommitted_execution() { + use crate::runtime::RestoreCommitFact as F; + for (commit, repaired, expected) in [ + (F::NotCommitted, false, server_api::WorkerRestoreState::RolledBack), + (F::NotCommitted, true, server_api::WorkerRestoreState::ReconciliationRequired), + (F::Committed, false, server_api::WorkerRestoreState::Accepted), + (F::LegacyUnknown, true, server_api::WorkerRestoreState::ReconciliationRequired), + ] { + let fixture = Fixture::no_model(); + let runtime = stopped_runtime(&fixture); + let worker_ref = fixture.request.worker_ref.clone(); + let scope = fixture.request.workspace_scope.as_ref().unwrap(); + let request = runtime.test_restore_request(&worker_ref); + assert!(runtime.coordinate_worker_restore_operation(&worker_ref, runtime_api::WorkerRestoreCoordinationRequest { + expected_observation_token: request.expected_observation_token.clone(), + request_id: request.request_id.clone(), + preparation: request.preparation.clone(), + }, Some(scope)).unwrap().0.is_none()); + let operation = runtime.test_stage_restore_checkpoint(&worker_ref, commit, repaired); + let before = fixture.canonical_files(); + let run = fixture.backend.factory.worker_restore_run_dir(&worker_ref, operation).unwrap(); + std::fs::create_dir_all(&run).unwrap(); + std::fs::write(run.join("before-restart.marker"), "disposable").unwrap(); + if commit == F::NotCommitted && !repaired { + // Fixture for an interrupted automatic restore. A rollback must + // not leave an active/Automatic projection that restores again. + let store = runtime.test_runtime_store(); + let mut record = store.read_worker_record(&worker_ref).unwrap(); + record.status = crate::catalog::WorkerStatus::Idle; + let PersistedWorkerExecutionState::ReconciliationRequired( + crate::fs_store::PersistedWorkerLifecycleOperation::Restore(saved), + ) = &mut record.execution_state else { panic!("pending fixture") }; + saved.mode = crate::fs_store::PersistedWorkerRestoreMode::Automatic; + saved.last_settled_status = crate::catalog::WorkerStatus::Idle; + saved.binding = Some(crate::fs_store::PersistedWorkerExecutionBinding {}); + store.write_worker_record(&record).unwrap(); + } + drop(runtime); + assert!(fixture.backend.pending_restore(&worker_ref).unwrap().is_none()); + assert!(fixture.backend.workers.lock().unwrap().is_empty()); + let options = FsRuntimeStoreOptions::new(fixture._root.path().join("catalog")).with_runtime_id("candidate-runtime"); + let restarted = EmbeddedRuntime::with_fs_store_and_execution_backend(options.clone(), fixture.backend.clone()).unwrap(); + if expected == server_api::WorkerRestoreState::ReconciliationRequired { + let reason = restarted.test_wait_restore_continuation_parked(&worker_ref, operation, WAIT); + assert!(!reason.is_empty()); + let pending = durable_operation(&restarted, &worker_ref).unwrap(); + assert_eq!(pending.operation_id, operation); + assert_eq!(pending.commit, commit); + assert_eq!(pending.source_repair_possible, repaired); + } + assert!(restarted.worker_detail(&worker_ref).is_ok(), "{commit:?}: {:?}", restarted.diagnostics()); + let receipt = restarted.restore_worker_operation(&worker_ref, request.clone()).unwrap(); + assert_eq!(receipt.state, expected, "{commit:?}: {receipt:?}; backend failure: {:?}", fixture.backend.completed_restore_failure(operation, &worker_ref)); + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), usize::from(commit == F::Committed)); + if commit == F::Committed { + let transport = fixture.backend.clone().attach_worker_protocol(&worker_ref).unwrap(); + transport.validate().unwrap(); + drop(transport); + } else { + assert_eq!(fixture.canonical_files(), before); + assert!(fixture.backend.workers.lock().unwrap().is_empty()); + assert!(fixture.backend.factory.observation_hub.get(&worker_ref).is_none()); + assert_eq!(run.exists(), commit == F::LegacyUnknown); + } + if expected == server_api::WorkerRestoreState::RolledBack { + assert_eq!(restarted.worker_detail(&worker_ref).unwrap().status, crate::catalog::WorkerStatus::Stopped); + drop(restarted); + let second = EmbeddedRuntime::with_fs_store_and_execution_backend(options, fixture.backend.clone()).unwrap(); + assert_eq!(second.restore_worker_operation_scoped(scope, &worker_ref, request).unwrap(), receipt); + assert_eq!(fixture.backend.factory.restore_faults.executions.load(Ordering::SeqCst), 0); + } else { + restarted.stop_worker_scoped(scope, &worker_ref, None).unwrap(); + } + fixture.assert_released(); + fixture.assert_quiet(); + } + } + #[test] #[serial_test::serial(worker_allocation)] fn stop_timeout_retains_autonomous_late_cleanup_and_blocks_remove_until_confirmed() {