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

workspace: gate workdir assignment

Browse Source
This commit is contained in:
Hare
2026-07-12 20:07:32 +09:00
parent 21802d68f8
commit 59b1b3df79
9 changed files with 161 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+2
View File
@@ -154,6 +154,8 @@ pub struct WorkingDirectorySummary {
pub status: WorkingDirectoryStatusKind,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cleanliness: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub primary_worker_id: Option<WorkerId>,
/// Backend projection metadata. Runtimes leave this absent; Workspace Browser
/// APIs fill it with `backend_managed` or `runtime_unmanaged`.
#[serde(default, skip_serializing_if = "Option::is_none")]
crates/worker-runtime/src/runtime.rs
+68 -3
View File
@@ -270,7 +270,8 @@ impl Runtime {
}
})?
};
Ok(backend.list_working_directories())
let statuses = backend.list_working_directories();
self.annotate_working_directory_statuses(statuses)
}
/// Get a Runtime-owned working directory status.
@@ -287,9 +288,10 @@ impl Runtime {
}
})?
};
backend
let status = backend
.working_directory(working_directory_id)
.map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string()))
.map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string()))?;
self.annotate_working_directory_status(status)
}
/// Cleanup a Runtime-owned working directory.
@@ -300,6 +302,11 @@ impl Runtime {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
if let Some(worker_id) = state.primary_worker_id_for_workdir(working_directory_id) {
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is assigned to worker {worker_id}"
)));
}
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory cleanup requires an execution backend".to_string(),
@@ -311,6 +318,26 @@ impl Runtime {
.map_err(|diagnostic| RuntimeError::InvalidRequest(diagnostic.to_string()))
}
fn annotate_working_directory_statuses(
&self,
statuses: Vec<CatalogWorkingDirectoryStatus>,
) -> Result<Vec<CatalogWorkingDirectoryStatus>, RuntimeError> {
statuses
.into_iter()
.map(|status| self.annotate_working_directory_status(status))
.collect()
}
fn annotate_working_directory_status(
&self,
mut status: CatalogWorkingDirectoryStatus,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let state = self.lock()?;
status.summary.primary_worker_id =
state.primary_worker_id_for_workdir(status.summary.working_directory_id.as_str());
Ok(status)
}
/// Create a Worker through the canonical profile-source + execution backend path.
pub fn create_worker(
&self,
@@ -321,6 +348,15 @@ impl Runtime {
state.ensure_running()?;
validate_create_worker_request(&request)?;
state.validate_worker_config_boundary(&request)?;
if let Some(working_directory_id) = requested_primary_workdir_id(&request) {
if let Some(owner_worker_id) =
state.primary_worker_id_for_workdir(working_directory_id)
{
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is already assigned to worker {owner_worker_id}"
)));
}
}
let backend = state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "worker creation requires an execution backend".to_string(),
@@ -1342,6 +1378,22 @@ impl RuntimeState {
})
}
fn primary_worker_id_for_workdir(&self, working_directory_id: &str) -> Option<WorkerId> {
self.workers.values().find_map(|worker| {
if worker
.execution
.working_directory
.as_ref()
.is_some_and(|binding| binding.summary.working_directory_id == working_directory_id)
|| requested_primary_workdir_id(&worker.request) == Some(working_directory_id)
{
Some(worker.worker_id)
} else {
None
}
})
}
fn push_event(
&mut self,
worker_ref: Option<WorkerRef>,
@@ -1510,6 +1562,19 @@ impl WorkerRecord {
}
}
fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> {
request
.working_directory
.as_ref()
.map(|claim| claim.working_directory_id.as_str())
.or_else(|| {
request
.working_directory_request
.as_ref()
.and_then(|request| request.backend_workdir_id.as_deref())
})
}
fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> {
match &request.profile_source {
crate::catalog::ProfileSourceArchiveSource::Embedded { archive } => {
crates/worker-runtime/src/worker_backend.rs
+2 -2
View File
@@ -866,8 +866,8 @@ mod tests {
use crate::Runtime as EmbeddedRuntime;
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, MaterializerKind, ProfileSelector, RepositorySelector,
WorkingDirectoryRepository, WorkingDirectoryRequest,
ConfigBundleRef, CreateWorkerRequest, MaterializerKind, ProfileSelector,
RepositorySelector, WorkingDirectoryRepository, WorkingDirectoryRequest,
};
use crate::execution::WorkerExecutionContext;
use crate::identity::RuntimeId;
crates/worker-runtime/src/working_directory.rs
+4 -2
View File
@@ -1,6 +1,6 @@
use crate::catalog::{
MaterializerKind, WorkingDirectoryCleanupTarget, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectoryStatusKind, WorkingDirectorySummary,
MaterializerKind, WorkingDirectoryCleanupTarget, WorkingDirectoryRequest,
WorkingDirectoryStatus, WorkingDirectoryStatusKind, WorkingDirectorySummary,
};
use crate::identity::WorkerRef;
use serde::{Deserialize, Serialize};
@@ -47,6 +47,7 @@ impl WorkingDirectory {
cleanup_target: Some(self.cleanup_target.clone()),
status: self.status.clone(),
cleanliness: None,
primary_worker_id: None,
management_kind: None,
}
}
@@ -220,6 +221,7 @@ impl LocalGitWorktreeMaterializer {
}),
status: WorkingDirectoryStatusKind::Corrupted,
cleanliness: Some("unknown".to_string()),
primary_worker_id: None,
management_kind: None,
},
}
crates/workspace-server/src/server.rs
+43 -11
View File
@@ -1709,8 +1709,7 @@ fn build_runtime_cleanup_plan(
reason: if blocking_reason.is_some() {
"Workdir cleanup is blocked until linked Worker state is safe".to_string()
} else if matches!(file_status.as_str(), "missing" | "not_found") {
"Not-found Workdir record can be deleted from the Backend registry"
.to_string()
"Not-found Workdir record can be deleted from the Backend registry".to_string()
} else if file_status == "corrupted" {
"Corrupted Workdir can be deleted from Runtime storage and Backend registry"
.to_string()
@@ -1848,9 +1847,10 @@ fn execute_runtime_cleanup(
));
}
cleanup_runtime_workdir_for_execution(api, runtime_id, candidate)?;
let deleted = api
.store
.delete_workdir_registry(&api.config.workspace_id, candidate.workdir_id.as_str())?;
let deleted = api.store.delete_workdir_registry(
&api.config.workspace_id,
candidate.workdir_id.as_str(),
)?;
if !deleted {
return Err(cleanup_api_error(
runtime_id,
@@ -1869,9 +1869,10 @@ fn execute_runtime_cleanup(
}
CleanupTargetKind::WorkdirCleanCleanup => {
cleanup_runtime_workdir_for_execution(api, runtime_id, candidate)?;
let deleted = api
.store
.delete_workdir_registry(&api.config.workspace_id, candidate.workdir_id.as_str())?;
let deleted = api.store.delete_workdir_registry(
&api.config.workspace_id,
candidate.workdir_id.as_str(),
)?;
if !deleted {
return Err(cleanup_api_error(
runtime_id,
@@ -1883,7 +1884,8 @@ fn execute_runtime_cleanup(
target_id: candidate.target_id.clone(),
action: candidate.action.clone(),
status: "deleted".to_string(),
message: "Workdir deleted from Runtime storage and Backend registry".to_string(),
message: "Workdir deleted from Runtime storage and Backend registry"
.to_string(),
});
}
CleanupTargetKind::WorkdirRecordDelete => {
@@ -3933,7 +3935,7 @@ fn worker_launch_options_response(api: &WorkspaceApi) -> WorkerLaunchOptionsResp
runtimes,
profiles: worker_profile_candidates_for_root(&api.config.workspace_root),
repositories: working_directory_repository_options(api),
working_directories: working_directory_summaries(api).unwrap_or_default(),
working_directories: available_working_directory_summaries(api).unwrap_or_default(),
diagnostics: Vec::new(),
}
}
@@ -3966,6 +3968,32 @@ fn working_directory_summaries(api: &WorkspaceApi) -> ApiResult<Vec<WorkingDirec
.collect::<Vec<_>>())
}
fn available_working_directory_summaries(
api: &WorkspaceApi,
) -> ApiResult<Vec<WorkingDirectorySummary>> {
let limit = api.config.max_records.min(200);
for worker in api.runtime.list_workers(limit).items {
let _ = sync_worker_observation(api, &worker);
}
let records = working_directory_summaries(api)?;
let mut available = Vec::new();
for summary in records {
if summary.status != WorkingDirectoryStatusKind::Active
|| summary.cleanliness.as_deref() != Some("clean")
{
continue;
}
let links = api.store.list_workdir_worker_links(
&api.config.workspace_id,
summary.working_directory_id.as_str(),
)?;
if links.is_empty() && summary.primary_worker_id.is_none() {
available.push(summary);
}
}
Ok(available)
}
fn runtime_working_directory_summaries(
api: &WorkspaceApi,
runtime_id: &str,
@@ -4235,7 +4263,10 @@ fn sync_runtime_workdir_observations(
&status.summary,
management_kind.as_str(),
);
preserve_workdir_identity_for_corrupted_summary(&mut updated, Some(&record));
preserve_workdir_identity_for_corrupted_summary(
&mut updated,
Some(&record),
);
api.store.upsert_workdir_registry(&updated)?;
}
} else {
@@ -4400,6 +4431,7 @@ fn workdir_summary_from_record(record: &WorkdirRegistryRecord) -> WorkingDirecto
}),
status,
cleanliness: Some(record.cleanliness.clone()),
primary_worker_id: None,
management_kind: Some(record.management_kind.clone()),
}
}