From 5894e73ab9cc9d78b7c70d65e2f5f95c8079a9d2 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 29 Jul 2026 02:54:23 +0900 Subject: [PATCH] runtime: expose restoreable workers --- crates/client/src/backend_runtime.rs | 142 ++++++++++++ crates/client/src/lib.rs | 7 +- crates/worker-runtime/src/catalog.rs | 45 ++++ crates/worker-runtime/src/http_server.rs | 24 ++- crates/worker-runtime/src/runtime.rs | 225 ++++++++++++++++++- crates/workspace-server/src/hosts.rs | 261 ++++++++++++++++++++++- crates/workspace-server/src/server.rs | 127 ++++++++++- 7 files changed, 809 insertions(+), 22 deletions(-) diff --git a/crates/client/src/backend_runtime.rs b/crates/client/src/backend_runtime.rs index 40eb2259..99fbe91d 100644 --- a/crates/client/src/backend_runtime.rs +++ b/crates/client/src/backend_runtime.rs @@ -168,6 +168,56 @@ pub struct BackendWorkerSummary { pub diagnostics: Vec, } +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct BackendWorkerRestoreableOperations { + pub can_attach: bool, + pub can_restore: bool, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct BackendWorkerRestoreableCandidate { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub host_id: String, + #[serde(default)] + pub display_name: String, + pub label: String, + #[serde(default)] + pub role: Option, + #[serde(default)] + pub profile: Option, + #[serde(default)] + pub singleton_key: Option, + #[serde(default)] + pub tags: Vec, + pub workspace: BackendWorkerWorkspaceSummary, + pub state: String, + pub restoreability: String, + pub operations: BackendWorkerRestoreableOperations, + #[serde(default)] + pub working_directory: Option, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct BackendWorkerRestoreResult { + pub state: String, + #[serde(default)] + pub worker: Option, + #[serde(default)] + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct BackendWorkerRestoreResponse { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub result: BackendWorkerRestoreResult, +} + #[derive(Debug)] pub struct BackendRuntimeClient { target: BackendRuntimeTarget, @@ -266,6 +316,45 @@ pub async fn list_backend_workers( }) } +pub async fn list_backend_restoreable_workers( + target: &BackendRuntimeListTarget, +) -> Result, BackendRuntimeClientError> +{ + validate_list_target(target)?; + let Some(runtime_id) = target.runtime_id.as_deref() else { + return Err(BackendRuntimeClientError::InvalidTarget( + "restoreable worker listing requires a runtime id".to_string(), + )); + }; + let http = reqwest::Client::new(); + let path = backend_runtime_restoreable_workers_path(target.workspace_id.as_deref(), runtime_id); + let url = join_base_and_path(&target.base_url, &path); + Ok(http + .get(url) + .send() + .await? + .error_for_status()? + .json::>() + .await?) +} + +pub async fn restore_backend_worker( + target: &BackendRuntimeTarget, +) -> Result { + validate_target(target)?; + let http = reqwest::Client::new(); + let path = backend_runtime_worker_restore_path(None, &target.runtime_id, &target.worker_id); + let url = join_base_and_path(&target.base_url, &path); + Ok(http + .post(url) + .json(&serde_json::json!({})) + .send() + .await? + .error_for_status()? + .json::() + .await?) +} + impl BackendRuntimeClient { pub async fn connect(target: BackendRuntimeTarget) -> Result { validate_target(&target)?; @@ -479,6 +568,43 @@ fn backend_runtime_workers_path(workspace_id: Option<&str>, runtime_id: &str) -> } } +fn backend_runtime_restoreable_workers_path( + workspace_id: Option<&str>, + runtime_id: &str, +) -> String { + match workspace_id { + Some(workspace_id) => format!( + "/api/w/{}/runtimes/{}/workers/restoreable", + path_segment_encode(workspace_id), + path_segment_encode(runtime_id) + ), + None => format!( + "/api/runtimes/{}/workers/restoreable", + path_segment_encode(runtime_id) + ), + } +} + +fn backend_runtime_worker_restore_path( + workspace_id: Option<&str>, + runtime_id: &str, + worker_id: &str, +) -> String { + match workspace_id { + Some(workspace_id) => format!( + "/api/w/{}/runtimes/{}/workers/{}/restore", + path_segment_encode(workspace_id), + path_segment_encode(runtime_id), + path_segment_encode(worker_id) + ), + None => format!( + "/api/runtimes/{}/workers/{}/restore", + path_segment_encode(runtime_id), + path_segment_encode(worker_id) + ), + } +} + fn protocol_ws_url(target: &BackendRuntimeTarget) -> String { let path = format!( "/api/runtimes/{}/workers/{}/protocol/ws", @@ -542,4 +668,20 @@ mod tests { "ws://127.0.0.1:8787/api/runtimes/runtime%2Fone/workers/worker%20one/protocol/ws" ); } + + #[test] + fn restoreable_workers_path_can_be_workspace_scoped() { + assert_eq!( + backend_runtime_restoreable_workers_path(Some("team main"), "runtime/one"), + "/api/w/team%20main/runtimes/runtime%2Fone/workers/restoreable" + ); + } + + #[test] + fn restore_worker_path_uses_backend_runtime_worker_identity() { + assert_eq!( + backend_runtime_worker_restore_path(None, "runtime/one", "worker one"), + "/api/runtimes/runtime%2Fone/workers/worker%20one/restore" + ); + } } diff --git a/crates/client/src/lib.rs b/crates/client/src/lib.rs index 84ffa477..9a63a0f7 100644 --- a/crates/client/src/lib.rs +++ b/crates/client/src/lib.rs @@ -23,8 +23,11 @@ pub use backend_auth::{ pub use backend_runtime::{ BackendRuntimeClient, BackendRuntimeClientError, BackendRuntimeListResponse, BackendRuntimeListTarget, BackendRuntimeSummary, BackendRuntimeTarget, - BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerSummary, - BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, list_backend_workers, + BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, + BackendWorkerRestoreResponse, BackendWorkerRestoreResult, BackendWorkerRestoreableCandidate, + BackendWorkerRestoreableOperations, BackendWorkerSummary, BackendWorkerWorkspaceSummary, + BackendWorkingDirectorySummary, list_backend_restoreable_workers, list_backend_workers, + restore_backend_worker, }; pub use runtime_command::WorkerRuntimeCommand; pub use target::{ diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index fa69b9da..f7e40543 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -257,6 +257,51 @@ pub struct WorkerDetail { pub last_event_id: u64, } +/// Current Runtime-side restoreability classification for a known Worker. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerRestoreability { + /// The Worker already has a live execution handle and can be attached to. + Alive, + /// The Worker has durable Runtime catalog state and can currently be restored. + Restoreable, + /// The Worker is known, but restore cannot be attempted in the current state. + Unrestorable, +} + +/// Operations currently available for a restoreability candidate. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRestoreableOperations { + pub can_attach: bool, + pub can_restore: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkerRestoreableDiagnosticSeverity { + Info, + Warning, + Error, +} + +/// Bounded diagnostic attached to a restoreability candidate. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRestoreableDiagnostic { + pub severity: WorkerRestoreableDiagnosticSeverity, + pub code: String, + pub message: String, +} + +/// Runtime-local Worker restoreability projection. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerRestoreableCandidate { + pub worker: WorkerSummary, + pub restoreability: WorkerRestoreability, + pub operations: WorkerRestoreableOperations, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub diagnostics: Vec, +} + /// Acknowledgement returned by stop/cancel lifecycle operations. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerLifecycleAck { diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 2c8a0797..41e36e4c 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -11,8 +11,8 @@ use crate::auth::{ RuntimeAuthContext, RuntimeHttpAuthConfig, unix_now_seconds, verify_capability_token, }; use crate::catalog::{ - ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary, - WorkingDirectoryRequest, WorkingDirectoryStatus, + ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, + WorkerRestoreableCandidate, WorkerSummary, WorkingDirectoryRequest, WorkingDirectoryStatus, }; use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; use crate::error::RuntimeError; @@ -174,6 +174,7 @@ pub fn runtime_http_router_with_auth( get(get_working_directory).delete(cleanup_working_directory), ) .route("/v1/workers", get(list_workers).post(create_worker)) + .route("/v1/workers/restoreable", get(list_restoreable_workers)) .route( "/v1/workers/{worker_id}", get(get_worker).delete(delete_worker), @@ -240,6 +241,12 @@ pub struct RuntimeHttpWorkersResponse { pub workers: Vec, } +/// `GET /v1/workers/restoreable` response. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeHttpRestoreableWorkersResponse { + pub workers: Vec, +} + /// `GET /v1/working-directories` response. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct RuntimeHttpWorkingDirectoriesResponse { @@ -380,6 +387,16 @@ async fn list_workers( Ok(Json(RuntimeHttpWorkersResponse { workers })) } +async fn list_restoreable_workers( + State(state): State, +) -> RestResult { + let workers = state + .runtime + .list_restoreable_workers() + .map_err(RuntimeHttpRestError::runtime)?; + Ok(Json(RuntimeHttpRestoreableWorkersResponse { workers })) +} + async fn list_working_directories( State(state): State, ) -> RestResult { @@ -829,6 +846,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s if path == "/v1/workers" && *method == Method::GET { return Some("workers:list"); } + if path == "/v1/workers/restoreable" && *method == Method::GET { + return Some("workers:list"); + } if path == "/v1/workers" && *method == Method::POST { return Some("workers:create"); } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index ca3f4642..dc031c3a 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1,15 +1,14 @@ use crate::catalog::{ - ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus, - WorkerSummary, WorkingDirectoryRequest, - WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, + ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerRestoreability, + WorkerRestoreableCandidate, WorkerRestoreableDiagnostic, WorkerRestoreableDiagnosticSeverity, + WorkerRestoreableOperations, WorkerStatus, WorkerSummary, WorkingDirectoryRequest, + WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkingDirectoryStatusKind, }; use crate::config_bundle::{ ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle, validate_config_bundle_ref, }; -#[cfg(feature = "fs-store")] -use crate::diagnostics::DiagnosticSeverity; -use crate::diagnostics::RuntimeDiagnostic; +use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic}; use crate::error::RuntimeError; use crate::execution::WorkerExecutionRestoreRequest; use crate::execution::{ @@ -437,6 +436,26 @@ impl Runtime { Ok(state.workers.values().map(WorkerRecord::summary).collect()) } + /// List Workers known to this Runtime as restore/attach candidates. + pub fn list_restoreable_workers( + &self, + ) -> Result, RuntimeError> { + let state = self.lock()?; + let runtime_running = state.status == RuntimeStatus::Running; + let execution_backend_available = state.execution_backend.is_some(); + Ok(state + .workers + .values() + .map(|worker| { + worker.restoreable_candidate( + runtime_running, + execution_backend_available, + &state.diagnostics, + ) + }) + .collect()) + } + /// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime. pub fn worker_detail(&self, worker_ref: &WorkerRef) -> Result { let state = self.lock()?; @@ -1687,6 +1706,110 @@ impl WorkerRecord { } } + fn restoreable_candidate( + &self, + runtime_running: bool, + execution_backend_available: bool, + runtime_diagnostics: &[RuntimeDiagnostic], + ) -> WorkerRestoreableCandidate { + let mut diagnostics = runtime_diagnostics + .iter() + .filter(|diagnostic| diagnostic.worker_ref.as_ref() == Some(&self.worker_ref)) + .take(8) + .map(worker_restoreable_diagnostic_from_runtime) + .collect::>(); + + let (restoreability, operations) = + if self.execution_handle.is_some() && self.status.is_active() { + ( + WorkerRestoreability::Alive, + WorkerRestoreableOperations { + can_attach: true, + can_restore: false, + }, + ) + } else if self.status == WorkerStatus::Cancelled { + diagnostics.push(worker_restoreable_diagnostic( + WorkerRestoreableDiagnosticSeverity::Warning, + "worker_cancelled", + format!( + "worker {} is cancelled and cannot be restored", + self.worker_id + ), + )); + ( + WorkerRestoreability::Unrestorable, + WorkerRestoreableOperations::default(), + ) + } else if let Some(workdir) = self.working_directory.as_ref() { + if matches!( + workdir.summary.status, + WorkingDirectoryStatusKind::Corrupted | WorkingDirectoryStatusKind::NotFound + ) { + diagnostics.push(worker_restoreable_diagnostic( + WorkerRestoreableDiagnosticSeverity::Error, + "worker_working_directory_unrestorable", + format!( + "worker {} working directory {} is {:?}", + self.worker_id, + workdir.summary.working_directory_id, + workdir.summary.status + ), + )); + ( + WorkerRestoreability::Unrestorable, + WorkerRestoreableOperations::default(), + ) + } else if runtime_running && execution_backend_available { + ( + WorkerRestoreability::Restoreable, + WorkerRestoreableOperations { + can_attach: false, + can_restore: true, + }, + ) + } else { + push_restore_unavailable_diagnostic( + &mut diagnostics, + self.worker_id, + runtime_running, + execution_backend_available, + ); + ( + WorkerRestoreability::Unrestorable, + WorkerRestoreableOperations::default(), + ) + } + } else if runtime_running && execution_backend_available { + ( + WorkerRestoreability::Restoreable, + WorkerRestoreableOperations { + can_attach: false, + can_restore: true, + }, + ) + } else { + push_restore_unavailable_diagnostic( + &mut diagnostics, + self.worker_id, + runtime_running, + execution_backend_available, + ); + ( + WorkerRestoreability::Unrestorable, + WorkerRestoreableOperations::default(), + ) + }; + + diagnostics.truncate(8); + WorkerRestoreableCandidate { + worker: self.summary(), + restoreability, + operations, + diagnostics, + } + } + #[cfg(feature = "fs-store")] fn persisted_record(&self) -> PersistedWorkerRecord { PersistedWorkerRecord { @@ -1699,6 +1822,55 @@ impl WorkerRecord { } } +fn worker_restoreable_diagnostic( + severity: WorkerRestoreableDiagnosticSeverity, + code: impl Into, + message: impl Into, +) -> WorkerRestoreableDiagnostic { + WorkerRestoreableDiagnostic { + severity, + code: code.into(), + message: message.into(), + } +} + +fn worker_restoreable_diagnostic_from_runtime( + diagnostic: &RuntimeDiagnostic, +) -> WorkerRestoreableDiagnostic { + worker_restoreable_diagnostic( + match diagnostic.severity { + DiagnosticSeverity::Info => WorkerRestoreableDiagnosticSeverity::Info, + DiagnosticSeverity::Warning => WorkerRestoreableDiagnosticSeverity::Warning, + DiagnosticSeverity::Error => WorkerRestoreableDiagnosticSeverity::Error, + }, + diagnostic.code.clone(), + diagnostic.message.clone(), + ) +} + +fn push_restore_unavailable_diagnostic( + diagnostics: &mut Vec, + worker_id: WorkerId, + runtime_running: bool, + execution_backend_available: bool, +) { + if !runtime_running { + diagnostics.push(worker_restoreable_diagnostic( + WorkerRestoreableDiagnosticSeverity::Error, + "runtime_not_running", + format!("worker {worker_id} cannot be restored because the runtime is not running"), + )); + } else if !execution_backend_available { + diagnostics.push(worker_restoreable_diagnostic( + WorkerRestoreableDiagnosticSeverity::Error, + "runtime_execution_backend_unavailable", + format!( + "worker {worker_id} cannot be restored because the runtime has no execution backend" + ), + )); + } +} + fn worker_status_from_run_state(run_state: WorkerExecutionRunState) -> WorkerStatus { match run_state { WorkerExecutionRunState::Idle => WorkerStatus::Idle, @@ -2049,6 +2221,47 @@ mod tests { assert_eq!(fetched.profile, detail.profile); } + #[test] + fn restoreable_worker_projection_reports_attach_and_restore_operations() { + let runtime = runtime_with_backend(); + let detail = runtime.create_worker(task_request("restoreable")).unwrap(); + + let alive = runtime.list_restoreable_workers().unwrap(); + assert_eq!(alive.len(), 1); + assert_eq!(alive[0].worker.worker_ref, detail.worker_ref); + assert_eq!(alive[0].restoreability, WorkerRestoreability::Alive); + assert!(alive[0].operations.can_attach); + assert!(!alive[0].operations.can_restore); + + runtime.stop_worker(&detail.worker_ref, None).unwrap(); + let stopped = runtime.list_restoreable_workers().unwrap(); + assert_eq!(stopped[0].restoreability, WorkerRestoreability::Restoreable); + assert!(!stopped[0].operations.can_attach); + assert!(stopped[0].operations.can_restore); + } + + #[test] + fn cancelled_worker_projection_is_unrestorable() { + let runtime = runtime_with_backend(); + let detail = runtime.create_worker(task_request("cancelled")).unwrap(); + runtime.cancel_worker(&detail.worker_ref, None).unwrap(); + + let candidates = runtime.list_restoreable_workers().unwrap(); + assert_eq!(candidates.len(), 1); + assert_eq!( + candidates[0].restoreability, + WorkerRestoreability::Unrestorable + ); + assert!(!candidates[0].operations.can_attach); + assert!(!candidates[0].operations.can_restore); + assert!( + candidates[0] + .diagnostics + .iter() + .any(|diagnostic| diagnostic.code == "worker_cancelled") + ); + } + #[test] fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() { let runtime = runtime_with_backend(); diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 0a7a7eee..8c4bf5c4 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -16,9 +16,12 @@ use std::{ use worker_runtime::auth::{CapabilityTokenSigner, capability_claims}; use worker_runtime::catalog::{ ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef, - ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail, - WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRequest, - WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef, + ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail, WorkerRestoreability, + WorkerRestoreableCandidate as EmbeddedWorkerRestoreableCandidate, + WorkerRestoreableDiagnostic as EmbeddedWorkerRestoreableDiagnostic, + WorkerRestoreableDiagnosticSeverity as EmbeddedWorkerRestoreableDiagnosticSeverity, + WorkerRestoreableOperations, WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, + WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef, }; use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; #[cfg(test)] @@ -31,11 +34,12 @@ use worker_runtime::execution::WorkerExecutionRunState; use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::http_server::{ RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, - RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerCompletionsRequest, - RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse, - RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest, - RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, - RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, + RuntimeHttpErrorResponse, RuntimeHttpRestoreableWorkersResponse, RuntimeHttpSummaryResponse, + RuntimeHttpWorkerCompletionsRequest, RuntimeHttpWorkerCompletionsResponse, + RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse, + RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse, + RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse, + RuntimeHttpWorkingDirectoryResponse, }; use worker_runtime::identity::{WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef}; use worker_runtime::interaction::{ @@ -253,6 +257,36 @@ pub struct WorkerSummary { pub diagnostics: Vec, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRestoreableCandidate { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub host_id: String, + pub display_name: String, + pub label: String, + pub role: Option, + pub profile: Option, + pub singleton_key: Option, + #[serde(default)] + pub tags: Vec, + pub workspace: WorkerWorkspaceSummary, + pub state: String, + pub restoreability: WorkerRestoreability, + pub operations: WorkerRestoreableOperations, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub working_directory: Option, + pub diagnostics: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkerRestoreResult { + pub state: WorkerOperationState, + #[serde(skip_serializing_if = "Option::is_none")] + pub worker: Option, + pub diagnostics: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct RuntimeList { pub items: Vec, @@ -270,6 +304,11 @@ fn is_retired_companion_worker(worker: &WorkerSummary) -> bool { || worker.profile.as_deref() == Some("builtin:companion") } +fn is_retired_companion_restoreable_worker(worker: &WorkerRestoreableCandidate) -> bool { + worker.role.as_deref() == Some("builtin:companion") + || worker.profile.as_deref() == Some("builtin:companion") +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerLookupResult { #[serde(skip_serializing_if = "Option::is_none")] @@ -576,8 +615,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { fn list_workers(&self, limit: usize) -> RuntimeList; + fn list_restoreable_workers(&self, _limit: usize) -> RuntimeList { + RuntimeList::new(Vec::new(), Vec::new()) + } + fn worker(&self, worker_id: &str) -> WorkerLookupResult; + fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { + WorkerRestoreResult { + state: WorkerOperationState::Unsupported, + worker: None, + diagnostics: vec![diagnostic( + "worker_restore_unsupported", + DiagnosticSeverity::Info, + format!("runtime does not implement worker restore for `{worker_id}`"), + )], + } + } + fn create_working_directory( &self, _request: WorkingDirectoryRequest, @@ -955,6 +1010,26 @@ impl RuntimeRegistry { Ok(RuntimeList::new(items, diagnostics)) } + pub fn list_restoreable_workers_for_runtime( + &self, + runtime_id: &str, + limit: usize, + ) -> Result, RuntimeRegistryError> { + validate_backend_identifier("runtime_id", runtime_id)?; + let runtime = self.runtime(runtime_id)?; + let worker_list = runtime.list_restoreable_workers(limit); + let mut items: Vec<_> = worker_list + .items + .into_iter() + .filter(|worker| !is_retired_companion_restoreable_worker(worker)) + .take(limit) + .collect(); + items.truncate(limit); + let mut diagnostics = worker_list.diagnostics; + diagnostics.truncate(MAX_DIAGNOSTICS); + Ok(RuntimeList::new(items, diagnostics)) + } + pub fn list_workers_for_host( &self, host_id: &str, @@ -1015,6 +1090,17 @@ impl RuntimeRegistry { Ok(worker) } + pub fn restore_worker( + &self, + runtime_id: &str, + worker_id: &str, + ) -> Result { + validate_backend_identifier("runtime_id", runtime_id)?; + validate_backend_identifier("worker_id", worker_id)?; + let runtime = self.runtime(runtime_id)?; + Ok(runtime.restore_worker(worker_id)) + } + pub fn spawn_worker( &self, runtime_id: &str, @@ -1414,6 +1500,39 @@ impl EmbeddedWorkerRuntime { diagnostics: embedded_worker_projection_diagnostics(), } } + + fn map_worker_restoreable_candidate( + &self, + candidate: EmbeddedWorkerRestoreableCandidate, + ) -> WorkerRestoreableCandidate { + let worker = self.map_worker_summary(candidate.worker); + let mut diagnostics = worker.diagnostics.clone(); + diagnostics.extend( + candidate + .diagnostics + .iter() + .map(embedded_restoreable_diagnostic), + ); + diagnostics.truncate(MAX_DIAGNOSTICS); + WorkerRestoreableCandidate { + workspace_id: self.workspace_id.clone(), + runtime_id: worker.runtime_id, + worker_id: worker.worker_id, + host_id: worker.host_id, + display_name: worker.display_name, + label: worker.label, + role: worker.role, + profile: worker.profile, + singleton_key: worker.singleton_key, + tags: worker.tags, + workspace: worker.workspace, + state: worker.state, + restoreability: candidate.restoreability, + operations: candidate.operations, + working_directory: worker.working_directory, + diagnostics, + } + } } impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { @@ -1501,6 +1620,23 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { } } + fn list_restoreable_workers(&self, limit: usize) -> RuntimeList { + if limit == 0 { + return RuntimeList::new(Vec::new(), Vec::new()); + } + match self.runtime.list_restoreable_workers() { + Ok(workers) => RuntimeList::new( + workers + .into_iter() + .take(limit) + .map(|worker| self.map_worker_restoreable_candidate(worker)) + .collect(), + Vec::new(), + ), + Err(err) => RuntimeList::new(Vec::new(), vec![embedded_runtime_diagnostic(&err)]), + } + } + fn worker(&self, worker_id: &str) -> WorkerLookupResult { let Some(worker_ref) = self.worker_ref(worker_id) else { return WorkerLookupResult { @@ -1528,6 +1664,32 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime { } } + fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { + let Some(worker_ref) = self.worker_ref(worker_id) else { + return WorkerRestoreResult { + state: WorkerOperationState::Rejected, + worker: None, + diagnostics: vec![diagnostic( + "embedded_worker_id_invalid", + DiagnosticSeverity::Warning, + "Worker id was empty and cannot be restored".to_string(), + )], + }; + }; + match self.runtime.restore_worker(&worker_ref) { + Ok(detail) => WorkerRestoreResult { + state: WorkerOperationState::Accepted, + worker: Some(self.map_worker_detail(detail)), + diagnostics: Vec::new(), + }, + Err(err) => WorkerRestoreResult { + state: WorkerOperationState::Rejected, + worker: None, + diagnostics: vec![embedded_runtime_diagnostic(&err)], + }, + } + } + fn create_working_directory( &self, _request: WorkingDirectoryRequest, @@ -2339,6 +2501,39 @@ impl RemoteWorkerRuntime { } } + fn map_worker_restoreable_candidate( + &self, + candidate: EmbeddedWorkerRestoreableCandidate, + ) -> WorkerRestoreableCandidate { + let worker = self.map_worker_summary(candidate.worker); + let mut diagnostics = worker.diagnostics.clone(); + diagnostics.extend( + candidate + .diagnostics + .iter() + .map(embedded_restoreable_diagnostic), + ); + diagnostics.truncate(MAX_DIAGNOSTICS); + WorkerRestoreableCandidate { + workspace_id: self.workspace_id.clone(), + runtime_id: worker.runtime_id, + worker_id: worker.worker_id, + host_id: worker.host_id, + display_name: worker.display_name, + label: worker.label, + role: worker.role, + profile: worker.profile, + singleton_key: worker.singleton_key, + tags: worker.tags, + workspace: worker.workspace, + state: worker.state, + restoreability: candidate.restoreability, + operations: candidate.operations, + working_directory: worker.working_directory, + diagnostics, + } + } + fn lifecycle_result_from_response( &self, worker_id: &str, @@ -2446,6 +2641,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } } + fn list_restoreable_workers(&self, limit: usize) -> RuntimeList { + if limit == 0 { + return RuntimeList::new(Vec::new(), Vec::new()); + } + match self.get_json::("/v1/workers/restoreable") { + Ok(response) => RuntimeList::new( + response + .workers + .into_iter() + .take(limit) + .map(|worker| self.map_worker_restoreable_candidate(worker)) + .collect(), + Vec::new(), + ), + Err(diagnostic) => RuntimeList::new(Vec::new(), vec![diagnostic]), + } + } + fn worker(&self, worker_id: &str) -> WorkerLookupResult { match self.get_json::(&format!("/v1/workers/{worker_id}")) { Ok(response) => WorkerLookupResult { @@ -2463,6 +2676,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } } + fn restore_worker(&self, worker_id: &str) -> WorkerRestoreResult { + match self.post_json::<_, RuntimeHttpWorkerResponse>( + &format!("/v1/workers/{worker_id}/restore"), + &serde_json::json!({}), + ) { + Ok(response) => WorkerRestoreResult { + state: WorkerOperationState::Accepted, + worker: Some(self.map_worker_detail(response.worker)), + diagnostics: Vec::new(), + }, + Err(diagnostic) => WorkerRestoreResult { + state: WorkerOperationState::Rejected, + worker: None, + diagnostics: vec![diagnostic], + }, + } + } + fn create_working_directory( &self, request: WorkingDirectoryRequest, @@ -2819,6 +3050,20 @@ fn embedded_worker_projection_diagnostics() -> Vec { )] } +fn embedded_restoreable_diagnostic( + restore_diagnostic: &EmbeddedWorkerRestoreableDiagnostic, +) -> RuntimeDiagnostic { + diagnostic( + restore_diagnostic.code.clone(), + match restore_diagnostic.severity { + EmbeddedWorkerRestoreableDiagnosticSeverity::Info => DiagnosticSeverity::Info, + EmbeddedWorkerRestoreableDiagnosticSeverity::Warning => DiagnosticSeverity::Warning, + EmbeddedWorkerRestoreableDiagnosticSeverity::Error => DiagnosticSeverity::Error, + }, + restore_diagnostic.message.clone(), + ) +} + fn default_profile_source_archive_source( profile: &ProfileSelector, ) -> Result { diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index fb85e4ec..33edf320 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -56,9 +56,9 @@ use crate::hosts::{ RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary, WorkerCompletionsRequest, WorkerCompletionsResult, WorkerImplementationSummary, WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest, - WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement, - WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest, - WorkerSummary, WorkerWorkspaceSummary, + WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, WorkerRestoreableCandidate, + WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, + WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary, }; use crate::identity::WorkspaceIdentity; use crate::memory_backend::execute_memory_backend_operation_with_authority; @@ -644,10 +644,18 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/runtimes/{runtime_id}/workers", get(list_runtime_workers).post(create_runtime_worker), ) + .route( + "/api/runtimes/{runtime_id}/workers/restoreable", + get(list_restoreable_runtime_workers), + ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/workers", get(scoped_list_runtime_workers).post(scoped_create_runtime_worker), ) + .route( + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/restoreable", + get(scoped_list_restoreable_runtime_workers), + ) .route( "/api/runtimes/{runtime_id}/config-bundles", post(sync_runtime_config_bundle), @@ -668,10 +676,18 @@ pub fn build_router(api: WorkspaceApi) -> Router { "/api/runtimes/{runtime_id}/workers/{worker_id}", get(get_runtime_worker), ) + .route( + "/api/runtimes/{runtime_id}/workers/{worker_id}/restore", + post(restore_runtime_worker), + ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}", get(scoped_get_runtime_worker), ) + .route( + "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/restore", + post(scoped_restore_runtime_worker), + ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/pin", put(scoped_pin_runtime_worker).delete(scoped_unpin_runtime_worker), @@ -786,6 +802,14 @@ pub struct RuntimeListResponse { pub diagnostics: Vec, } +#[derive(Debug, Serialize, Deserialize)] +pub struct WorkerRestoreResponse { + pub workspace_id: String, + pub runtime_id: String, + pub worker_id: String, + pub result: WorkerRestoreResult, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum CleanupTargetKind { @@ -2738,6 +2762,14 @@ async fn scoped_list_runtime_workers( list_runtime_workers(State(api), AxumPath(path.runtime_id)).await } +async fn scoped_list_restoreable_runtime_workers( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult>> { + validate_workspace_scope(&api, &path.workspace_id)?; + list_restoreable_runtime_workers(State(api), AxumPath(path.runtime_id)).await +} + async fn scoped_create_runtime_worker( State(api): State, AxumPath(path): AxumPath, @@ -2778,6 +2810,14 @@ async fn scoped_get_runtime_worker( get_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await } +async fn scoped_restore_runtime_worker( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + restore_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await +} + async fn scoped_pin_runtime_worker( State(api): State, AxumPath(path): AxumPath, @@ -4322,6 +4362,39 @@ async fn get_runtime_worker( ))) } +async fn restore_runtime_worker( + State(api): State, + AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>, +) -> ApiResult> { + let mut result = api + .runtime + .restore_worker(&runtime_id, &worker_id) + .map_err(|err| err.into_error())?; + if let Some(worker) = result.worker.as_ref() { + let record = sync_worker_observation(&api, worker)?; + let links = api.store.list_worker_workdir_links( + &api.config.workspace_id, + record.runtime_id.as_str(), + record.runtime_worker_id, + )?; + let workdirs = api + .store + .list_workdir_registry(&api.config.workspace_id, 500)?; + result.worker = Some(merge_worker_registry_projection( + Some(worker), + &record, + links, + &workdirs, + )); + } + Ok(Json(WorkerRestoreResponse { + workspace_id: api.workspace_id().to_string(), + runtime_id, + worker_id, + result, + })) +} + #[derive(Debug, Serialize, Deserialize)] pub struct RuntimeConfigBundleSyncRequest { pub bundle: ConfigBundle, @@ -4393,6 +4466,24 @@ async fn list_runtime_workers( })) } +async fn list_restoreable_runtime_workers( + State(api): State, + AxumPath(runtime_id): AxumPath, +) -> ApiResult>> { + let limit = api.config.max_records.min(200); + let worker_list = api + .runtime + .list_restoreable_workers_for_runtime(&runtime_id, limit) + .map_err(|err| err.into_error())?; + Ok(Json(RuntimeListResponse { + workspace_id: api.workspace_id().to_string(), + limit, + items: worker_list.items, + source: "runtime_registry_restoreable".to_string(), + diagnostics: worker_list.diagnostics, + })) +} + async fn create_runtime_worker( State(api): State, AxumPath(runtime_id): AxumPath, @@ -9322,6 +9413,31 @@ mod tests { assert_eq!(worker["worker_id"], worker_id); assert_eq!(worker["runtime_id"], "embedded-worker-runtime"); + let alive_candidates = get_json( + app.clone(), + "/api/runtimes/embedded-worker-runtime/workers/restoreable", + ) + .await; + assert_eq!(alive_candidates["items"][0]["worker_id"], worker_id); + assert_eq!(alive_candidates["items"][0]["restoreability"], "alive"); + assert_eq!( + alive_candidates["items"][0]["operations"]["can_attach"], + true + ); + assert_eq!( + alive_candidates["items"][0]["operations"]["can_restore"], + false + ); + + let restored = post_json( + app.clone(), + &format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/restore"), + json!({}), + ) + .await; + assert_eq!(restored["result"]["state"], "accepted"); + assert_eq!(restored["result"]["worker"]["worker_id"], worker_id); + let accepted = post_json( app.clone(), &format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/input"), @@ -9373,7 +9489,10 @@ mod tests { .unwrap(); assert_eq!(wrong_runtime.status(), StatusCode::NOT_FOUND); - let projected = format!("{}{}{}{}", embedded_summary, spawned, worker, accepted); + let projected = format!( + "{}{}{}{}{}{}", + embedded_summary, spawned, worker, alive_candidates, restored, accepted + ); for forbidden in [ dir.path().to_string_lossy().as_ref(), "metadata.json",