diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index a2b544fb..d927a4de 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -164,6 +164,49 @@ pub struct WorkerExecutionStatus { pub binding: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub working_directory: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub restore_dry_check: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum WorkerRestoreDryCheckStatus { + Valid, + Invalid, + Unavailable, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRestoreDryCheck { + pub status: WorkerRestoreDryCheckStatus, + pub code: String, + pub message: String, +} + +impl WorkerRestoreDryCheck { + pub fn valid(message: impl Into) -> Self { + Self { + status: WorkerRestoreDryCheckStatus::Valid, + code: "restore_dry_check_valid".to_string(), + message: message.into(), + } + } + + pub fn invalid(code: impl Into, message: impl Into) -> Self { + Self { + status: WorkerRestoreDryCheckStatus::Invalid, + code: code.into(), + message: message.into(), + } + } + + pub fn unavailable(code: impl Into, message: impl Into) -> Self { + Self { + status: WorkerRestoreDryCheckStatus::Unavailable, + code: code.into(), + message: message.into(), + } + } } impl WorkerExecutionStatus { @@ -177,18 +220,21 @@ impl WorkerExecutionStatus { run_state, binding: None, working_directory: None, + restore_dry_check: None, } } pub fn stopped_from(mut previous: Self) -> Self { previous.backend = WorkerExecutionBackendKind::Stopped; previous.run_state = WorkerExecutionRunState::Stopped; + previous.restore_dry_check = None; previous } pub fn corrupted(mut previous: Self) -> Self { previous.backend = WorkerExecutionBackendKind::Corrupted; previous.run_state = WorkerExecutionRunState::Errored; + previous.restore_dry_check = None; previous } @@ -206,6 +252,14 @@ impl WorkerExecutionStatus { self.run_state = result.run_state; self } + + pub fn with_restore_dry_check( + mut self, + restore_dry_check: Option, + ) -> Self { + self.restore_dry_check = restore_dry_check; + self + } } /// Opaque per-Worker execution handle returned by a backend. @@ -308,6 +362,16 @@ pub struct WorkerExecutionSpawnRequest { pub config_bundle: Option, } +/// Request passed to a [`WorkerExecutionBackend`] when validating a restore without side effects. +#[derive(Clone, Debug)] +pub struct WorkerExecutionRestoreDryRequest { + pub worker_ref: WorkerRef, + pub request: crate::catalog::CreateWorkerRequest, + pub previous_execution: WorkerExecutionStatus, + pub working_directory: Option, + pub config_bundle: Option, +} + /// Request passed to a [`WorkerExecutionBackend`] when restoring a persisted Worker. #[derive(Clone, Debug)] pub struct WorkerExecutionRestoreRequest { @@ -350,6 +414,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult; + fn dry_restore_worker( + &self, + _request: WorkerExecutionRestoreDryRequest, + ) -> WorkerRestoreDryCheck { + WorkerRestoreDryCheck::unavailable( + "restore_dry_check_unsupported", + "execution backend does not support side-effect-free restore validation", + ) + } + fn restore_worker( &self, _request: WorkerExecutionRestoreRequest, @@ -473,6 +547,13 @@ impl WorkerExecutionBackendRef { self.backend.spawn_worker(request) } + pub(crate) fn dry_restore_worker( + &self, + request: WorkerExecutionRestoreDryRequest, + ) -> WorkerRestoreDryCheck { + self.backend.dry_restore_worker(request) + } + pub(crate) fn restore_worker( &self, request: WorkerExecutionRestoreRequest, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 8f987acb..a4b1fe09 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -11,14 +11,13 @@ use crate::config_bundle::{ use crate::diagnostics::DiagnosticSeverity; use crate::diagnostics::RuntimeDiagnostic; use crate::error::RuntimeError; -#[cfg(feature = "fs-store")] -use crate::execution::WorkerExecutionRestoreRequest; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionBackendKind, WorkerExecutionBackendRef, WorkerExecutionBindingIdentity, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, WorkerExecutionStatus, + WorkerExecutionSpawnResult, WorkerExecutionStatus, WorkerRestoreDryCheckStatus, }; +use crate::execution::{WorkerExecutionRestoreDryRequest, WorkerExecutionRestoreRequest}; #[cfg(feature = "fs-store")] use crate::fs_store::{ FsRuntimeStore, FsRuntimeStoreOptions, PersistedRuntimeState, PersistedWorkerRecord, @@ -434,11 +433,16 @@ impl Runtime { /// List Workers known to this Runtime. pub fn list_workers(&self) -> Result, RuntimeError> { let state = self.lock()?; - Ok(state - .workers - .values() - .map(|worker| worker.summary()) - .collect()) + let mut summaries = Vec::with_capacity(state.workers.len()); + for worker in state.workers.values() { + let restore_dry_check = corrupted_worker_restore_dry_check( + state.execution_backend.as_ref(), + worker, + &state.config_bundles, + ); + summaries.push(worker.summary_with_restore_dry_check(restore_dry_check)); + } + Ok(summaries) } /// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime. @@ -513,6 +517,7 @@ impl Runtime { run_state: dispatch_result.run_state, binding: worker.execution.binding.clone(), working_directory: worker.execution.working_directory.clone(), + restore_dry_check: None, }; let status = worker.status; @@ -670,6 +675,7 @@ impl Runtime { run_state: result.run_state, binding: worker.execution.binding.clone(), working_directory: worker.execution.working_directory.clone(), + restore_dry_check: None, }; state.persist_worker(&worker_ref.worker_id)?; Ok(()) @@ -1039,6 +1045,7 @@ impl Runtime { worker_ref: WorkerRef, request: CreateWorkerRequest, previous_execution: WorkerExecutionStatus, + config_bundle: Option, } let candidates = { @@ -1105,10 +1112,17 @@ impl Runtime { continue; } } + let config_bundle = worker + .request + .config_bundle + .as_ref() + .and_then(|bundle_ref| state.config_bundles.get(&bundle_ref.id)) + .cloned(); candidates.push(RestoreCandidate { worker_ref: worker.worker_ref.clone(), request: worker.request.clone(), previous_execution: worker.execution.clone(), + config_bundle, }); } candidates @@ -1122,6 +1136,36 @@ impl Runtime { let Some(backend) = backend else { return Ok(()); }; + let dry_check = backend.dry_restore_worker(WorkerExecutionRestoreDryRequest { + worker_ref: candidate.worker_ref.clone(), + request: candidate.request.clone(), + previous_execution: candidate.previous_execution.clone(), + working_directory: None, + config_bundle: candidate.config_bundle.clone(), + }); + match dry_check.status { + WorkerRestoreDryCheckStatus::Valid => {} + WorkerRestoreDryCheckStatus::Invalid => { + let mut state = self.lock()?; + state.record_restore_failure( + &candidate.worker_ref, + WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + dry_check.message, + ), + )?; + continue; + } + WorkerRestoreDryCheckStatus::Unavailable => { + let mut state = self.lock()?; + state.record_restore_unavailable( + &candidate.worker_ref, + candidate.previous_execution.clone(), + dry_check.message, + )?; + continue; + } + } let request = WorkerExecutionRestoreRequest { worker_ref: candidate.worker_ref.clone(), request: candidate.request, @@ -1618,6 +1662,33 @@ impl RuntimeState { Ok(()) } + #[cfg(feature = "fs-store")] + fn record_restore_unavailable( + &mut self, + worker_ref: &WorkerRef, + previous_execution: WorkerExecutionStatus, + message: String, + ) -> Result<(), RuntimeError> { + let diagnostic_id = self.next_diagnostic_id; + self.next_diagnostic_id += 1; + self.diagnostics.push(RuntimeDiagnostic { + id: diagnostic_id, + severity: DiagnosticSeverity::Warning, + code: "worker_execution_restore_unavailable".to_string(), + message: format!( + "worker {} execution restore deferred: {message}", + worker_ref.worker_id + ), + worker_ref: Some(worker_ref.clone()), + }); + let worker = self.worker_mut(worker_ref)?; + worker.execution_handle = None; + worker.execution = WorkerExecutionStatus::stopped_from(previous_execution); + self.persist_runtime_snapshot()?; + self.persist_worker(&worker_ref.worker_id)?; + Ok(()) + } + fn push_event( &mut self, worker_ref: Option, @@ -1744,13 +1815,50 @@ struct WorkerRecord { last_event_id: u64, } +fn corrupted_worker_restore_dry_check( + execution_backend: Option<&WorkerExecutionBackendRef>, + worker: &WorkerRecord, + config_bundles: &BTreeMap, +) -> Option { + if worker.execution.backend != WorkerExecutionBackendKind::Corrupted { + return None; + } + let Some(execution_backend) = execution_backend else { + return Some(crate::execution::WorkerRestoreDryCheck::unavailable( + "restore_dry_check_backend_unavailable", + "execution backend is unavailable; restore dry-test was not run", + )); + }; + let config_bundle = worker + .request + .config_bundle + .as_ref() + .and_then(|bundle_ref| config_bundles.get(&bundle_ref.id)) + .cloned(); + Some( + execution_backend.dry_restore_worker(WorkerExecutionRestoreDryRequest { + worker_ref: worker.worker_ref.clone(), + request: worker.request.clone(), + previous_execution: worker.execution.clone(), + working_directory: None, + config_bundle, + }), + ) +} + impl WorkerRecord { - fn summary(&self) -> WorkerSummary { + fn summary_with_restore_dry_check( + &self, + restore_dry_check: Option, + ) -> WorkerSummary { WorkerSummary { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, - execution: self.execution.clone(), + execution: self + .execution + .clone() + .with_restore_dry_check(restore_dry_check), profile: self.request.profile.clone(), profile_source: self.request.profile_source.reference(), config_bundle: self.request.config_bundle.clone(), @@ -1888,9 +1996,11 @@ mod tests { }; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle, - WorkerExecutionRestoreRequest, WorkerExecutionRunState, + WorkerExecutionRestoreDryRequest, WorkerExecutionRestoreRequest, WorkerExecutionRunState, }; use std::collections::BTreeMap; + #[cfg(feature = "fs-store")] + use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; fn task_request(_objective: &str) -> CreateWorkerRequest { @@ -1959,6 +2069,8 @@ mod tests { dispatch_result: Mutex>, restore_result: Mutex>, restore_count: Mutex, + dry_restore_result: Mutex>, + dry_restore_count: Mutex, contexts: Mutex>, #[cfg(feature = "ws-server")] snapshots: Mutex>, @@ -1969,6 +2081,18 @@ mod tests { *self.dispatch_result.lock().unwrap() = Some(result); } + fn set_dry_restore_result(&self, result: crate::execution::WorkerRestoreDryCheck) { + *self.dry_restore_result.lock().unwrap() = Some(result); + } + + fn dry_restore_count(&self) -> u64 { + *self.dry_restore_count.lock().unwrap() + } + + fn restore_count(&self) -> u64 { + *self.restore_count.lock().unwrap() + } + #[cfg(feature = "ws-server")] fn set_worker_snapshot(&self, worker_ref: &WorkerRef, snapshot: protocol::Event) { self.snapshots @@ -2009,6 +2133,20 @@ mod tests { } } + fn dry_restore_worker( + &self, + _request: WorkerExecutionRestoreDryRequest, + ) -> crate::execution::WorkerRestoreDryCheck { + *self.dry_restore_count.lock().unwrap() += 1; + self.dry_restore_result + .lock() + .unwrap() + .clone() + .unwrap_or_else(|| { + crate::execution::WorkerRestoreDryCheck::valid("test dry restore ok") + }) + } + fn restore_worker( &self, request: WorkerExecutionRestoreRequest, @@ -2121,6 +2259,35 @@ mod tests { assert_eq!(fetched.profile, detail.profile); } + #[test] + fn list_corrupted_worker_runs_restore_dry_check_and_reports_error() { + let (runtime, backend) = runtime_and_backend(); + let detail = runtime + .create_worker(task_request("restore dry-check")) + .unwrap(); + backend.set_dry_restore_result(crate::execution::WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + "missing Worker metadata for worker-1", + )); + { + let mut state = runtime.lock().unwrap(); + let worker = state.worker_mut(&detail.worker_ref).unwrap(); + worker.execution = WorkerExecutionStatus::corrupted(worker.execution.clone()); + } + + let list = runtime.list_workers().unwrap(); + + assert_eq!(backend.dry_restore_count(), 1); + let dry_check = list[0].execution.restore_dry_check.as_ref().unwrap(); + assert_eq!( + dry_check.status, + crate::execution::WorkerRestoreDryCheckStatus::Invalid + ); + assert_eq!(dry_check.code, "restore_dry_check_failed"); + assert!(dry_check.message.contains("missing Worker metadata")); + assert_eq!(backend.restore_count(), 0); + } + #[test] fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() { let runtime = runtime_with_backend(); @@ -2763,6 +2930,63 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[cfg(feature = "fs-store")] + #[test] + fn fs_store_marks_worker_corrupted_when_restore_dry_check_fails() { + let root = fs_store_root("execution-dry-restore-failed"); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + display_name: None, + limits: RuntimeLimits::default(), + }, + Arc::new(TestExecutionBackend::default()), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("dry restore failure")) + .unwrap(); + drop(runtime); + + let restoring_backend = Arc::new(TestExecutionBackend::default()); + restoring_backend.set_dry_restore_result(crate::execution::WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + "missing Worker metadata for dry restore failure", + )); + let restored = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + display_name: None, + limits: RuntimeLimits::default(), + }, + restoring_backend.clone(), + ) + .unwrap(); + + assert_eq!(restoring_backend.dry_restore_count(), 1); + assert_eq!(restoring_backend.restore_count(), 0); + let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap(); + assert_eq!(restored_worker.status, WorkerStatus::Running); + assert_eq!( + restored_worker.execution.backend, + WorkerExecutionBackendKind::Corrupted + ); + assert!( + restored + .diagnostics() + .unwrap() + .iter() + .any( + |diagnostic| diagnostic.code == "worker_execution_restore_failed" + && diagnostic.message.contains("missing Worker metadata") + && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) + ) + ); + + let _ = std::fs::remove_dir_all(root); + } + #[cfg(feature = "fs-store")] #[test] fn fs_store_marks_worker_corrupted_when_execution_restore_fails() { diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index aa9230f5..4949a899 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -20,8 +20,9 @@ use crate::catalog::{ }; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, - WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionRunState, - WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, + WorkerExecutionRestoreDryRequest, WorkerExecutionRestoreRequest, WorkerExecutionResult, + WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, + WorkerRestoreDryCheck, }; use crate::interaction::{WorkerInput, WorkerInputKind}; use crate::resource::{BackendResourceClient, ProfileSourceArchiveCache}; @@ -32,7 +33,7 @@ use async_trait::async_trait; use manifest::paths; use protocol::{Method, Segment, WorkerStatus}; use session_store::FsStore; -use session_store::{CombinedStore, FsWorkerStore}; +use session_store::{CombinedStore, FsWorkerStore, Store, WorkerMetadataStore}; use tokio::runtime::Runtime; #[cfg(feature = "ws-server")] use tokio::sync::broadcast; @@ -60,6 +61,16 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { &self, request: WorkerExecutionRestoreRequest, ) -> Result; + + async fn dry_restore_controller( + &self, + _request: WorkerExecutionRestoreDryRequest, + ) -> WorkerRestoreDryCheck { + WorkerRestoreDryCheck::unavailable( + "restore_dry_check_unsupported", + "runtime worker factory does not support side-effect-free restore validation", + ) + } } /// Production factory that resolves a normal Worker profile and spawns it under @@ -328,6 +339,83 @@ async fn fetch_profile_source_archive_http( ) } +impl ProfileRuntimeWorkerFactory { + async fn try_dry_restore_controller( + &self, + request: WorkerExecutionRestoreDryRequest, + ) -> Result<(), String> { + let WorkerExecutionRestoreDryRequest { + worker_ref, + request, + previous_execution: _, + working_directory, + config_bundle: _, + } = request; + let store_dir = self.store_dir()?; + let session_store = FsStore::new(&store_dir).map_err(|err| { + format!( + "failed to initialize session store at {}: {err}", + store_dir.display() + ) + })?; + let worker_metadata_dir = self.worker_metadata_dir(&store_dir); + let worker_metadata_store = FsWorkerStore::new(&worker_metadata_dir).map_err(|err| { + format!( + "failed to initialize worker metadata store at {}: {err}", + worker_metadata_dir.display() + ) + })?; + let worker_name = Self::runtime_worker_name_for_ref(&worker_ref); + let active_segment = worker_metadata_store + .read_by_name(&worker_name) + .map_err(|err| format!("failed to read Worker metadata for {worker_name}: {err}"))? + .ok_or_else(|| format!("missing Worker metadata for {worker_name}"))? + .active + .ok_or_else(|| format!("Worker metadata for {worker_name} has no active segment"))?; + let active_segment_id = active_segment + .segment_id + .ok_or_else(|| format!("Worker metadata for {worker_name} has no active segment id"))?; + let raw_entries = session_store + .read_all(active_segment.session_id, active_segment_id) + .map_err(|err| { + format!( + "failed to read active segment {} for session {}: {err}", + active_segment_id, active_segment.session_id + ) + })?; + let state = session_store::collect_state(&raw_entries); + if state.entries_count == 0 { + return Err(format!("active segment {active_segment_id} is empty")); + } + let worker_root = working_directory + .as_ref() + .map(|binding| binding.root.clone()) + .unwrap_or_else(|| self.profile_base_dir.clone()); + let profile = self.runtime_profile_for_request(&request); + let selector = profile.as_deref().unwrap_or("builtin:default"); + let archive = self + .resolve_profile_source_archive(&request.profile_source) + .await?; + let manifest = archive + .resolve_profile(selector, &worker_root, &worker_name) + .map_err(|err| format!("failed to resolve profile source archive: {err}"))?; + if working_directory.is_some() { + worker::entrypoint::resolve_runtime_profile_manifest_from_manifest( + manifest, + &worker_root, + &worker_name, + )?; + } else { + worker::entrypoint::resolve_runtime_profile_manifest_from_manifest_without_filesystem( + manifest, + &worker_root, + &worker_name, + )?; + } + Ok(()) + } +} + #[async_trait] impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { async fn spawn_controller( @@ -411,6 +499,16 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { Ok(handle) } + async fn dry_restore_controller( + &self, + request: WorkerExecutionRestoreDryRequest, + ) -> WorkerRestoreDryCheck { + match self.try_dry_restore_controller(request).await { + Ok(()) => WorkerRestoreDryCheck::valid("restore dry-check succeeded"), + Err(message) => WorkerRestoreDryCheck::invalid("restore_dry_check_failed", message), + } + } + async fn restore_controller( &self, request: WorkerExecutionRestoreRequest, @@ -923,6 +1021,75 @@ where ) } + fn dry_restore_worker( + &self, + mut request: WorkerExecutionRestoreDryRequest, + ) -> WorkerRestoreDryCheck { + let working_directory = match request.previous_execution.working_directory.clone() { + Some(status) => { + let Some(materializer) = self.working_directory_materializer.as_ref() else { + return WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + "persisted worker has a working directory binding, but no materializer is configured for this runtime backend", + ); + }; + let relative_cwd = request + .request + .working_directory + .as_ref() + .and_then(|working_directory| working_directory.relative_cwd.as_deref()); + match materializer + .bind_working_directory(&status.summary.working_directory_id, relative_cwd) + { + Ok(binding) => Some(binding), + Err(error) => { + return WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + error.to_string(), + ); + } + } + } + None if request.request.working_directory_request.is_some() => { + return WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + "persisted worker requested a working directory, but no persisted working directory binding is available to restore", + ); + } + None if request.request.working_directory.is_some() => { + let Some(materializer) = self.working_directory_materializer.as_ref() else { + return WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + "persisted worker has a working directory claim, but no materializer is configured for this runtime backend", + ); + }; + let working_directory = + request.request.working_directory.as_ref().expect("checked"); + match materializer.bind_working_directory( + &working_directory.working_directory_id, + working_directory.relative_cwd.as_deref(), + ) { + Ok(binding) => Some(binding), + Err(error) => { + return WorkerRestoreDryCheck::invalid( + "restore_dry_check_failed", + error.to_string(), + ); + } + } + } + None => None, + }; + request.working_directory = working_directory; + let factory = self.factory.clone(); + self.run_on_adapter_runtime( + async move { Ok(factory.dry_restore_controller(request).await) }, + ) + .unwrap_or_else(|message| { + WorkerRestoreDryCheck::unavailable("restore_dry_check_unavailable", message) + }) + } + fn restore_worker( &self, mut request: WorkerExecutionRestoreRequest, @@ -1217,6 +1384,7 @@ mod tests { WorkingDirectoryRequest, }; use crate::execution::WorkerExecutionContext; + use crate::identity::WorkerRef; use crate::management::RuntimeOptions; use crate::observation::WorkerObservationCursor; use crate::working_directory::LocalGitWorktreeMaterializer; @@ -1262,6 +1430,19 @@ mod tests { } } + #[cfg(feature = "ws-server")] + fn test_execution_context(worker_ref: WorkerRef) -> WorkerExecutionContext { + WorkerExecutionContext::new( + worker_ref, + Arc::new(|_, _| panic!("unused test event sink")), + ) + } + + #[cfg(not(feature = "ws-server"))] + fn test_execution_context(worker_ref: WorkerRef) -> WorkerExecutionContext { + WorkerExecutionContext::new(worker_ref) + } + struct MockFactory { client: MockClient, runtime_base: PathBuf, @@ -1535,7 +1716,7 @@ mod tests { let request = WorkerExecutionSpawnRequest { worker_ref: worker_ref.clone(), request: create_request("1"), - context: WorkerExecutionContext::new(worker_ref, Arc::new(|_, _| panic!("unused"))), + context: test_execution_context(worker_ref), working_directory: None, config_bundle: None, }; diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 9987c24d..4fe4fe3d 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -2749,6 +2749,17 @@ fn embedded_worker_projection_diagnostics( } } + if let Some(restore_dry_check) = execution.restore_dry_check.as_ref() { + diagnostics.push(diagnostic( + "embedded_worker_execution_restore_dry_check", + DiagnosticSeverity::Error, + format!( + "Worker restore dry-test {}: {}", + restore_dry_check.code, restore_dry_check.message + ), + )); + } + diagnostics } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 7607c577..535e6200 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -6915,6 +6915,15 @@ mod tests { .cleanup_working_directory(working_directory_id) } + fn dry_restore_worker( + &self, + _request: worker_runtime::execution::WorkerExecutionRestoreDryRequest, + ) -> worker_runtime::execution::WorkerRestoreDryCheck { + worker_runtime::execution::WorkerRestoreDryCheck::valid( + "deterministic test backend dry restore ok", + ) + } + fn spawn_worker( &self, request: worker_runtime::execution::WorkerExecutionSpawnRequest,