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

feat: migrate runtime worker domains to ServerApi

Browse Source
This commit is contained in:
Hare
2026-09-23 08:24:39 +09:00
parent 7a975765ac
commit ad12d77ad4
28 changed files with 10269 additions and 896 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/client/src/backend_runtime.rs
+142 -190
View File
@@ -88,28 +88,25 @@ impl BackendRuntimeTarget {
content: Vec<u8>,
) -> Result<protocol::UploadedFileRef, BackendRuntimeClientError> {
let api = BackendApiClient::from_stored_token(&self.base_url)?;
let client = runtime_server_api_client(&self.base_url, &api)?;
let grant = client
.runtime_worker_attachment_upload_grant(
self.workspace_id.clone(),
self.runtime_id.clone(),
self.worker_id.clone(),
server_api::WorkerFileUploadQuery {
file_name: file_name.to_string(),
media_type: media_type.to_string(),
upload_id: Some(upload_id.to_string()),
},
)
.await?;
let worker_path = format!(
"/api/w/{}/runtimes/{}/workers/{}",
path_segment_encode(&self.workspace_id),
path_segment_encode(&self.runtime_id),
path_segment_encode(&self.worker_id),
);
let grant_path = format!(
"{worker_path}/attachment-upload-grants?file_name={}&media_type={}&upload_id={}",
path_segment_encode(file_name),
path_segment_encode(media_type),
path_segment_encode(&upload_id),
);
let grant_response = api
.request(HttpMethod::POST, &grant_path)?
.send()
.await
.map_err(BackendRuntimeClientError::Http)?;
api.check_status(grant_response.status())?;
let grant = grant_response
.json::<AttachmentUploadGrantResponse>()
.await
.map_err(BackendRuntimeClientError::Http)?;
let upload_path = format!(
"{worker_path}/attachment-uploads/{}",
path_segment_encode(&grant.upload_id),
@@ -133,19 +130,14 @@ impl BackendRuntimeTarget {
upload_id: &str,
) -> Result<(), BackendRuntimeClientError> {
let api = BackendApiClient::from_stored_token(&self.base_url)?;
let path = format!(
"/api/w/{}/runtimes/{}/workers/{}/attachment-uploads/{}",
path_segment_encode(&self.workspace_id),
path_segment_encode(&self.runtime_id),
path_segment_encode(&self.worker_id),
path_segment_encode(upload_id),
);
let response = api
.request(HttpMethod::DELETE, &path)?
.send()
.await
.map_err(BackendRuntimeClientError::Http)?;
api.check_status(response.status())?;
runtime_server_api_client(&self.base_url, &api)?
.runtime_worker_attachment_upload_cancel(
self.workspace_id.clone(),
self.runtime_id.clone(),
self.worker_id.clone(),
upload_id.to_string(),
)
.await?;
Ok(())
}
@@ -154,30 +146,18 @@ impl BackendRuntimeTarget {
artifact_id: &str,
) -> Result<(), BackendRuntimeClientError> {
let api = BackendApiClient::from_stored_token(&self.base_url)?;
let path = format!(
"/api/w/{}/runtimes/{}/workers/{}/attachments/{}",
path_segment_encode(&self.workspace_id),
path_segment_encode(&self.runtime_id),
path_segment_encode(&self.worker_id),
path_segment_encode(artifact_id),
);
let response = api
.request(HttpMethod::DELETE, &path)?
.send()
.await
.map_err(BackendRuntimeClientError::Http)?;
api.check_status(response.status())?;
runtime_server_api_client(&self.base_url, &api)?
.runtime_worker_attachment_delete(
self.workspace_id.clone(),
self.runtime_id.clone(),
self.worker_id.clone(),
artifact_id.to_string(),
)
.await?;
Ok(())
}
}
#[derive(Deserialize)]
struct AttachmentUploadGrantResponse {
upload_id: String,
#[allow(dead_code)]
expires_at_ms: u64,
}
#[derive(Deserialize)]
struct UploadedFileResponse {
file: protocol::UploadedFileRef,
@@ -280,6 +260,7 @@ pub enum BackendRuntimeClientError {
InvalidTarget(String),
Api(BackendApiClientError),
SessionApi(server_api::ServerApiClientError),
ContractApi(server_api::client_support::ClientError<server_api::RepositoryApiError>),
Http(reqwest::Error),
Protocol(String),
}
@@ -290,6 +271,7 @@ impl fmt::Display for BackendRuntimeClientError {
Self::InvalidTarget(message) => f.write_str(message),
Self::Api(error) => write!(f, "{error}"),
Self::SessionApi(error) => write!(f, "{error}"),
Self::ContractApi(error) => write!(f, "{error}"),
Self::Http(error) => write!(f, "{error}"),
Self::Protocol(message) => f.write_str(message),
}
@@ -304,6 +286,16 @@ impl From<server_api::ServerApiClientError> for BackendRuntimeClientError {
}
}
impl From<server_api::client_support::ClientError<server_api::RepositoryApiError>>
for BackendRuntimeClientError
{
fn from(
value: server_api::client_support::ClientError<server_api::RepositoryApiError>,
) -> Self {
Self::ContractApi(value)
}
}
impl From<BackendApiClientError> for BackendRuntimeClientError {
fn from(error: BackendApiClientError) -> Self {
Self::Api(error)
@@ -318,6 +310,34 @@ impl From<reqwest::Error> for BackendRuntimeClientError {
const WORKER_SESSION_RESPONSE_LIMIT: usize = 16 * 1024 * 1024;
fn runtime_server_api_client(
base_url: &str,
backend: &BackendApiClient,
) -> Result<server_api::ServerApiClient<StoredBearerAuthorizer>, BackendRuntimeClientError> {
server_api::ServerApiClient::builder(base_url)
.map_err(|error| BackendRuntimeClientError::Protocol(error.to_string()))?
.client(backend.asynchronous_client())
.authorizer(StoredBearerAuthorizer {
authorization: backend.authorization_header_value(),
})
.response_body_limit(WORKER_SESSION_RESPONSE_LIMIT)
.build()
.map_err(|error| BackendRuntimeClientError::Protocol(error.to_string()))
}
fn server_api_error_is_auth(
error: &server_api::client_support::ClientError<server_api::RepositoryApiError>,
) -> bool {
matches!(
error,
server_api::client_support::ClientError::Public { status, .. }
if matches!(
*status,
reqwest::StatusCode::UNAUTHORIZED | reqwest::StatusCode::FORBIDDEN
)
)
}
#[derive(Clone, Debug)]
struct StoredBearerAuthorizer {
authorization: String,
@@ -351,15 +371,7 @@ async fn observe_backend_worker_session_with_client(
target: &BackendRuntimeTarget,
backend: &BackendApiClient,
) -> Result<server_api::WorkspaceWorkerSessionResponse, BackendRuntimeClientError> {
let client = server_api::ServerApiClient::builder(&target.base_url)
.map_err(|error| BackendRuntimeClientError::Protocol(error.to_string()))?
.authorizer(StoredBearerAuthorizer {
authorization: backend.authorization_header_value(),
})
.response_body_limit(WORKER_SESSION_RESPONSE_LIMIT)
.build()
.map_err(|error| BackendRuntimeClientError::Protocol(error.to_string()))?;
client
runtime_server_api_client(&target.base_url, backend)?
.worker_session(
target.workspace_id.clone(),
target.runtime_id.clone(),
@@ -426,80 +438,50 @@ pub async fn list_backend_workers(
) -> Result<BackendRuntimeListResponse<BackendWorkerSummary>, BackendRuntimeClientError> {
validate_list_target(target)?;
let api = BackendApiClient::from_stored_token(&target.base_url)?;
let client = runtime_server_api_client(&target.base_url, &api)?;
let workspace_id = target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope");
if let Some(runtime_id) = target.runtime_id.as_deref() {
let path = backend_runtime_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
runtime_id,
);
let response = api.request(HttpMethod::GET, &path)?.send().await?;
api.check_status(response.status())?;
return Ok(response
.json::<BackendRuntimeListResponse<BackendWorkerSummary>>()
.await?);
return client
.runtime_worker_list(
workspace_id.to_string(),
runtime_id.to_string(),
server_api::RuntimeWorkersQuery::default(),
)
.await
.map(runtime_worker_list_response)
.map_err(Into::into);
}
let runtime_path = backend_runtimes_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
);
let response = api.request(HttpMethod::GET, &runtime_path)?.send().await?;
api.check_status(response.status())?;
let runtimes = response
.json::<BackendRuntimeListResponse<BackendRuntimeSummary>>()
.await?;
let runtimes = client.runtime_list(workspace_id.to_string()).await?;
let mut items = Vec::new();
let mut diagnostics = runtimes.diagnostics;
for runtime in runtimes.items {
let path = backend_runtime_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
&runtime.runtime_id,
);
let response = match api.request(HttpMethod::GET, &path)?.send().await {
Ok(response) => response,
Err(error) => {
diagnostics.push(BackendDiagnostic {
code: "runtime_worker_list_failed".to_string(),
severity: BackendDiagnosticSeverity::Error,
message: format!(
"failed to list workers for runtime {}: {error}",
runtime.runtime_id
),
});
continue;
for resource in runtimes.items {
let runtime = resource.runtime;
match client
.runtime_worker_list(
workspace_id.to_string(),
runtime.runtime_id.clone(),
server_api::RuntimeWorkersQuery::default(),
)
.await
{
Ok(response) => {
diagnostics.extend(response.diagnostics);
items.extend(response.items);
}
};
if matches!(
response.status(),
reqwest::StatusCode::UNAUTHORIZED | reqwest::StatusCode::FORBIDDEN
) {
api.check_status(response.status())?;
}
if !response.status().is_success() {
diagnostics.push(BackendDiagnostic {
Err(error) if server_api_error_is_auth(&error) => return Err(error.into()),
Err(error) => diagnostics.push(BackendDiagnostic {
code: "runtime_worker_list_failed".to_string(),
severity: BackendDiagnosticSeverity::Error,
message: format!(
"failed to list workers for runtime {}: Backend returned HTTP {}",
runtime.runtime_id,
response.status().as_u16()
"failed to list workers for runtime {}: {error}",
runtime.runtime_id
),
});
continue;
}),
}
let response = response
.json::<BackendRuntimeListResponse<BackendWorkerSummary>>()
.await?;
diagnostics.extend(response.diagnostics);
items.extend(response.items);
}
Ok(BackendRuntimeListResponse {
@@ -521,21 +503,21 @@ pub async fn list_backend_stopped_workers(
));
};
let api = BackendApiClient::from_stored_token(&target.base_url)?;
let path = backend_runtime_workers_path(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope"),
runtime_id,
);
let response = api
.request(HttpMethod::GET, &format!("{path}?status=stopped"))?
.send()
.await?;
api.check_status(response.status())?;
Ok(response
.json::<BackendRuntimeListResponse<BackendWorkerSummary>>()
.await?)
runtime_server_api_client(&target.base_url, &api)?
.runtime_worker_list(
target
.workspace_id
.as_deref()
.expect("validated Backend Workspace scope")
.to_string(),
runtime_id.to_string(),
server_api::RuntimeWorkersQuery {
status: Some(server_api::RuntimeWorkersStatusFilter::Stopped),
},
)
.await
.map(runtime_worker_list_response)
.map_err(Into::into)
}
pub async fn restore_backend_worker(
@@ -543,18 +525,18 @@ pub async fn restore_backend_worker(
) -> Result<BackendWorkerRestoreResponse, BackendRuntimeClientError> {
validate_target(target)?;
let api = BackendApiClient::from_stored_token(&target.base_url)?;
let path = backend_runtime_worker_restore_path(
&target.workspace_id,
&target.runtime_id,
&target.worker_id,
);
let response = api
.request(HttpMethod::POST, &path)?
.json(&serde_json::json!({}))
.send()
.await?;
let response = api.require_success(response).await?;
Ok(response.json::<BackendWorkerRestoreResponse>().await?)
runtime_server_api_client(&target.base_url, &api)?
.runtime_worker_restore(
target.workspace_id.clone(),
target.runtime_id.clone(),
target.worker_id.clone(),
server_api::RestoreTicketAssignmentQuery {
ticket_id: None,
assignment_operation_id: None,
},
)
.await
.map_err(Into::into)
}
pub async fn connect_backend_runtime(
@@ -599,6 +581,18 @@ fn protocol_connect_error_message(
)
}
fn runtime_worker_list_response(
response: server_api::RuntimeWorkerListResponse,
) -> BackendRuntimeListResponse<BackendWorkerSummary> {
BackendRuntimeListResponse {
workspace_id: response.workspace_id,
limit: response.limit,
items: response.items,
source: response.source,
diagnostics: response.diagnostics,
}
}
fn validate_target(target: &BackendRuntimeTarget) -> Result<(), BackendRuntimeClientError> {
if target.base_url.trim().is_empty() {
return Err(BackendRuntimeClientError::InvalidTarget(
@@ -697,31 +691,6 @@ fn backend_workspace_workers_launch_options_path(workspace_id: &str) -> String {
)
}
fn backend_runtimes_path(workspace_id: &str) -> String {
format!("/api/w/{}/runtimes", path_segment_encode(workspace_id))
}
fn backend_runtime_workers_path(workspace_id: &str, runtime_id: &str) -> String {
format!(
"/api/w/{}/runtimes/{}/workers",
path_segment_encode(workspace_id),
path_segment_encode(runtime_id)
)
}
fn backend_runtime_worker_restore_path(
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
) -> String {
format!(
"/api/w/{}/runtimes/{}/workers/{}/restore",
path_segment_encode(workspace_id),
path_segment_encode(runtime_id),
path_segment_encode(worker_id)
)
}
fn protocol_ws_request(
target: &BackendRuntimeTarget,
api: &BackendApiClient,
@@ -1051,21 +1020,4 @@ mod tests {
serde_json::json!(64);
assert!(serde_json::from_value::<BackendWorkerSummary>(stale).is_err());
}
#[test]
fn workers_path_requires_workspace_scope_for_status_queries() {
let path = backend_runtime_workers_path("team main", "runtime/one");
assert_eq!(
format!("{path}?status=stopped"),
"/api/w/team%20main/runtimes/runtime%2Fone/workers?status=stopped"
);
}
#[test]
fn restore_worker_path_requires_workspace_scope() {
assert_eq!(
backend_runtime_worker_restore_path("team main", "runtime/one", "worker one"),
"/api/w/team%20main/runtimes/runtime%2Fone/workers/worker%20one/restore"
);
}
}
crates/client/src/backend_workspace.rs
+27
View File
@@ -31,6 +31,9 @@ pub enum BackendWorkspaceClientError {
Api(BackendApiClientError),
Http(reqwest::Error),
ServerApi(server_api::client_support::ClientError<server_api::RepositoryApiError>),
RuntimeManagementApi(
server_api::client_support::ClientError<server_api::RuntimeManagementApiError>,
),
}
impl fmt::Display for BackendWorkspaceClientError {
@@ -40,6 +43,7 @@ impl fmt::Display for BackendWorkspaceClientError {
Self::Api(error) => write!(f, "{error}"),
Self::Http(error) => write!(f, "{error}"),
Self::ServerApi(error) => write!(f, "{error}"),
Self::RuntimeManagementApi(error) => write!(f, "{error}"),
}
}
}
@@ -91,6 +95,29 @@ pub(crate) fn server_client_error(
}
}
pub(crate) fn runtime_management_client_error(
backend: &BackendApiClient,
error: server_api::client_support::ClientError<server_api::RuntimeManagementApiError>,
) -> BackendWorkspaceClientError {
let status = match &error {
server_api::client_support::ClientError::Public { status, .. } => Some(*status),
server_api::client_support::ClientError::Failure(_) => None,
};
match status {
Some(reqwest::StatusCode::UNAUTHORIZED) => {
BackendWorkspaceClientError::Api(BackendApiClientError::Unauthorized {
origin: backend.origin().clone(),
})
}
Some(reqwest::StatusCode::FORBIDDEN) => {
BackendWorkspaceClientError::Api(BackendApiClientError::Forbidden {
origin: backend.origin().clone(),
})
}
_ => BackendWorkspaceClientError::RuntimeManagementApi(error),
}
}
#[derive(Clone, Debug)]
pub(crate) struct ServerBearerAuthorizer {
authorization: String,
crates/client/src/workspace_product.rs
+36 -11
View File
@@ -345,24 +345,40 @@ impl BackendWorkspaceProductClient {
pub fn list_runtimes(
&self,
) -> Result<ListResponse<WorkspaceRuntimeResource>, BackendWorkspaceClientError> {
self.get_json("/runtimes")
let workspace_id = self.workspace_id.clone();
let response =
self.generated(move |client| async move { client.runtime_list(workspace_id).await })?;
Ok(ListResponse {
workspace_id: response.workspace_id,
limit: response.limit,
items: response.items,
source: response.source,
diagnostics: response.diagnostics,
})
}
pub fn runtime_detail(
&self,
runtime_id: &str,
) -> Result<WorkspaceRuntimeDetail, BackendWorkspaceClientError> {
self.get_json(&format!("/runtimes/{}", encode_path_segment(runtime_id)))
let workspace_id = self.workspace_id.clone();
let runtime_id = runtime_id.to_string();
self.generated(move |client| async move {
client.runtime_detail(workspace_id, runtime_id).await
})
}
pub fn reveal_runtime_trust_key(
&self,
runtime_id: &str,
) -> Result<RuntimeTrustKeyRevealResponse, BackendWorkspaceClientError> {
self.get_json(&format!(
"/runtimes/{}/trust-key",
encode_path_segment(runtime_id)
))
let workspace_id = self.workspace_id.clone();
let runtime_id = runtime_id.to_string();
self.generated(move |client| async move {
client
.runtime_trust_key_reveal(workspace_id, runtime_id)
.await
})
}
pub fn revoke_runtime_trust_key(
@@ -370,11 +386,20 @@ impl BackendWorkspaceProductClient {
runtime_id: &str,
request: &RevokeRuntimeTrustKeyRequest,
) -> Result<WorkspaceRuntimeDetail, BackendWorkspaceClientError> {
self.send_json(
Method::DELETE,
&format!("/runtimes/{}/trust-key", encode_path_segment(runtime_id)),
Some(request),
)
let client = crate::backend_workspace::server_api_client(&self.api)?;
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| BackendWorkspaceClientError::InvalidTarget(error.to_string()))?;
runtime
.block_on(client.runtime_trust_key_revoke(
self.workspace_id.clone(),
runtime_id.to_string(),
request.clone(),
))
.map_err(|error| {
crate::backend_workspace::runtime_management_client_error(&self.api, error)
})
}
pub fn memory_document(&self) -> Result<MemoryDocumentResponse, BackendWorkspaceClientError> {