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

runtime: own working directories

Browse Source
This commit is contained in:
Hare
2026-07-09 19:04:37 +09:00
parent afc43780a7
commit fa233ba025
9 changed files with 798 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/execution.rs
+63 -2
View File
@@ -1,11 +1,11 @@
use crate::catalog::{CreateWorkerRequest, WorkingDirectoryStatus};
use crate::catalog::{CreateWorkerRequest, WorkingDirectoryRequest, WorkingDirectoryStatus};
use crate::config_bundle::ConfigBundle;
use crate::error::RuntimeError;
use crate::identity::WorkerRef;
use crate::interaction::WorkerInput;
#[cfg(feature = "ws-server")]
use crate::observation::WorkerObservationEvent;
use crate::working_directory::WorkingDirectoryBinding;
use crate::working_directory::{WorkingDirectoryBinding, WorkingDirectoryDiagnostic};
use serde::{Deserialize, Serialize};
use std::fmt;
use std::sync::Arc;
@@ -313,6 +313,42 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult;
fn create_working_directory(
&self,
_request: &WorkingDirectoryRequest,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_unsupported",
"Worker execution backend does not support working directory materialization",
))
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
Vec::new()
}
fn working_directory(
&self,
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_unknown",
format!("working directory `{working_directory_id}` is not known to this Runtime"),
))
}
fn cleanup_working_directory(
&self,
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_cleanup_unsupported",
format!(
"working directory `{working_directory_id}` cannot be cleaned up by this backend"
),
))
}
fn dispatch_input(
&self,
handle: &WorkerExecutionHandle,
@@ -358,6 +394,31 @@ impl WorkerExecutionBackendRef {
self.backend.spawn_worker(request)
}
pub(crate) fn create_working_directory(
&self,
request: &WorkingDirectoryRequest,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
self.backend.create_working_directory(request)
}
pub(crate) fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.backend.list_working_directories()
}
pub(crate) fn working_directory(
&self,
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
self.backend.working_directory(working_directory_id)
}
pub(crate) fn cleanup_working_directory(
&self,
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
self.backend.cleanup_working_directory(working_directory_id)
}
pub(crate) fn dispatch_input(
&self,
handle: &WorkerExecutionHandle,
crates/worker-runtime/src/http_server.rs
+73
View File
@@ -9,6 +9,7 @@
use crate::Runtime;
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
WorkingDirectoryRequest, WorkingDirectoryStatus,
};
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
use crate::error::RuntimeError;
@@ -138,6 +139,14 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option<String>) -> Rou
"/v1/config-bundles/{bundle_id}/availability",
get(check_config_bundle),
)
.route(
"/v1/working-directories",
get(list_working_directories).post(create_working_directory),
)
.route(
"/v1/working-directories/{working_directory_id}",
get(get_working_directory).delete(cleanup_working_directory),
)
.route("/v1/workers", get(list_workers).post(create_worker))
.route("/v1/workers/{worker_id}", get(get_worker))
.route("/v1/workers/{worker_id}/input", post(send_worker_input))
@@ -197,6 +206,18 @@ pub struct RuntimeHttpWorkersResponse {
pub workers: Vec<WorkerSummary>,
}
/// `GET /v1/working-directories` response.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkingDirectoriesResponse {
pub working_directories: Vec<WorkingDirectoryStatus>,
}
/// Working directory response used by create/detail/delete endpoints.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkingDirectoryResponse {
pub working_directory: WorkingDirectoryStatus,
}
/// Worker detail response used by create/detail endpoints.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkerResponse {
@@ -399,6 +420,58 @@ async fn list_workers(
Ok(Json(RuntimeHttpWorkersResponse { workers }))
}
async fn list_working_directories(
State(state): State<RuntimeHttpState>,
) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
let working_directories = state
.runtime
.list_working_directories()
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkingDirectoriesResponse {
working_directories,
}))
}
async fn create_working_directory(
State(state): State<RuntimeHttpState>,
body: Result<Json<WorkingDirectoryRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
let working_directory = state
.runtime
.create_working_directory(request)
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkingDirectoryResponse {
working_directory,
}))
}
async fn get_working_directory(
State(state): State<RuntimeHttpState>,
Path(working_directory_id): Path<String>,
) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
let working_directory = state
.runtime
.working_directory(&working_directory_id)
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkingDirectoryResponse {
working_directory,
}))
}
async fn cleanup_working_directory(
State(state): State<RuntimeHttpState>,
Path(working_directory_id): Path<String>,
) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
let working_directory = state
.runtime
.cleanup_working_directory(&working_directory_id)
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkingDirectoryResponse {
working_directory,
}))
}
async fn get_worker(
State(state): State<RuntimeHttpState>,
Path(worker_id): Path<String>,
crates/worker-runtime/src/runtime.rs
+75 -1
View File
@@ -1,6 +1,7 @@
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck,
WorkerStatus, WorkerSummary, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
WorkerStatus, WorkerSummary, WorkingDirectoryRequest,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
};
use crate::config_bundle::{
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
@@ -234,6 +235,79 @@ impl Runtime {
Ok(event_id)
}
/// Create a Runtime-owned working directory through the attached execution backend.
pub fn create_working_directory(
&self,
request: WorkingDirectoryRequest,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory creation requires an execution backend".to_string(),
}
})?
};
backend
.create_working_directory(&request)
.map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string()))
}
/// List Runtime-owned working directories through the attached execution backend.
pub fn list_working_directories(
&self,
) -> Result<Vec<CatalogWorkingDirectoryStatus>, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory listing requires an execution backend".to_string(),
}
})?
};
Ok(backend.list_working_directories())
}
/// Get a Runtime-owned working directory status.
pub fn working_directory(
&self,
working_directory_id: &str,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory lookup requires an execution backend".to_string(),
}
})?
};
backend
.working_directory(working_directory_id)
.map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string()))
}
/// Cleanup a Runtime-owned working directory.
pub fn cleanup_working_directory(
&self,
working_directory_id: &str,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory cleanup requires an execution backend".to_string(),
}
})?
};
backend
.cleanup_working_directory(working_directory_id)
.map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string()))
}
/// Create a Worker through the canonical ConfigBundle + execution backend path.
pub fn create_worker(
&self,
crates/worker-runtime/src/worker_backend.rs
+50 -1
View File
@@ -14,6 +14,7 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;
use crate::catalog::{WorkingDirectoryRequest, WorkingDirectoryStatus};
use crate::config_bundle::verified_profile_source_archive;
use crate::execution::{
WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult,
@@ -24,7 +25,9 @@ use crate::resource::{
BackendResourceClient, BackendResourceError, ProfileSourceArchiveCache,
build_profile_source_archive_fetch_request, profile_source_archive_from_response,
};
use crate::working_directory::{WorkingDirectoryBinding, WorkingDirectoryMaterializer};
use crate::working_directory::{
WorkingDirectoryBinding, WorkingDirectoryDiagnostic, WorkingDirectoryMaterializer,
};
use async_trait::async_trait;
use manifest::paths;
use protocol::{Method, Segment, WorkerStatus};
@@ -441,6 +444,52 @@ where
&self.backend_id
}
fn create_working_directory(
&self,
request: &WorkingDirectoryRequest,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_materializer_unavailable",
"working directory materialization requested, but no materializer is configured for this runtime backend",
));
};
Ok(materializer.create(request)?.status())
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.working_directory_materializer
.as_ref()
.and_then(|materializer| materializer.list_working_directories().ok())
.unwrap_or_default()
}
fn working_directory(
&self,
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_materializer_unavailable",
"working directory lookup requested, but no materializer is configured for this runtime backend",
));
};
materializer.working_directory_status(working_directory_id)
}
fn cleanup_working_directory(
&self,
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_materializer_unavailable",
"working directory cleanup requested, but no materializer is configured for this runtime backend",
));
};
materializer.cleanup_working_directory(working_directory_id)
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
if self
.workers
crates/worker-runtime/src/working_directory.rs
+4
View File
@@ -95,6 +95,10 @@ pub struct WorkingDirectoryDiagnostic {
}
impl WorkingDirectoryDiagnostic {
pub fn rejected(code: impl Into<String>, message: impl Into<String>) -> Self {
Self::new(code, message)
}
fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
Self {
code: code.into(),