runtime: tolerate corrupt worker snapshots

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<PersistedRuntimeState, RuntimeError> {
         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::<RuntimeEvent>(&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 {
+                record_worker_load_diagnostic(
-                    operation: "read worker",
+                    &mut snapshot,
-                    path,
+                    None,
-                    message: "worker path exists but is not a directory".to_string(),
+                    "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")?;
+                match read_json(&worker_snapshot_path, "read worker snapshot") {
-            worker_snapshot.validate(&self.runtime_id, &worker_snapshot_path)?;
+                    Ok(snapshot) => snapshot,
-            let transcript =
+                    Err(_error) => {
-                read_json_lines::<TranscriptEntry>(&path.join(TRANSCRIPT_FILE), "read transcript")?;
+                        record_worker_load_diagnostic(
-            for entry in &transcript {
+                            &mut snapshot,
-                self.ensure_worker_ref(&entry.worker_ref)?;
+                            None,
-                if entry.worker_ref.worker_id != worker_snapshot.worker_id {
+                            "ignored corrupt worker snapshot while loading runtime store",
-                    return Err(RuntimeError::StoreCorrupt {
+                        );
-                        operation: "read transcript",
+                        continue;
-                        path: path.join(TRANSCRIPT_FILE),
-                        message: format!(
-                            "transcript entry belongs to worker {}, expected {}",
-                            entry.worker_ref.worker_id, worker_snapshot.worker_id
-                        ),
-                    });
                     }
+                };
+            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::<TranscriptEntry>(
+                &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 {
+                record_worker_load_diagnostic(
-                    operation: "read workers",
+                    &mut snapshot,
-                    path: workers_dir.clone(),
+                    None,
-                    message: "duplicate worker id in store".to_string(),
+                    "ignored duplicate worker snapshot while loading runtime store",
-                });
+                );
             }
         }
 
@@ -399,6 +436,22 @@ struct RuntimeSnapshot {
     diagnostics: Vec<RuntimeDiagnostic>,
 }
 
+fn record_worker_load_diagnostic(
+    snapshot: &mut RuntimeSnapshot,
+    worker_ref: Option<WorkerRef>,
+    message: impl Into<String>,
+) {
+    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 {

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();
+        .expect("invalid worker snapshot should not make runtime store unreadable");
-        assert!(matches!(err, RuntimeError::StoreMissing { .. }));
+        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);
     }
 }

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<AtomicUsize>,
-    }
-
-    #[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 {