From 91d1b1fedd18728cefff3133a4c9e40520b4b87d Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 26 Sep 2026 21:40:21 +0900 Subject: [PATCH] fix: make worker aggregate persistence atomic --- crates/worker-runtime/src/error.rs | 11 + crates/worker-runtime/src/fs_store.rs | 585 +++++++++++++++++------ crates/worker-runtime/src/http_server.rs | 4 + crates/worker-runtime/src/runtime.rs | 238 +++++---- 4 files changed, 610 insertions(+), 228 deletions(-) diff --git a/crates/worker-runtime/src/error.rs b/crates/worker-runtime/src/error.rs index 7f320c3e8..c0f108386 100644 --- a/crates/worker-runtime/src/error.rs +++ b/crates/worker-runtime/src/error.rs @@ -96,6 +96,17 @@ pub enum RuntimeError { source: std::io::Error, }, + #[error( + "runtime store {operation} may have committed at {} before durability confirmation failed: {source}", + path.display() + )] + StoreCommitOutcomeUnknown { + operation: &'static str, + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("runtime store {operation} missing data at {}", path.display())] StoreMissing { operation: &'static str, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 7d783f757..28e7b278b 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -18,11 +18,12 @@ use std::fs::{self, File, OpenOptions}; use std::io::{BufReader, Read, Write}; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU8, AtomicU64, Ordering}; -const SCHEMA_VERSION: u32 = 8; -const PREVIOUS_SCHEMA_VERSION: u32 = 7; -const LEGACY_SCHEMA_VERSION: u32 = 6; +const SCHEMA_VERSION: u32 = 9; +const PREVIOUS_SCHEMA_VERSION: u32 = 8; +const LEGACY_SCHEMA_VERSION: u32 = 7; +const OLDEST_SCHEMA_VERSION: u32 = 6; const RUNTIME_FILE: &str = "runtime.json"; const WORKERS_DIR: &str = "workers"; const ORPHANED_WORKERS_DIR: &str = "orphaned-workers"; @@ -35,6 +36,24 @@ const MAX_PERSISTED_WORKER_RECORDS: usize = 4096; static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1); +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +#[repr(u8)] +pub(crate) enum AtomicWriteFault { + None = 0, + BeforeRename = 1, + AfterRename = 2, +} + +impl AtomicWriteFault { + fn take(slot: &AtomicU8) -> Self { + match slot.swap(Self::None as u8, Ordering::AcqRel) { + value if value == Self::BeforeRename as u8 => Self::BeforeRename, + value if value == Self::AfterRename as u8 => Self::AfterRename, + _ => Self::None, + } + } +} + /// Options for constructing a filesystem-backed Runtime store. #[derive(Clone, Debug, PartialEq, Eq)] pub struct FsRuntimeStoreOptions { @@ -77,6 +96,7 @@ impl Drop for RuntimeStoreOwnerLock { pub struct FsRuntimeStore { root: PathBuf, _owner_lock: Option>, + worker_write_fault: Arc, } impl PartialEq for FsRuntimeStore { @@ -111,6 +131,11 @@ impl FsRuntimeStore { &self.root } + #[cfg(test)] + pub(crate) fn fail_next_worker_write(&self, fault: AtomicWriteFault) { + self.worker_write_fault.store(fault as u8, Ordering::Release); + } + pub(crate) fn open_or_create( root: PathBuf, runtime_id: &str, @@ -169,6 +194,7 @@ impl FsRuntimeStore { let store = Self { root, _owner_lock: Some(owner_lock), + worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), }; let state = if snapshot_exists { Some(store.load_runtime_state()?) @@ -197,18 +223,12 @@ impl FsRuntimeStore { path: worker_dir.clone(), source, })?; - atomic_write_json( + atomic_write_json_with_fault( &worker_dir.join(WORKER_FILE), - &WorkerIdentityRecord::from_persisted(worker), - "write Worker identity", + &WorkerAggregateRecord::from_persisted(worker), + "write Worker aggregate", + AtomicWriteFault::take(&self.worker_write_fault), )?; - 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(()) } @@ -285,44 +305,32 @@ impl FsRuntimeStore { ); continue; } - let identity_path = path.join(WORKER_FILE); - let identity: WorkerIdentityRecord = - match read_bounded_json(&identity_path, "read Worker identity") { - Ok(identity) => identity, + let aggregate_path = path.join(WORKER_FILE); + let aggregate: WorkerAggregateRecord = + match read_bounded_json(&aggregate_path, "read Worker aggregate") { + Ok(aggregate) => aggregate, Err(_error) => { record_worker_load_diagnostic( &mut snapshot, None, - "ignored corrupt Worker identity while loading Runtime store", + "ignored corrupt Worker aggregate while loading Runtime store", ); continue; } }; - if identity.validate(&identity_path).is_err() { - record_worker_load_diagnostic( - &mut snapshot, - 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 - }); + let worker_ref = aggregate.worker_ref.clone(); + let worker = match aggregate.validate(&aggregate_path) { + Ok(worker) => worker, + Err(_error) => { + record_worker_load_diagnostic( + &mut snapshot, + Some(worker_ref), + "ignored invalid Worker aggregate while loading Runtime store", + ); + continue; + } + }; remove_legacy_observations(&path); - let worker = identity.into_persisted(execution_state); if workers.insert(worker.worker_id.clone(), worker).is_some() { record_worker_load_diagnostic( &mut snapshot, @@ -557,8 +565,8 @@ fn plan_runtime_store_migration( format!("Runtime store schema version {schema_version} is out of range"), ) })?; - let staging = migration_sibling(root, "schema-v8-staging")?; - let backup = migration_sibling(root, "pre-schema-v8-backup")?; + let staging = migration_sibling(root, "schema-v9-staging")?; + let backup = migration_sibling(root, "pre-schema-v9-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -587,12 +595,12 @@ fn plan_runtime_store_migration( } if !matches!( current_schema_version, - LEGACY_SCHEMA_VERSION | PREVIOUS_SCHEMA_VERSION + OLDEST_SCHEMA_VERSION | LEGACY_SCHEMA_VERSION | PREVIOUS_SCHEMA_VERSION ) { return Err(runtime_store_corrupt( &runtime_path, format!( - "unsupported Runtime store schema version {schema_version}; expected {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, or {SCHEMA_VERSION}" + "unsupported Runtime store schema version {schema_version}; expected {OLDEST_SCHEMA_VERSION}, {LEGACY_SCHEMA_VERSION}, {PREVIOUS_SCHEMA_VERSION}, or {SCHEMA_VERSION}" ), )); } @@ -697,23 +705,22 @@ fn plan_runtime_store_migration( 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 record")?; - let migrated = migrate_worker_document( + let migrated = migrate_worker_record( snapshot, current_schema_version, worker.legacy_mapping.as_ref(), &snapshot_path, )?; - let identity = validate_migrated_worker_documents(&migrated, &snapshot_path)?; - if identity.worker_id != worker.worker_id { + if migrated.worker_id != worker.worker_id { return Err(runtime_store_corrupt( &snapshot_path, format!( "Worker identity {} does not match directory identity {}", - identity.worker_id, worker.worker_id + migrated.worker_id, worker.worker_id ), )); } - worker.workspace_id = worker.workspace_id.clone().or(identity.workspace_id); + worker.workspace_id = worker.workspace_id.clone().or(migrated.workspace_id); let metadata_path = worker.source_dir.join(WORKER_METADATA_FILE); if metadata_path.is_file() { @@ -807,7 +814,7 @@ fn migrate_schema_v6_worker_document_to_v7( } object.insert( "schema_version".to_string(), - serde_json::Value::from(PREVIOUS_SCHEMA_VERSION), + serde_json::Value::from(LEGACY_SCHEMA_VERSION), ); Ok(document) } @@ -818,19 +825,19 @@ fn migrate_worker_document( _mapping: Option<&LegacyWorkerIdentityMapping>, identity_path: &Path, ) -> Result { - let mut document = if source_schema_version == LEGACY_SCHEMA_VERSION { + let mut document = if source_schema_version == OLDEST_SCHEMA_VERSION { migrate_schema_v6_worker_document_to_v7(document, identity_path)? } else { document }; if !matches!( source_schema_version, - LEGACY_SCHEMA_VERSION | PREVIOUS_SCHEMA_VERSION + OLDEST_SCHEMA_VERSION | LEGACY_SCHEMA_VERSION ) { return Err(runtime_store_corrupt( identity_path, format!( - "unsupported Worker identity schema {source_schema_version}; expected {LEGACY_SCHEMA_VERSION} or {PREVIOUS_SCHEMA_VERSION}" + "unsupported Worker identity schema {source_schema_version}; expected {OLDEST_SCHEMA_VERSION} or {LEGACY_SCHEMA_VERSION}" ), )); } @@ -879,11 +886,11 @@ fn migrate_worker_document( execution.insert("request".to_string(), request_value); execution.insert( "schema_version".to_string(), - serde_json::Value::from(SCHEMA_VERSION), + serde_json::Value::from(PREVIOUS_SCHEMA_VERSION), ); object.insert( "schema_version".to_string(), - serde_json::Value::from(SCHEMA_VERSION), + serde_json::Value::from(PREVIOUS_SCHEMA_VERSION), ); let migrated = MigratedWorkerDocuments { identity: document, @@ -904,7 +911,7 @@ fn validate_migrated_worker_documents( format!("decode migrated Worker identity: {error}"), ) })?; - identity.validate(identity_path)?; + identity.validate_for_schema(identity_path, PREVIOUS_SCHEMA_VERSION)?; let execution_path = identity_path.with_file_name(WORKER_EXECUTION_FILE); let execution: WorkerExecutionRecord = serde_json::from_value(documents.execution.clone()) .map_err(|error| { @@ -913,10 +920,56 @@ fn validate_migrated_worker_documents( format!("decode migrated Worker execution record: {error}"), ) })?; - execution.validate(&identity, &execution_path)?; + execution.validate_for_schema(&identity, &execution_path, PREVIOUS_SCHEMA_VERSION)?; Ok(identity) } +fn migrate_worker_record( + identity_document: serde_json::Value, + source_schema_version: u32, + mapping: Option<&LegacyWorkerIdentityMapping>, + identity_path: &Path, +) -> Result { + if source_schema_version == PREVIOUS_SCHEMA_VERSION { + let identity: WorkerIdentityRecord = serde_json::from_value(identity_document).map_err(|error| { + runtime_store_corrupt( + identity_path, + format!("decode schema-v8 Worker identity: {error}"), + ) + })?; + identity.validate_for_schema(identity_path, PREVIOUS_SCHEMA_VERSION)?; + let execution_path = identity_path.with_file_name(WORKER_EXECUTION_FILE); + let execution: WorkerExecutionRecord = + read_bounded_json(&execution_path, "read schema-v8 Worker execution record")?; + execution.validate_for_schema(&identity, &execution_path, PREVIOUS_SCHEMA_VERSION)?; + return Ok(WorkerAggregateRecord::from_v8(identity, execution)); + } + + let documents = migrate_worker_document( + identity_document, + source_schema_version, + mapping, + identity_path, + )?; + let identity: WorkerIdentityRecord = serde_json::from_value(documents.identity).map_err(|error| { + runtime_store_corrupt( + identity_path, + format!("decode validated schema-v8 Worker identity: {error}"), + ) + })?; + let execution_path = identity_path.with_file_name(WORKER_EXECUTION_FILE); + let execution: WorkerExecutionRecord = + serde_json::from_value(documents.execution).map_err(|error| { + runtime_store_corrupt( + &execution_path, + format!("decode validated schema-v8 Worker execution record: {error}"), + ) + })?; + identity.validate_for_schema(identity_path, PREVIOUS_SCHEMA_VERSION)?; + execution.validate_for_schema(&identity, &execution_path, PREVIOUS_SCHEMA_VERSION)?; + Ok(WorkerAggregateRecord::from_v8(identity, execution)) +} + fn runtime_worker_name(worker_id: WorkerId) -> String { format!("worker-runtime-{worker_id}") } @@ -1178,6 +1231,7 @@ fn validate_current_runtime_store(root: &Path) -> Result<(), RuntimeError> { let store = FsRuntimeStore { root: root.to_path_buf(), _owner_lock: None, + worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), }; store.load_runtime_state().map(|_| ()) } @@ -1186,6 +1240,7 @@ fn validate_completed_migration_staging(root: &Path) -> Result<(), RuntimeError> let store = FsRuntimeStore { root: root.to_path_buf(), _owner_lock: None, + worker_write_fault: Arc::new(AtomicU8::new(AtomicWriteFault::None as u8)), }; let state = store.load_runtime_state()?; if let Some(worker) = state.workers.values().find(|worker| { @@ -1224,8 +1279,8 @@ fn remove_runtime_migration_directory(path: &Path) -> Result<(), RuntimeError> { } fn recover_runtime_store_migration(root: &Path) -> Result<(), RuntimeError> { - let staging = migration_sibling(root, "schema-v8-staging")?; - let backup = migration_sibling(root, "pre-schema-v8-backup")?; + let staging = migration_sibling(root, "schema-v9-staging")?; + let backup = migration_sibling(root, "pre-schema-v9-backup")?; let root_exists = migration_directory_exists(root)?; let staging_exists = migration_directory_exists(&staging)?; let backup_exists = migration_directory_exists(&backup)?; @@ -1427,8 +1482,8 @@ fn migrate_runtime_store( if !plan.migration_required { return Ok(plan); } - let staging = migration_sibling(root, "schema-v8-staging")?; - let backup = migration_sibling(root, "pre-schema-v8-backup")?; + let staging = migration_sibling(root, "schema-v9-staging")?; + let backup = migration_sibling(root, "pre-schema-v9-backup")?; if staging.exists() || backup.exists() { return Err(runtime_store_corrupt( root, @@ -1556,7 +1611,7 @@ fn migrate_runtime_store_in_place( ), ) })?; - let documents = migrate_worker_document( + let aggregate = migrate_worker_record( snapshot, plan.current_schema_version, planned_worker.legacy_mapping.as_ref(), @@ -1585,17 +1640,24 @@ fn migrate_runtime_store_in_place( fs::rename(source_dir, &migrated_dir) .map_err(|error| runtime_io_error("rename", source_dir, error))?; } - let migrated_identity_path = migrated_dir.join(WORKER_FILE); + let migrated_aggregate_path = migrated_dir.join(WORKER_FILE); atomic_write_json( - &migrated_identity_path, - &documents.identity, - "migrate Worker identity", - )?; - atomic_write_json( - &migrated_dir.join(WORKER_EXECUTION_FILE), - &documents.execution, - "migrate Worker execution record", + &migrated_aggregate_path, + &aggregate, + "migrate Worker aggregate", )?; + let legacy_execution_path = migrated_dir.join(WORKER_EXECUTION_FILE); + match fs::remove_file(&legacy_execution_path) { + Ok(()) => {} + Err(source) if source.kind() == std::io::ErrorKind::NotFound => {} + Err(source) => { + return Err(runtime_io_error( + "remove migrated Worker execution record", + &legacy_execution_path, + source, + )); + } + } if let Some(metadata) = metadata { atomic_write_json( &migrated_dir.join(WORKER_METADATA_FILE), @@ -1709,6 +1771,121 @@ impl RuntimeSnapshot { } } +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct WorkerAggregateRecord { + schema_version: u32, + worker_ref: WorkerRef, + worker_id: WorkerId, + 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + workspace_id: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + workdir_attachments: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + logical_workdir_attachments: Vec, + execution_state: WorkerExecutionStateRecord, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(tag = "state", content = "execution", rename_all = "snake_case")] +enum WorkerExecutionStateRecord { + Available(WorkerExecutionData), + Unavailable, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct WorkerExecutionData { + request: CreateWorkerRequest, + #[serde(default, skip_serializing_if = "Option::is_none")] + binding: Option, + restore_intent: WorkerRestoreIntent, +} + +impl WorkerAggregateRecord { + fn from_persisted(worker: &PersistedWorkerRecord) -> Self { + let execution_state = match &worker.execution_state { + PersistedWorkerExecutionState::Available(execution) => { + WorkerExecutionStateRecord::Available(WorkerExecutionData { + request: execution.request.clone(), + binding: execution.binding.clone(), + restore_intent: execution.restore_intent, + }) + } + PersistedWorkerExecutionState::Unavailable => WorkerExecutionStateRecord::Unavailable, + }; + Self { + schema_version: SCHEMA_VERSION, + worker_ref: worker.worker_ref.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, + workspace_id: worker.workspace_id.clone(), + workdir_attachments: worker.workdir_attachments.clone(), + logical_workdir_attachments: worker.logical_workdir_attachments.clone(), + execution_state, + } + } + + fn from_v8(identity: WorkerIdentityRecord, execution: WorkerExecutionRecord) -> Self { + let persisted = identity.into_persisted(PersistedWorkerExecutionState::Available( + PersistedWorkerExecution { + request: execution.request, + binding: execution.binding, + restore_intent: execution.restore_intent, + }, + )); + Self::from_persisted(&persisted) + } + + fn validate(self, path: &Path) -> Result { + let identity = WorkerIdentityRecord { + schema_version: self.schema_version, + 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, + workspace_id: self.workspace_id, + workdir_attachments: self.workdir_attachments, + logical_workdir_attachments: self.logical_workdir_attachments, + legacy_working_directory: None, + }; + identity.validate_for_schema(path, SCHEMA_VERSION)?; + let execution_state = match self.execution_state { + WorkerExecutionStateRecord::Available(execution) => { + let execution = WorkerExecutionRecord { + schema_version: SCHEMA_VERSION, + request: execution.request, + binding: execution.binding, + restore_intent: execution.restore_intent, + }; + PersistedWorkerExecutionState::Available(execution.validate_for_schema( + &identity, + path, + SCHEMA_VERSION, + )?) + } + WorkerExecutionStateRecord::Unavailable => PersistedWorkerExecutionState::Unavailable, + }; + Ok(identity.into_persisted(execution_state)) + } +} + #[derive(Clone, Debug, Serialize, Deserialize)] #[serde(deny_unknown_fields)] struct WorkerIdentityRecord { @@ -1819,32 +1996,14 @@ impl<'de> Deserialize<'de> for WorkerExecutionRecord { } impl WorkerIdentityRecord { - fn from_persisted(worker: &PersistedWorkerRecord) -> Self { - Self { - schema_version: SCHEMA_VERSION, - worker_ref: worker.worker_ref.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, - workspace_id: worker.workspace_id.clone(), - workdir_attachments: worker.workdir_attachments.clone(), - logical_workdir_attachments: worker.logical_workdir_attachments.clone(), - legacy_working_directory: None, - } - } - - fn validate(&self, path: &Path) -> Result<(), RuntimeError> { - if self.schema_version != SCHEMA_VERSION { + fn validate_for_schema(&self, path: &Path, expected_schema: u32) -> Result<(), RuntimeError> { + if self.schema_version != expected_schema { return Err(RuntimeError::StoreCorrupt { operation: "read Worker identity", path: path.to_path_buf(), message: format!( "unsupported schema version {}, expected {}", - self.schema_version, SCHEMA_VERSION + self.schema_version, expected_schema ), }); } @@ -1912,27 +2071,19 @@ impl WorkerIdentityRecord { } 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( + fn validate_for_schema( &self, identity: &WorkerIdentityRecord, path: &Path, + expected_schema: u32, ) -> Result { - if self.schema_version != SCHEMA_VERSION { + if self.schema_version != expected_schema { 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 + self.schema_version, expected_schema ), }); } @@ -2068,6 +2219,18 @@ where } fn atomic_write_json(path: &Path, value: &T, operation: &'static str) -> Result<(), RuntimeError> +where + T: Serialize, +{ + atomic_write_json_with_fault(path, value, operation, AtomicWriteFault::None) +} + +fn atomic_write_json_with_fault( + path: &Path, + value: &T, + operation: &'static str, + fault: AtomicWriteFault, +) -> Result<(), RuntimeError> where T: Serialize, { @@ -2112,12 +2275,32 @@ where source, })?; drop(file); + if fault == AtomicWriteFault::BeforeRename { + return Err(RuntimeError::StoreIo { + operation, + path: path.to_path_buf(), + source: std::io::Error::other("injected failure before atomic rename"), + }); + } fs::rename(&tmp_path, path).map_err(|source| RuntimeError::StoreIo { operation, path: path.to_path_buf(), source, })?; - sync_directory(parent, operation) + if fault == AtomicWriteFault::AfterRename { + return Err(RuntimeError::StoreCommitOutcomeUnknown { + operation, + path: path.to_path_buf(), + source: std::io::Error::other("injected failure after atomic rename"), + }); + } + File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(|source| RuntimeError::StoreCommitOutcomeUnknown { + operation, + path: path.to_path_buf(), + source, + }) })(); if write_result.is_err() { @@ -2152,7 +2335,7 @@ fn sync_directory(path: &Path, operation: &'static str) -> Result<(), RuntimeErr mod tests { use super::*; - fn write_empty_schema_v7_store(root: &Path) { + fn write_empty_schema_v8_store(root: &Path) { fs::create_dir_all(root.join(WORKERS_DIR)).unwrap(); fs::write( root.join(RUNTIME_FILE), @@ -2378,7 +2561,7 @@ mod tests { fn schema_v7_worker_document(worker_id: WorkerId) -> serde_json::Value { serde_json::json!({ - "schema_version": PREVIOUS_SCHEMA_VERSION, + "schema_version": LEGACY_SCHEMA_VERSION, "worker_ref": { "worker_id": worker_id }, "worker_id": worker_id, "request": { @@ -2410,7 +2593,7 @@ mod tests { fn schema_v6_worker_document(worker_id: WorkerId) -> serde_json::Value { let mut document = schema_v7_worker_document(worker_id); - document["schema_version"] = serde_json::json!(LEGACY_SCHEMA_VERSION); + document["schema_version"] = serde_json::json!(OLDEST_SCHEMA_VERSION); document["execution"]["last_run_generation"] = serde_json::json!(7); document["execution"]["binding"] = serde_json::json!({ "run_generation": 7 }); document @@ -2557,8 +2740,8 @@ mod tests { fn startup_recovers_staging_created_before_source_backup() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); - write_empty_schema_v7_store(&root); - let staging = migration_sibling(&root, "schema-v8-staging").unwrap(); + write_empty_schema_v8_store(&root); + let staging = migration_sibling(&root, "schema-v9-staging").unwrap(); fs::create_dir_all(&staging).unwrap(); fs::write(staging.join("partial"), b"partial").unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -2572,8 +2755,8 @@ mod tests { fn startup_restores_backup_when_source_was_renamed_before_activation() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); - write_empty_schema_v7_store(&root); - let backup = migration_sibling(&root, "pre-schema-v8-backup").unwrap(); + write_empty_schema_v8_store(&root); + let backup = migration_sibling(&root, "pre-schema-v9-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -2586,11 +2769,11 @@ mod tests { fn startup_promotes_valid_staging_after_source_backup() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); - write_empty_schema_v7_store(&root); - let staging = migration_sibling(&root, "schema-v8-staging").unwrap(); + write_empty_schema_v8_store(&root); + let staging = migration_sibling(&root, "schema-v9-staging").unwrap(); copy_runtime_tree(&root, &root, &staging).unwrap(); migrate_runtime_store_in_place(&staging, "test-runtime").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v8-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v9-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -2604,11 +2787,11 @@ mod tests { fn startup_restores_backup_when_staging_is_incomplete() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); - write_empty_schema_v7_store(&root); - let staging = migration_sibling(&root, "schema-v8-staging").unwrap(); + write_empty_schema_v8_store(&root); + let staging = migration_sibling(&root, "schema-v9-staging").unwrap(); fs::create_dir_all(&staging).unwrap(); fs::write(staging.join(RUNTIME_FILE), b"{\"schema_version\":8}").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v8-backup").unwrap(); + let backup = migration_sibling(&root, "pre-schema-v9-backup").unwrap(); fs::rename(&root, &backup).unwrap(); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); @@ -2623,10 +2806,10 @@ mod tests { fn startup_removes_backup_left_after_activation() { let temp = tempfile::tempdir().unwrap(); let root = temp.path().join("runtime-store"); - write_empty_schema_v7_store(&root); + write_empty_schema_v8_store(&root); migrate_runtime_store_in_place(&root, "test-runtime").unwrap(); - let backup = migration_sibling(&root, "pre-schema-v8-backup").unwrap(); - write_empty_schema_v7_store(&backup); + let backup = migration_sibling(&root, "pre-schema-v9-backup").unwrap(); + write_empty_schema_v8_store(&backup); let store = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); store.store.load_runtime_state().unwrap(); @@ -2649,6 +2832,109 @@ mod tests { assert!(error.to_string().contains("exceeds limit")); } + #[test] + fn worker_aggregate_write_distinguishes_pre_and_post_rename_failures() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime-store"); + let opened = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); + let worker_id = WorkerId::now_v7(); + let aggregate_path = root + .join(WORKERS_DIR) + .join(worker_id.to_string()) + .join(WORKER_FILE); + let mut worker = migrate_worker_record( + schema_v7_worker_document(worker_id), + LEGACY_SCHEMA_VERSION, + None, + &aggregate_path, + ) + .unwrap() + .validate(&aggregate_path) + .unwrap(); + + opened + .store + .fail_next_worker_write(AtomicWriteFault::BeforeRename); + let error = opened.store.write_worker_record(&worker).unwrap_err(); + assert!(matches!(error, RuntimeError::StoreIo { .. })); + assert!(!aggregate_path.exists()); + + worker.created_at_ms = Some(1); + opened.store.write_worker_record(&worker).unwrap(); + worker.created_at_ms = Some(2); + opened + .store + .fail_next_worker_write(AtomicWriteFault::AfterRename); + let error = opened.store.write_worker_record(&worker).unwrap_err(); + assert!(matches!( + error, + RuntimeError::StoreCommitOutcomeUnknown { .. } + )); + let committed: WorkerAggregateRecord = + read_bounded_json(&aggregate_path, "read test Worker aggregate").unwrap(); + assert_eq!(committed.created_at_ms, Some(2)); + assert!(!aggregate_path.with_file_name(WORKER_EXECUTION_FILE).exists()); + } + + #[test] + fn schema_v8_pair_migrates_to_one_worker_aggregate() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime-store"); + write_empty_schema_v8_store(&root); + let worker_id = WorkerId::now_v7(); + let worker_dir = root.join(WORKERS_DIR).join(worker_id.to_string()); + fs::create_dir_all(&worker_dir).unwrap(); + let identity_path = worker_dir.join(WORKER_FILE); + let pair = migrate_worker_document( + schema_v7_worker_document(worker_id), + LEGACY_SCHEMA_VERSION, + None, + &identity_path, + ) + .unwrap(); + atomic_write_json(&identity_path, &pair.identity, "write schema-v8 identity").unwrap(); + atomic_write_json( + &worker_dir.join(WORKER_EXECUTION_FILE), + &pair.execution, + "write schema-v8 execution", + ) + .unwrap(); + + let opened = FsRuntimeStore::open_or_create(root.clone(), "test-runtime").unwrap(); + let state = opened.state.unwrap(); + let worker = state.workers.get(&worker_id).unwrap(); + assert!(matches!( + worker.execution_state, + PersistedWorkerExecutionState::Available(_) + )); + assert!(!worker_dir.join(WORKER_EXECUTION_FILE).exists()); + let aggregate: WorkerAggregateRecord = + read_bounded_json(&identity_path, "read schema-v9 aggregate").unwrap(); + assert_eq!(aggregate.schema_version, SCHEMA_VERSION); + } + + #[test] + fn schema_v8_migration_fails_closed_without_execution_record() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime-store"); + write_empty_schema_v8_store(&root); + let worker_id = WorkerId::now_v7(); + let worker_dir = root.join(WORKERS_DIR).join(worker_id.to_string()); + fs::create_dir_all(&worker_dir).unwrap(); + let identity_path = worker_dir.join(WORKER_FILE); + let pair = migrate_worker_document( + schema_v7_worker_document(worker_id), + LEGACY_SCHEMA_VERSION, + None, + &identity_path, + ) + .unwrap(); + atomic_write_json(&identity_path, &pair.identity, "write schema-v8 identity").unwrap(); + + let error = FsRuntimeStore::open_or_create(root, "test-runtime").unwrap_err(); + assert!(matches!(error, RuntimeError::StoreMissing { .. })); + } + #[test] fn schema_v6_worker_migrates_through_v7_to_split_v8_records() { let worker_id = WorkerId::now_v7(); @@ -2660,16 +2946,22 @@ mod tests { .join(WORKER_FILE); let migrated = migrate_worker_document( schema_v6_worker_document(worker_id), - LEGACY_SCHEMA_VERSION, + OLDEST_SCHEMA_VERSION, None, &identity_path, ) .unwrap(); - assert_eq!(migrated.identity["schema_version"], SCHEMA_VERSION); + assert_eq!( + migrated.identity["schema_version"], + PREVIOUS_SCHEMA_VERSION + ); 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["schema_version"], + PREVIOUS_SCHEMA_VERSION + ); assert_eq!( migrated.execution["request"]["worker_id"], worker_id.to_string() @@ -2696,7 +2988,7 @@ mod tests { source["execution"]["restore_intent"] = serde_json::json!("explicit"); let migrated = - migrate_worker_document(source, LEGACY_SCHEMA_VERSION, None, &identity_path).unwrap(); + migrate_worker_document(source, OLDEST_SCHEMA_VERSION, None, &identity_path).unwrap(); assert_eq!(migrated.identity["status"], "stopped"); assert_eq!(migrated.execution["binding"], serde_json::Value::Null); @@ -2711,7 +3003,7 @@ mod tests { let error = migrate_worker_document( source, - LEGACY_SCHEMA_VERSION, + OLDEST_SCHEMA_VERSION, None, Path::new("worker.json"), ) @@ -2727,11 +3019,11 @@ mod tests { #[test] fn schema_v6_runtime_store_is_migrated_in_place() { let root = tempfile::tempdir().unwrap(); - write_empty_schema_v7_store(root.path()); + write_empty_schema_v8_store(root.path()); let runtime_path = root.path().join(RUNTIME_FILE); let mut runtime_document: serde_json::Value = read_json(&runtime_path, "read test Runtime").unwrap(); - runtime_document["schema_version"] = serde_json::json!(LEGACY_SCHEMA_VERSION); + runtime_document["schema_version"] = serde_json::json!(OLDEST_SCHEMA_VERSION); atomic_write_json(&runtime_path, &runtime_document, "write test Runtime").unwrap(); let worker_id = WorkerId::now_v7(); @@ -2767,14 +3059,15 @@ mod tests { .workers .contains_key(&worker_id) ); - assert!(worker_dir.join(WORKER_EXECUTION_FILE).exists()); - let execution: serde_json::Value = read_json( - &worker_dir.join(WORKER_EXECUTION_FILE), - "read migrated test execution", - ) - .unwrap(); - assert_eq!(execution["schema_version"], SCHEMA_VERSION); - assert_eq!(execution["binding"], serde_json::json!({})); + assert!(!worker_dir.join(WORKER_EXECUTION_FILE).exists()); + let aggregate: serde_json::Value = + read_json(&worker_dir.join(WORKER_FILE), "read migrated Worker aggregate").unwrap(); + assert_eq!(aggregate["schema_version"], SCHEMA_VERSION); + assert_eq!(aggregate["execution_state"]["state"], "available"); + assert_eq!( + aggregate["execution_state"]["execution"]["binding"], + serde_json::json!({}) + ); assert!( root.path() .join(ORPHANED_WORKERS_DIR) @@ -2795,17 +3088,23 @@ mod tests { let migrated = migrate_worker_document( schema_v7_worker_document(worker_id), - PREVIOUS_SCHEMA_VERSION, + LEGACY_SCHEMA_VERSION, None, &identity_path, ) .unwrap(); - assert_eq!(migrated.identity["schema_version"], SCHEMA_VERSION); + assert_eq!( + migrated.identity["schema_version"], + PREVIOUS_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["schema_version"], + PREVIOUS_SCHEMA_VERSION + ); assert_eq!( migrated.execution["request"]["worker_id"], worker_id.to_string() @@ -2824,7 +3123,7 @@ mod tests { 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) + let error = migrate_worker_document(source, LEGACY_SCHEMA_VERSION, None, &identity_path) .unwrap_err(); assert!( diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 27a41fef0..8b564cb7a 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2500,6 +2500,7 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode { | RuntimeError::UnsupportedConfigDeclaration { .. } => StatusCode::BAD_REQUEST, RuntimeError::WorkingDirectory(_) => StatusCode::BAD_REQUEST, RuntimeError::StoreIo { .. } + | RuntimeError::StoreCommitOutcomeUnknown { .. } | RuntimeError::StoreMissing { .. } | RuntimeError::StoreCorrupt { .. } | RuntimeError::WorkerDeletePersistenceFailed { .. } @@ -2541,6 +2542,9 @@ fn code_for_runtime_error(error: &RuntimeError) -> String { "unsupported_config_declaration".to_string() } RuntimeError::StoreIo { .. } => "store_io".to_string(), + RuntimeError::StoreCommitOutcomeUnknown { .. } => { + "store_commit_outcome_unknown".to_string() + } RuntimeError::StoreMissing { .. } => "store_missing".to_string(), RuntimeError::StoreCorrupt { .. } => "store_corrupt".to_string(), RuntimeError::StatePoisoned => "state_poisoned".to_string(), diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 1934aab76..5d517209a 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1724,6 +1724,30 @@ impl Runtime { ); match commit { Ok(worker) => Ok(RuntimeWorkerRestoreResult::accepted(worker)), + Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => { + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %error, + "Worker restore commit outcome is unknown; preserving execution evidence" + ); + if let Err(retain_error) = self.retain_restore_execution_evidence( + worker_ref, + handle, + worker_state, + workdir_attachments, + ) { + tracing::error!( + worker_id = %worker_ref.worker_id, + error = %retain_error, + "failed to persist restore execution evidence after an uncertain commit" + ); + } + Ok(RuntimeWorkerRestoreResult::failed( + WorkerRestoreState::ReconciliationRequired, + "worker_restore_commit_outcome_unknown", + "Worker restore commit outcome is uncertain; reread or retry the Worker restore", + )) + } Err(error) => { match self.cleanup_failed_restore(&backend, worker_ref, &handle) { Ok(()) => { @@ -4488,6 +4512,9 @@ fn runtime_worker_create_failure_fields( ("unsupported_config_declaration", None, None) } RuntimeError::StoreIo { .. } => ("store_io", None, None), + RuntimeError::StoreCommitOutcomeUnknown { .. } => { + ("store_commit_outcome_unknown", None, None) + } RuntimeError::StoreMissing { .. } => ("store_missing", None, None), RuntimeError::StoreCorrupt { .. } => ("store_corrupt", None, None), RuntimeError::StatePoisoned => ("state_poisoned", None, None), @@ -7807,7 +7834,7 @@ mod tests { assert!( error .to_string() - .contains("unsupported Runtime store schema version 2; expected 6, 7, or 8") + .contains("unsupported Runtime store schema version 2; expected 6, 7, 8, or 9") ); let _ = std::fs::remove_dir_all(root); @@ -7846,21 +7873,24 @@ 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_identity: serde_json::Value = + let worker_aggregate: serde_json::Value = serde_json::from_slice(&std::fs::read(worker_store_dir.join("worker.json")).unwrap()) .unwrap(); - 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_aggregate["schema_version"], serde_json::json!(9)); + assert_eq!(worker_aggregate["status"], serde_json::json!("stopped")); assert_eq!( - worker_execution["restore_intent"], + worker_aggregate["execution_state"]["state"], + serde_json::json!("available") + ); + assert_eq!( + worker_aggregate["execution_state"]["execution"]["binding"], + serde_json::json!({}) + ); + assert_eq!( + worker_aggregate["execution_state"]["execution"]["restore_intent"], serde_json::json!("explicit") ); + assert!(!worker_store_dir.join("execution.json").exists()); assert!(!root.join("events.jsonl").exists()); std::fs::write( worker_store_dir.join("observations.jsonl"), @@ -7978,12 +8008,10 @@ mod tests { missing_identity_error, RuntimeError::WorkerNotFound { .. } )); - assert!( - !restored - .worker_detail(&recoverable_legacy.worker_ref) - .unwrap() - .execution_metadata_available - ); + assert!(matches!( + restored.worker_detail(&recoverable_legacy.worker_ref), + Err(RuntimeError::WorkerNotFound { .. }) + )); let _ = std::fs::remove_dir_all(root); } @@ -8207,8 +8235,8 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_retains_identity_when_execution_metadata_is_corrupt() { - let root = fs_store_root("corrupt-execution-retains-identity"); + fn fs_store_rejects_corrupt_worker_aggregate() { + let root = fs_store_root("corrupt-worker-aggregate"); let options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), runtime_id: "test-runtime".to_string(), @@ -8233,13 +8261,13 @@ mod tests { 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); + let aggregate_path = worker_dir.join("worker.json"); + let mut aggregate: serde_json::Value = + serde_json::from_slice(&std::fs::read(&aggregate_path).unwrap()).unwrap(); + aggregate["execution_state"]["execution"]["restore_intent"] = serde_json::json!(17); std::fs::write( - &execution_path, - serde_json::to_vec_pretty(&execution).unwrap(), + &aggregate_path, + serde_json::to_vec_pretty(&aggregate).unwrap(), ) .unwrap(); @@ -8249,33 +8277,17 @@ mod tests { ) .unwrap(); let workers = restored.list_workers().unwrap(); - assert_eq!(workers.len(), 2); - let worker_summary = workers + assert_eq!(workers.len(), 1); + assert!(workers.iter().all(|summary| summary.worker_id != worker.worker_id)); + assert!(matches!( + restored.worker_detail(&worker.worker_ref), + Err(RuntimeError::WorkerNotFound { .. }) + )); + assert!(restored + .diagnostics() + .unwrap() .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()); + .any(|diagnostic| diagnostic.code == "worker_record_unavailable")); let restored_detail = restored .restore_worker(&restorable.worker_ref) @@ -8316,7 +8328,7 @@ mod tests { runtime.stop_runtime().unwrap(); - let identity: serde_json::Value = serde_json::from_slice( + let aggregate: serde_json::Value = serde_json::from_slice( &std::fs::read( root.join("workers") .join(worker.worker_id.to_string()) @@ -8325,17 +8337,18 @@ mod tests { .unwrap(), ) .unwrap(); - 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")); + assert_eq!(aggregate["status"], serde_json::json!("idle")); + assert_eq!( + aggregate["execution_state"]["execution"]["restore_intent"], + serde_json::json!("automatic") + ); + assert!( + !root + .join("workers") + .join(worker.worker_id.to_string()) + .join("execution.json") + .exists() + ); let _ = std::fs::remove_dir_all(root); } @@ -8395,7 +8408,7 @@ mod tests { #[cfg(feature = "fs-store")] #[test] - fn fs_store_migrates_schema_v7_workers_into_identity_and_execution_records() { + fn fs_store_migrates_schema_v7_workers_into_one_aggregate() { let root = fs_store_root("schema-v7-split-records"); let options = crate::fs_store::FsRuntimeStoreOptions { root: root.clone(), @@ -8427,21 +8440,20 @@ mod tests { ) .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"); + serde_json::from_slice(&std::fs::read(&worker_path).expect("worker aggregate")) + .expect("worker aggregate json"); worker_identity["schema_version"] = serde_json::json!(7); + let mut worker_execution = worker_identity + .as_object_mut() + .expect("worker aggregate object") + .remove("execution_state") + .and_then(|state| state.get("execution").cloned()) + .expect("schema v9 execution data"); 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"); + .expect("schema v9 execution request"); worker_identity["request"] = request; worker_identity["execution"] = worker_execution; std::fs::write( @@ -8449,7 +8461,7 @@ mod tests { serde_json::to_vec_pretty(&worker_identity).expect("worker record bytes"), ) .expect("write worker record"); - std::fs::remove_file(&execution_path).expect("remove split execution record"); + assert!(!execution_path.exists()); let plan = crate::fs_store::FsRuntimeStore::migration_plan(&options) .expect("read-only migration preflight"); @@ -8469,21 +8481,77 @@ mod tests { migrated.worker_detail(&worker.worker_ref).unwrap().status, WorkerStatus::Idle ); - let migrated_identity: serde_json::Value = + let migrated_aggregate: serde_json::Value = serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap(); - 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_aggregate["schema_version"], serde_json::json!(9)); + assert!(migrated_aggregate.get("request").is_none()); + assert!(migrated_aggregate.get("execution").is_none()); assert_eq!( - migrated_execution["request"]["worker_id"], + migrated_aggregate["execution_state"]["state"], + serde_json::json!("available") + ); + assert_eq!( + migrated_aggregate["execution_state"]["execution"]["request"]["worker_id"], worker.worker_id.to_string() ); - assert_eq!(migrated_execution["binding"], serde_json::json!({})); assert_eq!( - migrated_execution["restore_intent"], + migrated_aggregate["execution_state"]["execution"]["binding"], + serde_json::json!({}) + ); + assert_eq!( + migrated_aggregate["execution_state"]["execution"]["restore_intent"], + serde_json::json!("automatic") + ); + assert!(!execution_path.exists()); + + let _ = std::fs::remove_dir_all(root); + } + + #[cfg(feature = "fs-store")] + #[test] + fn fs_store_does_not_roll_back_an_unknown_restore_commit() { + let root = fs_store_root("restore-commit-outcome-unknown"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = Runtime::with_fs_store_and_execution_backend( + crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + runtime_id: "test-runtime".to_string(), + display_name: None, + }, + backend.clone(), + ) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("uncertain restore commit")) + .unwrap(); + runtime.stop_worker(&worker.worker_ref, None).unwrap(); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); + + runtime_store(&runtime) + .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename); + let result = runtime + .restore_worker_operation(&worker.worker_ref) + .unwrap(); + + assert_eq!(result.state, WorkerRestoreState::ReconciliationRequired); + assert_eq!(*backend.stop_count.lock().unwrap(), 1); + assert_eq!( + runtime.worker_detail(&worker.worker_ref).unwrap().status, + WorkerStatus::Idle + ); + let aggregate: serde_json::Value = serde_json::from_slice( + &std::fs::read( + root.join("workers") + .join(worker.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(), + ) + .unwrap(); + assert_eq!(aggregate["status"], serde_json::json!("idle")); + assert_eq!( + aggregate["execution_state"]["execution"]["restore_intent"], serde_json::json!("automatic") );