Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

Address runtime API review findings

Browse Source
This commit is contained in:
Hare
2026-09-18 02:11:20 +09:00
parent d837684a33
commit fbdb6ccc22
4 changed files with 402 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/runtime-api/src/lib.rs
+55 -19
View File
@@ -638,28 +638,13 @@ pub struct RemainingRuntimeRoute {
pub const REMAINING_RUNTIME_ROUTES: &[RemainingRuntimeRoute] = &[
RemainingRuntimeRoute {
method: "POST",
path: "/v1/bootstrap",
reason: "bootstrap lifecycle",
},
RemainingRuntimeRoute {
method: "GET",
path: "/v1/protocol/ws",
reason: "WebSocket protocol transport",
},
RemainingRuntimeRoute {
method: "GET",
path: "/v1/workers/{worker_id}/protocol/ws",
reason: "WebSocket protocol transport",
path: "/v1/workspace-runtime-verification/challenge",
reason: "Workspace verification handshake",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/workers/{worker_id}/attachments",
reason: "binary attachment transport",
},
RemainingRuntimeRoute {
method: "DELETE",
path: "/v1/workers/{worker_id}/attachments/{artifact_id}",
reason: "binary attachment transport",
path: "/v1/workspace-runtime-verification/acknowledgement",
reason: "Workspace verification handshake",
},
RemainingRuntimeRoute {
method: "GET",
@@ -691,6 +676,56 @@ pub const REMAINING_RUNTIME_ROUTES: &[RemainingRuntimeRoute] = &[
path: "/v1/working-directories",
reason: "Workdir transport",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/working-directories/repository-access",
reason: "Workdir transport",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/repositories/ssh/probe",
reason: "Repository SSH transport",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/repository-refs/observe",
reason: "Repository observation transport",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/working-directories/{working_directory_id}/sessions",
reason: "Workdir session transport",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/workdir-sessions/{session_id}/operations",
reason: "Workdir session transport",
},
RemainingRuntimeRoute {
method: "DELETE",
path: "/v1/workdir-sessions/{session_id}",
reason: "Workdir session transport",
},
RemainingRuntimeRoute {
method: "GET",
path: "/v1/working-directories/{working_directory_id}",
reason: "Workdir transport",
},
RemainingRuntimeRoute {
method: "DELETE",
path: "/v1/working-directories/{working_directory_id}",
reason: "Workdir transport",
},
RemainingRuntimeRoute {
method: "POST",
path: "/v1/workers/{worker_id}/attachments",
reason: "Binary attachment transport",
},
RemainingRuntimeRoute {
method: "DELETE",
path: "/v1/workers/{worker_id}/attachments/{artifact_id}",
reason: "Binary attachment transport",
},
];
#[cfg(test)]
@@ -866,6 +901,7 @@ mod tests {
#[test]
fn remaining_route_inventory_does_not_overlap_contract() {
assert_eq!(REMAINING_RUNTIME_ROUTES.len(), 18);
for remaining in REMAINING_RUNTIME_ROUTES {
assert!(!RuntimeApiMetadata::OPERATIONS.iter().any(|operation| {
format!("{:?}", operation.method).eq_ignore_ascii_case(remaining.method)
crates/worker-runtime/src/http_server.rs
+32
View File
@@ -348,11 +348,13 @@ struct RuntimeHttpConfigBundleAvailabilityQuery {
digest: String,
}
#[allow(dead_code)]
#[derive(Clone, Debug, Default, Deserialize)]
struct RuntimeHttpWorkersQuery {
status: Option<RuntimeHttpWorkerStatusFilter>,
}
#[allow(dead_code)]
#[derive(Clone, Copy, Debug, Deserialize)]
#[serde(rename_all = "snake_case")]
enum RuntimeHttpWorkerStatusFilter {
@@ -646,6 +648,7 @@ async fn post_workspace_verification_acknowledgement(
}))
}
#[allow(dead_code)]
async fn get_runtime_ping(
State(state): State<RuntimeHttpState>,
Extension(auth): Extension<RuntimeAuthContext>,
@@ -688,6 +691,7 @@ async fn get_runtime_ping(
}))
}
#[allow(dead_code)]
async fn get_runtime(
State(state): State<RuntimeHttpState>,
) -> RestResult<RuntimeHttpSummaryResponse> {
@@ -767,6 +771,7 @@ async fn check_config_bundle(
}))
}
#[allow(dead_code)]
async fn list_workers(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1107,6 +1112,7 @@ async fn cleanup_working_directory(
}))
}
#[allow(dead_code)]
async fn get_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1121,6 +1127,7 @@ async fn get_worker(
Ok(Json(RuntimeHttpWorkerResponse { worker }))
}
#[allow(dead_code)]
async fn delete_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1135,6 +1142,7 @@ async fn delete_worker(
Ok(Json(RuntimeHttpWorkerDeleteResponse { worker }))
}
#[allow(dead_code)]
async fn create_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1149,6 +1157,7 @@ async fn create_worker(
Ok(Json(RuntimeHttpWorkerResponse { worker }))
}
#[allow(dead_code)]
async fn replace_worker_workspace_api(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1171,6 +1180,7 @@ async fn replace_worker_workspace_api(
Ok(Json(RuntimeHttpWorkerResponse { worker }))
}
#[allow(dead_code)]
async fn restore_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1734,6 +1744,7 @@ fn protocol_error_event(message: impl Into<String>) -> protocol::Event {
}
}
#[allow(dead_code)]
async fn worker_retention_inventory(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1754,6 +1765,7 @@ async fn worker_retention_inventory(
.map_err(RuntimeHttpRestError::runtime)
}
#[allow(dead_code)]
async fn execute_worker_retention(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1789,6 +1801,7 @@ async fn execute_worker_retention(
.map_err(RuntimeHttpRestError::runtime)
}
#[allow(dead_code)]
async fn send_worker_input(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1805,6 +1818,7 @@ async fn send_worker_input(
Ok(Json(RuntimeHttpWorkerInputResponse { ack }))
}
#[allow(dead_code)]
async fn worker_completions(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1941,6 +1955,7 @@ async fn delete_worker_uploaded_file(
}))
}
#[allow(dead_code)]
async fn stop_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -1959,6 +1974,7 @@ async fn stop_worker(
Ok(Json(RuntimeHttpWorkerLifecycleResponse { ack }))
}
#[allow(dead_code)]
async fn cancel_worker(
State(state): State<RuntimeHttpState>,
auth: Option<Extension<RuntimeAuthContext>>,
@@ -3252,6 +3268,22 @@ mod tests {
let token = "local-token";
let app = runtime_http_router(runtime.clone(), token.to_string());
let malformed_response = app
.clone()
.oneshot(
Request::builder()
.method(Method::POST)
.uri("/v1/workers")
.header(header::AUTHORIZATION, format!("Bearer {token}"))
.body(Body::from("{"))
.unwrap(),
)
.await
.unwrap();
assert_eq!(malformed_response.status(), StatusCode::BAD_REQUEST);
let malformed: RuntimeHttpErrorResponse = read_json(malformed_response).await;
assert_eq!(malformed.error.code, "invalid_json");
let response = authed_json_request(
app.clone(),
Method::POST,
crates/worker-runtime/src/http_server/runtime_management_api.rs
+253 -121
View File
@@ -50,7 +50,17 @@ async fn runtime_api_wire_compatibility(
mut request: axum::extract::Request,
next: Next,
) -> Response {
if !matches!(*request.method(), Method::GET | Method::HEAD) {
let is_json_request = !matches!(*request.method(), Method::GET | Method::HEAD);
let is_query_request = request.method() == Method::GET;
if request.uri().path() == "/v1/ping" && !request.headers().contains_key("x-yoi-workspace-id") {
return RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"runtime_ping_workspace_scope_required",
"Runtime ping requires an authenticated Workspace scope",
)
.into_response();
}
if is_json_request {
if !request.headers().contains_key(header::CONTENT_TYPE) {
request.headers_mut().insert(
header::CONTENT_TYPE,
@@ -73,11 +83,28 @@ async fn runtime_api_wire_compatibility(
request = axum::extract::Request::from_parts(parts, Body::from(body));
}
let response = next.run(request).await;
if response.status() == StatusCode::UNPROCESSABLE_ENTITY {
let is_json_error = response
.headers()
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| value.starts_with("application/json"));
if matches!(
response.status(),
StatusCode::BAD_REQUEST | StatusCode::UNPROCESSABLE_ENTITY
) && !is_json_error
{
return RuntimeHttpRestError::new(
StatusCode::BAD_REQUEST,
"invalid_json",
"request body must be valid JSON",
if is_query_request {
"invalid_query"
} else {
"invalid_json"
},
if is_query_request {
"request query parameters are invalid"
} else {
"request body must be valid JSON"
},
)
.into_response();
}
@@ -125,79 +152,117 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
&self,
workspace_id: String,
) -> Result<runtime_api::RuntimePingResponse, runtime_api::RuntimeApiError> {
let mut headers = HeaderMap::new();
headers.insert(
"x-yoi-workspace-id",
workspace_id.parse().map_err(|error| {
let Extension(auth) = required_auth()?;
if workspace_id != auth.workspace_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"runtime_ping_workspace_scope_mismatch",
"Runtime ping Workspace scope does not match the authenticated capability",
));
}
let runtime_id = self
.state
.workspace_auth
.as_ref()
.map(|auth| auth.signer.runtime_id().trim())
.filter(|runtime_id| !runtime_id.is_empty())
.ok_or_else(|| {
runtime_api::RuntimeApiError::new(
StatusCode::BAD_REQUEST.as_u16(),
"invalid_workspace_header",
format!("invalid Workspace ping header: {error}"),
StatusCode::SERVICE_UNAVAILABLE.as_u16(),
"runtime_ping_identity_unavailable",
"Runtime ping identity is not configured",
)
})?,
);
let Json(value) = get_runtime_ping(State(self.state.clone()), required_auth()?, headers)
.await
.map_err(api_error)?;
response(value)
})?;
Ok(runtime_api::RuntimePingResponse {
runtime_id: runtime_id.to_string(),
protocol_version: RUNTIME_HTTP_PROTOCOL_VERSION,
})
}
async fn runtime_summary(
&self,
) -> Result<runtime_api::RuntimeSummaryResponse, runtime_api::RuntimeApiError> {
let Json(value) = get_runtime(State(self.state.clone()))
.await
let runtime = self
.state
.runtime
.summary()
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let runtime = response(runtime)?;
Ok(runtime_api::RuntimeSummaryResponse { runtime })
}
async fn list_workers(
&self,
query: runtime_api::WorkerListQuery,
) -> Result<runtime_api::WorkersResponse, runtime_api::RuntimeApiError> {
let Json(value) = list_workers(
State(self.state.clone()),
auth_extension(),
Ok(Query(request(query)?)),
)
.await
let scope =
auth_workspace_scope(&self.state, auth_extension().as_ref()).map_err(api_error)?;
let workers = match (query.status, scope.as_ref()) {
(Some(runtime_api::WorkerStatusFilter::Stopped), Some(scope)) => {
self.state.runtime.list_stopped_workers_scoped(scope)
}
(Some(runtime_api::WorkerStatusFilter::Stopped), None) => {
self.state.runtime.list_stopped_workers()
}
(None, Some(scope)) => self.state.runtime.list_workers_scoped(scope),
(None, None) => self.state.runtime.list_workers(),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let workers = response(workers)?;
Ok(runtime_api::WorkersResponse { workers })
}
async fn get_worker(
&self,
worker_id: String,
) -> Result<runtime_api::WorkerResponse, runtime_api::RuntimeApiError> {
let Json(value) = get_worker(State(self.state.clone()), auth_extension(), Path(worker_id))
.await
.map_err(api_error)?;
response(value)
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self.state.runtime.worker_detail_scoped(&scope, &worker_ref),
None => self.state.runtime.worker_detail(&worker_ref),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
let worker = response(worker)?;
Ok(runtime_api::WorkerResponse { worker })
}
async fn create_worker(
&self,
value: runtime_api::CreateWorkerRequest,
) -> Result<runtime_api::WorkerResponse, runtime_api::RuntimeApiError> {
let Json(value) = create_worker(
State(self.state.clone()),
auth_extension(),
Ok(Json(request(value)?)),
)
.await
let request: CreateWorkerRequest = request(value)?;
let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self.state.runtime.create_worker_scoped(&scope, request),
None => self.state.runtime.create_worker(request),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let worker = response(worker)?;
Ok(runtime_api::WorkerResponse { worker })
}
async fn delete_worker(
&self,
worker_id: String,
) -> Result<runtime_api::WorkerDeleteResponse, runtime_api::RuntimeApiError> {
let Json(value) =
delete_worker(State(self.state.clone()), auth_extension(), Path(worker_id))
.await
.map_err(api_error)?;
response(value)
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self.state.runtime.delete_worker_scoped(&scope, &worker_ref),
None => self.state.runtime.delete_worker(&worker_ref),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
let worker = response(worker)?;
Ok(runtime_api::WorkerDeleteResponse { worker })
}
async fn send_worker_input(
@@ -205,15 +270,21 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
input: runtime_api::WorkerInput,
) -> Result<runtime_api::WorkerInputResponse, runtime_api::RuntimeApiError> {
let Json(value) = send_worker_input(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
Ok(Json(request(input)?)),
)
.await
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let input: WorkerInput = request(input)?;
let ack = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self
.state
.runtime
.send_input_scoped(&scope, &worker_ref, input),
None => self.state.runtime.send_input(&worker_ref, input),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let ack = response(ack)?;
Ok(runtime_api::WorkerInputResponse { ack })
}
async fn stop_worker(
@@ -221,23 +292,20 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
value: runtime_api::WorkerLifecycleRequest,
) -> Result<runtime_api::WorkerLifecycleResponse, runtime_api::RuntimeApiError> {
let body = serde_json::to_vec(&request::<_, RuntimeHttpWorkerLifecycleRequest>(value)?)
.map_err(|error| {
runtime_api::RuntimeApiError::new(
500,
"runtime_request_conversion_failed",
error.to_string(),
)
})?;
let Json(value) = stop_worker(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
Bytes::from(body),
)
.await
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let ack = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self
.state
.runtime
.stop_worker_scoped(&scope, &worker_ref, value.reason),
None => self.state.runtime.stop_worker(&worker_ref, value.reason),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let ack = response(ack)?;
Ok(runtime_api::WorkerLifecycleResponse { ack })
}
async fn cancel_worker(
@@ -245,23 +313,21 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
value: runtime_api::WorkerLifecycleRequest,
) -> Result<runtime_api::WorkerLifecycleResponse, runtime_api::RuntimeApiError> {
let body = serde_json::to_vec(&request::<_, RuntimeHttpWorkerLifecycleRequest>(value)?)
.map_err(|error| {
runtime_api::RuntimeApiError::new(
500,
"runtime_request_conversion_failed",
error.to_string(),
)
})?;
let Json(value) = cancel_worker(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
Bytes::from(body),
)
.await
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let ack = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => {
self.state
.runtime
.cancel_worker_scoped(&scope, &worker_ref, value.reason)
}
None => self.state.runtime.cancel_worker(&worker_ref, value.reason),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let ack = response(ack)?;
Ok(runtime_api::WorkerLifecycleResponse { ack })
}
async fn restore_worker(
@@ -269,11 +335,20 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
_request: runtime_api::EmptyObjectRequest,
) -> Result<runtime_api::WorkerResponse, runtime_api::RuntimeApiError> {
let Json(value) =
restore_worker(State(self.state.clone()), auth_extension(), Path(worker_id))
.await
.map_err(api_error)?;
response(value)
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self
.state
.runtime
.restore_worker_scoped(&scope, &worker_ref),
None => self.state.runtime.restore_worker(&worker_ref),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
let worker = response(worker)?;
Ok(runtime_api::WorkerResponse { worker })
}
async fn replace_worker_workspace_api(
@@ -281,15 +356,25 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
value: runtime_api::WorkerWorkspaceApiRequest,
) -> Result<runtime_api::WorkerResponse, runtime_api::RuntimeApiError> {
let Json(value) = replace_worker_workspace_api(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
Ok(Json(request(value)?)),
)
.await
let workspace_api: WorkspaceApiRef = request(value.workspace_api)?;
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let worker = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => self.state.runtime.replace_worker_workspace_api_scoped(
&scope,
&worker_ref,
workspace_api,
),
None => self
.state
.runtime
.replace_worker_workspace_api(&worker_ref, workspace_api),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let worker = response(worker)?;
Ok(runtime_api::WorkerResponse { worker })
}
async fn complete_worker_arguments(
@@ -297,29 +382,53 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
value: runtime_api::CompletionRequest,
) -> Result<runtime_api::CompletionResponse, runtime_api::RuntimeApiError> {
let Json(value) = worker_completions(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
Ok(Json(request(value)?)),
)
.await
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let kind = value.kind;
let prefix = value.prefix;
let entries = match auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
{
Some(scope) => {
self.state
.runtime
.worker_completions_scoped(&scope, &worker_ref, kind, &prefix)
}
None => self
.state
.runtime
.worker_completions(&worker_ref, kind, &prefix),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(value)
let entries = response(entries)?;
Ok(runtime_api::CompletionResponse {
kind,
prefix,
entries,
})
}
async fn retention_inventory(
&self,
worker_id: String,
) -> Result<runtime_api::WorkerRetentionInventory, runtime_api::RuntimeApiError> {
let Json(value) = worker_retention_inventory(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
)
.await
.map_err(api_error)?;
response(value)
let scope = auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
.ok_or_else(|| {
runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"workspace_scope_required",
"Worker retention inventory requires workspace-scoped authorization",
)
})?;
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let inventory = self
.state
.runtime
.worker_retention_inventory(&scope.workspace_id, &worker_ref)
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(inventory)
}
async fn execute_retention(
@@ -327,14 +436,37 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
worker_id: String,
value: runtime_api::WorkerRetentionExecutionRequest,
) -> Result<runtime_api::WorkerRetentionExecutionResult, runtime_api::RuntimeApiError> {
let Json(value) = execute_worker_retention(
State(self.state.clone()),
auth_extension(),
Path(worker_id),
Ok(Json(request(value)?)),
)
.await
.map_err(api_error)?;
response(value)
let scope = auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
.ok_or_else(|| {
runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"workspace_scope_required",
"Worker retention execution requires workspace-scoped authorization",
)
})?;
let worker_ref = worker_ref_for(&self.state.runtime, worker_id).map_err(api_error)?;
let request: WorkerRetentionExecutionRequest = request(value)?;
if request.workspace_id != scope.workspace_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"workspace_scope_mismatch",
"Worker retention request workspace does not match the authenticated workspace scope",
));
}
if request.worker_id != worker_ref.worker_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::BAD_REQUEST.as_u16(),
"worker_id_mismatch",
"Worker retention request worker_id does not match route worker_id",
));
}
let result = self
.state
.runtime
.execute_worker_retention(&request)
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
response(result)
}
}
crates/workspace-server/src/hosts.rs
+62 -54
View File
@@ -46,10 +46,7 @@ use worker_runtime::http_server::{
RUNTIME_PING_PERMISSION, RUNTIME_WORKSPACE_SCOPE_HEADER,
RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
RuntimeHttpErrorResponse, RuntimeHttpPingResponse, RuntimeHttpRepositoryAccessResponse,
RuntimeHttpSummaryResponse, RuntimeHttpUploadedFileDeleteResponse,
RuntimeHttpUploadedFileResponse, RuntimeHttpWorkerCompletionsResponse,
RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse,
RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
RuntimeHttpUploadedFileDeleteResponse, RuntimeHttpUploadedFileResponse,
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
};
@@ -3981,7 +3978,7 @@ impl RemoteWorkerRuntime {
fn lifecycle_result_from_response(
&self,
worker_id: &str,
response: RuntimeHttpWorkerLifecycleResponse,
response: worker_runtime::catalog::WorkerLifecycleAck,
) -> WorkerLifecycleResult {
WorkerLifecycleResult {
state: WorkerOperationState::Accepted,
@@ -3991,7 +3988,7 @@ impl RemoteWorkerRuntime {
DiagnosticSeverity::Info,
format!(
"Remote Runtime acknowledged lifecycle operation for '{worker_id}' with status {}",
embedded_worker_status_label(response.ack.status)
embedded_worker_status_label(response.status)
),
)],
}
@@ -4010,25 +4007,27 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
|client| async move { client.runtime_summary().await },
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpSummaryResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::management::RuntimeSummary>(
value.runtime,
)
}) {
Ok(response) => RuntimeSummary {
runtime_id: self.runtime_id.clone(),
label: response
.runtime
.display_name
.unwrap_or_else(|| self.display_name.clone()),
kind: "remote_worker_runtime".to_string(),
status: embedded_runtime_status_label(response.runtime.status).to_string(),
status: embedded_runtime_status_label(response.status).to_string(),
source: RuntimeSourceSummary::remote_http(),
host_ids: if limit == 0 {
Vec::new()
} else {
vec![self.host_id.clone()]
},
worker_creation_available: response.runtime.worker_creation_available,
os: response.runtime.os,
arch: response.runtime.arch,
worker_creation_available: response.worker_creation_available,
os: response.os,
arch: response.arch,
diagnostics: Vec::new(),
},
Err(diagnostic) => RuntimeSummary {
@@ -4162,11 +4161,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.await
},
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkersResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, Vec<worker_runtime::catalog::WorkerSummary>>(
value.workers,
)
}) {
Ok(response) => RuntimeList::new(
response
.workers
.into_iter()
.take(limit)
.map(|worker| self.map_worker_summary(worker))
@@ -4193,11 +4194,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.await
},
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkersResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, Vec<worker_runtime::catalog::WorkerSummary>>(
value.workers,
)
}) {
Ok(response) => RuntimeList::new(
response
.workers
.into_iter()
.take(limit)
.map(|worker| self.map_worker_summary(worker))
@@ -4216,10 +4219,11 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move { client.get_worker(worker_id_owned).await },
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker)
}) {
Ok(response) => WorkerLookupResult {
worker: Some(self.map_worker_detail(response.worker)),
worker: Some(self.map_worker_detail(response)),
diagnostics: Vec::new(),
},
Err(diagnostic) if diagnostic.code == "worker_not_found" => WorkerLookupResult {
@@ -4245,11 +4249,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.await
},
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker)
}) {
Ok(response) => WorkerRestoreResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response.worker)),
worker: Some(self.map_worker_detail(response)),
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerRestoreResult {
@@ -4265,10 +4270,8 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
worker_id: &str,
workspace_api: WorkspaceApiRef,
) -> WorkerWorkspaceApiResult {
let request = match runtime_contract_convert(RuntimeHttpWorkerWorkspaceApiRequest {
workspace_api,
}) {
Ok(request) => request,
let request = match runtime_contract_convert(workspace_api) {
Ok(workspace_api) => runtime_api::WorkerWorkspaceApiRequest { workspace_api },
Err(diagnostic) => {
return WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
@@ -4288,11 +4291,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
.await
},
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker)
}) {
Ok(response) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response.worker)),
worker: Some(self.map_worker_detail(response)),
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerWorkspaceApiResult {
@@ -4506,11 +4510,12 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move { client.create_worker(create).await },
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker)
}) {
Ok(response) => WorkerSpawnResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response.worker)),
worker: Some(self.map_worker_detail(response)),
acceptance_evidence: vec![WorkerSpawnAcceptanceEvidence {
kind: "remote_runtime_worker_created".to_string(),
detail: "worker-runtime REST create endpoint accepted the Worker".to_string(),
@@ -4589,7 +4594,9 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
move |client| async move { client.stop_worker(worker_id_owned, body).await },
)
.and_then(|value| {
runtime_contract_convert::<_, RuntimeHttpWorkerLifecycleResponse>(value)
runtime_contract_convert::<_, worker_runtime::catalog::WorkerLifecycleAck>(
value.ack,
)
}) {
Ok(response) => self.lifecycle_result_from_response(worker_id, response),
Err(diagnostic) => remote_lifecycle_rejected(&self.runtime_id, worker_id, diagnostic),
@@ -4612,7 +4619,9 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
move |client| async move { client.cancel_worker(worker_id_owned, body).await },
)
.and_then(|value| {
runtime_contract_convert::<_, RuntimeHttpWorkerLifecycleResponse>(value)
runtime_contract_convert::<_, worker_runtime::catalog::WorkerLifecycleAck>(
value.ack,
)
}) {
Ok(response) => self.lifecycle_result_from_response(worker_id, response),
Err(diagnostic) => remote_lifecycle_rejected(&self.runtime_id, worker_id, diagnostic),
@@ -4627,15 +4636,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move { client.delete_worker(worker_id_owned).await },
)
.and_then(|value| runtime_contract_convert::<_, RuntimeHttpWorkerDeleteResponse>(value))
{
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::management::WorkerDeleteResult>(
value.worker,
)
}) {
Ok(response) => WorkerDeleteResult {
state: WorkerOperationState::Accepted,
worker: RuntimeWorkerRef::new(
self.runtime_id.clone(),
response.worker.worker_id.to_string(),
response.worker_id.to_string(),
),
deleted: response.worker.deleted,
deleted: response.deleted,
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerDeleteResult {
@@ -4799,19 +4811,15 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
let failure_kind = request.kind;
let failure_prefix = request.prefix.clone();
let worker_id_owned = worker_id.to_string();
match self
.run_runtime_api(
self.request_timeout,
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move {
client
.complete_worker_arguments(worker_id_owned, request)
.await
},
)
.and_then(|value| {
runtime_contract_convert::<_, RuntimeHttpWorkerCompletionsResponse>(value)
}) {
match self.run_runtime_api(
self.request_timeout,
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move {
client
.complete_worker_arguments(worker_id_owned, request)
.await
},
) {
Ok(response) => WorkerCompletionsResult {
worker: RuntimeWorkerRef::new(self.runtime_id.clone(), worker_id.to_string()),
kind: response.kind,