diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 710af418..4578eb18 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -1,6 +1,6 @@ use crate::catalog::{CreateWorkerRequest, WorkerStatus}; use crate::config_bundle::ConfigBundle; -use crate::diagnostics::RuntimeDiagnostic; +use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::error::RuntimeError; use crate::execution::WorkerExecutionStatus; use crate::identity::{RuntimeId, WorkerId, WorkerRef}; @@ -244,7 +244,7 @@ impl FsRuntimeStore { pub(crate) fn load_runtime_state(&self) -> Result { let runtime_path = self.runtime_path(); - let snapshot: RuntimeSnapshot = read_json(&runtime_path, "read runtime snapshot")?; + let mut snapshot: RuntimeSnapshot = read_json(&runtime_path, "read runtime snapshot")?; snapshot.validate(&self.runtime_id, &runtime_path)?; let events = read_json_lines::(&self.events_path(), "read events")?; @@ -281,38 +281,75 @@ impl FsRuntimeStore { for entry in worker_dirs { let path = entry.path(); if !path.is_dir() { - return Err(RuntimeError::StoreCorrupt { - operation: "read worker", - path, - message: "worker path exists but is not a directory".to_string(), - }); + record_worker_load_diagnostic( + &mut snapshot, + None, + "ignored invalid worker store entry while loading runtime store", + ); + continue; } let worker_snapshot_path = path.join(WORKER_FILE); let worker_snapshot: WorkerSnapshot = - read_json(&worker_snapshot_path, "read worker snapshot")?; - worker_snapshot.validate(&self.runtime_id, &worker_snapshot_path)?; - let transcript = - read_json_lines::(&path.join(TRANSCRIPT_FILE), "read transcript")?; - for entry in &transcript { - self.ensure_worker_ref(&entry.worker_ref)?; - if entry.worker_ref.worker_id != worker_snapshot.worker_id { - return Err(RuntimeError::StoreCorrupt { - operation: "read transcript", - path: path.join(TRANSCRIPT_FILE), - message: format!( - "transcript entry belongs to worker {}, expected {}", - entry.worker_ref.worker_id, worker_snapshot.worker_id - ), - }); + match read_json(&worker_snapshot_path, "read worker snapshot") { + Ok(snapshot) => snapshot, + Err(_error) => { + record_worker_load_diagnostic( + &mut snapshot, + None, + "ignored corrupt worker snapshot while loading runtime store", + ); + continue; + } + }; + if worker_snapshot + .validate(&self.runtime_id, &worker_snapshot_path) + .is_err() + { + record_worker_load_diagnostic( + &mut snapshot, + Some(worker_snapshot.worker_ref.clone()), + "ignored invalid worker snapshot while loading runtime store", + ); + continue; + } + let transcript = match read_json_lines::( + &path.join(TRANSCRIPT_FILE), + "read transcript", + ) { + Ok(transcript) => transcript, + Err(_error) => { + record_worker_load_diagnostic( + &mut snapshot, + Some(worker_snapshot.worker_ref.clone()), + "ignored worker with unreadable transcript while loading runtime store", + ); + continue; } + }; + let mut transcript_valid = true; + for entry in &transcript { + if self.ensure_worker_ref(&entry.worker_ref).is_err() + || entry.worker_ref.worker_id != worker_snapshot.worker_id + { + transcript_valid = false; + break; + } + } + if !transcript_valid { + record_worker_load_diagnostic( + &mut snapshot, + Some(worker_snapshot.worker_ref.clone()), + "ignored worker with invalid transcript while loading runtime store", + ); + continue; } let worker = worker_snapshot.into_persisted(transcript); if workers.insert(worker.worker_id.clone(), worker).is_some() { - return Err(RuntimeError::StoreCorrupt { - operation: "read workers", - path: workers_dir.clone(), - message: "duplicate worker id in store".to_string(), - }); + record_worker_load_diagnostic( + &mut snapshot, + None, + "ignored duplicate worker snapshot while loading runtime store", + ); } } @@ -399,6 +436,22 @@ struct RuntimeSnapshot { diagnostics: Vec, } +fn record_worker_load_diagnostic( + snapshot: &mut RuntimeSnapshot, + worker_ref: Option, + message: impl Into, +) { + let id = snapshot.next_diagnostic_id; + snapshot.next_diagnostic_id = snapshot.next_diagnostic_id.saturating_add(1); + snapshot.diagnostics.push(RuntimeDiagnostic { + id, + worker_ref, + severity: DiagnosticSeverity::Warning, + code: "worker_snapshot_ignored".to_string(), + message: message.into(), + }); +} + impl RuntimeSnapshot { fn from_persisted(state: &PersistedRuntimeState) -> Self { Self { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 7e63998c..f3b069ff 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2416,14 +2416,21 @@ mod tests { worker_dirs.sort_by_key(|entry| entry.path()); std::fs::remove_file(worker_dirs[0].path().join("worker.json")).unwrap(); drop(missing_runtime); - let err = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { + let loaded = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions { root: missing_root.clone(), runtime_id: Some(missing_runtime_id), display_name: None, limits: RuntimeLimits::default(), }) - .unwrap_err(); - assert!(matches!(err, RuntimeError::StoreMissing { .. })); + .expect("invalid worker snapshot should not make runtime store unreadable"); + assert!(loaded.list_workers().unwrap().is_empty()); + assert!( + loaded + .diagnostics() + .unwrap() + .iter() + .any(|diagnostic| diagnostic.code == "worker_snapshot_ignored") + ); let _ = std::fs::remove_dir_all(missing_root); } } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index b2372515..2384fa13 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -789,7 +789,7 @@ where #[cfg(test)] mod tests { use super::*; - use std::collections::{BTreeMap, VecDeque}; + use std::collections::BTreeMap; use std::fs; use std::pin::Pin; use std::process::Command; @@ -800,7 +800,6 @@ mod tests { ConfigBundleRef, CreateWorkerRequest, DirtyStatePolicy, MaterializerKind, ProfileSelector, RepositorySelector, WorkingDirectoryRepository, WorkingDirectoryRequest, }; - use crate::execution::WorkerExecutionContext; use crate::identity::RuntimeId; use crate::management::RuntimeOptions; use crate::observation::{TranscriptQuery, TranscriptRole}; @@ -967,105 +966,6 @@ mod tests { .unwrap() } - fn handle_for_archive( - archive: &crate::profile_archive::ProfileSourceArchive, - ) -> crate::resource::BackendResourceHandle { - crate::resource::BackendResourceHandle { - kind: crate::resource::BackendResourceKind::ProfileSourceArchive, - workspace_id: "workspace-test".to_string(), - scope_id: Some("workspace-profile-source".to_string()), - runtime_id: Some("runtime-test".to_string()), - worker_id: Some("worker-test".to_string()), - resource_id: archive.reference.id.clone(), - digest: archive.reference.digest.clone(), - operation: crate::resource::BackendResourceOperation::FetchArchive, - expires_at_unix_seconds: 4_102_444_800, - nonce: "nonce-test".to_string(), - revision: archive.reference.digest.clone(), - generation: Some(1), - max_bytes: crate::resource::DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES, - content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), - redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly, - audit_correlation_id: "audit-test".to_string(), - profile_source_graph: Some(archive.reference.source_graph.clone()), - } - } - - fn response_for_archive( - handle: &crate::resource::BackendResourceHandle, - archive: &crate::profile_archive::ProfileSourceArchive, - ) -> crate::resource::BackendResourceFetchResponse { - crate::resource::BackendResourceFetchResponse { - kind: crate::resource::BackendResourceKind::ProfileSourceArchive, - resource_id: handle.resource_id.clone(), - digest: handle.digest.clone(), - content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), - bytes: archive.content.clone(), - audit_correlation_id: handle.audit_correlation_id.clone(), - } - } - - #[derive(Clone)] - struct SequencedResourceClient { - responses: Arc< - Mutex< - VecDeque< - Result< - crate::resource::BackendResourceFetchResponse, - crate::resource::BackendResourceError, - >, - >, - >, - >, - call_count: Arc, - } - - #[async_trait] - impl crate::resource::BackendResourceClient for SequencedResourceClient { - async fn fetch_resource( - &self, - _request: crate::resource::BackendResourceFetchRequest, - ) -> Result< - crate::resource::BackendResourceFetchResponse, - crate::resource::BackendResourceError, - > { - self.call_count.fetch_add(1, Ordering::SeqCst); - self.responses - .lock() - .unwrap() - .pop_front() - .expect("missing sequenced resource response") - } - } - - fn spawn_request_with_bundle( - bundle: crate::config_bundle::ConfigBundle, - ) -> WorkerExecutionSpawnRequest { - let worker_ref = crate::identity::WorkerRef::new( - RuntimeId::new("runtime-test").unwrap(), - crate::identity::WorkerId::new("worker-test").unwrap(), - ); - WorkerExecutionSpawnRequest { - worker_ref: worker_ref.clone(), - request: CreateWorkerRequest { - profile: ProfileSelector::RuntimeDefault, - profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { - archive: bundle.profile_source_archive.clone().unwrap(), - }, - config_bundle: Some(ConfigBundleRef { - id: bundle.metadata.id.clone(), - digest: bundle.metadata.digest.clone(), - }), - initial_input: None, - working_directory_request: None, - working_directory: None, - }, - context: WorkerExecutionContext::new(worker_ref, Arc::new(|_, _| panic!("unused"))), - working_directory: None, - config_bundle: Some(bundle), - } - } - fn create_request(_name: &str) -> CreateWorkerRequest { let bundle = test_bundle(); CreateWorkerRequest {