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

feat: support multiple worker workdir attachments

Browse Source
This commit is contained in:
Hare
2026-09-19 23:27:57 +09:00
parent 308ae19f5e
commit 37df3ec545
52 changed files with 3715 additions and 1404 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+24 -9
View File
@@ -161,6 +161,8 @@ pub struct WorkingDirectoryRepositoryAccessRequest {
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryRequest {
pub repository: WorkingDirectoryRepository,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display_name: Option<String>,
#[serde(default)]
pub materializer: MaterializerKind,
/// Backend-assigned stable Workdir id. Runtimes use this when present so the
@@ -197,12 +199,25 @@ pub struct RepositoryRefObservation {
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryClaim {
pub struct WorkingDirectoryAttachmentClaim {
pub alias: workdir::WorkdirAttachmentAlias,
pub working_directory_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub relative_cwd: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryAttachmentRequest {
pub alias: workdir::WorkdirAttachmentAlias,
pub working_directory: WorkingDirectoryRequest,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryAttachmentStatus {
pub alias: workdir::WorkdirAttachmentAlias,
pub working_directory: WorkingDirectoryStatus,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryStatus {
pub summary: WorkingDirectorySummary,
@@ -247,10 +262,10 @@ pub struct CreateWorkerRequest {
pub config_bundle: Option<ConfigBundleRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub initial_input: Option<WorkerInput>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory_request: Option<WorkingDirectoryRequest>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectoryClaim>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
/// Backend-only feature enablement. Grants still define local Runtime peers;
/// the Workspace provider reauthorizes its dynamic set per operation.
#[serde(default, skip_serializing_if = "is_false")]
@@ -308,8 +323,8 @@ pub struct WorkerSummary {
pub worker_state: Option<protocol::WorkerStateSnapshot>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectoryStatus>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
pub profile: ProfileSelector,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display_name: Option<String>,
@@ -333,8 +348,8 @@ pub struct WorkerDetail {
pub worker_state: Option<protocol::WorkerStateSnapshot>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectoryStatus>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
pub profile: ProfileSelector,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display_name: Option<String>,
crates/worker-runtime/src/execution.rs
+8 -7
View File
@@ -12,6 +12,7 @@ use crate::observation::WorkerObservationEvent;
use crate::working_directory::{WorkingDirectoryBinding, WorkingDirectoryDiagnostic};
use protocol::{Method, UploadedFileRef};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::fmt;
use std::sync::Arc;
use workdir::WorkdirSessionHandle;
@@ -246,7 +247,7 @@ pub struct WorkerExecutionSpawnRequest {
pub request: crate::catalog::CreateWorkerRequest,
pub workspace_scope: Option<crate::runtime::RuntimeWorkspaceScope>,
pub context: WorkerExecutionContext,
pub working_directory: Option<WorkingDirectoryBinding>,
pub workdir_attachments: BTreeMap<workdir::WorkdirAttachmentAlias, WorkingDirectoryBinding>,
pub config_bundle: Option<ConfigBundle>,
}
@@ -257,8 +258,8 @@ pub struct WorkerExecutionRestoreRequest {
pub request: crate::catalog::CreateWorkerRequest,
pub workspace_scope: Option<crate::runtime::RuntimeWorkspaceScope>,
pub context: WorkerExecutionContext,
pub previous_working_directory: Option<WorkingDirectoryStatus>,
pub working_directory: Option<WorkingDirectoryBinding>,
pub previous_workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
pub workdir_attachments: BTreeMap<workdir::WorkdirAttachmentAlias, WorkingDirectoryBinding>,
pub config_bundle: Option<ConfigBundle>,
}
@@ -284,7 +285,7 @@ pub enum WorkerExecutionSpawnResult {
Connected {
handle: WorkerExecutionHandle,
worker_state: protocol::WorkerStateSnapshot,
working_directory: Option<WorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
},
/// A read-only preflight rejected the operation before live work started.
Rejected(WorkerExecutionResult),
@@ -297,7 +298,7 @@ pub enum WorkerExecutionSpawnResult {
result: WorkerExecutionResult,
handle: Option<WorkerExecutionHandle>,
worker_state: Option<protocol::WorkerStateSnapshot>,
working_directory: Option<WorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
},
/// Legacy spawn failure. Restore implementations must use `RolledBack` or
/// `ReconciliationRequired` after crossing their live side-effect boundary.
@@ -308,12 +309,12 @@ impl WorkerExecutionSpawnResult {
pub fn connected(
handle: WorkerExecutionHandle,
worker_state: protocol::WorkerStateSnapshot,
working_directory: Option<WorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
) -> Self {
Self::Connected {
handle,
worker_state,
working_directory,
workdir_attachments,
}
}
}
crates/worker-runtime/src/fs_store.rs
+161 -7
View File
@@ -1,6 +1,7 @@
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerRestoreIntent, WorkerStatus,
WorkingDirectoryStatus,
WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
WorkingDirectoryAttachmentStatus, WorkingDirectoryRequest,
};
use crate::config_bundle::ConfigBundle;
use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
@@ -442,7 +443,7 @@ pub(crate) struct PersistedWorkerRecord {
pub(crate) status: WorkerStatus,
pub(crate) execution_state: PersistedWorkerExecutionState,
pub(crate) workspace_id: Option<String>,
pub(crate) working_directory: Option<WorkingDirectoryStatus>,
pub(crate) workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
}
fn runtime_io_error(operation: &'static str, path: &Path, source: std::io::Error) -> RuntimeError {
@@ -1685,11 +1686,18 @@ struct WorkerIdentityRecord {
status: WorkerStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
working_directory: Option<WorkingDirectoryStatus>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
/// Schema-v8 compatibility input. New records serialize only `workdir_attachments`.
#[serde(
default,
rename = "working_directory",
skip_serializing_if = "Option::is_none"
)]
legacy_working_directory: Option<crate::catalog::WorkingDirectoryStatus>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
#[derive(Clone, Debug, Serialize)]
#[serde(deny_unknown_fields)]
struct WorkerExecutionRecord {
schema_version: u32,
@@ -1699,6 +1707,72 @@ struct WorkerExecutionRecord {
restore_intent: WorkerRestoreIntent,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct RawWorkerExecutionRecord {
schema_version: u32,
request: serde_json::Value,
#[serde(default)]
binding: Option<PersistedWorkerExecutionBinding>,
restore_intent: WorkerRestoreIntent,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct LegacyWorkingDirectoryClaim {
working_directory_id: String,
#[serde(default)]
relative_cwd: Option<String>,
}
impl<'de> Deserialize<'de> for WorkerExecutionRecord {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let raw = RawWorkerExecutionRecord::deserialize(deserializer)?;
let mut request_value = raw.request;
let request_object = request_value
.as_object_mut()
.ok_or_else(|| serde::de::Error::custom("Worker create request must be an object"))?;
let legacy_request = request_object.remove("working_directory_request");
let legacy_claim = request_object.remove("working_directory");
let mut request: CreateWorkerRequest =
serde_json::from_value(request_value).map_err(serde::de::Error::custom)?;
if request.workdir_attachment_requests.is_empty() && request.workdir_attachments.is_empty()
{
let alias = workdir::WorkdirAttachmentAlias::new("workdir")
.expect("static compatibility alias is valid");
if let Some(value) = legacy_request.filter(|value| !value.is_null()) {
let working_directory: WorkingDirectoryRequest =
serde_json::from_value(value).map_err(serde::de::Error::custom)?;
request
.workdir_attachment_requests
.push(WorkingDirectoryAttachmentRequest {
alias,
working_directory,
});
} else if let Some(value) = legacy_claim.filter(|value| !value.is_null()) {
let claim: LegacyWorkingDirectoryClaim =
serde_json::from_value(value).map_err(serde::de::Error::custom)?;
request
.workdir_attachments
.push(WorkingDirectoryAttachmentClaim {
alias,
working_directory_id: claim.working_directory_id,
relative_cwd: claim.relative_cwd,
});
}
}
Ok(Self {
schema_version: raw.schema_version,
request,
binding: raw.binding,
restore_intent: raw.restore_intent,
})
}
}
impl WorkerIdentityRecord {
fn from_persisted(worker: &PersistedWorkerRecord) -> Self {
Self {
@@ -1712,7 +1786,8 @@ impl WorkerIdentityRecord {
created_at_ms: worker.created_at_ms,
status: worker.status,
workspace_id: worker.workspace_id.clone(),
working_directory: worker.working_directory.clone(),
workdir_attachments: worker.workdir_attachments.clone(),
legacy_working_directory: None,
}
}
@@ -1760,6 +1835,19 @@ impl WorkerIdentityRecord {
self,
execution_state: PersistedWorkerExecutionState,
) -> PersistedWorkerRecord {
let workdir_attachments = if self.workdir_attachments.is_empty() {
self.legacy_working_directory
.map(|working_directory| {
vec![WorkingDirectoryAttachmentStatus {
alias: workdir::WorkdirAttachmentAlias::new("workdir")
.expect("static compatibility alias is valid"),
working_directory,
}]
})
.unwrap_or_default()
} else {
self.workdir_attachments
};
PersistedWorkerRecord {
worker_ref: self.worker_ref,
worker_id: self.worker_id,
@@ -1771,7 +1859,7 @@ impl WorkerIdentityRecord {
status: self.status,
execution_state,
workspace_id: self.workspace_id,
working_directory: self.working_directory,
workdir_attachments,
}
}
}
@@ -2215,6 +2303,72 @@ mod tests {
document
}
#[test]
fn schema_v8_singular_workdir_fields_normalize_to_one_attachment() {
let worker_id = WorkerId::now_v7();
let workdir_status = serde_json::json!({
"summary": {
"working_directory_id": "wd-legacy",
"repository_id": "repo",
"materializer_kind": "runtime_git_clone",
"status": "active"
}
});
let identity: WorkerIdentityRecord = serde_json::from_value(serde_json::json!({
"schema_version": SCHEMA_VERSION,
"worker_ref": { "worker_id": worker_id },
"worker_id": worker_id,
"profile": { "kind": "builtin", "value": "builtin:coder" },
"display_name": null,
"profile_source": {
"id": "archive-1",
"digest": "sha256:archive",
"size_bytes": 0,
"source_graph": {
"source_count": 0,
"total_source_bytes": 0,
"entrypoints": {},
"import_count": 0
}
},
"config_bundle": null,
"status": "stopped",
"working_directory": workdir_status
}))
.unwrap();
let persisted = identity.into_persisted(PersistedWorkerExecutionState::Unavailable);
assert_eq!(persisted.workdir_attachments.len(), 1);
assert_eq!(persisted.workdir_attachments[0].alias.as_str(), "workdir");
assert_eq!(
persisted.workdir_attachments[0]
.working_directory
.summary
.working_directory_id,
"wd-legacy"
);
let mut request = schema_v7_worker_document(worker_id)["request"].clone();
request["working_directory"] = serde_json::json!({
"working_directory_id": "wd-legacy",
"relative_cwd": "crates/yoi"
});
let execution: WorkerExecutionRecord = serde_json::from_value(serde_json::json!({
"schema_version": SCHEMA_VERSION,
"request": request,
"binding": null,
"restore_intent": "explicit"
}))
.unwrap();
assert!(execution.request.workdir_attachment_requests.is_empty());
assert_eq!(execution.request.workdir_attachments.len(), 1);
let attachment = &execution.request.workdir_attachments[0];
assert_eq!(attachment.alias.as_str(), "workdir");
assert_eq!(attachment.working_directory_id, "wd-legacy");
assert_eq!(attachment.relative_cwd.as_deref(), Some("crates/yoi"));
let serialized = serde_json::to_value(execution).unwrap();
assert!(serialized["request"].get("working_directory").is_none());
}
#[test]
fn startup_recovers_staging_created_before_source_backup() {
let temp = tempfile::tempdir().unwrap();
crates/worker-runtime/src/http_server.rs
+30 -14
View File
@@ -2282,7 +2282,11 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
if path.ends_with("/workspace-api") {
return Some("workers:create");
}
if path.ends_with("/input") || path.ends_with("/restore") || path.contains("/attachments") {
if path.ends_with("/input")
|| path.ends_with("/restore")
|| path.contains("/attachments")
|| path.contains("/workdir-attachments")
{
return Some("workers:input");
}
if path.ends_with("/stop") || path.ends_with("/cancel") {
@@ -3072,8 +3076,8 @@ mod tests {
digest: bundle.metadata.digest,
}),
initial_input: None,
working_directory_request: None,
working_directory: None,
workdir_attachment_requests: Vec::new(),
workdir_attachments: Vec::new(),
worker_observation_enabled: false,
worker_observation_grants: Vec::new(),
workspace_api: None,
@@ -3319,10 +3323,16 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
},
)
.collect(),
}
}
@@ -3335,7 +3345,7 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request.previous_working_directory,
workdir_attachments: request.previous_workdir_attachments,
}
}
@@ -3907,10 +3917,16 @@ mod ws_tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
},
)
.collect(),
}
}
@@ -4020,8 +4036,8 @@ mod ws_tests {
digest: bundle.metadata.digest,
}),
initial_input: None,
working_directory_request: None,
working_directory: None,
workdir_attachment_requests: Vec::new(),
workdir_attachments: Vec::new(),
worker_observation_enabled: false,
worker_observation_grants: Vec::new(),
workspace_api: None,
crates/worker-runtime/src/http_server/runtime_management_api.rs
+25
View File
@@ -423,6 +423,31 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
Ok(runtime_api::WorkerResponse { worker })
}
async fn replace_worker_workdir_attachments(
&self,
worker_id: String,
value: runtime_api::WorkerWorkdirAttachmentsRequest,
) -> Result<runtime_api::WorkerResponse, runtime_api::RuntimeApiError> {
let attachments = request(value.workdir_attachments)?;
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_workdir_attachments_scoped(&scope, &worker_ref, attachments),
None => self
.state
.runtime
.replace_worker_workdir_attachments(&worker_ref, attachments),
}
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)?;
let worker = response(worker)?;
Ok(runtime_api::WorkerResponse { worker })
}
async fn complete_worker_arguments(
&self,
worker_id: String,
crates/worker-runtime/src/runtime.rs
+412 -147
View File
@@ -1,7 +1,8 @@
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource,
RepositoryRefObservation, RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck,
WorkerRestoreIntent, WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest,
WorkerRestoreIntent, WorkerStatus, WorkerSummary, WorkingDirectoryAttachmentClaim,
WorkingDirectoryAttachmentStatus, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
WorkspaceApiRef,
};
@@ -43,7 +44,7 @@ use crate::retention::{
use protocol::subscription::{
EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot,
SubscriptionValidationError, SubscriptionWorkdirId, SubscriptionWorker, SubscriptionWorkerId,
SubscriptionWorkerState,
SubscriptionWorkerState, SubscriptionWorkerWorkdirAttachment,
};
use protocol::{Event, Method};
use server_api::WorkerRestoreState;
@@ -729,8 +730,9 @@ impl Runtime {
.get(&worker_ref.worker_id)
.is_some_and(|worker| {
worker.belongs_to_workspace(&scope.workspace_id)
&& worker.working_directory.as_ref().is_some_and(|status| {
status.summary.working_directory_id == working_directory_id
&& worker.workdir_attachments.iter().any(|attachment| {
attachment.working_directory.summary.working_directory_id
== working_directory_id
})
});
if !owns_workdir {
@@ -764,7 +766,7 @@ 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) {
if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is assigned to worker {worker_id}"
)));
@@ -792,12 +794,8 @@ impl Runtime {
fn annotate_working_directory_status(
&self,
mut status: CatalogWorkingDirectoryStatus,
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())
.map(|worker_id| worker_id.to_string());
Ok(status)
}
@@ -1023,10 +1021,8 @@ impl Runtime {
return Ok(existing.detail());
}
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)
{
for working_directory_id in requested_workdir_ids(&request) {
if let Some(owner_worker_id) = state.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}"
)));
@@ -1064,7 +1060,7 @@ impl Runtime {
execution_metadata_available: true,
execution_bound: true,
restore_intent: WorkerRestoreIntent::Explicit,
working_directory: None,
workdir_attachments: Vec::new(),
execution_handle: None,
internal_workers: InternalWorkerActivityProjection::default(),
};
@@ -1076,19 +1072,19 @@ impl Runtime {
request,
workspace_scope: scope.cloned(),
context: self.execution_context(worker_ref.clone()),
working_directory: None,
workdir_attachments: BTreeMap::new(),
config_bundle,
};
(backend, worker_ref, spawn_request)
};
let spawn_result = backend.spawn_worker(spawn_request);
let (handle, initial_worker_state, working_directory) = match spawn_result {
let (handle, initial_worker_state, workdir_attachments) = match spawn_result {
WorkerExecutionSpawnResult::Connected {
handle,
worker_state,
working_directory,
} => (handle, worker_state, working_directory),
workdir_attachments,
} => (handle, worker_state, workdir_attachments),
WorkerExecutionSpawnResult::Rejected(result)
| WorkerExecutionSpawnResult::RolledBack(result)
| WorkerExecutionSpawnResult::Errored(result) => {
@@ -1101,8 +1097,30 @@ impl Runtime {
result,
});
}
WorkerExecutionSpawnResult::ReconciliationRequired { result, .. } => {
self.rollback_failed_create(&worker_ref)?;
WorkerExecutionSpawnResult::ReconciliationRequired {
result,
handle,
worker_state,
workdir_attachments,
} => {
let retain_result =
if let (Some(handle), Some(worker_state)) = (handle, worker_state) {
self.retain_restore_execution_evidence(
&worker_ref,
handle,
worker_state,
workdir_attachments,
)
} else {
self.retain_create_reconciliation_pending(&worker_ref, workdir_attachments)
};
if let Err(error) = retain_result {
tracing::error!(
worker_id = %worker_ref.worker_id,
%error,
"failed to persist Worker create reconciliation evidence"
);
}
return Err(RuntimeError::WorkerExecutionRejected {
worker_id: worker_ref.worker_id.clone(),
operation: result.operation,
@@ -1160,7 +1178,7 @@ impl Runtime {
&worker_ref,
handle.clone(),
initial_worker_state.clone(),
working_directory,
workdir_attachments,
dispatch_result,
) {
Ok(detail) => detail,
@@ -1179,7 +1197,7 @@ impl Runtime {
&worker_ref,
handle.clone(),
initial_worker_state,
working_directory,
workdir_attachments,
WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn),
) {
Ok(detail) => Ok(detail),
@@ -1425,6 +1443,119 @@ impl Runtime {
Ok(state.worker(worker_ref)?.detail())
}
pub fn replace_worker_workdir_attachments_scoped(
&self,
scope: &RuntimeWorkspaceScope,
worker_ref: &WorkerRef,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> Result<WorkerDetail, RuntimeError> {
self.ensure_worker_in_workspace(scope, worker_ref)?;
self.replace_worker_workdir_attachments(worker_ref, attachments)
}
/// Persist the exact current attachment set used by live mutation and future restore.
pub fn replace_worker_workdir_attachments(
&self,
worker_ref: &WorkerRef,
mut attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> Result<WorkerDetail, RuntimeError> {
let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?;
let _operation_guard = operation_lock
.lock()
.map_err(|_| RuntimeError::StatePoisoned)?;
let backend = {
let state = self.lock()?;
let worker = state.worker(worker_ref)?;
if worker.execution_handle.is_none() || !worker.status.is_active() {
return Err(RuntimeError::WorkerExecutionUnavailable {
worker_id: worker.worker_id,
message: "Workdir attachments can only be changed for a live Worker"
.to_string(),
});
}
let request = worker.request.as_ref().ok_or_else(|| {
RuntimeError::WorkerExecutionUnavailable {
worker_id: worker.worker_id,
message: "persisted Worker restore request is unavailable".to_string(),
}
})?;
for attachment in &mut attachments {
if attachment.relative_cwd.is_none()
&& let Some(previous) = request.workdir_attachments.iter().find(|previous| {
previous.alias == attachment.alias
&& previous.working_directory_id == attachment.working_directory_id
})
{
attachment.relative_cwd = previous.relative_cwd.clone();
}
}
let mut candidate = request.clone();
candidate.workdir_attachment_requests.clear();
candidate.workdir_attachments = attachments.clone();
validate_create_worker_request(&candidate)?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "Workdir attachment mutation requires an execution backend"
.to_string(),
}
})?
};
let statuses = attachments
.iter()
.map(|attachment| {
backend
.working_directory(&attachment.working_directory_id)
.map(|working_directory| WorkingDirectoryAttachmentStatus {
alias: attachment.alias.clone(),
working_directory,
})
.map_err(RuntimeError::from)
})
.collect::<Result<Vec<_>, RuntimeError>>()?;
if let Some(inactive) = statuses.iter().find(|attachment| {
attachment.working_directory.summary.status
!= workdir::workspace::WorkingDirectoryStatusKind::Active
}) {
return Err(RuntimeError::InvalidRequest(format!(
"working directory {} is not active ({})",
inactive.working_directory.summary.working_directory_id,
inactive.working_directory.summary.status
)));
}
let mut state = self.lock()?;
for attachment in &attachments {
if let Some(owner_worker_id) =
state.worker_id_for_workdir(&attachment.working_directory_id)
&& owner_worker_id != worker_ref.worker_id
{
return Err(RuntimeError::InvalidRequest(format!(
"working directory {} is already assigned to worker {owner_worker_id}",
attachment.working_directory_id
)));
}
}
let previous = state.worker(worker_ref)?.clone();
{
let worker = state.worker_mut(worker_ref)?;
let request = worker
.request
.as_mut()
.expect("restore request checked above");
request.workdir_attachment_requests.clear();
request.workdir_attachments = attachments;
worker.workdir_attachments = statuses;
}
if let Err(error) = state.persist_worker(&worker_ref.worker_id) {
state.workers.insert(worker_ref.worker_id, previous);
return Err(error);
}
state.publish_worker_upsert(worker_ref.worker_id)?;
Ok(state.worker(worker_ref)?.detail())
}
/// Observe a Worker's live protocol availability or its retained Session.
/// The same lifecycle lock used by stop/restore fences the decision and read.
pub fn worker_session_scoped(
@@ -1570,8 +1701,8 @@ impl Runtime {
request: worker_request,
workspace_scope,
context: self.execution_context(worker_ref.clone()),
previous_working_directory: worker.working_directory.clone(),
working_directory: None,
previous_workdir_attachments: worker.workdir_attachments.clone(),
workdir_attachments: BTreeMap::new(),
config_bundle: None,
};
(backend, request)
@@ -1589,14 +1720,14 @@ impl Runtime {
WorkerExecutionSpawnResult::Connected {
handle,
worker_state,
working_directory,
workdir_attachments,
} => {
let commit = self.commit_restored_worker_execution(
worker_ref,
handle.clone(),
worker_state.clone(),
WorkerStatus::Idle,
working_directory.clone(),
workdir_attachments.clone(),
);
match commit {
Ok(worker) => Ok(RuntimeWorkerRestoreResult::accepted(worker)),
@@ -1625,7 +1756,7 @@ impl Runtime {
worker_ref,
handle,
worker_state,
working_directory,
workdir_attachments,
) {
tracing::error!(
worker_id = %worker_ref.worker_id,
@@ -1663,14 +1794,14 @@ impl Runtime {
result: _result,
handle,
worker_state,
working_directory,
workdir_attachments,
} => {
if let (Some(handle), Some(worker_state)) = (handle, worker_state) {
if let Err(retain_error) = self.retain_restore_execution_evidence(
worker_ref,
handle,
worker_state,
working_directory,
workdir_attachments,
) {
tracing::error!(
worker_id = %worker_ref.worker_id,
@@ -2079,7 +2210,7 @@ impl Runtime {
worker_ref: &WorkerRef,
handle: WorkerExecutionHandle,
initial_worker_state: protocol::WorkerStateSnapshot,
working_directory: Option<CatalogWorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
result: WorkerExecutionResult,
) -> Result<WorkerDetail, RuntimeError> {
let mut state = self.lock()?;
@@ -2094,7 +2225,7 @@ impl Runtime {
let _ = worker.apply_worker_state(snapshot);
}
worker.restore_intent = restore_intent_for_status(worker.status);
worker.working_directory = working_directory;
worker.workdir_attachments = workdir_attachments;
worker.detail()
};
state.publish_worker_upsert(worker_ref.worker_id)?;
@@ -2367,9 +2498,13 @@ impl Runtime {
state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if worker.execution_handle.is_some() && worker.status.is_active() {
if worker.status.is_active()
&& (worker.execution_handle.is_some()
|| (!worker.execution_metadata_available
&& !worker.workdir_attachments.is_empty()))
{
return Err(RuntimeError::InvalidRequest(format!(
"worker {} is running and must be stopped before deletion",
"worker {} is active or requires reconciliation and must be stopped before deletion",
worker_ref.worker_id
)));
}
@@ -2398,9 +2533,12 @@ impl Runtime {
state.ensure_running()?;
state.ensure_worker_ref(worker_ref)?;
let worker = state.worker(worker_ref)?;
if worker.execution_handle.is_some() && worker.status.is_active() {
if worker.status.is_active()
&& (worker.execution_handle.is_some()
|| (!worker.execution_metadata_available && !worker.workdir_attachments.is_empty()))
{
return Err(RuntimeError::InvalidRequest(format!(
"worker {} became active before deletion",
"worker {} became active or reconciliation-required before deletion",
worker_ref.worker_id
)));
}
@@ -2662,7 +2800,7 @@ impl Runtime {
handle: WorkerExecutionHandle,
worker_state: protocol::WorkerStateSnapshot,
status: WorkerStatus,
working_directory: Option<CatalogWorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
) -> Result<WorkerDetail, RuntimeError> {
let mut state = self.lock()?;
state.ensure_worker_ref(worker_ref)?;
@@ -2679,7 +2817,7 @@ impl Runtime {
candidate.status = status;
let _ = candidate.apply_worker_state(&worker_state);
candidate.restore_intent = restore_intent_for_status(candidate.status);
candidate.working_directory = working_directory;
candidate.workdir_attachments = workdir_attachments;
// Validate the exact subscription projection before persistence. The
// subsequent publish uses the same pure projection while the Worker
@@ -2708,7 +2846,7 @@ impl Runtime {
worker_ref: &WorkerRef,
handle: WorkerExecutionHandle,
worker_state: protocol::WorkerStateSnapshot,
working_directory: Option<CatalogWorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let mut candidate = state.worker(worker_ref)?.clone();
@@ -2718,7 +2856,26 @@ impl Runtime {
candidate.status = WorkerStatus::Idle;
let _ = candidate.apply_worker_state(&worker_state);
candidate.restore_intent = WorkerRestoreIntent::Automatic;
candidate.working_directory = working_directory;
candidate.workdir_attachments = workdir_attachments;
state.workers.insert(worker_ref.worker_id, candidate);
state.publish_worker_upsert(worker_ref.worker_id)?;
state.persist_worker(&worker_ref.worker_id)?;
Ok(())
}
fn retain_create_reconciliation_pending(
&self,
worker_ref: &WorkerRef,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
) -> Result<(), RuntimeError> {
let mut state = self.lock()?;
let mut candidate = state.worker(worker_ref)?.clone();
candidate.execution_handle = None;
candidate.execution_metadata_available = false;
candidate.execution_bound = true;
candidate.status = WorkerStatus::Idle;
candidate.restore_intent = WorkerRestoreIntent::Explicit;
candidate.workdir_attachments = workdir_attachments;
state.workers.insert(worker_ref.worker_id, candidate);
state.publish_worker_upsert(worker_ref.worker_id)?;
state.persist_worker(&worker_ref.worker_id)?;
@@ -3078,7 +3235,7 @@ impl RuntimeState {
execution_metadata_available,
execution_bound,
restore_intent,
working_directory: worker.working_directory,
workdir_attachments: worker.workdir_attachments,
execution_handle: None,
internal_workers: InternalWorkerActivityProjection::default(),
},
@@ -3394,18 +3551,25 @@ impl RuntimeState {
) -> Result<SubscriptionWorker, RuntimeError> {
let worker_id = SubscriptionWorkerId::new(worker.worker_id.to_string())
.map_err(subscription_validation_error)?;
let repository_id = worker
.working_directory
.as_ref()
.map(|working_directory| working_directory.summary.repository_id.clone());
let working_directory_id = worker
.working_directory
.as_ref()
.map(|working_directory| {
SubscriptionWorkdirId::new(working_directory.summary.working_directory_id.clone())
.map_err(subscription_validation_error)
let workdir_attachments = worker
.workdir_attachments
.iter()
.map(|attachment| {
Ok(SubscriptionWorkerWorkdirAttachment {
alias: attachment.alias.to_string(),
repository_id: Some(attachment.working_directory.summary.repository_id.clone()),
repository_key: None,
working_directory_id: SubscriptionWorkdirId::new(
attachment
.working_directory
.summary
.working_directory_id
.clone(),
)
.map_err(subscription_validation_error)?,
})
})
.transpose()?;
.collect::<Result<Vec<_>, RuntimeError>>()?;
let profile = match &worker.profile {
ProfileSelector::Builtin(name) | ProfileSelector::Named(name) => Some(name.clone()),
};
@@ -3425,9 +3589,7 @@ impl RuntimeState {
workspace_id: worker.workspace_id.clone(),
display_name: worker.display_name.clone(),
profile,
repository_id,
repository_key: None,
working_directory_id,
workdir_attachments,
})
}
@@ -3525,22 +3687,15 @@ impl RuntimeState {
}
}
fn primary_worker_id_for_workdir(&self, working_directory_id: &str) -> Option<WorkerId> {
fn worker_id_for_workdir(&self, working_directory_id: &str) -> Option<WorkerId> {
self.workers.values().find_map(|worker| {
if worker
.working_directory
.as_ref()
.is_some_and(|binding| binding.summary.working_directory_id == working_directory_id)
|| worker
.request
.as_ref()
.and_then(requested_primary_workdir_id)
== Some(working_directory_id)
{
Some(worker.worker_id)
} else {
None
}
let has_status = worker.workdir_attachments.iter().any(|attachment| {
attachment.working_directory.summary.working_directory_id == working_directory_id
});
let has_request = worker.request.as_ref().is_some_and(|request| {
requested_workdir_ids(request).contains(&working_directory_id)
});
(has_status || has_request).then_some(worker.worker_id)
})
}
@@ -3976,7 +4131,7 @@ struct WorkerRecord {
execution_metadata_available: bool,
execution_bound: bool,
restore_intent: WorkerRestoreIntent,
working_directory: Option<CatalogWorkingDirectoryStatus>,
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
execution_handle: Option<WorkerExecutionHandle>,
internal_workers: InternalWorkerActivityProjection,
}
@@ -3999,7 +4154,7 @@ impl WorkerRecord {
execution_metadata_available: self.execution_metadata_available,
worker_state: self.worker_state.clone(),
workspace_id: self.workspace_id.clone(),
working_directory: self.working_directory.clone(),
workdir_attachments: self.workdir_attachments.clone(),
profile: self.profile.clone(),
display_name: self.display_name.clone(),
profile_source: self.profile_source.clone(),
@@ -4016,7 +4171,7 @@ impl WorkerRecord {
execution_metadata_available: self.execution_metadata_available,
worker_state: self.worker_state.clone(),
workspace_id: self.workspace_id.clone(),
working_directory: self.working_directory.clone(),
workdir_attachments: self.workdir_attachments.clone(),
profile: self.profile.clone(),
display_name: self.display_name.clone(),
profile_source: self.profile_source.clone(),
@@ -4049,7 +4204,7 @@ impl WorkerRecord {
status: self.status,
execution_state,
workspace_id: self.workspace_id.clone(),
working_directory: self.working_directory.clone(),
workdir_attachments: self.workdir_attachments.clone(),
}
}
}
@@ -4100,25 +4255,28 @@ fn repository_resource_error(error: BackendResourceError) -> RuntimeError {
fn durable_create_worker_request(request: &CreateWorkerRequest) -> CreateWorkerRequest {
let mut durable = request.clone();
if let Some(working_directory) = durable.working_directory_request.as_mut()
&& let Some(materialization) = working_directory.materialization.as_mut()
{
materialization.ssh = None;
for attachment in &mut durable.workdir_attachment_requests {
if let Some(materialization) = attachment.working_directory.materialization.as_mut() {
materialization.ssh = None;
}
}
durable
}
fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> {
fn requested_workdir_ids(request: &CreateWorkerRequest) -> Vec<&str> {
request
.working_directory
.as_ref()
.workdir_attachments
.iter()
.map(|claim| claim.working_directory_id.as_str())
.or_else(|| {
.chain(
request
.working_directory_request
.as_ref()
.and_then(|request| request.backend_workdir_id.as_deref())
})
.workdir_attachment_requests
.iter()
.filter_map(|attachment| {
attachment.working_directory.backend_workdir_id.as_deref()
}),
)
.collect()
}
fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> {
@@ -4127,6 +4285,42 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R
"create_fingerprint must not be empty".to_string(),
));
}
let mut aliases = std::collections::BTreeSet::new();
let mut workdir_ids = std::collections::BTreeSet::new();
for (alias, workdir_id) in request
.workdir_attachments
.iter()
.map(|attachment| {
(
&attachment.alias,
Some(attachment.working_directory_id.as_str()),
)
})
.chain(
request
.workdir_attachment_requests
.iter()
.map(|attachment| {
(
&attachment.alias,
attachment.working_directory.backend_workdir_id.as_deref(),
)
}),
)
{
if !aliases.insert(alias.as_str()) {
return Err(RuntimeError::InvalidRequest(format!(
"duplicate Workdir attachment alias `{alias}`"
)));
}
if let Some(workdir_id) = workdir_id
&& !workdir_ids.insert(workdir_id)
{
return Err(RuntimeError::InvalidRequest(format!(
"working directory `{workdir_id}` cannot be attached more than once"
)));
}
}
match &request.profile_source {
crate::catalog::ProfileSourceArchiveSource::Embedded { archive } => {
archive.verify().map_err(|err| {
@@ -4420,8 +4614,8 @@ mod tests {
use crate::catalog::{
ConfigBundleRef, MaterializerKind, ProfileSelector, RepositoryMaterializationContext,
RepositorySshCredentialCandidate, RepositorySshMaterializationAccess, SensitiveString,
WorkingDirectoryClaim, WorkingDirectoryRepository, WorkingDirectoryRequest,
WorkspaceApiRef,
WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
};
use crate::config_bundle::{
ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
@@ -4913,8 +5107,8 @@ mod tests {
digest: bundle.metadata.digest,
}),
initial_input: None,
working_directory_request: None,
working_directory: None,
workdir_attachment_requests: Vec::new(),
workdir_attachments: Vec::new(),
worker_observation_enabled: false,
worker_observation_grants: Vec::new(),
workspace_api: None,
@@ -4929,60 +5123,64 @@ mod tests {
#[test]
fn durable_worker_request_omits_repository_credentials() {
let mut request = task_request("worker-secret-redaction");
request.working_directory_request = Some(WorkingDirectoryRequest {
repository: WorkingDirectoryRepository {
id: "repository-1".to_string(),
provider: "git".to_string(),
source: server_api::RepositorySource {
kind: server_api::RepositorySourceKind::Ssh,
uri: "ssh://git@example.test/repo.git".to_string(),
request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory: WorkingDirectoryRequest {
repository: WorkingDirectoryRepository {
id: "repository-1".to_string(),
provider: "git".to_string(),
source: server_api::RepositorySource {
kind: server_api::RepositorySourceKind::Ssh,
uri: "ssh://git@example.test/repo.git".to_string(),
},
source_revision: 1,
source_fingerprint: "sha256:source".to_string(),
selector: None,
},
source_revision: 1,
source_fingerprint: "sha256:source".to_string(),
selector: None,
},
materializer: MaterializerKind::RuntimeGitClone,
backend_workdir_id: Some("working-directory-1".to_string()),
materialization: Some(RepositoryMaterializationContext {
workspace_id: "workspace-1".to_string(),
runtime_id: "runtime-1".to_string(),
operation_id: "operation-1".to_string(),
config_revision: 1,
config_projection_digest: "sha256:projection".to_string(),
ssh: Some(RepositorySshMaterializationAccess {
credential_candidates: vec![RepositorySshCredentialCandidate {
credential_id: "credential-1".to_string(),
credential_revision: 1,
private_key: SensitiveString::new("private-key-bytes"),
}],
host_trust_id: "host-trust-1".to_string(),
host_trust_revision: 1,
access: server_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repository-1".to_string(),
repository_source_fingerprint: "sha256:source".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
known_hosts_entry: SensitiveString::new("known-hosts-entry"),
display_name: None,
materializer: MaterializerKind::RuntimeGitClone,
backend_workdir_id: Some("working-directory-1".to_string()),
materialization: Some(RepositoryMaterializationContext {
workspace_id: "workspace-1".to_string(),
runtime_id: "runtime-1".to_string(),
operation_id: "operation-1".to_string(),
config_revision: 1,
config_projection_digest: "sha256:projection".to_string(),
ssh: Some(RepositorySshMaterializationAccess {
credential_candidates: vec![RepositorySshCredentialCandidate {
credential_id: "credential-1".to_string(),
credential_revision: 1,
private_key: SensitiveString::new("private-key-bytes"),
}],
host_trust_id: "host-trust-1".to_string(),
host_trust_revision: 1,
access: server_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repository-1".to_string(),
repository_source_fingerprint: "sha256:source".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
known_hosts_entry: SensitiveString::new("known-hosts-entry"),
}),
}),
}),
});
},
}];
let durable = durable_create_worker_request(&request);
assert!(
request
.working_directory_request
.as_ref()
.and_then(|working_directory| working_directory.materialization.as_ref())
.workdir_attachment_requests
.first()
.and_then(|attachment| attachment.working_directory.materialization.as_ref())
.and_then(|materialization| materialization.ssh.as_ref())
.is_some()
);
assert!(
durable
.working_directory_request
.as_ref()
.and_then(|working_directory| working_directory.materialization.as_ref())
.workdir_attachment_requests
.first()
.and_then(|attachment| attachment.working_directory.materialization.as_ref())
.and_then(|materialization| materialization.ssh.as_ref())
.is_none()
);
@@ -5172,6 +5370,7 @@ mod tests {
source_fingerprint: "sha256:source".to_string(),
selector: None,
},
display_name: None,
materializer: MaterializerKind::RuntimeGitClone,
backend_workdir_id: Some("working-directory-1".to_string()),
materialization: Some(RepositoryMaterializationContext {
@@ -5342,6 +5541,7 @@ mod tests {
#[derive(Default)]
struct TestExecutionBackend {
spawn_result: Mutex<Option<WorkerExecutionSpawnResult>>,
dispatch_result: Mutex<Option<WorkerExecutionResult>>,
stop_result: Mutex<Option<WorkerExecutionResult>>,
stop_gate: Mutex<Option<Arc<RestoreGate>>>,
@@ -5447,6 +5647,9 @@ mod tests {
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
if let Some(result) = self.spawn_result.lock().unwrap().take() {
return result;
}
self.config_bundles
.lock()
.unwrap()
@@ -5460,10 +5663,16 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
},
)
.collect(),
}
}
@@ -5502,10 +5711,16 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
},
)
.collect(),
}
}
@@ -6263,10 +6478,11 @@ mod tests {
let runtime = runtime_with_backend();
let mut request = task_request("idempotent");
request.create_fingerprint = "sha256:input-1".to_string();
request.working_directory = Some(WorkingDirectoryClaim {
request.workdir_attachments = vec![WorkingDirectoryAttachmentClaim {
alias: workdir::WorkdirAttachmentAlias::new("workdir").unwrap(),
working_directory_id: "workdir-idempotent".to_string(),
relative_cwd: None,
});
}];
let first = runtime.create_worker(request.clone()).unwrap();
let workdir_count_after_first = runtime.list_working_directories().unwrap().len();
@@ -6631,6 +6847,49 @@ mod tests {
assert!(runtime.list_workers().unwrap().is_empty());
}
#[test]
fn reconciliation_required_create_retains_durable_worker_evidence() {
let (runtime, backend) = runtime_and_backend();
runtime.store_config_bundle(test_bundle()).unwrap();
*backend.spawn_result.lock().unwrap() =
Some(WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Spawn,
"Workdir cleanup could not be proven",
),
handle: None,
worker_state: None,
workdir_attachments: vec![WorkingDirectoryAttachmentStatus {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory: serde_json::from_value(serde_json::json!({
"summary": {
"working_directory_id": "wd-uncertain",
"repository_id": "repo",
"materializer_kind": "runtime_git_clone",
"status": "active"
}
}))
.unwrap(),
}],
});
let request = task_request("uncertain create cleanup");
let worker_ref = WorkerRef::new(request.worker_id);
let error = runtime.create_worker(request).unwrap_err();
assert!(matches!(
error,
RuntimeError::WorkerExecutionRejected { .. }
));
let retained = runtime.worker_detail(&worker_ref).unwrap();
assert_eq!(retained.status, WorkerStatus::Idle);
assert!(!retained.execution_metadata_available);
assert!(matches!(
runtime.delete_worker(&worker_ref),
Err(RuntimeError::InvalidRequest(_))
));
}
#[test]
fn create_worker_without_execution_backend_is_rejected_and_not_persisted() {
let runtime = Runtime::new_memory();
@@ -6804,10 +7063,16 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
},
)
.collect(),
}
}
@@ -7667,7 +7932,7 @@ mod tests {
),
handle: None,
worker_state: None,
working_directory: None,
workdir_attachments: Vec::new(),
},
);
crates/worker-runtime/src/worker_backend.rs
+470 -195
View File
@@ -7,7 +7,7 @@
//! socket paths, session paths, manifests, credentials, or handles leave this
//! module.
use std::collections::HashMap;
use std::collections::{BTreeMap, HashMap};
use std::future::Future;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
@@ -17,8 +17,9 @@ use std::time::Duration;
use crate::auth::{BACKEND_RESOURCE_FETCH_PERMISSION, RuntimeIdentityMaterial};
use crate::catalog::{
CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation,
RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectoryStatusKind,
RepositoryRefObservationRequest, WorkingDirectoryAttachmentStatus,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectoryStatusKind,
};
use crate::config_bundle::{ConfigBundle, workspace_config_etag};
use crate::execution::{
@@ -44,6 +45,45 @@ use protocol::{Event, Method, Segment, WorkerCommandEnvelope};
static NEXT_INTERNAL_COMMAND_ID: AtomicU64 = AtomicU64::new(1);
fn rollback_materialized_spawn_workdirs(
materializer: &dyn WorkingDirectoryMaterializer,
bindings: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
workdir_ids: &[String],
failure_context: &str,
) -> Result<(), WorkerExecutionSpawnResult> {
let mut cleanup_failures = Vec::new();
let mut uncertain_attachments = Vec::new();
for workdir_id in workdir_ids.iter().rev() {
if let Err(error) = materializer.cleanup_working_directory(workdir_id) {
cleanup_failures.push(format!("{workdir_id}: {error}"));
if let Some((alias, binding)) = bindings
.iter()
.find(|(_, binding)| binding.working_directory.id == *workdir_id)
{
uncertain_attachments.push(WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
});
}
}
}
if cleanup_failures.is_empty() {
return Ok(());
}
Err(WorkerExecutionSpawnResult::ReconciliationRequired {
result: WorkerExecutionResult::errored(
WorkerExecutionOperation::Spawn,
format!(
"{failure_context}; Workdir cleanup could not be proven: {}",
cleanup_failures.join("; ")
),
),
handle: None,
worker_state: None,
workdir_attachments: uncertain_attachments,
})
}
fn next_internal_command(
state: &RwLock<protocol::WorkerStateSnapshot>,
) -> Result<WorkerCommandEnvelope, String> {
@@ -68,7 +108,10 @@ use session_store::{FsStore, FsWorkerStore};
use tokio::runtime::Runtime;
#[cfg(feature = "ws-server")]
use tokio::sync::broadcast;
use workdir::{LocalWorkdirSession, Workdir, WorkdirSessionCapabilities, WorkdirSessionHandle};
use workdir::{
LocalWorkdirSession, Workdir, WorkdirAttachmentAlias, WorkdirSessionCapabilities,
WorkdirSessionHandle, WorkdirSessionRouter,
};
#[cfg(test)]
use worker::WorkerController;
@@ -730,6 +773,29 @@ fn runtime_local_workdir_session(
))
}
fn runtime_local_workdir_router(
attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
scope: manifest::SharedScope,
) -> Result<Arc<WorkdirSessionRouter>, String> {
let router = Arc::new(WorkdirSessionRouter::new());
for (alias, binding) in attachments {
router
.attach(
alias.clone(),
runtime_local_workdir_session(
&binding.working_directory.id,
binding.root(),
binding.cwd(),
scope.clone(),
binding.command_environment(),
binding.session_resources(),
),
)
.map_err(|error| format!("bind Workdir attachment `{alias}`: {error}"))?;
}
Ok(router)
}
fn bind_workspace_memory_settings(
manifest: &mut manifest::WorkerManifest,
request: &CreateWorkerRequest,
@@ -845,15 +911,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
) -> Result<RuntimeWorkerController, String> {
let worker_name = Self::runtime_worker_name(&request);
let profile = Self::runtime_profile(&request);
let has_local_filesystem = request.working_directory.is_some();
let worker_root = request
.working_directory
.as_ref()
let has_local_filesystem = request.workdir_attachments.len() == 1;
let only_binding = (request.workdir_attachments.len() == 1)
.then(|| request.workdir_attachments.values().next())
.flatten();
let worker_root = only_binding
.map(|binding| binding.root().to_path_buf())
.unwrap_or_else(|| self.profile_base_dir.clone());
let filesystem_authority = request
.working_directory
.as_ref()
let filesystem_authority = only_binding
.map(|binding| {
WorkerFilesystemAuthority::local(
binding.root().to_path_buf(),
@@ -965,18 +1030,10 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
})?;
let worker = prepared.worker_mut();
validate_worker_memory_settings(worker.manifest(), &request.request)?;
if let Some(binding) = request.working_directory.as_ref() {
worker.bind_workdir_session(Some(runtime_local_workdir_session(
&binding.working_directory.id,
binding.root(),
binding.cwd(),
worker.scope().clone(),
binding.command_environment(),
binding.session_resources(),
)));
} else {
worker.bind_workdir_session(None);
}
worker.bind_workdir_sessions(runtime_local_workdir_router(
&request.workdir_attachments,
worker.scope().clone(),
)?);
if let (Some(runtime_id), Some(workspace_id)) =
(observation_runtime_id, observation_workspace_id.clone())
&& observation_enabled
@@ -1095,9 +1152,10 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
request: WorkerExecutionRestoreRequest,
) -> Result<RuntimeWorkerController, String> {
let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref);
let filesystem_authority = request
.working_directory
.as_ref()
let only_binding = (request.workdir_attachments.len() == 1)
.then(|| request.workdir_attachments.values().next())
.flatten();
let filesystem_authority = only_binding
.map(|binding| {
WorkerFilesystemAuthority::local(
binding.root().to_path_buf(),
@@ -1201,18 +1259,10 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
};
validate_worker_memory_settings(worker.manifest(), &request.request)?;
let flow_transition_enabled = worker.manifest().feature.flow.enabled;
if let Some(binding) = request.working_directory.as_ref() {
worker.bind_workdir_session(Some(runtime_local_workdir_session(
&binding.working_directory.id,
binding.root(),
binding.cwd(),
worker.scope().clone(),
binding.command_environment(),
binding.session_resources(),
)));
} else {
worker.bind_workdir_session(None);
}
worker.bind_workdir_sessions(runtime_local_workdir_router(
&request.workdir_attachments,
worker.scope().clone(),
)?);
if let (Some(runtime_id), Some(workspace_id)) =
(observation_runtime_id, observation_workspace_id.clone())
&& observation_enabled
@@ -1738,7 +1788,7 @@ where
handle: WorkerHandle,
shutdown: Arc<tokio::sync::Mutex<Option<worker::ShutdownReceiver>>>,
controller_task: tokio::task::JoinHandle<()>,
working_directory: Option<WorkingDirectoryBinding>,
workdir_attachments: BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
workspace_client: Option<Arc<dyn WorkspaceClient>>,
) -> WorkerExecutionSpawnResult {
#[cfg(feature = "ws-server")]
@@ -1818,8 +1868,13 @@ where
Err(_) => {
let retained_state =
worker_state.read().ok().map(|state| state.clone());
let retained_working_directory =
working_directory.as_ref().map(|binding| binding.status());
let retained_workdir_attachments = workdir_attachments
.iter()
.map(|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
})
.collect();
let retained_handle = self
.retain_uncertain_unconnected_controller(
&worker_ref,
@@ -1834,7 +1889,7 @@ where
result,
handle: retained_handle,
worker_state: retained_state,
working_directory: retained_working_directory,
workdir_attachments: retained_workdir_attachments,
}
}
}
@@ -1878,7 +1933,7 @@ where
result,
handle: Some(existing_handle),
worker_state: existing_worker_state,
working_directory: None,
workdir_attachments: Vec::new(),
},
}
} else {
@@ -1904,7 +1959,15 @@ where
WorkerExecutionSpawnResult::Connected {
handle: WorkerExecutionHandle::new(worker_ref, self.backend_id()),
worker_state: connected_worker_state,
working_directory: working_directory.map(|binding| binding.status()),
workdir_attachments: workdir_attachments
.into_iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias,
working_directory: binding.status(),
},
)
.collect(),
}
}
}
@@ -2118,67 +2181,134 @@ where
}
let mut request = request;
let mut rollback_working_directory = None;
let working_directory = match (
request.request.working_directory_request.as_ref(),
request.request.working_directory.as_ref(),
) {
(Some(_), Some(_)) => {
let Some(materializer) = self.working_directory_materializer.as_ref().or_else(|| {
(request.request.workdir_attachment_requests.is_empty()
&& request.request.workdir_attachments.is_empty())
.then_some(())
.and(None)
}) else {
if request.request.workdir_attachment_requests.is_empty()
&& request.request.workdir_attachments.is_empty()
{
// No provider is needed for a Workdir-less Worker.
let factory = self.factory.clone();
let bridge_context = request.context.clone();
let worker_ref = request.worker_ref.clone();
let spawn_result = self
.run_cancellable_on_adapter_runtime(self.spawn_restore_timeout, async move {
factory.spawn_controller(request).await
});
let controller = match spawn_result {
Ok(controller) => controller,
Err(message) => {
return WorkerExecutionSpawnResult::Errored(
WorkerExecutionResult::errored(
WorkerExecutionOperation::Spawn,
message,
),
);
}
};
return self.connect_handle(
WorkerExecutionOperation::Spawn,
worker_ref,
bridge_context,
controller.handle,
controller.shutdown,
controller.controller_task,
BTreeMap::new(),
Some(controller.workspace_client),
);
}
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
"Workdir attachments were requested, but no materializer is configured for this Runtime backend",
));
};
let mut rollback_workdirs: Vec<String> = Vec::new();
for attachment in request.request.workdir_attachment_requests.clone() {
if request.workdir_attachments.contains_key(&attachment.alias) {
let message = format!("duplicate Workdir attachment alias `{}`", attachment.alias);
if let Err(result) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
) {
return result;
}
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
"worker spawn cannot specify both working_directory_request and working_directory",
message,
));
}
(Some(working_directory_request), None) => {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
match materializer.materialize(&request.worker_ref, &attachment.working_directory) {
Ok(binding) => {
rollback_workdirs.push(binding.working_directory.id.clone());
request
.workdir_attachments
.insert(attachment.alias, binding);
}
Err(error) => {
let message = error.to_string();
if let Err(result) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
) {
return result;
}
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
"working directory materialization requested, but no materializer is configured for this runtime backend",
message,
));
};
match materializer.materialize(&request.worker_ref, working_directory_request) {
Ok(binding) => {
request.working_directory = Some(binding.clone());
rollback_working_directory = Some(binding.clone());
Some(binding)
}
Err(error) => {
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
error.to_string(),
),
);
}
}
}
(None, Some(working_directory)) => {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
"working directory working_directory requested, but no materializer is configured for this runtime backend",
));
};
match materializer.bind_working_directory(
&working_directory.working_directory_id,
working_directory.relative_cwd.as_deref(),
}
for attachment in request.request.workdir_attachments.clone() {
if request.workdir_attachments.contains_key(&attachment.alias) {
let message = format!("duplicate Workdir attachment alias `{}`", attachment.alias);
if let Err(result) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
) {
Ok(binding) => {
request.working_directory = Some(binding.clone());
Some(binding)
}
Err(error) => {
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
error.to_string(),
),
);
return result;
}
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message,
));
}
match materializer.bind_working_directory(
&attachment.working_directory_id,
attachment.relative_cwd.as_deref(),
) {
Ok(binding) => {
request
.workdir_attachments
.insert(attachment.alias, binding);
}
Err(error) => {
let message = error.to_string();
if let Err(result) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&request.workdir_attachments,
&rollback_workdirs,
&message,
) {
return result;
}
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Spawn,
message,
));
}
}
(None, None) => None,
};
}
let workdir_attachments = request.workdir_attachments.clone();
let factory = self.factory.clone();
let bridge_context = request.context.clone();
@@ -2191,11 +2321,13 @@ where
let controller = match spawn_result {
Ok(controller) => controller,
Err(message) => {
if let (Some(materializer), Some(binding)) = (
self.working_directory_materializer.as_ref(),
rollback_working_directory.as_ref(),
if let Err(result) = rollback_materialized_spawn_workdirs(
materializer.as_ref(),
&workdir_attachments,
&rollback_workdirs,
&message,
) {
let _ = materializer.cleanup_working_directory(&binding.working_directory.id);
return result;
}
return WorkerExecutionSpawnResult::Errored(WorkerExecutionResult::errored(
WorkerExecutionOperation::Spawn,
@@ -2211,7 +2343,7 @@ where
controller.handle,
controller.shutdown,
controller.controller_task,
working_directory,
workdir_attachments,
Some(controller.workspace_client),
)
}
@@ -2232,46 +2364,56 @@ where
}
drop(workers);
if let Some(previous) = request.previous_working_directory.as_ref() {
if !request.previous_workdir_attachments.is_empty() {
let materializer = self
.working_directory_materializer
.as_ref()
.ok_or_else(|| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir binding cannot be restored by this Runtime",
"Persisted Worker Workdir attachments cannot be restored by this Runtime",
)
})?;
let current = materializer
.working_directory_status(&previous.summary.working_directory_id)
.map_err(|message| {
WorkerExecutionResult::rejected(
for previous in &request.previous_workdir_attachments {
let current = materializer
.working_directory_status(
&previous.working_directory.summary.working_directory_id,
)
.map_err(|message| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!(
"Persisted Worker Workdir attachment `{}` is unavailable: {message}",
previous.alias
),
)
})?;
if current.summary.status != WorkingDirectoryStatusKind::Active {
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!("Persisted Worker Workdir is unavailable: {message}"),
)
})?;
if current.summary.status != WorkingDirectoryStatusKind::Active {
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir is not active",
));
format!(
"Persisted Worker Workdir attachment `{}` is not active",
previous.alias
),
));
}
}
}
if request.previous_working_directory.is_none()
&& request.request.working_directory_request.is_some()
if request.previous_workdir_attachments.is_empty()
&& !request.request.workdir_attachment_requests.is_empty()
{
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir allocation is unavailable",
"Persisted Worker Workdir allocations are unavailable",
));
}
if request.previous_working_directory.is_none()
&& request.request.working_directory.is_some()
if request.previous_workdir_attachments.is_empty()
&& !request.request.workdir_attachments.is_empty()
&& self.working_directory_materializer.is_none()
{
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir claim cannot be restored by this Runtime",
"Persisted Worker Workdir claims cannot be restored by this Runtime",
));
}
if let Some(workspace_api) = request.request.workspace_api.as_ref() {
@@ -2321,58 +2463,40 @@ where
}
drop(workers);
let working_directory = match request.previous_working_directory.clone() {
Some(status) => {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"persisted worker has a working directory binding, but no materializer is configured for this runtime backend",
));
};
let relative_cwd = request
.request
.working_directory
.as_ref()
.and_then(|working_directory| working_directory.relative_cwd.as_deref());
match materializer
.bind_working_directory(&status.summary.working_directory_id, relative_cwd)
{
Ok(binding) => {
request.working_directory = Some(binding.clone());
Some(binding)
}
Err(error) => {
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
error.to_string(),
),
);
}
}
}
None if request.request.working_directory_request.is_some() => {
let mut workdir_attachments = BTreeMap::new();
if !request.previous_workdir_attachments.is_empty() {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"persisted worker requested a working directory, but no persisted working directory binding is available to restore",
"persisted Worker has Workdir attachments, but no materializer is configured for this Runtime backend",
));
}
None if request.request.working_directory.is_some() => {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"persisted worker has a working directory claim, but no materializer is configured for this runtime backend",
));
};
let working_directory =
request.request.working_directory.as_ref().expect("checked");
};
for status in request.previous_workdir_attachments.clone() {
let relative_cwd = request
.request
.workdir_attachments
.iter()
.find(|claim| claim.alias == status.alias)
.and_then(|claim| claim.relative_cwd.as_deref());
match materializer.bind_working_directory(
&working_directory.working_directory_id,
working_directory.relative_cwd.as_deref(),
&status.working_directory.summary.working_directory_id,
relative_cwd,
) {
Ok(binding) => {
request.working_directory = Some(binding.clone());
Some(binding)
if workdir_attachments
.insert(status.alias.clone(), binding)
.is_some()
{
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!(
"duplicate persisted Workdir attachment alias `{}`",
status.alias
),
),
);
}
}
Err(error) => {
return WorkerExecutionSpawnResult::Rejected(
@@ -2384,8 +2508,43 @@ where
}
}
}
None => None,
};
} else if !request.request.workdir_attachments.is_empty() {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"persisted Worker has Workdir claims, but no materializer is configured for this Runtime backend",
));
};
for claim in request.request.workdir_attachments.clone() {
match materializer.bind_working_directory(
&claim.working_directory_id,
claim.relative_cwd.as_deref(),
) {
Ok(binding) => {
if workdir_attachments
.insert(claim.alias.clone(), binding)
.is_some()
{
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!("duplicate Workdir attachment alias `{}`", claim.alias),
),
);
}
}
Err(error) => {
return WorkerExecutionSpawnResult::Rejected(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
error.to_string(),
),
);
}
}
}
}
request.workdir_attachments = workdir_attachments.clone();
let factory = self.factory.clone();
let bridge_context = request.context.clone();
@@ -2405,7 +2564,7 @@ where
),
handle: None,
worker_state: None,
working_directory: None,
workdir_attachments: Vec::new(),
};
}
};
@@ -2417,7 +2576,7 @@ where
controller.handle,
controller.shutdown,
controller.controller_task,
working_directory,
workdir_attachments,
Some(controller.workspace_client),
)
}
@@ -2739,8 +2898,8 @@ mod tests {
use crate::Runtime as EmbeddedRuntime;
use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, MaterializerKind, ProfileSelector,
RepositorySelector, WorkingDirectoryClaim, WorkingDirectoryRepository,
WorkingDirectoryRequest, WorkspaceApiRef,
RepositorySelector, WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
};
use crate::execution::WorkerExecutionContext;
use crate::identity::WorkerId;
@@ -3279,18 +3438,17 @@ mod tests {
FsStore::new(&self.store_dir).map_err(|err| err.to_string())?,
FsWorkerStore::new(&self.worker_metadata_dir).map_err(|err| err.to_string())?,
);
let filesystem_authority = request
.working_directory
.as_ref()
let only_workdir = (request.workdir_attachments.len() == 1)
.then(|| request.workdir_attachments.values().next())
.flatten();
let filesystem_authority = only_workdir
.map(|binding| {
let cwd = binding.cwd().to_path_buf();
self.observed_cwds.lock().unwrap().push(cwd.clone());
WorkerFilesystemAuthority::local(binding.root().to_path_buf(), cwd)
})
.unwrap_or(WorkerFilesystemAuthority::None);
let scope_root = request
.working_directory
.as_ref()
let scope_root = only_workdir
.map(|binding| binding.root().to_path_buf())
.unwrap_or_else(|| self.cwd.clone());
let workspace_backend_ref = RuntimeWorkspaceBackendRef::from_worker_request(
@@ -3348,7 +3506,7 @@ mod tests {
request: request.request,
workspace_scope: request.workspace_scope,
context: request.context,
working_directory: request.working_directory,
workdir_attachments: request.workdir_attachments,
config_bundle: request.config_bundle,
};
self.spawn_controller(request).await
@@ -3513,8 +3671,8 @@ mod tests {
digest: bundle.metadata.digest,
}),
initial_input: None,
working_directory_request: None,
working_directory: None,
workdir_attachment_requests: Vec::new(),
workdir_attachments: Vec::new(),
worker_observation_enabled: false,
worker_observation_grants: Vec::new(),
workspace_api: None,
@@ -3595,6 +3753,7 @@ mod tests {
source_fingerprint: "sha256:test".to_string(),
selector: Some(RepositorySelector::from("HEAD")),
},
display_name: None,
materializer: MaterializerKind::RuntimeGitClone,
backend_workdir_id: None,
materialization: None,
@@ -3698,7 +3857,7 @@ mod tests {
request: create_request("1"),
workspace_scope: None,
context: test_execution_context(worker_ref),
working_directory: None,
workdir_attachments: BTreeMap::new(),
config_bundle: None,
};
@@ -3830,8 +3989,8 @@ mod tests {
"server-main",
)),
context: test_execution_context(worker_ref),
previous_working_directory: None,
working_directory: None,
previous_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
config_bundle: None,
})
.await
@@ -3920,8 +4079,8 @@ mod tests {
request: create_request("embedded restore"),
workspace_scope: None,
context: test_execution_context(worker_ref),
previous_working_directory: None,
working_directory: None,
previous_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
config_bundle: None,
})
.await
@@ -3988,7 +4147,7 @@ mod tests {
&[
"WorkerBootstrap::new(",
".prepare()",
"worker.bind_workdir_session(",
"worker.bind_workdir_sessions(",
"worker.bind_worker_observation_provider(",
"install_runtime_flow_transition_feature()",
"prepared.start()",
@@ -3998,7 +4157,7 @@ mod tests {
restore,
&[
"Worker::restore_from_worker_metadata_with_context(",
"worker.bind_workdir_session(",
"worker.bind_workdir_sessions(",
"worker.bind_worker_observation_provider(",
"install_runtime_flow_transition_feature()",
"PreparedWorker::new(",
@@ -4354,7 +4513,10 @@ mod tests {
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let mut request = create_request("chat");
request.working_directory_request = Some(working_directory_request(repo.path()));
request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory: working_directory_request(repo.path()),
}];
let detail = runtime.create_worker(request).unwrap();
runtime
@@ -4376,7 +4538,7 @@ mod tests {
);
}
assert!(detail.working_directory.is_some());
assert_eq!(detail.workdir_attachments.len(), 1);
let cwds = observed_cwds.lock().unwrap();
assert_eq!(cwds.len(), 1);
let cwd = &cwds[0];
@@ -4387,6 +4549,71 @@ mod tests {
observed_workspace_clients.lock().unwrap().as_slice(),
&[("unavailable".to_string(), None, false)]
);
let workdir_id = detail.workdir_attachments[0]
.working_directory
.summary
.working_directory_id
.clone();
let second = runtime.create_worker(create_request("second")).unwrap();
let conflict = runtime
.replace_worker_workdir_attachments(
&second.worker_ref,
vec![WorkingDirectoryAttachmentClaim {
alias: workdir::WorkdirAttachmentAlias::new("conflict").unwrap(),
working_directory_id: workdir_id.clone(),
relative_cwd: None,
}],
)
.unwrap_err();
assert!(matches!(
conflict,
crate::error::RuntimeError::InvalidRequest(_)
));
let inactive = runtime
.create_working_directory(working_directory_request(repo.path()))
.unwrap();
let inactive_id = inactive.summary.working_directory_id;
runtime.cleanup_working_directory(&inactive_id).unwrap();
let inactive_error = runtime
.replace_worker_workdir_attachments(
&second.worker_ref,
vec![WorkingDirectoryAttachmentClaim {
alias: workdir::WorkdirAttachmentAlias::new("inactive").unwrap(),
working_directory_id: inactive_id,
relative_cwd: None,
}],
)
.unwrap_err();
assert!(matches!(
inactive_error,
crate::error::RuntimeError::InvalidRequest(_)
| crate::error::RuntimeError::WorkingDirectory(_)
));
let detached = runtime
.replace_worker_workdir_attachments(&detail.worker_ref, Vec::new())
.unwrap();
assert!(detached.workdir_attachments.is_empty());
runtime.stop_worker(&detail.worker_ref, None).unwrap();
let restored_without_workdir = runtime.restore_worker(&detail.worker_ref).unwrap();
assert!(restored_without_workdir.workdir_attachments.is_empty());
let alias = workdir::WorkdirAttachmentAlias::new("restored").unwrap();
runtime
.replace_worker_workdir_attachments(
&detail.worker_ref,
vec![WorkingDirectoryAttachmentClaim {
alias: alias.clone(),
working_directory_id: workdir_id,
relative_cwd: None,
}],
)
.unwrap();
runtime.stop_worker(&detail.worker_ref, None).unwrap();
let stopped = runtime.worker_detail(&detail.worker_ref).unwrap();
assert_eq!(stopped.workdir_attachments.len(), 1);
assert_eq!(stopped.workdir_attachments[0].alias, alias);
}
#[test]
@@ -4620,12 +4847,13 @@ mod tests {
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let mut request = create_request("chat");
request.working_directory_request = Some(working_directory_request(repo.path()));
request.workdir_attachment_requests = vec![WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory: working_directory_request(repo.path()),
}];
let detail = runtime.create_worker(request).unwrap();
let workdir_id = detail
let workdir_id = detail.workdir_attachments[0]
.working_directory
.as_ref()
.unwrap()
.summary
.working_directory_id
.clone();
@@ -4642,7 +4870,6 @@ mod tests {
crate::catalog::WorkingDirectoryStatusKind::Active
);
assert_eq!(status.summary.cleanliness.as_deref(), Some("clean"));
assert_eq!(status.summary.primary_worker_id, None);
}
#[test]
@@ -4663,10 +4890,11 @@ mod tests {
let clone_root = materialized_clone_root(runtime_base.path(), &workdir_id);
assert!(clone_root.join("README.md").exists());
let mut request = create_request("chat");
request.working_directory = Some(WorkingDirectoryClaim {
request.workdir_attachments = vec![WorkingDirectoryAttachmentClaim {
alias: workdir::WorkdirAttachmentAlias::new("workdir").unwrap(),
working_directory_id: workdir_id.clone(),
relative_cwd: None,
});
}];
let error = runtime.create_worker(request).unwrap_err();
@@ -4691,7 +4919,17 @@ mod tests {
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let mut request = create_request("chat");
request.working_directory_request = Some(working_directory_request(repo.path()));
let second_repo = create_clean_repo();
request.workdir_attachment_requests = vec![
WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory: working_directory_request(repo.path()),
},
WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("docs").unwrap(),
working_directory: working_directory_request(second_repo.path()),
},
];
let error = runtime.create_worker(request).unwrap_err();
@@ -4708,4 +4946,41 @@ mod tests {
assert_eq!(remaining_workdirs, 0);
assert!(!working_directories_root.join(".repository-cache").exists());
}
#[test]
fn second_materialization_failure_rolls_back_first_attachment() {
let runtime_base = tempfile::tempdir().unwrap();
let repo = create_clean_repo();
let missing_repo = runtime_base.path().join("missing-repository");
let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
.unwrap()
.with_working_directory_materializer(RuntimeGitMaterializer::new(runtime_base.path()));
let runtime =
EmbeddedRuntime::with_execution_backend(RuntimeOptions::default(), Arc::new(backend))
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let mut request = create_request("chat");
request.workdir_attachment_requests = vec![
WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory: working_directory_request(repo.path()),
},
WorkingDirectoryAttachmentRequest {
alias: workdir::WorkdirAttachmentAlias::new("docs").unwrap(),
working_directory: working_directory_request(&missing_repo),
},
];
let error = runtime.create_worker(request).unwrap_err();
assert!(format!("{error:?}").contains("repository"));
let remaining_workdirs = fs::read_dir(runtime_base.path())
.map(|entries| {
entries
.flatten()
.filter(|entry| !entry.file_name().to_string_lossy().starts_with('.'))
.count()
})
.unwrap_or(0);
assert_eq!(remaining_workdirs, 0);
}
}
crates/worker-runtime/src/working_directory.rs
+7 -2
View File
@@ -57,6 +57,8 @@ pub struct WorkingDirectoryEvidence {
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectory {
pub id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display_name: Option<String>,
pub repository_id: String,
pub materializer_kind: MaterializerKind,
pub evidence: WorkingDirectoryEvidence,
@@ -68,6 +70,7 @@ impl WorkingDirectory {
pub fn status_summary(&self) -> WorkingDirectorySummary {
WorkingDirectorySummary {
working_directory_id: self.id.clone(),
display_name: self.display_name.clone(),
repository_id: self.repository_id.clone(),
creation_selector: self.evidence.requested_selector.clone(),
creation_ref: Some(self.evidence.resolved_commit.clone()),
@@ -80,7 +83,6 @@ impl WorkingDirectory {
cleanup_target: Some(self.cleanup_target.clone()),
status: self.status.clone(),
cleanliness: None,
primary_worker_id: None,
occupied_by: None,
}
}
@@ -717,6 +719,7 @@ impl RuntimeGitMaterializer {
WorkingDirectoryStatus {
summary: WorkingDirectorySummary {
working_directory_id: working_directory_id.to_string(),
display_name: None,
repository_id: "unknown".to_string(),
creation_selector: None,
creation_ref: None,
@@ -733,7 +736,6 @@ impl RuntimeGitMaterializer {
}),
status: WorkingDirectoryStatusKind::Corrupted,
cleanliness: Some("unknown".to_string()),
primary_worker_id: None,
occupied_by: None,
},
}
@@ -1072,6 +1074,7 @@ impl RuntimeGitMaterializer {
let context = request.materialization.as_ref();
let working_directory = WorkingDirectory {
id: working_directory_id.clone(),
display_name: request.display_name.clone(),
repository_id: request.repository.id.clone(),
materializer_kind: MaterializerKind::RuntimeGitClone,
evidence: WorkingDirectoryEvidence {
@@ -1211,6 +1214,7 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
let working_request = WorkingDirectoryRequest {
repository: request.repository.clone(),
display_name: None,
materializer: MaterializerKind::RuntimeGitClone,
backend_workdir_id: None,
materialization: request.materialization.clone(),
@@ -3316,6 +3320,7 @@ mod tests {
source_fingerprint: "sha256:test".to_string(),
selector: Some(RepositorySelector::from("HEAD")),
},
display_name: None,
materializer: MaterializerKind::RuntimeGitClone,
backend_workdir_id: None,
materialization: None,