diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index d09869450..148309e32 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -208,7 +208,9 @@ impl WorkerExecutionHandle { /// Runtime hooks available to an execution backend for one Worker. #[cfg(feature = "ws-server")] type WorkerObservationPublisher = Arc< - dyn Fn(WorkerRef, protocol::Event) -> Result + Send + Sync, + dyn Fn(WorkerRef, protocol::Event) -> Result + + Send + + Sync, >; #[cfg(feature = "ws-server")] @@ -279,12 +281,10 @@ impl WorkerExecutionContext { .map_err(|_| RuntimeError::StatePoisoned)?; match &*state { WorkerExecutionPublicationState::Pending(_) => { - let WorkerExecutionPublicationState::Pending(events) = - std::mem::replace( - &mut *state, - WorkerExecutionPublicationState::Activating(VecDeque::new()), - ) - else { + let WorkerExecutionPublicationState::Pending(events) = std::mem::replace( + &mut *state, + WorkerExecutionPublicationState::Activating(VecDeque::new()), + ) else { unreachable!() }; *state = WorkerExecutionPublicationState::Activating(events); @@ -662,10 +662,7 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { Vec::new() } - fn stop_worker_operation( - &self, - request: WorkerExecutionStopRequest, - ) -> WorkerExecutionResult { + fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult { let Some(handle) = request.handle.as_ref() else { return WorkerExecutionResult::unsupported( WorkerExecutionOperation::Stop, diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 682984fbc..8869484b4 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -134,7 +134,8 @@ impl FsRuntimeStore { #[cfg(test)] pub(crate) fn fail_next_worker_write(&self, fault: AtomicWriteFault) { - self.worker_write_fault.store(fault as u8, Ordering::Release); + self.worker_write_fault + .store(fault as u8, Ordering::Release); } pub(crate) fn open_or_create( @@ -967,12 +968,13 @@ fn migrate_worker_record( 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}"), - ) - })?; + 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 = @@ -987,12 +989,13 @@ fn migrate_worker_record( 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 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| { @@ -1965,8 +1968,9 @@ impl WorkerAggregateRecord { return Err(RuntimeError::StoreCorrupt { operation: "read Worker lifecycle reconciliation", path: path.to_path_buf(), - message: "Worker reconciliation status does not match the last settled status" - .to_string(), + message: + "Worker reconciliation status does not match the last settled status" + .to_string(), }); } WorkerExecutionRecord { @@ -2971,7 +2975,11 @@ mod tests { 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()); + assert!( + !aggregate_path + .with_file_name(WORKER_EXECUTION_FILE) + .exists() + ); } #[test] @@ -3050,10 +3058,7 @@ mod tests { ) .unwrap(); - assert_eq!( - migrated.identity["schema_version"], - PREVIOUS_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!( @@ -3158,8 +3163,11 @@ mod tests { .contains_key(&worker_id) ); 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(); + 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!( @@ -3192,10 +3200,7 @@ mod tests { ) .unwrap(); - assert_eq!( - migrated.identity["schema_version"], - PREVIOUS_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()); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 49b4ef6e5..27d0754a3 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1775,13 +1775,12 @@ impl Runtime { } if !reconcile - && let Err(error) = - self.begin_restore_operation( - worker_ref, - operation_id, - operation_mode, - candidate_context.clone(), - ) + && let Err(error) = self.begin_restore_operation( + worker_ref, + operation_id, + operation_mode, + candidate_context.clone(), + ) { if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. }) { return Ok(RuntimeWorkerRestoreResult::failed( @@ -1813,8 +1812,7 @@ impl Runtime { ); match commit { Ok(worker) => { - if let Err(error) = - backend.activate_restored_worker(operation_id, &handle) + if let Err(error) = backend.activate_restored_worker(operation_id, &handle) { tracing::error!( worker_id = %worker_ref.worker_id, @@ -1865,7 +1863,8 @@ impl Runtime { error = %error, "Worker restore commit failed and was rolled back" ); - if let Err(settle_error) = self.settle_restore_rollback(worker_ref) { + if let Err(settle_error) = self.settle_restore_rollback(worker_ref) + { tracing::error!( worker_id = %worker_ref.worker_id, error = %settle_error, @@ -1983,11 +1982,8 @@ impl Runtime { "failed to persist restore execution evidence; in-process reconciliation remains required" ); } - } else if let Err(retain_error) = - self.retain_restore_reconciliation_pending( - worker_ref, - candidate_context.clone(), - ) + } else if let Err(retain_error) = self + .retain_restore_reconciliation_pending(worker_ref, candidate_context.clone()) { tracing::error!( worker_id = %worker_ref.worker_id, @@ -2002,10 +1998,9 @@ impl Runtime { )) } WorkerExecutionSpawnResult::Errored(_result) => { - if let Err(retain_error) = self.retain_restore_reconciliation_pending( - worker_ref, - candidate_context.clone(), - ) { + if let Err(retain_error) = self + .retain_restore_reconciliation_pending(worker_ref, candidate_context.clone()) + { tracing::error!( worker_id = %worker_ref.worker_id, error = %retain_error, @@ -3110,7 +3105,11 @@ impl Runtime { message: "runtime has no execution backend".to_string(), } })?; - (backend, worker.execution_handle.clone(), pending.operation_id) + ( + backend, + worker.execution_handle.clone(), + pending.operation_id, + ) }; let result = backend.stop_worker_operation(WorkerExecutionStopRequest { operation_id, @@ -3725,37 +3724,36 @@ impl RuntimeState { None, None, ), - PersistedWorkerExecutionState::ReconciliationRequired(operation) => { - match operation { - PersistedWorkerLifecycleOperation::Restore(operation) => { - let restore_intent = - restore_intent_for_status(operation.last_settled_status); - ( - Some(operation.request), - true, - operation.binding.is_some(), - restore_intent, - Some(PendingWorkerRestore { - operation_id: operation.operation_id, - mode: operation.mode.into(), - last_settled_status: operation.last_settled_status, - }), - None, - ) - } - PersistedWorkerLifecycleOperation::Stop(operation) => ( + PersistedWorkerExecutionState::ReconciliationRequired(operation) => match operation + { + PersistedWorkerLifecycleOperation::Restore(operation) => { + let restore_intent = + restore_intent_for_status(operation.last_settled_status); + ( Some(operation.request), true, operation.binding.is_some(), - operation.restore_intent, - None, - Some(PendingWorkerStop { + restore_intent, + Some(PendingWorkerRestore { operation_id: operation.operation_id, + mode: operation.mode.into(), last_settled_status: operation.last_settled_status, }), - ), + None, + ) } - } + PersistedWorkerLifecycleOperation::Stop(operation) => ( + Some(operation.request), + true, + operation.binding.is_some(), + operation.restore_intent, + None, + Some(PendingWorkerStop { + operation_id: operation.operation_id, + last_settled_status: operation.last_settled_status, + }), + ), + }, PersistedWorkerExecutionState::Unavailable => ( None, false, @@ -4782,8 +4780,7 @@ impl WorkerRecord { last_settled_status: pending.last_settled_status, }), ) - } else if let (Some(pending), Some(request)) = - (self.pending_restore, self.request.clone()) + } else if let (Some(pending), Some(request)) = (self.pending_restore, self.request.clone()) { PersistedWorkerExecutionState::ReconciliationRequired( PersistedWorkerLifecycleOperation::Restore(PersistedWorkerRestoreOperation { @@ -7397,11 +7394,8 @@ mod tests { display_name: None, }; let backend = Arc::new(TestExecutionBackend::default()); - let runtime = Runtime::with_fs_store_and_execution_backend( - options.clone(), - backend.clone(), - ) - .unwrap(); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()).unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); let worker = runtime .create_worker(task_request("reconcile pending stop")) @@ -7409,9 +7403,7 @@ mod tests { runtime_store(&runtime) .fail_next_worker_write(crate::fs_store::AtomicWriteFault::AfterRename); - let error = runtime - .stop_worker(&worker.worker_ref, None) - .unwrap_err(); + let error = runtime.stop_worker(&worker.worker_ref, None).unwrap_err(); assert!(matches!( error, RuntimeError::StoreCommitOutcomeUnknown { .. } @@ -7480,11 +7472,9 @@ mod tests { drop(reconciled); let restarted_backend = Arc::new(TestExecutionBackend::default()); - let restarted = Runtime::with_fs_store_and_execution_backend( - options, - restarted_backend.clone(), - ) - .unwrap(); + let restarted = + Runtime::with_fs_store_and_execution_backend(options, restarted_backend.clone()) + .unwrap(); assert_eq!(*restarted_backend.stop_count.lock().unwrap(), 0); assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 0); assert_eq!( @@ -7645,19 +7635,20 @@ mod tests { let cursor_before = runtime .worker_observation_cursor_now(&worker.worker_ref) .unwrap(); - backend.restore_event.lock().unwrap().replace( - protocol::Event::WorkerState { + backend + .restore_event + .lock() + .unwrap() + .replace(protocol::Event::WorkerState { snapshot: protocol::WorkerStatus::Running.into(), - }, - ); + }); let gate = Arc::new(RestoreGate::default()); *backend.restore_gate.lock().unwrap() = Some(gate.clone()); let restoring_runtime = runtime.clone(); let restoring_ref = worker.worker_ref.clone(); - let restoring = std::thread::spawn(move || { - restoring_runtime.restore_worker_operation(&restoring_ref) - }); + let restoring = + std::thread::spawn(move || restoring_runtime.restore_worker_operation(&restoring_ref)); assert!(gate.wait_for_entered(1, std::time::Duration::from_secs(2))); let pending = runtime.worker_detail(&worker.worker_ref).unwrap(); @@ -8928,12 +8919,16 @@ mod tests { .create_worker(task_request("rolled back restore")) .unwrap(); runtime.stop_worker(&worker.worker_ref, None).unwrap(); - backend.restore_result.lock().unwrap().replace( - WorkerExecutionSpawnResult::RolledBack(WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - "controller launch failed and cleanup completed", - )), - ); + backend + .restore_result + .lock() + .unwrap() + .replace(WorkerExecutionSpawnResult::RolledBack( + WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + "controller launch failed and cleanup completed", + ), + )); let result = runtime .restore_worker_operation(&worker.worker_ref) @@ -9149,16 +9144,22 @@ mod tests { .unwrap(); let workers = restored.list_workers().unwrap(); assert_eq!(workers.len(), 1); - assert!(workers.iter().all(|summary| summary.worker_id != worker.worker_id)); + 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() - .any(|diagnostic| diagnostic.code == "worker_record_unavailable")); + assert!( + restored + .diagnostics() + .unwrap() + .iter() + .any(|diagnostic| diagnostic.code == "worker_record_unavailable") + ); let restored_detail = restored .restore_worker(&restorable.worker_ref) diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 568246be3..4f62c9199 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1442,8 +1442,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { } let workspace_client = worker.workspace_client_handle(); - let run_dir = - self.worker_restore_run_dir(&request.worker_ref, request.operation_id)?; + let run_dir = self.worker_restore_run_dir(&request.worker_ref, request.operation_id)?; let bash_output_dir = bash_output_dir_for_worker_id(&request.worker_ref.worker_id); let started = PreparedWorker::new( worker, @@ -1483,11 +1482,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { workspace_id: Option<&str>, handle: &WorkerHandle, ) { - self.observation_hub.register( - worker_ref.clone(), - workspace_id.map(str::to_string), - handle, - ); + self.observation_hub + .register(worker_ref.clone(), workspace_id.map(str::to_string), handle); } } @@ -2953,13 +2949,10 @@ where self.send_method(WorkerExecutionOperation::ProtocolMethod, worker, method) } - fn stop_worker_operation( - &self, - request: WorkerExecutionStopRequest, - ) -> WorkerExecutionResult { - let handle = request.handle.unwrap_or_else(|| { - WorkerExecutionHandle::new(request.worker_ref, self.backend_id()) - }); + fn stop_worker_operation(&self, request: WorkerExecutionStopRequest) -> WorkerExecutionResult { + let handle = request + .handle + .unwrap_or_else(|| WorkerExecutionHandle::new(request.worker_ref, self.backend_id())); self.stop_worker(&handle) } @@ -3865,9 +3858,7 @@ mod tests { worker_ref, Arc::new(|worker_ref, payload| { Ok(crate::observation::WorkerObservationEvent::new( - 1, - worker_ref, - payload, + 1, worker_ref, payload, )) }), ) @@ -4894,11 +4885,9 @@ mod tests { observed_workspace_clients: observed_workspace_clients.clone(), }; let backend = Arc::new(WorkerRuntimeExecutionBackend::new(factory).unwrap()); - let runtime = EmbeddedRuntime::with_execution_backend( - RuntimeOptions::default(), - backend.clone(), - ) - .unwrap(); + let runtime = + EmbeddedRuntime::with_execution_backend(RuntimeOptions::default(), backend.clone()) + .unwrap(); runtime.store_config_bundle(test_bundle()).unwrap(); let request = create_request("idempotent restore reconciliation"); let worker = runtime.create_worker(request.clone()).unwrap(); diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 2f263ff81..a9a769d0e 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -2222,8 +2222,7 @@ impl EmbeddedWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: summary.workspace_id.clone(), }, - availability: - protocol::subscription::SubscriptionWorkerAvailability::Observed, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, state: embedded_worker_state_label( summary.status, summary.execution_metadata_available, @@ -2271,8 +2270,7 @@ impl EmbeddedWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: detail.workspace_id.clone(), }, - availability: - protocol::subscription::SubscriptionWorkerAvailability::Observed, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, state: embedded_worker_state_label(detail.status, detail.execution_metadata_available) .to_string(), worker_state: detail.worker_state.clone(), @@ -4064,8 +4062,7 @@ impl RemoteWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: summary.workspace_id.clone(), }, - availability: - protocol::subscription::SubscriptionWorkerAvailability::Observed, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, state: embedded_worker_state_label( summary.status, summary.execution_metadata_available, @@ -4111,8 +4108,7 @@ impl RemoteWorkerRuntime { identity: "runtime_registry_worker".to_string(), workspace_id: detail.workspace_id.clone(), }, - availability: - protocol::subscription::SubscriptionWorkerAvailability::Observed, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, state: embedded_worker_state_label(detail.status, detail.execution_metadata_available) .to_string(), worker_state: detail.worker_state.clone(), @@ -5979,18 +5975,14 @@ mod tests { #[test] fn embedded_store_commit_unknown_diagnostic_is_path_free() { - let diagnostic = embedded_runtime_diagnostic( - &EmbeddedRuntimeError::StoreCommitOutcomeUnknown { + let diagnostic = + embedded_runtime_diagnostic(&EmbeddedRuntimeError::StoreCommitOutcomeUnknown { operation: "sync Worker aggregate directory", path: std::path::PathBuf::from("/private/runtime/workers/worker.json"), source: std::io::Error::other("raw fsync failure"), - }, - ); + }); - assert_eq!( - diagnostic.code, - "embedded_runtime_store_commit_unknown" - ); + assert_eq!(diagnostic.code, "embedded_runtime_store_commit_unknown"); assert_eq!(diagnostic.severity, HostDiagnosticSeverity::Error); assert!(!diagnostic.message.contains("/private/runtime")); assert!(!diagnostic.message.contains("raw fsync failure")); @@ -6547,8 +6539,7 @@ mod tests { identity: host_id.to_string(), workspace_id: None, }, - availability: - protocol::subscription::SubscriptionWorkerAvailability::Observed, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, state: "available".to_string(), worker_state: None, last_seen_at: None,