From 7a1b5e97c14dae5c687ecfebb97ba7795b8f100d Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 28 Jul 2026 19:21:09 +0900 Subject: [PATCH] refactor: make runtime worker state authoritative --- crates/worker-runtime/src/catalog.rs | 11 +- crates/worker-runtime/src/execution.rs | 220 +----- crates/worker-runtime/src/fs_store.rs | 142 +--- crates/worker-runtime/src/http_server.rs | 66 +- crates/worker-runtime/src/main.rs | 3 +- crates/worker-runtime/src/runtime.rs | 792 +++++++------------- crates/worker-runtime/src/worker_backend.rs | 266 ++----- crates/worker/src/controller.rs | 35 +- crates/worker/src/runtime/dir.rs | 42 +- crates/workspace-server/src/hosts.rs | 199 ++--- crates/workspace-server/src/server.rs | 20 +- 11 files changed, 530 insertions(+), 1266 deletions(-) diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 868993a9..fa69b9da 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -1,4 +1,3 @@ -use crate::execution::WorkerExecutionStatus; use crate::identity::{WorkerId, WorkerRef}; use crate::interaction::WorkerInput; use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef}; @@ -211,14 +210,16 @@ pub struct CreateWorkerRequest { #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum WorkerStatus { + Idle, Running, + Paused, Stopped, Cancelled, } impl WorkerStatus { pub fn is_active(self) -> bool { - matches!(self, Self::Running) + matches!(self, Self::Idle | Self::Running | Self::Paused) } } @@ -228,7 +229,8 @@ pub struct WorkerSummary { pub worker_ref: WorkerRef, pub worker_id: WorkerId, pub status: WorkerStatus, - pub execution: WorkerExecutionStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory: Option, pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] pub display_name: Option, @@ -244,7 +246,8 @@ pub struct WorkerDetail { pub worker_ref: WorkerRef, pub worker_id: WorkerId, pub status: WorkerStatus, - pub execution: WorkerExecutionStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory: Option, pub profile: ProfileSelector, #[serde(default, skip_serializing_if = "Option::is_none")] pub display_name: Option, diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index d927a4de..6f5144a8 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -11,44 +11,6 @@ use serde::{Deserialize, Serialize}; use std::fmt; use std::sync::Arc; -/// Persisted execution lifecycle visible through Worker catalog/detail responses. -/// -/// This is intentionally a worker lifecycle projection, not a transport/backend -/// handle state. Runtime restart boundaries invalidate live handles, so a -/// persisted `alive` worker is restored on startup; restore deferral keeps the -/// worker `stopped`, while structural restore failure marks it `corrupted`. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum WorkerExecutionBackendKind { - /// Restoreable persisted state exists, but no live execution handle is attached. - #[default] - #[serde(alias = "unconnected", alias = "stale")] - Stopped, - /// A live execution handle is currently attached. Legacy `connected` maps here. - #[serde(alias = "connected")] - Alive, - /// Persisted execution state is structurally invalid and cannot be restored. - Corrupted, -} - -/// Durable, non-authority execution binding projection. -/// -/// This records only enough identity to restore through the same backend kind. -/// It is not a live handle and must not contain sockets, paths, credentials, or -/// provider-private authority. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct WorkerExecutionBindingIdentity { - pub backend_id: String, -} - -impl WorkerExecutionBindingIdentity { - pub fn from_handle(handle: &WorkerExecutionHandle) -> Self { - Self { - backend_id: handle.backend_id.clone(), - } - } -} - /// Current execution-side run state for a Worker. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -155,113 +117,6 @@ impl WorkerExecutionResult { } } -/// Execution status surfaced in Worker summary/detail responses. -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] -pub struct WorkerExecutionStatus { - pub backend: WorkerExecutionBackendKind, - pub run_state: WorkerExecutionRunState, - #[serde(default, skip_serializing_if = "Option::is_none")] - 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 { - pub fn stopped() -> Self { - Self::default() - } - - pub fn alive(run_state: WorkerExecutionRunState) -> Self { - Self { - backend: WorkerExecutionBackendKind::Alive, - 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 - } - - pub fn with_binding(mut self, binding: WorkerExecutionBindingIdentity) -> Self { - self.binding = Some(binding); - self - } - - pub fn with_working_directory(mut self, status: WorkingDirectoryStatus) -> Self { - self.working_directory = Some(status); - self - } - - pub fn with_result(mut self, result: WorkerExecutionResult) -> Self { - 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. /// /// The handle is a typed token for routing calls back into the same backend. It @@ -362,23 +217,13 @@ 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 { pub worker_ref: WorkerRef, pub request: crate::catalog::CreateWorkerRequest, pub context: WorkerExecutionContext, - pub previous_execution: WorkerExecutionStatus, + pub previous_working_directory: Option, pub working_directory: Option, pub config_bundle: Option, } @@ -414,16 +259,6 @@ 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, @@ -536,10 +371,6 @@ impl WorkerExecutionBackendRef { }) } - pub(crate) fn backend_id(&self) -> &str { - &self.backend_id - } - pub(crate) fn spawn_worker( &self, request: WorkerExecutionSpawnRequest, @@ -547,13 +378,6 @@ 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, @@ -635,45 +459,3 @@ impl fmt::Debug for WorkerExecutionBackendRef { .finish_non_exhaustive() } } - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - #[test] - fn execution_backend_kind_accepts_legacy_values() { - let connected: WorkerExecutionStatus = serde_json::from_value(json!({ - "backend": "connected", - "run_state": "idle", - "binding": { "backend_id": "worker-crate" } - })) - .unwrap(); - assert_eq!(connected.backend, WorkerExecutionBackendKind::Alive); - - let stale: WorkerExecutionStatus = serde_json::from_value(json!({ - "backend": "stale", - "run_state": "stopped", - "binding": { "backend_id": "worker-crate" } - })) - .unwrap(); - assert_eq!(stale.backend, WorkerExecutionBackendKind::Stopped); - - let unconnected: WorkerExecutionStatus = serde_json::from_value(json!({ - "backend": "unconnected", - "run_state": "stopped" - })) - .unwrap(); - assert_eq!(unconnected.backend, WorkerExecutionBackendKind::Stopped); - } - - #[test] - fn execution_status_serializes_without_last_result() { - let status = WorkerExecutionStatus::alive(WorkerExecutionRunState::Idle).with_result( - WorkerExecutionResult::rejected(WorkerExecutionOperation::Input, "transient"), - ); - let serialized = serde_json::to_value(status).unwrap(); - assert_eq!(serialized["backend"], "alive"); - assert!(serialized.get("last_result").is_none()); - } -} diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 745e196e..adc036da 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -1,12 +1,9 @@ -use crate::catalog::{CreateWorkerRequest, WorkerStatus}; +use crate::catalog::{CreateWorkerRequest, WorkingDirectoryStatus}; use crate::config_bundle::ConfigBundle; use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::error::RuntimeError; -use crate::execution::WorkerExecutionStatus; use crate::identity::{WorkerId, WorkerRef}; use crate::management::{RuntimeBackendKind, RuntimeLimits, RuntimeStatus}; -#[cfg(feature = "ws-server")] -use crate::observation::WorkerObservationEvent; use crate::observation::{EventCursor, RuntimeEvent, RuntimeEventBatch}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; @@ -21,8 +18,7 @@ const EVENTS_FILE: &str = "events.jsonl"; const WORKERS_DIR: &str = "workers"; const LEGACY_RUNTIMES_DIR: &str = "runtimes"; const WORKER_FILE: &str = "worker.json"; -#[cfg(feature = "ws-server")] -const OBSERVATIONS_FILE: &str = "observations.jsonl"; +const LEGACY_OBSERVATIONS_FILE: &str = "observations.jsonl"; static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1); @@ -153,11 +149,7 @@ impl FsRuntimeStore { &WorkerSnapshot::from_persisted(worker), "write worker snapshot", )?; - #[cfg(feature = "ws-server")] - ensure_file_exists( - &worker_dir.join(OBSERVATIONS_FILE), - "create observations log", - )?; + remove_legacy_observations(&worker_dir); Ok(()) } @@ -180,19 +172,6 @@ impl FsRuntimeStore { append_json_line(&self.events_path(), event, "append event") } - #[cfg(feature = "ws-server")] - pub(crate) fn append_worker_observation_event( - &self, - event: &WorkerObservationEvent, - ) -> Result<(), RuntimeError> { - self.ensure_worker_ref(&event.worker_ref)?; - append_json_line( - &self.observations_path(&event.worker_ref.worker_id), - event, - "append worker observation", - ) - } - pub(crate) fn load_runtime_state(&self) -> Result { let runtime_path = self.runtime_path(); let mut snapshot: RuntimeSnapshot = read_json(&runtime_path, "read runtime snapshot")?; @@ -229,9 +208,6 @@ impl FsRuntimeStore { })?; worker_dirs.sort_by_key(|entry| entry.path()); - #[cfg(feature = "ws-server")] - let mut observation_events = Vec::new(); - for entry in worker_dirs { let path = entry.path(); if !path.is_dir() { @@ -263,43 +239,7 @@ impl FsRuntimeStore { ); continue; } - #[cfg(feature = "ws-server")] - let worker_observations = match read_json_lines::( - &path.join(OBSERVATIONS_FILE), - "read worker observations", - ) { - Ok(events) => events, - Err(_error) => { - record_worker_load_diagnostic( - &mut snapshot, - Some(worker_snapshot.worker_ref.clone()), - "ignored worker with unreadable observations while loading runtime store", - ); - continue; - } - }; - #[cfg(feature = "ws-server")] - let mut observations_valid = true; - #[cfg(feature = "ws-server")] - for event in &worker_observations { - if self.ensure_worker_ref(&event.worker_ref).is_err() - || event.worker_ref.worker_id != worker_snapshot.worker_id - { - observations_valid = false; - break; - } - } - #[cfg(feature = "ws-server")] - if !observations_valid { - record_worker_load_diagnostic( - &mut snapshot, - Some(worker_snapshot.worker_ref.clone()), - "ignored worker with invalid observations while loading runtime store", - ); - continue; - } - #[cfg(feature = "ws-server")] - observation_events.extend(worker_observations); + remove_legacy_observations(&path); let worker = worker_snapshot.into_persisted(); if workers.insert(worker.worker_id.clone(), worker).is_some() { record_worker_load_diagnostic( @@ -310,15 +250,7 @@ impl FsRuntimeStore { } } - #[cfg(feature = "ws-server")] - observation_events.sort_by_key(|event| event.sequence); - - Ok(snapshot.into_persisted( - events, - workers, - #[cfg(feature = "ws-server")] - observation_events, - )) + Ok(snapshot.into_persisted(events, workers)) } fn ensure_worker_ref(&self, _worker_ref: &WorkerRef) -> Result<(), RuntimeError> { @@ -336,10 +268,12 @@ impl FsRuntimeStore { fn worker_dir(&self, worker_id: &WorkerId) -> PathBuf { self.root.join(WORKERS_DIR).join(worker_id.to_string()) } +} - #[cfg(feature = "ws-server")] - fn observations_path(&self, worker_id: &WorkerId) -> PathBuf { - self.worker_dir(worker_id).join(OBSERVATIONS_FILE) +fn remove_legacy_observations(worker_dir: &Path) { + let path = worker_dir.join(LEGACY_OBSERVATIONS_FILE); + if path.is_file() { + let _ = fs::remove_file(path); } } @@ -360,8 +294,6 @@ pub(crate) struct PersistedRuntimeState { pub(crate) workers: BTreeMap, pub(crate) config_bundles: BTreeMap, pub(crate) events: Vec, - #[cfg(feature = "ws-server")] - pub(crate) observation_events: Vec, pub(crate) diagnostics: Vec, } @@ -369,9 +301,8 @@ pub(crate) struct PersistedRuntimeState { pub(crate) struct PersistedWorkerRecord { pub(crate) worker_ref: WorkerRef, pub(crate) worker_id: WorkerId, - pub(crate) status: WorkerStatus, pub(crate) request: CreateWorkerRequest, - pub(crate) execution: WorkerExecutionStatus, + pub(crate) working_directory: Option, pub(crate) last_event_id: u64, } @@ -447,7 +378,6 @@ impl RuntimeSnapshot { self, events: Vec, workers: BTreeMap, - #[cfg(feature = "ws-server")] observation_events: Vec, ) -> PersistedRuntimeState { PersistedRuntimeState { display_name: self.display_name, @@ -459,8 +389,6 @@ impl RuntimeSnapshot { workers, config_bundles: self.config_bundles, events, - #[cfg(feature = "ws-server")] - observation_events, diagnostics: self.diagnostics, } } @@ -471,22 +399,31 @@ struct WorkerSnapshot { schema_version: u32, worker_ref: WorkerRef, worker_id: WorkerId, - status: WorkerStatus, request: CreateWorkerRequest, - #[serde(default = "WorkerExecutionStatus::stopped")] - execution: WorkerExecutionStatus, + #[serde(default, skip_serializing_if = "Option::is_none")] + working_directory: Option, + /// One-way migration input for schema-v1 snapshots. New snapshots never + /// write the removed execution projection. + #[serde(default, rename = "execution", skip_serializing)] + legacy_execution: Option, last_event_id: u64, } +#[derive(Clone, Debug, Deserialize)] +struct LegacyWorkerExecutionProjection { + #[serde(default)] + working_directory: Option, +} + impl WorkerSnapshot { fn from_persisted(worker: &PersistedWorkerRecord) -> Self { Self { schema_version: SCHEMA_VERSION, worker_ref: worker.worker_ref.clone(), worker_id: worker.worker_id.clone(), - status: worker.status, request: worker.request.clone(), - execution: worker.execution.clone(), + working_directory: worker.working_directory.clone(), + legacy_execution: None, last_event_id: worker.last_event_id, } } @@ -519,9 +456,11 @@ impl WorkerSnapshot { PersistedWorkerRecord { worker_ref: self.worker_ref, worker_id: self.worker_id, - status: self.status, request: self.request, - execution: self.execution, + working_directory: self.working_directory.or_else(|| { + self.legacy_execution + .and_then(|execution| execution.working_directory) + }), last_event_id: self.last_event_id, } } @@ -759,29 +698,6 @@ where }) } -fn ensure_file_exists(path: &Path, operation: &'static str) -> Result<(), RuntimeError> { - let parent = path.parent().ok_or_else(|| RuntimeError::StoreCorrupt { - operation, - path: path.to_path_buf(), - message: "path has no parent directory".to_string(), - })?; - fs::create_dir_all(parent).map_err(|source| RuntimeError::StoreIo { - operation, - path: parent.to_path_buf(), - source, - })?; - OpenOptions::new() - .create(true) - .append(true) - .open(path) - .and_then(|file| file.sync_all()) - .map_err(|source| RuntimeError::StoreIo { - operation, - path: path.to_path_buf(), - source, - }) -} - fn tmp_path_for(path: &Path) -> PathBuf { let sequence = NEXT_TMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); let file_name = path diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 4ba47685..2c8a0797 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -179,6 +179,7 @@ pub fn runtime_http_router_with_auth( get(get_worker).delete(delete_worker), ) .route("/v1/workers/{worker_id}/input", post(send_worker_input)) + .route("/v1/workers/{worker_id}/restore", post(restore_worker)) .route( "/v1/workers/{worker_id}/completions", post(worker_completions), @@ -467,6 +468,18 @@ async fn create_worker( Ok(Json(RuntimeHttpWorkerResponse { worker })) } +async fn restore_worker( + State(state): State, + Path(worker_id): Path, +) -> RestResult { + let worker_ref = worker_ref_for(&state.runtime, worker_id)?; + let worker = state + .runtime + .restore_worker(&worker_ref) + .map_err(RuntimeHttpRestError::runtime)?; + Ok(Json(RuntimeHttpWorkerResponse { worker })) +} + #[cfg(feature = "ws-server")] async fn worker_protocol_ws( State(state): State, @@ -512,6 +525,22 @@ async fn worker_protocol_ws_session( } }, }; + // Observation cursors are process-local. After a Runtime restart (or + // bounded backlog expiry), the current Worker snapshot is authoritative + // and replay resumes from the current in-memory tail. + if runtime + .read_worker_observation_events(&worker_ref, cursor) + .is_err() + { + cursor = match runtime.worker_observation_cursor_now(&worker_ref) { + Ok(cursor) => cursor, + Err(error) => { + let event = protocol_error_event(error.to_string()); + let _ = send_protocol_event(&mut socket, &event).await; + return; + } + }; + } let mut receiver = match runtime.subscribe_worker_observation() { Ok(receiver) => receiver, @@ -806,7 +835,7 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s if path.starts_with("/v1/config-bundles") || path.starts_with("/v1/working-directories") { return Some("workers:create"); } - if path.ends_with("/input") { + if path.ends_with("/input") || path.ends_with("/restore") { return Some("workers:input"); } if path.ends_with("/stop") || path.ends_with("/cancel") { @@ -948,14 +977,14 @@ pub enum RuntimeHttpServerError { #[cfg(test)] mod tests { use super::*; - use crate::catalog::{ConfigBundleRef, ProfileSelector}; + use crate::catalog::{ConfigBundleRef, ProfileSelector, WorkerStatus}; use crate::config_bundle::{ ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor, }; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, - WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, + WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionRunState, + WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; use crate::management::RuntimeOptions; use axum::body::to_bytes; @@ -1038,6 +1067,17 @@ mod tests { } } + fn restore_worker( + &self, + request: WorkerExecutionRestoreRequest, + ) -> WorkerExecutionSpawnResult { + WorkerExecutionSpawnResult::Connected { + handle: WorkerExecutionHandle::new(request.worker_ref, self.backend_id()), + run_state: WorkerExecutionRunState::Idle, + working_directory: request.previous_working_directory, + } + } + fn dispatch_input( &self, _handle: &WorkerExecutionHandle, @@ -1156,6 +1196,24 @@ mod tests { let stop: RuntimeHttpWorkerLifecycleResponse = read_json(response).await; assert_eq!(stop.ack.worker_ref, created.worker.worker_ref); + let response = empty_request( + app.clone(), + Method::POST, + &format!("/v1/workers/{}/restore", created.worker.worker_id), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let restored: RuntimeHttpWorkerResponse = read_json(response).await; + assert_eq!(restored.worker.status, WorkerStatus::Idle); + + let response = empty_request( + app.clone(), + Method::POST, + &format!("/v1/workers/{}/stop", created.worker.worker_id), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let response = empty_request( app.clone(), Method::POST, diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index ebf458f4..8ab36bc2 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -82,8 +82,7 @@ fn build_runtime(config: &ProcessConfig) -> Result { let fs_paths = config.resolved_fs_paths(); let mut factory = ProfileRuntimeWorkerFactory::new(fs_paths.worker_dir.join("worker-root")) .with_store_dir(fs_paths.worker_dir.join("sessions")) - .with_worker_metadata_dir(fs_paths.worker_dir.join("metadata")) - .with_runtime_base_dir(fs_paths.worker_dir.join("runtime")); + .with_worker_metadata_dir(fs_paths.worker_dir.join("metadata")); if let Some(endpoint) = config.backend_resource_endpoint.clone() { factory = factory.with_resource_client(Arc::new( worker_runtime::resource::HttpBackendResourceClient::new( diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 9d192eb2..ca3f4642 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -11,13 +11,12 @@ use crate::config_bundle::{ use crate::diagnostics::DiagnosticSeverity; use crate::diagnostics::RuntimeDiagnostic; use crate::error::RuntimeError; +use crate::execution::WorkerExecutionRestoreRequest; use crate::execution::{ - WorkerExecutionBackend, WorkerExecutionBackendKind, WorkerExecutionBackendRef, - WorkerExecutionBindingIdentity, WorkerExecutionHandle, WorkerExecutionOperation, - WorkerExecutionResult, WorkerExecutionRunState, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, WorkerExecutionStatus, WorkerRestoreDryCheckStatus, + WorkerExecutionBackend, WorkerExecutionBackendRef, WorkerExecutionHandle, + WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionRunState, + WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; -use crate::execution::{WorkerExecutionRestoreDryRequest, WorkerExecutionRestoreRequest}; #[cfg(feature = "fs-store")] use crate::fs_store::{ FsRuntimeStore, FsRuntimeStoreOptions, PersistedRuntimeState, PersistedWorkerRecord, @@ -131,7 +130,9 @@ impl Runtime { let mut cancelled_worker_count = 0; for worker in state.workers.values() { match worker.status { - WorkerStatus::Running => active_worker_count += 1, + WorkerStatus::Idle | WorkerStatus::Running | WorkerStatus::Paused => { + active_worker_count += 1; + } WorkerStatus::Stopped => stopped_worker_count += 1, WorkerStatus::Cancelled => cancelled_worker_count += 1, } @@ -354,9 +355,9 @@ impl Runtime { let record = WorkerRecord { worker_ref: worker_ref.clone(), worker_id: worker_id.clone(), - status: WorkerStatus::Running, + status: WorkerStatus::Stopped, request: request.clone(), - execution: WorkerExecutionStatus::stopped(), + working_directory: None, execution_handle: None, last_event_id: event_id, }; @@ -433,16 +434,7 @@ impl Runtime { /// List Workers known to this Runtime. pub fn list_workers(&self) -> Result, RuntimeError> { let state = self.lock()?; - 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) + Ok(state.workers.values().map(WorkerRecord::summary).collect()) } /// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime. @@ -452,16 +444,105 @@ impl Runtime { Ok(worker.detail()) } + /// Attach a live execution to a persisted Worker definition. + /// + /// Current liveness is never read from disk. If a handle is already + /// present this is idempotent; otherwise the configured backend is tried. + pub fn restore_worker(&self, worker_ref: &WorkerRef) -> Result { + let (backend, request) = { + let state = self.lock()?; + state.ensure_running()?; + let worker = state.worker(worker_ref)?; + if worker.execution_handle.is_some() { + return Ok(worker.detail()); + } + if worker.status == WorkerStatus::Cancelled { + return Err(RuntimeError::InvalidRequest(format!( + "worker {} is cancelled", + worker_ref.worker_id + ))); + } + let backend = state.execution_backend.clone().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: worker_ref.worker_id.clone(), + message: "runtime has no execution backend".to_string(), + } + })?; + let config_bundle = worker + .request + .config_bundle + .as_ref() + .and_then(|bundle_ref| state.config_bundles.get(&bundle_ref.id)) + .cloned(); + let request = WorkerExecutionRestoreRequest { + worker_ref: worker_ref.clone(), + request: worker.request.clone(), + context: self.execution_context(worker_ref.clone()), + previous_working_directory: worker.working_directory.clone(), + working_directory: None, + config_bundle, + }; + (backend, request) + }; + + match backend.restore_worker(request) { + WorkerExecutionSpawnResult::Connected { + handle, + run_state, + working_directory, + } => { + self.commit_restored_worker_execution( + worker_ref, + handle, + run_state, + working_directory, + )?; + self.worker_detail(worker_ref) + } + WorkerExecutionSpawnResult::Rejected(result) + | WorkerExecutionSpawnResult::Errored(result) => { + #[cfg(feature = "fs-store")] + { + self.lock()? + .record_restore_failure(worker_ref, result.clone())?; + } + Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id.clone(), + operation: result.operation, + outcome: result.outcome, + message: result.message_or_default(), + result, + }) + } + } + } + + fn ensure_worker_execution(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { + let has_handle = { + let state = self.lock()?; + state + .worker(worker_ref)? + .execution_handle + .as_ref() + .is_some() + }; + if has_handle { + return Ok(()); + } + self.restore_worker(worker_ref).map(|_| ()) + } + /// Accept input into a Worker. pub fn send_input( &self, worker_ref: &WorkerRef, input: WorkerInput, ) -> Result { + validate_worker_input(&input)?; + self.ensure_worker_execution(worker_ref)?; let (backend, handle) = { - let mut state = self.lock()?; + let state = self.lock()?; state.ensure_running()?; - validate_worker_input(&input)?; state.ensure_worker_ref(worker_ref)?; let worker = state.worker(worker_ref)?; if !worker.status.is_active() { @@ -475,17 +556,9 @@ impl Runtime { match (backend, handle) { (Some(backend), Some(handle)) => (backend, handle), _ => { - let worker = state.worker_mut(worker_ref)?; - worker.execution = - if worker.execution.backend == WorkerExecutionBackendKind::Corrupted { - worker.execution.clone() - } else { - WorkerExecutionStatus::stopped_from(worker.execution.clone()) - }; - state.persist_worker(&worker_ref.worker_id)?; return Err(RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id.clone(), - message: "worker has no execution backend".to_string(), + message: "worker has no live execution handle".to_string(), }); } } @@ -512,27 +585,16 @@ impl Runtime { ); let worker = state.worker_mut(worker_ref)?; worker.last_event_id = event_id; - worker.execution = WorkerExecutionStatus { - backend: WorkerExecutionBackendKind::Alive, - run_state: dispatch_result.run_state, - binding: worker.execution.binding.clone(), - working_directory: worker.execution.working_directory.clone(), - restore_dry_check: None, - }; - + worker.status = worker_status_from_run_state(dispatch_result.run_state); let status = worker.status; #[cfg(feature = "ws-server")] - let observation = { + { let payload = input_protocol_event(&input); - Some(state.push_worker_observation_event(worker_ref.clone(), payload)) - }; + state.push_worker_observation_event(worker_ref.clone(), payload); + } state.persist_runtime_snapshot()?; state.persist_worker(&worker_ref.worker_id)?; state.persist_event_by_id(event_id)?; - #[cfg(feature = "ws-server")] - if let Some(observation) = observation.as_ref() { - state.persist_worker_observation_event(observation)?; - } Ok(WorkerInteractionAck { worker_ref: worker_ref.clone(), @@ -578,9 +640,14 @@ impl Runtime { let entries = self.worker_completions(worker_ref, kind, &prefix)?; return Ok(vec![Event::Completions { kind, entries }]); } + if matches!(&method, Method::Shutdown) { + self.stop_worker(worker_ref, Some("worker protocol shutdown".to_string()))?; + return Ok(Vec::new()); + } + self.ensure_worker_execution(worker_ref)?; let (backend, handle) = { - let mut state = self.lock()?; + let state = self.lock()?; state.ensure_running()?; state.ensure_worker_ref(worker_ref)?; let worker = state.worker(worker_ref)?; @@ -595,17 +662,9 @@ impl Runtime { match (backend, handle) { (Some(backend), Some(handle)) => (backend, handle), _ => { - let worker = state.worker_mut(worker_ref)?; - worker.execution = - if worker.execution.backend == WorkerExecutionBackendKind::Corrupted { - worker.execution.clone() - } else { - WorkerExecutionStatus::stopped_from(worker.execution.clone()) - }; - state.persist_worker(&worker_ref.worker_id)?; return Err(RuntimeError::WorkerExecutionUnavailable { worker_id: worker_ref.worker_id.clone(), - message: "worker has no execution backend".to_string(), + message: "worker has no live execution handle".to_string(), }); } } @@ -633,18 +692,14 @@ impl Runtime { handle: WorkerExecutionHandle, run_state: WorkerExecutionRunState, working_directory: Option, - result: WorkerExecutionResult, + _result: WorkerExecutionResult, ) -> Result { let mut state = self.lock()?; let detail = { - let binding = WorkerExecutionBindingIdentity::from_handle(&handle); let worker = state.worker_mut(worker_ref)?; worker.execution_handle = Some(handle); - let mut execution = WorkerExecutionStatus::alive(run_state).with_binding(binding); - if let Some(status) = working_directory { - execution = execution.with_working_directory(status); - } - worker.execution = execution.with_result(result); + worker.status = worker_status_from_run_state(run_state); + worker.working_directory = working_directory; worker.detail() }; state.persist_runtime_snapshot()?; @@ -670,14 +725,9 @@ impl Runtime { ) -> Result<(), RuntimeError> { let mut state = self.lock()?; let worker = state.worker_mut(worker_ref)?; - worker.execution = WorkerExecutionStatus { - backend: WorkerExecutionBackendKind::Alive, - 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)?; + if result.is_accepted() { + worker.status = worker_status_from_run_state(result.run_state); + } Ok(()) } @@ -713,10 +763,8 @@ impl Runtime { | WorkerExecutionOperation::ProtocolMethod => return Ok(()), }; if result.is_accepted() { - self.record_execution_result(worker_ref, result)?; return Ok(()); } - self.record_execution_result(worker_ref, result.clone())?; Err(RuntimeError::WorkerExecutionRejected { worker_id: worker_ref.worker_id.clone(), operation: result.operation, @@ -939,13 +987,8 @@ impl Runtime { ) -> Result { let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; - let execution_state_changed = - state.project_protocol_event_to_execution(worker_ref, &payload); + state.project_protocol_event_to_status(worker_ref, &payload); let event = state.push_worker_observation_event(worker_ref.clone(), payload); - if execution_state_changed { - state.persist_worker(&worker_ref.worker_id)?; - } - state.persist_worker_observation_event(&event)?; Ok(event) } @@ -962,9 +1005,7 @@ impl Runtime { ) -> Result<(), RuntimeError> { let mut state = self.lock()?; state.ensure_worker_ref(worker_ref)?; - let event = - state.push_worker_observation_event(worker_ref.clone(), input_protocol_event(&input)); - state.persist_worker_observation_event(&event)?; + state.push_worker_observation_event(worker_ref.clone(), input_protocol_event(&input)); Ok(()) } @@ -1002,6 +1043,7 @@ impl Runtime { let event_id = state.push_event(Some(worker_ref.clone()), event_kind, reason); let worker = state.worker_mut(worker_ref)?; worker.status = status; + worker.execution_handle = None; worker.last_event_id = event_id; let status = worker.status; state.persist_runtime_snapshot()?; @@ -1044,88 +1086,34 @@ impl Runtime { struct RestoreCandidate { worker_ref: WorkerRef, request: CreateWorkerRequest, - previous_execution: WorkerExecutionStatus, + previous_working_directory: Option, config_bundle: Option, } let candidates = { - let mut state = self.lock()?; - let Some(backend) = state.execution_backend.clone() else { - let worker_ids: Vec<_> = state.workers.keys().cloned().collect(); - for worker_id in worker_ids { - let Some(worker) = state.workers.get(&worker_id) else { - continue; - }; - if worker.execution.backend != WorkerExecutionBackendKind::Alive { - continue; - } - let worker_ref = worker.worker_ref.clone(); - let worker = state.worker_mut(&worker_ref)?; - worker.execution = - if worker.execution.backend == WorkerExecutionBackendKind::Corrupted { - worker.execution.clone() - } else { - WorkerExecutionStatus::stopped_from(worker.execution.clone()) - }; - state.persist_worker(&worker_ref.worker_id)?; - } + let state = self.lock()?; + if state.execution_backend.is_none() { return Ok(()); - }; - let backend_id = backend.backend_id().to_string(); - let mut candidates = Vec::new(); - let worker_ids: Vec<_> = state.workers.keys().cloned().collect(); - for worker_id in worker_ids { - let Some(worker) = state.workers.get(&worker_id) else { - continue; - }; - if !worker.status.is_active() - || worker.execution_handle.is_some() - || !matches!( - worker.execution.backend, - WorkerExecutionBackendKind::Alive | WorkerExecutionBackendKind::Stopped - ) - || worker - .execution - .binding - .as_ref() - .is_none_or(|binding| binding.backend_id != backend_id) - { - continue; - } - if let Some(working_directory) = worker.execution.working_directory.as_ref() { - if let Some(owner) = state.active_primary_worker_id_for_workdir_excluding( - &working_directory.summary.working_directory_id, - &worker.worker_id, - ) { - let worker_ref = worker.worker_ref.clone(); - let message = format!( - "worker {} cannot restore working directory {} because active worker {} is already the primary assignment", - worker.worker_id, working_directory.summary.working_directory_id, owner - ); - state.record_restore_failure( - &worker_ref, - WorkerExecutionResult::rejected( - WorkerExecutionOperation::Restore, - message, - ), - )?; - 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 + state + .workers + .values() + .filter(|worker| worker.execution_handle.is_none()) + .map(|worker| { + let config_bundle = worker + .request + .config_bundle + .as_ref() + .and_then(|bundle_ref| state.config_bundles.get(&bundle_ref.id)) + .cloned(); + RestoreCandidate { + worker_ref: worker.worker_ref.clone(), + request: worker.request.clone(), + previous_working_directory: worker.working_directory.clone(), + config_bundle, + } + }) + .collect::>() }; for candidate in candidates { @@ -1136,43 +1124,13 @@ 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, context: self.execution_context(candidate.worker_ref.clone()), - previous_execution: candidate.previous_execution, + previous_working_directory: candidate.previous_working_directory, working_directory: None, - config_bundle: None, + config_bundle: candidate.config_bundle, }; match backend.restore_worker(request) { WorkerExecutionSpawnResult::Connected { @@ -1195,7 +1153,6 @@ impl Runtime { Ok(()) } - #[cfg(feature = "fs-store")] fn commit_restored_worker_execution( &self, worker_ref: &WorkerRef, @@ -1212,13 +1169,9 @@ impl Runtime { ); { let worker = state.worker_mut(worker_ref)?; - worker.execution_handle = Some(handle.clone()); - let mut execution = WorkerExecutionStatus::alive(run_state) - .with_binding(WorkerExecutionBindingIdentity::from_handle(&handle)); - if let Some(status) = working_directory { - execution = execution.with_working_directory(status); - } - worker.execution = execution; + worker.execution_handle = Some(handle); + worker.status = worker_status_from_run_state(run_state); + worker.working_directory = working_directory; worker.last_event_id = event_id; } state.persist_runtime_snapshot()?; @@ -1327,52 +1280,23 @@ impl RuntimeState { store: FsRuntimeStore, ) -> Result { let mut workers = BTreeMap::new(); - let mut diagnostics = persisted.diagnostics; - let mut next_diagnostic_id = persisted.next_diagnostic_id; + let diagnostics = persisted.diagnostics; + let next_diagnostic_id = persisted.next_diagnostic_id; for (worker_id, worker) in persisted.workers { - let execution = if worker.execution.backend == WorkerExecutionBackendKind::Alive - && worker.execution.binding.is_none() - { - diagnostics.push(RuntimeDiagnostic { - id: next_diagnostic_id, - severity: DiagnosticSeverity::Error, - code: "worker_execution_binding_missing".to_string(), - message: format!( - "worker {} was persisted as alive but has no execution binding identity", - worker.worker_id - ), - worker_ref: Some(worker.worker_ref.clone()), - }); - next_diagnostic_id += 1; - WorkerExecutionStatus::corrupted(worker.execution) - } else { - worker.execution - }; workers.insert( worker_id, WorkerRecord { worker_ref: worker.worker_ref, worker_id: worker.worker_id, - status: worker.status, + status: WorkerStatus::Stopped, request: worker.request, - execution, + working_directory: worker.working_directory, execution_handle: None, last_event_id: worker.last_event_id, }, ); } - #[cfg(feature = "ws-server")] - let next_observation_sequence = persisted - .observation_events - .iter() - .map(|event| event.sequence) - .max() - .map(|sequence| sequence.saturating_add(1)) - .unwrap_or(1); - #[cfg(feature = "ws-server")] - let observation_events = persisted.observation_events.into_iter().collect(); - Ok(Self { display_name: persisted.display_name, backend: RuntimeBackendKind::FsStore, @@ -1388,9 +1312,9 @@ impl RuntimeState { events: persisted.events, diagnostics, #[cfg(feature = "ws-server")] - next_observation_sequence, + next_observation_sequence: 1, #[cfg(feature = "ws-server")] - observation_events, + observation_events: VecDeque::new(), #[cfg(feature = "ws-server")] observation_tx: broadcast::channel(256).0, }) @@ -1412,8 +1336,6 @@ impl RuntimeState { .collect(), config_bundles: self.config_bundles.clone(), events: self.events.clone(), - #[cfg(feature = "ws-server")] - observation_events: Vec::new(), diagnostics: self.diagnostics.clone(), } } @@ -1473,25 +1395,6 @@ impl RuntimeState { Ok(()) } - #[cfg(all(feature = "fs-store", feature = "ws-server"))] - fn persist_worker_observation_event( - &self, - event: &WorkerObservationEvent, - ) -> Result<(), RuntimeError> { - if let Some(store) = self.fs_store() { - store.append_worker_observation_event(event)?; - } - Ok(()) - } - - #[cfg(all(not(feature = "fs-store"), feature = "ws-server"))] - fn persist_worker_observation_event( - &self, - _event: &WorkerObservationEvent, - ) -> Result<(), RuntimeError> { - Ok(()) - } - #[cfg(feature = "fs-store")] fn persist_workers(&self) -> Result<(), RuntimeError> { if self.fs_store().is_some() { @@ -1595,31 +1498,6 @@ impl RuntimeState { fn primary_worker_id_for_workdir(&self, working_directory_id: &str) -> Option { self.workers.values().find_map(|worker| { if worker - .execution - .working_directory - .as_ref() - .is_some_and(|binding| binding.summary.working_directory_id == working_directory_id) - || requested_primary_workdir_id(&worker.request) == Some(working_directory_id) - { - Some(worker.worker_id) - } else { - None - } - }) - } - - #[cfg(feature = "fs-store")] - fn active_primary_worker_id_for_workdir_excluding( - &self, - working_directory_id: &str, - excluded_worker_id: &WorkerId, - ) -> Option { - self.workers.values().find_map(|worker| { - if worker.worker_id == *excluded_worker_id || !worker.status.is_active() { - return None; - } - if worker - .execution .working_directory .as_ref() .is_some_and(|binding| binding.summary.working_directory_id == working_directory_id) @@ -1656,36 +1534,8 @@ impl RuntimeState { }); let worker = self.worker_mut(worker_ref)?; worker.execution_handle = None; - worker.execution = WorkerExecutionStatus::corrupted(worker.execution.clone()); + worker.status = WorkerStatus::Stopped; self.persist_runtime_snapshot()?; - self.persist_worker(&worker_ref.worker_id)?; - 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(()) } @@ -1759,48 +1609,41 @@ impl RuntimeState { } #[cfg(feature = "ws-server")] - fn project_protocol_event_to_execution( + fn project_protocol_event_to_status( &mut self, worker_ref: &WorkerRef, event: &protocol::Event, - ) -> bool { + ) { let Some(worker) = self.workers.get_mut(&worker_ref.worker_id) else { - return false; + return; }; - let status = &mut worker.execution; - let next_run_state = match event { + let next_status = match event { protocol::Event::Status { status: protocol::WorkerStatus::Running, - } => Some(WorkerExecutionRunState::Busy), + } => Some(WorkerStatus::Running), protocol::Event::Status { status: protocol::WorkerStatus::Idle, - } => Some(WorkerExecutionRunState::Idle), + } => Some(WorkerStatus::Idle), protocol::Event::Status { status: protocol::WorkerStatus::Paused, - } => Some(WorkerExecutionRunState::Busy), + } => Some(WorkerStatus::Paused), protocol::Event::Snapshot { status, .. } => match status { - protocol::WorkerStatus::Running => Some(WorkerExecutionRunState::Busy), - protocol::WorkerStatus::Idle => Some(WorkerExecutionRunState::Idle), - protocol::WorkerStatus::Paused => Some(WorkerExecutionRunState::Busy), + protocol::WorkerStatus::Running => Some(WorkerStatus::Running), + protocol::WorkerStatus::Idle => Some(WorkerStatus::Idle), + protocol::WorkerStatus::Paused => Some(WorkerStatus::Paused), }, protocol::Event::RunEnd { result } => match result { protocol::RunResult::Finished | protocol::RunResult::RolledBack => { - Some(WorkerExecutionRunState::Idle) + Some(WorkerStatus::Idle) } - protocol::RunResult::Paused => Some(WorkerExecutionRunState::Busy), - protocol::RunResult::LimitReached => Some(WorkerExecutionRunState::Errored), + protocol::RunResult::Paused => Some(WorkerStatus::Paused), + protocol::RunResult::LimitReached => Some(WorkerStatus::Idle), }, - protocol::Event::Error { .. } => Some(WorkerExecutionRunState::Errored), _ => None, }; - let Some(next_run_state) = next_run_state else { - return false; - }; - if status.run_state == next_run_state { - return false; + if let Some(next_status) = next_status { + worker.status = next_status; } - status.run_state = next_run_state; - true } } @@ -1810,55 +1653,18 @@ struct WorkerRecord { worker_id: WorkerId, status: WorkerStatus, request: CreateWorkerRequest, - execution: WorkerExecutionStatus, + working_directory: Option, execution_handle: Option, 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_with_restore_dry_check( - &self, - restore_dry_check: Option, - ) -> WorkerSummary { + fn summary(&self) -> WorkerSummary { WorkerSummary { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, - execution: self - .execution - .clone() - .with_restore_dry_check(restore_dry_check), + working_directory: self.working_directory.clone(), profile: self.request.profile.clone(), display_name: self.request.display_name.clone(), profile_source: self.request.profile_source.reference(), @@ -1872,7 +1678,7 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, - execution: self.execution.clone(), + working_directory: self.working_directory.clone(), profile: self.request.profile.clone(), display_name: self.request.display_name.clone(), profile_source: self.request.profile_source.reference(), @@ -1886,14 +1692,23 @@ impl WorkerRecord { PersistedWorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id.clone(), - status: self.status, request: self.request.clone(), - execution: self.execution.clone(), + working_directory: self.working_directory.clone(), last_event_id: self.last_event_id, } } } +fn worker_status_from_run_state(run_state: WorkerExecutionRunState) -> WorkerStatus { + match run_state { + WorkerExecutionRunState::Idle => WorkerStatus::Idle, + WorkerExecutionRunState::Busy => WorkerStatus::Running, + WorkerExecutionRunState::Stopped + | WorkerExecutionRunState::Rejected + | WorkerExecutionRunState::Errored => WorkerStatus::Stopped, + } +} + fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> { request .working_directory @@ -1998,7 +1813,7 @@ mod tests { }; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle, - WorkerExecutionRestoreDryRequest, WorkerExecutionRestoreRequest, WorkerExecutionRunState, + WorkerExecutionRestoreRequest, WorkerExecutionRunState, }; use std::collections::BTreeMap; #[cfg(feature = "fs-store")] @@ -2072,8 +1887,6 @@ 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>, @@ -2084,18 +1897,6 @@ 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 @@ -2136,20 +1937,6 @@ 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, @@ -2249,7 +2036,7 @@ mod tests { let runtime = runtime_with_backend(); let detail = runtime.create_worker(task_request("implement v0")).unwrap(); - assert_eq!(detail.status, WorkerStatus::Running); + assert_eq!(detail.status, WorkerStatus::Idle); assert_eq!(detail.config_bundle.as_ref().unwrap().id, "bundle-1"); let list = runtime.list_workers().unwrap(); @@ -2262,35 +2049,6 @@ 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(); @@ -2415,7 +2173,7 @@ mod tests { } )); let refreshed = runtime.worker_detail(&detail.worker_ref).unwrap(); - assert_eq!(refreshed.execution.run_state, WorkerExecutionRunState::Busy); + assert_eq!(refreshed.status, WorkerStatus::Idle); #[cfg(feature = "ws-server")] assert!( runtime @@ -2541,7 +2299,53 @@ mod tests { )); assert_eq!( runtime.worker_detail(&detail.worker_ref).unwrap().status, - WorkerStatus::Running + WorkerStatus::Idle + ); + } + + #[test] + fn protocol_shutdown_transitions_worker_to_stopped() { + let runtime = runtime_with_backend(); + let detail = runtime + .create_worker(task_request("shutdown from client")) + .unwrap(); + + runtime + .send_protocol_method(&detail.worker_ref, Method::Shutdown) + .unwrap(); + + assert_eq!( + runtime.worker_detail(&detail.worker_ref).unwrap().status, + WorkerStatus::Stopped + ); + let events = runtime + .read_events(&runtime.event_cursor_from_start().unwrap(), 16) + .unwrap() + .events; + assert!(events.iter().any(|event| { + event.kind == RuntimeEventKind::WorkerStopped + && event.worker_ref.as_ref() == Some(&detail.worker_ref) + })); + } + + #[test] + fn input_restores_stopped_worker_without_persisted_connection_state() { + let (runtime, backend) = runtime_and_backend(); + let detail = runtime + .create_worker(task_request("restore on input")) + .unwrap(); + runtime + .send_protocol_method(&detail.worker_ref, Method::Shutdown) + .unwrap(); + + runtime + .send_input(&detail.worker_ref, WorkerInput::user("wake up")) + .unwrap(); + + assert_eq!(*backend.restore_count.lock().unwrap(), 1); + assert_eq!( + runtime.worker_detail(&detail.worker_ref).unwrap().status, + WorkerStatus::Idle ); } @@ -2765,7 +2569,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_restores_workers_events_and_protocol_observations() { + fn fs_store_restores_workers_and_events_but_not_protocol_observations() { let root = fs_store_root("restore"); let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { @@ -2794,6 +2598,17 @@ mod tests { runtime .stop_worker(&worker.worker_ref, Some("finished".to_string())) .unwrap(); + let worker_store_dir = root.join("workers").join(worker.worker_id.to_string()); + let worker_snapshot: serde_json::Value = + serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap()) + .unwrap(); + assert!(worker_snapshot.get("status").is_none()); + assert!(worker_snapshot.get("execution").is_none()); + std::fs::write( + worker_store_dir.join("observations.jsonl"), + b"{\"legacy\":true}\n", + ) + .unwrap(); let store = runtime_store(&runtime); drop(runtime); @@ -2805,32 +2620,13 @@ mod tests { .unwrap(); let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap(); assert_eq!(restored_worker.status, WorkerStatus::Stopped); - assert_eq!( - restored_worker.execution.backend, - WorkerExecutionBackendKind::Stopped - ); - assert_eq!( - restored_worker - .execution - .binding - .as_ref() - .map(|binding| binding.backend_id.as_str()), - Some("test-execution-backend") - ); + assert!(!worker_store_dir.join("observations.jsonl").exists()); #[cfg(feature = "ws-server")] { let observations = restored .read_worker_observation_events(&worker.worker_ref, WorkerObservationCursor::zero()) .unwrap(); - assert_eq!(observations.len(), 2); - assert!(matches!( - observations[0].payload, - protocol::Event::UserMessage { .. } - )); - assert!(matches!( - observations[1].payload, - protocol::Event::SystemItem { .. } - )); + assert!(observations.is_empty()); } let cursor = restored.event_cursor_from_start().unwrap(); @@ -2852,12 +2648,12 @@ mod tests { }, ) .unwrap(); - assert_eq!(observation.sequence, 3); + assert_eq!(observation.sequence, 1); let observations = restored .read_worker_observation_events(&worker.worker_ref, WorkerObservationCursor::zero()) .unwrap(); - assert_eq!(observations.len(), 3); - assert_eq!(observations[2].cursor, observation.cursor); + assert_eq!(observations.len(), 1); + assert_eq!(observations[0].cursor, observation.cursor); } let _ = std::fs::remove_dir_all(root); @@ -2889,10 +2685,7 @@ mod tests { }) .unwrap(); let stopped_worker = backendless.worker_detail(&worker.worker_ref).unwrap(); - assert_eq!( - stopped_worker.execution.backend, - WorkerExecutionBackendKind::Stopped - ); + assert_eq!(stopped_worker.status, WorkerStatus::Stopped); drop(backendless); let restoring_backend = Arc::new(TestExecutionBackend::default()); @@ -2908,12 +2701,7 @@ mod tests { assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1); let restored_worker = restored.worker_detail(&worker.worker_ref).unwrap(); - assert_eq!(restored_worker.status, WorkerStatus::Running); - assert_eq!( - restored_worker.execution.backend, - WorkerExecutionBackendKind::Alive - ); - assert!(restored_worker.execution.binding.is_some()); + assert_eq!(restored_worker.status, WorkerStatus::Idle); restored .send_input(&worker.worker_ref, WorkerInput::user("after restart")) .unwrap(); @@ -2935,64 +2723,7 @@ mod tests { #[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() { + fn fs_store_stops_worker_and_reports_when_execution_restore_fails() { let root = fs_store_root("execution-restore-failed"); let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { @@ -3026,11 +2757,7 @@ mod tests { assert_eq!(*restoring_backend.restore_count.lock().unwrap(), 1); 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_eq!(restored_worker.status, WorkerStatus::Stopped); assert!( restored .diagnostics() @@ -3047,10 +2774,7 @@ mod tests { WorkerInput::user("after failed restore"), ) .unwrap_err(); - assert!(matches!( - err, - RuntimeError::WorkerExecutionUnavailable { .. } - )); + assert!(matches!(err, RuntimeError::WorkerExecutionRejected { .. })); let _ = std::fs::remove_dir_all(root); } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index b70eda4c..89ee8a48 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -10,7 +10,7 @@ use std::collections::HashMap; use std::future::Future; use std::path::PathBuf; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; @@ -20,9 +20,8 @@ use crate::catalog::{ }; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, - WorkerExecutionRestoreDryRequest, WorkerExecutionRestoreRequest, WorkerExecutionResult, - WorkerExecutionRunState, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, - WorkerRestoreDryCheck, + WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionRunState, + WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; use crate::interaction::{WorkerInput, WorkerInputKind}; use crate::resource::{BackendResourceClient, ProfileSourceArchiveCache}; @@ -33,7 +32,7 @@ use async_trait::async_trait; use manifest::paths; use protocol::{Method, Segment, WorkerStatus}; use session_store::FsStore; -use session_store::{CombinedStore, FsWorkerStore, Store, WorkerMetadataStore}; +use session_store::{CombinedStore, FsWorkerStore}; use tokio::runtime::Runtime; #[cfg(feature = "ws-server")] use tokio::sync::broadcast; @@ -47,6 +46,42 @@ use worker::{ const DEFAULT_BACKEND_ID: &str = "worker-crate"; const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10); +static NEXT_RUNTIME_ARTIFACT_ROOT: AtomicU64 = AtomicU64::new(1); + +#[derive(Clone)] +enum RuntimeArtifactRoot { + Owned(Arc), + External(PathBuf), +} + +impl RuntimeArtifactRoot { + fn owned() -> Self { + let sequence = NEXT_RUNTIME_ARTIFACT_ROOT.fetch_add(1, Ordering::Relaxed); + Self::Owned(Arc::new(OwnedRuntimeArtifactRoot { + path: std::env::temp_dir().join(format!( + "yoi-worker-runtime-artifacts-{}-{sequence}", + std::process::id() + )), + })) + } + + fn path(&self) -> &std::path::Path { + match self { + Self::Owned(root) => &root.path, + Self::External(path) => path, + } + } +} + +struct OwnedRuntimeArtifactRoot { + path: PathBuf, +} + +impl Drop for OwnedRuntimeArtifactRoot { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.path); + } +} /// Factory seam used by [`WorkerRuntimeExecutionBackend`] to construct a real /// controller-backed Worker for a Runtime catalog entry. @@ -61,16 +96,6 @@ 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 @@ -81,7 +106,7 @@ pub struct ProfileRuntimeWorkerFactory { store_dir: Option, worker_metadata_dir: Option, profile: Option, - runtime_base_dir: Option, + runtime_base_dir: RuntimeArtifactRoot, resource_client: Option>, profile_archive_cache: Arc, } @@ -94,7 +119,7 @@ impl ProfileRuntimeWorkerFactory { store_dir: None, worker_metadata_dir: None, profile: None, - runtime_base_dir: None, + runtime_base_dir: RuntimeArtifactRoot::owned(), resource_client: None, profile_archive_cache: Arc::new(ProfileSourceArchiveCache::default()), } @@ -118,7 +143,7 @@ impl ProfileRuntimeWorkerFactory { } pub fn with_runtime_base_dir(mut self, runtime_base_dir: impl Into) -> Self { - self.runtime_base_dir = Some(runtime_base_dir.into()); + self.runtime_base_dir = RuntimeArtifactRoot::External(runtime_base_dir.into()); self } @@ -146,10 +171,7 @@ impl ProfileRuntimeWorkerFactory { } fn runtime_base_dir(&self) -> Result { - self.runtime_base_dir - .clone() - .or_else(|| worker::runtime::dir::default_base().ok()) - .ok_or_else(|| "could not resolve worker runtime directory".to_string()) + Ok(self.runtime_base_dir.path().to_path_buf()) } fn runtime_worker_name_for_ref(worker_ref: &crate::identity::WorkerRef) -> String { @@ -339,52 +361,6 @@ 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; - 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")); - } - Ok(()) - } -} - #[async_trait] impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { async fn spawn_controller( @@ -462,22 +438,12 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .map_err(|err| format!("failed to create Worker from profile: {err}"))?; let runtime_base = self.runtime_base_dir()?; - let (handle, _shutdown_rx) = WorkerController::spawn(worker, &runtime_base) + let (handle, _shutdown_rx) = WorkerController::spawn_runtime_managed(worker, &runtime_base) .await .map_err(|err| format!("failed to spawn Worker controller: {err}"))?; 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, @@ -557,7 +523,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { }; let runtime_base = self.runtime_base_dir()?; - let (handle, _shutdown_rx) = WorkerController::spawn(worker, &runtime_base) + let (handle, _shutdown_rx) = WorkerController::spawn_runtime_managed(worker, &runtime_base) .await .map_err(|err| format!("failed to spawn restored Worker controller: {err}"))?; Ok(handle) @@ -968,80 +934,11 @@ 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, ) -> WorkerExecutionSpawnResult { - let working_directory = match request.previous_execution.working_directory.clone() { + let working_directory = match request.previous_working_directory.clone() { Some(status) => { let Some(materializer) = self.working_directory_materializer.as_ref() else { return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected( @@ -1341,6 +1238,7 @@ mod tests { use llm_engine::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent}; use llm_engine::llm_client::{ClientError, LlmClient, Request}; use manifest::{Scope, WorkerManifest}; + use session_store::WorkerMetadataStore; #[derive(Clone)] struct MockClient { @@ -1462,9 +1360,10 @@ mod tests { ) .await .map_err(|err| err.to_string())?; - let (handle, _shutdown_rx) = WorkerController::spawn(worker, &self.runtime_base) - .await - .map_err(|err| err.to_string())?; + let (handle, _shutdown_rx) = + WorkerController::spawn_runtime_managed(worker, &self.runtime_base) + .await + .map_err(|err| err.to_string())?; Ok(handle) } async fn restore_controller( @@ -1689,60 +1588,6 @@ mod tests { .expect("embedded archive should resolve without Backend resource client"); } - #[tokio::test] - async fn dry_restore_validates_saved_worker_state_without_profile_source_resolution() { - let root = tempfile::tempdir().unwrap(); - let store_dir = root.path().join("sessions"); - let worker_metadata_dir = root.path().join("workers"); - let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(1)); - let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); - let session_id = session_store::new_session_id(); - let segment_id = session_store::new_segment_id(); - FsStore::new(&store_dir) - .unwrap() - .create_segment( - session_id, - segment_id, - &[session_store::LogEntry::Invoke { - ts: 1, - trigger: protocol::InvokeKind::UserSend, - }], - ) - .unwrap(); - FsWorkerStore::new(&worker_metadata_dir) - .unwrap() - .set_active( - &worker_name, - Some(session_store::WorkerActiveSegmentRef::active_segment( - session_id, segment_id, - )), - None, - ) - .unwrap(); - - let request = create_request("restore"); - let result = ProfileRuntimeWorkerFactory::new(root.path()) - .with_store_dir(&store_dir) - .with_worker_metadata_dir(&worker_metadata_dir) - .dry_restore_controller(WorkerExecutionRestoreDryRequest { - worker_ref, - request, - previous_execution: crate::execution::WorkerExecutionStatus::alive( - WorkerExecutionRunState::Idle, - ), - working_directory: None, - config_bundle: None, - }) - .await; - - assert_eq!( - result.status, - crate::execution::WorkerRestoreDryCheckStatus::Valid, - "{}", - result.message - ); - } - #[tokio::test] async fn restore_pending_worker_uses_saved_manifest_snapshot() { let root = tempfile::tempdir().unwrap(); @@ -1793,9 +1638,7 @@ mod tests { worker_ref: worker_ref.clone(), request, context: test_execution_context(worker_ref), - previous_execution: crate::execution::WorkerExecutionStatus::alive( - WorkerExecutionRunState::Idle, - ), + previous_working_directory: None, working_directory: None, config_bundle: None, }) @@ -1948,7 +1791,7 @@ mod tests { ); } - assert!(detail.execution.working_directory.is_some()); + assert!(detail.working_directory.is_some()); let cwds = observed_cwds.lock().unwrap(); assert_eq!(cwds.len(), 1); let cwd = &cwds[0]; @@ -1991,7 +1834,6 @@ mod tests { request.working_directory_request = Some(working_directory_request(repo.path())); let detail = runtime.create_worker(request).unwrap(); let workdir_id = detail - .execution .working_directory .as_ref() .unwrap() diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 3f18f901..d8bae4b0 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -193,8 +193,36 @@ pub struct WorkerController; impl WorkerController { pub async fn spawn( + worker: Worker, + runtime_base: &Path, + ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> + where + C: LlmClient + Clone + 'static, + St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, + { + Self::spawn_inner(worker, runtime_base, false).await + } + + /// Spawn a Worker owned by `worker-runtime`. + /// + /// The controller still uses an ephemeral directory for Unix sockets and + /// tool spill artifacts, but does not write legacy pid/status/manifest + /// liveness projections. + pub async fn spawn_runtime_managed( + worker: Worker, + runtime_base: &Path, + ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> + where + C: LlmClient + Clone + 'static, + St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, + { + Self::spawn_inner(worker, runtime_base, true).await + } + + async fn spawn_inner( mut worker: Worker, runtime_base: &Path, + runtime_managed: bool, ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> where C: LlmClient + Clone + 'static, @@ -214,8 +242,11 @@ impl WorkerController { // the spawn-tool factories need its socket path, and before the // initial status/history writes consume the greeting we build // after registration is complete. - let runtime_dir = - Arc::new(RuntimeDir::create(runtime_base, &worker.manifest().worker.name).await?); + let runtime_dir = Arc::new(if runtime_managed { + RuntimeDir::create_transient(runtime_base, &worker.manifest().worker.name).await? + } else { + RuntimeDir::create(runtime_base, &worker.manifest().worker.name).await? + }); let spawner_name = worker.manifest().worker.name.clone(); let self_parent_socket = worker.callback_socket().cloned(); diff --git a/crates/worker/src/runtime/dir.rs b/crates/worker/src/runtime/dir.rs index cafa45fe..ed41d0b4 100644 --- a/crates/worker/src/runtime/dir.rs +++ b/crates/worker/src/runtime/dir.rs @@ -41,6 +41,7 @@ pub struct SpawnedWorkerRecord { /// The directory is removed on drop. pub struct RuntimeDir { path: PathBuf, + write_legacy_snapshots: bool, } impl RuntimeDir { @@ -52,7 +53,23 @@ impl RuntimeDir { let pid = std::process::id().to_string(); fs::write(path.join("pid"), pid.as_bytes()).await?; - Ok(Self { path }) + Ok(Self { + path, + write_legacy_snapshots: true, + }) + } + + /// Create an ephemeral Runtime-owned artifact directory. + /// + /// Runtime-managed Workers keep their status in the owning Runtime and do + /// not materialize legacy pid/status/manifest projections. + pub async fn create_transient(base: &Path, worker_name: &str) -> Result { + let path = base.join(worker_name); + fs::create_dir_all(&path).await?; + Ok(Self { + path, + write_legacy_snapshots: false, + }) } /// Create in the default base directory resolved via @@ -64,12 +81,18 @@ impl RuntimeDir { /// Write status.json atomically. pub async fn write_status(&self, state: &WorkerSharedState) -> Result<(), io::Error> { + if !self.write_legacy_snapshots { + return Ok(()); + } let content = state.status_json(); atomic_write(&self.path.join("status.json"), content.as_bytes()).await } /// Write manifest.toml (typically once at startup). pub async fn write_manifest(&self, toml: &str) -> Result<(), io::Error> { + if !self.write_legacy_snapshots { + return Ok(()); + } atomic_write(&self.path.join("manifest.toml"), toml.as_bytes()).await } @@ -200,6 +223,23 @@ mod tests { assert_eq!(content, "[engine]\nname = \"test\""); } + #[tokio::test] + async fn transient_directory_does_not_write_liveness_snapshots() { + let tmp = tempfile::tempdir().unwrap(); + let rt = RuntimeDir::create_transient(tmp.path(), "runtime-worker") + .await + .unwrap(); + + rt.write_status(&test_state()).await.unwrap(); + rt.write_manifest("[worker]\nname = \"runtime-worker\"") + .await + .unwrap(); + + assert!(!rt.path().join("pid").exists()); + assert!(!rt.path().join("status.json").exists()); + assert!(!rt.path().join("manifest.toml").exists()); + } + #[tokio::test] async fn write_spawned_workers_creates_file() { use manifest::{Permission, ScopeRule}; diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index fad21e8e..0a7a7eee 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -26,9 +26,8 @@ use worker_runtime::config_bundle::{ ConfigBundleMetadata, ConfigBundleProvenance, ConfigProfileDescriptor, }; use worker_runtime::error::RuntimeError as EmbeddedRuntimeError; -use worker_runtime::execution::{ - WorkerExecutionBackendKind, WorkerExecutionRunState, WorkerExecutionStatus, -}; +#[cfg(test)] +use worker_runtime::execution::WorkerExecutionRunState; use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::http_server::{ RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, @@ -1332,12 +1331,8 @@ impl EmbeddedWorkerRuntime { Some(EmbeddedWorkerRef::new(EmbeddedWorkerId::parse(worker_id)?)) } - fn can_stop_embedded_worker( - &self, - status: EmbeddedWorkerStatus, - execution: &worker_runtime::execution::WorkerExecutionStatus, - ) -> bool { - runtime_worker_can_stop(self.execution_enabled, status, execution) + fn can_stop_embedded_worker(&self, status: EmbeddedWorkerStatus) -> bool { + runtime_worker_can_stop(self.execution_enabled, status) } fn map_worker_summary(&self, summary: worker_runtime::catalog::WorkerSummary) -> WorkerSummary { @@ -1363,8 +1358,7 @@ impl EmbeddedWorkerRuntime { visibility: "backend_internal".to_string(), identity: "runtime_registry_worker".to_string(), }, - state: embedded_worker_execution_status_label(summary.status, &summary.execution) - .to_string(), + state: embedded_worker_status_label(summary.status).to_string(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -1373,15 +1367,11 @@ impl EmbeddedWorkerRuntime { display_hint: "backend-internal worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: self.can_stop_embedded_worker(summary.status, &summary.execution), + can_stop: self.can_stop_embedded_worker(summary.status), can_spawn_followup: false, }, - working_directory: summary - .execution - .working_directory - .clone() - .map(|status| status.summary), - diagnostics: embedded_worker_projection_diagnostics(&summary.execution), + working_directory: summary.working_directory.map(|status| status.summary), + diagnostics: embedded_worker_projection_diagnostics(), } } @@ -1408,8 +1398,7 @@ impl EmbeddedWorkerRuntime { visibility: "backend_internal".to_string(), identity: "runtime_registry_worker".to_string(), }, - state: embedded_worker_execution_status_label(detail.status, &detail.execution) - .to_string(), + state: embedded_worker_status_label(detail.status).to_string(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -1418,15 +1407,11 @@ impl EmbeddedWorkerRuntime { display_hint: "backend-internal worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: self.can_stop_embedded_worker(detail.status, &detail.execution), + can_stop: self.can_stop_embedded_worker(detail.status), can_spawn_followup: false, }, - working_directory: detail - .execution - .working_directory - .clone() - .map(|status| status.summary), - diagnostics: embedded_worker_projection_diagnostics(&detail.execution), + working_directory: detail.working_directory.map(|status| status.summary), + diagnostics: embedded_worker_projection_diagnostics(), } } } @@ -2289,8 +2274,7 @@ impl RemoteWorkerRuntime { visibility: "remote_runtime".to_string(), identity: "runtime_registry_worker".to_string(), }, - state: embedded_worker_execution_status_label(summary.status, &summary.execution) - .to_string(), + state: embedded_worker_status_label(summary.status).to_string(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -2299,10 +2283,10 @@ impl RemoteWorkerRuntime { display_hint: "Backend-proxied remote worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: runtime_worker_can_stop(true, summary.status, &summary.execution), + can_stop: runtime_worker_can_stop(true, summary.status), can_spawn_followup: false, }, - working_directory: summary.execution.working_directory.clone().map(|status| status.summary), + working_directory: summary.working_directory.map(|status| status.summary), diagnostics: vec![diagnostic( "remote_runtime_projection", DiagnosticSeverity::Info, @@ -2334,8 +2318,7 @@ impl RemoteWorkerRuntime { visibility: "remote_runtime".to_string(), identity: "runtime_registry_worker".to_string(), }, - state: embedded_worker_execution_status_label(detail.status, &detail.execution) - .to_string(), + state: embedded_worker_status_label(detail.status).to_string(), last_seen_at: None, pinned: false, retention_state: "transient".to_string(), @@ -2344,10 +2327,10 @@ impl RemoteWorkerRuntime { display_hint: "Backend-proxied remote worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: runtime_worker_can_stop(true, detail.status, &detail.execution), + can_stop: runtime_worker_can_stop(true, detail.status), can_spawn_followup: false, }, - working_directory: detail.execution.working_directory.clone().map(|status| status.summary), + working_directory: detail.working_directory.map(|status| status.summary), diagnostics: vec![diagnostic( "remote_runtime_projection", DiagnosticSeverity::Info, @@ -2814,94 +2797,26 @@ fn embedded_runtime_status_label(status: RuntimeStatus) -> &'static str { } } -fn runtime_worker_can_stop( - execution_enabled: bool, - status: EmbeddedWorkerStatus, - execution: &WorkerExecutionStatus, -) -> bool { - execution_enabled - && status == EmbeddedWorkerStatus::Running - && execution.backend == worker_runtime::execution::WorkerExecutionBackendKind::Alive - && matches!( - execution.run_state, - WorkerExecutionRunState::Idle | WorkerExecutionRunState::Busy - ) +fn runtime_worker_can_stop(execution_enabled: bool, status: EmbeddedWorkerStatus) -> bool { + execution_enabled && status.is_active() } fn embedded_worker_status_label(status: EmbeddedWorkerStatus) -> &'static str { match status { + EmbeddedWorkerStatus::Idle => "idle", EmbeddedWorkerStatus::Running => "running", + EmbeddedWorkerStatus::Paused => "paused", EmbeddedWorkerStatus::Stopped => "stopped", EmbeddedWorkerStatus::Cancelled => "cancelled", } } -fn embedded_worker_projection_diagnostics( - execution: &WorkerExecutionStatus, -) -> Vec { - let mut diagnostics = vec![diagnostic( +fn embedded_worker_projection_diagnostics() -> Vec { + vec![diagnostic( "embedded_runtime_projection", DiagnosticSeverity::Info, "Worker identity is projected only as runtime_id plus worker_id; embedded runtime internals remain backend-private".to_string(), - )]; - - match execution.backend { - WorkerExecutionBackendKind::Stopped => diagnostics.push(diagnostic( - "embedded_worker_execution_stopped", - DiagnosticSeverity::Info, - "Worker execution is stopped; the runtime will restore it before dispatching input or protocol methods".to_string(), - )), - WorkerExecutionBackendKind::Corrupted => diagnostics.push(diagnostic( - "embedded_worker_execution_corrupted", - DiagnosticSeverity::Error, - "Worker execution state is corrupted and cannot be restored without repair".to_string(), - )), - WorkerExecutionBackendKind::Alive => { - if execution.run_state == WorkerExecutionRunState::Rejected { - diagnostics.push(diagnostic( - "embedded_worker_execution_rejected", - DiagnosticSeverity::Warning, - "Worker execution rejected the last transient operation; retry or inspect runtime logs".to_string(), - )); - } - } - } - - 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 -} - -fn embedded_worker_execution_status_label( - status: EmbeddedWorkerStatus, - execution: &WorkerExecutionStatus, -) -> &'static str { - match status { - EmbeddedWorkerStatus::Stopped => "stopped", - EmbeddedWorkerStatus::Cancelled => "cancelled", - EmbeddedWorkerStatus::Running => match execution.backend { - worker_runtime::execution::WorkerExecutionBackendKind::Stopped => "stopped", - worker_runtime::execution::WorkerExecutionBackendKind::Corrupted => "corrupted", - worker_runtime::execution::WorkerExecutionBackendKind::Alive => { - match execution.run_state { - WorkerExecutionRunState::Idle => "idle", - WorkerExecutionRunState::Busy => "running", - WorkerExecutionRunState::Stopped => "stopped", - WorkerExecutionRunState::Rejected => "rejected", - WorkerExecutionRunState::Errored => "errored", - } - } - }, - } + )] } fn default_profile_source_archive_source( @@ -4501,7 +4416,7 @@ mod tests { } #[test] - fn remote_runtime_projection_blocks_stopped_and_corrupted_execution_stop() { + fn remote_runtime_projection_uses_canonical_worker_status_for_stop_capability() { let (base_url, server) = serve_mock_http(vec![ mock_response( "GET", @@ -4510,30 +4425,10 @@ mod tests { 200, json!({ "workers": [ - worker_json_with_execution( - "remote:primary", - "1", - "stopped", - "stopped", - ), - worker_json_with_execution( - "remote:primary", - "2", - "corrupted", - "errored", - ), - worker_json_with_execution( - "remote:primary", - "3", - "alive", - "rejected", - ), - worker_json_with_execution( - "remote:primary", - "4", - "alive", - "errored", - ) + worker_json_with_status("remote:primary", "1", "stopped"), + worker_json_with_status("remote:primary", "2", "cancelled"), + worker_json_with_status("remote:primary", "3", "paused"), + worker_json_with_status("remote:primary", "4", "idle") ] }) .to_string(), @@ -4544,12 +4439,7 @@ mod tests { true, 200, json!({ - "worker": worker_json_with_execution( - "remote:primary", - "1", - "stopped", - "stopped", - )}) + "worker": worker_json_with_status("remote:primary", "1", "stopped")}) .to_string(), ), ]); @@ -4569,17 +4459,14 @@ mod tests { let workers = registry.list_workers(10); assert_eq!(workers.items.len(), 4); - for worker in &workers.items { - assert!( - !worker.capabilities.can_stop, - "{} should not be stoppable", - worker.worker_id - ); - } + assert!(!workers.items[0].capabilities.can_stop); + assert!(!workers.items[1].capabilities.can_stop); + assert!(workers.items[2].capabilities.can_stop); + assert!(workers.items[3].capabilities.can_stop); assert_eq!(workers.items[0].state, "stopped"); - assert_eq!(workers.items[1].state, "corrupted"); - assert_eq!(workers.items[2].state, "rejected"); - assert_eq!(workers.items[3].state, "errored"); + assert_eq!(workers.items[1].state, "cancelled"); + assert_eq!(workers.items[2].state, "paused"); + assert_eq!(workers.items[3].state, "idle"); let stopped_detail = registry.worker("remote:primary", "1").unwrap(); assert!(!stopped_detail.capabilities.can_stop); @@ -4764,22 +4651,20 @@ mod tests { } fn worker_json(runtime_id: &str, worker_id: &str) -> serde_json::Value { - worker_json_with_execution(runtime_id, worker_id, "alive", "idle") + worker_json_with_status(runtime_id, worker_id, "idle") } - fn worker_json_with_execution( + fn worker_json_with_status( runtime_id: &str, worker_id: &str, - backend: &str, - run_state: &str, + status: &str, ) -> serde_json::Value { let worker_id = worker_id.parse::().unwrap(); json!({ "worker_ref": { "runtime_id": runtime_id, "worker_id": worker_id }, "runtime_id": runtime_id, "worker_id": worker_id, - "status": "running", - "execution": { "backend": backend, "run_state": run_state }, + "status": status, "intent": { "kind": "role", "role": "coder", "purpose": "remote test" }, "profile": { "kind": "builtin", "value": "coder" }, "profile_source": { diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index d925734e..fb85e4ec 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -7192,15 +7192,6 @@ 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, @@ -9024,8 +9015,7 @@ mod tests { } #[tokio::test] - async fn embedded_runtime_fs_store_restores_catalog_config_bundle_and_corrupts_failed_execution() - { + async fn embedded_runtime_fs_store_restores_catalog_and_stops_failed_execution() { let dir = tempfile::tempdir().unwrap(); let config = test_server_config(dir.path().join("workspace")); let store_root = config.embedded_runtime_store_root.clone(); @@ -9112,14 +9102,8 @@ mod tests { .runtime .worker("embedded-worker-runtime", &worker_id) .expect("restored worker"); - assert_eq!(restored_worker.state, "corrupted"); + assert_eq!(restored_worker.state, "stopped"); assert!(!restored_worker.capabilities.can_stop); - assert!( - restored_worker - .diagnostics - .iter() - .any(|diagnostic| diagnostic.code == "embedded_worker_execution_corrupted") - ); let bundles = restored .runtime