From c4622e9e3af72add43d91a07e90ebd271846a045 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 17 Sep 2026 03:25:40 +0900 Subject: [PATCH] refactor: separate Worker identity from execution state --- crates/worker-runtime/src/catalog.rs | 22 +- crates/worker-runtime/src/fs_store.rs | 637 +++++++++++++++++--------- crates/worker-runtime/src/main.rs | 12 +- crates/worker-runtime/src/runtime.rs | 433 +++++++++++------ crates/workspace-server/src/hosts.rs | 114 +++-- 5 files changed, 845 insertions(+), 373 deletions(-) diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index a2646b0e..f9cfa9b5 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -266,11 +266,11 @@ pub struct CreateWorkerRequest { pub memory_settings: Option, } -/// Worker lifecycle status for the in-memory embedded runtime. +/// Last persisted Worker lifecycle status. /// -/// Run termination details are carried separately by the Worker protocol. In -/// particular, cancellation returns a Worker to `Idle`; it is not a lifecycle -/// state of its own. +/// This is not proof that the current Runtime process holds a live execution handle. Run +/// termination details remain separate Worker protocol state; in particular, cancellation +/// returns a Worker to `Idle` and is not a lifecycle state of its own. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum WorkerStatus { @@ -293,12 +293,17 @@ pub(crate) enum WorkerRestoreIntent { Explicit, } -/// Lightweight catalog row. +/// Lightweight persisted Worker identity projection. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerSummary { pub worker_ref: WorkerRef, pub worker_id: WorkerId, pub status: WorkerStatus, + /// Creation timestamp in Unix epoch milliseconds for records created on this schema. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_at_ms: Option, + /// Whether the persisted execution metadata was valid when this identity was loaded. + pub execution_metadata_available: bool, #[serde(default, skip_serializing_if = "Option::is_none")] pub worker_state: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -313,12 +318,17 @@ pub struct WorkerSummary { pub config_bundle: Option, } -/// Full Worker catalog/lifecycle detail. +/// Full persisted Worker identity and lifecycle detail. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerDetail { pub worker_ref: WorkerRef, pub worker_id: WorkerId, pub status: WorkerStatus, + /// Creation timestamp in Unix epoch milliseconds for records created on this schema. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_at_ms: Option, + /// Whether the persisted execution metadata was valid when this identity was loaded. + pub execution_metadata_available: bool, #[serde(default, skip_serializing_if = "Option::is_none")] pub worker_state: Option, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 8068fd68..aabcff61 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -1,5 +1,6 @@ use crate::catalog::{ - CreateWorkerRequest, WorkerRestoreIntent, WorkerStatus, WorkingDirectoryStatus, + ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerRestoreIntent, WorkerStatus, + WorkingDirectoryStatus, }; use crate::config_bundle::ConfigBundle; use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; @@ -8,22 +9,26 @@ use crate::identity::{ LegacyWorkerIdentityMapping, WorkerId, WorkerRef, legacy_worker_identity_mapping_digest, }; use crate::management::{RuntimeBackendKind, RuntimeStatus}; +use crate::profile_archive::ProfileSourceArchiveRef; use fs4::fs_std::FileExt; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; use std::fs::{self, File, OpenOptions}; -use std::io::{BufReader, Write}; +use std::io::{BufReader, Read, Write}; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; -const SCHEMA_VERSION: u32 = 7; -const PREVIOUS_SCHEMA_VERSION: u32 = 6; +const SCHEMA_VERSION: u32 = 8; +const PREVIOUS_SCHEMA_VERSION: u32 = 7; const RUNTIME_FILE: &str = "runtime.json"; const WORKERS_DIR: &str = "workers"; const WORKER_FILE: &str = "worker.json"; +const WORKER_EXECUTION_FILE: &str = "execution.json"; const WORKER_METADATA_FILE: &str = "metadata.json"; const LEGACY_OBSERVATIONS_FILE: &str = "observations.jsonl"; +const MAX_WORKER_RECORD_BYTES: u64 = 8 * 1024 * 1024; +const MAX_PERSISTED_WORKER_RECORDS: usize = 4096; static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1); @@ -165,27 +170,34 @@ impl FsRuntimeStore { atomic_write_json(&self.runtime_path(), &snapshot, "write runtime snapshot") } - pub(crate) fn write_worker_snapshot( + pub(crate) fn write_worker_record( &self, worker: &PersistedWorkerRecord, ) -> Result<(), RuntimeError> { self.ensure_worker_ref(&worker.worker_ref)?; let worker_dir = self.worker_dir(&worker.worker_id); fs::create_dir_all(&worker_dir).map_err(|source| RuntimeError::StoreIo { - operation: "create worker store", + operation: "create Worker store", path: worker_dir.clone(), source, })?; atomic_write_json( &worker_dir.join(WORKER_FILE), - &WorkerSnapshot::from_persisted(worker), - "write worker snapshot", + &WorkerIdentityRecord::from_persisted(worker), + "write Worker identity", )?; + if let PersistedWorkerExecutionState::Available(execution) = &worker.execution_state { + atomic_write_json( + &worker_dir.join(WORKER_EXECUTION_FILE), + &WorkerExecutionRecord::from_persisted(execution), + "write Worker execution record", + )?; + } remove_legacy_observations(&worker_dir); Ok(()) } - pub(crate) fn delete_worker_snapshot(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> { + pub(crate) fn delete_worker_record(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> { let worker_dir = self.worker_dir(worker_id); if !worker_dir.exists() { return Ok(()); @@ -232,44 +244,74 @@ impl FsRuntimeStore { })?; worker_dirs.sort_by_key(|entry| entry.path()); + if worker_dirs.len() > MAX_PERSISTED_WORKER_RECORDS { + return Err(RuntimeError::StoreCorrupt { + operation: "read Worker identities", + path: workers_dir, + message: format!( + "Worker record count {} exceeds limit {MAX_PERSISTED_WORKER_RECORDS}", + worker_dirs.len() + ), + }); + } + for entry in worker_dirs { let path = entry.path(); - if !path.is_dir() { + let is_directory = entry + .file_type() + .map(|file_type| file_type.is_dir() && !file_type.is_symlink()) + .unwrap_or(false); + if !is_directory { record_worker_load_diagnostic( &mut snapshot, None, - "ignored invalid worker store entry while loading runtime store", + "ignored invalid Worker store entry while loading Runtime store", ); continue; } - let worker_snapshot_path = path.join(WORKER_FILE); - let worker_snapshot: WorkerSnapshot = - match read_json(&worker_snapshot_path, "read worker snapshot") { - Ok(snapshot) => snapshot, + let identity_path = path.join(WORKER_FILE); + let identity: WorkerIdentityRecord = + match read_bounded_json(&identity_path, "read Worker identity") { + Ok(identity) => identity, Err(_error) => { record_worker_load_diagnostic( &mut snapshot, None, - "ignored corrupt worker snapshot while loading runtime store", + "ignored corrupt Worker identity while loading Runtime store", ); continue; } }; - if worker_snapshot.validate(&worker_snapshot_path).is_err() { + if identity.validate(&identity_path).is_err() { record_worker_load_diagnostic( &mut snapshot, - Some(worker_snapshot.worker_ref.clone()), - "ignored invalid worker snapshot while loading runtime store", + Some(identity.worker_ref.clone()), + "ignored invalid Worker identity while loading Runtime store", ); continue; } + let execution_path = path.join(WORKER_EXECUTION_FILE); + let execution_state = read_bounded_json::( + &execution_path, + "read Worker execution record", + ) + .and_then(|execution| execution.validate(&identity, &execution_path)) + .map(PersistedWorkerExecutionState::Available) + .unwrap_or_else(|_error| { + record_worker_load_diagnostic( + &mut snapshot, + Some(identity.worker_ref.clone()), + "Worker execution record is unavailable; retained Worker identity", + ); + PersistedWorkerExecutionState::Unavailable + }); remove_legacy_observations(&path); - let worker = worker_snapshot.into_persisted(); + let worker = identity.into_persisted(execution_state); if workers.insert(worker.worker_id.clone(), worker).is_some() { record_worker_load_diagnostic( &mut snapshot, None, - "ignored duplicate worker snapshot while loading runtime store", + "ignored duplicate Worker identity while loading Runtime store", ); } } @@ -375,17 +417,28 @@ pub(crate) struct PersistedWorkerExecutionBinding {} #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct PersistedWorkerExecution { + pub(crate) request: CreateWorkerRequest, pub(crate) binding: Option, pub(crate) restore_intent: WorkerRestoreIntent, } +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum PersistedWorkerExecutionState { + Available(PersistedWorkerExecution), + Unavailable, +} + #[derive(Clone, Debug)] pub(crate) struct PersistedWorkerRecord { pub(crate) worker_ref: WorkerRef, pub(crate) worker_id: WorkerId, - pub(crate) request: CreateWorkerRequest, + pub(crate) profile: ProfileSelector, + pub(crate) display_name: Option, + pub(crate) profile_source: ProfileSourceArchiveRef, + pub(crate) config_bundle: Option, + pub(crate) created_at_ms: Option, pub(crate) status: WorkerStatus, - pub(crate) execution: PersistedWorkerExecution, + pub(crate) execution_state: PersistedWorkerExecutionState, pub(crate) workspace_id: Option, pub(crate) working_directory: Option, } @@ -462,8 +515,8 @@ fn plan_runtime_store_migration( format!("Runtime store schema version {schema_version} is out of range"), ) })?; - let staging = migration_sibling(root, "schema-v7-staging")?; - let backup = migration_sibling(root, "pre-schema-v7-backup")?; + let staging = migration_sibling(root, "schema-v8-staging")?; + let backup = migration_sibling(root, "pre-schema-v8-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -523,14 +576,14 @@ fn plan_runtime_store_migration( if !snapshot_path .try_exists() .map_err(|source| RuntimeError::StoreIo { - operation: "inspect Worker snapshot", + operation: "inspect Worker record", path: snapshot_path.clone(), source, })? { continue; } - let snapshot: serde_json::Value = read_json(&snapshot_path, "read Worker snapshot")?; + let snapshot: serde_json::Value = read_json(&snapshot_path, "read Worker record")?; let (worker_id, workspace_id, legacy_mapping) = if current_schema_version == 1 { let legacy_worker_id = name.parse::().map_err(|_| { runtime_store_corrupt( @@ -545,7 +598,7 @@ fn plan_runtime_store_migration( .ok_or_else(|| { runtime_store_corrupt( &snapshot_path, - "legacy Worker snapshot is missing workspace_id; unscoped Workers require an explicit migration disposition" + "legacy Worker record is missing workspace_id; unscoped Workers require an explicit migration disposition" .to_string(), ) })? @@ -595,33 +648,24 @@ fn plan_runtime_store_migration( let mut migrated_worker_aggregate_count = 0; for worker in &mut planned { let snapshot_path = worker.source_dir.join(WORKER_FILE); - let snapshot: serde_json::Value = read_json(&snapshot_path, "read Worker snapshot")?; + let snapshot: serde_json::Value = read_json(&snapshot_path, "read Worker record")?; let migrated = migrate_worker_document( snapshot, current_schema_version, worker.legacy_mapping.as_ref(), &snapshot_path, )?; - let snapshot = validate_migrated_worker_document(&migrated, &snapshot_path)?; - if snapshot.worker_id != worker.worker_id { + let identity = validate_migrated_worker_documents(&migrated, &snapshot_path)?; + if identity.worker_id != worker.worker_id { return Err(runtime_store_corrupt( &snapshot_path, format!( - "Worker snapshot id {} does not match directory identity {}", - snapshot.worker_id, worker.worker_id + "Worker identity {} does not match directory identity {}", + identity.worker_id, worker.worker_id ), )); } - worker.workspace_id = worker - .workspace_id - .clone() - .or(snapshot.workspace_id) - .or_else(|| { - snapshot - .request - .workspace_api - .map(|workspace_api| workspace_api.workspace_id) - }); + worker.workspace_id = worker.workspace_id.clone().or(identity.workspace_id); let metadata_path = worker.source_dir.join(WORKER_METADATA_FILE); if metadata_path.is_file() { @@ -655,123 +699,107 @@ struct DiagnosticWorkerRefMigrationCounts { cleared: usize, } +#[derive(Clone, Debug)] +struct MigratedWorkerDocuments { + identity: serde_json::Value, + execution: serde_json::Value, +} + fn migrate_worker_document( mut document: serde_json::Value, source_schema_version: u32, _mapping: Option<&LegacyWorkerIdentityMapping>, - snapshot_path: &Path, -) -> Result { + identity_path: &Path, +) -> Result { if source_schema_version != PREVIOUS_SCHEMA_VERSION { return Err(runtime_store_corrupt( - snapshot_path, + identity_path, format!( - "unsupported Worker snapshot schema {source_schema_version}; expected {PREVIOUS_SCHEMA_VERSION}" + "unsupported Worker identity schema {source_schema_version}; expected {PREVIOUS_SCHEMA_VERSION}" ), )); } let object = document.as_object_mut().ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker snapshot must be an object".to_string(), - ) + runtime_store_corrupt(identity_path, "Worker record must be an object".to_string()) })?; - if let Some(run_generation) = object.remove("run_generation") - && run_generation.as_u64().is_none() - { - return Err(runtime_store_corrupt( - snapshot_path, - "Worker snapshot run_generation must be an unsigned integer".to_string(), - )); - } - let execution = object - .get_mut("execution") - .and_then(serde_json::Value::as_object_mut) + let mut execution = object + .remove("execution") + .and_then(|value| value.as_object().cloned()) .ok_or_else(|| { runtime_store_corrupt( - snapshot_path, - "Worker snapshot execution must be an object".to_string(), + identity_path, + "Worker record execution must be an object".to_string(), ) })?; - let last_run_generation = execution - .remove("last_run_generation") - .and_then(|value| value.as_u64()) - .ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker execution last_run_generation must be an unsigned integer".to_string(), - ) - })?; - let binding = execution.get_mut("binding").ok_or_else(|| { + let request_value = object.remove("request").ok_or_else(|| { runtime_store_corrupt( - snapshot_path, - "Worker execution is missing binding".to_string(), + identity_path, + "Worker record request must be present".to_string(), ) })?; - if let Some(binding_object) = binding.as_object_mut() { - let binding_run_generation = binding_object - .remove("run_generation") - .and_then(|value| value.as_u64()) - .ok_or_else(|| { - runtime_store_corrupt( - snapshot_path, - "Worker execution binding run_generation must be an unsigned integer" - .to_string(), - ) - })?; - if binding_run_generation != last_run_generation { - return Err(runtime_store_corrupt( - snapshot_path, - format!( - "execution binding run_generation {binding_run_generation} does not match last_run_generation {last_run_generation}" - ), - )); - } - if !binding_object.is_empty() { - return Err(runtime_store_corrupt( - snapshot_path, - "Worker execution binding contains unsupported fields".to_string(), - )); - } - } else if !binding.is_null() { - return Err(runtime_store_corrupt( - snapshot_path, - "Worker execution binding must be an object or null".to_string(), - )); - } - if !execution.contains_key("restore_intent") { - return Err(runtime_store_corrupt( - snapshot_path, - "Worker execution is missing restore_intent".to_string(), - )); - } - if execution - .keys() - .any(|key| key != "binding" && key != "restore_intent") - { - return Err(runtime_store_corrupt( - snapshot_path, - "Worker execution contains unsupported fields".to_string(), - )); - } + let request: CreateWorkerRequest = + serde_json::from_value(request_value.clone()).map_err(|error| { + runtime_store_corrupt( + identity_path, + format!("decode Worker execution request: {error}"), + ) + })?; + object.insert( + "profile".to_string(), + serde_json::to_value(&request.profile).expect("Profile selector serializes"), + ); + object.insert( + "display_name".to_string(), + serde_json::to_value(&request.display_name).expect("display name serializes"), + ); + object.insert( + "profile_source".to_string(), + serde_json::to_value(request.profile_source.reference()) + .expect("Profile source reference serializes"), + ); + object.insert( + "config_bundle".to_string(), + serde_json::to_value(&request.config_bundle).expect("config bundle serializes"), + ); + execution.insert("request".to_string(), request_value); + execution.insert( + "schema_version".to_string(), + serde_json::Value::from(SCHEMA_VERSION), + ); object.insert( "schema_version".to_string(), serde_json::Value::from(SCHEMA_VERSION), ); - Ok(document) + let migrated = MigratedWorkerDocuments { + identity: document, + execution: serde_json::Value::Object(execution), + }; + validate_migrated_worker_documents(&migrated, identity_path)?; + Ok(migrated) } -fn validate_migrated_worker_document( - document: &serde_json::Value, - snapshot_path: &Path, -) -> Result { - let snapshot: WorkerSnapshot = serde_json::from_value(document.clone()).map_err(|error| { - runtime_store_corrupt( - snapshot_path, - format!("decode migrated Worker snapshot: {error}"), - ) - })?; - snapshot.validate(snapshot_path)?; - Ok(snapshot) +fn validate_migrated_worker_documents( + documents: &MigratedWorkerDocuments, + identity_path: &Path, +) -> Result { + let identity: WorkerIdentityRecord = serde_json::from_value(documents.identity.clone()) + .map_err(|error| { + runtime_store_corrupt( + identity_path, + format!("decode migrated Worker identity: {error}"), + ) + })?; + identity.validate(identity_path)?; + let execution_path = identity_path.with_file_name(WORKER_EXECUTION_FILE); + let execution: WorkerExecutionRecord = serde_json::from_value(documents.execution.clone()) + .map_err(|error| { + runtime_store_corrupt( + &execution_path, + format!("decode migrated Worker execution record: {error}"), + ) + })?; + execution.validate(&identity, &execution_path)?; + Ok(identity) } fn runtime_worker_name(worker_id: WorkerId) -> String { @@ -1125,8 +1153,8 @@ fn migrate_runtime_store( if !plan.migration_required { return Ok(plan); } - let staging = migration_sibling(root, "schema-v7-staging")?; - let backup = migration_sibling(root, "pre-schema-v7-backup")?; + let staging = migration_sibling(root, "schema-v8-staging")?; + let backup = migration_sibling(root, "pre-schema-v8-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -1208,12 +1236,12 @@ fn migrate_runtime_store_in_place( runtime_store_corrupt( &source_snapshot_path, format!( - "decode Worker snapshot {}: {error}", + "decode Worker record {}: {error}", source_snapshot_path.display() ), ) })?; - let snapshot = migrate_worker_document( + let documents = migrate_worker_document( snapshot, plan.current_schema_version, planned_worker.legacy_mapping.as_ref(), @@ -1242,12 +1270,17 @@ fn migrate_runtime_store_in_place( fs::rename(source_dir, &migrated_dir) .map_err(|error| runtime_io_error("rename", source_dir, error))?; } - let migrated_snapshot_path = migrated_dir.join(WORKER_FILE); + let migrated_identity_path = migrated_dir.join(WORKER_FILE); atomic_write_json( - &migrated_snapshot_path, - &snapshot, + &migrated_identity_path, + &documents.identity, "migrate Worker identity", )?; + atomic_write_json( + &migrated_dir.join(WORKER_EXECUTION_FILE), + &documents.execution, + "migrate Worker execution record", + )?; if let Some(metadata) = metadata { atomic_write_json( &migrated_dir.join(WORKER_METADATA_FILE), @@ -1296,7 +1329,7 @@ fn record_worker_load_diagnostic( id, worker_ref, severity: DiagnosticSeverity::Warning, - code: "worker_snapshot_ignored".to_string(), + code: "worker_record_unavailable".to_string(), message: message.into(), }); } @@ -1352,28 +1385,46 @@ impl RuntimeSnapshot { } #[derive(Clone, Debug, Serialize, Deserialize)] -struct WorkerSnapshot { +#[serde(deny_unknown_fields)] +struct WorkerIdentityRecord { schema_version: u32, worker_ref: WorkerRef, worker_id: WorkerId, - request: CreateWorkerRequest, + profile: ProfileSelector, + display_name: Option, + profile_source: ProfileSourceArchiveRef, + config_bundle: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + created_at_ms: Option, status: WorkerStatus, - execution: PersistedWorkerExecution, #[serde(default, skip_serializing_if = "Option::is_none")] workspace_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] working_directory: Option, } -impl WorkerSnapshot { +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct WorkerExecutionRecord { + schema_version: u32, + request: CreateWorkerRequest, + #[serde(default, skip_serializing_if = "Option::is_none")] + binding: Option, + restore_intent: WorkerRestoreIntent, +} + +impl WorkerIdentityRecord { fn from_persisted(worker: &PersistedWorkerRecord) -> Self { Self { schema_version: SCHEMA_VERSION, worker_ref: worker.worker_ref.clone(), - worker_id: worker.worker_id.clone(), - request: worker.request.clone(), + worker_id: worker.worker_id, + profile: worker.profile.clone(), + display_name: worker.display_name.clone(), + profile_source: worker.profile_source.clone(), + config_bundle: worker.config_bundle.clone(), + created_at_ms: worker.created_at_ms, status: worker.status, - execution: worker.execution.clone(), workspace_id: worker.workspace_id.clone(), working_directory: worker.working_directory.clone(), } @@ -1382,7 +1433,7 @@ impl WorkerSnapshot { fn validate(&self, path: &Path) -> Result<(), RuntimeError> { if self.schema_version != SCHEMA_VERSION { return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", + operation: "read Worker identity", path: path.to_path_buf(), message: format!( "unsupported schema version {}, expected {}", @@ -1392,7 +1443,7 @@ impl WorkerSnapshot { } if self.worker_ref.worker_id != self.worker_id { return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", + operation: "read Worker identity", path: path.to_path_buf(), message: format!( "worker_ref id {} does not match worker_id {}", @@ -1400,11 +1451,94 @@ impl WorkerSnapshot { ), }); } - match (self.status, self.execution.restore_intent) { + let expected_name = self.worker_id.to_string(); + if path + .parent() + .and_then(Path::file_name) + .and_then(|name| name.to_str()) + != Some(expected_name.as_str()) + { + return Err(RuntimeError::StoreCorrupt { + operation: "read Worker identity", + path: path.to_path_buf(), + message: format!( + "Worker identity location does not match worker_id {}", + self.worker_id + ), + }); + } + Ok(()) + } + + fn into_persisted( + self, + execution_state: PersistedWorkerExecutionState, + ) -> PersistedWorkerRecord { + PersistedWorkerRecord { + worker_ref: self.worker_ref, + worker_id: self.worker_id, + profile: self.profile, + display_name: self.display_name, + profile_source: self.profile_source, + config_bundle: self.config_bundle, + created_at_ms: self.created_at_ms, + status: self.status, + execution_state, + workspace_id: self.workspace_id, + working_directory: self.working_directory, + } + } +} + +impl WorkerExecutionRecord { + fn from_persisted(execution: &PersistedWorkerExecution) -> Self { + Self { + schema_version: SCHEMA_VERSION, + request: execution.request.clone(), + binding: execution.binding.clone(), + restore_intent: execution.restore_intent, + } + } + + fn validate( + &self, + identity: &WorkerIdentityRecord, + path: &Path, + ) -> Result { + if self.schema_version != SCHEMA_VERSION { + return Err(RuntimeError::StoreCorrupt { + operation: "read Worker execution record", + path: path.to_path_buf(), + message: format!( + "unsupported schema version {}, expected {}", + self.schema_version, SCHEMA_VERSION + ), + }); + } + let request_workspace_id = self + .request + .workspace_api + .as_ref() + .map(|workspace_api| workspace_api.workspace_id.as_str()); + if identity.workspace_id.as_deref() != request_workspace_id + || self.request.worker_id != identity.worker_id + || self.request.profile != identity.profile + || self.request.display_name != identity.display_name + || self.request.profile_source.reference() != identity.profile_source + || self.request.config_bundle != identity.config_bundle + { + return Err(RuntimeError::StoreCorrupt { + operation: "read Worker execution record", + path: path.to_path_buf(), + message: "Worker execution request does not match persisted Worker identity" + .to_string(), + }); + } + match (identity.status, self.restore_intent) { (status, WorkerRestoreIntent::Automatic) if status.is_active() => { - if self.execution.binding.is_none() { + if self.binding.is_none() { return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", + operation: "read Worker execution record", path: path.to_path_buf(), message: "automatic restore intent requires an execution binding" .to_string(), @@ -1414,35 +1548,80 @@ impl WorkerSnapshot { (WorkerStatus::Stopped, WorkerRestoreIntent::Explicit) => {} _ => { return Err(RuntimeError::StoreCorrupt { - operation: "read worker snapshot", + operation: "read Worker execution record", path: path.to_path_buf(), message: format!( - "worker status {:?} conflicts with restore intent {:?}", - self.status, self.execution.restore_intent + "Worker status {:?} conflicts with restore intent {:?}", + identity.status, self.restore_intent ), }); } } - Ok(()) + Ok(PersistedWorkerExecution { + request: self.request.clone(), + binding: self.binding.clone(), + restore_intent: self.restore_intent, + }) } +} - fn into_persisted(self) -> PersistedWorkerRecord { - let workspace_id = self.workspace_id.or_else(|| { - self.request - .workspace_api - .as_ref() - .map(|workspace_api| workspace_api.workspace_id.clone()) +fn read_bounded_json(path: &Path, operation: &'static str) -> Result +where + T: for<'de> Deserialize<'de>, +{ + let metadata = fs::symlink_metadata(path).map_err(|source| match source.kind() { + std::io::ErrorKind::NotFound => RuntimeError::StoreMissing { + operation, + path: path.to_path_buf(), + }, + _ => RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + }, + })?; + if !metadata.file_type().is_file() || metadata.file_type().is_symlink() { + return Err(RuntimeError::StoreCorrupt { + operation, + path: path.to_path_buf(), + message: "record is not a regular file".to_string(), }); - PersistedWorkerRecord { - worker_ref: self.worker_ref, - worker_id: self.worker_id, - request: self.request, - status: self.status, - execution: self.execution, - workspace_id, - working_directory: self.working_directory, - } } + if metadata.len() > MAX_WORKER_RECORD_BYTES { + return Err(RuntimeError::StoreCorrupt { + operation, + path: path.to_path_buf(), + message: format!( + "record size {} exceeds limit {MAX_WORKER_RECORD_BYTES}", + metadata.len() + ), + }); + } + let file = File::open(path).map_err(|source| RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + })?; + let mut bytes = Vec::with_capacity(metadata.len() as usize); + file.take(MAX_WORKER_RECORD_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|source| RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source, + })?; + if bytes.len() as u64 > MAX_WORKER_RECORD_BYTES { + return Err(RuntimeError::StoreCorrupt { + operation, + path: path.to_path_buf(), + message: format!("record exceeds limit {MAX_WORKER_RECORD_BYTES} while reading"), + }); + } + serde_json::from_slice(&bytes).map_err(|source| RuntimeError::StoreCorrupt { + operation, + path: path.to_path_buf(), + message: source.to_string(), + }) } fn read_json(path: &Path, operation: &'static str) -> Result @@ -1668,50 +1847,100 @@ mod tests { assert_eq!(plan.worker_count, 0); } - #[test] - fn schema_v6_worker_migration_removes_generation_and_preserves_active_restore() { - let path = Path::new("worker.json"); - let source = serde_json::json!({ + fn schema_v7_worker_document(worker_id: WorkerId) -> serde_json::Value { + serde_json::json!({ "schema_version": PREVIOUS_SCHEMA_VERSION, - "run_generation": 7, + "worker_ref": { "worker_id": worker_id }, + "worker_id": worker_id, + "request": { + "worker_id": worker_id, + "create_fingerprint": "test-create", + "profile": { "kind": "builtin", "value": "builtin:coder" }, + "profile_source": { + "kind": "workspace_config", + "archive": { + "id": "archive-1", + "digest": "sha256:archive", + "size_bytes": 0, + "source_graph": { + "source_count": 0, + "total_source_bytes": 0, + "entrypoints": {}, + "import_count": 0 + } + } + } + }, "status": "running", "execution": { - "last_run_generation": 7, - "binding": { "run_generation": 7 }, + "binding": {}, "restore_intent": "automatic" } - }); - - let migrated = - migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, path).unwrap(); - - assert_eq!(migrated["schema_version"], SCHEMA_VERSION); - assert_eq!(migrated["status"], "running"); - assert_eq!(migrated["execution"]["binding"], serde_json::json!({})); - assert_eq!(migrated["execution"]["restore_intent"], "automatic"); - assert!(migrated.get("run_generation").is_none()); - assert!(migrated["execution"].get("last_run_generation").is_none()); + }) } #[test] - fn schema_v6_worker_migration_rejects_mismatched_generation_state() { - let path = Path::new("worker.json"); - let source = serde_json::json!({ - "schema_version": PREVIOUS_SCHEMA_VERSION, - "execution": { - "last_run_generation": 7, - "binding": { "run_generation": 6 }, - "restore_intent": "automatic" - } - }); + fn bounded_worker_record_read_rejects_oversize_before_parse() { + let root = tempfile::tempdir().unwrap(); + let path = root.path().join(WORKER_FILE); + File::create(&path) + .unwrap() + .set_len(MAX_WORKER_RECORD_BYTES + 1) + .unwrap(); let error = - migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, path).unwrap_err(); + read_bounded_json::(&path, "read Worker identity").unwrap_err(); + + assert!(matches!(error, RuntimeError::StoreCorrupt { .. })); + assert!(error.to_string().contains("exceeds limit")); + } + + #[test] + fn schema_v7_worker_migration_splits_identity_from_execution_metadata() { + let root = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::now_v7(); + let worker_dir = root.path().join(WORKERS_DIR).join(worker_id.to_string()); + fs::create_dir_all(&worker_dir).unwrap(); + let identity_path = worker_dir.join(WORKER_FILE); + + let migrated = migrate_worker_document( + schema_v7_worker_document(worker_id), + PREVIOUS_SCHEMA_VERSION, + None, + &identity_path, + ) + .unwrap(); + + assert_eq!(migrated.identity["schema_version"], SCHEMA_VERSION); + assert_eq!(migrated.identity["status"], "running"); + assert!(migrated.identity.get("request").is_none()); + assert!(migrated.identity.get("execution").is_none()); + assert_eq!(migrated.execution["schema_version"], SCHEMA_VERSION); + assert_eq!( + migrated.execution["request"]["worker_id"], + worker_id.to_string() + ); + assert_eq!(migrated.execution["binding"], serde_json::json!({})); + assert_eq!(migrated.execution["restore_intent"], "automatic"); + } + + #[test] + fn schema_v7_worker_migration_rejects_invalid_execution_metadata() { + let root = tempfile::tempdir().unwrap(); + let worker_id = WorkerId::now_v7(); + let worker_dir = root.path().join(WORKERS_DIR).join(worker_id.to_string()); + fs::create_dir_all(&worker_dir).unwrap(); + let identity_path = worker_dir.join(WORKER_FILE); + let mut source = schema_v7_worker_document(worker_id); + source["execution"]["restore_intent"] = serde_json::json!(17); + + let error = migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, &identity_path) + .unwrap_err(); assert!( error .to_string() - .contains("does not match last_run_generation") + .contains("decode migrated Worker execution record") ); } } diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index 1c498d02..adec4f1b 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -1,7 +1,7 @@ -// Worker-backed Runtime REST process wrapper. +// Worker-backed Runtime host service. // -// This binary starts a Runtime command API with a real worker execution backend. -// A REST Runtime process that cannot spawn Workers is not a valid Runtime for the +// This binary starts a Runtime command API with a real Worker execution backend. +// A Runtime service that cannot create and restore Workers is not available to the // Workspace Browser. use std::collections::VecDeque; @@ -1152,7 +1152,7 @@ fn usage() -> &'static str { yoi-runtime migrate --dry-run [--runtime-id ] [OPTIONS] Starts a worker-backed Runtime REST command API for a trusted backend/proxy. -Browsers must not connect to this Runtime process directly. +Browsers must not connect to this Runtime service directly. Options: --bind Bind socket address (default: 127.0.0.1:38800) @@ -1280,7 +1280,7 @@ mod tests { std::fs::write( root.join("runtime.json"), serde_json::to_vec_pretty(&serde_json::json!({ - "schema_version": 6, + "schema_version": 7, "display_name": "local", "backend": "fs_store", "status": "running", @@ -1319,7 +1319,7 @@ mod tests { std::fs::write( root.join("runtime.json"), serde_json::to_vec_pretty(&serde_json::json!({ - "schema_version": 6, + "schema_version": 7, "display_name": "local", "backend": "fs_store", "status": 3, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index df94617d..4e040b71 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -20,7 +20,7 @@ use crate::execution::{ #[cfg(feature = "fs-store")] use crate::fs_store::{ FsRuntimeStore, FsRuntimeStoreOptions, PersistedRuntimeState, PersistedWorkerExecution, - PersistedWorkerExecutionBinding, PersistedWorkerRecord, + PersistedWorkerExecutionBinding, PersistedWorkerExecutionState, PersistedWorkerRecord, }; use crate::identity::{WorkerId, WorkerRef}; use crate::interaction::{WorkerInput, WorkerInputKind, WorkerInteractionAck}; @@ -29,6 +29,7 @@ use crate::management::{ }; #[cfg(feature = "ws-server")] use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; +use crate::profile_archive::ProfileSourceArchiveRef; use crate::resource::{ BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceKind, REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret, @@ -289,6 +290,9 @@ impl Runtime { let mut active_worker_count = 0; let mut stopped_worker_count = 0; for worker in state.workers.values() { + if !worker.execution_metadata_available { + continue; + } match worker.status { WorkerStatus::Idle | WorkerStatus::Running | WorkerStatus::Paused => { active_worker_count += 1; @@ -787,7 +791,13 @@ impl Runtime { request.worker_id ))); } - if existing.request.create_fingerprint != request.create_fingerprint { + let existing_request = existing.request.as_ref().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: existing.worker_id, + message: "persisted Worker restore request is unavailable".to_string(), + } + })?; + if existing_request.create_fingerprint != request.create_fingerprint { return Err(RuntimeError::InvalidRequest(format!( "worker {} was already created with a different fingerprint", request.worker_id @@ -951,7 +961,13 @@ impl Runtime { request.worker_id ))); } - if existing.request.create_fingerprint != request.create_fingerprint { + let existing_request = existing.request.as_ref().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: existing.worker_id, + message: "persisted Worker restore request is unavailable".to_string(), + } + })?; + if existing_request.create_fingerprint != request.create_fingerprint { return Err(RuntimeError::InvalidRequest(format!( "worker {} was already created with a different fingerprint", request.worker_id @@ -986,7 +1002,19 @@ impl Runtime { status: WorkerStatus::Stopped, worker_state: None, workspace_id: scope.map(|scope| scope.workspace_id.clone()), - request: durable_request, + profile: durable_request.profile.clone(), + display_name: durable_request.display_name.clone(), + profile_source: durable_request.profile_source.reference(), + config_bundle: durable_request.config_bundle.clone(), + request: Some(durable_request), + created_at_ms: Some( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(u128::from(u64::MAX)) as u64, + ), + execution_metadata_available: true, execution_bound: true, restore_intent: WorkerRestoreIntent::Explicit, working_directory: None, @@ -1029,7 +1057,11 @@ impl Runtime { if let Some(mut initial_input) = { let state = self.lock()?; - state.worker(&worker_ref)?.request.initial_input.clone() + state + .worker(&worker_ref)? + .request + .as_ref() + .and_then(|request| request.initial_input.clone()) } { let expected_submission_id = initial_input .submission_request_id @@ -1291,7 +1323,13 @@ impl Runtime { let previous_workspace_api = { let state = self.lock()?; let worker = state.worker(worker_ref)?; - if let Some(existing) = worker.request.workspace_api.as_ref() + let request = worker.request.as_ref().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: worker.worker_id, + message: "persisted Worker restore request is unavailable".to_string(), + } + })?; + if let Some(existing) = request.workspace_api.as_ref() && (existing.workspace_id != workspace_api.workspace_id || existing.base_url.trim_end_matches('/') != workspace_api.base_url.trim_end_matches('/')) @@ -1301,14 +1339,26 @@ impl Runtime { .to_string(), )); } - worker.request.workspace_api.clone() + request.workspace_api.clone() }; { let mut state = self.lock()?; - state.worker_mut(worker_ref)?.request.workspace_api = Some(workspace_api); - if let Err(error) = state.persist_runtime_snapshot() { - state.worker_mut(worker_ref)?.request.workspace_api = previous_workspace_api; + let worker = state.worker_mut(worker_ref)?; + let request = worker.request.as_mut().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: worker.worker_id, + message: "persisted Worker restore request is unavailable".to_string(), + } + })?; + request.workspace_api = Some(workspace_api); + if let Err(error) = state.persist_worker(&worker_ref.worker_id) { + state + .worker_mut(worker_ref)? + .request + .as_mut() + .expect("restore request existed before persistence") + .workspace_api = previous_workspace_api; return Err(error); } } @@ -1371,7 +1421,10 @@ impl Runtime { }); } match mode { - WorkerRestoreMode::Explicit if worker.status != WorkerStatus::Stopped => { + WorkerRestoreMode::Explicit + if worker.execution_metadata_available + && worker.status != WorkerStatus::Stopped => + { return Err(RuntimeError::InvalidRequest(format!( "worker {} is not stopped", worker_ref.worker_id @@ -1385,7 +1438,13 @@ impl Runtime { } _ => {} } - (worker.request.clone(), worker.working_directory.clone()) + let request = worker.request.clone().ok_or_else(|| { + RuntimeError::WorkerExecutionUnavailable { + worker_id: worker.worker_id, + message: "persisted Worker restore request is unavailable".to_string(), + } + })?; + (request, worker.working_directory.clone()) }; let backend = state.execution_backend.clone().ok_or_else(|| { RuntimeError::WorkerExecutionUnavailable { @@ -1829,6 +1888,7 @@ impl Runtime { let detail = { let worker = state.worker_mut(worker_ref)?; worker.execution_handle = Some(handle); + worker.execution_metadata_available = true; worker.execution_bound = true; worker.status = WorkerStatus::Idle; let _ = worker.apply_worker_state(&initial_worker_state); @@ -1867,7 +1927,7 @@ impl Runtime { fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { let mut state = self.lock()?; if state.workers.contains_key(&worker_ref.worker_id) { - state.delete_worker_snapshot(&worker_ref.worker_id)?; + state.delete_worker_record(&worker_ref.worker_id)?; let record = state .workers .remove(&worker_ref.worker_id) @@ -2109,7 +2169,7 @@ impl Runtime { state.ensure_running()?; state.ensure_worker_ref(worker_ref)?; let worker = state.worker(worker_ref)?; - if worker.status.is_active() { + if worker.execution_handle.is_some() && worker.status.is_active() { return Err(RuntimeError::InvalidRequest(format!( "worker {} is running and must be stopped before deletion", worker_ref.worker_id @@ -2140,13 +2200,13 @@ impl Runtime { state.ensure_running()?; state.ensure_worker_ref(worker_ref)?; let worker = state.worker(worker_ref)?; - if worker.status.is_active() { + if worker.execution_handle.is_some() && worker.status.is_active() { return Err(RuntimeError::InvalidRequest(format!( "worker {} became active before deletion", worker_ref.worker_id ))); } - state.delete_worker_snapshot(&worker_ref.worker_id)?; + state.delete_worker_record(&worker_ref.worker_id)?; let removed = state.workers.remove(&worker_ref.worker_id).ok_or_else(|| { RuntimeError::WorkerNotFound { worker_id: worker_ref.worker_id, @@ -2410,6 +2470,7 @@ impl Runtime { ))); } worker.execution_handle = Some(handle); + worker.execution_metadata_available = true; worker.execution_bound = true; worker.status = status; let _ = worker.apply_worker_state(&worker_state); @@ -2445,6 +2506,7 @@ impl Runtime { let mut state = self.lock()?; let worker = state.worker_mut(worker_ref)?; worker.execution_handle = None; + worker.execution_bound = false; worker.status = WorkerStatus::Stopped; worker.worker_state = None; worker.restore_intent = WorkerRestoreIntent::Explicit; @@ -2738,6 +2800,18 @@ impl RuntimeState { let diagnostics = persisted.diagnostics; let next_diagnostic_id = persisted.next_diagnostic_id; for (worker_id, worker) in persisted.workers { + let (request, execution_metadata_available, execution_bound, restore_intent) = + match worker.execution_state { + PersistedWorkerExecutionState::Available(execution) => ( + Some(execution.request), + true, + execution.binding.is_some(), + execution.restore_intent, + ), + PersistedWorkerExecutionState::Unavailable => { + (None, false, false, restore_intent_for_status(worker.status)) + } + }; workers.insert( worker_id, WorkerRecord { @@ -2746,9 +2820,15 @@ impl RuntimeState { status: worker.status, worker_state: None, workspace_id: worker.workspace_id, - request: worker.request, - execution_bound: worker.execution.binding.is_some(), - restore_intent: worker.execution.restore_intent, + profile: worker.profile, + display_name: worker.display_name, + profile_source: worker.profile_source, + config_bundle: worker.config_bundle, + request, + created_at_ms: worker.created_at_ms, + execution_metadata_available, + execution_bound, + restore_intent, working_directory: worker.working_directory, execution_handle: None, internal_workers: InternalWorkerActivityProjection::default(), @@ -2826,15 +2906,15 @@ impl RuntimeState { .ok_or_else(|| RuntimeError::WorkerNotFound { worker_id: *worker_id, })?; - store.write_worker_snapshot(&worker.persisted_record())?; + store.write_worker_record(&worker.persisted_record())?; } Ok(()) } #[cfg(feature = "fs-store")] - fn delete_worker_snapshot(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> { + fn delete_worker_record(&self, worker_id: &WorkerId) -> Result<(), RuntimeError> { if let Some(store) = self.fs_store() { - store.delete_worker_snapshot(worker_id)?; + store.delete_worker_record(worker_id)?; } Ok(()) } @@ -2860,7 +2940,7 @@ impl RuntimeState { } #[cfg(not(feature = "fs-store"))] - fn delete_worker_snapshot(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> { + fn delete_worker_record(&self, _worker_id: &WorkerId) -> Result<(), RuntimeError> { Ok(()) } @@ -3064,7 +3144,7 @@ impl RuntimeState { .map_err(subscription_validation_error) }) .transpose()?; - let profile = match &worker.request.profile { + let profile = match &worker.profile { ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => Some(name.clone()), }; Ok(SubscriptionWorker { @@ -3080,7 +3160,7 @@ impl RuntimeState { state: subscription_worker_state(worker.status), has_running_internal_workers: worker.internal_workers.has_running_worker(), workspace_id: worker.workspace_id.clone(), - display_name: worker.request.display_name.clone(), + display_name: worker.display_name.clone(), profile, repository_id, repository_key: None, @@ -3188,7 +3268,11 @@ impl RuntimeState { .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) + || worker + .request + .as_ref() + .and_then(requested_primary_workdir_id) + == Some(working_directory_id) { Some(worker.worker_id) } else { @@ -3233,6 +3317,7 @@ impl RuntimeState { }); let worker = self.worker_mut(worker_ref)?; worker.execution_handle = None; + worker.execution_bound = false; worker.status = WorkerStatus::Stopped; worker.restore_intent = WorkerRestoreIntent::Explicit; worker.internal_workers.clear(); @@ -3585,7 +3670,13 @@ struct WorkerRecord { status: WorkerStatus, worker_state: Option, workspace_id: Option, - request: CreateWorkerRequest, + profile: ProfileSelector, + display_name: Option, + profile_source: ProfileSourceArchiveRef, + config_bundle: Option, + request: Option, + created_at_ms: Option, + execution_metadata_available: bool, execution_bound: bool, restore_intent: WorkerRestoreIntent, working_directory: Option, @@ -3607,13 +3698,15 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, + created_at_ms: self.created_at_ms, + execution_metadata_available: self.execution_metadata_available, worker_state: self.worker_state.clone(), workspace_id: self.workspace_id.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(), - config_bundle: self.request.config_bundle.clone(), + profile: self.profile.clone(), + display_name: self.display_name.clone(), + profile_source: self.profile_source.clone(), + config_bundle: self.config_bundle.clone(), } } @@ -3622,29 +3715,42 @@ impl WorkerRecord { worker_ref: self.worker_ref.clone(), worker_id: self.worker_id, status: self.status, + created_at_ms: self.created_at_ms, + execution_metadata_available: self.execution_metadata_available, worker_state: self.worker_state.clone(), workspace_id: self.workspace_id.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(), - config_bundle: self.request.config_bundle.clone(), + profile: self.profile.clone(), + display_name: self.display_name.clone(), + profile_source: self.profile_source.clone(), + config_bundle: self.config_bundle.clone(), } } #[cfg(feature = "fs-store")] fn persisted_record(&self) -> PersistedWorkerRecord { + let execution_state = match (self.execution_metadata_available, self.request.clone()) { + (true, Some(request)) => { + PersistedWorkerExecutionState::Available(PersistedWorkerExecution { + request, + binding: self + .execution_bound + .then_some(PersistedWorkerExecutionBinding {}), + restore_intent: self.restore_intent, + }) + } + _ => PersistedWorkerExecutionState::Unavailable, + }; PersistedWorkerRecord { worker_ref: self.worker_ref.clone(), - worker_id: self.worker_id.clone(), - request: self.request.clone(), + worker_id: self.worker_id, + profile: self.profile.clone(), + display_name: self.display_name.clone(), + profile_source: self.profile_source.clone(), + config_bundle: self.config_bundle.clone(), + created_at_ms: self.created_at_ms, status: self.status, - execution: PersistedWorkerExecution { - binding: self - .execution_bound - .then_some(PersistedWorkerExecutionBinding {}), - restore_intent: self.restore_intent, - }, + execution_state, workspace_id: self.workspace_id.clone(), working_directory: self.working_directory.clone(), } @@ -5433,7 +5539,8 @@ mod tests { .worker(&worker.worker_ref) .unwrap() .request - .workspace_api, + .as_ref() + .and_then(|request| request.workspace_api.clone()), Some(replacement) ); } @@ -6744,7 +6851,7 @@ mod tests { assert!( error .to_string() - .contains("unsupported Runtime store schema version 2; expected 6 or 7") + .contains("unsupported Runtime store schema version 2; expected 7 or 8") ); let _ = std::fs::remove_dir_all(root); @@ -6783,22 +6890,19 @@ mod tests { .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 = + let worker_identity: serde_json::Value = serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap()) .unwrap(); - assert_eq!(worker_snapshot["schema_version"], serde_json::json!(7)); - assert_eq!(worker_snapshot["status"], serde_json::json!("stopped")); - assert!( - worker_snapshot["execution"] - .get("last_run_generation") - .is_none() - ); + let worker_execution: serde_json::Value = serde_json::from_slice( + &std::fs::read(worker_store_dir.join("execution.json")).unwrap(), + ) + .unwrap(); + assert_eq!(worker_identity["schema_version"], serde_json::json!(8)); + assert_eq!(worker_identity["status"], serde_json::json!("stopped")); + assert_eq!(worker_execution["schema_version"], serde_json::json!(8)); + assert_eq!(worker_execution["binding"], serde_json::json!({})); assert_eq!( - worker_snapshot["execution"]["binding"], - serde_json::json!({}) - ); - assert_eq!( - worker_snapshot["execution"]["restore_intent"], + worker_execution["restore_intent"], serde_json::json!("explicit") ); assert!(!root.join("events.jsonl").exists()); @@ -6858,7 +6962,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_restores_workspace_scope_and_hides_legacy_workers_from_scoped_access() { + fn fs_store_restores_explicit_workspace_identity_without_inference() { let root = fs_store_root("workspace-scope"); let runtime = Runtime::with_fs_store_and_execution_backend( crate::fs_store::FsRuntimeStoreOptions { @@ -6908,26 +7012,22 @@ mod tests { .worker_detail_scoped(&scope("workspace-a", "server-a"), &legacy.worker_ref) .unwrap_err(); assert!(matches!(legacy_error, RuntimeError::WorkerNotFound { .. })); - let recovered_legacy = restored + let missing_identity_error = restored .worker_detail_scoped( &scope("workspace-b", "server-b"), &recoverable_legacy.worker_ref, ) - .unwrap(); - assert_eq!( - recovered_legacy.workspace_id.as_deref(), - Some("workspace-b") - ); - let stolen_legacy_error = restored - .worker_detail_scoped( - &scope("workspace-b", "server-c"), - &recoverable_legacy.worker_ref, - ) .unwrap_err(); assert!(matches!( - stolen_legacy_error, - RuntimeError::WorkspaceOwnerMismatch { .. } + missing_identity_error, + RuntimeError::WorkerNotFound { .. } )); + assert!( + !restored + .worker_detail(&recoverable_legacy.worker_ref) + .unwrap() + .execution_metadata_available + ); let _ = std::fs::remove_dir_all(root); } @@ -7032,8 +7132,8 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_current_schema_requires_lifecycle_authority() { - let root = fs_store_root("current-schema-requires-lifecycle"); + fn fs_store_retains_identity_when_execution_metadata_is_corrupt() { + let root = fs_store_root("corrupt-execution-retains-identity"); let options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), runtime_id: "test-runtime".to_string(), @@ -7046,20 +7146,25 @@ mod tests { .unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); let worker = runtime - .create_worker(task_request("missing lifecycle authority")) + .create_worker(task_request("corrupt execution metadata")) .unwrap(); + let restorable = runtime + .create_worker(task_request("restore persisted Worker identity")) + .unwrap(); + runtime + .stop_worker(&restorable.worker_ref, None) + .expect("stop Worker before Runtime restart"); drop(runtime); - let worker_path = root - .join("workers") - .join(worker.worker_id.to_string()) - .join("worker.json"); - let mut worker_json: serde_json::Value = - serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - worker_json.as_object_mut().unwrap().remove("status"); + let worker_dir = root.join("workers").join(worker.worker_id.to_string()); + let restorable_worker_dir = root.join("workers").join(restorable.worker_id.to_string()); + let execution_path = worker_dir.join("execution.json"); + let mut execution: serde_json::Value = + serde_json::from_slice(&std::fs::read(&execution_path).unwrap()).unwrap(); + execution["restore_intent"] = serde_json::json!(17); std::fs::write( - &worker_path, - serde_json::to_vec_pretty(&worker_json).unwrap(), + &execution_path, + serde_json::to_vec_pretty(&execution).unwrap(), ) .unwrap(); @@ -7068,14 +7173,50 @@ mod tests { Arc::new(TestExecutionBackend::default()), ) .unwrap(); - assert!(restored.list_workers().unwrap().is_empty()); + let workers = restored.list_workers().unwrap(); + assert_eq!(workers.len(), 2); + let worker_summary = workers + .iter() + .find(|summary| summary.worker_id == worker.worker_id) + .expect("retained Worker identity"); + assert!(!worker_summary.execution_metadata_available); + assert!( + !restored + .worker_detail(&worker.worker_ref) + .unwrap() + .execution_metadata_available + ); + assert!(restored.diagnostics().unwrap().iter().any(|diagnostic| { + diagnostic.code == "worker_record_unavailable" + && diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref) + })); + + assert!(matches!( + restored.send_input(&worker.worker_ref, WorkerInput::user("hello")), + Err(RuntimeError::WorkerExecutionUnavailable { .. }) + )); + assert!(matches!( + restored.worker_observation_snapshot(&worker.worker_ref), + Err(RuntimeError::WorkerExecutionUnavailable { .. }) + )); + assert!(restored.delete_worker(&worker.worker_ref).unwrap().deleted); + assert!(!worker_dir.exists()); + + let restored_detail = restored + .restore_worker(&restorable.worker_ref) + .expect("explicit restore reconstructs execution metadata"); + assert!(restored_detail.execution_metadata_available); + restored + .stop_worker(&restorable.worker_ref, None) + .expect("stop restored Worker"); assert!( restored - .diagnostics() + .delete_worker(&restorable.worker_ref) .unwrap() - .iter() - .any(|diagnostic| diagnostic.code == "worker_snapshot_ignored") + .deleted ); + assert!(restored.list_workers().unwrap().is_empty()); + assert!(!restorable_worker_dir.exists()); let _ = std::fs::remove_dir_all(root); } @@ -7100,7 +7241,7 @@ mod tests { runtime.stop_runtime().unwrap(); - let snapshot: serde_json::Value = serde_json::from_slice( + let identity: serde_json::Value = serde_json::from_slice( &std::fs::read( root.join("workers") .join(worker.worker_id.to_string()) @@ -7109,11 +7250,17 @@ mod tests { .unwrap(), ) .unwrap(); - assert_eq!(snapshot["status"], serde_json::json!("idle")); - assert_eq!( - snapshot["execution"]["restore_intent"], - serde_json::json!("automatic") - ); + let execution: serde_json::Value = serde_json::from_slice( + &std::fs::read( + root.join("workers") + .join(worker.worker_id.to_string()) + .join("execution.json"), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(identity["status"], serde_json::json!("idle")); + assert_eq!(execution["restore_intent"], serde_json::json!("automatic")); let _ = std::fs::remove_dir_all(root); } @@ -7173,8 +7320,8 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_migrates_schema_v6_workers_without_losing_automatic_restore() { - let root = fs_store_root("schema-v6-no-generation"); + fn fs_store_migrates_schema_v7_workers_into_identity_and_execution_records() { + let root = fs_store_root("schema-v7-split-records"); let options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), runtime_id: "test-runtime".to_string(), @@ -7187,34 +7334,57 @@ mod tests { .unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); let worker = runtime - .create_worker(task_request("schema v6 worker")) + .create_worker(task_request("schema v7 worker")) .unwrap(); drop(runtime); - let runtime_path = root.join("runtime.json"); - let worker_path = root - .join("workers") - .join(worker.worker_id.to_string()) - .join("worker.json"); - let mut runtime_json: serde_json::Value = - serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap(); - runtime_json["schema_version"] = serde_json::json!(6); - std::fs::write( - &runtime_path, - serde_json::to_vec_pretty(&runtime_json).unwrap(), + let worker_dir = root.join("workers").join(worker.worker_id.to_string()); + let worker_path = worker_dir.join("worker.json"); + let execution_path = worker_dir.join("execution.json"); + let mut runtime_snapshot: serde_json::Value = serde_json::from_slice( + &std::fs::read(root.join("runtime.json")).expect("runtime snapshot"), ) - .unwrap(); - let mut worker_json: serde_json::Value = - serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - worker_json["schema_version"] = serde_json::json!(6); - worker_json["run_generation"] = serde_json::json!(1); - worker_json["execution"]["last_run_generation"] = serde_json::json!(1); - worker_json["execution"]["binding"] = serde_json::json!({"run_generation": 1}); + .expect("runtime snapshot json"); + runtime_snapshot["schema_version"] = serde_json::json!(7); + std::fs::write( + root.join("runtime.json"), + serde_json::to_vec_pretty(&runtime_snapshot).expect("runtime snapshot bytes"), + ) + .expect("write runtime snapshot"); + let mut worker_identity: serde_json::Value = + serde_json::from_slice(&std::fs::read(&worker_path).expect("worker identity")) + .expect("worker identity json"); + let mut worker_execution: serde_json::Value = + serde_json::from_slice(&std::fs::read(&execution_path).expect("worker execution")) + .expect("worker execution json"); + worker_identity["schema_version"] = serde_json::json!(7); + let request = worker_execution + .as_object_mut() + .expect("worker execution object") + .remove("request") + .expect("schema v8 execution request"); + worker_execution + .as_object_mut() + .expect("worker execution object") + .remove("schema_version"); + worker_identity["request"] = request; + worker_identity["execution"] = worker_execution; std::fs::write( &worker_path, - serde_json::to_vec_pretty(&worker_json).unwrap(), + serde_json::to_vec_pretty(&worker_identity).expect("worker record bytes"), ) - .unwrap(); + .expect("write worker record"); + std::fs::remove_file(&execution_path).expect("remove split execution record"); + + let plan = crate::fs_store::FsRuntimeStore::migration_plan(&options) + .expect("read-only migration preflight"); + assert!(plan.migration_required); + let preflight_runtime: serde_json::Value = serde_json::from_slice( + &std::fs::read(root.join("runtime.json")).expect("preflight runtime snapshot"), + ) + .expect("preflight runtime json"); + assert_eq!(preflight_runtime["schema_version"], serde_json::json!(7)); + assert!(!execution_path.exists()); let backend = Arc::new(TestExecutionBackend::default()); let migrated = @@ -7224,18 +7394,21 @@ mod tests { migrated.worker_detail(&worker.worker_ref).unwrap().status, WorkerStatus::Idle ); - let migrated_json: serde_json::Value = + let migrated_identity: serde_json::Value = serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - assert_eq!(migrated_json["schema_version"], serde_json::json!(7)); - assert_eq!(migrated_json["execution"]["binding"], serde_json::json!({})); - assert!(migrated_json.get("run_generation").is_none()); - assert!( - migrated_json["execution"] - .get("last_run_generation") - .is_none() - ); + let migrated_execution: serde_json::Value = + serde_json::from_slice(&std::fs::read(&execution_path).unwrap()).unwrap(); + assert_eq!(migrated_identity["schema_version"], serde_json::json!(8)); + assert!(migrated_identity.get("request").is_none()); + assert!(migrated_identity.get("execution").is_none()); + assert_eq!(migrated_execution["schema_version"], serde_json::json!(8)); assert_eq!( - migrated_json["execution"]["restore_intent"], + migrated_execution["request"]["worker_id"], + worker.worker_id.to_string() + ); + assert_eq!(migrated_execution["binding"], serde_json::json!({})); + assert_eq!( + migrated_execution["restore_intent"], serde_json::json!("automatic") ); @@ -7342,7 +7515,7 @@ mod tests { .unwrap(); missing_runtime.store_config_bundle(test_bundle()).unwrap(); missing_runtime - .create_worker(task_request("missing worker snapshot")) + .create_worker(task_request("missing Worker identity")) .unwrap(); let missing_store = runtime_store(&missing_runtime); let mut worker_dirs = std::fs::read_dir(missing_store.runtime_dir().join("workers")) @@ -7358,14 +7531,14 @@ mod tests { runtime_id: "test-runtime".to_string(), display_name: None, }) - .expect("invalid worker snapshot should not make runtime store unreadable"); + .expect("invalid Worker identity should not make Runtime store unreadable"); assert!(loaded.list_workers().unwrap().is_empty()); assert!( loaded .diagnostics() .unwrap() .iter() - .any(|diagnostic| diagnostic.code == "worker_snapshot_ignored") + .any(|diagnostic| diagnostic.code == "worker_record_unavailable") ); let _ = std::fs::remove_dir_all(missing_root); } diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 28f7249f..63b38857 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -2129,7 +2129,11 @@ impl EmbeddedWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: summary.workspace_id.clone(), }, - state: embedded_worker_status_label(summary.status).to_string(), + state: embedded_worker_state_label( + summary.status, + summary.execution_metadata_available, + ) + .to_string(), worker_state: summary.worker_state.clone(), last_seen_at: None, pinned: false, @@ -2139,11 +2143,14 @@ impl EmbeddedWorkerRuntime { display_hint: "backend-internal worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: self.can_stop_embedded_worker(summary.status), + can_stop: summary.execution_metadata_available + && self.can_stop_embedded_worker(summary.status), can_spawn_followup: false, }, working_directory: summary.working_directory.map(|status| status.summary), - diagnostics: embedded_worker_projection_diagnostics(), + diagnostics: embedded_worker_projection_diagnostics( + summary.execution_metadata_available, + ), } } @@ -2169,7 +2176,8 @@ impl EmbeddedWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: detail.workspace_id.clone(), }, - state: embedded_worker_status_label(detail.status).to_string(), + state: embedded_worker_state_label(detail.status, detail.execution_metadata_available) + .to_string(), worker_state: detail.worker_state.clone(), last_seen_at: None, pinned: false, @@ -2179,11 +2187,14 @@ impl EmbeddedWorkerRuntime { display_hint: "backend-internal worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: self.can_stop_embedded_worker(detail.status), + can_stop: detail.execution_metadata_available + && self.can_stop_embedded_worker(detail.status), can_spawn_followup: false, }, working_directory: detail.working_directory.map(|status| status.summary), - diagnostics: embedded_worker_projection_diagnostics(), + diagnostics: embedded_worker_projection_diagnostics( + detail.execution_metadata_available, + ), } } } @@ -3914,7 +3925,11 @@ impl RemoteWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: summary.workspace_id.clone(), }, - state: embedded_worker_status_label(summary.status).to_string(), + state: embedded_worker_state_label( + summary.status, + summary.execution_metadata_available, + ) + .to_string(), worker_state: summary.worker_state.clone(), last_seen_at: None, pinned: false, @@ -3924,15 +3939,12 @@ impl RemoteWorkerRuntime { display_hint: "Backend-proxied remote worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: runtime_worker_can_stop(true, summary.status), + can_stop: summary.execution_metadata_available + && runtime_worker_can_stop(true, summary.status), can_spawn_followup: false, }, working_directory: summary.working_directory.map(|status| status.summary), - diagnostics: vec![diagnostic( - "remote_runtime_projection", - DiagnosticSeverity::Info, - "Remote Worker identity is projected only as runtime_id plus worker_id; endpoint and credentials remain backend-private".to_string(), - )], + diagnostics: remote_worker_projection_diagnostics(summary.execution_metadata_available), } } @@ -3958,7 +3970,8 @@ impl RemoteWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: detail.workspace_id.clone(), }, - state: embedded_worker_status_label(detail.status).to_string(), + state: embedded_worker_state_label(detail.status, detail.execution_metadata_available) + .to_string(), worker_state: detail.worker_state.clone(), last_seen_at: None, pinned: false, @@ -3968,15 +3981,12 @@ impl RemoteWorkerRuntime { display_hint: "Backend-proxied remote worker-runtime Worker".to_string(), }, capabilities: WorkerCapabilitySummary { - can_stop: runtime_worker_can_stop(true, detail.status), + can_stop: detail.execution_metadata_available + && runtime_worker_can_stop(true, detail.status), can_spawn_followup: false, }, working_directory: detail.working_directory.map(|status| status.summary), - diagnostics: vec![diagnostic( - "remote_runtime_projection", - DiagnosticSeverity::Info, - "Remote Worker identity is projected only as runtime_id plus worker_id; endpoint and credentials remain backend-private".to_string(), - )], + diagnostics: remote_worker_projection_diagnostics(detail.execution_metadata_available), } } @@ -4696,12 +4706,49 @@ fn embedded_worker_status_label(status: EmbeddedWorkerStatus) -> &'static str { } } -fn embedded_worker_projection_diagnostics() -> Vec { - vec![diagnostic( +fn embedded_worker_state_label( + status: EmbeddedWorkerStatus, + execution_metadata_available: bool, +) -> &'static str { + if !execution_metadata_available { + return "execution_unavailable"; + } + embedded_worker_status_label(status) +} + +fn execution_metadata_diagnostic(execution_metadata_available: bool) -> Option { + (!execution_metadata_available).then(|| { + diagnostic( + "worker_execution_unavailable", + DiagnosticSeverity::Error, + "Persisted Worker identity is available, but execution metadata is unavailable" + .to_string(), + ) + }) +} + +fn embedded_worker_projection_diagnostics( + execution_metadata_available: bool, +) -> Vec { + let mut diagnostics = 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(), - )] + )]; + diagnostics.extend(execution_metadata_diagnostic(execution_metadata_available)); + diagnostics +} + +fn remote_worker_projection_diagnostics( + execution_metadata_available: bool, +) -> Vec { + let mut diagnostics = vec![diagnostic( + "remote_runtime_projection", + DiagnosticSeverity::Info, + "Remote Worker identity is projected only as runtime_id plus worker_id; endpoint and credentials remain backend-private".to_string(), + )]; + diagnostics.extend(execution_metadata_diagnostic(execution_metadata_available)); + diagnostics } fn spawn_config_bundle_ref(request: &WorkerSpawnRequest) -> Option { @@ -6787,11 +6834,14 @@ mod tests { } #[test] - fn remote_runtime_projection_uses_canonical_worker_status_for_stop_capability() { - let worker_ids = (1..=4) + fn remote_runtime_projection_uses_execution_availability_and_canonical_status() { + let worker_ids = (1..=5) .map(|value| EmbeddedWorkerId::from_legacy_u64(value).to_string()) .collect::>(); let worker_id = worker_ids[0].clone(); + let mut execution_unavailable = + worker_json_with_status("remote:primary", &worker_ids[4], "running"); + execution_unavailable["execution_metadata_available"] = serde_json::json!(false); let (base_url, server) = serve_mock_http(vec![ mock_response( "GET", @@ -6803,7 +6853,8 @@ mod tests { worker_json_with_status("remote:primary", &worker_ids[0], "stopped"), worker_json_with_status("remote:primary", &worker_ids[1], "running"), worker_json_with_status("remote:primary", &worker_ids[2], "paused"), - worker_json_with_status("remote:primary", &worker_ids[3], "idle") + worker_json_with_status("remote:primary", &worker_ids[3], "idle"), + execution_unavailable ] }) .to_string(), @@ -6838,15 +6889,23 @@ mod tests { )]); let workers = registry.list_workers(10); - assert_eq!(workers.items.len(), 4); + assert_eq!(workers.items.len(), 5); 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!(!workers.items[4].capabilities.can_stop); assert_eq!(workers.items[0].state, "stopped"); assert_eq!(workers.items[1].state, "running"); assert_eq!(workers.items[2].state, "paused"); assert_eq!(workers.items[3].state, "idle"); + assert_eq!(workers.items[4].state, "execution_unavailable"); + assert!( + workers.items[4] + .diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "worker_execution_unavailable") + ); let stopped_detail = registry .worker(&RuntimeWorkerRef::new("remote:primary", &worker_id)) @@ -7119,6 +7178,7 @@ mod tests { "runtime_id": runtime_id, "worker_id": worker_id, "status": status, + "execution_metadata_available": true, "intent": { "kind": "role", "role": "coder", "purpose": "remote test" }, "profile": { "kind": "builtin", "value": "coder" }, "profile_source": {