diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index d3304617e..729243ba6 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -518,9 +518,10 @@ pub struct WorkerRestorePreparation { pub workspace_api: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub workdir_attachments: Option>, + /// Retained in saved Restore intents; Restore does not apply Repository access. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub repository_access: Vec, - /// 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, } diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index f15d5ca4e..6c511ec97 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -432,6 +432,9 @@ pub struct WorkerExecutionRestoreRequest { pub previous_workdir_attachments: Vec, pub logical_workdir_attachments: Vec, pub workdir_attachments: BTreeMap, + /// Process-owned sessions that open providers only on a Workdir operation. + pub deferred_workdir_sessions: + BTreeMap, pub config_bundle: Option, } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index c9307a5dc..74093ce51 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -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, + restoring: bool, ) -> Result { 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 { - 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 { 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] diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 78c2380d1..a482902c6 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -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, logical_attachments: &[crate::catalog::LogicalWorkdirAttachment], + deferred_sessions: &BTreeMap, claims: &[crate::catalog::WorkingDirectoryAttachmentClaim], workspace_client: Arc, ) -> Result, 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, +) -> Vec { + bindings + .iter() + .map(|(alias, binding)| WorkingDirectoryAttachmentStatus { + alias: alias.clone(), + working_directory: binding.status(), + }) + .collect() +} + +fn deferred_restore_workdir_sessions( + request: &WorkerExecutionRestoreRequest, + materializer: Option>, +) -> Result, 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>>, controller_task: tokio::task::JoinHandle<()>, restore_operation_id: Option, - workdir_attachments: BTreeMap, + workdir_attachment_statuses: Vec, workspace_client: Option>, ) -> 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, restore_operation_id: Option, - workdir_attachments: BTreeMap, + workdir_attachment_statuses: Vec, workspace_client: Option>, ) -> 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::>(); 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>); + + 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 { + 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()); } } diff --git a/crates/worker-runtime/src/worker_backend/deferred_workdir.rs b/crates/worker-runtime/src/worker_backend/deferred_workdir.rs new file mode 100644 index 000000000..eebedde57 --- /dev/null +++ b/crates/worker-runtime/src/worker_backend/deferred_workdir.rs @@ -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 + Send + Sync; + +pub(super) struct DeferredWorkdirSession { + workdir: Workdir, + capabilities: WorkdirSessionCapabilities, + open: Arc, + state: Mutex, + events: broadcast::Sender, +} + +#[derive(Default)] +struct State { + session: Option, + event_bridge: Option>, + 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 + 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 { + 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> { + Some(self.events.subscribe()) + } + + fn command_snapshot(&self) -> Vec { + 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(); + } +} diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index fe8b6ca7c..a8ddde657 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -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> { - 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::>>() - .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; diff --git a/crates/workspace-server/src/server/tests/restore_workdir_tests.rs b/crates/workspace-server/src/server/tests/restore_workdir_tests.rs new file mode 100644 index 000000000..f11276a78 --- /dev/null +++ b/crates/workspace-server/src/server/tests/restore_workdir_tests.rs @@ -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(); +} diff --git a/crates/workspace-server/src/server/tests/worker_operations_tests.rs b/crates/workspace-server/src/server/tests/worker_operations_tests.rs index 574c48b8e..9febd5fb0 100644 --- a/crates/workspace-server/src/server/tests/worker_operations_tests.rs +++ b/crates/workspace-server/src/server/tests/worker_operations_tests.rs @@ -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 + ); + } }