runtime: expose restoreable workers
This commit is contained in:
parent
98c1599d1a
commit
5894e73ab9
|
|
@ -168,6 +168,56 @@ pub struct BackendWorkerSummary {
|
||||||
pub diagnostics: Vec<BackendDiagnostic>,
|
pub diagnostics: Vec<BackendDiagnostic>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub profile: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub singleton_key: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub tags: Vec<String>,
|
||||||
|
pub workspace: BackendWorkerWorkspaceSummary,
|
||||||
|
pub state: String,
|
||||||
|
pub restoreability: String,
|
||||||
|
pub operations: BackendWorkerRestoreableOperations,
|
||||||
|
#[serde(default)]
|
||||||
|
pub working_directory: Option<BackendWorkingDirectorySummary>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub diagnostics: Vec<BackendDiagnostic>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Deserialize, PartialEq, Eq)]
|
||||||
|
pub struct BackendWorkerRestoreResult {
|
||||||
|
pub state: String,
|
||||||
|
#[serde(default)]
|
||||||
|
pub worker: Option<BackendWorkerSummary>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub diagnostics: Vec<BackendDiagnostic>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[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)]
|
#[derive(Debug)]
|
||||||
pub struct BackendRuntimeClient {
|
pub struct BackendRuntimeClient {
|
||||||
target: BackendRuntimeTarget,
|
target: BackendRuntimeTarget,
|
||||||
|
|
@ -266,6 +316,45 @@ pub async fn list_backend_workers(
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub async fn list_backend_restoreable_workers(
|
||||||
|
target: &BackendRuntimeListTarget,
|
||||||
|
) -> Result<BackendRuntimeListResponse<BackendWorkerRestoreableCandidate>, 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::<BackendRuntimeListResponse<BackendWorkerRestoreableCandidate>>()
|
||||||
|
.await?)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn restore_backend_worker(
|
||||||
|
target: &BackendRuntimeTarget,
|
||||||
|
) -> Result<BackendWorkerRestoreResponse, BackendRuntimeClientError> {
|
||||||
|
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::<BackendWorkerRestoreResponse>()
|
||||||
|
.await?)
|
||||||
|
}
|
||||||
|
|
||||||
impl BackendRuntimeClient {
|
impl BackendRuntimeClient {
|
||||||
pub async fn connect(target: BackendRuntimeTarget) -> Result<Self, BackendRuntimeClientError> {
|
pub async fn connect(target: BackendRuntimeTarget) -> Result<Self, BackendRuntimeClientError> {
|
||||||
validate_target(&target)?;
|
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 {
|
fn protocol_ws_url(target: &BackendRuntimeTarget) -> String {
|
||||||
let path = format!(
|
let path = format!(
|
||||||
"/api/runtimes/{}/workers/{}/protocol/ws",
|
"/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"
|
"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"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -23,8 +23,11 @@ pub use backend_auth::{
|
||||||
pub use backend_runtime::{
|
pub use backend_runtime::{
|
||||||
BackendRuntimeClient, BackendRuntimeClientError, BackendRuntimeListResponse,
|
BackendRuntimeClient, BackendRuntimeClientError, BackendRuntimeListResponse,
|
||||||
BackendRuntimeListTarget, BackendRuntimeSummary, BackendRuntimeTarget,
|
BackendRuntimeListTarget, BackendRuntimeSummary, BackendRuntimeTarget,
|
||||||
BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary, BackendWorkerSummary,
|
BackendWorkerCapabilitySummary, BackendWorkerImplementationSummary,
|
||||||
BackendWorkerWorkspaceSummary, BackendWorkingDirectorySummary, list_backend_workers,
|
BackendWorkerRestoreResponse, BackendWorkerRestoreResult, BackendWorkerRestoreableCandidate,
|
||||||
|
BackendWorkerRestoreableOperations, BackendWorkerSummary, BackendWorkerWorkspaceSummary,
|
||||||
|
BackendWorkingDirectorySummary, list_backend_restoreable_workers, list_backend_workers,
|
||||||
|
restore_backend_worker,
|
||||||
};
|
};
|
||||||
pub use runtime_command::WorkerRuntimeCommand;
|
pub use runtime_command::WorkerRuntimeCommand;
|
||||||
pub use target::{
|
pub use target::{
|
||||||
|
|
|
||||||
|
|
@ -257,6 +257,51 @@ pub struct WorkerDetail {
|
||||||
pub last_event_id: u64,
|
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<WorkerRestoreableDiagnostic>,
|
||||||
|
}
|
||||||
|
|
||||||
/// Acknowledgement returned by stop/cancel lifecycle operations.
|
/// Acknowledgement returned by stop/cancel lifecycle operations.
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
pub struct WorkerLifecycleAck {
|
pub struct WorkerLifecycleAck {
|
||||||
|
|
|
||||||
|
|
@ -11,8 +11,8 @@ use crate::auth::{
|
||||||
RuntimeAuthContext, RuntimeHttpAuthConfig, unix_now_seconds, verify_capability_token,
|
RuntimeAuthContext, RuntimeHttpAuthConfig, unix_now_seconds, verify_capability_token,
|
||||||
};
|
};
|
||||||
use crate::catalog::{
|
use crate::catalog::{
|
||||||
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
|
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck,
|
||||||
WorkingDirectoryRequest, WorkingDirectoryStatus,
|
WorkerRestoreableCandidate, WorkerSummary, WorkingDirectoryRequest, WorkingDirectoryStatus,
|
||||||
};
|
};
|
||||||
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
|
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
|
||||||
use crate::error::RuntimeError;
|
use crate::error::RuntimeError;
|
||||||
|
|
@ -174,6 +174,7 @@ pub fn runtime_http_router_with_auth(
|
||||||
get(get_working_directory).delete(cleanup_working_directory),
|
get(get_working_directory).delete(cleanup_working_directory),
|
||||||
)
|
)
|
||||||
.route("/v1/workers", get(list_workers).post(create_worker))
|
.route("/v1/workers", get(list_workers).post(create_worker))
|
||||||
|
.route("/v1/workers/restoreable", get(list_restoreable_workers))
|
||||||
.route(
|
.route(
|
||||||
"/v1/workers/{worker_id}",
|
"/v1/workers/{worker_id}",
|
||||||
get(get_worker).delete(delete_worker),
|
get(get_worker).delete(delete_worker),
|
||||||
|
|
@ -240,6 +241,12 @@ pub struct RuntimeHttpWorkersResponse {
|
||||||
pub workers: Vec<WorkerSummary>,
|
pub workers: Vec<WorkerSummary>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// `GET /v1/workers/restoreable` response.
|
||||||
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
pub struct RuntimeHttpRestoreableWorkersResponse {
|
||||||
|
pub workers: Vec<WorkerRestoreableCandidate>,
|
||||||
|
}
|
||||||
|
|
||||||
/// `GET /v1/working-directories` response.
|
/// `GET /v1/working-directories` response.
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
pub struct RuntimeHttpWorkingDirectoriesResponse {
|
pub struct RuntimeHttpWorkingDirectoriesResponse {
|
||||||
|
|
@ -380,6 +387,16 @@ async fn list_workers(
|
||||||
Ok(Json(RuntimeHttpWorkersResponse { workers }))
|
Ok(Json(RuntimeHttpWorkersResponse { workers }))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn list_restoreable_workers(
|
||||||
|
State(state): State<RuntimeHttpState>,
|
||||||
|
) -> RestResult<RuntimeHttpRestoreableWorkersResponse> {
|
||||||
|
let workers = state
|
||||||
|
.runtime
|
||||||
|
.list_restoreable_workers()
|
||||||
|
.map_err(RuntimeHttpRestError::runtime)?;
|
||||||
|
Ok(Json(RuntimeHttpRestoreableWorkersResponse { workers }))
|
||||||
|
}
|
||||||
|
|
||||||
async fn list_working_directories(
|
async fn list_working_directories(
|
||||||
State(state): State<RuntimeHttpState>,
|
State(state): State<RuntimeHttpState>,
|
||||||
) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
|
) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
|
||||||
|
|
@ -829,6 +846,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
|
||||||
if path == "/v1/workers" && *method == Method::GET {
|
if path == "/v1/workers" && *method == Method::GET {
|
||||||
return Some("workers:list");
|
return Some("workers:list");
|
||||||
}
|
}
|
||||||
|
if path == "/v1/workers/restoreable" && *method == Method::GET {
|
||||||
|
return Some("workers:list");
|
||||||
|
}
|
||||||
if path == "/v1/workers" && *method == Method::POST {
|
if path == "/v1/workers" && *method == Method::POST {
|
||||||
return Some("workers:create");
|
return Some("workers:create");
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,15 +1,14 @@
|
||||||
use crate::catalog::{
|
use crate::catalog::{
|
||||||
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus,
|
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerRestoreability,
|
||||||
WorkerSummary, WorkingDirectoryRequest,
|
WorkerRestoreableCandidate, WorkerRestoreableDiagnostic, WorkerRestoreableDiagnosticSeverity,
|
||||||
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
|
WorkerRestoreableOperations, WorkerStatus, WorkerSummary, WorkingDirectoryRequest,
|
||||||
|
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkingDirectoryStatusKind,
|
||||||
};
|
};
|
||||||
use crate::config_bundle::{
|
use crate::config_bundle::{
|
||||||
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
|
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
|
||||||
validate_config_bundle_ref,
|
validate_config_bundle_ref,
|
||||||
};
|
};
|
||||||
#[cfg(feature = "fs-store")]
|
use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
|
||||||
use crate::diagnostics::DiagnosticSeverity;
|
|
||||||
use crate::diagnostics::RuntimeDiagnostic;
|
|
||||||
use crate::error::RuntimeError;
|
use crate::error::RuntimeError;
|
||||||
use crate::execution::WorkerExecutionRestoreRequest;
|
use crate::execution::WorkerExecutionRestoreRequest;
|
||||||
use crate::execution::{
|
use crate::execution::{
|
||||||
|
|
@ -437,6 +436,26 @@ impl Runtime {
|
||||||
Ok(state.workers.values().map(WorkerRecord::summary).collect())
|
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<Vec<WorkerRestoreableCandidate>, 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.
|
/// Fetch Worker detail. The supplied [`WorkerRef`] must match this Runtime.
|
||||||
pub fn worker_detail(&self, worker_ref: &WorkerRef) -> Result<WorkerDetail, RuntimeError> {
|
pub fn worker_detail(&self, worker_ref: &WorkerRef) -> Result<WorkerDetail, RuntimeError> {
|
||||||
let state = self.lock()?;
|
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::<Vec<_>>();
|
||||||
|
|
||||||
|
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")]
|
#[cfg(feature = "fs-store")]
|
||||||
fn persisted_record(&self) -> PersistedWorkerRecord {
|
fn persisted_record(&self) -> PersistedWorkerRecord {
|
||||||
PersistedWorkerRecord {
|
PersistedWorkerRecord {
|
||||||
|
|
@ -1699,6 +1822,55 @@ impl WorkerRecord {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn worker_restoreable_diagnostic(
|
||||||
|
severity: WorkerRestoreableDiagnosticSeverity,
|
||||||
|
code: impl Into<String>,
|
||||||
|
message: impl Into<String>,
|
||||||
|
) -> 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<WorkerRestoreableDiagnostic>,
|
||||||
|
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 {
|
fn worker_status_from_run_state(run_state: WorkerExecutionRunState) -> WorkerStatus {
|
||||||
match run_state {
|
match run_state {
|
||||||
WorkerExecutionRunState::Idle => WorkerStatus::Idle,
|
WorkerExecutionRunState::Idle => WorkerStatus::Idle,
|
||||||
|
|
@ -2049,6 +2221,47 @@ mod tests {
|
||||||
assert_eq!(fetched.profile, detail.profile);
|
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]
|
#[test]
|
||||||
fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() {
|
fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() {
|
||||||
let runtime = runtime_with_backend();
|
let runtime = runtime_with_backend();
|
||||||
|
|
|
||||||
|
|
@ -16,9 +16,12 @@ use std::{
|
||||||
use worker_runtime::auth::{CapabilityTokenSigner, capability_claims};
|
use worker_runtime::auth::{CapabilityTokenSigner, capability_claims};
|
||||||
use worker_runtime::catalog::{
|
use worker_runtime::catalog::{
|
||||||
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef,
|
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef,
|
||||||
ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail,
|
ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail, WorkerRestoreability,
|
||||||
WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRequest,
|
WorkerRestoreableCandidate as EmbeddedWorkerRestoreableCandidate,
|
||||||
WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef,
|
WorkerRestoreableDiagnostic as EmbeddedWorkerRestoreableDiagnostic,
|
||||||
|
WorkerRestoreableDiagnosticSeverity as EmbeddedWorkerRestoreableDiagnosticSeverity,
|
||||||
|
WorkerRestoreableOperations, WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim,
|
||||||
|
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef,
|
||||||
};
|
};
|
||||||
use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
|
use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|
@ -31,11 +34,12 @@ use worker_runtime::execution::WorkerExecutionRunState;
|
||||||
use worker_runtime::fs_store::FsRuntimeStoreOptions;
|
use worker_runtime::fs_store::FsRuntimeStoreOptions;
|
||||||
use worker_runtime::http_server::{
|
use worker_runtime::http_server::{
|
||||||
RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
|
RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
|
||||||
RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerCompletionsRequest,
|
RuntimeHttpErrorResponse, RuntimeHttpRestoreableWorkersResponse, RuntimeHttpSummaryResponse,
|
||||||
RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse,
|
RuntimeHttpWorkerCompletionsRequest, RuntimeHttpWorkerCompletionsResponse,
|
||||||
RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest,
|
RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse,
|
||||||
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse,
|
RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse,
|
||||||
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
|
RuntimeHttpWorkerResponse, RuntimeHttpWorkersResponse, RuntimeHttpWorkingDirectoriesResponse,
|
||||||
|
RuntimeHttpWorkingDirectoryResponse,
|
||||||
};
|
};
|
||||||
use worker_runtime::identity::{WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef};
|
use worker_runtime::identity::{WorkerId as EmbeddedWorkerId, WorkerRef as EmbeddedWorkerRef};
|
||||||
use worker_runtime::interaction::{
|
use worker_runtime::interaction::{
|
||||||
|
|
@ -253,6 +257,36 @@ pub struct WorkerSummary {
|
||||||
pub diagnostics: Vec<RuntimeDiagnostic>,
|
pub diagnostics: Vec<RuntimeDiagnostic>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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<String>,
|
||||||
|
pub profile: Option<String>,
|
||||||
|
pub singleton_key: Option<String>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub tags: Vec<String>,
|
||||||
|
pub workspace: WorkerWorkspaceSummary,
|
||||||
|
pub state: String,
|
||||||
|
pub restoreability: WorkerRestoreability,
|
||||||
|
pub operations: WorkerRestoreableOperations,
|
||||||
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
|
pub working_directory: Option<WorkingDirectorySummary>,
|
||||||
|
pub diagnostics: Vec<RuntimeDiagnostic>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
|
pub struct WorkerRestoreResult {
|
||||||
|
pub state: WorkerOperationState,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
pub worker: Option<WorkerSummary>,
|
||||||
|
pub diagnostics: Vec<RuntimeDiagnostic>,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
pub struct RuntimeList<T> {
|
pub struct RuntimeList<T> {
|
||||||
pub items: Vec<T>,
|
pub items: Vec<T>,
|
||||||
|
|
@ -270,6 +304,11 @@ fn is_retired_companion_worker(worker: &WorkerSummary) -> bool {
|
||||||
|| worker.profile.as_deref() == Some("builtin:companion")
|
|| 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)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
pub struct WorkerLookupResult {
|
pub struct WorkerLookupResult {
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
|
@ -576,8 +615,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
|
||||||
|
|
||||||
fn list_workers(&self, limit: usize) -> RuntimeList<WorkerSummary>;
|
fn list_workers(&self, limit: usize) -> RuntimeList<WorkerSummary>;
|
||||||
|
|
||||||
|
fn list_restoreable_workers(&self, _limit: usize) -> RuntimeList<WorkerRestoreableCandidate> {
|
||||||
|
RuntimeList::new(Vec::new(), Vec::new())
|
||||||
|
}
|
||||||
|
|
||||||
fn worker(&self, worker_id: &str) -> WorkerLookupResult;
|
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(
|
fn create_working_directory(
|
||||||
&self,
|
&self,
|
||||||
_request: WorkingDirectoryRequest,
|
_request: WorkingDirectoryRequest,
|
||||||
|
|
@ -955,6 +1010,26 @@ impl RuntimeRegistry {
|
||||||
Ok(RuntimeList::new(items, diagnostics))
|
Ok(RuntimeList::new(items, diagnostics))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn list_restoreable_workers_for_runtime(
|
||||||
|
&self,
|
||||||
|
runtime_id: &str,
|
||||||
|
limit: usize,
|
||||||
|
) -> Result<RuntimeList<WorkerRestoreableCandidate>, 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(
|
pub fn list_workers_for_host(
|
||||||
&self,
|
&self,
|
||||||
host_id: &str,
|
host_id: &str,
|
||||||
|
|
@ -1015,6 +1090,17 @@ impl RuntimeRegistry {
|
||||||
Ok(worker)
|
Ok(worker)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub fn restore_worker(
|
||||||
|
&self,
|
||||||
|
runtime_id: &str,
|
||||||
|
worker_id: &str,
|
||||||
|
) -> Result<WorkerRestoreResult, RuntimeRegistryError> {
|
||||||
|
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(
|
pub fn spawn_worker(
|
||||||
&self,
|
&self,
|
||||||
runtime_id: &str,
|
runtime_id: &str,
|
||||||
|
|
@ -1414,6 +1500,39 @@ impl EmbeddedWorkerRuntime {
|
||||||
diagnostics: embedded_worker_projection_diagnostics(),
|
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 {
|
impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
|
||||||
|
|
@ -1501,6 +1620,23 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn list_restoreable_workers(&self, limit: usize) -> RuntimeList<WorkerRestoreableCandidate> {
|
||||||
|
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 {
|
fn worker(&self, worker_id: &str) -> WorkerLookupResult {
|
||||||
let Some(worker_ref) = self.worker_ref(worker_id) else {
|
let Some(worker_ref) = self.worker_ref(worker_id) else {
|
||||||
return WorkerLookupResult {
|
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(
|
fn create_working_directory(
|
||||||
&self,
|
&self,
|
||||||
_request: WorkingDirectoryRequest,
|
_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(
|
fn lifecycle_result_from_response(
|
||||||
&self,
|
&self,
|
||||||
worker_id: &str,
|
worker_id: &str,
|
||||||
|
|
@ -2446,6 +2641,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn list_restoreable_workers(&self, limit: usize) -> RuntimeList<WorkerRestoreableCandidate> {
|
||||||
|
if limit == 0 {
|
||||||
|
return RuntimeList::new(Vec::new(), Vec::new());
|
||||||
|
}
|
||||||
|
match self.get_json::<RuntimeHttpRestoreableWorkersResponse>("/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 {
|
fn worker(&self, worker_id: &str) -> WorkerLookupResult {
|
||||||
match self.get_json::<RuntimeHttpWorkerResponse>(&format!("/v1/workers/{worker_id}")) {
|
match self.get_json::<RuntimeHttpWorkerResponse>(&format!("/v1/workers/{worker_id}")) {
|
||||||
Ok(response) => WorkerLookupResult {
|
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(
|
fn create_working_directory(
|
||||||
&self,
|
&self,
|
||||||
request: WorkingDirectoryRequest,
|
request: WorkingDirectoryRequest,
|
||||||
|
|
@ -2819,6 +3050,20 @@ fn embedded_worker_projection_diagnostics() -> Vec<RuntimeDiagnostic> {
|
||||||
)]
|
)]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
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(
|
fn default_profile_source_archive_source(
|
||||||
profile: &ProfileSelector,
|
profile: &ProfileSelector,
|
||||||
) -> Result<ProfileSourceArchiveSource, String> {
|
) -> Result<ProfileSourceArchiveSource, String> {
|
||||||
|
|
|
||||||
|
|
@ -56,9 +56,9 @@ use crate::hosts::{
|
||||||
RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary,
|
RuntimeRegistryError, RuntimeRegistryUnregisterResult, RuntimeSummary, WorkerCapabilitySummary,
|
||||||
WorkerCompletionsRequest, WorkerCompletionsResult, WorkerImplementationSummary,
|
WorkerCompletionsRequest, WorkerCompletionsResult, WorkerImplementationSummary,
|
||||||
WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest,
|
WorkerInputKind, WorkerInputRequest, WorkerInputResult, WorkerLifecycleRequest,
|
||||||
WorkerLifecycleResult, WorkerOperationState, WorkerSpawnAcceptanceRequirement,
|
WorkerLifecycleResult, WorkerOperationState, WorkerRestoreResult, WorkerRestoreableCandidate,
|
||||||
WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult, WorkerSpawnWorkingDirectoryRequest,
|
WorkerSpawnAcceptanceRequirement, WorkerSpawnIntent, WorkerSpawnRequest, WorkerSpawnResult,
|
||||||
WorkerSummary, WorkerWorkspaceSummary,
|
WorkerSpawnWorkingDirectoryRequest, WorkerSummary, WorkerWorkspaceSummary,
|
||||||
};
|
};
|
||||||
use crate::identity::WorkspaceIdentity;
|
use crate::identity::WorkspaceIdentity;
|
||||||
use crate::memory_backend::execute_memory_backend_operation_with_authority;
|
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",
|
"/api/runtimes/{runtime_id}/workers",
|
||||||
get(list_runtime_workers).post(create_runtime_worker),
|
get(list_runtime_workers).post(create_runtime_worker),
|
||||||
)
|
)
|
||||||
|
.route(
|
||||||
|
"/api/runtimes/{runtime_id}/workers/restoreable",
|
||||||
|
get(list_restoreable_runtime_workers),
|
||||||
|
)
|
||||||
.route(
|
.route(
|
||||||
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers",
|
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers",
|
||||||
get(scoped_list_runtime_workers).post(scoped_create_runtime_worker),
|
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(
|
.route(
|
||||||
"/api/runtimes/{runtime_id}/config-bundles",
|
"/api/runtimes/{runtime_id}/config-bundles",
|
||||||
post(sync_runtime_config_bundle),
|
post(sync_runtime_config_bundle),
|
||||||
|
|
@ -668,10 +676,18 @@ pub fn build_router(api: WorkspaceApi) -> Router {
|
||||||
"/api/runtimes/{runtime_id}/workers/{worker_id}",
|
"/api/runtimes/{runtime_id}/workers/{worker_id}",
|
||||||
get(get_runtime_worker),
|
get(get_runtime_worker),
|
||||||
)
|
)
|
||||||
|
.route(
|
||||||
|
"/api/runtimes/{runtime_id}/workers/{worker_id}/restore",
|
||||||
|
post(restore_runtime_worker),
|
||||||
|
)
|
||||||
.route(
|
.route(
|
||||||
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}",
|
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}",
|
||||||
get(scoped_get_runtime_worker),
|
get(scoped_get_runtime_worker),
|
||||||
)
|
)
|
||||||
|
.route(
|
||||||
|
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/restore",
|
||||||
|
post(scoped_restore_runtime_worker),
|
||||||
|
)
|
||||||
.route(
|
.route(
|
||||||
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/pin",
|
"/api/w/{workspace_id}/runtimes/{runtime_id}/workers/{worker_id}/pin",
|
||||||
put(scoped_pin_runtime_worker).delete(scoped_unpin_runtime_worker),
|
put(scoped_pin_runtime_worker).delete(scoped_unpin_runtime_worker),
|
||||||
|
|
@ -786,6 +802,14 @@ pub struct RuntimeListResponse<T> {
|
||||||
pub diagnostics: Vec<RuntimeDiagnostic>,
|
pub diagnostics: Vec<RuntimeDiagnostic>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
#[serde(rename_all = "snake_case")]
|
#[serde(rename_all = "snake_case")]
|
||||||
pub enum CleanupTargetKind {
|
pub enum CleanupTargetKind {
|
||||||
|
|
@ -2738,6 +2762,14 @@ async fn scoped_list_runtime_workers(
|
||||||
list_runtime_workers(State(api), AxumPath(path.runtime_id)).await
|
list_runtime_workers(State(api), AxumPath(path.runtime_id)).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn scoped_list_restoreable_runtime_workers(
|
||||||
|
State(api): State<WorkspaceApi>,
|
||||||
|
AxumPath(path): AxumPath<ScopedRuntimePath>,
|
||||||
|
) -> ApiResult<Json<RuntimeListResponse<WorkerRestoreableCandidate>>> {
|
||||||
|
validate_workspace_scope(&api, &path.workspace_id)?;
|
||||||
|
list_restoreable_runtime_workers(State(api), AxumPath(path.runtime_id)).await
|
||||||
|
}
|
||||||
|
|
||||||
async fn scoped_create_runtime_worker(
|
async fn scoped_create_runtime_worker(
|
||||||
State(api): State<WorkspaceApi>,
|
State(api): State<WorkspaceApi>,
|
||||||
AxumPath(path): AxumPath<ScopedRuntimePath>,
|
AxumPath(path): AxumPath<ScopedRuntimePath>,
|
||||||
|
|
@ -2778,6 +2810,14 @@ async fn scoped_get_runtime_worker(
|
||||||
get_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await
|
get_runtime_worker(State(api), AxumPath((path.runtime_id, path.worker_id))).await
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn scoped_restore_runtime_worker(
|
||||||
|
State(api): State<WorkspaceApi>,
|
||||||
|
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
|
||||||
|
) -> ApiResult<Json<WorkerRestoreResponse>> {
|
||||||
|
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(
|
async fn scoped_pin_runtime_worker(
|
||||||
State(api): State<WorkspaceApi>,
|
State(api): State<WorkspaceApi>,
|
||||||
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
|
AxumPath(path): AxumPath<ScopedRuntimeWorkerPath>,
|
||||||
|
|
@ -4322,6 +4362,39 @@ async fn get_runtime_worker(
|
||||||
)))
|
)))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn restore_runtime_worker(
|
||||||
|
State(api): State<WorkspaceApi>,
|
||||||
|
AxumPath((runtime_id, worker_id)): AxumPath<(String, String)>,
|
||||||
|
) -> ApiResult<Json<WorkerRestoreResponse>> {
|
||||||
|
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)]
|
#[derive(Debug, Serialize, Deserialize)]
|
||||||
pub struct RuntimeConfigBundleSyncRequest {
|
pub struct RuntimeConfigBundleSyncRequest {
|
||||||
pub bundle: ConfigBundle,
|
pub bundle: ConfigBundle,
|
||||||
|
|
@ -4393,6 +4466,24 @@ async fn list_runtime_workers(
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn list_restoreable_runtime_workers(
|
||||||
|
State(api): State<WorkspaceApi>,
|
||||||
|
AxumPath(runtime_id): AxumPath<String>,
|
||||||
|
) -> ApiResult<Json<RuntimeListResponse<WorkerRestoreableCandidate>>> {
|
||||||
|
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(
|
async fn create_runtime_worker(
|
||||||
State(api): State<WorkspaceApi>,
|
State(api): State<WorkspaceApi>,
|
||||||
AxumPath(runtime_id): AxumPath<String>,
|
AxumPath(runtime_id): AxumPath<String>,
|
||||||
|
|
@ -9322,6 +9413,31 @@ mod tests {
|
||||||
assert_eq!(worker["worker_id"], worker_id);
|
assert_eq!(worker["worker_id"], worker_id);
|
||||||
assert_eq!(worker["runtime_id"], "embedded-worker-runtime");
|
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(
|
let accepted = post_json(
|
||||||
app.clone(),
|
app.clone(),
|
||||||
&format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/input"),
|
&format!("/api/runtimes/embedded-worker-runtime/workers/{worker_id}/input"),
|
||||||
|
|
@ -9373,7 +9489,10 @@ mod tests {
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(wrong_runtime.status(), StatusCode::NOT_FOUND);
|
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 [
|
for forbidden in [
|
||||||
dir.path().to_string_lossy().as_ref(),
|
dir.path().to_string_lossy().as_ref(),
|
||||||
"metadata.json",
|
"metadata.json",
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue
Block a user