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

fix: defer restored workdir access until use

Browse Source
This commit is contained in:
Hare
2026-10-20 22:00:58 +09:00
parent 3cfe04be3a
commit 338e39bcdb
8 changed files with 933 additions and 914 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/runtime-api/src/lib.rs
+2 -1
View File
@@ -518,9 +518,10 @@ pub struct WorkerRestorePreparation {
pub workspace_api: Option<WorkspaceApiRef>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workdir_attachments: Option<Vec<LogicalWorkdirAttachment>>,
/// Retained in saved Restore intents; Restore does not apply Repository access.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub repository_access: Vec<WorkingDirectoryRepositoryAccessRequest>,
/// Read-only references resolved by Workspace only after this operation owns admission.
/// Retained references from saved resource preparation, not Restore prerequisites.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub repository_access_workdirs: Vec<RepositoryAccessWorkdirReference>,
}
crates/worker-runtime/src/execution.rs
+3
View File
@@ -432,6 +432,9 @@ pub struct WorkerExecutionRestoreRequest {
pub previous_workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
pub logical_workdir_attachments: Vec<crate::catalog::LogicalWorkdirAttachment>,
pub workdir_attachments: BTreeMap<workdir::WorkdirAttachmentAlias, WorkingDirectoryBinding>,
/// Process-owned sessions that open providers only on a Workdir operation.
pub deferred_workdir_sessions:
BTreeMap<workdir::WorkdirAttachmentAlias, workdir::WorkdirSessionHandle>,
pub config_bundle: Option<ConfigBundle>,
}
crates/worker-runtime/src/runtime.rs
+106 -364
View File
@@ -1865,20 +1865,23 @@ impl Runtime {
if let Some(scope) = scope {
self.ensure_worker_in_workspace(scope, worker_ref)?;
}
self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments)
self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments, false)
}
fn replace_worker_workdir_attachments_under_lock(
&self,
worker_ref: &WorkerRef,
mut attachments: Vec<LogicalWorkdirAttachment>,
restoring: bool,
) -> Result<WorkerDetail, RuntimeError> {
validate_logical_workdir_attachments(&attachments)?;
attachments.sort_by(|left, right| left.alias.cmp(&right.alias));
let mut state = self.lock()?;
for attachment in &attachments {
state.ensure_workdir_not_removing(&attachment.working_directory_id)?;
if !restoring {
for attachment in &attachments {
state.ensure_workdir_not_removing(&attachment.working_directory_id)?;
}
}
let previous = state.worker(worker_ref)?.clone();
{
@@ -2146,7 +2149,6 @@ impl Runtime {
),
RuntimeError,
> {
let mut preparation_authorized = !coordinate && request.preparation.is_some();
let operation_lock = self.worker_operation_lock(worker_ref.worker_id)?;
let _operation_guard = operation_lock
.lock()
@@ -2191,7 +2193,7 @@ impl Runtime {
let conflict = || RuntimeError::RestoreObservationConflict {
worker_id: worker_ref.worker_id,
};
if let Some(mut owner) = candidate
if let Some(owner) = candidate
.restore_guard
.owners
.get(&request.request_id)
@@ -2204,13 +2206,6 @@ impl Runtime {
if let Some(receipt) = &owner.receipt {
return Ok((Some(receipt.clone()), None));
}
if let Some(preparation) = &owner.request.preparation {
for id in
preparation_workdir_ids(preparation, state.runtime_identity.as_deref())
{
state.ensure_workdir_not_removing(id)?;
}
}
if candidate.restore_guard.active_request_id.as_deref() != Some(&request.request_id)
|| candidate
.pending_restore
@@ -2218,151 +2213,6 @@ impl Runtime {
{
return Err(conflict());
}
if coordinate && !owner.preparation_applied {
if let Some(preparation) = owner.request.preparation.as_mut() {
if preparation.repository_access_workdirs.is_empty()
&& !preparation.repository_access.is_empty()
{
preparation.repository_access_workdirs = preparation
.repository_access
.iter()
.map(|access| runtime_api::RepositoryAccessWorkdirReference {
runtime_id: access.materialization.runtime_id.clone(),
working_directory_id: access.working_directory_id.clone(),
})
.collect();
candidate
.restore_guard
.owners
.insert(request.request_id.clone(), owner.clone());
match state.persist_worker_record(&candidate) {
Ok(()) => {
state
.workers
.insert(worker_ref.worker_id, candidate.clone());
}
Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
state.workers.insert(worker_ref.worker_id, candidate);
return Ok((
Some(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_preparation_commit_unknown",
sanitize_worker_create_failure_message(
&error.to_string(),
),
)),
None,
));
}
Err(error) => {
return Ok((
Some(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_preparation_commit_failed",
sanitize_worker_create_failure_message(
&error.to_string(),
),
)),
None,
));
}
}
}
}
}
if !owner.preparation_applied {
if let (Some(original), Some(incoming)) =
(&owner.request.preparation, &request.preparation)
{
// Server-regenerated bindings are not part of the public fingerprint.
// Reuse admitted bindings and accept only resource completion for those references.
let references_match = incoming.repository_access.len()
== original.repository_access_workdirs.len()
&& original.repository_access_workdirs.iter().all(|reference| {
incoming
.repository_access
.iter()
.filter(|access| {
reference.working_directory_id
== access.working_directory_id
&& reference.runtime_id
== access.materialization.runtime_id
})
.count()
== 1
});
let snapshot_matches = original.repository_access.is_empty()
|| (original.repository_access.len()
== incoming.repository_access.len()
&& original
.repository_access
.iter()
.zip(&incoming.repository_access)
.all(|(original, incoming)| {
let mut normalized = incoming.clone();
if let (Some(old), Some(new)) = (
&original.materialization.ssh,
&mut normalized.materialization.ssh,
) {
new.secret_resource = old.secret_resource.clone();
new.expires_at_epoch_seconds =
old.expires_at_epoch_seconds;
}
&normalized == original
}));
if !original.repository_access_workdirs.is_empty() {
preparation_authorized =
!coordinate && references_match && snapshot_matches;
}
if references_match
&& snapshot_matches
&& original.repository_access != incoming.repository_access
{
candidate
.restore_guard
.owners
.get_mut(&request.request_id)
.unwrap()
.request
.preparation
.as_mut()
.unwrap()
.repository_access = incoming.repository_access.clone();
match state.persist_worker_record(&candidate) {
Ok(()) => {
state
.workers
.insert(worker_ref.worker_id, candidate.clone());
}
Err(error @ RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
state.workers.insert(worker_ref.worker_id, candidate);
return Ok((
Some(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_preparation_commit_unknown",
sanitize_worker_create_failure_message(
&error.to_string(),
),
)),
None,
));
}
Err(error) => {
return Ok((
Some(RuntimeWorkerRestoreResult::failed(
WorkerRestoreState::ReconciliationRequired,
"worker_restore_preparation_commit_failed",
sanitize_worker_create_failure_message(
&error.to_string(),
),
)),
None,
));
}
}
}
}
}
} else {
// Startup/older persisted pending operations have a lifecycle
// owner but no public request receipt. Bind recovery to that
@@ -2417,13 +2267,6 @@ impl Runtime {
if recovering_pending {
request.preparation = None;
}
if let Some(preparation) = &request.preparation {
for id in
preparation_workdir_ids(preparation, state.runtime_identity.as_deref())
{
state.ensure_workdir_not_removing(id)?;
}
}
let pending = candidate
.pending_restore
.unwrap_or_else(|| PendingWorkerRestore {
@@ -2481,12 +2324,8 @@ impl Runtime {
return Ok((None, owner.request.preparation.clone()));
}
}
self.restore_admitted_worker_under_lock_with_preparation(
worker_ref,
WorkerRestoreMode::Explicit,
preparation_authorized,
)
.map(|result| (Some(result), None))
self.restore_admitted_worker_under_lock(worker_ref, WorkerRestoreMode::Explicit)
.map(|result| (Some(result), None))
}
#[cfg(test)]
@@ -2514,15 +2353,6 @@ impl Runtime {
&self,
worker_ref: &WorkerRef,
mode: WorkerRestoreMode,
) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
self.restore_admitted_worker_under_lock_with_preparation(worker_ref, mode, false)
}
fn restore_admitted_worker_under_lock_with_preparation(
&self,
worker_ref: &WorkerRef,
mode: WorkerRestoreMode,
preparation_authorized: bool,
) -> Result<RuntimeWorkerRestoreResult, RuntimeError> {
self.lock()?
.ensure_subject_not_reserved(worker_ref, false)?;
@@ -2587,15 +2417,7 @@ impl Runtime {
message,
)
};
if owner.request.preparation.as_ref().is_some_and(|prep| {
prep.repository_access.len() < prep.repository_access_workdirs.len()
|| (!prep.repository_access_workdirs.is_empty() && !preparation_authorized)
}) {
return Ok(uncertain(
"Admitted Restore is awaiting Workspace resource completion".into(),
));
}
// Complete idempotent binding/access effects under the original pending_restore ID.
// Apply reference updates under the original pending_restore ID; no provider is opened.
let mut effects_started = false;
if let Err(error) = self.apply_restore_preparation_under_lock(
worker_ref,
@@ -2736,42 +2558,10 @@ impl Runtime {
.map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
validate_logical_workdir_attachments(&attachments)?;
*effects_started = true;
self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments)?;
}
for access in preparation.repository_access {
if preparation
.repository_access_workdirs
.iter()
.any(|reference| {
reference.working_directory_id == access.working_directory_id
&& reference.runtime_id == access.materialization.runtime_id
})
{
// Workspace completed these at the Workdir owner Runtime after snapshot persistence.
// Never send cross-Runtime credentials to this Worker's Runtime.
continue;
}
let access: WorkingDirectoryRepositoryAccessRequest = serde_json::from_value(
serde_json::to_value(access)
.map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?,
)
.map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
*effects_started = true;
let runtime = self.clone();
// The public embedded interface is synchronous. A separate IO executor
// avoids nested Tokio block_on while retaining the lifecycle lock.
std::thread::spawn(move || {
let executor = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|error| RuntimeError::InvalidRequest(error.to_string()))?;
executor.block_on(
runtime.authorize_working_directory_repository_access_from_resource(access),
)
})
.join()
.map_err(|_| RuntimeError::StatePoisoned)??;
self.replace_worker_workdir_attachments_under_lock(worker_ref, attachments, true)?;
}
// Retained legacy SSH preparation is not a prerequisite or an effect
// of restoring a Worker. Sessions resolve live access when used.
Ok(())
}
@@ -2856,6 +2646,7 @@ impl Runtime {
previous_workdir_attachments: worker.workdir_attachments.clone(),
logical_workdir_attachments: worker.logical_workdir_attachments.clone(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
};
(
@@ -2868,28 +2659,6 @@ impl Runtime {
)
};
let _workdir_operation = self.reserve_workdir_use(
requested_workdir_ids(&request.request)
.into_iter()
.chain(
request
.previous_workdir_attachments
.iter()
.map(|attachment| {
attachment
.working_directory
.summary
.working_directory_id
.as_str()
}),
)
.chain(
request
.logical_workdir_attachments
.iter()
.map(|attachment| attachment.working_directory_id.as_str()),
),
)?;
if !reconcile && let Err(result) = backend.preflight_restore(&request) {
// A preflight may attest an already-connected Controller snapshot.
// Repeated restore is read-only; do not replace its operation owner.
@@ -8932,7 +8701,7 @@ mod tests {
}
#[test]
fn active_workdir_cleanup_rejects_same_id_admissions_without_blocking_other_workers() {
fn active_workdir_cleanup_blocks_access_but_allows_reference_only_restore() {
let (runtime, backend) = runtime_and_backend();
let owner = scope("workspace-a", "server-a");
let worker = runtime
@@ -8992,36 +8761,43 @@ mod tests {
.stop_worker_scoped(&owner, &worker.worker_ref, None)
.unwrap();
let restore = runtime.test_restore_request(&worker.worker_ref);
assert_workdir_busy(
runtime
.coordinate_worker_restore_operation(
&worker.worker_ref,
runtime_api::WorkerRestoreCoordinationRequest {
expected_observation_token: restore.expected_observation_token,
request_id: restore.request_id,
preparation: Some(runtime_api::WorkerRestorePreparation {
workdir_attachments: Some(vec![
runtime_api::LogicalWorkdirAttachment {
alias: attachment.alias,
working_directory_id: attachment.working_directory_id,
capabilities: attachment.capabilities,
},
]),
..Default::default()
}),
},
Some(&owner),
)
.unwrap_err(),
);
assert!(
let (result, preparation) = runtime
.coordinate_worker_restore_operation(
&worker.worker_ref,
runtime_api::WorkerRestoreCoordinationRequest {
expected_observation_token: restore.expected_observation_token.clone(),
request_id: restore.request_id.clone(),
preparation: Some(runtime_api::WorkerRestorePreparation {
workdir_attachments: Some(vec![runtime_api::LogicalWorkdirAttachment {
alias: attachment.alias.clone(),
working_directory_id: attachment.working_directory_id.clone(),
capabilities: attachment.capabilities,
}]),
..Default::default()
}),
},
Some(&owner),
)
.unwrap();
assert!(result.is_none());
assert!(preparation.is_some());
let restored = runtime
.restore_worker_operation_scoped(&owner, &worker.worker_ref, restore)
.unwrap();
assert_eq!(restored.state, WorkerRestoreState::Accepted);
assert_eq!(
runtime
.lock()
.unwrap()
.worker(&worker.worker_ref)
.unwrap()
.pending_restore
.is_none()
.logical_workdir_attachments,
vec![attachment]
);
assert_workdir_busy(
runtime
.open_workdir_session_scoped(&owner, id, None)
.unwrap_err(),
);
assert_eq!(runtime.list_workers_scoped(&owner).unwrap().len(), 1);
// Different IDs and Worker creation remain available during cleanup.
@@ -10783,102 +10559,68 @@ mod tests {
}
#[test]
fn restore_guard_resource_snapshot_requires_workspace_completion_before_execution() {
let (runtime, backend) = runtime_and_backend();
let scope = scope("workspace-a", "server-a");
let worker = runtime
.create_worker_scoped(
&scope,
scoped_task_request("resource coordination", "workspace-a"),
)
.unwrap();
runtime.stop_worker(&worker.worker_ref, None).unwrap();
let mut request = runtime.test_restore_request(&worker.worker_ref);
let mut preparation = runtime_api::WorkerRestorePreparation::default();
preparation.repository_access_workdirs.push(
runtime_api::RepositoryAccessWorkdirReference {
runtime_id: "peer-runtime".into(),
working_directory_id: "peer-workdir".into(),
},
);
let mut coordination = runtime_api::WorkerRestoreCoordinationRequest {
expected_observation_token: request.expected_observation_token.clone(),
request_id: request.request_id.clone(),
preparation: Some(preparation.clone()),
};
runtime
.coordinate_worker_restore_operation(
&worker.worker_ref,
coordination.clone(),
Some(&scope),
)
.unwrap();
let operation_id = runtime
.lock()
.unwrap()
.worker(&worker.worker_ref)
.unwrap()
.pending_restore
.unwrap()
.operation_id;
// Model a retained resource snapshot before the peer authorization callback completes.
preparation
.repository_access
.push(runtime_api::WorkingDirectoryRepositoryAccessRequest {
working_directory_id: "peer-workdir".into(),
materialization: runtime_api::RepositoryMaterializationContext {
workspace_id: "workspace-a".into(),
runtime_id: "peer-runtime".into(),
operation_id: format!("worker-restore:{}", request.request_id),
config_projection_digest: "snapshot".into(),
ssh: None,
fn retained_restore_resource_snapshot_does_not_require_workspace_completion() {
for completed in [false, true] {
let (runtime, backend) = runtime_and_backend();
let scope = scope("workspace-a", "server-a");
let worker = runtime
.create_worker_scoped(
&scope,
scoped_task_request("resource coordination", "workspace-a"),
)
.unwrap();
runtime.stop_worker(&worker.worker_ref, None).unwrap();
let mut request = runtime.test_restore_request(&worker.worker_ref);
let mut preparation = runtime_api::WorkerRestorePreparation::default();
preparation.repository_access_workdirs.push(
runtime_api::RepositoryAccessWorkdirReference {
runtime_id: "offline-peer".into(),
working_directory_id: "peer-workdir".into(),
},
});
coordination.preparation = Some(preparation.clone());
assert!(
runtime
.coordinate_worker_restore_operation(&worker.worker_ref, coordination, Some(&scope))
.unwrap()
.1
.is_some()
);
request.preparation = None;
assert_eq!(
runtime
);
if completed {
preparation.repository_access.push(
runtime_api::WorkingDirectoryRepositoryAccessRequest {
working_directory_id: "peer-workdir".into(),
materialization: runtime_api::RepositoryMaterializationContext {
workspace_id: "workspace-a".into(),
runtime_id: "offline-peer".into(),
operation_id: format!("worker-restore:{}", request.request_id),
config_projection_digest: "snapshot".into(),
ssh: None,
},
},
);
}
for include_resources in [false, completed] {
let mut admitted = preparation.clone();
if !include_resources {
admitted.repository_access.clear();
}
runtime
.coordinate_worker_restore_operation(
&worker.worker_ref,
runtime_api::WorkerRestoreCoordinationRequest {
expected_observation_token: request.expected_observation_token.clone(),
request_id: request.request_id.clone(),
preparation: Some(admitted),
},
Some(&scope),
)
.unwrap();
}
request.preparation = None;
let restored = runtime
.restore_worker_operation_scoped(&scope, &worker.worker_ref, request.clone())
.unwrap()
.state,
WorkerRestoreState::ReconciliationRequired
);
assert_eq!(
runtime
.restore_admitted_worker_under_lock(&worker.worker_ref, WorkerRestoreMode::Explicit)
.unwrap()
.state,
WorkerRestoreState::ReconciliationRequired
);
assert_eq!(*backend.restore_count.lock().unwrap(), 0);
request.preparation = Some(runtime_api::WorkerRestorePreparation::default());
assert_eq!(
runtime
.restore_worker_operation_scoped(&scope, &worker.worker_ref, request.clone())
.unwrap()
.state,
WorkerRestoreState::ReconciliationRequired
);
assert_eq!(*backend.restore_count.lock().unwrap(), 0);
request.preparation = Some(preparation);
assert_eq!(
runtime
.unwrap();
assert_eq!(restored.state, WorkerRestoreState::Accepted);
let replay = runtime
.restore_worker_operation_scoped(&scope, &worker.worker_ref, request)
.unwrap()
.state,
WorkerRestoreState::Accepted
);
assert_eq!(
backend.restore_operation_ids.lock().unwrap().as_slice(),
&[operation_id]
);
.unwrap();
assert_eq!(replay, restored);
assert_eq!(*backend.restore_count.lock().unwrap(), 1);
assert!(backend.repository_accesses.lock().unwrap().is_empty());
}
}
#[test]
crates/worker-runtime/src/worker_backend.rs
+321 -176
View File
@@ -7,7 +7,7 @@
//!
//! Opens Session and metadata stores within the aggregate through trusted WorkerRef, with no
//! process-global root fallback. Restore preserves identity while rebuilding Workspace connections
//! and Workdir bindings. Candidate controller observation registration waits for activation after
//! and deferred Workdir references. Candidate controller observation registration waits for activation after
//! catalog commit. Cleanup checks the actual controller and session resources, deleting only
//! operation-owned disposable run artifacts. It implies neither Session/metadata deletion nor
//! retention of all past runs.
@@ -15,6 +15,8 @@
//! WorkerHandle is owned internally; external operations take WorkerRef. Socket/session paths,
//! Manifests, credentials, and handles are not passed to browser/API projections.
mod deferred_workdir;
use std::collections::{BTreeMap, HashMap};
use std::future::Future;
use std::path::{Path, PathBuf};
@@ -1145,6 +1147,7 @@ fn runtime_local_workdir_router(
fn restored_workdir_router(
local_attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
logical_attachments: &[crate::catalog::LogicalWorkdirAttachment],
deferred_sessions: &BTreeMap<WorkdirAttachmentAlias, WorkdirSessionHandle>,
claims: &[crate::catalog::WorkingDirectoryAttachmentClaim],
workspace_client: Arc<dyn WorkspaceClient>,
) -> Result<Arc<WorkdirSessionRouter>, String> {
@@ -1167,12 +1170,33 @@ fn restored_workdir_router(
&local_capabilities,
Some(workspace_client.clone()),
)?;
for (alias, session) in deferred_sessions {
if let Some(attachment) = logical_attachments.iter().find(|item| &item.alias == alias) {
if session.workdir().id().as_str() != attachment.working_directory_id {
return Err(format!("conflicting restored Workdir attachment {alias}"));
}
// Workspace sessions reacquire live Repository access at operation time.
// A Runtime-local opener cannot renew Host-owned SSH credentials.
continue;
}
if local_attachments.contains_key(alias) {
return Err(format!("duplicate restored Workdir attachment {alias}"));
}
let source = worker::authorized_workdir::authorized_workdir_session(
session.clone(),
workspace_client.clone(),
session.workdir().id().to_string(),
);
router
.attach(alias.clone(), source)
.map_err(|error| error.to_string())?;
}
for attachment in logical_attachments {
if let Some(local) = local_attachments.get(&attachment.alias) {
if local.working_directory.id != attachment.working_directory_id {
if let Some(session) = router.session(&attachment.alias) {
if session.workdir().id().as_str() != attachment.working_directory_id {
return Err(format!(
"logical Workdir attachment `{}` conflicts with Runtime-local Workdir {}",
attachment.alias, local.working_directory.id
"conflicting restored Workdir attachment {}",
attachment.alias
));
}
continue;
@@ -1197,6 +1221,104 @@ fn restored_workdir_router(
Ok(router)
}
fn workdir_attachment_statuses(
bindings: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
) -> Vec<WorkingDirectoryAttachmentStatus> {
bindings
.iter()
.map(|(alias, binding)| WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
})
.collect()
}
fn deferred_restore_workdir_sessions(
request: &WorkerExecutionRestoreRequest,
materializer: Option<Arc<dyn WorkingDirectoryMaterializer>>,
) -> Result<BTreeMap<WorkdirAttachmentAlias, WorkdirSessionHandle>, String> {
let mut references = BTreeMap::new();
for previous in &request.previous_workdir_attachments {
if references
.insert(
previous.alias.clone(),
previous
.working_directory
.summary
.working_directory_id
.clone(),
)
.is_some()
{
return Err(format!(
"duplicate persisted Workdir alias `{}`",
previous.alias
));
}
}
for claim in &request.request.workdir_attachments {
if let Some(previous) = references.get(&claim.alias) {
if previous != &claim.working_directory_id {
return Err(format!(
"conflicting persisted Workdir alias `{}`",
claim.alias
));
}
} else {
references.insert(claim.alias.clone(), claim.working_directory_id.clone());
}
}
references
.into_iter()
.map(|(alias, workdir_id)| {
let capabilities = request
.logical_workdir_attachments
.iter()
.find(|attachment| attachment.alias == alias)
.map(|attachment| attachment.capabilities)
.unwrap_or_else(|| {
initial_workdir_session_capabilities(
&alias,
&request.request.workdir_attachments,
)
});
let relative_cwd = request
.request
.workdir_attachments
.iter()
.find(|claim| claim.alias == alias)
.and_then(|claim| claim.relative_cwd.clone());
let materializer = materializer.clone();
let id = workdir_id.clone();
let session = deferred_workdir::DeferredWorkdirSession::handle(
&workdir_id,
capabilities,
move || {
let materializer = materializer
.as_ref()
.ok_or_else(|| "Workdir provider is not configured".to_string())?;
let binding = materializer
.bind_working_directory(&id, relative_cwd.as_deref())
.map_err(|error| error.to_string())?;
let scope = runtime_attachment_scope(binding.root(), capabilities)?;
Ok(runtime_local_workdir_session(
&id,
binding.root(),
binding.cwd(),
scope,
capabilities,
binding.command_environment(),
binding
.session_resources()
.map_err(|error| error.to_string())?,
))
},
);
Ok((alias, session))
})
.collect()
}
fn validate_backend_job_profile(
manifest: &manifest::WorkerManifest,
request: &CreateWorkerRequest,
@@ -1838,6 +1960,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
let workdir_sessions = restored_workdir_router(
&request.workdir_attachments,
&request.logical_workdir_attachments,
&request.deferred_workdir_sessions,
&request.request.workdir_attachments,
worker.workspace_client_handle(),
)?;
@@ -2926,7 +3049,7 @@ where
shutdown: Arc<tokio::sync::Mutex<Option<worker::ShutdownReceiver>>>,
controller_task: tokio::task::JoinHandle<()>,
restore_operation_id: Option<crate::execution::WorkerLifecycleOperationId>,
workdir_attachments: BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
workdir_attachment_statuses: Vec<WorkingDirectoryAttachmentStatus>,
workspace_client: Option<Arc<dyn WorkspaceClient>>,
) -> WorkerExecutionSpawnResult {
self.connect_controller_scope(
@@ -2938,7 +3061,7 @@ where
RuntimeExecutionTaskScope::new(controller_task),
Arc::new(AtomicBool::new(false)),
restore_operation_id,
workdir_attachments,
workdir_attachment_statuses,
workspace_client,
)
}
@@ -2956,7 +3079,7 @@ where
tasks: RuntimeExecutionTaskScope,
shutdown_requested: Arc<AtomicBool>,
restore_operation_id: Option<crate::execution::WorkerLifecycleOperationId>,
workdir_attachments: BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
workdir_attachment_statuses: Vec<WorkingDirectoryAttachmentStatus>,
workspace_client: Option<Arc<dyn WorkspaceClient>>,
) -> WorkerExecutionSpawnResult {
#[cfg(feature = "ws-server")]
@@ -2969,15 +3092,6 @@ where
#[cfg(not(feature = "ws-server"))]
let worker_state_snapshot = handle.shared_state.snapshot();
let worker_state = Arc::new(RwLock::new(worker_state_snapshot));
let workdir_attachment_statuses = workdir_attachments
.iter()
.map(
|(alias, binding)| crate::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
},
)
.collect::<Vec<_>>();
let mut workers = self
.workers
.lock()
@@ -3862,7 +3976,7 @@ where
controller.shutdown,
controller.controller_task,
None,
BTreeMap::new(),
Vec::new(),
Some(controller.workspace_client),
);
}
@@ -3990,7 +4104,7 @@ where
controller.shutdown,
controller.controller_task,
None,
workdir_attachments,
workdir_attachment_statuses(&workdir_attachments),
Some(controller.workspace_client),
)
}
@@ -4064,54 +4178,6 @@ where
}
drop(workers);
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 attachments cannot be restored by this Runtime",
)
})?;
for previous in &request.previous_workdir_attachments {
let relative_cwd = request
.request
.workdir_attachments
.iter()
.find(|claim| claim.alias == previous.alias)
.and_then(|claim| claim.relative_cwd.as_deref());
// Recovery attachments remain CleanupPending across restart.
// Validate current bindability/identity, not the old state name.
// Preflight is read-only; actual restore binds again below.
materializer.preflight_bind_working_directory(
&previous.working_directory.summary.working_directory_id,
relative_cwd,
).map_err(|message| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!("Persisted Worker Workdir attachment `{}` is unavailable: {message}", previous.alias),
)
})?;
}
}
if request.previous_workdir_attachments.is_empty()
&& !request.request.workdir_attachment_requests.is_empty()
{
return Err(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"Persisted Worker Workdir allocations are unavailable",
));
}
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 claims cannot be restored by this Runtime",
));
}
if let Some(workspace_api) = request.request.workspace_api.as_ref() {
let Some(scope) = request.workspace_scope.as_ref() else {
return Err(WorkerExecutionResult::rejected(
@@ -4203,88 +4269,18 @@ where
}
drop(workers);
let mut workdir_attachments = BTreeMap::new();
if !request.previous_workdir_attachments.is_empty() {
let Some(materializer) = self.working_directory_materializer.as_ref() else {
request.deferred_workdir_sessions = match deferred_restore_workdir_sessions(
&request,
self.working_directory_materializer.clone(),
) {
Ok(sessions) => sessions,
Err(message) => {
return WorkerExecutionSpawnResult::Rejected(WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
"persisted Worker has Workdir attachments, but no materializer is configured for this Runtime backend",
message,
));
};
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(
&status.working_directory.summary.working_directory_id,
relative_cwd,
) {
Ok(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(
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
error.to_string(),
),
);
}
}
}
} 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 worker_ref = request.worker_ref.clone();
let operation_id = request.operation_id;
@@ -4424,7 +4420,7 @@ where
controller.shutdown,
controller.controller_task,
Some(operation_id),
pending.request.workdir_attachments.clone(),
pending.request.previous_workdir_attachments.clone(),
Some(controller.workspace_client),
);
if let Err(message) = self.remove_pending_restore(&worker_ref) {
@@ -5312,6 +5308,7 @@ mod tests {
previous_workdir_attachments: Vec::new(),
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
}
}
@@ -5991,7 +5988,7 @@ mod tests {
candidate.tasks.clone(),
candidate.shutdown_requested.clone(),
Some(request.operation_id),
BTreeMap::new(),
Vec::new(),
candidate.workspace_client.clone(),
)
};
@@ -6122,7 +6119,7 @@ mod tests {
candidate.tasks.clone(),
candidate.shutdown_requested.clone(),
Some(request.operation_id),
BTreeMap::new(),
Vec::new(),
candidate.workspace_client.clone(),
)
};
@@ -6494,6 +6491,67 @@ mod tests {
assert!(!root.path().join("nonexistent-current-profiles").exists());
}
#[test]
#[serial_test::serial(worker_allocation)]
fn profile_backend_restores_saved_workdir_reference_without_a_materializer() {
let root = tempfile::tempdir().unwrap();
let store = root.path().join("runtime");
let setup = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.unwrap();
let mut request = setup.block_on(saved_subjektiv_restore_fixture(&store, true));
let alias = WorkdirAttachmentAlias::new("offline").unwrap();
request
.request
.workdir_attachments
.push(WorkingDirectoryAttachmentClaim {
alias,
working_directory_id: "unavailable-checkout".into(),
relative_cwd: Some("unavailable-cwd".into()),
capabilities: WorkdirSessionCapabilities::READ_ONLY,
});
let backend = WorkerRuntimeExecutionBackend::new(
ProfileRuntimeWorkerFactory::new(root.path().join("missing-profiles"))
.with_runtime_store_dir(&store)
.with_runtime_id("runtime-subjektiv-test")
.with_controller_transport(WorkerControllerTransport::InProcess),
)
.unwrap();
backend.preflight_restore(&request).unwrap();
let restored = backend.restore_worker(request.clone());
assert!(
matches!(restored, WorkerExecutionSpawnResult::Connected { .. }),
"{restored:?}"
);
let session = backend.workers.lock().unwrap()[&request.worker_ref]
.handle
.shared_state
.fs_view()
.unwrap()
.session()
.clone();
assert_eq!(session.workdir().id().as_str(), "unavailable-checkout");
backend
.run_on_adapter_runtime(async move {
assert!(
session
.read(workdir::ReadRequest {
path: workdir::WorkdirPath::new("README.md").unwrap(),
offset: 0,
limit: 10,
max_bytes: 1024,
})
.await
.is_err()
);
Ok(())
})
.unwrap();
assert!(backend.stop_worker(&request.worker_ref).is_accepted());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial_test::serial(worker_allocation)]
async fn profile_factory_pending_subjektiv_restore_requires_operation_owned_launch_material() {
@@ -6893,6 +6951,7 @@ mod tests {
working_directory_id: "remote-workdir".to_string(),
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
}],
&BTreeMap::new(),
&[],
client,
)
@@ -6906,6 +6965,77 @@ mod tests {
);
}
#[tokio::test]
async fn restored_workspace_attachment_uses_host_operations_instead_of_local_opener() {
#[derive(Debug, Default)]
struct OfflineWorkspaceClient(Mutex<Vec<worker::WorkspaceRequest>>);
impl WorkspaceClient for OfflineWorkspaceClient {
fn workspace_id(&self) -> Option<&str> {
Some("workspace-a")
}
fn kind(&self) -> &str {
"offline-test"
}
fn is_available(&self) -> bool {
false
}
fn execute(
&self,
request: worker::WorkspaceRequest,
) -> Result<worker::WorkspaceResponse, worker::WorkspaceClientError> {
self.0.lock().unwrap().push(request);
Err(worker::WorkspaceClientError::Unavailable(
"provider offline".into(),
))
}
}
let client = Arc::new(OfflineWorkspaceClient::default());
let alias = WorkdirAttachmentAlias::new("checkout").unwrap();
let deferred = deferred_workdir::DeferredWorkdirSession::handle(
"saved-ssh-workdir",
WorkdirSessionCapabilities::READ_ONLY,
|| panic!("Host-owned Repository access must not use the local opener"),
);
let router = restored_workdir_router(
&BTreeMap::new(),
&[LogicalWorkdirAttachment {
alias: alias.clone(),
working_directory_id: "saved-ssh-workdir".into(),
capabilities: WorkdirSessionCapabilities::READ_ONLY,
}],
&BTreeMap::from([(alias.clone(), deferred)]),
&[],
client.clone(),
)
.unwrap();
assert!(client.0.lock().unwrap().is_empty());
let session = router.session(&alias).unwrap();
assert_eq!(session.workdir().id().as_str(), "saved-ssh-workdir");
assert_eq!(
session.capabilities(),
WorkdirSessionCapabilities::READ_ONLY
);
assert!(
session
.read(workdir::ReadRequest {
path: workdir::WorkdirPath::new("README.md").unwrap(),
offset: 0,
limit: 10,
max_bytes: 1024,
})
.await
.is_err()
);
let requests = client.0.lock().unwrap();
assert_eq!(requests.len(), 1);
assert_eq!(
requests[0].path,
"/api/w/workspace-a/workers/self/workdir-session/operations"
);
}
#[test]
fn runtime_attachment_scope_matches_session_capabilities() {
let root = tempfile::tempdir().unwrap();
@@ -6949,7 +7079,8 @@ mod tests {
// Workspace-bound client pretending to authorize operations.
let client = WorkerWorkspaceContext::no_workspace().client_handle();
let router = restored_workdir_router(&attachments, &logical, &[], client).unwrap();
let router =
restored_workdir_router(&attachments, &logical, &BTreeMap::new(), &[], client).unwrap();
let session = router.session(&alias).unwrap();
assert_eq!(
@@ -8418,6 +8549,7 @@ mod tests {
previous_workdir_attachments: Vec::new(),
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
}
}
@@ -8989,6 +9121,7 @@ mod tests {
previous_workdir_attachments: Vec::new(),
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
})
.await
@@ -9082,6 +9215,7 @@ mod tests {
previous_workdir_attachments: Vec::new(),
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
})
.await
@@ -9305,6 +9439,7 @@ mod tests {
previous_workdir_attachments: Vec::new(),
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
};
@@ -9693,6 +9828,7 @@ mod tests {
previous_workdir_attachments: detail.workdir_attachments,
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
deferred_workdir_sessions: BTreeMap::new(),
config_bundle: None,
};
drop(runtime);
@@ -9706,7 +9842,7 @@ mod tests {
#[cfg(target_os = "linux")]
#[test]
fn cleanup_pending_owner_restores_after_restart_with_validated_dirty_checkout() {
fn cleanup_pending_owner_restores_without_opening_dirty_checkout() {
let (root, backend, request, checkout, observed) = pending_owner_restore_fixture();
let before = persisted_files(&root.path().join("workdirs"));
backend.preflight_restore(&request).unwrap();
@@ -9731,10 +9867,7 @@ mod tests {
restored[0].working_directory.summary.cleanliness.as_deref(),
Some("dirty")
);
assert_eq!(
observed.lock().unwrap().as_slice(),
&[checkout.clone(), checkout.clone()]
);
assert_eq!(observed.lock().unwrap().as_slice(), &[checkout.clone()]);
assert_eq!(
fs::read_to_string(checkout.join("README.md")).unwrap(),
"owner recovery in progress"
@@ -9744,7 +9877,7 @@ mod tests {
#[cfg(target_os = "linux")]
#[test]
fn cleanup_pending_owner_restore_rejects_incomplete_replaced_and_raced_checkouts() {
fn damaged_checkout_does_not_block_restore_but_is_rejected_when_used() {
for damage in ["missing_git", "replacement", "missing_after_preflight"] {
let (root, backend, request, checkout, observed) = pending_owner_restore_fixture();
if damage == "missing_after_preflight" {
@@ -9757,36 +9890,48 @@ mod tests {
} else {
fs::remove_dir_all(checkout.join(".git")).unwrap();
}
let before = persisted_files(root.path());
let error = backend.preflight_restore(&request).unwrap_err();
assert_eq!(
error.outcome,
crate::execution::WorkerExecutionOutcome::Rejected,
"{damage}"
);
let before = persisted_files(&root.path().join("workdirs"));
backend.preflight_restore(&request).unwrap();
let sessions = deferred_restore_workdir_sessions(
&request,
backend.working_directory_materializer.clone(),
)
.unwrap();
let alias = WorkdirAttachmentAlias::new("checkout").unwrap();
let session = sessions[&alias].clone();
assert!(matches!(
backend.restore_worker(request.clone()),
WorkerExecutionSpawnResult::Connected { .. }
));
let error = backend
.run_on_adapter_runtime(async move {
let result = session
.read(workdir::ReadRequest {
path: workdir::WorkdirPath::new("README.md").unwrap(),
offset: 0,
limit: 10,
max_bytes: 1024,
})
.await;
session.close().await.unwrap();
Ok(result.unwrap_err().to_string())
})
.unwrap();
assert!(
error
.message_or_default()
.contains("working_directory_cleanup_"),
"{error:?}"
);
assert!(
matches!(
backend.restore_worker(request),
WorkerExecutionSpawnResult::Rejected(_)
),
"{damage}"
error.contains("working_directory_cleanup_"),
"{damage}: {error}"
);
assert_eq!(
observed.lock().unwrap().len(),
1,
"must not start a Controller on invalid checkout: {damage}"
"restore must not bind a cwd"
);
assert_eq!(
persisted_files(root.path()),
persisted_files(&root.path().join("workdirs")),
before,
"must preserve damaged/replaced checkout: {damage}"
"damaged checkout must be retained"
);
assert!(backend.stop_worker(&request.worker_ref).is_accepted());
}
}
crates/worker-runtime/src/worker_backend/deferred_workdir.rs
+256
View File
@@ -0,0 +1,256 @@
//! Restore retains attachment identity without opening a provider. Failed opens
//! are retryable at operation time; closing an unopened session never opens it.
use std::sync::Arc;
use async_trait::async_trait;
use tokio::sync::{Mutex, broadcast};
use workdir::*;
type OpenSession = dyn Fn() -> Result<WorkdirSessionHandle, String> + Send + Sync;
pub(super) struct DeferredWorkdirSession {
workdir: Workdir,
capabilities: WorkdirSessionCapabilities,
open: Arc<OpenSession>,
state: Mutex<State>,
events: broadcast::Sender<CommandEvent>,
}
#[derive(Default)]
struct State {
session: Option<WorkdirSessionHandle>,
event_bridge: Option<tokio::task::JoinHandle<()>>,
closed: bool,
}
impl Drop for State {
fn drop(&mut self) {
if let Some(task) = &self.event_bridge {
task.abort();
}
}
}
impl std::fmt::Debug for DeferredWorkdirSession {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DeferredWorkdirSession")
.field("workdir", &self.workdir)
.finish_non_exhaustive()
}
}
impl DeferredWorkdirSession {
pub(super) fn handle(
workdir_id: &str,
capabilities: WorkdirSessionCapabilities,
open: impl Fn() -> Result<WorkdirSessionHandle, String> + Send + Sync + 'static,
) -> WorkdirSessionHandle {
Arc::new(Self {
workdir: Workdir::new(workdir_id),
capabilities,
open: Arc::new(open),
state: Mutex::new(State::default()),
events: broadcast::channel(1024).0,
})
}
async fn session(&self) -> Result<WorkdirSessionHandle, WorkdirError> {
let mut state = self.state.lock().await;
if state.closed {
return Err(WorkdirError::SessionClosed);
}
if let Some(session) = &state.session {
return Ok(session.clone());
}
let open = self.open.clone();
let session = tokio::task::spawn_blocking(move || open())
.await
.map_err(|_| WorkdirError::Unavailable("Workdir session opener failed".into()))?
.map_err(WorkdirError::Unavailable)?;
if session.workdir().id() != self.workdir.id() {
session.close().await?;
return Err(WorkdirError::Unavailable(
"Workdir session identity mismatch".into(),
));
}
if let Some(mut source) = session.subscribe_command_events() {
let events = self.events.clone();
state.event_bridge = Some(tokio::spawn(async move {
loop {
match source.recv().await {
Ok(event) => {
let _ = events.send(event);
}
Err(broadcast::error::RecvError::Lagged(_)) => continue,
Err(broadcast::error::RecvError::Closed) => break,
}
}
}));
}
state.session = Some(session.clone());
Ok(session)
}
}
macro_rules! impl_deferred_session {
($( $method:ident($request:ident: $request_type:ty) -> $result:ty; )*) => {
#[async_trait]
impl WorkdirSession for DeferredWorkdirSession {
fn workdir(&self) -> &Workdir {
&self.workdir
}
fn capabilities(&self) -> WorkdirSessionCapabilities {
self.capabilities
}
$(
async fn $method(&self, $request: $request_type) -> Result<$result, WorkdirError> {
self.session().await?.$method($request).await
}
)*
fn subscribe_command_events(&self) -> Option<broadcast::Receiver<CommandEvent>> {
Some(self.events.subscribe())
}
fn command_snapshot(&self) -> Vec<CommandSnapshot> {
self.state.try_lock().ok().and_then(|state| {
state.session.as_ref().map(|session| session.command_snapshot())
}).unwrap_or_default()
}
async fn close(&self) -> Result<(), WorkdirError> {
let mut state = self.state.lock().await;
state.closed = true;
if let Some(session) = &state.session {
session.close().await?;
}
if let Some(task) = state.event_bridge.take() {
task.abort();
let _ = task.await;
}
Ok(())
}
}
};
}
impl_deferred_session! {
authorize_scope_path(request: WorkdirScopeAuthorizationRequest) -> ();
scope_rules_overlap(request: WorkdirScopeOverlapRequest) -> bool;
checkout_search(request: CheckoutSearchRequest) -> CheckoutSearchResult;
checkout_observe(path: WorkdirPath) -> CheckoutObservation;
checkout_execute(request: CheckoutRequest) -> CheckoutResult;
stat(request: StatRequest) -> StatResult;
read(request: ReadRequest) -> ReadResult;
read_bytes(request: ReadBytesRequest) -> ReadBytesResult;
write(request: WriteRequest) -> WriteResult;
edit(request: EditRequest) -> EditResult;
list(request: ListRequest) -> ListResult;
glob(request: GlobRequest) -> GlobResult;
grep(request: GrepRequest) -> GrepResult;
start_command(request: CommandRequest) -> CommandHandle;
command_status(handle: CommandHandle) -> CommandStatus;
command_output(request: CommandOutputRequest) -> CommandOutput;
cancel_command(handle: CommandHandle) -> ();
}
#[cfg(test)]
mod tests {
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
fn read_request() -> ReadRequest {
ReadRequest {
path: WorkdirPath::new("README.md").unwrap(),
offset: 0,
limit: 10,
max_bytes: 1024,
}
}
#[tokio::test]
async fn unopened_session_close_does_not_connect_and_is_terminal() {
let attempts = Arc::new(AtomicUsize::new(0));
let observed = attempts.clone();
let session = DeferredWorkdirSession::handle(
"offline",
WorkdirSessionCapabilities::READ_ONLY,
move || {
observed.fetch_add(1, Ordering::SeqCst);
Err("offline".into())
},
);
assert_eq!(session.workdir().id().as_str(), "offline");
assert!(session.subscribe_command_events().is_some());
session.close().await.unwrap();
session.close().await.unwrap();
assert!(matches!(
session.read(read_request()).await,
Err(WorkdirError::SessionClosed)
));
assert_eq!(attempts.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn failed_open_retries_on_use_and_keeps_read_only_capabilities() {
let temp = tempfile::tempdir().unwrap();
std::fs::write(temp.path().join("README.md"), "restored").unwrap();
let root = temp.path().to_path_buf();
let attempts = Arc::new(AtomicUsize::new(0));
let observed = attempts.clone();
let session = DeferredWorkdirSession::handle(
"saved",
WorkdirSessionCapabilities::READ_ONLY,
move || {
if observed.fetch_add(1, Ordering::SeqCst) == 0 {
return Err("offline".into());
}
let scope = super::super::runtime_attachment_scope(
&root,
WorkdirSessionCapabilities::READ_ONLY,
)?;
Ok(super::super::runtime_local_workdir_session(
"saved",
&root,
&root,
scope,
WorkdirSessionCapabilities::READ_ONLY,
Default::default(),
Vec::new(),
))
},
);
assert_eq!(attempts.load(Ordering::SeqCst), 0);
assert!(matches!(
session.read(read_request()).await,
Err(WorkdirError::Unavailable(_))
));
assert_eq!(
session.read(read_request()).await.unwrap().bytes,
b"restored"
);
assert_eq!(
session.read(read_request()).await.unwrap().bytes,
b"restored"
);
assert_eq!(attempts.load(Ordering::SeqCst), 2);
assert!(
session
.write(WriteRequest {
path: WorkdirPath::new("README.md").unwrap(),
content: b"forbidden".to_vec(),
expected_hash: None
})
.await
.is_err()
);
assert_eq!(
std::fs::read(temp.path().join("README.md")).unwrap(),
b"restored"
);
session.close().await.unwrap();
}
}
crates/workspace-server/src/server.rs
+28 -242
View File
@@ -6079,12 +6079,11 @@ impl WorkspaceApi {
/// Binds observation-based Restore intent to a Runtime-owned operation.
///
/// First queries existing owners/receipts without preparation, handling replay and recovery
/// before new-launch eligibility. New intent rereads singleton, Workdir, and External grants
/// to build references. SSH resources are issued only after Runtime admits the same observation
/// token under its operation lock. Replay does not replace saved preparation; incomplete auth
/// preparation remains ReconciliationRequired. Runtime owns lifecycle settlement, while the
/// returned summary comes from current observation.
/// Existing owners reach receipts/recovery before new-launch eligibility.
/// Restore retains attachment references without opening providers, resolving
/// Repository credentials, or requiring materialization. Workdir operations
/// validate current authority and availability when a session is used.
/// Runtime owns lifecycle settlement; summaries come from current observation.
fn restore_workspace_worker(
&self,
worker: &RuntimeWorkerRef,
@@ -6117,82 +6116,11 @@ impl WorkspaceApi {
result.worker = self.current_restore_summary(worker, result.worker.is_some());
return Ok(result);
}
let mut preparation = if let Some(preparation) = owned_preparation {
let preparation = if let Some(preparation) = owned_preparation {
preparation
} else {
self.store
.require_current_worker_singleton_owner(&self.config.workspace_id, worker)?;
let mut repository_access_workdirs = Vec::new();
for link in self
.store
.list_worker_workdir_links(&self.config.workspace_id, worker)?
.into_iter()
.filter(|link| link.unlinked_at.is_none())
{
let record = self
.store
.get_workdir_registry(&self.config.workspace_id, &link.workdir_id)?
.ok_or_else(|| Error::RuntimeOperationFailed {
runtime_id: "workspace-backend".to_string(),
code: "working_directory_not_found".to_string(),
message: format!("Unknown Workdir `{}`", link.workdir_id),
})?;
match record.source {
WorkdirRegistrySource::Repository {
runtime_id,
repository_id,
} => {
if self
.require_configured_workspace_repository(&repository_id)?
.source
.kind
== server_api::RepositorySourceKind::Ssh
{
repository_access_workdirs.push(
runtime_api::RepositoryAccessWorkdirReference {
runtime_id,
working_directory_id: link.workdir_id.clone(),
},
);
}
}
WorkdirRegistrySource::WorkspaceConfig { .. } => continue,
WorkdirRegistrySource::ExternalGrant { grant_id } => {
let grant = self
.store
.get_external_workdir_grant(&self.config.workspace_id, &grant_id)?
.ok_or_else(|| Error::RuntimeOperationFailed {
runtime_id: "external-workdir-provider".to_string(),
code: "external_workdir_unavailable".to_string(),
message: "External Workdir grant is unavailable".to_string(),
})?;
let grant_expired = external_workdir_grant_is_expired(&grant)?;
let connection = self
.external_workdir_providers
.lock()
.expect("External Workdir provider registry poisoned")
.connection(&grant_id);
if grant.status != "online"
|| grant_expired
|| !connection.is_some_and(|connection| {
connection.generation == grant.generation
&& connection.provider_instance_id == grant.provider_instance_id
&& connection.workdir_id == link.workdir_id
&& connection
.expires_at
.is_none_or(|expires_at| expires_at > Utc::now())
})
{
return Err(Error::RuntimeOperationFailed {
runtime_id: "external-workdir-provider".to_string(),
code: "external_workdir_unavailable".to_string(),
message: "External Workdir provider is offline".to_string(),
}
.into());
}
}
}
}
let preparation = runtime_api::WorkerRestorePreparation {
workspace_api: Some(runtime_contract_restore(
self.workspace_api_ref(&worker.runtime_id),
@@ -6201,7 +6129,7 @@ impl WorkspaceApi {
restore_logical_workdir_attachments(self, worker)?,
)?),
repository_access: Vec::new(),
repository_access_workdirs,
repository_access_workdirs: Vec::new(),
};
let (result, preparation) = self
.runtime
@@ -6222,60 +6150,6 @@ impl WorkspaceApi {
Error::Config("Runtime did not retain admitted Restore preparation".into())
})?
};
// Only the admitted operation may issue SSH handles. Reissue from the pinned
// credential snapshot after interruption; never renew candidate bindings.
let completion =
complete_restore_repository_access(self, &request.request_id, &mut preparation);
if let Err(error) = completion {
return Ok(InternalWorkerRestoreResult {
state: server_api::WorkerRestoreState::ReconciliationRequired,
worker: None,
diagnostics: vec![RuntimeDiagnostic::new(
"worker_restore_preparation_pending",
"warning",
sanitize_backend_error(&error.error.to_string()),
)],
});
}
// Persist the first resource snapshot at the admitted owner BEFORE any
// Workdir Runtime authorization effect. Replays renew only its opaque handles.
let (result, retained) = self
.runtime
.coordinate_worker_restore(
worker,
runtime_api::WorkerRestoreCoordinationRequest {
expected_observation_token: request.expected_observation_token.clone(),
request_id: request.request_id.clone(),
preparation: Some(preparation),
},
)
.map_err(RuntimeRegistryError::into_error)?;
if let Some(mut result) = result {
result.worker = self.current_restore_summary(worker, result.worker.is_some());
return Ok(result);
}
let preparation = retained.ok_or_else(|| {
Error::Config("Runtime did not retain Restore resource snapshot".into())
})?;
for access in &preparation.repository_access {
let access: worker_runtime::catalog::WorkingDirectoryRepositoryAccessRequest =
runtime_contract_restore(access.clone())?;
let runtime_id = access.materialization.runtime_id.clone();
if let Err(error) = self
.runtime
.authorize_working_directory_repository_access(&runtime_id, access)
{
return Ok(InternalWorkerRestoreResult {
state: server_api::WorkerRestoreState::ReconciliationRequired,
worker: None,
diagnostics: vec![RuntimeDiagnostic::new(
"worker_restore_repository_access_pending",
"warning",
sanitize_backend_error(&error.message()),
)],
});
}
}
let mut restored = self
.runtime
.restore_worker(
@@ -20315,44 +20189,30 @@ fn restore_logical_workdir_attachments(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
) -> ApiResult<Vec<LogicalWorkdirAttachment>> {
api.store
let mut attachments = Vec::new();
for link in api
.store
.list_worker_workdir_links(&api.config.workspace_id, worker)?
.into_iter()
.filter(|link| link.unlinked_at.is_none())
.filter(|link| {
api.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)
.ok()
.flatten()
.is_none_or(|record| {
!matches!(record.source, WorkdirRegistrySource::WorkspaceConfig { .. })
})
})
.map(|link| {
let workdir = api
.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Worker attachment references unknown Workdir {}",
link.workdir_id
))
})?;
let capabilities = link
.capabilities
.intersection(workdir_source_capabilities(api, &workdir)?)
.intersection(ticket_target_capabilities_for_workdir(
api, worker, &workdir,
)?);
Ok(LogicalWorkdirAttachment {
alias: workdir::WorkdirAttachmentAlias::new(link.alias)
.map_err(|error| Error::InvalidInput(error.to_string()))?,
working_directory_id: link.workdir_id,
capabilities,
{
if api
.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)?
.is_some_and(|record| {
matches!(record.source, WorkdirRegistrySource::WorkspaceConfig { .. })
})
})
.collect::<Result<Vec<_>>>()
.map_err(ApiError::from)
{
continue;
}
attachments.push(LogicalWorkdirAttachment {
alias: workdir::WorkdirAttachmentAlias::new(link.alias)
.map_err(|error| Error::InvalidInput(error.to_string()))?,
working_directory_id: link.workdir_id,
capabilities: link.capabilities,
});
}
Ok(attachments)
}
fn prepare_runtime_worker_workdir_attachments(
@@ -36648,81 +36508,6 @@ fn authorize_repository_materialization(
Ok(())
}
fn complete_restore_repository_access(
api: &WorkspaceApi,
request_id: &str,
preparation: &mut runtime_api::WorkerRestorePreparation,
) -> ApiResult<()> {
if preparation.repository_access.is_empty() {
for reference in &preparation.repository_access_workdirs {
if let Some(access) = repository_access_request_for_workdir(
api,
&reference.runtime_id,
&reference.working_directory_id,
&format!("worker-restore:{request_id}"),
)? {
preparation
.repository_access
.push(runtime_contract_restore(access)?);
}
}
return Ok(());
}
for prepared in &mut preparation.repository_access {
let mut access: worker_runtime::catalog::WorkingDirectoryRepositoryAccessRequest =
runtime_contract_restore(prepared.clone())?;
let Some(ssh) = access.materialization.ssh.as_mut() else {
continue;
};
let mut leases = Vec::new();
for candidate in &ssh.credential_candidates {
leases.push(
api.repository_secrets
.lease_ssh_materialization_access_fingerprints(
api.workspace_id(),
&candidate.credential_id,
&candidate.public_key_fingerprint,
&ssh.host_trust_id,
&ssh.host_key_fingerprint,
)?,
);
}
let known_hosts_entry = leases
.first()
.ok_or_else(|| Error::Config("Admitted SSH snapshot has no candidates".into()))?
.known_hosts_entry
.clone();
ssh.expires_at_epoch_seconds = repository_access_expiry();
ssh.secret_resource = api
.resource_broker
.issue_repository_ssh_access_handle(
api.workspace_id(),
&access.materialization.runtime_id,
format!(
"repository-ssh-access:{}",
access.materialization.operation_id
),
i64::try_from(ssh.expires_at_epoch_seconds).unwrap_or(i64::MAX),
worker_runtime::resource::RepositorySshAccessSecret {
credential_candidates: leases
.into_iter()
.map(
|lease| worker_runtime::resource::RepositorySshAccessSecretCandidate {
credential_id: lease.credential_id,
public_key_fingerprint: lease.credential_fingerprint.clone(),
private_key: lease.private_key.as_str().to_owned(),
},
)
.collect(),
known_hosts_entry,
},
)
.map_err(|error| Error::Config(error.to_string()))?;
*prepared = runtime_contract_restore(access)?;
}
Ok(())
}
fn repository_access_request_for_workdir(
api: &WorkspaceApi,
runtime_id: &str,
@@ -37544,6 +37329,7 @@ mod tests {
mod embedded_owned_execution_tests;
mod external_checkout_tests;
mod protocol_backpressure_tests;
mod restore_workdir_tests;
mod restored_parent_control_tests;
mod subject_authorization_tests;
mod subject_spawn_tests;
crates/workspace-server/src/server/tests/restore_workdir_tests.rs
+194
View File
@@ -0,0 +1,194 @@
//! Restore entry -> real SQLite attachment registry -> embedded Runtime restore
//! coordination. Only execution is deterministic; Workdir providers are absent.
//! This suite covers restore preparation, not HTTP authentication or model turns.
use super::*;
#[tokio::test]
async fn restore_keeps_offline_external_and_unmaterialized_repository_attachments() {
let temp = tempfile::tempdir().unwrap();
init_clean_git_workspace(temp.path());
let (api, execution) = test_api_with_recording_backend(temp.path()).await;
execution.accept_restores();
let Json(created) = create_workspace_worker(
State(api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(),
display_name: "Restore without providers".into(),
singleton_key: None,
profile: Some("builtin:coder".into()),
ticket_assignment: None,
initial_submit: Vec::new(),
workdir_attachments: Vec::new(),
feature_connections: Default::default(),
control_operation_id: None,
execution_grant: None,
}),
)
.await
.unwrap();
let worker = RuntimeWorkerRef::new(&created.runtime_id, &created.worker_id);
let grant = ExternalWorkdirGrantRecord {
grant_id: "restore-external-grant".into(),
workspace_id: TEST_WORKSPACE_ID.into(),
workdir_id: "restore-external".into(),
provider_instance_id: "offline-provider".into(),
display_name: "Offline external".into(),
permissions: "read_only".into(),
created_by: test_browser_request_actor().account_id,
created_at: TEST_CREATED_AT.into(),
expires_at: None,
generation: 1,
status: "pending".into(),
updated_at: TEST_CREATED_AT.into(),
};
for (alias, workdir_id, source) in [
(
"external",
grant.workdir_id.as_str(),
WorkdirRegistrySource::ExternalGrant {
grant_id: grant.grant_id.clone(),
},
),
(
"repository",
"restore-missing-repository",
WorkdirRegistrySource::Repository {
runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(),
repository_id: test_repository_id(&api),
},
),
] {
let record = WorkdirRegistryRecord {
workspace_id: TEST_WORKSPACE_ID.into(),
workdir_id: workdir_id.into(),
display_name: Some(alias.into()),
source,
creation_selector: None,
creation_ref: None,
creation_tree: None,
current_selector: None,
current_ref: None,
current_tree: None,
observed_at_epoch_seconds: None,
materialization_status: "not_found".into(),
cleanliness: "unknown".into(),
created_at: TEST_CREATED_AT.into(),
updated_at: TEST_CREATED_AT.into(),
};
if alias == "external" {
api.store
.create_external_workdir_grant(&grant, &record)
.unwrap();
assert!(
api.store
.activate_external_workdir_grant(
TEST_WORKSPACE_ID,
&grant.grant_id,
&grant.provider_instance_id,
1,
TEST_CREATED_AT,
)
.unwrap()
);
} else {
api.store.upsert_workdir_registry(&record).unwrap();
}
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.into(),
worker: worker.clone(),
workdir_id: workdir_id.into(),
alias: alias.into(),
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
linked_at: TEST_CREATED_AT.into(),
unlinked_at: None,
})
.unwrap();
}
let links = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap();
api.runtime
.stop_worker(
&worker,
WorkerLifecycleRequest {
reason: None,
ticket_assignment: None,
},
)
.unwrap();
let result = api
.restore_workspace_worker(
&worker,
server_api::WorkerRestoreRequest {
expected_observation_token: api
.runtime
.worker(&worker)
.unwrap()
.restore_observation_token
.unwrap(),
request_id: uuid::Uuid::now_v7().to_string(),
},
)
.unwrap();
assert_eq!(result.state, server_api::WorkerRestoreState::Accepted);
let attachments = restore_logical_workdir_attachments(&api, &worker).unwrap();
assert_eq!(attachments.len(), 2);
assert!(
attachments
.iter()
.any(|attachment| attachment.alias.as_str() == "external"
&& attachment.working_directory_id == grant.workdir_id)
);
assert!(
attachments
.iter()
.any(|attachment| attachment.alias.as_str() == "repository"
&& attachment.working_directory_id == "restore-missing-repository")
);
assert_eq!(
api.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap(),
links
);
let lock = current_worker_session_lock(&api, &worker);
let guard = Arc::new(lock.lock_owned().await);
for link in &links {
let error = open_current_worker_workdir_session_locked(&api, &worker, link, &guard)
.await
.unwrap_err();
match link.alias.as_str() {
"external" => assert!(matches!(
error,
Error::RuntimeOperationFailed { code, .. }
if code == "external_workdir_provider_offline"
)),
"repository" => assert!(matches!(
error,
Error::WorkdirSessionOpenFailed { transport, .. }
if transport.code == workdir::http::WorkdirTransportErrorCode::Unavailable
)),
alias => panic!("unexpected attachment: {alias}"),
}
}
assert_eq!(
api.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap(),
links
);
api.runtime
.stop_worker(
&worker,
WorkerLifecycleRequest {
reason: None,
ticket_assignment: None,
},
)
.unwrap();
}
crates/workspace-server/src/server/tests/worker_operations_tests.rs
+23 -131
View File
@@ -1593,7 +1593,7 @@ async fn restore_guard_workspace_interrupted_preparation_recovers_admitted_snaps
}
#[tokio::test]
async fn restore_guard_cross_runtime_ssh_completion_retries_pinned_snapshot_at_workdir_owner() {
async fn restore_keeps_cross_runtime_ssh_attachment_without_issuing_repository_access() {
let workspace = tempfile::tempdir().unwrap();
let mut config = test_server_config(workspace.path());
let source = server_api::RepositorySource {
@@ -1619,39 +1619,6 @@ async fn restore_guard_cross_runtime_ssh_completion_retries_pinned_snapshot_at_w
.stop(&WorkerOperationContext::Backend, operation_lifecycle())
.await
.unwrap();
let original_key = api
.repository_secrets
.generate_credential(
api.workspace_id(),
GenerateRepositorySshCredentialRequest {
operation_id: "cross-runtime-default-key".into(),
credential_id:
crate::repository_access::WORKSPACE_DEFAULT_REPOSITORY_SSH_CREDENTIAL_ID.into(),
name: "Restore default".into(),
},
"owner-account",
)
.unwrap();
let public_key = api
.repository_secrets
.credential_public_key(api.workspace_id(), &original_key.credential_id)
.unwrap()
.unwrap()
.public_key;
api.repository_secrets
.put_host_trust(
api.workspace_id(),
PutRepositorySshHostTrustRequest {
operation_id: "cross-runtime-trust".into(),
host_trust_id: "cross-runtime-host".into(),
hostname: "example.invalid".into(),
port: 22,
host_key: public_key,
expected_fingerprint: None,
},
"owner-account",
)
.unwrap();
let record = restore_test_workdir(
&api,
"cross-runtime-restore-workdir",
@@ -1663,102 +1630,27 @@ async fn restore_guard_cross_runtime_ssh_completion_retries_pinned_snapshot_at_w
peer.set_workdir_repository(&record.workdir_id, &test_repository_id(&api));
api.store.upsert_workdir_registry(&record).unwrap();
attach_restore_test_workdir(&api, &identity, &record);
let links = api
.store
.list_worker_workdir_links(api.workspace_id(), &identity)
.unwrap();
let request = restore_request_for(&api, &identity, "cross-runtime-restore-request");
assert_eq!(
handle
.restore(&WorkerOperationContext::Backend, request.clone())
.await
.unwrap()
.state,
server_api::WorkerRestoreState::ReconciliationRequired
);
assert!(
execution.restore_operations.lock().unwrap().is_empty(),
"credential completion must precede execution"
);
let first = peer.repository_access_requests.lock().unwrap()[0].clone();
assert_eq!(
first.materialization.runtime_id,
WorkdirlessFixtureRuntime::RUNTIME_ID
);
assert_ne!(first.materialization.runtime_id, identity.runtime_id);
// Rotate current host trust while the accepted operation is uncertain. Its
// original credential/trust fingerprints, source and binding must remain pinned.
let alternate = api
.repository_secrets
.generate_credential(
api.workspace_id(),
GenerateRepositorySshCredentialRequest {
operation_id: "cross-runtime-alternate-key".into(),
credential_id: "cross-runtime-alternate".into(),
name: "Alternate".into(),
},
"owner-account",
)
.unwrap();
let alternate_public = api
.repository_secrets
.credential_public_key(api.workspace_id(), &alternate.credential_id)
.unwrap()
.unwrap()
.public_key;
let original_trust_fingerprint = first
.materialization
.ssh
.as_ref()
.unwrap()
.host_key_fingerprint
.clone();
api.repository_secrets
.put_host_trust(
api.workspace_id(),
PutRepositorySshHostTrustRequest {
operation_id: "cross-runtime-trust-rotate".into(),
host_trust_id: "cross-runtime-host".into(),
hostname: "example.invalid".into(),
port: 22,
host_key: alternate_public,
expected_fingerprint: Some(original_trust_fingerprint),
},
"owner-account",
)
.unwrap();
*peer.reject_repository_access.lock().unwrap() = false;
assert_eq!(
handle
.restore(&WorkerOperationContext::Backend, request.clone())
.await
.unwrap()
.state,
server_api::WorkerRestoreState::Accepted
);
let attempts = peer.repository_access_requests.lock().unwrap();
assert_eq!(attempts.len(), 2);
let second = &attempts[1];
assert_eq!(
second.materialization.operation_id,
first.materialization.operation_id
);
let mut normalized = second.clone();
let old_ssh = first.materialization.ssh.as_ref().unwrap();
let new_ssh = normalized.materialization.ssh.as_mut().unwrap();
assert_ne!(new_ssh.secret_resource.nonce, old_ssh.secret_resource.nonce);
new_ssh.secret_resource = old_ssh.secret_resource.clone();
new_ssh.expires_at_epoch_seconds = old_ssh.expires_at_epoch_seconds;
assert_eq!(normalized, first, "only opaque handle and expiry may renew");
drop(attempts);
assert_eq!(execution.restore_operations.lock().unwrap().len(), 1);
assert_eq!(
handle
.restore(&WorkerOperationContext::Backend, request)
.await
.unwrap()
.state,
server_api::WorkerRestoreState::Accepted
);
assert_eq!(
peer.repository_access_requests.lock().unwrap().len(),
2,
"terminal receipt must not renew access"
);
for _ in 0..2 {
assert_eq!(
handle
.restore(&WorkerOperationContext::Backend, request.clone())
.await
.unwrap()
.state,
server_api::WorkerRestoreState::Accepted
);
assert!(peer.repository_access_requests.lock().unwrap().is_empty());
assert_eq!(execution.restore_operations.lock().unwrap().len(), 1);
assert_eq!(
api.store
.list_worker_workdir_links(api.workspace_id(), &identity)
.unwrap(),
links
);
}
}