Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

refactor: separate Worker identity from execution state

Browse Source
This commit is contained in:
Hare
2026-09-17 03:25:40 +09:00
parent bc0342b03e
commit c4622e9e3a
5 changed files with 845 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+16 -6
View File
@@ -266,11 +266,11 @@ pub struct CreateWorkerRequest {
pub memory_settings: Option<manifest::WorkspaceMemorySettingsSnapshot>, pub memory_settings: Option<manifest::WorkspaceMemorySettingsSnapshot>,
} }
/// 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 /// This is not proof that the current Runtime process holds a live execution handle. Run
/// particular, cancellation returns a Worker to `Idle`; it is not a lifecycle /// termination details remain separate Worker protocol state; in particular, cancellation
/// state of its own. /// returns a Worker to `Idle` and is not a lifecycle state of its own.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum WorkerStatus { pub enum WorkerStatus {
@@ -293,12 +293,17 @@ pub(crate) enum WorkerRestoreIntent {
Explicit, Explicit,
} }
/// Lightweight catalog row. /// Lightweight persisted Worker identity projection.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerSummary { pub struct WorkerSummary {
pub worker_ref: WorkerRef, pub worker_ref: WorkerRef,
pub worker_id: WorkerId, pub worker_id: WorkerId,
pub status: WorkerStatus, 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<u64>,
/// 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")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_state: Option<protocol::WorkerStateSnapshot>, pub worker_state: Option<protocol::WorkerStateSnapshot>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -313,12 +318,17 @@ pub struct WorkerSummary {
pub config_bundle: Option<ConfigBundleRef>, pub config_bundle: Option<ConfigBundleRef>,
} }
/// Full Worker catalog/lifecycle detail. /// Full persisted Worker identity and lifecycle detail.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkerDetail { pub struct WorkerDetail {
pub worker_ref: WorkerRef, pub worker_ref: WorkerRef,
pub worker_id: WorkerId, pub worker_id: WorkerId,
pub status: WorkerStatus, 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<u64>,
/// 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")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub worker_state: Option<protocol::WorkerStateSnapshot>, pub worker_state: Option<protocol::WorkerStateSnapshot>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
crates/worker-runtime/src/fs_store.rs
+427 -198
View File
@@ -1,5 +1,6 @@
use crate::catalog::{ use crate::catalog::{
CreateWorkerRequest, WorkerRestoreIntent, WorkerStatus, WorkingDirectoryStatus, ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerRestoreIntent, WorkerStatus,
WorkingDirectoryStatus,
}; };
use crate::config_bundle::ConfigBundle; use crate::config_bundle::ConfigBundle;
use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
@@ -8,22 +9,26 @@ use crate::identity::{
LegacyWorkerIdentityMapping, WorkerId, WorkerRef, legacy_worker_identity_mapping_digest, LegacyWorkerIdentityMapping, WorkerId, WorkerRef, legacy_worker_identity_mapping_digest,
}; };
use crate::management::{RuntimeBackendKind, RuntimeStatus}; use crate::management::{RuntimeBackendKind, RuntimeStatus};
use crate::profile_archive::ProfileSourceArchiveRef;
use fs4::fs_std::FileExt; use fs4::fs_std::FileExt;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions}; use std::fs::{self, File, OpenOptions};
use std::io::{BufReader, Write}; use std::io::{BufReader, Read, Write};
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::Arc; use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
const SCHEMA_VERSION: u32 = 7; const SCHEMA_VERSION: u32 = 8;
const PREVIOUS_SCHEMA_VERSION: u32 = 6; const PREVIOUS_SCHEMA_VERSION: u32 = 7;
const RUNTIME_FILE: &str = "runtime.json"; const RUNTIME_FILE: &str = "runtime.json";
const WORKERS_DIR: &str = "workers"; const WORKERS_DIR: &str = "workers";
const WORKER_FILE: &str = "worker.json"; const WORKER_FILE: &str = "worker.json";
const WORKER_EXECUTION_FILE: &str = "execution.json";
const WORKER_METADATA_FILE: &str = "metadata.json"; const WORKER_METADATA_FILE: &str = "metadata.json";
const LEGACY_OBSERVATIONS_FILE: &str = "observations.jsonl"; 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); static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
@@ -165,27 +170,34 @@ impl FsRuntimeStore {
atomic_write_json(&self.runtime_path(), &snapshot, "write runtime snapshot") atomic_write_json(&self.runtime_path(), &snapshot, "write runtime snapshot")
} }
pub(crate) fn write_worker_snapshot( pub(crate) fn write_worker_record(
&self, &self,
worker: &PersistedWorkerRecord, worker: &PersistedWorkerRecord,
) -> Result<(), RuntimeError> { ) -> Result<(), RuntimeError> {
self.ensure_worker_ref(&worker.worker_ref)?; self.ensure_worker_ref(&worker.worker_ref)?;
let worker_dir = self.worker_dir(&worker.worker_id); let worker_dir = self.worker_dir(&worker.worker_id);
fs::create_dir_all(&worker_dir).map_err(|source| RuntimeError::StoreIo { fs::create_dir_all(&worker_dir).map_err(|source| RuntimeError::StoreIo {
operation: "create worker store", operation: "create Worker store",
path: worker_dir.clone(), path: worker_dir.clone(),
source, source,
})?; })?;
atomic_write_json( atomic_write_json(
&worker_dir.join(WORKER_FILE), &worker_dir.join(WORKER_FILE),
&WorkerSnapshot::from_persisted(worker), &WorkerIdentityRecord::from_persisted(worker),
"write worker snapshot", "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); remove_legacy_observations(&worker_dir);
Ok(()) 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); let worker_dir = self.worker_dir(worker_id);
if !worker_dir.exists() { if !worker_dir.exists() {
return Ok(()); return Ok(());
@@ -232,44 +244,74 @@ impl FsRuntimeStore {
})?; })?;
worker_dirs.sort_by_key(|entry| entry.path()); 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 { for entry in worker_dirs {
let path = entry.path(); 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( record_worker_load_diagnostic(
&mut snapshot, &mut snapshot,
None, None,
"ignored invalid worker store entry while loading runtime store", "ignored invalid Worker store entry while loading Runtime store",
); );
continue; continue;
} }
let worker_snapshot_path = path.join(WORKER_FILE); let identity_path = path.join(WORKER_FILE);
let worker_snapshot: WorkerSnapshot = let identity: WorkerIdentityRecord =
match read_json(&worker_snapshot_path, "read worker snapshot") { match read_bounded_json(&identity_path, "read Worker identity") {
Ok(snapshot) => snapshot, Ok(identity) => identity,
Err(_error) => { Err(_error) => {
record_worker_load_diagnostic( record_worker_load_diagnostic(
&mut snapshot, &mut snapshot,
None, None,
"ignored corrupt worker snapshot while loading runtime store", "ignored corrupt Worker identity while loading Runtime store",
); );
continue; continue;
} }
}; };
if worker_snapshot.validate(&worker_snapshot_path).is_err() { if identity.validate(&identity_path).is_err() {
record_worker_load_diagnostic( record_worker_load_diagnostic(
&mut snapshot, &mut snapshot,
Some(worker_snapshot.worker_ref.clone()), Some(identity.worker_ref.clone()),
"ignored invalid worker snapshot while loading runtime store", "ignored invalid Worker identity while loading Runtime store",
); );
continue; continue;
} }
let execution_path = path.join(WORKER_EXECUTION_FILE);
let execution_state = read_bounded_json::<WorkerExecutionRecord>(
&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); 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() { if workers.insert(worker.worker_id.clone(), worker).is_some() {
record_worker_load_diagnostic( record_worker_load_diagnostic(
&mut snapshot, &mut snapshot,
None, 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)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub(crate) struct PersistedWorkerExecution { pub(crate) struct PersistedWorkerExecution {
pub(crate) request: CreateWorkerRequest,
pub(crate) binding: Option<PersistedWorkerExecutionBinding>, pub(crate) binding: Option<PersistedWorkerExecutionBinding>,
pub(crate) restore_intent: WorkerRestoreIntent, pub(crate) restore_intent: WorkerRestoreIntent,
} }
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) enum PersistedWorkerExecutionState {
Available(PersistedWorkerExecution),
Unavailable,
}
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
pub(crate) struct PersistedWorkerRecord { pub(crate) struct PersistedWorkerRecord {
pub(crate) worker_ref: WorkerRef, pub(crate) worker_ref: WorkerRef,
pub(crate) worker_id: WorkerId, pub(crate) worker_id: WorkerId,
pub(crate) request: CreateWorkerRequest, pub(crate) profile: ProfileSelector,
pub(crate) display_name: Option<String>,
pub(crate) profile_source: ProfileSourceArchiveRef,
pub(crate) config_bundle: Option<ConfigBundleRef>,
pub(crate) created_at_ms: Option<u64>,
pub(crate) status: WorkerStatus, pub(crate) status: WorkerStatus,
pub(crate) execution: PersistedWorkerExecution, pub(crate) execution_state: PersistedWorkerExecutionState,
pub(crate) workspace_id: Option<String>, pub(crate) workspace_id: Option<String>,
pub(crate) working_directory: Option<WorkingDirectoryStatus>, pub(crate) working_directory: Option<WorkingDirectoryStatus>,
} }
@@ -462,8 +515,8 @@ fn plan_runtime_store_migration(
format!("Runtime store schema version {schema_version} is out of range"), format!("Runtime store schema version {schema_version} is out of range"),
) )
})?; })?;
let staging = migration_sibling(root, "schema-v7-staging")?; let staging = migration_sibling(root, "schema-v8-staging")?;
let backup = migration_sibling(root, "pre-schema-v7-backup")?; let backup = migration_sibling(root, "pre-schema-v8-backup")?;
if staging.exists() || backup.exists() { if staging.exists() || backup.exists() {
return Err(runtime_store_corrupt( return Err(runtime_store_corrupt(
root, root,
@@ -523,14 +576,14 @@ fn plan_runtime_store_migration(
if !snapshot_path if !snapshot_path
.try_exists() .try_exists()
.map_err(|source| RuntimeError::StoreIo { .map_err(|source| RuntimeError::StoreIo {
operation: "inspect Worker snapshot", operation: "inspect Worker record",
path: snapshot_path.clone(), path: snapshot_path.clone(),
source, source,
})? })?
{ {
continue; 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 (worker_id, workspace_id, legacy_mapping) = if current_schema_version == 1 {
let legacy_worker_id = name.parse::<u64>().map_err(|_| { let legacy_worker_id = name.parse::<u64>().map_err(|_| {
runtime_store_corrupt( runtime_store_corrupt(
@@ -545,7 +598,7 @@ fn plan_runtime_store_migration(
.ok_or_else(|| { .ok_or_else(|| {
runtime_store_corrupt( runtime_store_corrupt(
&snapshot_path, &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(), .to_string(),
) )
})? })?
@@ -595,33 +648,24 @@ fn plan_runtime_store_migration(
let mut migrated_worker_aggregate_count = 0; let mut migrated_worker_aggregate_count = 0;
for worker in &mut planned { for worker in &mut planned {
let snapshot_path = worker.source_dir.join(WORKER_FILE); 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( let migrated = migrate_worker_document(
snapshot, snapshot,
current_schema_version, current_schema_version,
worker.legacy_mapping.as_ref(), worker.legacy_mapping.as_ref(),
&snapshot_path, &snapshot_path,
)?; )?;
let snapshot = validate_migrated_worker_document(&migrated, &snapshot_path)?; let identity = validate_migrated_worker_documents(&migrated, &snapshot_path)?;
if snapshot.worker_id != worker.worker_id { if identity.worker_id != worker.worker_id {
return Err(runtime_store_corrupt( return Err(runtime_store_corrupt(
&snapshot_path, &snapshot_path,
format!( format!(
"Worker snapshot id {} does not match directory identity {}", "Worker identity {} does not match directory identity {}",
snapshot.worker_id, worker.worker_id identity.worker_id, worker.worker_id
), ),
)); ));
} }
worker.workspace_id = worker worker.workspace_id = worker.workspace_id.clone().or(identity.workspace_id);
.workspace_id
.clone()
.or(snapshot.workspace_id)
.or_else(|| {
snapshot
.request
.workspace_api
.map(|workspace_api| workspace_api.workspace_id)
});
let metadata_path = worker.source_dir.join(WORKER_METADATA_FILE); let metadata_path = worker.source_dir.join(WORKER_METADATA_FILE);
if metadata_path.is_file() { if metadata_path.is_file() {
@@ -655,123 +699,107 @@ struct DiagnosticWorkerRefMigrationCounts {
cleared: usize, cleared: usize,
} }
#[derive(Clone, Debug)]
struct MigratedWorkerDocuments {
identity: serde_json::Value,
execution: serde_json::Value,
}
fn migrate_worker_document( fn migrate_worker_document(
mut document: serde_json::Value, mut document: serde_json::Value,
source_schema_version: u32, source_schema_version: u32,
_mapping: Option<&LegacyWorkerIdentityMapping>, _mapping: Option<&LegacyWorkerIdentityMapping>,
snapshot_path: &Path, identity_path: &Path,
) -> Result<serde_json::Value, RuntimeError> { ) -> Result<MigratedWorkerDocuments, RuntimeError> {
if source_schema_version != PREVIOUS_SCHEMA_VERSION { if source_schema_version != PREVIOUS_SCHEMA_VERSION {
return Err(runtime_store_corrupt( return Err(runtime_store_corrupt(
snapshot_path, identity_path,
format!( 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(|| { let object = document.as_object_mut().ok_or_else(|| {
runtime_store_corrupt( runtime_store_corrupt(identity_path, "Worker record must be an object".to_string())
snapshot_path,
"Worker snapshot must be an object".to_string(),
)
})?; })?;
if let Some(run_generation) = object.remove("run_generation") let mut execution = object
&& run_generation.as_u64().is_none() .remove("execution")
{ .and_then(|value| value.as_object().cloned())
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)
.ok_or_else(|| { .ok_or_else(|| {
runtime_store_corrupt( runtime_store_corrupt(
snapshot_path, identity_path,
"Worker snapshot execution must be an object".to_string(), "Worker record execution must be an object".to_string(),
) )
})?; })?;
let last_run_generation = execution let request_value = object.remove("request").ok_or_else(|| {
.remove("last_run_generation")
.and_then(|value| value.as_u64())
.ok_or_else(|| {
runtime_store_corrupt( runtime_store_corrupt(
snapshot_path, identity_path,
"Worker execution last_run_generation must be an unsigned integer".to_string(), "Worker record request must be present".to_string(),
) )
})?; })?;
let binding = execution.get_mut("binding").ok_or_else(|| { let request: CreateWorkerRequest =
serde_json::from_value(request_value.clone()).map_err(|error| {
runtime_store_corrupt( runtime_store_corrupt(
snapshot_path, identity_path,
"Worker execution is missing binding".to_string(), format!("decode Worker execution request: {error}"),
) )
})?; })?;
if let Some(binding_object) = binding.as_object_mut() { object.insert(
let binding_run_generation = binding_object "profile".to_string(),
.remove("run_generation") serde_json::to_value(&request.profile).expect("Profile selector serializes"),
.and_then(|value| value.as_u64()) );
.ok_or_else(|| { object.insert(
runtime_store_corrupt( "display_name".to_string(),
snapshot_path, serde_json::to_value(&request.display_name).expect("display name serializes"),
"Worker execution binding run_generation must be an unsigned integer" );
.to_string(), object.insert(
) "profile_source".to_string(),
})?; serde_json::to_value(request.profile_source.reference())
if binding_run_generation != last_run_generation { .expect("Profile source reference serializes"),
return Err(runtime_store_corrupt( );
snapshot_path, object.insert(
format!( "config_bundle".to_string(),
"execution binding run_generation {binding_run_generation} does not match last_run_generation {last_run_generation}" serde_json::to_value(&request.config_bundle).expect("config bundle serializes"),
), );
)); execution.insert("request".to_string(), request_value);
} execution.insert(
if !binding_object.is_empty() { "schema_version".to_string(),
return Err(runtime_store_corrupt( serde_json::Value::from(SCHEMA_VERSION),
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(),
));
}
object.insert( object.insert(
"schema_version".to_string(), "schema_version".to_string(),
serde_json::Value::from(SCHEMA_VERSION), 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( fn validate_migrated_worker_documents(
document: &serde_json::Value, documents: &MigratedWorkerDocuments,
snapshot_path: &Path, identity_path: &Path,
) -> Result<WorkerSnapshot, RuntimeError> { ) -> Result<WorkerIdentityRecord, RuntimeError> {
let snapshot: WorkerSnapshot = serde_json::from_value(document.clone()).map_err(|error| { let identity: WorkerIdentityRecord = serde_json::from_value(documents.identity.clone())
.map_err(|error| {
runtime_store_corrupt( runtime_store_corrupt(
snapshot_path, identity_path,
format!("decode migrated Worker snapshot: {error}"), format!("decode migrated Worker identity: {error}"),
) )
})?; })?;
snapshot.validate(snapshot_path)?; identity.validate(identity_path)?;
Ok(snapshot) 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 { fn runtime_worker_name(worker_id: WorkerId) -> String {
@@ -1125,8 +1153,8 @@ fn migrate_runtime_store(
if !plan.migration_required { if !plan.migration_required {
return Ok(plan); return Ok(plan);
} }
let staging = migration_sibling(root, "schema-v7-staging")?; let staging = migration_sibling(root, "schema-v8-staging")?;
let backup = migration_sibling(root, "pre-schema-v7-backup")?; let backup = migration_sibling(root, "pre-schema-v8-backup")?;
if staging.exists() || backup.exists() { if staging.exists() || backup.exists() {
return Err(runtime_store_corrupt( return Err(runtime_store_corrupt(
root, root,
@@ -1208,12 +1236,12 @@ fn migrate_runtime_store_in_place(
runtime_store_corrupt( runtime_store_corrupt(
&source_snapshot_path, &source_snapshot_path,
format!( format!(
"decode Worker snapshot {}: {error}", "decode Worker record {}: {error}",
source_snapshot_path.display() source_snapshot_path.display()
), ),
) )
})?; })?;
let snapshot = migrate_worker_document( let documents = migrate_worker_document(
snapshot, snapshot,
plan.current_schema_version, plan.current_schema_version,
planned_worker.legacy_mapping.as_ref(), planned_worker.legacy_mapping.as_ref(),
@@ -1242,12 +1270,17 @@ fn migrate_runtime_store_in_place(
fs::rename(source_dir, &migrated_dir) fs::rename(source_dir, &migrated_dir)
.map_err(|error| runtime_io_error("rename", source_dir, error))?; .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( atomic_write_json(
&migrated_snapshot_path, &migrated_identity_path,
&snapshot, &documents.identity,
"migrate Worker identity", "migrate Worker identity",
)?; )?;
atomic_write_json(
&migrated_dir.join(WORKER_EXECUTION_FILE),
&documents.execution,
"migrate Worker execution record",
)?;
if let Some(metadata) = metadata { if let Some(metadata) = metadata {
atomic_write_json( atomic_write_json(
&migrated_dir.join(WORKER_METADATA_FILE), &migrated_dir.join(WORKER_METADATA_FILE),
@@ -1296,7 +1329,7 @@ fn record_worker_load_diagnostic(
id, id,
worker_ref, worker_ref,
severity: DiagnosticSeverity::Warning, severity: DiagnosticSeverity::Warning,
code: "worker_snapshot_ignored".to_string(), code: "worker_record_unavailable".to_string(),
message: message.into(), message: message.into(),
}); });
} }
@@ -1352,28 +1385,46 @@ impl RuntimeSnapshot {
} }
#[derive(Clone, Debug, Serialize, Deserialize)] #[derive(Clone, Debug, Serialize, Deserialize)]
struct WorkerSnapshot { #[serde(deny_unknown_fields)]
struct WorkerIdentityRecord {
schema_version: u32, schema_version: u32,
worker_ref: WorkerRef, worker_ref: WorkerRef,
worker_id: WorkerId, worker_id: WorkerId,
request: CreateWorkerRequest, profile: ProfileSelector,
display_name: Option<String>,
profile_source: ProfileSourceArchiveRef,
config_bundle: Option<ConfigBundleRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
created_at_ms: Option<u64>,
status: WorkerStatus, status: WorkerStatus,
execution: PersistedWorkerExecution,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
workspace_id: Option<String>, workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
working_directory: Option<WorkingDirectoryStatus>, working_directory: Option<WorkingDirectoryStatus>,
} }
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<PersistedWorkerExecutionBinding>,
restore_intent: WorkerRestoreIntent,
}
impl WorkerIdentityRecord {
fn from_persisted(worker: &PersistedWorkerRecord) -> Self { fn from_persisted(worker: &PersistedWorkerRecord) -> Self {
Self { Self {
schema_version: SCHEMA_VERSION, schema_version: SCHEMA_VERSION,
worker_ref: worker.worker_ref.clone(), worker_ref: worker.worker_ref.clone(),
worker_id: worker.worker_id.clone(), worker_id: worker.worker_id,
request: worker.request.clone(), 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, status: worker.status,
execution: worker.execution.clone(),
workspace_id: worker.workspace_id.clone(), workspace_id: worker.workspace_id.clone(),
working_directory: worker.working_directory.clone(), working_directory: worker.working_directory.clone(),
} }
@@ -1382,7 +1433,7 @@ impl WorkerSnapshot {
fn validate(&self, path: &Path) -> Result<(), RuntimeError> { fn validate(&self, path: &Path) -> Result<(), RuntimeError> {
if self.schema_version != SCHEMA_VERSION { if self.schema_version != SCHEMA_VERSION {
return Err(RuntimeError::StoreCorrupt { return Err(RuntimeError::StoreCorrupt {
operation: "read worker snapshot", operation: "read Worker identity",
path: path.to_path_buf(), path: path.to_path_buf(),
message: format!( message: format!(
"unsupported schema version {}, expected {}", "unsupported schema version {}, expected {}",
@@ -1392,7 +1443,7 @@ impl WorkerSnapshot {
} }
if self.worker_ref.worker_id != self.worker_id { if self.worker_ref.worker_id != self.worker_id {
return Err(RuntimeError::StoreCorrupt { return Err(RuntimeError::StoreCorrupt {
operation: "read worker snapshot", operation: "read Worker identity",
path: path.to_path_buf(), path: path.to_path_buf(),
message: format!( message: format!(
"worker_ref id {} does not match worker_id {}", "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();
(status, WorkerRestoreIntent::Automatic) if status.is_active() => { if path
if self.execution.binding.is_none() { .parent()
.and_then(Path::file_name)
.and_then(|name| name.to_str())
!= Some(expected_name.as_str())
{
return Err(RuntimeError::StoreCorrupt { return Err(RuntimeError::StoreCorrupt {
operation: "read worker snapshot", 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<PersistedWorkerExecution, RuntimeError> {
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.binding.is_none() {
return Err(RuntimeError::StoreCorrupt {
operation: "read Worker execution record",
path: path.to_path_buf(), path: path.to_path_buf(),
message: "automatic restore intent requires an execution binding" message: "automatic restore intent requires an execution binding"
.to_string(), .to_string(),
@@ -1414,35 +1548,80 @@ impl WorkerSnapshot {
(WorkerStatus::Stopped, WorkerRestoreIntent::Explicit) => {} (WorkerStatus::Stopped, WorkerRestoreIntent::Explicit) => {}
_ => { _ => {
return Err(RuntimeError::StoreCorrupt { return Err(RuntimeError::StoreCorrupt {
operation: "read worker snapshot", operation: "read Worker execution record",
path: path.to_path_buf(), path: path.to_path_buf(),
message: format!( message: format!(
"worker status {:?} conflicts with restore intent {:?}", "Worker status {:?} conflicts with restore intent {:?}",
self.status, self.execution.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 { fn read_bounded_json<T>(path: &Path, operation: &'static str) -> Result<T, RuntimeError>
let workspace_id = self.workspace_id.or_else(|| { where
self.request T: for<'de> Deserialize<'de>,
.workspace_api {
.as_ref() let metadata = fs::symlink_metadata(path).map_err(|source| match source.kind() {
.map(|workspace_api| workspace_api.workspace_id.clone()) 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<T>(path: &Path, operation: &'static str) -> Result<T, RuntimeError> fn read_json<T>(path: &Path, operation: &'static str) -> Result<T, RuntimeError>
@@ -1668,50 +1847,100 @@ mod tests {
assert_eq!(plan.worker_count, 0); assert_eq!(plan.worker_count, 0);
} }
#[test] fn schema_v7_worker_document(worker_id: WorkerId) -> serde_json::Value {
fn schema_v6_worker_migration_removes_generation_and_preserves_active_restore() { serde_json::json!({
let path = Path::new("worker.json");
let source = serde_json::json!({
"schema_version": PREVIOUS_SCHEMA_VERSION, "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", "status": "running",
"execution": { "execution": {
"last_run_generation": 7, "binding": {},
"binding": { "run_generation": 7 },
"restore_intent": "automatic" "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] #[test]
fn schema_v6_worker_migration_rejects_mismatched_generation_state() { fn bounded_worker_record_read_rejects_oversize_before_parse() {
let path = Path::new("worker.json"); let root = tempfile::tempdir().unwrap();
let source = serde_json::json!({ let path = root.path().join(WORKER_FILE);
"schema_version": PREVIOUS_SCHEMA_VERSION, File::create(&path)
"execution": { .unwrap()
"last_run_generation": 7, .set_len(MAX_WORKER_RECORD_BYTES + 1)
"binding": { "run_generation": 6 }, .unwrap();
"restore_intent": "automatic"
}
});
let error = let error =
migrate_worker_document(source, PREVIOUS_SCHEMA_VERSION, None, path).unwrap_err(); read_bounded_json::<serde_json::Value>(&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!( assert!(
error error
.to_string() .to_string()
.contains("does not match last_run_generation") .contains("decode migrated Worker execution record")
); );
} }
} }
crates/worker-runtime/src/main.rs
+6 -6
View File
@@ -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. // 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 // A Runtime service that cannot create and restore Workers is not available to the
// Workspace Browser. // Workspace Browser.
use std::collections::VecDeque; use std::collections::VecDeque;
@@ -1152,7 +1152,7 @@ fn usage() -> &'static str {
yoi-runtime migrate --dry-run [--runtime-id <ID>] [OPTIONS] yoi-runtime migrate --dry-run [--runtime-id <ID>] [OPTIONS]
Starts a worker-backed Runtime REST command API for a trusted backend/proxy. 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: Options:
--bind <ADDR> Bind socket address (default: 127.0.0.1:38800) --bind <ADDR> Bind socket address (default: 127.0.0.1:38800)
@@ -1280,7 +1280,7 @@ mod tests {
std::fs::write( std::fs::write(
root.join("runtime.json"), root.join("runtime.json"),
serde_json::to_vec_pretty(&serde_json::json!({ serde_json::to_vec_pretty(&serde_json::json!({
"schema_version": 6, "schema_version": 7,
"display_name": "local", "display_name": "local",
"backend": "fs_store", "backend": "fs_store",
"status": "running", "status": "running",
@@ -1319,7 +1319,7 @@ mod tests {
std::fs::write( std::fs::write(
root.join("runtime.json"), root.join("runtime.json"),
serde_json::to_vec_pretty(&serde_json::json!({ serde_json::to_vec_pretty(&serde_json::json!({
"schema_version": 6, "schema_version": 7,
"display_name": "local", "display_name": "local",
"backend": "fs_store", "backend": "fs_store",
"status": 3, "status": 3,
crates/worker-runtime/src/runtime.rs
+302 -129
View File
@@ -20,7 +20,7 @@ use crate::execution::{
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
use crate::fs_store::{ use crate::fs_store::{
FsRuntimeStore, FsRuntimeStoreOptions, PersistedRuntimeState, PersistedWorkerExecution, FsRuntimeStore, FsRuntimeStoreOptions, PersistedRuntimeState, PersistedWorkerExecution,
PersistedWorkerExecutionBinding, PersistedWorkerRecord, PersistedWorkerExecutionBinding, PersistedWorkerExecutionState, PersistedWorkerRecord,
}; };
use crate::identity::{WorkerId, WorkerRef}; use crate::identity::{WorkerId, WorkerRef};
use crate::interaction::{WorkerInput, WorkerInputKind, WorkerInteractionAck}; use crate::interaction::{WorkerInput, WorkerInputKind, WorkerInteractionAck};
@@ -29,6 +29,7 @@ use crate::management::{
}; };
#[cfg(feature = "ws-server")] #[cfg(feature = "ws-server")]
use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; use crate::observation::{WorkerObservationCursor, WorkerObservationEvent};
use crate::profile_archive::ProfileSourceArchiveRef;
use crate::resource::{ use crate::resource::{
BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceKind, BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceKind,
REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret, REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret,
@@ -289,6 +290,9 @@ impl Runtime {
let mut active_worker_count = 0; let mut active_worker_count = 0;
let mut stopped_worker_count = 0; let mut stopped_worker_count = 0;
for worker in state.workers.values() { for worker in state.workers.values() {
if !worker.execution_metadata_available {
continue;
}
match worker.status { match worker.status {
WorkerStatus::Idle | WorkerStatus::Running | WorkerStatus::Paused => { WorkerStatus::Idle | WorkerStatus::Running | WorkerStatus::Paused => {
active_worker_count += 1; active_worker_count += 1;
@@ -787,7 +791,13 @@ impl Runtime {
request.worker_id 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!( return Err(RuntimeError::InvalidRequest(format!(
"worker {} was already created with a different fingerprint", "worker {} was already created with a different fingerprint",
request.worker_id request.worker_id
@@ -951,7 +961,13 @@ impl Runtime {
request.worker_id 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!( return Err(RuntimeError::InvalidRequest(format!(
"worker {} was already created with a different fingerprint", "worker {} was already created with a different fingerprint",
request.worker_id request.worker_id
@@ -986,7 +1002,19 @@ impl Runtime {
status: WorkerStatus::Stopped, status: WorkerStatus::Stopped,
worker_state: None, worker_state: None,
workspace_id: scope.map(|scope| scope.workspace_id.clone()), 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, execution_bound: true,
restore_intent: WorkerRestoreIntent::Explicit, restore_intent: WorkerRestoreIntent::Explicit,
working_directory: None, working_directory: None,
@@ -1029,7 +1057,11 @@ impl Runtime {
if let Some(mut initial_input) = { if let Some(mut initial_input) = {
let state = self.lock()?; 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 let expected_submission_id = initial_input
.submission_request_id .submission_request_id
@@ -1291,7 +1323,13 @@ impl Runtime {
let previous_workspace_api = { let previous_workspace_api = {
let state = self.lock()?; let state = self.lock()?;
let worker = state.worker(worker_ref)?; 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.workspace_id != workspace_api.workspace_id
|| existing.base_url.trim_end_matches('/') || existing.base_url.trim_end_matches('/')
!= workspace_api.base_url.trim_end_matches('/')) != workspace_api.base_url.trim_end_matches('/'))
@@ -1301,14 +1339,26 @@ impl Runtime {
.to_string(), .to_string(),
)); ));
} }
worker.request.workspace_api.clone() request.workspace_api.clone()
}; };
{ {
let mut state = self.lock()?; let mut state = self.lock()?;
state.worker_mut(worker_ref)?.request.workspace_api = Some(workspace_api); let worker = state.worker_mut(worker_ref)?;
if let Err(error) = state.persist_runtime_snapshot() { let request = worker.request.as_mut().ok_or_else(|| {
state.worker_mut(worker_ref)?.request.workspace_api = previous_workspace_api; 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); return Err(error);
} }
} }
@@ -1371,7 +1421,10 @@ impl Runtime {
}); });
} }
match mode { 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!( return Err(RuntimeError::InvalidRequest(format!(
"worker {} is not stopped", "worker {} is not stopped",
worker_ref.worker_id 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(|| { let backend = state.execution_backend.clone().ok_or_else(|| {
RuntimeError::WorkerExecutionUnavailable { RuntimeError::WorkerExecutionUnavailable {
@@ -1829,6 +1888,7 @@ impl Runtime {
let detail = { let detail = {
let worker = state.worker_mut(worker_ref)?; let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = Some(handle); worker.execution_handle = Some(handle);
worker.execution_metadata_available = true;
worker.execution_bound = true; worker.execution_bound = true;
worker.status = WorkerStatus::Idle; worker.status = WorkerStatus::Idle;
let _ = worker.apply_worker_state(&initial_worker_state); let _ = worker.apply_worker_state(&initial_worker_state);
@@ -1867,7 +1927,7 @@ impl Runtime {
fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> {
let mut state = self.lock()?; let mut state = self.lock()?;
if state.workers.contains_key(&worker_ref.worker_id) { 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 let record = state
.workers .workers
.remove(&worker_ref.worker_id) .remove(&worker_ref.worker_id)
@@ -2109,7 +2169,7 @@ impl Runtime {
state.ensure_running()?; state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?; state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(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!( return Err(RuntimeError::InvalidRequest(format!(
"worker {} is running and must be stopped before deletion", "worker {} is running and must be stopped before deletion",
worker_ref.worker_id worker_ref.worker_id
@@ -2140,13 +2200,13 @@ impl Runtime {
state.ensure_running()?; state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?; state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(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!( return Err(RuntimeError::InvalidRequest(format!(
"worker {} became active before deletion", "worker {} became active before deletion",
worker_ref.worker_id 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(|| { let removed = state.workers.remove(&worker_ref.worker_id).ok_or_else(|| {
RuntimeError::WorkerNotFound { RuntimeError::WorkerNotFound {
worker_id: worker_ref.worker_id, worker_id: worker_ref.worker_id,
@@ -2410,6 +2470,7 @@ impl Runtime {
))); )));
} }
worker.execution_handle = Some(handle); worker.execution_handle = Some(handle);
worker.execution_metadata_available = true;
worker.execution_bound = true; worker.execution_bound = true;
worker.status = status; worker.status = status;
let _ = worker.apply_worker_state(&worker_state); let _ = worker.apply_worker_state(&worker_state);
@@ -2445,6 +2506,7 @@ impl Runtime {
let mut state = self.lock()?; let mut state = self.lock()?;
let worker = state.worker_mut(worker_ref)?; let worker = state.worker_mut(worker_ref)?;
worker.execution_handle = None; worker.execution_handle = None;
worker.execution_bound = false;
worker.status = WorkerStatus::Stopped; worker.status = WorkerStatus::Stopped;
worker.worker_state = None; worker.worker_state = None;
worker.restore_intent = WorkerRestoreIntent::Explicit; worker.restore_intent = WorkerRestoreIntent::Explicit;
@@ -2738,6 +2800,18 @@ impl RuntimeState {
let diagnostics = persisted.diagnostics; let diagnostics = persisted.diagnostics;
let next_diagnostic_id = persisted.next_diagnostic_id; let next_diagnostic_id = persisted.next_diagnostic_id;
for (worker_id, worker) in persisted.workers { 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( workers.insert(
worker_id, worker_id,
WorkerRecord { WorkerRecord {
@@ -2746,9 +2820,15 @@ impl RuntimeState {
status: worker.status, status: worker.status,
worker_state: None, worker_state: None,
workspace_id: worker.workspace_id, workspace_id: worker.workspace_id,
request: worker.request, profile: worker.profile,
execution_bound: worker.execution.binding.is_some(), display_name: worker.display_name,
restore_intent: worker.execution.restore_intent, 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, working_directory: worker.working_directory,
execution_handle: None, execution_handle: None,
internal_workers: InternalWorkerActivityProjection::default(), internal_workers: InternalWorkerActivityProjection::default(),
@@ -2826,15 +2906,15 @@ impl RuntimeState {
.ok_or_else(|| RuntimeError::WorkerNotFound { .ok_or_else(|| RuntimeError::WorkerNotFound {
worker_id: *worker_id, worker_id: *worker_id,
})?; })?;
store.write_worker_snapshot(&worker.persisted_record())?; store.write_worker_record(&worker.persisted_record())?;
} }
Ok(()) Ok(())
} }
#[cfg(feature = "fs-store")] #[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() { if let Some(store) = self.fs_store() {
store.delete_worker_snapshot(worker_id)?; store.delete_worker_record(worker_id)?;
} }
Ok(()) Ok(())
} }
@@ -2860,7 +2940,7 @@ impl RuntimeState {
} }
#[cfg(not(feature = "fs-store"))] #[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(()) Ok(())
} }
@@ -3064,7 +3144,7 @@ impl RuntimeState {
.map_err(subscription_validation_error) .map_err(subscription_validation_error)
}) })
.transpose()?; .transpose()?;
let profile = match &worker.request.profile { let profile = match &worker.profile {
ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => Some(name.clone()), ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => Some(name.clone()),
}; };
Ok(SubscriptionWorker { Ok(SubscriptionWorker {
@@ -3080,7 +3160,7 @@ impl RuntimeState {
state: subscription_worker_state(worker.status), state: subscription_worker_state(worker.status),
has_running_internal_workers: worker.internal_workers.has_running_worker(), has_running_internal_workers: worker.internal_workers.has_running_worker(),
workspace_id: worker.workspace_id.clone(), workspace_id: worker.workspace_id.clone(),
display_name: worker.request.display_name.clone(), display_name: worker.display_name.clone(),
profile, profile,
repository_id, repository_id,
repository_key: None, repository_key: None,
@@ -3188,7 +3268,11 @@ impl RuntimeState {
.working_directory .working_directory
.as_ref() .as_ref()
.is_some_and(|binding| binding.summary.working_directory_id == working_directory_id) .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) Some(worker.worker_id)
} else { } else {
@@ -3233,6 +3317,7 @@ impl RuntimeState {
}); });
let worker = self.worker_mut(worker_ref)?; let worker = self.worker_mut(worker_ref)?;
worker.execution_handle = None; worker.execution_handle = None;
worker.execution_bound = false;
worker.status = WorkerStatus::Stopped; worker.status = WorkerStatus::Stopped;
worker.restore_intent = WorkerRestoreIntent::Explicit; worker.restore_intent = WorkerRestoreIntent::Explicit;
worker.internal_workers.clear(); worker.internal_workers.clear();
@@ -3585,7 +3670,13 @@ struct WorkerRecord {
status: WorkerStatus, status: WorkerStatus,
worker_state: Option<protocol::WorkerStateSnapshot>, worker_state: Option<protocol::WorkerStateSnapshot>,
workspace_id: Option<String>, workspace_id: Option<String>,
request: CreateWorkerRequest, profile: ProfileSelector,
display_name: Option<String>,
profile_source: ProfileSourceArchiveRef,
config_bundle: Option<ConfigBundleRef>,
request: Option<CreateWorkerRequest>,
created_at_ms: Option<u64>,
execution_metadata_available: bool,
execution_bound: bool, execution_bound: bool,
restore_intent: WorkerRestoreIntent, restore_intent: WorkerRestoreIntent,
working_directory: Option<CatalogWorkingDirectoryStatus>, working_directory: Option<CatalogWorkingDirectoryStatus>,
@@ -3607,13 +3698,15 @@ impl WorkerRecord {
worker_ref: self.worker_ref.clone(), worker_ref: self.worker_ref.clone(),
worker_id: self.worker_id, worker_id: self.worker_id,
status: self.status, status: self.status,
created_at_ms: self.created_at_ms,
execution_metadata_available: self.execution_metadata_available,
worker_state: self.worker_state.clone(), worker_state: self.worker_state.clone(),
workspace_id: self.workspace_id.clone(), workspace_id: self.workspace_id.clone(),
working_directory: self.working_directory.clone(), working_directory: self.working_directory.clone(),
profile: self.request.profile.clone(), profile: self.profile.clone(),
display_name: self.request.display_name.clone(), display_name: self.display_name.clone(),
profile_source: self.request.profile_source.reference(), profile_source: self.profile_source.clone(),
config_bundle: self.request.config_bundle.clone(), config_bundle: self.config_bundle.clone(),
} }
} }
@@ -3622,29 +3715,42 @@ impl WorkerRecord {
worker_ref: self.worker_ref.clone(), worker_ref: self.worker_ref.clone(),
worker_id: self.worker_id, worker_id: self.worker_id,
status: self.status, status: self.status,
created_at_ms: self.created_at_ms,
execution_metadata_available: self.execution_metadata_available,
worker_state: self.worker_state.clone(), worker_state: self.worker_state.clone(),
workspace_id: self.workspace_id.clone(), workspace_id: self.workspace_id.clone(),
working_directory: self.working_directory.clone(), working_directory: self.working_directory.clone(),
profile: self.request.profile.clone(), profile: self.profile.clone(),
display_name: self.request.display_name.clone(), display_name: self.display_name.clone(),
profile_source: self.request.profile_source.reference(), profile_source: self.profile_source.clone(),
config_bundle: self.request.config_bundle.clone(), config_bundle: self.config_bundle.clone(),
} }
} }
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
fn persisted_record(&self) -> PersistedWorkerRecord { fn persisted_record(&self) -> PersistedWorkerRecord {
PersistedWorkerRecord { let execution_state = match (self.execution_metadata_available, self.request.clone()) {
worker_ref: self.worker_ref.clone(), (true, Some(request)) => {
worker_id: self.worker_id.clone(), PersistedWorkerExecutionState::Available(PersistedWorkerExecution {
request: self.request.clone(), request,
status: self.status,
execution: PersistedWorkerExecution {
binding: self binding: self
.execution_bound .execution_bound
.then_some(PersistedWorkerExecutionBinding {}), .then_some(PersistedWorkerExecutionBinding {}),
restore_intent: self.restore_intent, restore_intent: self.restore_intent,
}, })
}
_ => PersistedWorkerExecutionState::Unavailable,
};
PersistedWorkerRecord {
worker_ref: self.worker_ref.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_state,
workspace_id: self.workspace_id.clone(), workspace_id: self.workspace_id.clone(),
working_directory: self.working_directory.clone(), working_directory: self.working_directory.clone(),
} }
@@ -5433,7 +5539,8 @@ mod tests {
.worker(&worker.worker_ref) .worker(&worker.worker_ref)
.unwrap() .unwrap()
.request .request
.workspace_api, .as_ref()
.and_then(|request| request.workspace_api.clone()),
Some(replacement) Some(replacement)
); );
} }
@@ -6744,7 +6851,7 @@ mod tests {
assert!( assert!(
error error
.to_string() .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); let _ = std::fs::remove_dir_all(root);
@@ -6783,22 +6890,19 @@ mod tests {
.stop_worker(&worker.worker_ref, Some("finished".to_string())) .stop_worker(&worker.worker_ref, Some("finished".to_string()))
.unwrap(); .unwrap();
let worker_store_dir = root.join("workers").join(worker.worker_id.to_string()); 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()) serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap())
.unwrap(); .unwrap();
assert_eq!(worker_snapshot["schema_version"], serde_json::json!(7)); let worker_execution: serde_json::Value = serde_json::from_slice(
assert_eq!(worker_snapshot["status"], serde_json::json!("stopped")); &std::fs::read(worker_store_dir.join("execution.json")).unwrap(),
assert!( )
worker_snapshot["execution"] .unwrap();
.get("last_run_generation") assert_eq!(worker_identity["schema_version"], serde_json::json!(8));
.is_none() 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!( assert_eq!(
worker_snapshot["execution"]["binding"], worker_execution["restore_intent"],
serde_json::json!({})
);
assert_eq!(
worker_snapshot["execution"]["restore_intent"],
serde_json::json!("explicit") serde_json::json!("explicit")
); );
assert!(!root.join("events.jsonl").exists()); assert!(!root.join("events.jsonl").exists());
@@ -6858,7 +6962,7 @@ mod tests {
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
#[test] #[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 root = fs_store_root("workspace-scope");
let runtime = Runtime::with_fs_store_and_execution_backend( let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions { crate::fs_store::FsRuntimeStoreOptions {
@@ -6908,26 +7012,22 @@ mod tests {
.worker_detail_scoped(&scope("workspace-a", "server-a"), &legacy.worker_ref) .worker_detail_scoped(&scope("workspace-a", "server-a"), &legacy.worker_ref)
.unwrap_err(); .unwrap_err();
assert!(matches!(legacy_error, RuntimeError::WorkerNotFound { .. })); assert!(matches!(legacy_error, RuntimeError::WorkerNotFound { .. }));
let recovered_legacy = restored let missing_identity_error = restored
.worker_detail_scoped( .worker_detail_scoped(
&scope("workspace-b", "server-b"), &scope("workspace-b", "server-b"),
&recoverable_legacy.worker_ref, &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(); .unwrap_err();
assert!(matches!( assert!(matches!(
stolen_legacy_error, missing_identity_error,
RuntimeError::WorkspaceOwnerMismatch { .. } RuntimeError::WorkerNotFound { .. }
)); ));
assert!(
!restored
.worker_detail(&recoverable_legacy.worker_ref)
.unwrap()
.execution_metadata_available
);
let _ = std::fs::remove_dir_all(root); let _ = std::fs::remove_dir_all(root);
} }
@@ -7032,8 +7132,8 @@ mod tests {
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
#[test] #[test]
fn fs_store_current_schema_requires_lifecycle_authority() { fn fs_store_retains_identity_when_execution_metadata_is_corrupt() {
let root = fs_store_root("current-schema-requires-lifecycle"); let root = fs_store_root("corrupt-execution-retains-identity");
let options = crate::fs_store::FsRuntimeStoreOptions { let options = crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(), root: root.clone(),
runtime_id: "test-runtime".to_string(), runtime_id: "test-runtime".to_string(),
@@ -7046,20 +7146,25 @@ mod tests {
.unwrap(); .unwrap();
runtime.store_config_bundle(test_bundle()).unwrap(); runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime let worker = runtime
.create_worker(task_request("missing lifecycle authority")) .create_worker(task_request("corrupt execution metadata"))
.unwrap(); .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); drop(runtime);
let worker_path = root let worker_dir = root.join("workers").join(worker.worker_id.to_string());
.join("workers") let restorable_worker_dir = root.join("workers").join(restorable.worker_id.to_string());
.join(worker.worker_id.to_string()) let execution_path = worker_dir.join("execution.json");
.join("worker.json"); let mut execution: serde_json::Value =
let mut worker_json: serde_json::Value = serde_json::from_slice(&std::fs::read(&execution_path).unwrap()).unwrap();
serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); execution["restore_intent"] = serde_json::json!(17);
worker_json.as_object_mut().unwrap().remove("status");
std::fs::write( std::fs::write(
&worker_path, &execution_path,
serde_json::to_vec_pretty(&worker_json).unwrap(), serde_json::to_vec_pretty(&execution).unwrap(),
) )
.unwrap(); .unwrap();
@@ -7068,14 +7173,50 @@ mod tests {
Arc::new(TestExecutionBackend::default()), Arc::new(TestExecutionBackend::default()),
) )
.unwrap(); .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!( assert!(
restored restored
.diagnostics() .delete_worker(&restorable.worker_ref)
.unwrap() .unwrap()
.iter() .deleted
.any(|diagnostic| diagnostic.code == "worker_snapshot_ignored")
); );
assert!(restored.list_workers().unwrap().is_empty());
assert!(!restorable_worker_dir.exists());
let _ = std::fs::remove_dir_all(root); let _ = std::fs::remove_dir_all(root);
} }
@@ -7100,7 +7241,7 @@ mod tests {
runtime.stop_runtime().unwrap(); 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( &std::fs::read(
root.join("workers") root.join("workers")
.join(worker.worker_id.to_string()) .join(worker.worker_id.to_string())
@@ -7109,11 +7250,17 @@ mod tests {
.unwrap(), .unwrap(),
) )
.unwrap(); .unwrap();
assert_eq!(snapshot["status"], serde_json::json!("idle")); let execution: serde_json::Value = serde_json::from_slice(
assert_eq!( &std::fs::read(
snapshot["execution"]["restore_intent"], root.join("workers")
serde_json::json!("automatic") .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); let _ = std::fs::remove_dir_all(root);
} }
@@ -7173,8 +7320,8 @@ mod tests {
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
#[test] #[test]
fn fs_store_migrates_schema_v6_workers_without_losing_automatic_restore() { fn fs_store_migrates_schema_v7_workers_into_identity_and_execution_records() {
let root = fs_store_root("schema-v6-no-generation"); let root = fs_store_root("schema-v7-split-records");
let options = crate::fs_store::FsRuntimeStoreOptions { let options = crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(), root: root.clone(),
runtime_id: "test-runtime".to_string(), runtime_id: "test-runtime".to_string(),
@@ -7187,34 +7334,57 @@ mod tests {
.unwrap(); .unwrap();
runtime.store_config_bundle(test_bundle()).unwrap(); runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime let worker = runtime
.create_worker(task_request("schema v6 worker")) .create_worker(task_request("schema v7 worker"))
.unwrap(); .unwrap();
drop(runtime); drop(runtime);
let runtime_path = root.join("runtime.json"); let worker_dir = root.join("workers").join(worker.worker_id.to_string());
let worker_path = root let worker_path = worker_dir.join("worker.json");
.join("workers") let execution_path = worker_dir.join("execution.json");
.join(worker.worker_id.to_string()) let mut runtime_snapshot: serde_json::Value = serde_json::from_slice(
.join("worker.json"); &std::fs::read(root.join("runtime.json")).expect("runtime snapshot"),
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(),
) )
.unwrap(); .expect("runtime snapshot json");
let mut worker_json: serde_json::Value = runtime_snapshot["schema_version"] = serde_json::json!(7);
serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); std::fs::write(
worker_json["schema_version"] = serde_json::json!(6); root.join("runtime.json"),
worker_json["run_generation"] = serde_json::json!(1); serde_json::to_vec_pretty(&runtime_snapshot).expect("runtime snapshot bytes"),
worker_json["execution"]["last_run_generation"] = serde_json::json!(1); )
worker_json["execution"]["binding"] = serde_json::json!({"run_generation": 1}); .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( std::fs::write(
&worker_path, &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 backend = Arc::new(TestExecutionBackend::default());
let migrated = let migrated =
@@ -7224,18 +7394,21 @@ mod tests {
migrated.worker_detail(&worker.worker_ref).unwrap().status, migrated.worker_detail(&worker.worker_ref).unwrap().status,
WorkerStatus::Idle 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(); serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap();
assert_eq!(migrated_json["schema_version"], serde_json::json!(7)); let migrated_execution: serde_json::Value =
assert_eq!(migrated_json["execution"]["binding"], serde_json::json!({})); serde_json::from_slice(&std::fs::read(&execution_path).unwrap()).unwrap();
assert!(migrated_json.get("run_generation").is_none()); assert_eq!(migrated_identity["schema_version"], serde_json::json!(8));
assert!( assert!(migrated_identity.get("request").is_none());
migrated_json["execution"] assert!(migrated_identity.get("execution").is_none());
.get("last_run_generation") assert_eq!(migrated_execution["schema_version"], serde_json::json!(8));
.is_none()
);
assert_eq!( 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") serde_json::json!("automatic")
); );
@@ -7342,7 +7515,7 @@ mod tests {
.unwrap(); .unwrap();
missing_runtime.store_config_bundle(test_bundle()).unwrap(); missing_runtime.store_config_bundle(test_bundle()).unwrap();
missing_runtime missing_runtime
.create_worker(task_request("missing worker snapshot")) .create_worker(task_request("missing Worker identity"))
.unwrap(); .unwrap();
let missing_store = runtime_store(&missing_runtime); let missing_store = runtime_store(&missing_runtime);
let mut worker_dirs = std::fs::read_dir(missing_store.runtime_dir().join("workers")) 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(), runtime_id: "test-runtime".to_string(),
display_name: None, 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.list_workers().unwrap().is_empty());
assert!( assert!(
loaded loaded
.diagnostics() .diagnostics()
.unwrap() .unwrap()
.iter() .iter()
.any(|diagnostic| diagnostic.code == "worker_snapshot_ignored") .any(|diagnostic| diagnostic.code == "worker_record_unavailable")
); );
let _ = std::fs::remove_dir_all(missing_root); let _ = std::fs::remove_dir_all(missing_root);
} }
crates/workspace-server/src/hosts.rs
+87 -27
View File
@@ -2129,7 +2129,11 @@ impl EmbeddedWorkerRuntime {
identity: "runtime_registry_worker".to_string(), identity: "runtime_registry_worker".to_string(),
workspace_id: summary.workspace_id.clone(), 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(), worker_state: summary.worker_state.clone(),
last_seen_at: None, last_seen_at: None,
pinned: false, pinned: false,
@@ -2139,11 +2143,14 @@ impl EmbeddedWorkerRuntime {
display_hint: "backend-internal worker-runtime Worker".to_string(), display_hint: "backend-internal worker-runtime Worker".to_string(),
}, },
capabilities: WorkerCapabilitySummary { 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, can_spawn_followup: false,
}, },
working_directory: summary.working_directory.map(|status| status.summary), 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(), identity: "runtime_registry_worker".to_string(),
workspace_id: detail.workspace_id.clone(), 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(), worker_state: detail.worker_state.clone(),
last_seen_at: None, last_seen_at: None,
pinned: false, pinned: false,
@@ -2179,11 +2187,14 @@ impl EmbeddedWorkerRuntime {
display_hint: "backend-internal worker-runtime Worker".to_string(), display_hint: "backend-internal worker-runtime Worker".to_string(),
}, },
capabilities: WorkerCapabilitySummary { 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, can_spawn_followup: false,
}, },
working_directory: detail.working_directory.map(|status| status.summary), 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(), identity: "runtime_registry_worker".to_string(),
workspace_id: summary.workspace_id.clone(), 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(), worker_state: summary.worker_state.clone(),
last_seen_at: None, last_seen_at: None,
pinned: false, pinned: false,
@@ -3924,15 +3939,12 @@ impl RemoteWorkerRuntime {
display_hint: "Backend-proxied remote worker-runtime Worker".to_string(), display_hint: "Backend-proxied remote worker-runtime Worker".to_string(),
}, },
capabilities: WorkerCapabilitySummary { 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, can_spawn_followup: false,
}, },
working_directory: summary.working_directory.map(|status| status.summary), working_directory: summary.working_directory.map(|status| status.summary),
diagnostics: vec![diagnostic( diagnostics: remote_worker_projection_diagnostics(summary.execution_metadata_available),
"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(),
)],
} }
} }
@@ -3958,7 +3970,8 @@ impl RemoteWorkerRuntime {
identity: "runtime_registry_worker".to_string(), identity: "runtime_registry_worker".to_string(),
workspace_id: detail.workspace_id.clone(), 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(), worker_state: detail.worker_state.clone(),
last_seen_at: None, last_seen_at: None,
pinned: false, pinned: false,
@@ -3968,15 +3981,12 @@ impl RemoteWorkerRuntime {
display_hint: "Backend-proxied remote worker-runtime Worker".to_string(), display_hint: "Backend-proxied remote worker-runtime Worker".to_string(),
}, },
capabilities: WorkerCapabilitySummary { 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, can_spawn_followup: false,
}, },
working_directory: detail.working_directory.map(|status| status.summary), working_directory: detail.working_directory.map(|status| status.summary),
diagnostics: vec![diagnostic( diagnostics: remote_worker_projection_diagnostics(detail.execution_metadata_available),
"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(),
)],
} }
} }
@@ -4696,12 +4706,49 @@ fn embedded_worker_status_label(status: EmbeddedWorkerStatus) -> &'static str {
} }
} }
fn embedded_worker_projection_diagnostics() -> Vec<RuntimeDiagnostic> { fn embedded_worker_state_label(
vec![diagnostic( 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<RuntimeDiagnostic> {
(!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<RuntimeDiagnostic> {
let mut diagnostics = vec![diagnostic(
"embedded_runtime_projection", "embedded_runtime_projection",
DiagnosticSeverity::Info, DiagnosticSeverity::Info,
"Worker identity is projected only as runtime_id plus worker_id; embedded runtime internals remain backend-private".to_string(), "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<RuntimeDiagnostic> {
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<ConfigBundleRef> { fn spawn_config_bundle_ref(request: &WorkerSpawnRequest) -> Option<ConfigBundleRef> {
@@ -6787,11 +6834,14 @@ mod tests {
} }
#[test] #[test]
fn remote_runtime_projection_uses_canonical_worker_status_for_stop_capability() { fn remote_runtime_projection_uses_execution_availability_and_canonical_status() {
let worker_ids = (1..=4) let worker_ids = (1..=5)
.map(|value| EmbeddedWorkerId::from_legacy_u64(value).to_string()) .map(|value| EmbeddedWorkerId::from_legacy_u64(value).to_string())
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let worker_id = worker_ids[0].clone(); 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![ let (base_url, server) = serve_mock_http(vec![
mock_response( mock_response(
"GET", "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[0], "stopped"),
worker_json_with_status("remote:primary", &worker_ids[1], "running"), 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[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(), .to_string(),
@@ -6838,15 +6889,23 @@ mod tests {
)]); )]);
let workers = registry.list_workers(10); 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[0].capabilities.can_stop);
assert!(workers.items[1].capabilities.can_stop); assert!(workers.items[1].capabilities.can_stop);
assert!(workers.items[2].capabilities.can_stop); assert!(workers.items[2].capabilities.can_stop);
assert!(workers.items[3].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[0].state, "stopped");
assert_eq!(workers.items[1].state, "running"); assert_eq!(workers.items[1].state, "running");
assert_eq!(workers.items[2].state, "paused"); assert_eq!(workers.items[2].state, "paused");
assert_eq!(workers.items[3].state, "idle"); 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 let stopped_detail = registry
.worker(&RuntimeWorkerRef::new("remote:primary", &worker_id)) .worker(&RuntimeWorkerRef::new("remote:primary", &worker_id))
@@ -7119,6 +7178,7 @@ mod tests {
"runtime_id": runtime_id, "runtime_id": runtime_id,
"worker_id": worker_id, "worker_id": worker_id,
"status": status, "status": status,
"execution_metadata_available": true,
"intent": { "kind": "role", "role": "coder", "purpose": "remote test" }, "intent": { "kind": "role", "role": "coder", "purpose": "remote test" },
"profile": { "kind": "builtin", "value": "coder" }, "profile": { "kind": "builtin", "value": "coder" },
"profile_source": { "profile_source": {