fix: decouple workdir attachments from materialization
This commit is contained in:
@@ -198,6 +198,14 @@ pub struct RepositoryRefObservation {
|
||||
pub observed_at_epoch_seconds: u64,
|
||||
}
|
||||
|
||||
/// Workspace-authoritative alias-to-Workdir mapping. This carries no claim that
|
||||
/// the Worker Runtime materialized or can bind the referenced Workdir locally.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct LogicalWorkdirAttachment {
|
||||
pub alias: workdir::WorkdirAttachmentAlias,
|
||||
pub working_directory_id: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct WorkingDirectoryAttachmentClaim {
|
||||
pub alias: workdir::WorkdirAttachmentAlias,
|
||||
|
||||
@@ -259,6 +259,7 @@ pub struct WorkerExecutionRestoreRequest {
|
||||
pub workspace_scope: Option<crate::runtime::RuntimeWorkspaceScope>,
|
||||
pub context: WorkerExecutionContext,
|
||||
pub previous_workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
|
||||
pub logical_workdir_attachments: Vec<crate::catalog::LogicalWorkdirAttachment>,
|
||||
pub workdir_attachments: BTreeMap<workdir::WorkdirAttachmentAlias, WorkingDirectoryBinding>,
|
||||
pub config_bundle: Option<ConfigBundle>,
|
||||
}
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use crate::catalog::{
|
||||
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerRestoreIntent, WorkerStatus,
|
||||
WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
|
||||
WorkingDirectoryAttachmentStatus, WorkingDirectoryRequest,
|
||||
ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, ProfileSelector,
|
||||
WorkerRestoreIntent, WorkerStatus, WorkingDirectoryAttachmentClaim,
|
||||
WorkingDirectoryAttachmentRequest, WorkingDirectoryAttachmentStatus, WorkingDirectoryRequest,
|
||||
};
|
||||
use crate::config_bundle::ConfigBundle;
|
||||
use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
|
||||
@@ -444,6 +444,7 @@ pub(crate) struct PersistedWorkerRecord {
|
||||
pub(crate) execution_state: PersistedWorkerExecutionState,
|
||||
pub(crate) workspace_id: Option<String>,
|
||||
pub(crate) workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
|
||||
pub(crate) logical_workdir_attachments: Vec<LogicalWorkdirAttachment>,
|
||||
}
|
||||
|
||||
fn runtime_io_error(operation: &'static str, path: &Path, source: std::io::Error) -> RuntimeError {
|
||||
@@ -1688,6 +1689,8 @@ struct WorkerIdentityRecord {
|
||||
workspace_id: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
logical_workdir_attachments: Vec<LogicalWorkdirAttachment>,
|
||||
/// Schema-v8 compatibility input. New records serialize only `workdir_attachments`.
|
||||
#[serde(
|
||||
default,
|
||||
@@ -1787,6 +1790,7 @@ impl WorkerIdentityRecord {
|
||||
status: worker.status,
|
||||
workspace_id: worker.workspace_id.clone(),
|
||||
workdir_attachments: worker.workdir_attachments.clone(),
|
||||
logical_workdir_attachments: worker.logical_workdir_attachments.clone(),
|
||||
legacy_working_directory: None,
|
||||
}
|
||||
}
|
||||
@@ -1860,6 +1864,7 @@ impl WorkerIdentityRecord {
|
||||
execution_state,
|
||||
workspace_id: self.workspace_id,
|
||||
workdir_attachments,
|
||||
logical_workdir_attachments: self.logical_workdir_attachments,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -2337,6 +2342,7 @@ mod tests {
|
||||
}))
|
||||
.unwrap();
|
||||
let persisted = identity.into_persisted(PersistedWorkerExecutionState::Unavailable);
|
||||
assert!(persisted.logical_workdir_attachments.is_empty());
|
||||
assert_eq!(persisted.workdir_attachments.len(), 1);
|
||||
assert_eq!(persisted.workdir_attachments[0].alias.as_str(), "workdir");
|
||||
assert_eq!(
|
||||
@@ -2369,6 +2375,44 @@ mod tests {
|
||||
assert!(serialized["request"].get("working_directory").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn worker_identity_round_trips_logical_workdir_attachments() {
|
||||
let worker_id = WorkerId::now_v7();
|
||||
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",
|
||||
"logical_workdir_attachments": [{
|
||||
"alias": "checkout",
|
||||
"working_directory_id": "remote-workdir"
|
||||
}]
|
||||
}))
|
||||
.unwrap();
|
||||
let persisted = identity.into_persisted(PersistedWorkerExecutionState::Unavailable);
|
||||
assert_eq!(
|
||||
persisted.logical_workdir_attachments,
|
||||
[LogicalWorkdirAttachment {
|
||||
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
|
||||
working_directory_id: "remote-workdir".to_string(),
|
||||
}]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn startup_recovers_staging_created_before_source_backup() {
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
|
||||
@@ -428,7 +428,7 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
|
||||
worker_id: String,
|
||||
value: runtime_api::WorkerWorkdirAttachmentsRequest,
|
||||
) -> Result<runtime_api::WorkerResponse, runtime_api::RuntimeApiError> {
|
||||
let attachments = request(value.workdir_attachments)?;
|
||||
let attachments = request(value.logical_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)?
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
use crate::catalog::{
|
||||
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource,
|
||||
RepositoryRefObservation, RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck,
|
||||
WorkerRestoreIntent, WorkerStatus, WorkerSummary, WorkingDirectoryAttachmentClaim,
|
||||
WorkingDirectoryAttachmentStatus, WorkingDirectoryRepositoryAccessRequest,
|
||||
WorkingDirectoryRequest, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus,
|
||||
WorkspaceApiRef,
|
||||
ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, ProfileSelector,
|
||||
ProfileSourceArchiveSource, RepositoryRefObservation, RepositoryRefObservationRequest,
|
||||
WorkerDetail, WorkerLifecycleAck, WorkerRestoreIntent, WorkerStatus, WorkerSummary,
|
||||
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
|
||||
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef,
|
||||
};
|
||||
use crate::config_bundle::{
|
||||
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
|
||||
@@ -1061,6 +1060,7 @@ impl Runtime {
|
||||
execution_bound: true,
|
||||
restore_intent: WorkerRestoreIntent::Explicit,
|
||||
workdir_attachments: Vec::new(),
|
||||
logical_workdir_attachments: Vec::new(),
|
||||
execution_handle: None,
|
||||
internal_workers: InternalWorkerActivityProjection::default(),
|
||||
};
|
||||
@@ -1447,106 +1447,95 @@ impl Runtime {
|
||||
&self,
|
||||
scope: &RuntimeWorkspaceScope,
|
||||
worker_ref: &WorkerRef,
|
||||
attachments: Vec<WorkingDirectoryAttachmentClaim>,
|
||||
attachments: Vec<LogicalWorkdirAttachment>,
|
||||
) -> 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.
|
||||
/// Persist the Workspace-authoritative logical attachment set used by live
|
||||
/// mutation and future restore. Runtime-local bindings remain a separate,
|
||||
/// optional implementation detail and never authorize logical attachment.
|
||||
pub fn replace_worker_workdir_attachments(
|
||||
&self,
|
||||
worker_ref: &WorkerRef,
|
||||
mut attachments: Vec<WorkingDirectoryAttachmentClaim>,
|
||||
mut attachments: Vec<LogicalWorkdirAttachment>,
|
||||
) -> 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
|
||||
)));
|
||||
}
|
||||
validate_logical_workdir_attachments(&attachments)?;
|
||||
attachments.sort_by(|left, right| left.alias.cmp(&right.alias));
|
||||
|
||||
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)?;
|
||||
if !attachments.is_empty() && worker.workspace_id.is_none() {
|
||||
return Err(RuntimeError::InvalidRequest(
|
||||
"logical Workdir attachments require Workspace authority".to_string(),
|
||||
));
|
||||
}
|
||||
if worker.request.is_none() {
|
||||
return Err(RuntimeError::WorkerExecutionUnavailable {
|
||||
worker_id: worker.worker_id,
|
||||
message: "persisted Worker restore request is unavailable".to_string(),
|
||||
});
|
||||
}
|
||||
let logical_by_alias = attachments
|
||||
.iter()
|
||||
.map(|attachment| {
|
||||
(
|
||||
attachment.alias.clone(),
|
||||
attachment.working_directory_id.as_str(),
|
||||
)
|
||||
})
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
for local in &worker.workdir_attachments {
|
||||
let local_id = local
|
||||
.working_directory
|
||||
.summary
|
||||
.working_directory_id
|
||||
.as_str();
|
||||
if let Some(logical_id) = logical_by_alias.get(&local.alias)
|
||||
&& *logical_id != local_id
|
||||
{
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"logical Workdir attachment alias `{}` conflicts with Runtime-local Workdir {}",
|
||||
local.alias, local_id
|
||||
)));
|
||||
}
|
||||
}
|
||||
let retained_local = worker
|
||||
.workdir_attachments
|
||||
.iter()
|
||||
.filter_map(|local| {
|
||||
let local_id = local
|
||||
.working_directory
|
||||
.summary
|
||||
.working_directory_id
|
||||
.as_str();
|
||||
(logical_by_alias.get(&local.alias).copied() == Some(local_id))
|
||||
.then(|| (local.alias.clone(), local_id.to_string()))
|
||||
})
|
||||
.collect::<BTreeMap<_, _>>();
|
||||
worker
|
||||
.workdir_attachments
|
||||
.retain(|local| retained_local.contains_key(&local.alias));
|
||||
let request = worker
|
||||
.request
|
||||
.as_mut()
|
||||
.expect("restore request checked above");
|
||||
request.workdir_attachment_requests.clear();
|
||||
request.workdir_attachments = attachments;
|
||||
worker.workdir_attachments = statuses;
|
||||
.expect("restore request existence checked above");
|
||||
request
|
||||
.workdir_attachment_requests
|
||||
.retain(|local| retained_local.contains_key(&local.alias));
|
||||
request.workdir_attachments.retain(|local| {
|
||||
retained_local
|
||||
.get(&local.alias)
|
||||
.is_some_and(|workdir_id| workdir_id == &local.working_directory_id)
|
||||
});
|
||||
worker.logical_workdir_attachments = attachments;
|
||||
}
|
||||
if let Err(error) = state.persist_worker(&worker_ref.worker_id) {
|
||||
state.workers.insert(worker_ref.worker_id, previous);
|
||||
@@ -1702,6 +1691,7 @@ impl Runtime {
|
||||
workspace_scope,
|
||||
context: self.execution_context(worker_ref.clone()),
|
||||
previous_workdir_attachments: worker.workdir_attachments.clone(),
|
||||
logical_workdir_attachments: worker.logical_workdir_attachments.clone(),
|
||||
workdir_attachments: BTreeMap::new(),
|
||||
config_bundle: None,
|
||||
};
|
||||
@@ -2225,6 +2215,8 @@ impl Runtime {
|
||||
let _ = worker.apply_worker_state(snapshot);
|
||||
}
|
||||
worker.restore_intent = restore_intent_for_status(worker.status);
|
||||
worker.logical_workdir_attachments =
|
||||
logical_workdir_attachments_from_statuses(&workdir_attachments);
|
||||
worker.workdir_attachments = workdir_attachments;
|
||||
worker.detail()
|
||||
};
|
||||
@@ -2875,6 +2867,8 @@ impl Runtime {
|
||||
candidate.execution_bound = true;
|
||||
candidate.status = WorkerStatus::Idle;
|
||||
candidate.restore_intent = WorkerRestoreIntent::Explicit;
|
||||
candidate.logical_workdir_attachments =
|
||||
logical_workdir_attachments_from_statuses(&workdir_attachments);
|
||||
candidate.workdir_attachments = workdir_attachments;
|
||||
state.workers.insert(worker_ref.worker_id, candidate);
|
||||
state.publish_worker_upsert(worker_ref.worker_id)?;
|
||||
@@ -3218,6 +3212,11 @@ impl RuntimeState {
|
||||
(None, false, false, restore_intent_for_status(worker.status))
|
||||
}
|
||||
};
|
||||
let logical_workdir_attachments = if worker.logical_workdir_attachments.is_empty() {
|
||||
logical_workdir_attachments_from_statuses(&worker.workdir_attachments)
|
||||
} else {
|
||||
worker.logical_workdir_attachments
|
||||
};
|
||||
workers.insert(
|
||||
worker_id,
|
||||
WorkerRecord {
|
||||
@@ -3236,6 +3235,7 @@ impl RuntimeState {
|
||||
execution_bound,
|
||||
restore_intent,
|
||||
workdir_attachments: worker.workdir_attachments,
|
||||
logical_workdir_attachments,
|
||||
execution_handle: None,
|
||||
internal_workers: InternalWorkerActivityProjection::default(),
|
||||
},
|
||||
@@ -4132,6 +4132,7 @@ struct WorkerRecord {
|
||||
execution_bound: bool,
|
||||
restore_intent: WorkerRestoreIntent,
|
||||
workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
|
||||
logical_workdir_attachments: Vec<LogicalWorkdirAttachment>,
|
||||
execution_handle: Option<WorkerExecutionHandle>,
|
||||
internal_workers: InternalWorkerActivityProjection,
|
||||
}
|
||||
@@ -4205,6 +4206,7 @@ impl WorkerRecord {
|
||||
execution_state,
|
||||
workspace_id: self.workspace_id.clone(),
|
||||
workdir_attachments: self.workdir_attachments.clone(),
|
||||
logical_workdir_attachments: self.logical_workdir_attachments.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4279,6 +4281,50 @@ fn requested_workdir_ids(request: &CreateWorkerRequest) -> Vec<&str> {
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn logical_workdir_attachments_from_statuses(
|
||||
attachments: &[crate::catalog::WorkingDirectoryAttachmentStatus],
|
||||
) -> Vec<LogicalWorkdirAttachment> {
|
||||
attachments
|
||||
.iter()
|
||||
.map(|attachment| LogicalWorkdirAttachment {
|
||||
alias: attachment.alias.clone(),
|
||||
working_directory_id: attachment
|
||||
.working_directory
|
||||
.summary
|
||||
.working_directory_id
|
||||
.clone(),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn validate_logical_workdir_attachments(
|
||||
attachments: &[LogicalWorkdirAttachment],
|
||||
) -> Result<(), RuntimeError> {
|
||||
let mut aliases = std::collections::BTreeSet::new();
|
||||
let mut workdir_ids = std::collections::BTreeSet::new();
|
||||
for attachment in attachments {
|
||||
let workdir_id = attachment.working_directory_id.trim();
|
||||
if workdir_id.is_empty() || workdir_id.chars().any(char::is_control) {
|
||||
return Err(RuntimeError::InvalidRequest(
|
||||
"logical Workdir attachment id must not be empty or contain control characters"
|
||||
.to_string(),
|
||||
));
|
||||
}
|
||||
if !aliases.insert(attachment.alias.as_str()) {
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"duplicate logical Workdir attachment alias `{}`",
|
||||
attachment.alias
|
||||
)));
|
||||
}
|
||||
if !workdir_ids.insert(workdir_id) {
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"logical working directory `{workdir_id}` cannot be attached more than once"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> {
|
||||
if request.create_fingerprint.trim().is_empty() {
|
||||
return Err(RuntimeError::InvalidRequest(
|
||||
@@ -4615,7 +4661,8 @@ mod tests {
|
||||
ConfigBundleRef, MaterializerKind, ProfileSelector, RepositoryMaterializationContext,
|
||||
RepositorySshCredentialCandidate, RepositorySshMaterializationAccess, SensitiveString,
|
||||
WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
|
||||
WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
|
||||
WorkingDirectoryAttachmentStatus, WorkingDirectoryRepository, WorkingDirectoryRequest,
|
||||
WorkspaceApiRef,
|
||||
};
|
||||
use crate::config_bundle::{
|
||||
ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
|
||||
@@ -6168,6 +6215,59 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn logical_workdir_attachments_do_not_require_local_materialization() {
|
||||
let runtime = runtime_with_backend();
|
||||
let workspace_scope = scope("workspace-a", "server-a");
|
||||
let worker = runtime
|
||||
.create_worker_scoped(
|
||||
&workspace_scope,
|
||||
scoped_task_request("logical attachment", "workspace-a"),
|
||||
)
|
||||
.unwrap();
|
||||
let attachment = LogicalWorkdirAttachment {
|
||||
alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
|
||||
working_directory_id: "workdir-on-another-runtime".to_string(),
|
||||
};
|
||||
|
||||
let detail = runtime
|
||||
.replace_worker_workdir_attachments_scoped(
|
||||
&workspace_scope,
|
||||
&worker.worker_ref,
|
||||
vec![attachment.clone()],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(detail.workdir_attachments.is_empty());
|
||||
{
|
||||
let state = runtime.lock().unwrap();
|
||||
assert_eq!(
|
||||
state
|
||||
.worker(&worker.worker_ref)
|
||||
.unwrap()
|
||||
.logical_workdir_attachments,
|
||||
[attachment.clone()]
|
||||
);
|
||||
}
|
||||
runtime.stop_worker(&worker.worker_ref, None).unwrap();
|
||||
runtime
|
||||
.replace_worker_workdir_attachments_scoped(
|
||||
&workspace_scope,
|
||||
&worker.worker_ref,
|
||||
vec![attachment.clone()],
|
||||
)
|
||||
.unwrap();
|
||||
runtime.restore_worker(&worker.worker_ref).unwrap();
|
||||
let state = runtime.lock().unwrap();
|
||||
assert_eq!(
|
||||
state
|
||||
.worker(&worker.worker_ref)
|
||||
.unwrap()
|
||||
.logical_workdir_attachments,
|
||||
[attachment]
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn workspace_owner_binding_rejects_other_backend_and_forgets_after_last_worker_delete() {
|
||||
let runtime = runtime_with_backend();
|
||||
|
||||
@@ -115,6 +115,7 @@ use workdir::{
|
||||
|
||||
#[cfg(test)]
|
||||
use worker::WorkerController;
|
||||
use worker::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession;
|
||||
use worker::feature::builtin::{
|
||||
CompositeWorkerObservationProvider, WorkerObservationError, WorkerObservationProvider,
|
||||
WorkerObservationSubject, WorkerObservationSubjectRef, WorkerSessionCapture,
|
||||
@@ -796,6 +797,42 @@ fn runtime_local_workdir_router(
|
||||
Ok(router)
|
||||
}
|
||||
|
||||
fn restored_workdir_router(
|
||||
local_attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
|
||||
logical_attachments: &[crate::catalog::LogicalWorkdirAttachment],
|
||||
scope: manifest::SharedScope,
|
||||
workspace_client: Arc<dyn WorkspaceClient>,
|
||||
) -> Result<Arc<WorkdirSessionRouter>, String> {
|
||||
let router = runtime_local_workdir_router(local_attachments, scope)?;
|
||||
for attachment in logical_attachments {
|
||||
if let Some(local) = local_attachments.get(&attachment.alias) {
|
||||
if local.working_directory.id != attachment.working_directory_id {
|
||||
return Err(format!(
|
||||
"logical Workdir attachment `{}` conflicts with Runtime-local Workdir {}",
|
||||
attachment.alias, local.working_directory.id
|
||||
));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
router
|
||||
.attach(
|
||||
attachment.alias.clone(),
|
||||
WorkspaceAttachedWorkdirSession::handle_for_workdir(
|
||||
workspace_client.clone(),
|
||||
attachment.alias.as_str(),
|
||||
&attachment.working_directory_id,
|
||||
),
|
||||
)
|
||||
.map_err(|error| {
|
||||
format!(
|
||||
"bind logical Workdir attachment `{}`: {error}",
|
||||
attachment.alias
|
||||
)
|
||||
})?;
|
||||
}
|
||||
Ok(router)
|
||||
}
|
||||
|
||||
fn bind_workspace_memory_settings(
|
||||
manifest: &mut manifest::WorkerManifest,
|
||||
request: &CreateWorkerRequest,
|
||||
@@ -1259,10 +1296,13 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
};
|
||||
validate_worker_memory_settings(worker.manifest(), &request.request)?;
|
||||
let flow_transition_enabled = worker.manifest().feature.flow.enabled;
|
||||
worker.bind_workdir_sessions(runtime_local_workdir_router(
|
||||
let workdir_sessions = restored_workdir_router(
|
||||
&request.workdir_attachments,
|
||||
&request.logical_workdir_attachments,
|
||||
worker.scope().clone(),
|
||||
)?);
|
||||
worker.workspace_client_handle(),
|
||||
)?;
|
||||
worker.bind_workdir_sessions(workdir_sessions);
|
||||
if let (Some(runtime_id), Some(workspace_id)) =
|
||||
(observation_runtime_id, observation_workspace_id.clone())
|
||||
&& observation_enabled
|
||||
@@ -2897,9 +2937,10 @@ mod tests {
|
||||
|
||||
use crate::Runtime as EmbeddedRuntime;
|
||||
use crate::catalog::{
|
||||
ConfigBundleRef, CreateWorkerRequest, MaterializerKind, ProfileSelector,
|
||||
RepositorySelector, WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
|
||||
WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
|
||||
ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, MaterializerKind,
|
||||
ProfileSelector, RepositorySelector, WorkingDirectoryAttachmentClaim,
|
||||
WorkingDirectoryAttachmentRequest, WorkingDirectoryRepository, WorkingDirectoryRequest,
|
||||
WorkspaceApiRef,
|
||||
};
|
||||
use crate::execution::WorkerExecutionContext;
|
||||
use crate::identity::WorkerId;
|
||||
@@ -2915,6 +2956,31 @@ mod tests {
|
||||
use manifest::{Scope, WorkerManifest};
|
||||
use session_store::{LogEntry, WorkerMetadataStore};
|
||||
|
||||
#[test]
|
||||
fn restored_router_uses_workspace_proxy_for_logical_only_attachment() {
|
||||
let root = tempfile::tempdir().unwrap();
|
||||
let scope = manifest::SharedScope::new(manifest::Scope::writable(root.path()).unwrap());
|
||||
let client = WorkerWorkspaceContext::local_filesystem(Some(
|
||||
WorkspaceId::new("workspace-a").unwrap(),
|
||||
))
|
||||
.client_handle();
|
||||
let alias = WorkdirAttachmentAlias::new("checkout").unwrap();
|
||||
|
||||
let router = restored_workdir_router(
|
||||
&BTreeMap::new(),
|
||||
&[LogicalWorkdirAttachment {
|
||||
alias: alias.clone(),
|
||||
working_directory_id: "remote-workdir".to_string(),
|
||||
}],
|
||||
scope,
|
||||
client,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let session = router.session(&alias).unwrap();
|
||||
assert_eq!(session.workdir().id().as_str(), "remote-workdir");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn profile_factory_routes_workspace_requests_by_workspace_id() {
|
||||
let profiles = tempfile::tempdir().unwrap();
|
||||
@@ -3990,6 +4056,7 @@ mod tests {
|
||||
)),
|
||||
context: test_execution_context(worker_ref),
|
||||
previous_workdir_attachments: Vec::new(),
|
||||
logical_workdir_attachments: Vec::new(),
|
||||
workdir_attachments: BTreeMap::new(),
|
||||
config_bundle: None,
|
||||
})
|
||||
@@ -4080,6 +4147,7 @@ mod tests {
|
||||
workspace_scope: None,
|
||||
context: test_execution_context(worker_ref),
|
||||
previous_workdir_attachments: Vec::new(),
|
||||
logical_workdir_attachments: Vec::new(),
|
||||
workdir_attachments: BTreeMap::new(),
|
||||
config_bundle: None,
|
||||
})
|
||||
@@ -4550,47 +4618,6 @@ mod tests {
|
||||
&[("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();
|
||||
@@ -4598,22 +4625,6 @@ mod tests {
|
||||
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]
|
||||
|
||||
Reference in New Issue
Block a user