From dc8e586a0bc505a234933844b0aa2a71cf922208 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 10 Oct 2026 12:04:56 +0900 Subject: [PATCH] fix: keep runtime responsive during workdir cleanup --- crates/workdir/src/local.rs | 174 ++++- crates/worker-runtime/src/http_server.rs | 430 ++++++++++- crates/worker-runtime/src/runtime.rs | 701 +++++++++++++++++- .../src/runtime/workdir_session.rs | 165 +++++ crates/worker-runtime/src/worker_backend.rs | 42 +- .../worker-runtime/src/working_directory.rs | 369 ++++++++- .../src/working_directory/cleanup.rs | 70 +- .../src/working_directory/occupancy.rs | 196 +++++ docs/development/workdir-removal.md | 24 +- 9 files changed, 2069 insertions(+), 102 deletions(-) create mode 100644 crates/worker-runtime/src/runtime/workdir_session.rs create mode 100644 crates/worker-runtime/src/working_directory/occupancy.rs diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs index 2d619655a..7d8a7697d 100644 --- a/crates/workdir/src/local.rs +++ b/crates/workdir/src/local.rs @@ -1770,6 +1770,9 @@ impl WorkdirSession for LocalWorkdirSession { async fn start_command(&self, request: CommandRequest) -> Result { self.ensure_capability(WorkdirSessionCapability::Command)?; + // Command admission, resource capture and registration must be atomic + // with terminal close, which drains commands and releases resources. + let _admission = self.inner.close_lock.lock().await; self.ensure_open()?; if let Some(spill_dir) = request.spill_dir.as_deref() && !self @@ -1794,6 +1797,15 @@ impl WorkdirSession for LocalWorkdirSession { let telemetry = self.inner.command_telemetry.clone(); let command_environment = self.inner.command_environment.clone(); let (cancel, cancel_rx) = watch::channel(false); + // A session drop only schedules command cancellation. Keep its Runtime + // resources (including cleanup exclusion) alive until the task actually + // finishes or its future is dropped, not just until the handle disappears. + let resources = self + .inner + .resources + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clone(); let task = tokio::spawn(async move { let output = run_command( cwd, @@ -1802,6 +1814,7 @@ impl WorkdirSession for LocalWorkdirSession { telemetry, command_environment, cancel_rx, + resources, ) .await; let _ = completion_tx.send(true); @@ -2131,6 +2144,7 @@ async fn run_command( telemetry: CommandTelemetry, command_environment: BTreeMap, mut cancel: watch::Receiver, + resources: Vec>, ) -> Result { let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?; let stderr = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?; @@ -2250,7 +2264,7 @@ async fn run_command( let stdout_path = stdout_path.to_path_buf(); let stderr_path = stderr_path.to_path_buf(); Some( - tokio::task::spawn_blocking(move || { + spawn_resource_owned_blocking(resources.clone(), move || { persist_command_output(&stdout_path, &stderr_path, &spill_dir) }) .await @@ -2272,6 +2286,18 @@ async fn run_command( }) } +// Blocking spills cannot be aborted with their async waiter. Transfer resource +// ownership into the actual filesystem task so cleanup waits for its completion. +fn spawn_resource_owned_blocking( + resources: Vec>, + effect: impl FnOnce() -> T + Send + 'static, +) -> JoinHandle { + tokio::task::spawn_blocking(move || { + let _resources = resources; + effect() + }) +} + #[derive(Debug, Default)] struct CommandOutputDecoder { read_offset: u64, @@ -4065,6 +4091,152 @@ mod tests { assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None))); } + #[tokio::test(flavor = "current_thread")] + async fn command_admission_serializes_with_close_before_capturing_resources() { + let dir = TempDir::new().unwrap(); + let session = Arc::new(make_fs(&dir)); + let held = session.inner.close_lock.lock().await; + let (closing, close_entered) = tokio::sync::oneshot::channel(); + let close_session = session.clone(); + let close = tokio::spawn(async move { + closing.send(()).unwrap(); + WorkdirSession::close(close_session.as_ref()).await + }); + close_entered.await.unwrap(); + let (starting, start_entered) = tokio::sync::oneshot::channel(); + let start_session = session.clone(); + let start = tokio::spawn(async move { + starting.send(()).unwrap(); + WorkdirSession::start_command( + start_session.as_ref(), + CommandRequest { + command: "true".into(), + timeout_secs: 5, + output_limit: 1024, + cwd: WorkdirPath::root(), + spill_dir: None, + tool_call_id: None, + }, + ) + .await + }); + start_entered.await.unwrap(); + assert!( + !start.is_finished(), + "command started outside terminal-close admission" + ); + drop(held); + tokio::time::timeout(Duration::from_secs(5), close) + .await + .unwrap() + .unwrap() + .unwrap(); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), start) + .await + .unwrap() + .unwrap(), + Err(WorkdirError::Unavailable(message)) if message.contains("is closed") + )); + assert!( + session.command_snapshot().is_empty(), + "closed admission launched a command" + ); + } + + #[tokio::test] + async fn blocking_output_spill_retains_resources_after_waiter_is_cancelled() { + #[derive(Debug)] + struct Resource(Option>); + impl Drop for Resource { + fn drop(&mut self) { + let _ = self.0.take().unwrap().send(()); + } + } + let (released, mut release) = tokio::sync::oneshot::channel(); + let resources: Vec> = + vec![Arc::new(Resource(Some(released)))]; + let (entered, started) = tokio::sync::oneshot::channel(); + let (finish, wait) = std::sync::mpsc::channel(); + // Exercise the resource-transfer policy at the same blocking boundary + // used by output persistence, without filesystem timing assumptions. + let waiter = tokio::spawn(async move { + spawn_resource_owned_blocking(resources, move || { + entered.send(()).unwrap(); + wait.recv_timeout(Duration::from_secs(5)).unwrap(); + }) + .await + .unwrap(); + }); + tokio::time::timeout(Duration::from_secs(5), started) + .await + .unwrap() + .unwrap(); + waiter.abort(); + assert!(waiter.await.unwrap_err().is_cancelled()); + assert!( + matches!( + release.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + ), + "blocking spill lost its resources when the waiter was cancelled" + ); + finish.send(()).unwrap(); + tokio::time::timeout(Duration::from_secs(5), release) + .await + .unwrap() + .unwrap(); + } + + #[tokio::test(flavor = "current_thread")] + async fn session_drop_retains_resources_until_command_task_cancellation_completes() { + #[derive(Debug)] + struct Resource(Option>); + impl Drop for Resource { + fn drop(&mut self) { + let _ = self.0.take().unwrap().send(()); + } + } + let dir = TempDir::new().unwrap(); + let (released, mut release) = tokio::sync::oneshot::channel(); + let session = LocalWorkdirSession::materialized_bound_with_environment( + Workdir::new("command-resource-session"), + dir.path().to_path_buf(), + dir.path().to_path_buf(), + SharedScope::new(Scope::writable(dir.path()).unwrap()), + WorkdirSessionCapabilities::ALL, + BTreeMap::new(), + vec![Arc::new(Resource(Some(released)))], + ); + WorkdirSession::start_command( + &session, + CommandRequest { + command: "true".into(), + timeout_secs: 5, + output_limit: 1024, + cwd: WorkdirPath::root(), + spill_dir: None, + tool_call_id: None, + }, + ) + .await + .unwrap(); + // The sole executor has not polled the command task, so dropping the + // session schedules its abort but cannot yet complete cancellation. + drop(session); + assert!( + matches!( + release.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + ), + "resources released before the aborted command future was dropped" + ); + tokio::time::timeout(Duration::from_secs(5), release) + .await + .unwrap() + .unwrap(); + } + #[tokio::test] async fn closing_session_releases_runtime_resources() { #[derive(Debug)] diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index df39f7651..336817c51 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -951,10 +951,12 @@ async fn observe_repository_ref( async fn list_working_directories( State(state): State, ) -> RestResult { - let working_directories = state - .runtime - .list_working_directories() - .map_err(RuntimeHttpRestError::runtime)?; + // Inventory providers may inspect the filesystem even for read-only requests. + let working_directories = + tokio::task::spawn_blocking(move || state.runtime.list_working_directories()) + .await + .map_err(RuntimeHttpRestError::working_directory_join)? + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkingDirectoriesResponse { working_directories, })) @@ -989,10 +991,11 @@ async fn get_working_directory( State(state): State, Path(working_directory_id): Path, ) -> RestResult { - let working_directory = state - .runtime - .working_directory(&working_directory_id) - .map_err(RuntimeHttpRestError::runtime)?; + let working_directory = + tokio::task::spawn_blocking(move || state.runtime.working_directory(&working_directory_id)) + .await + .map_err(RuntimeHttpRestError::working_directory_join)? + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkingDirectoryResponse { working_directory, })) @@ -1130,10 +1133,16 @@ async fn cleanup_working_directory( State(state): State, Path(working_directory_id): Path, ) -> RestResult { - let working_directory = state - .runtime - .cleanup_working_directory(&working_directory_id) - .map_err(RuntimeHttpRestError::runtime)?; + // Keep Runtime's occupancy check and deletion together, but do not run + // synchronous filesystem/Git work on the HTTP executor. + let working_directory = tokio::task::spawn_blocking(move || { + state + .runtime + .cleanup_working_directory(&working_directory_id) + }) + .await + .map_err(RuntimeHttpRestError::working_directory_join)? + .map_err(RuntimeHttpRestError::runtime)?; Ok(Json(RuntimeHttpWorkingDirectoryResponse { working_directory, })) @@ -2443,6 +2452,16 @@ impl RuntimeHttpRestError { Self::new(status, code, error.to_string()) } + fn working_directory_join(_error: tokio::task::JoinError) -> Self { + // A panic payload can contain paths or credentials. A missing result + // also does not prove that cleanup made no changes; do not claim rollback. + Self::new( + StatusCode::INTERNAL_SERVER_ERROR, + "working_directory_task_failed", + "Working directory operation failed to return a result", + ) + } + fn json_rejection(error: JsonRejection) -> Self { Self::new( StatusCode::BAD_REQUEST, @@ -4341,34 +4360,379 @@ mod tests { assert!(!error.to_string().contains('/')); } + enum CleanupHttpOutcome { + Removed, + Rejected(&'static str), + Panicked, + } + + struct CleanupHttpGate { + entered: tokio::sync::oneshot::Sender<()>, + release: std::sync::mpsc::Receiver<()>, + } + + struct CleanupHttpBackend { + gate: Mutex>, + outcome: CleanupHttpOutcome, + } + + fn removed_working_directory() -> WorkingDirectoryStatus { + WorkingDirectoryStatus { + summary: crate::catalog::WorkingDirectorySummary { + working_directory_id: "cleanup-test".to_string(), + display_name: None, + repository_id: "repository-test".to_string(), + creation_selector: None, + creation_ref: None, + creation_tree: None, + current_selector: None, + current_ref: None, + current_tree: None, + observed_at_epoch_seconds: None, + materializer_kind: crate::catalog::MaterializerKind::RuntimeGitClone, + cleanup_target: None, + status: crate::catalog::WorkingDirectoryStatusKind::NotFound, + cleanliness: None, + occupied_by: None, + }, + } + } + + impl WorkerExecutionBackend for CleanupHttpBackend { + fn backend_id(&self) -> &str { + "http-cleanup-test" + } + + fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { + AcceptingBackend.spawn_worker(request) + } + + fn dispatch_input( + &self, + worker_ref: &WorkerRef, + input: WorkerInput, + ) -> WorkerExecutionResult { + AcceptingBackend.dispatch_input(worker_ref, input) + } + + fn cleanup_working_directory( + &self, + working_directory_id: &str, + ) -> Result + { + assert_eq!(working_directory_id, "cleanup-test"); + let gate = self.gate.lock().unwrap().take(); + if let Some(gate) = gate { + gate.entered.send(()).expect("cleanup entry receiver"); + // A watchdog also bounds a regression that blocks the sole + // Tokio thread, where an async timeout could not make progress. + gate.release + .recv_timeout(std::time::Duration::from_secs(10)) + .expect("cleanup must be released after observations complete"); + } + match self.outcome { + CleanupHttpOutcome::Removed => Ok(removed_working_directory()), + CleanupHttpOutcome::Rejected(code) => Err( + crate::working_directory::WorkingDirectoryDiagnostic::rejected( + code, + "Resolve the current blocker and retry removal.", + ), + ), + CleanupHttpOutcome::Panicked => panic!("private provider path /secret/cleanup"), + } + } + } + + fn cleanup_http_router( + outcome: CleanupHttpOutcome, + gate: Option, + ) -> (Runtime, Router) { + let runtime = Runtime::with_execution_backend( + RuntimeOptions::default(), + Arc::new(CleanupHttpBackend { + gate: Mutex::new(gate), + outcome, + }), + ) + .unwrap(); + let identity = RuntimeIdentityMaterial::generate("runtime-cleanup-test").unwrap(); + let workspace_identity = + RuntimeIdentityMaterial::generate("workspace-cleanup-key").unwrap(); + let workspace_public_key = + crate::auth::decode_public_key(&workspace_identity.public_key).unwrap(); + let workspace_fingerprint = format!( + "sha256:{}", + crate::workspace_issuer::hex_lower(&sha2::Sha256::digest(workspace_public_key)) + ); + let app = runtime_http_router_with_workspace_auth( + runtime.clone(), + Some("cleanup-token".to_string()), + WorkspaceRuntimeHttpAuth { + verifier: WorkspaceCapabilityVerifier::new( + vec![WorkspaceIssuerTrustRecord { + workspace_id: "local".to_string(), + backend_url: "https://backend.test".to_string(), + key_id: "workspace-cleanup-key".to_string(), + algorithm: "ed25519".to_string(), + public_key: workspace_identity.public_key, + public_key_fingerprint: workspace_fingerprint, + identity_revision: 1, + trust_generation: 1, + state: WorkspaceIssuerTrustState::Active, + registered_at_unix: 1, + updated_at_unix: 1, + }], + Arc::new(InMemoryWorkspaceClaimReplayProtection::default()), + ) + .unwrap(), + signer: RuntimeVerificationSigner::from_identity(&identity).unwrap(), + verifications: Arc::new(InMemoryWorkspaceRuntimeVerificationAuthority::default()), + }, + ); + (runtime, app) + } + + #[tokio::test(flavor = "current_thread")] + async fn slow_working_directory_cleanup_does_not_block_http_ping() { + let (entered, started) = tokio::sync::oneshot::channel(); + let (release, wait_for_release) = std::sync::mpsc::channel(); + let (runtime, app) = cleanup_http_router( + CleanupHttpOutcome::Removed, + Some(CleanupHttpGate { + entered, + release: wait_for_release, + }), + ); + runtime + .store_config_bundle(test_bundle(ProfileSelector::Builtin( + "builtin:coder".to_string(), + ))) + .unwrap(); + let scope = RuntimeWorkspaceScope::new("local", "local-token"); + let mut request = task_request("unrelated Worker remains observable during cleanup"); + request.workspace_api = Some(WorkspaceApiRef { + workspace_id: "local".to_string(), + base_url: "https://backend.test".to_string(), + }); + let worker = runtime.create_worker_scoped(&scope, request).unwrap(); + assert!(worker.workdir_attachments.is_empty()); + let worker_path = format!("/v1/workers/{}", worker.worker_id); + let path = "/v1/working-directories/cleanup-test"; + let unauthorized = empty_request(app.clone(), Method::DELETE, path).await; + assert_eq!(unauthorized.status(), StatusCode::UNAUTHORIZED); + let error: RuntimeHttpErrorResponse = read_json(unauthorized).await; + assert_eq!(error.error.code, "unauthorized"); + + // A synchronous RuntimeState lock can stall this sole executor thread, + // including Tokio timeouts. Release from an independent thread on either + // completion, assertion unwind, or watchdog expiry, and report expiry as + // failure even if observations resume after the forced release. + let (observed, wait_for_observations) = std::sync::mpsc::channel(); + let watchdog = std::thread::spawn(move || { + let timed_out = matches!( + wait_for_observations.recv_timeout(std::time::Duration::from_secs(5)), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) + ); + let _ = release.send(()); + timed_out + }); + let cleanup = tokio::spawn(authed_empty_request( + app.clone(), + Method::DELETE, + path, + "cleanup-token", + )); + tokio::time::timeout(std::time::Duration::from_secs(2), started) + .await + .expect("cleanup provider must be entered") + .expect("cleanup entry signal"); + + // Unlike ping, these HTTP routes synchronously acquire RuntimeState. + // Exercise existing Worker paths so authorization/not-found shortcuts + // cannot make a global occupancy-lock regression pass unnoticed. + for observation_path in [ + "/v1/workers".to_string(), + worker_path.clone(), + format!("{worker_path}/session?workspace_id=local"), + ] { + let response = tokio::time::timeout( + std::time::Duration::from_secs(2), + authed_empty_request(app.clone(), Method::GET, &observation_path, "cleanup-token"), + ) + .await + .unwrap_or_else(|_| panic!("{observation_path} must finish while cleanup is gated")); + let status = response.status(); + let body: serde_json::Value = read_json(response).await; + assert_eq!(status, StatusCode::OK, "{observation_path}: {body}"); + if observation_path == "/v1/workers" { + assert_eq!(body["workers"].as_array().unwrap().len(), 1, "{body}"); + assert_eq!( + body["workers"][0]["worker_id"], + worker.worker_id.to_string() + ); + } else if observation_path == worker_path { + assert_eq!(body["worker"]["worker_id"], worker.worker_id.to_string()); + } else { + // This minimal fake has no session store. Its normal, typed + // availability result still requires the real scoped HTTP read. + assert_eq!( + body, + serde_json::to_value(runtime_api::WorkerSessionAvailability::Unavailable { + reason: runtime_api::WorkerSessionUnavailableReason::StorageUnavailable, + message: "retained session storage is unavailable".to_string(), + }) + .unwrap(), + ); + } + } + + // Exercise the same scoped snapshot acquisition used by Runtime WS + // subscriptions, without requiring a socket or the ws-server feature. + let subscription = runtime + .subscribe_event_selector_scoped( + &scope, + protocol::subscription::EventSubscriptionSelector::RuntimeWorkers, + ) + .expect("Runtime subscription snapshot must complete while cleanup is gated"); + let protocol::subscription::SubscriptionSnapshot::Workers { workers } = + subscription.snapshot() + else { + panic!("expected Runtime Worker subscription snapshot"); + }; + assert_eq!(workers.len(), 1); + assert_eq!(workers[0].worker_id.as_str(), worker.worker_id.to_string()); + + let ping = tokio::time::timeout( + std::time::Duration::from_secs(2), + app.oneshot( + Request::builder() + .method(Method::GET) + .uri("/v1/ping") + .header(header::AUTHORIZATION, "Bearer cleanup-token") + .header(RUNTIME_WORKSPACE_SCOPE_HEADER, "local") + .body(Body::empty()) + .unwrap(), + ), + ) + .await + .expect("ping must finish while cleanup is gated") + .unwrap(); + let ping_status = ping.status(); + let ping_body: serde_json::Value = read_json(ping).await; + assert_eq!(ping_status, StatusCode::OK, "{ping_body}"); + assert_eq!(ping_body["runtime_id"], "runtime-cleanup-test"); + assert_eq!(ping_body["protocol_version"], RUNTIME_HTTP_PROTOCOL_VERSION); + assert!( + !cleanup.is_finished(), + "cleanup must still be gated after ping", + ); + // The watchdog owns the release sender; dropping `observed` on an + // assertion failure also releases the provider instead of leaking a task. + let observations_completed = observed.send(()).is_ok(); + let watchdog_expired = watchdog.join().unwrap(); + assert!( + observations_completed && !watchdog_expired, + "Worker observations, subscription snapshot, and ping must finish before cleanup is released", + ); + let response = tokio::time::timeout(std::time::Duration::from_secs(2), cleanup) + .await + .expect("released cleanup must finish") + .unwrap(); + let status = response.status(); + let body: serde_json::Value = read_json(response).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body, + serde_json::to_value(RuntimeHttpWorkingDirectoryResponse { + working_directory: removed_working_directory(), + }) + .unwrap(), + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn working_directory_cleanup_panic_returns_bounded_http_error() { + let (_, app) = cleanup_http_router(CleanupHttpOutcome::Panicked, None); + let response = authed_empty_request( + app, + Method::DELETE, + "/v1/working-directories/cleanup-test", + "cleanup-token", + ) + .await; + let status = response.status(); + let body: serde_json::Value = read_json(response).await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR, "{body}"); + assert_eq!( + body, + serde_json::json!({ + "error": { + "code": "working_directory_task_failed", + "message": "Working directory operation failed to return a result", + }, + }), + ); + } + #[tokio::test] async fn cleanup_cause_codes_survive_http_error_serialization() { let cases = [ - ("mount_present", StatusCode::CONFLICT), - ("mount_check_unavailable", StatusCode::SERVICE_UNAVAILABLE), - ("permission_denied", StatusCode::FORBIDDEN), - ("resource_busy", StatusCode::CONFLICT), - ("storage_unavailable", StatusCode::SERVICE_UNAVAILABLE), - ("ownership_unknown", StatusCode::CONFLICT), - ("changes_present", StatusCode::CONFLICT), - ("changes_unknown", StatusCode::CONFLICT), + ("working_directory_not_found", StatusCode::NOT_FOUND), + ( + "working_directory_cleanup_mount_present", + StatusCode::CONFLICT, + ), + ( + "working_directory_cleanup_mount_check_unavailable", + StatusCode::SERVICE_UNAVAILABLE, + ), + ( + "working_directory_cleanup_permission_denied", + StatusCode::FORBIDDEN, + ), + ( + "working_directory_cleanup_resource_busy", + StatusCode::CONFLICT, + ), + ( + "working_directory_cleanup_storage_unavailable", + StatusCode::SERVICE_UNAVAILABLE, + ), + ( + "working_directory_cleanup_ownership_unknown", + StatusCode::CONFLICT, + ), + ( + "working_directory_cleanup_changes_present", + StatusCode::CONFLICT, + ), + ( + "working_directory_cleanup_changes_unknown", + StatusCode::CONFLICT, + ), ]; - for (cause, status) in cases { - let code = format!("working_directory_cleanup_{cause}"); + for (code, expected_status) in cases { + let (_, app) = cleanup_http_router(CleanupHttpOutcome::Rejected(code), None); + let response = authed_empty_request( + app, + Method::DELETE, + "/v1/working-directories/cleanup-test", + "cleanup-token", + ) + .await; + let status = response.status(); + let json: serde_json::Value = read_json(response).await; + assert_eq!(status, expected_status, "{json}"); + assert_eq!(json["error"]["code"], code, "{json}"); let diagnostic = crate::working_directory::WorkingDirectoryDiagnostic::rejected( - &code, + code, "Resolve the current blocker and retry removal.", ); - let response = - RuntimeHttpRestError::runtime(RuntimeError::WorkingDirectory(diagnostic)) - .into_response(); - assert_eq!(response.status(), status); - let body = axum::body::to_bytes(response.into_body(), 2048) - .await - .unwrap(); - let json: serde_json::Value = serde_json::from_slice(&body).unwrap(); - assert_eq!(json["error"]["code"], code, "{json}"); - assert!(json["error"]["message"].as_str().unwrap().contains("retry")); + assert_eq!( + json["error"]["message"], + RuntimeError::WorkingDirectory(diagnostic).to_string(), + "{json}", + ); } } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 3489140e9..1c812321c 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -59,6 +59,54 @@ use tokio::sync::broadcast; use tokio::sync::mpsc; use uuid::Uuid; +mod workdir_session; + +#[derive(Debug)] +enum WorkdirOperation { + Use(usize), + Cleanup, +} + +/// Transient admission authority, not persisted cleanup or recovery state. +#[derive(Debug)] +struct WorkdirOperationGuard { + state: Weak>, + ids: Vec, + cleanup: bool, +} + +impl Drop for WorkdirOperationGuard { + fn drop(&mut self) { + let Some(state) = self.state.upgrade() else { + return; + }; + let mut state = state + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + for id in &self.ids { + match state.workdir_operations.get_mut(id) { + Some(WorkdirOperation::Use(count)) if !self.cleanup && *count > 1 => *count -= 1, + Some(WorkdirOperation::Use(_)) if !self.cleanup => { + state.workdir_operations.remove(id); + } + Some(WorkdirOperation::Cleanup) if self.cleanup => { + state.workdir_operations.remove(id); + } + _ => {} + } + } + } +} + +fn workdir_operation_busy() -> RuntimeError { + RuntimeError::WorkingDirectory( + crate::working_directory::WorkingDirectoryDiagnostic::rejected( + "working_directory_cleanup_resource_busy", + "Workdir has an in-progress operation; retry after it finishes", + ), + ) +} + /// Workspace-scoped Runtime authorization context supplied by a trusted backend. #[derive(Clone, Debug, Eq, PartialEq)] pub struct RuntimeWorkspaceScope { @@ -296,6 +344,37 @@ impl Drop for WorkerOperationLease { } impl Runtime { + fn reserve_workdir_use<'a>( + &self, + ids: impl IntoIterator, + ) -> Result { + let mut ids: Vec = ids.into_iter().map(str::to_string).collect(); + ids.sort(); + ids.dedup(); + let mut state = self.lock()?; + state.ensure_running()?; + for id in &ids { + state.ensure_workdir_not_removing(id)?; + } + for id in &ids { + match state + .workdir_operations + .entry(id.clone()) + .or_insert(WorkdirOperation::Use(0)) + { + WorkdirOperation::Use(count) => *count += 1, + WorkdirOperation::Cleanup => { + unreachable!("cleanup admission checked under state lock") + } + } + } + Ok(WorkdirOperationGuard { + state: Arc::downgrade(&self.inner), + ids, + cleanup: false, + }) + } + /// Create a memory-backed Runtime with generated identity. pub fn new_memory() -> Self { Self::with_options(RuntimeOptions::default()) @@ -515,8 +594,15 @@ impl Runtime { /// Create a Runtime-owned working directory through the attached execution backend. pub fn create_working_directory( &self, - request: WorkingDirectoryRequest, + mut request: WorkingDirectoryRequest, ) -> Result { + let id = request + .backend_workdir_id + .get_or_insert_with(|| { + crate::working_directory::next_working_directory_id(&request.repository.id) + }) + .clone(); + let _operation = self.reserve_workdir_use([id.as_str()])?; let backend = { let state = self.lock()?; state.ensure_running()?; @@ -581,6 +667,7 @@ impl Runtime { &self, request: WorkingDirectoryRepositoryAccessRequest, ) -> Result<(), RuntimeError> { + let _operation = self.reserve_workdir_use([request.working_directory_id.as_str()])?; let backend = { let state = self.lock()?; state.ensure_running()?; @@ -775,6 +862,7 @@ impl Runtime { working_directory_id: &str, owner_worker_ref: Option<&WorkerRef>, ) -> Result { + let operation = self.reserve_workdir_use([working_directory_id])?; let backend = { let mut state = self.lock()?; state.ensure_running()?; @@ -809,9 +897,10 @@ impl Runtime { backend .working_directory(working_directory_id) .map_err(RuntimeError::WorkingDirectory)?; - backend + let session = backend .open_workdir_session(working_directory_id) - .map_err(RuntimeError::WorkingDirectory) + .map_err(RuntimeError::WorkingDirectory)?; + Ok(workdir_session::retain_operation(session, operation)) } /// Cleanup a Runtime-owned working directory. @@ -819,30 +908,40 @@ impl Runtime { &self, working_directory_id: &str, ) -> Result { - // Hold occupancy authority through the side effect, preventing an - // attachment from entering between the check and physical removal. - // This synchronous provider call must not reenter Runtime. The Git - // materializer does filesystem/Git work only. The global guard also - // delays unrelated state operations during hashing/unlink; releasing it - // would require equivalent per-workdir exclusion on ALL attachment paths, - // not just moving the occupied check before the provider call. - let state = self.lock()?; - state.ensure_running()?; - if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) { - return Err(RuntimeError::InvalidRequest(format!( - "working directory {working_directory_id} is assigned to worker {worker_id}" - ))); - } - let backend = state.execution_backend.clone().ok_or_else(|| { - RuntimeError::ExecutionBackendUnavailable { - message: "working directory cleanup requires an execution backend".to_string(), + // Admit cleanup atomically with attachment/use admission, but reserve + // only this Workdir while the provider hashes/unlinks. No filesystem + // work or wait for a Workdir operation may hold the global state mutex. + let backend = { + let mut state = self.lock()?; + state.ensure_running()?; + if state.workdir_operations.contains_key(working_directory_id) { + return Err(workdir_operation_busy()); } - })?; - let result = backend + if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) { + return Err(RuntimeError::InvalidRequest(format!( + "working directory {working_directory_id} is assigned to worker {worker_id}" + ))); + } + let backend = state.execution_backend.clone().ok_or_else(|| { + RuntimeError::ExecutionBackendUnavailable { + message: "working directory cleanup requires an execution backend".to_string(), + } + })?; + state + .workdir_operations + .insert(working_directory_id.to_string(), WorkdirOperation::Cleanup); + backend + }; + // Construct/drop the RAII owner outside the state guard: failure, panic + // and request cancellation must not leave stale exclusion or deadlock. + let _operation = WorkdirOperationGuard { + state: Arc::downgrade(&self.inner), + ids: vec![working_directory_id.to_string()], + cleanup: true, + }; + backend .cleanup_working_directory(working_directory_id) - .map_err(RuntimeError::from); - drop(state); - result + .map_err(RuntimeError::from) } fn annotate_working_directory_statuses( @@ -1041,7 +1140,7 @@ impl Runtime { fn create_worker_with_workspace_inner( &self, - request: CreateWorkerRequest, + mut request: CreateWorkerRequest, scope: Option<&RuntimeWorkspaceScope>, ) -> Result { let operation_lock = self.worker_operation_lock(request.worker_id)?; @@ -1051,6 +1150,17 @@ impl Runtime { if let Some(existing) = self.existing_worker_for_create(&request, scope)? { return Ok(existing); } + // Assign generated materializations before publishing the durable + // request, so cleanup cannot discover an unclaimed creation mid-spawn. + for attachment in &mut request.workdir_attachment_requests { + let directory = &mut attachment.working_directory; + if directory.backend_workdir_id.is_none() { + directory.backend_workdir_id = Some( + crate::working_directory::next_working_directory_id(&directory.repository.id), + ); + } + } + let _workdir_operation = self.reserve_workdir_use(requested_workdir_ids(&request))?; self.refresh_workspace_config(&request)?; let (backend, worker_ref, spawn_request) = { let mut state = self.lock()?; @@ -1565,6 +1675,9 @@ impl Runtime { 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)?; + } let previous = state.worker(worker_ref)?.clone(); { let worker = state.worker_mut(worker_ref)?; @@ -1875,6 +1988,13 @@ 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 @@ -2074,6 +2194,19 @@ impl Runtime { state.ensure_workspace_owner(scope, true)?; state.persist_runtime_snapshot()?; } + // Recovery uses the persisted operation, not new preparation. + // Discard it before checking Workdir admission as well as before + // publishing the owner; unrelated cleanup must not reject recovery. + 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 { @@ -2083,9 +2216,6 @@ impl Runtime { }); let operation_id = pending.operation_id; candidate.pending_restore = Some(pending); - if recovering_pending { - request.preparation = None; - } candidate.restore_guard.rotate(); candidate.restore_guard.active_request_id = Some(request.request_id.clone()); candidate.restore_guard.owners.insert( @@ -2517,6 +2647,28 @@ 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. @@ -4853,6 +5005,7 @@ struct RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: u64, workers: BTreeMap, + workdir_operations: BTreeMap, workspace_owners: BTreeMap, config_bundles: BTreeMap, workspace_config_latest: BTreeMap, @@ -4882,6 +5035,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), + workdir_operations: BTreeMap::new(), workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), workspace_config_latest: BTreeMap::new(), @@ -4912,6 +5066,7 @@ impl RuntimeState { #[cfg(feature = "fs-store")] next_diagnostic_id: 1, workers: BTreeMap::new(), + workdir_operations: BTreeMap::new(), workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), workspace_config_latest: BTreeMap::new(), @@ -5056,6 +5211,7 @@ impl RuntimeState { workspace_backend_resource_clients: BTreeMap::new(), next_diagnostic_id, workers, + workdir_operations: BTreeMap::new(), config_bundles: BTreeMap::new(), workspace_config_latest: BTreeMap::new(), workspace_config_fetch_gates: BTreeMap::new(), @@ -5470,6 +5626,16 @@ impl RuntimeState { } } + fn ensure_workdir_not_removing(&self, id: &str) -> Result<(), RuntimeError> { + if matches!( + self.workdir_operations.get(id), + Some(WorkdirOperation::Cleanup) + ) { + return Err(workdir_operation_busy()); + } + Ok(()) + } + fn worker_id_for_workdir(&self, working_directory_id: &str) -> Option { self.workers.values().find_map(|worker| { let has_status = worker.workdir_attachments.iter().any(|attachment| { @@ -5478,7 +5644,23 @@ impl RuntimeState { let has_request = worker.request.as_ref().is_some_and(|request| { requested_workdir_ids(request).contains(&working_directory_id) }); - (has_status || has_request).then_some(worker.worker_id) + let has_logical = worker + .logical_workdir_attachments + .iter() + .any(|attachment| attachment.working_directory_id == working_directory_id); + let has_preparation = worker + .restore_guard + .active_request_id + .as_ref() + .and_then(|id| worker.restore_guard.owners.get(id)) + .filter(|owner| owner.receipt.is_none()) + .and_then(|owner| owner.request.preparation.as_ref()) + .is_some_and(|preparation| { + preparation_workdir_ids(preparation, self.runtime_identity.as_deref()) + .contains(&working_directory_id) + }); + (has_status || has_request || has_logical || has_preparation) + .then_some(worker.worker_id) }) } @@ -6179,6 +6361,33 @@ fn durable_create_worker_request(request: &CreateWorkerRequest) -> CreateWorkerR durable } +fn preparation_workdir_ids<'a>( + preparation: &'a runtime_api::WorkerRestorePreparation, + runtime_id: Option<&str>, +) -> Vec<&'a str> { + let local = |owner: &str| runtime_id.is_none_or(|runtime_id| runtime_id == owner); + preparation + .workdir_attachments + .iter() + .flatten() + .map(|attachment| attachment.working_directory_id.as_str()) + .chain( + preparation + .repository_access_workdirs + .iter() + .filter(|reference| local(&reference.runtime_id)) + .map(|reference| reference.working_directory_id.as_str()), + ) + .chain( + preparation + .repository_access + .iter() + .filter(|access| local(&access.materialization.runtime_id)) + .map(|access| access.working_directory_id.as_str()), + ) + .collect() +} + fn requested_workdir_ids(request: &CreateWorkerRequest) -> Vec<&str> { request .workdir_attachments @@ -7641,6 +7850,35 @@ mod tests { } } + struct WorkdirEffectGate { + entered: std::sync::mpsc::Sender, + release: std::sync::mpsc::Receiver<()>, + } + + impl WorkdirEffectGate { + fn wait(slot: &Mutex>, id: &str) { + let gate = slot.lock().unwrap().take(); + if let Some(gate) = gate { + gate.entered.send(id.to_string()).unwrap(); + gate.release + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("Workdir effect did not receive release before timeout"); + } + } + + fn install( + slot: &Mutex>, + ) -> ( + std::sync::mpsc::Receiver, + std::sync::mpsc::Sender<()>, + ) { + let (entered, receiver) = std::sync::mpsc::channel(); + let (sender, release) = std::sync::mpsc::channel(); + *slot.lock().unwrap() = Some(Self { entered, release }); + (receiver, sender) + } + } + #[derive(Default)] struct TestExecutionBackend { spawn_result: Mutex>, @@ -7673,6 +7911,10 @@ mod tests { repository_access_available: AtomicBool, working_directory_requests: Mutex>, cleanup_occupancy_probe: Mutex>>>, + cleanup_gate: Mutex>, + workdir_use_gate: Mutex>, + cleanup_calls: std::sync::atomic::AtomicUsize, + cleanup_panics: AtomicBool, preserve_submission_acknowledgement_id: AtomicBool, #[cfg(feature = "ws-server")] snapshots: Mutex>, @@ -7717,10 +7959,21 @@ mod tests { &self, _id: &str, ) -> Result { + self.cleanup_calls.fetch_add(1, Ordering::SeqCst); + WorkdirEffectGate::wait(&self.cleanup_gate, _id); + if self.cleanup_panics.load(Ordering::SeqCst) { + panic!("cleanup provider failed after admission"); + } if let Some(state) = self.cleanup_occupancy_probe.lock().unwrap().as_ref() { + let state = state + .try_lock() + .expect("physical cleanup must not hold global Runtime state"); assert!( - matches!(state.try_lock(), Err(std::sync::TryLockError::WouldBlock)), - "occupancy must remain exclusive during physical cleanup" + matches!( + state.workdir_operations.get(_id), + Some(WorkdirOperation::Cleanup) + ), + "the target Workdir must remain reserved during physical cleanup" ); } Err(WorkingDirectoryDiagnostic::rejected( @@ -7749,6 +8002,10 @@ mod tests { &self, request: &WorkingDirectoryRequest, ) -> Result { + WorkdirEffectGate::wait( + &self.workdir_use_gate, + request.backend_workdir_id.as_deref().unwrap(), + ); self.working_directory_requests .lock() .unwrap() @@ -7763,6 +8020,7 @@ mod tests { &self, request: &WorkingDirectoryRepositoryAccessRequest, ) -> Result<(), WorkingDirectoryDiagnostic> { + WorkdirEffectGate::wait(&self.workdir_use_gate, &request.working_directory_id); self.repository_accesses .lock() .unwrap() @@ -8068,7 +8326,7 @@ mod tests { } #[test] - fn workdir_removal_keeps_occupancy_exclusive_during_provider_side_effect() { + fn workdir_removal_reserves_only_its_target_and_releases_on_failure() { let (runtime, backend) = runtime_and_backend(); *backend.cleanup_occupancy_probe.lock().unwrap() = Some(runtime.inner.clone()); let error = runtime @@ -8078,11 +8336,384 @@ mod tests { matches!(error, RuntimeError::WorkingDirectory(ref diagnostic) if diagnostic.code == "working_directory_cleanup_resource_busy") ); - // The lock is released on failure, permitting normal attachment/retry. - assert!(runtime.inner.try_lock().is_ok()); + // Failure releases the per-target reservation for normal use/retry. + assert!(runtime.inner.lock().unwrap().workdir_operations.is_empty()); *backend.cleanup_occupancy_probe.lock().unwrap() = None; } + fn workdir_effect_request(id: Option<&str>) -> WorkingDirectoryRequest { + WorkingDirectoryRequest { + repository: WorkingDirectoryRepository { + id: "repository-1".into(), + provider: "git".into(), + source: server_api::RepositorySource { + kind: server_api::RepositorySourceKind::LocalPath, + uri: "/unused-test-repository".into(), + }, + source_revision: 1, + source_fingerprint: "sha256:test-source".into(), + selector: None, + }, + display_name: None, + materializer: MaterializerKind::RuntimeGitClone, + backend_workdir_id: id.map(str::to_string), + materialization: None, + } + } + + fn workdir_access_request(id: &str) -> WorkingDirectoryRepositoryAccessRequest { + WorkingDirectoryRepositoryAccessRequest { + working_directory_id: id.into(), + materialization: RepositoryMaterializationContext { + workspace_id: "workspace-a".into(), + runtime_id: "runtime-1".into(), + operation_id: "operation-1".into(), + config_revision: 1, + config_projection_digest: "sha256:test-projection".into(), + ssh: None, + }, + } + } + + fn assert_workdir_busy(error: RuntimeError) { + assert!( + matches!(error, RuntimeError::WorkingDirectory(ref diagnostic) + if diagnostic.code == "working_directory_cleanup_resource_busy"), + "{error:?}" + ); + } + + #[test] + fn active_workdir_cleanup_rejects_same_id_admissions_without_blocking_other_workers() { + let (runtime, backend) = runtime_and_backend(); + let owner = scope("workspace-a", "server-a"); + let worker = runtime + .create_worker_scoped(&owner, scoped_task_request("unrelated", "workspace-a")) + .unwrap(); + let id = "workdir-cleanup-active"; + let (entered, release) = WorkdirEffectGate::install(&backend.cleanup_gate); + let cleanup_runtime = runtime.clone(); + let cleanup = std::thread::spawn(move || cleanup_runtime.cleanup_working_directory(id)); + assert_eq!( + entered + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(), + id + ); + + assert_eq!(runtime.list_workers_scoped(&owner).unwrap().len(), 1); + assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err()); + assert_workdir_busy( + runtime + .create_working_directory(workdir_effect_request(Some(id))) + .unwrap_err(), + ); + assert_workdir_busy( + runtime + .authorize_working_directory_repository_access(workdir_access_request(id)) + .unwrap_err(), + ); + assert_workdir_busy( + runtime + .open_workdir_session_scoped(&owner, id, None) + .unwrap_err(), + ); + let mut create = scoped_task_request("conflicting attachment", "workspace-a"); + create.workdir_attachments = vec![WorkingDirectoryAttachmentClaim { + alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(), + working_directory_id: id.into(), + relative_cwd: None, + capabilities: workdir::WorkdirSessionCapabilities::ALL, + }]; + assert_workdir_busy(runtime.create_worker_scoped(&owner, create).unwrap_err()); + let attachment = LogicalWorkdirAttachment { + alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(), + working_directory_id: id.into(), + capabilities: workdir::WorkdirSessionCapabilities::ALL, + }; + assert_workdir_busy( + runtime + .replace_worker_workdir_attachments_scoped( + &owner, + &worker.worker_ref, + vec![attachment.clone()], + ) + .unwrap_err(), + ); + runtime + .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!( + runtime + .lock() + .unwrap() + .worker(&worker.worker_ref) + .unwrap() + .pending_restore + .is_none() + ); + assert_eq!(runtime.list_workers_scoped(&owner).unwrap().len(), 1); + // Different IDs and Worker creation remain available during cleanup. + runtime + .create_worker_scoped( + &owner, + scoped_task_request("another unrelated", "workspace-a"), + ) + .unwrap(); + let other = runtime + .create_working_directory(workdir_effect_request(Some("workdir-other"))) + .unwrap_err(); + assert!( + matches!(other, RuntimeError::WorkingDirectory(ref diagnostic) if diagnostic.code == "working_directory_unsupported") + ); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1); + release.send(()).unwrap(); + assert_workdir_busy(cleanup.join().unwrap().unwrap_err()); + assert!(runtime.lock().unwrap().workdir_operations.is_empty()); + } + + #[test] + fn in_flight_creation_and_access_exclude_cleanup_including_generated_ids() { + for operation in ["explicit-create", "generated-create", "access"] { + let (runtime, backend) = runtime_and_backend(); + let (entered, release) = WorkdirEffectGate::install(&backend.workdir_use_gate); + let effect_runtime = runtime.clone(); + let effect = std::thread::spawn(move || { + if operation == "access" { + effect_runtime.authorize_working_directory_repository_access( + workdir_access_request("workdir-in-use"), + ) + } else { + effect_runtime + .create_working_directory(workdir_effect_request( + (operation == "explicit-create").then_some("workdir-in-use"), + )) + .map(|_| ()) + } + }); + let id = entered + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + assert!(!id.is_empty()); + assert_workdir_busy(runtime.cleanup_working_directory(&id).unwrap_err()); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 0); + runtime + .create_worker(task_request("unrelated during effect")) + .unwrap(); + release.send(()).unwrap(); + assert!(effect.join().unwrap().is_err()); + // Failed provider effects release their reservation for ordinary retry. + assert_workdir_busy(runtime.cleanup_working_directory(&id).unwrap_err()); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1); + } + } + + #[test] + fn logical_attachments_and_unsettled_restore_preparation_exclude_cleanup() { + for pending in [false, true] { + let (runtime, backend) = runtime_and_backend(); + let owner = scope("workspace-a", "server-a"); + let worker = runtime + .create_worker_scoped(&owner, scoped_task_request("logical owner", "workspace-a")) + .unwrap(); + let id = "workdir-logical-owner"; + let attachment = LogicalWorkdirAttachment { + alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(), + working_directory_id: id.into(), + capabilities: workdir::WorkdirSessionCapabilities::ALL, + }; + if pending { + runtime + .stop_worker_scoped(&owner, &worker.worker_ref, None) + .unwrap(); + let restore = runtime.test_restore_request(&worker.worker_ref); + let preparation = 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() + }; + let coordinated = 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(preparation.clone()), + }, + Some(&owner), + ) + .unwrap(); + assert_eq!(coordinated, (None, Some(preparation))); + } else { + runtime + .replace_worker_workdir_attachments_scoped( + &owner, + &worker.worker_ref, + vec![attachment], + ) + .unwrap(); + } + assert!( + matches!(runtime.cleanup_working_directory(id).unwrap_err(), RuntimeError::InvalidRequest(message) if message.contains("assigned to worker")) + ); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 0); + } + } + + #[test] + fn unowned_restore_recovery_ignores_new_preparation_for_a_workdir_being_cleaned() { + let (runtime, backend) = runtime_and_backend(); + let owner = scope("workspace-a", "server-a"); + let worker = runtime + .create_worker_scoped( + &owner, + scoped_task_request("recover persisted operation", "workspace-a"), + ) + .unwrap(); + runtime + .stop_worker_scoped(&owner, &worker.worker_ref, None) + .unwrap(); + // Persisted lifecycle recovery may have no request owner (e.g. after an + // interrupted older admission). New preparation is not its authority. + runtime + .lock() + .unwrap() + .worker_mut(&worker.worker_ref) + .unwrap() + .pending_restore = Some(PendingWorkerRestore { + operation_id: WorkerLifecycleOperationId::new(), + mode: WorkerRestoreMode::Explicit, + last_settled_status: WorkerStatus::Stopped, + }); + let id = "workdir-irrelevant-to-recovery"; + let (entered, release) = WorkdirEffectGate::install(&backend.cleanup_gate); + let cleanup_runtime = runtime.clone(); + let cleanup = std::thread::spawn(move || cleanup_runtime.cleanup_working_directory(id)); + entered + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + let restore = runtime.test_restore_request(&worker.worker_ref); + let (result, preparation) = 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: workdir::WorkdirAttachmentAlias::new("ignored").unwrap(), + working_directory_id: id.into(), + capabilities: workdir::WorkdirSessionCapabilities::ALL, + }]), + ..Default::default() + }), + }, + Some(&owner), + ) + .unwrap(); + assert_eq!(result.unwrap().state, WorkerRestoreState::Accepted); + assert!(preparation.is_none()); + assert!( + runtime + .lock() + .unwrap() + .worker(&worker.worker_ref) + .unwrap() + .logical_workdir_attachments + .is_empty() + ); + release.send(()).unwrap(); + assert_workdir_busy(cleanup.join().unwrap().unwrap_err()); + } + + #[tokio::test] + async fn live_workdir_session_clones_exclude_cleanup_until_close_or_final_drop() { + for close in [false, true] { + let (runtime, backend) = runtime_and_backend(); + let root = tempfile::tempdir().unwrap(); + let id = "workdir-live-session"; + let provider: workdir::WorkdirSessionHandle = Arc::new( + workdir::LocalWorkdirSession::materialized_bound_with_environment( + workdir::Workdir::new(id), + root.path().to_path_buf(), + root.path().to_path_buf(), + manifest::SharedScope::new(manifest::Scope::writable(root.path()).unwrap()), + workdir::WorkdirSessionCapabilities::ALL, + BTreeMap::new(), + Vec::new(), + ), + ); + let session = workdir_session::retain_operation( + provider, + runtime.reserve_workdir_use([id]).unwrap(), + ); + let clone = session.clone(); + drop(session); + assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err()); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 0); + if close { + clone.close().await.unwrap(); + clone.close().await.unwrap(); + // A closed wrapper may remain alive without reserving storage. + assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err()); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1); + } else { + drop(clone); + assert_workdir_busy(runtime.cleanup_working_directory(id).unwrap_err()); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 1); + } + } + } + + #[test] + fn workdir_cleanup_panic_releases_reservation_without_poisoning_runtime() { + let (runtime, backend) = runtime_and_backend(); + backend.cleanup_panics.store(true, Ordering::SeqCst); + assert!( + std::panic::catch_unwind(std::panic::AssertUnwindSafe( + || runtime.cleanup_working_directory("workdir-panic") + )) + .is_err() + ); + assert!(runtime.lock().unwrap().workdir_operations.is_empty()); + backend.cleanup_panics.store(false, Ordering::SeqCst); + runtime + .create_worker(task_request("after cleanup panic")) + .unwrap(); + assert_workdir_busy( + runtime + .cleanup_working_directory("workdir-panic") + .unwrap_err(), + ); + assert_eq!(backend.cleanup_calls.load(Ordering::SeqCst), 2); + } + #[test] fn unknown_typed_intent_is_rejected_before_backend_dispatch() { let (runtime, backend) = runtime_and_backend(); diff --git a/crates/worker-runtime/src/runtime/workdir_session.rs b/crates/worker-runtime/src/runtime/workdir_session.rs new file mode 100644 index 000000000..5a3249c0b --- /dev/null +++ b/crates/worker-runtime/src/runtime/workdir_session.rs @@ -0,0 +1,165 @@ +//! Retain a Runtime active-use reservation for a live provider session. +//! +//! Cloned handles share the reservation. A successful provider close releases +//! it even while closed handles remain; a failed or cancelled close retains it +//! until a later successful close or the final wrapper is dropped. Dropping the +//! wrapper does not close the provider or wait for outstanding commands, so +//! callers must keep a session handle alive while using provider-owned work. + +use super::WorkdirOperationGuard; +use async_trait::async_trait; +use std::sync::{Arc, Mutex}; +use tokio::sync::broadcast; +use workdir::{ + CheckoutObservation, CheckoutRequest, CheckoutResult, CheckoutSearchRequest, + CheckoutSearchResult, CommandEvent, CommandHandle, CommandOutput, CommandOutputRequest, + CommandRequest, CommandSnapshot, CommandStatus, EditRequest, EditResult, GlobRequest, + GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, ReadBytesRequest, + ReadBytesResult, ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirError, + WorkdirPath, WorkdirScopeAuthorizationRequest, WorkdirScopeOverlapRequest, WorkdirSession, + WorkdirSessionCapabilities, WorkdirSessionHandle, WriteRequest, WriteResult, +}; + +pub(super) fn retain_operation( + session: WorkdirSessionHandle, + operation: WorkdirOperationGuard, +) -> WorkdirSessionHandle { + Arc::new(RetainedWorkdirSession { + inner: session, + operation: Mutex::new(Some(operation)), + }) +} + +struct RetainedWorkdirSession { + inner: WorkdirSessionHandle, + operation: Mutex>, +} + +impl std::fmt::Debug for RetainedWorkdirSession { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Do not expose provider internals or reservation state containing host paths. + f.debug_struct("RetainedWorkdirSession") + .field("workdir", self.inner.workdir()) + .field("capabilities", &self.inner.capabilities()) + .finish_non_exhaustive() + } +} + +#[async_trait] +impl WorkdirSession for RetainedWorkdirSession { + fn workdir(&self) -> &Workdir { + self.inner.workdir() + } + + fn capabilities(&self) -> WorkdirSessionCapabilities { + self.inner.capabilities() + } + + async fn authorize_scope_path( + &self, + request: WorkdirScopeAuthorizationRequest, + ) -> Result<(), WorkdirError> { + self.inner.authorize_scope_path(request).await + } + + async fn scope_rules_overlap( + &self, + request: WorkdirScopeOverlapRequest, + ) -> Result { + self.inner.scope_rules_overlap(request).await + } + + async fn checkout_search( + &self, + request: CheckoutSearchRequest, + ) -> Result { + self.inner.checkout_search(request).await + } + + async fn checkout_observe( + &self, + path: WorkdirPath, + ) -> Result { + self.inner.checkout_observe(path).await + } + + async fn checkout_execute( + &self, + request: CheckoutRequest, + ) -> Result { + self.inner.checkout_execute(request).await + } + + async fn stat(&self, request: StatRequest) -> Result { + self.inner.stat(request).await + } + + async fn read(&self, request: ReadRequest) -> Result { + self.inner.read(request).await + } + + async fn read_bytes(&self, request: ReadBytesRequest) -> Result { + self.inner.read_bytes(request).await + } + + async fn write(&self, request: WriteRequest) -> Result { + self.inner.write(request).await + } + + async fn edit(&self, request: EditRequest) -> Result { + self.inner.edit(request).await + } + + async fn list(&self, request: ListRequest) -> Result { + self.inner.list(request).await + } + + async fn glob(&self, request: GlobRequest) -> Result { + self.inner.glob(request).await + } + + async fn grep(&self, request: GrepRequest) -> Result { + self.inner.grep(request).await + } + + async fn start_command(&self, request: CommandRequest) -> Result { + self.inner.start_command(request).await + } + + async fn command_status(&self, handle: CommandHandle) -> Result { + self.inner.command_status(handle).await + } + + async fn command_output( + &self, + request: CommandOutputRequest, + ) -> Result { + self.inner.command_output(request).await + } + + async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> { + self.inner.cancel_command(handle).await + } + + fn subscribe_command_events(&self) -> Option> { + self.inner.subscribe_command_events() + } + + fn command_snapshot(&self) -> Vec { + self.inner.command_snapshot() + } + + async fn close(&self) -> Result<(), WorkdirError> { + // Provider close must succeed before releasing the active-use reservation. + // No wrapper-local terminal checks may bypass the provider's own checks. + self.inner.close().await?; + let operation = self + .operation + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); + // Release outside the mutex: guard Drop may acquire Runtime state locks. + drop(operation); + Ok(()) + } +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 9ac79973b..801838b70 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1025,7 +1025,9 @@ fn runtime_local_workdir_router( command_output_scope.clone(), capabilities, binding.command_environment(), - binding.session_resources(), + binding.session_resources().map_err(|error| { + format!("open Workdir attachment `{alias}` resources: {error}") + })?, ), ) .map_err(|error| format!("bind Workdir attachment `{alias}`: {error}"))?; @@ -3453,7 +3455,7 @@ where scope, WorkdirSessionCapabilities::ALL, binding.command_environment(), - binding.session_resources(), + binding.session_resources()?, )) } @@ -9501,6 +9503,42 @@ mod tests { assert_eq!(status.summary.cleanliness.as_deref(), Some("clean")); } + #[cfg(target_os = "linux")] + #[tokio::test] + async fn local_sessions_exclude_cleanup_until_close_releases_resources() { + let runtime_base = tempfile::tempdir().unwrap(); + let repo = create_clean_repo(); + let materializer = RuntimeGitMaterializer::new(runtime_base.path()); + let binding = materializer + .create(&working_directory_request(repo.path())) + .unwrap(); + let id = &binding.working_directory.id; + let backend = WorkerRuntimeExecutionBackend::new(FailingFactory) + .unwrap() + .with_working_directory_materializer(materializer.clone()); + let session = backend.open_workdir_session(id).unwrap(); + let session_clone = session.clone(); + drop(session); + assert_eq!( + materializer.cleanup_working_directory(id).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + + let alias = WorkdirAttachmentAlias::new("checkout").unwrap(); + let attachments = BTreeMap::from([(alias, binding.clone())]); + let scope = manifest::SharedScope::new(manifest::Scope::writable(binding.root()).unwrap()); + let router = runtime_local_workdir_router(&attachments, &BTreeMap::new(), scope).unwrap(); + session_clone.close().await.unwrap(); + assert_eq!( + materializer.cleanup_working_directory(id).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + router.close_all().await.unwrap(); + // Closed handles, router and passive bindings can remain alive. + materializer.cleanup_working_directory(id).unwrap(); + assert!(!binding.working_directory_root().exists()); + } + #[test] fn spawn_failure_with_existing_working_directory_preserves_workdir() { let runtime_base = tempfile::tempdir().unwrap(); diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index 178850a55..dc27551a2 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -23,6 +23,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use workdir::WorkdirSessionResource; mod cleanup; +mod occupancy; #[cfg(all(test, target_os = "linux"))] pub(crate) use cleanup::tests::fail_once as fail_workdir_cleanup_once; @@ -101,6 +102,9 @@ pub struct WorkingDirectoryBinding { working_directory_root: PathBuf, command_environment: BTreeMap, session_resources: Vec>, + occupancy: occupancy::Occupancy, + occupancy_id: String, + generation: occupancy::Generation, } impl WorkingDirectoryBinding { @@ -120,8 +124,23 @@ impl WorkingDirectoryBinding { self.command_environment.clone() } - pub fn session_resources(&self) -> Vec> { - self.session_resources.clone() + /// Activate resources for a local session. Cloning a binding alone is passive; + /// the returned lease excludes cleanup until the last resource clone drops. + pub fn session_resources( + &self, + ) -> Result>, WorkingDirectoryDiagnostic> { + let lease = self + .occupancy + .acquire_session_use(&self.occupancy_id, &self.generation)?; + if !binding_paths_are_available(self) { + return Err(WorkingDirectoryDiagnostic::new( + "working_directory_not_found", + "Workdir binding is no longer available; acquire a fresh binding", + )); + } + let mut resources = self.session_resources.clone(); + resources.push(Arc::new(lease)); + Ok(resources) } pub fn status(&self) -> WorkingDirectoryStatus { @@ -583,6 +602,7 @@ fn repository_access_now_epoch_seconds() -> u64 { pub struct RuntimeGitMaterializer { runtime_root: PathBuf, repository_access: RepositoryAccessExpiryScheduler, + occupancy: occupancy::Occupancy, } impl RuntimeGitMaterializer { @@ -590,6 +610,7 @@ impl RuntimeGitMaterializer { Self { runtime_root: runtime_root.into(), repository_access: RepositoryAccessExpiryScheduler::new(), + occupancy: occupancy::Occupancy::default(), } } @@ -702,6 +723,9 @@ impl RuntimeGitMaterializer { working_directory_root, command_environment: BTreeMap::new(), session_resources: Vec::new(), + occupancy: self.occupancy.clone(), + occupancy_id: working_directory_id.to_string(), + generation: self.occupancy.generation(&working_directory_id), }) } @@ -904,6 +928,7 @@ impl RuntimeGitMaterializer { request: &WorkingDirectoryRequest, ) -> Result { validate_working_directory_id(&working_directory_id)?; + let _creation_lease = self.occupancy.acquire_creation(&working_directory_id)?; Self::validate_plain_http_source(request)?; let request = self.request_with_authorized_repository_access(&working_directory_id, request)?; @@ -911,20 +936,30 @@ impl RuntimeGitMaterializer { let selector = request.repository.selector.as_deref().unwrap_or("HEAD"); let working_directory_root = self.working_directory_root(&working_directory_id); let checkout_root = working_directory_root.join(CHECKOUT_DIR); - if checkout_root.exists() { + if working_directory_root.symlink_metadata().is_ok() { return Err(WorkingDirectoryDiagnostic::new( "working_directory_exists", "working directory target already exists; cleanup or choose a new working_directory", )); } - fs::create_dir_all(&working_directory_root).map_err(|_| { + fs::create_dir_all(&self.runtime_root).map_err(|_| { + WorkingDirectoryDiagnostic::new( + "working_directory_create_failed", + "failed to create working directory storage", + ) + })?; + // Prepare fallible access before owning a root that must not be adopted + // by a retry after failure. + let access = RepositoryCommandAccess::prepare(&self.runtime_root, &request)?; + // Atomically own this new root before any rollback can remove it. Do + // not adopt an existing partial-cleanup directory or another creator's root. + fs::create_dir(&working_directory_root).map_err(|_| { WorkingDirectoryDiagnostic::new( "working_directory_create_failed", "failed to create working directory; backend-private path details were omitted", ) })?; - let access = RepositoryCommandAccess::prepare(&self.runtime_root, &request)?; let mut clone = repository_git_command(&request, access.as_ref()); clone .args([ @@ -1022,6 +1057,9 @@ impl RuntimeGitMaterializer { working_directory_root: working_directory_root.clone(), command_environment: BTreeMap::new(), session_resources: Vec::new(), + occupancy: self.occupancy.clone(), + occupancy_id: working_directory_id.to_string(), + generation: self.occupancy.generation(&working_directory_id), }; if let Err(error) = self.write_record(&binding) { let _ = fs::remove_dir_all(&working_directory_root); @@ -1046,7 +1084,10 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer { worker_ref: &WorkerRef, request: &WorkingDirectoryRequest, ) -> Result { - let working_directory_id = Self::working_directory_id(worker_ref, &request.repository.id); + let working_directory_id = request + .backend_workdir_id + .clone() + .unwrap_or_else(|| Self::working_directory_id(worker_ref, &request.repository.id)); self.materialize_with_working_directory_id(working_directory_id, request) } @@ -1066,6 +1107,7 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer { request: &WorkingDirectoryRepositoryAccessRequest, ) -> Result<(), WorkingDirectoryDiagnostic> { validate_working_directory_id(&request.working_directory_id)?; + let _use_lease = self.occupancy.acquire_use(&request.working_directory_id)?; let ssh = request.materialization.ssh.as_ref().ok_or_else(|| { WorkingDirectoryDiagnostic::new( "working_directory_remote_repository_access_required", @@ -1187,6 +1229,8 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer { working_directory_id: &str, relative_cwd: Option<&str>, ) -> Result { + validate_working_directory_id(working_directory_id)?; + let _use_lease = self.occupancy.acquire_use(working_directory_id)?; let binding = self.checked_binding(working_directory_id, relative_cwd)?; self.bind_repository_access(working_directory_id, binding) } @@ -1281,7 +1325,10 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer { working_directory_id: &str, ) -> Result { validate_working_directory_id(working_directory_id)?; - cleanup::remove(self, working_directory_id) + let _cleanup_lease = self.occupancy.acquire_cleanup(working_directory_id)?; + let status = cleanup::remove(self, working_directory_id)?; + self.occupancy.invalidate_generation(working_directory_id); + Ok(status) } fn cleanup(&self, binding: &WorkingDirectoryBinding) -> Result<(), WorkingDirectoryDiagnostic> { @@ -2878,7 +2925,7 @@ fn sanitize_path_component(value: &str) -> String { } } -fn next_working_directory_id(_repository_id: &str) -> String { +pub(crate) fn next_working_directory_id(_repository_id: &str) -> String { let now = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|duration| duration.as_millis() as u64) @@ -3933,7 +3980,8 @@ mod tests { #[cfg(target_os = "linux")] { // Real SSH reauthorization updated access evidence repeatedly, but - // did not replace this clean materialization or require manual recovery. + // did not replace this materialization or require manual recovery. + // Status does not inspect content; cleanup still verifies it below. assert_eq!( restored .working_directory_status(&id) @@ -3941,7 +3989,7 @@ mod tests { .summary .cleanliness .as_deref(), - Some("clean") + Some("unknown") ); restored.cleanup_working_directory(&id).unwrap(); assert!(!restored.working_directory_root(&id).exists()); @@ -4443,6 +4491,307 @@ mod tests { assert!(root.exists()); } + fn materialization_context( + ssh: Option, + ) -> crate::catalog::RepositoryMaterializationContext { + crate::catalog::RepositoryMaterializationContext { + workspace_id: "workspace-1".to_string(), + runtime_id: "runtime-1".to_string(), + operation_id: "operation-1".to_string(), + config_revision: 7, + config_projection_digest: "sha256:projection".to_string(), + ssh, + } + } + + #[cfg(target_os = "linux")] + #[test] + fn live_session_resources_block_cleanup_until_last_clone_drops() { + let repo = create_clean_repo(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let binding = materializer.create(&request(repo.path())).unwrap(); + let id = &binding.working_directory.id; + // Read bindings and materializer clones must share the same admission. + let restored = materializer.bind_working_directory(id, None).unwrap(); + let resources = restored.session_resources().unwrap(); + let resource_clone = resources.clone(); + let second_resources = binding.session_resources().unwrap(); + let clone = materializer.clone(); + + for _ in 0..2 { + assert_eq!( + clone.cleanup_working_directory(id).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + } + assert!(binding.root().join("README.md").exists()); + drop(resources); + drop(second_resources); + assert_eq!( + clone.cleanup_working_directory(id).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + drop(resource_clone); + clone.cleanup_working_directory(id).unwrap(); + // Both bindings are still alive and passive after resource release. + assert!(!binding.working_directory_root().exists()); + assert!(!restored.root().exists()); + } + + #[test] + fn active_cleanup_rejects_creation_and_access_before_effects() { + let repo = tempfile::tempdir().unwrap(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let id = "workdir-cleanup-active"; + let cleanup = materializer.occupancy.acquire_cleanup(id).unwrap(); + let mut create_request = request(repo.path()); + create_request.backend_workdir_id = Some(id.to_string()); + let access_request = WorkingDirectoryRepositoryAccessRequest { + working_directory_id: id.to_string(), + materialization: materialization_context(Some(repository_ssh_access())), + }; + + assert_eq!( + materializer.create(&create_request).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + assert_eq!( + materializer + .materialize(&worker_ref(1), &create_request) + .unwrap_err() + .code, + "working_directory_cleanup_resource_busy" + ); + assert_eq!( + materializer + .authorize_repository_access(&access_request) + .unwrap_err() + .code, + "working_directory_cleanup_resource_busy" + ); + assert_eq!( + materializer + .bind_working_directory(id, None) + .unwrap_err() + .code, + "working_directory_cleanup_resource_busy" + ); + assert!(!materializer.working_directory_root(id).exists()); + assert_eq!(materializer.repository_access.pending_count(), 0); + drop(cleanup); + // The cleanup reservation was transient, not a persisted lifecycle. + assert_ne!( + materializer.create(&create_request).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn active_cleanup_rejects_existing_binding_resources_but_not_other_ids() { + let repo = create_clean_repo(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let binding = materializer.create(&request(repo.path())).unwrap(); + let id = &binding.working_directory.id; + let cleanup = materializer.occupancy.acquire_cleanup(id).unwrap(); + assert_eq!( + binding.session_resources().unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + assert_eq!( + materializer.cleanup_working_directory(id).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + let other = materializer.create(&request(repo.path())).unwrap(); + let other_resources = other.session_resources().unwrap(); + drop(other_resources); + materializer + .cleanup_working_directory(&other.working_directory.id) + .unwrap(); + drop(cleanup); + let resources = binding.session_resources().unwrap(); + drop(resources); + materializer.cleanup_working_directory(id).unwrap(); + } + + #[cfg(target_os = "linux")] + #[test] + fn cleanup_error_releases_lease_for_session_and_safe_retry() { + let repo = create_clean_repo(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let binding = materializer.create(&request(repo.path())).unwrap(); + let changed = binding.root().join("untracked.txt"); + fs::write(&changed, "preserve me\n").unwrap(); + assert_eq!( + materializer + .cleanup_working_directory(&binding.working_directory.id) + .unwrap_err() + .code, + "working_directory_cleanup_changes_present" + ); + let resources = binding.session_resources().unwrap(); + fs::remove_file(changed).unwrap(); + drop(resources); + materializer + .cleanup_working_directory(&binding.working_directory.id) + .unwrap(); + } + + #[cfg(target_os = "linux")] + #[test] + fn removed_binding_cannot_activate_against_recreated_same_id() { + let repo = create_clean_repo(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let mut create = request(repo.path()); + create.backend_workdir_id = Some("workdir-reused-id".into()); + let old = materializer.create(&create).unwrap(); + materializer + .cleanup_working_directory(&old.working_directory.id) + .unwrap(); + assert_eq!( + old.session_resources().unwrap_err().code, + "working_directory_not_found" + ); + let fresh = materializer.create(&create).unwrap(); + assert_eq!( + old.session_resources().unwrap_err().code, + "working_directory_not_found" + ); + let resources = fresh.session_resources().unwrap(); + assert_eq!( + materializer + .cleanup_working_directory(&fresh.working_directory.id) + .unwrap_err() + .code, + "working_directory_cleanup_resource_busy" + ); + drop(resources); + materializer + .cleanup_working_directory(&fresh.working_directory.id) + .unwrap(); + } + + #[test] + fn same_id_creation_is_exclusive_and_never_adopts_existing_partial_root() { + let repo = tempfile::tempdir().unwrap(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let id = "workdir-creation-owner"; + let mut create = request(repo.path()); + create.backend_workdir_id = Some(id.into()); + let creation = materializer.occupancy.acquire_creation(id).unwrap(); + assert_eq!( + materializer.clone().create(&create).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + assert_eq!( + materializer + .bind_working_directory(id, None) + .unwrap_err() + .code, + "working_directory_cleanup_resource_busy" + ); + assert_eq!( + materializer.cleanup_working_directory(id).unwrap_err().code, + "working_directory_cleanup_resource_busy" + ); + assert!(!materializer.working_directory_root(id).exists()); + drop(creation); + let partial = materializer.working_directory_root(id); + fs::create_dir(&partial).unwrap(); + fs::write( + partial.join("owner-content"), + "preserve partial removal state", + ) + .unwrap(); + assert_eq!( + materializer.create(&create).unwrap_err().code, + "working_directory_exists" + ); + assert_eq!( + fs::read_to_string(partial.join("owner-content")).unwrap(), + "preserve partial removal state" + ); + } + + #[test] + fn failed_access_preparation_does_not_strand_a_creation_root() { + let repo = tempfile::tempdir().unwrap(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let mut create = request(repo.path()); + create.backend_workdir_id = Some("workdir-access-setup-failure".into()); + let mut ssh = repository_ssh_access(); + ssh.known_hosts_entry = + crate::catalog::SensitiveString::new("example.test ssh-ed25519 placeholder"); + create.repository.source = server_api::RepositorySource { + kind: server_api::RepositorySourceKind::Ssh, + uri: ssh.repository_uri.clone(), + }; + create.materialization = Some(materialization_context(Some(ssh))); + // A local storage failure stops access setup before any SSH transport or + // agent can start. The materialization must remain retryable, not adopted. + fs::write( + runtime_root.path().join(REPOSITORY_ACCESS_DIR), + "storage obstruction", + ) + .unwrap(); + assert_eq!( + materializer.create(&create).unwrap_err().code, + "working_directory_repository_access_setup_failed" + ); + assert!( + !materializer + .working_directory_root(create.backend_workdir_id.as_deref().unwrap()) + .exists() + ); + } + + #[test] + fn failed_creation_and_access_release_use_leases() { + let repo = tempfile::tempdir().unwrap(); + let runtime_root = tempfile::tempdir().unwrap(); + let materializer = RuntimeGitMaterializer::new(runtime_root.path()); + let id = "workdir-effect-errors"; + let mut create_request = request(repo.path()); + create_request.backend_workdir_id = Some(id.to_string()); + // Validation fails after admission but before effects. + create_request.repository.provider = "unsupported".to_string(); + assert_eq!( + materializer.create(&create_request).unwrap_err().code, + "working_directory_repository_provider_unsupported" + ); + drop(materializer.occupancy.acquire_cleanup(id).unwrap()); + // A clone failure has already created the target and must also release + // admission after rolling its filesystem effects back. + create_request.repository.provider = "git".to_string(); + assert_eq!( + materializer.create(&create_request).unwrap_err().code, + "working_directory_repository_clone_failed" + ); + assert!(!materializer.working_directory_root(id).exists()); + drop(materializer.occupancy.acquire_cleanup(id).unwrap()); + let access_request = WorkingDirectoryRepositoryAccessRequest { + working_directory_id: id.to_string(), + materialization: materialization_context(None), + }; + assert_eq!( + materializer + .authorize_repository_access(&access_request) + .unwrap_err() + .code, + "working_directory_remote_repository_access_required" + ); + drop(materializer.occupancy.acquire_cleanup(id).unwrap()); + assert!(materializer.bind_working_directory(id, None).is_err()); + drop(materializer.occupancy.acquire_cleanup(id).unwrap()); + } + #[test] fn cleanup_working_directory_removes_clone_and_record() { let repo = create_clean_repo(); diff --git a/crates/worker-runtime/src/working_directory/cleanup.rs b/crates/worker-runtime/src/working_directory/cleanup.rs index 037f01343..a9a2c3a50 100644 --- a/crates/worker-runtime/src/working_directory/cleanup.rs +++ b/crates/worker-runtime/src/working_directory/cleanup.rs @@ -76,19 +76,11 @@ pub(super) fn remaining_status( } let mut summary = saved.working_directory.status_summary(); summary.status = WorkingDirectoryStatusKind::CleanupPending; - summary.cleanliness = Some( - match check_mounts( - &materializer.working_directory_root(id), - &fs::read("/proc/self/mountinfo").ok()?, - ) - .and_then(|()| retry_inventory(materializer, id, &tree, &saved)) - { - Ok(_) => "clean", - Err(error) if error.code == "working_directory_cleanup_changes_present" => "dirty", - Err(_) => "unknown", - } - .to_string(), - ); + // Observation is not removal admission: do not inventory or hash the + // surviving checkout just to list it. The witness can be stale, so neither + // its old cleanliness nor a successful identity check proves current content + // is safe to remove. Actual retries still perform the full safety checks. + summary.cleanliness = Some("unknown".to_string()); Some(WorkingDirectoryStatus { summary }) } @@ -974,7 +966,7 @@ pub(super) mod tests { } #[test] - fn cleanup_pending_rechecks_current_cleanliness_and_retries_without_state_edit() { + fn cleanup_pending_observation_is_unknown_and_retry_checks_safety() { let (_runtime, materializer, binding) = fixture(); let id = &binding.working_directory.id; fail_once(&materializer, id); @@ -983,12 +975,54 @@ pub(super) mod tests { status.summary.status, WorkingDirectoryStatusKind::CleanupPending ); - assert_eq!(status.summary.cleanliness.as_deref(), Some("clean")); + assert_eq!(status.summary.cleanliness.as_deref(), Some("unknown")); let removed = materializer.cleanup_working_directory(id).unwrap(); assert_eq!(removed.summary.status, WorkingDirectoryStatusKind::NotFound); assert!(!binding.working_directory_root.exists()); } + #[test] + fn cleanup_pending_observation_does_not_open_checkout_content() { + use std::os::fd::{AsRawFd, FromRawFd}; + use std::os::unix::ffi::OsStrExt; + + let (_runtime, materializer, binding) = fixture(); + let id = &binding.working_directory.id; + fail_once(&materializer, id); + + // Observe actual file opens rather than a timing threshold or a large + // fixture. Inventory would open this surviving tracked file to hash it. + // SAFETY: inotify_init1 has no pointer arguments. + let fd = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) }; + assert!(fd >= 0, "{}", io::Error::last_os_error()); + // SAFETY: fd is a freshly created descriptor, owned only by this File. + let mut events = unsafe { fs::File::from_raw_fd(fd) }; + let path = + std::ffi::CString::new(binding.root.join("README.md").as_os_str().as_bytes()).unwrap(); + // SAFETY: the descriptor is live and path is a valid terminated string. + let watch = + unsafe { libc::inotify_add_watch(events.as_raw_fd(), path.as_ptr(), libc::IN_OPEN) }; + assert!(watch >= 0, "{}", io::Error::last_os_error()); + + let detail = materializer.working_directory_status(id).unwrap(); + let listed = materializer.list_working_directories().unwrap(); + let mut buffer = [0u8; 4096]; + assert_eq!( + events + .read(&mut buffer) + .expect_err("status observation opened surviving checkout content") + .kind(), + io::ErrorKind::WouldBlock + ); + assert_eq!( + detail.summary.status, + WorkingDirectoryStatusKind::CleanupPending + ); + assert_eq!(detail.summary.cleanliness.as_deref(), Some("unknown")); + assert_eq!(listed.len(), 1); + assert_eq!(listed[0].summary, detail.summary); + } + #[test] fn partial_deletion_missing_checkout_and_restart_finish_with_durable_content_authority() { for removed in [ @@ -1021,7 +1055,7 @@ pub(super) mod tests { retained.summary.cleanup_target.as_ref(), Some(&binding.working_directory.cleanup_target) ); - assert_eq!(retained.summary.cleanliness.as_deref(), Some("clean")); + assert_eq!(retained.summary.cleanliness.as_deref(), Some("unknown")); assert_eq!( restarted.list_working_directories().unwrap()[0].summary, retained.summary @@ -1074,7 +1108,7 @@ pub(super) mod tests { .summary .cleanliness .as_deref(), - Some("dirty") + Some("unknown") ); assert_eq!( materializer.cleanup_working_directory(id).unwrap_err().code, @@ -1117,7 +1151,7 @@ pub(super) mod tests { .summary .cleanliness .as_deref(), - Some("clean"), + Some("unknown"), "{resolution}" ); // Renewal itself can fail at unlink. Its new witness must survive a diff --git a/crates/worker-runtime/src/working_directory/occupancy.rs b/crates/worker-runtime/src/working_directory/occupancy.rs new file mode 100644 index 000000000..c3e477bb5 --- /dev/null +++ b/crates/worker-runtime/src/working_directory/occupancy.rs @@ -0,0 +1,196 @@ +//! Transient, clone-shared admission for local filesystem effects. Bindings are +//! passive; only effects and activated session resources own leases. The mutex +//! protects admission bookkeeping only, never filesystem work. +use super::WorkingDirectoryDiagnostic; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +#[derive(Clone, Debug, Default)] +pub(super) struct Occupancy { + entries: Arc>>, + generations: Arc>>>, +} + +/// Passive bindings identify a materialization generation, not merely its ID. +#[derive(Clone, Debug)] +pub(super) struct Generation(Arc<()>); + +#[derive(Debug)] +enum Entry { + Use(usize), + Create, + Cleanup, +} + +#[derive(Debug)] +pub(super) struct Lease { + occupancy: Occupancy, + id: String, +} + +fn busy() -> WorkingDirectoryDiagnostic { + WorkingDirectoryDiagnostic::new( + "working_directory_cleanup_resource_busy", + "A filesystem resource is busy or changed during removal. Release the resource and retry.", + ) +} + +impl Occupancy { + pub(super) fn acquire_use(&self, id: &str) -> Result { + let mut entries = self + .entries + .lock() + .unwrap_or_else(|error| error.into_inner()); + match entries.get_mut(id) { + Some(Entry::Cleanup | Entry::Create) => return Err(busy()), + Some(Entry::Use(count)) => *count += 1, + None => { + entries.insert(id.to_string(), Entry::Use(1)); + } + } + Ok(Lease { + occupancy: self.clone(), + id: id.to_string(), + }) + } + + pub(super) fn generation(&self, id: &str) -> Generation { + let mut generations = self + .generations + .lock() + .unwrap_or_else(|error| error.into_inner()); + Generation( + generations + .entry(id.to_string()) + .or_insert_with(|| Arc::new(())) + .clone(), + ) + } + + // Called only after successful removal, while exclusive cleanup is held. + pub(super) fn invalidate_generation(&self, id: &str) { + self.generations + .lock() + .unwrap_or_else(|error| error.into_inner()) + .remove(id); + } + + pub(super) fn acquire_session_use( + &self, + id: &str, + generation: &Generation, + ) -> Result { + let lease = self.acquire_use(id)?; + let valid = self + .generations + .lock() + .unwrap_or_else(|error| error.into_inner()) + .get(id) + .is_some_and(|current| Arc::ptr_eq(current, &generation.0)); + if !valid { + return Err(WorkingDirectoryDiagnostic::new( + "working_directory_not_found", + "Workdir binding is stale; acquire a fresh binding before opening a session", + )); + } + Ok(lease) + } + + pub(super) fn acquire_creation(&self, id: &str) -> Result { + let mut entries = self + .entries + .lock() + .unwrap_or_else(|error| error.into_inner()); + if entries.contains_key(id) { + return Err(busy()); + } + entries.insert(id.to_string(), Entry::Create); + Ok(Lease { + occupancy: self.clone(), + id: id.to_string(), + }) + } + + pub(super) fn acquire_cleanup(&self, id: &str) -> Result { + let mut entries = self + .entries + .lock() + .unwrap_or_else(|error| error.into_inner()); + if entries.contains_key(id) { + return Err(busy()); + } + entries.insert(id.to_string(), Entry::Cleanup); + Ok(Lease { + occupancy: self.clone(), + id: id.to_string(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn use_and_cleanup_leases_release_during_unwind() { + let occupancy = Occupancy::default(); + for operation in ["use", "create", "cleanup"] { + let result = std::panic::catch_unwind(|| { + let _lease = match operation { + "cleanup" => occupancy.acquire_cleanup("workdir-panic").unwrap(), + "create" => occupancy.acquire_creation("workdir-panic").unwrap(), + _ => occupancy.acquire_use("workdir-panic").unwrap(), + }; + panic!("effect failed after admission"); + }); + assert!(result.is_err()); + drop(occupancy.acquire_use("workdir-panic").unwrap()); + drop(occupancy.acquire_cleanup("workdir-panic").unwrap()); + } + } + + #[test] + fn simultaneous_use_and_cleanup_admit_only_one_direction() { + let occupancy = Occupancy::default(); + let barrier = Arc::new(std::sync::Barrier::new(2)); + std::thread::scope(|scope| { + let use_occupancy = occupancy.clone(); + let use_barrier = barrier.clone(); + let use_thread = scope.spawn(move || { + use_barrier.wait(); + let result = use_occupancy.acquire_use("workdir-race"); + // Keep the winner's lease until both admission attempts finish. + use_barrier.wait(); + result + }); + barrier.wait(); + let cleanup = occupancy.acquire_cleanup("workdir-race"); + barrier.wait(); + let use_result = use_thread.join().unwrap(); + assert_ne!(use_result.is_ok(), cleanup.is_ok()); + let error = if let Err(error) = use_result { + error + } else { + cleanup.unwrap_err() + }; + assert_eq!(error.code, "working_directory_cleanup_resource_busy"); + }); + drop(occupancy.acquire_cleanup("workdir-race").unwrap()); + } +} + +impl Drop for Lease { + fn drop(&mut self) { + let mut entries = self + .occupancy + .entries + .lock() + .unwrap_or_else(|error| error.into_inner()); + match entries.get_mut(&self.id) { + Some(Entry::Use(count)) if *count > 1 => *count -= 1, + _ => { + entries.remove(&self.id); + } + } + } +} diff --git a/docs/development/workdir-removal.md b/docs/development/workdir-removal.md index 450d0ac5f..d0f6f6a9f 100644 --- a/docs/development/workdir-removal.md +++ b/docs/development/workdir-removal.md @@ -44,6 +44,11 @@ removal commits registry deletion only after Runtime confirms physical deletion or authoritative not-found. An unknown result leaves the registry in place so retry can reconcile the provider's current state. +A retained partial-removal Workdir is listed as `cleanup_pending` with cleanliness +`unknown`. List/detail observation checks its saved identity but does not read or +hash surviving checkout content. The saved removal witness is not a current +cleanliness assertion; only an actual removal retry runs the full content checks. + ## Runtime safety boundary The Runtime keeps removal content/identity evidence outside the tree being @@ -55,9 +60,22 @@ ignored/untracked content. Do not edit this witness manually. Mount-safe removal currently requires Linux `openat2` and `/proc` inspection. Unavailable inspection fails closed with `mount_check_unavailable`; there is no fallback to recursive deletion across an unchecked filesystem boundary. -Removal retains the Runtime occupancy mutex through its filesystem effect. -Unrelated Runtime state operations can therefore wait during hashing/unlink; -releasing it would require equivalent exclusion on all attachment paths. +Runtime HTTP list, detail and removal requests run synchronous provider work on +blocking workers rather than the HTTP executor. Removal reserves only the target +Workdir under the Runtime state mutex, then releases that mutex before content +inspection and unlink. Unrelated Worker list/detail, session observation and +subscription snapshots remain available while removal runs. Conflicting use, +attachment, creation or removal of the same Workdir is rejected promptly rather +than waiting under the global mutex. + +Provider leases protect in-flight creation/access and live session resources, +including their command tasks. Cloned sessions retain protection until safe close +or final resource release; passive bindings alone do not prevent removal. Removed +bindings cannot activate against a recreated Workdir with the same ID. Creation +owns a fresh root exclusively so failed creation cannot roll back another +creator's checkout or adopt an existing partial-removal directory. Reservations +are transient and released on provider success, error or unwind; they do not +replace the durable cleanup witness or attachment authority. ## Diagnostics