From defaf8f369524f711ec9ca6ebf943040656a8804 Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 10 Oct 2026 13:19:52 +0900 Subject: [PATCH] fix: allow concurrent Workspace requests with deletion admission fencing --- crates/workspace-server/Cargo.toml | 2 +- crates/workspace-server/src/server.rs | 196 ++-- .../src/server/tests/subject_spawn_tests.rs | 38 +- .../server/tests/workspace_admission_tests.rs | 1012 +++++++++++++++++ .../src/server/workspace_admission.rs | 104 ++ crates/workspace-server/src/worker_source.rs | 1 + .../src/worker_source/runtime_request.rs | 28 +- .../workspace-request-admission.md | 34 + .../t-732-subworker-checkout-attachment.md | 9 + 9 files changed, 1300 insertions(+), 124 deletions(-) create mode 100644 crates/workspace-server/src/server/tests/workspace_admission_tests.rs create mode 100644 crates/workspace-server/src/server/workspace_admission.rs create mode 100644 docs/development/workspace-request-admission.md create mode 100644 docs/report/t-732-subworker-checkout-attachment.md diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 7b50a34cf..390ad81dd 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -62,5 +62,5 @@ tempfile.workspace = true tools.workspace = true yoi = { path = "../yoi" } tower = { workspace = true, features = ["util"] } -tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } +tokio = { workspace = true, features = ["macros", "rt-multi-thread", "test-util"] } webauthn-authenticator-rs = { version = "0.5.5", default-features = false, features = ["softpasskey"] } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 134cf130d..6fd3e3741 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1594,6 +1594,7 @@ pub struct WorkspaceApi { pub(crate) worker_projection: Arc, resource_broker: BackendResourceBroker, workdir_sessions: Arc>, + memory_operation_lock: Arc>, external_workdir_providers: Arc>, external_workdir_expiry_tasks: Arc>>, workdir_session_locks: Arc>>>>, @@ -2164,36 +2165,8 @@ impl WorkerRemovalService { } } -/// Session attribution is a synchronous callback from Worker create/restore. -/// It must not wait for the request that is waiting for that callback. Deletion -/// acquires both lanes, in request-first order, so it cannot fence out a callback -/// needed by an in-flight create/restore while waiting for that request to finish. -#[derive(Default)] -struct WorkspaceMutationGate { - requests: AsyncMutex<()>, - session_attributions: AsyncMutex<()>, -} - -impl WorkspaceMutationGate { - fn request_lock(&self, method: &Method, path: &str, workspace_id: &str) -> &AsyncMutex<()> { - if *method == Method::POST && path == format!("/api/w/{workspace_id}/subjektiv/sessions") { - &self.session_attributions - } else { - &self.requests - } - } - - async fn lock_deletion( - &self, - ) -> ( - tokio::sync::MutexGuard<'_, ()>, - tokio::sync::MutexGuard<'_, ()>, - ) { - let requests = self.requests.lock().await; - let attributions = self.session_attributions.lock().await; - (requests, attributions) - } -} +mod workspace_admission; +use workspace_admission::WorkspaceAdmission; #[derive(Clone)] pub struct WorkspaceServerApi { @@ -2203,19 +2176,20 @@ pub struct WorkspaceServerApi { signing_materials: Arc, routers: Arc>>, apis: Arc>>, - mutation_locks: Arc>>>, + admission_gates: Arc>>>, running_deletions: Arc>>, + proof_clock: Arc i64 + Send + Sync>, hook_handles: Arc>>, } -async fn workspace_mutation_lock( - locks: &Arc>>>, +async fn workspace_admission_gate( + locks: &Arc>>>, workspace_id: &str, -) -> Arc { +) -> Arc { let mut locks = locks.lock().await; locks .entry(workspace_id.to_string()) - .or_insert_with(|| Arc::new(WorkspaceMutationGate::default())) + .or_insert_with(|| Arc::new(WorkspaceAdmission::default())) .clone() } @@ -2232,14 +2206,17 @@ impl WorkspaceServerApi { store, routers: Arc::new(AsyncMutex::new(HashMap::new())), apis: Arc::new(AsyncMutex::new(HashMap::new())), - mutation_locks: Arc::new(AsyncMutex::new(HashMap::new())), + admission_gates: Arc::new(AsyncMutex::new(HashMap::new())), running_deletions: Arc::new(AsyncMutex::new(HashSet::new())), + proof_clock: Arc::new(|| { + i64::try_from(worker_runtime::auth::unix_now_seconds()).unwrap_or(i64::MAX) + }), hook_handles: Arc::new(AsyncMutex::new(HashMap::new())), } } - async fn mutation_lock(&self, workspace_id: &str) -> Arc { - workspace_mutation_lock(&self.mutation_locks, workspace_id).await + async fn admission_for_workspace(&self, workspace_id: &str) -> Arc { + workspace_admission_gate(&self.admission_gates, workspace_id).await } async fn api_for_workspace(&self, workspace_id: &str) -> Result> { @@ -2479,7 +2456,7 @@ impl WorkspaceServerApi { handle.abort(); } self.apis.lock().await.remove(&completed.workspace_id); - self.mutation_locks + self.admission_gates .lock() .await .remove(&completed.workspace_id); @@ -2735,7 +2712,7 @@ async fn authorize_scoped_workspace_request( } else { worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION }; - let source = crate::worker_source::verify_runtime_request_source_proof_with_store( + let source = crate::worker_source::verify_runtime_request_source_proof_with_clock( api.store.as_ref(), api.template.as_ref(), &proof, @@ -2744,6 +2721,7 @@ async fn authorize_scoped_workspace_request( &method, &path, &digest, + api.proof_clock.as_ref(), ) .await .map_err(|error| { @@ -2757,23 +2735,7 @@ async fn authorize_scoped_workspace_request( )); } request.extensions_mut().insert(source); - if !matches!( - *request.method(), - Method::GET | Method::HEAD | Method::OPTIONS - ) && !api - .store - .get_workspace(workspace_id) - .await - .map_err(server_error_response)? - .is_some_and(|workspace| workspace.state == "active") - { - return Err(repository_api_rejection( - &request_path, - StatusCode::CONFLICT, - "Workspace is deleting and no longer accepts mutations", - )); - } - return Ok(()); + return require_active_scoped_workspace(api, workspace_id, &request_path).await; } let actor = resolve_server_actor(api, request.headers()) @@ -2809,21 +2771,32 @@ async fn authorize_scoped_workspace_request( } } request.extensions_mut().insert(actor); - if mutating - && !api - .store - .get_workspace(workspace_id) - .await - .map_err(server_error_response)? - .is_some_and(|workspace| workspace.state == "active") + require_active_scoped_workspace(api, workspace_id, &request_path).await +} + +async fn require_active_scoped_workspace( + api: &WorkspaceServerApi, + workspace_id: &str, + path: &str, +) -> std::result::Result<(), Response> { + match api + .store + .get_workspace(workspace_id) + .await + .map_err(server_error_response)? { - return Err(repository_api_rejection( - &request_path, + Some(workspace) if workspace.state == "active" => Ok(()), + Some(_) => Err(repository_api_rejection( + path, StatusCode::CONFLICT, - "Workspace is deleting and no longer accepts mutations", - )); + "Workspace is deleting and no longer accepts requests", + )), + None => Err(repository_api_rejection( + path, + StatusCode::NOT_FOUND, + "workspace was not found", + )), } - Ok(()) } async fn authorize_workspace_api_request( @@ -3011,17 +2984,15 @@ async fn enforce_server_cookie_mutation_origin( next.run(request).await } -fn workspace_request_requires_mutation_lock( - method: &Method, - path: &str, - workspace_id: &str, -) -> bool { - if matches!(*method, Method::GET | Method::HEAD | Method::OPTIONS) { - return false; - } - // Fetching a one-time Runtime resource consumes only an in-memory broker handle. It is a - // callback required to finish an already-gated mutation, not a Workspace state mutation. - path != format!("/api/runtime/v1/workspaces/{workspace_id}/resources/fetch") +fn workspace_request_is_callback(method: &Method, path: &str, workspace_id: &str) -> bool { + // Only these exact authenticated callback endpoints can reenter an admitted + // create/restore while deletion drains. They still hold a lifecycle lease. + (*method == Method::POST + && (path == format!("/api/w/{workspace_id}/subjektiv/sessions") + || path == format!("/api/runtime/v1/workspaces/{workspace_id}/resources/fetch"))) + || (*method == Method::GET + && (path == format!("/api/w/{workspace_id}/runtime-config") + || path == format!("/api/w/{workspace_id}/worker-discovery/workers"))) } async fn dispatch_workspace_request( @@ -3030,19 +3001,25 @@ async fn dispatch_workspace_request( ) -> Response { let path = request.uri().path().to_owned(); let workspace_id = scoped_workspace_id(&path); - let mutation_gate = if let Some(workspace_id) = workspace_id - && workspace_request_requires_mutation_lock(request.method(), &path, workspace_id) - { - Some((api.mutation_lock(workspace_id).await, workspace_id)) - } else { - None - }; - let _mutation_guard = if let Some((gate, workspace_id)) = mutation_gate.as_ref() { - Some( - gate.request_lock(request.method(), &path, workspace_id) - .lock() - .await, - ) + // Admission is immediate, shared, and cancellation-safe. No resource or + // business-operation lock may precede proof validation. The lease covers + // authentication and dispatch, so deletion cannot race the state check. + let _request_lease = if let Some(workspace_id) = workspace_id { + let gate = api.admission_for_workspace(workspace_id).await; + match gate.admit(workspace_request_is_callback( + request.method(), + &path, + workspace_id, + )) { + Some(lease) => Some(lease), + None => { + return repository_api_rejection( + &path, + StatusCode::CONFLICT, + "Workspace deletion is draining requests", + ); + } + } } else { None }; @@ -3596,6 +3573,7 @@ impl WorkspaceApi { worker_projection, resource_broker, workdir_sessions: Arc::new(Mutex::new(WorkdirSessionRegistry::default())), + memory_operation_lock: Arc::new(Mutex::new(())), external_workdir_providers: Arc::new(Mutex::new(ExternalProviderRegistry::default())), external_workdir_expiry_tasks: Arc::new(Mutex::new(HashMap::new())), workdir_session_locks: Arc::new(Mutex::new(HashMap::new())), @@ -7372,8 +7350,8 @@ impl server_api::ServerApi for ServerApiContractService { Vec::new(), ) })?; - let mutation_lock = api.mutation_lock(&workspace_id).await; - let _mutation_guard = mutation_lock.lock_deletion().await; + let admission_for_workspace = api.admission_for_workspace(&workspace_id).await; + let _mutation_guard = admission_for_workspace.lock_deletion().await; let existing = api .store .workspace_deletion_operation(&actor.account_id, &request.operation_id) @@ -18861,6 +18839,12 @@ async fn scoped_memory_backend_operation( Json(operation): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; + // The legacy memory document edit is a read/replace/write across store + // calls. Keep that resource atomic without serializing unrelated requests. + let _memory_guard = api + .memory_operation_lock + .lock() + .expect("Memory operation lock poisoned"); let result = execute_memory_backend_operation_with_authority(&api.authority, operation); let response = match result { Ok(result) => MemoryBackendHttpResponse::Ok { result }, @@ -32915,6 +32899,7 @@ mod tests { mod workdir_delegation_tests; mod worker_drive_tests; mod worker_operations_tests; + mod workspace_admission_tests; include!("server_workspace_config_tests.rs"); use super::*; use axum::body::{Body, to_bytes}; @@ -35177,6 +35162,7 @@ mod tests { repository_access_requests: Arc>>, reject_repository_access: Arc>, + lifecycle_callback: Option>, next_restore_outcome: Arc>>, restore_attempts: Arc>, } @@ -35430,6 +35416,9 @@ mod tests { }], }; } + if let Some(callback) = &self.lifecycle_callback { + callback(&binding.worker_id.to_string()); + } assert!(request.workdir_attachment_requests.is_empty()); assert!(request.resolved_workdir_attachment_requests.is_empty()); let mut worker = Self::worker_summary(binding, &request); @@ -35539,6 +35528,9 @@ mod tests { _request: runtime_api::WorkerRestoreRequest, ) -> InternalWorkerRestoreResult { *self.restore_attempts.lock().unwrap() += 1; + if let Some(callback) = &self.lifecycle_callback { + callback(worker_id); + } let worker = self.worker(worker_id).worker; InternalWorkerRestoreResult { state: if let Some(outcome) = self.next_restore_outcome.lock().unwrap().take() { @@ -35654,18 +35646,18 @@ mod tests { } #[test] - fn runtime_resource_fetch_bypasses_workspace_mutation_lock() { - assert!(!workspace_request_requires_mutation_lock( + fn runtime_resource_fetch_is_a_drained_workspace_callback() { + assert!(workspace_request_is_callback( &Method::POST, "/api/runtime/v1/workspaces/workspace-a/resources/fetch", "workspace-a" )); - assert!(workspace_request_requires_mutation_lock( + assert!(!workspace_request_is_callback( &Method::POST, "/api/w/workspace-a/working-directories", "workspace-a" )); - assert!(!workspace_request_requires_mutation_lock( + assert!(!workspace_request_is_callback( &Method::GET, "/api/w/workspace-a/working-directories", "workspace-a" @@ -35675,11 +35667,11 @@ mod tests { #[tokio::test] async fn workspace_mutation_gate_serializes_deletion_with_active_mutations() { let locks = Arc::new(AsyncMutex::new(HashMap::new())); - let active_mutation = workspace_mutation_lock(&locks, "workspace-a").await; - let deletion = workspace_mutation_lock(&locks, "workspace-a").await; + let active_mutation = workspace_admission_gate(&locks, "workspace-a").await; + let deletion = workspace_admission_gate(&locks, "workspace-a").await; assert!(Arc::ptr_eq(&active_mutation, &deletion)); - let active_guard = active_mutation.requests.lock().await; + let active_guard = active_mutation.admit(false).unwrap(); let (acquired_tx, mut acquired_rx) = tokio::sync::oneshot::channel(); let waiter = tokio::spawn(async move { let _deletion_guard = deletion.lock_deletion().await; diff --git a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs index 44d7d7811..b54c4bdad 100644 --- a/crates/workspace-server/src/server/tests/subject_spawn_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_spawn_tests.rs @@ -38,12 +38,8 @@ async fn subject_session_callback_completes_while_create_and_deletion_are_waitin TEST_WORKSPACE_ID.to_string(), build_inner_router(api.clone()), ); - let gate = server.mutation_lock(TEST_WORKSPACE_ID).await; - let create_path = format!("/api/w/{TEST_WORKSPACE_ID}/workers"); - let create = gate - .request_lock(&Method::POST, &create_path, TEST_WORKSPACE_ID) - .lock() - .await; + let gate = server.admission_for_workspace(TEST_WORKSPACE_ID).await; + let create = gate.admit(false).unwrap(); let mut deletion = Box::pin(gate.lock_deletion()); assert!(futures::poll!(&mut deletion).is_pending()); let app = workspace_server_router(server); @@ -95,29 +91,32 @@ async fn subject_session_callback_completes_while_create_and_deletion_are_waitin let _deletion = tokio::time::timeout(std::time::Duration::from_secs(2), deletion) .await .unwrap(); - assert!(gate.session_attributions.try_lock().is_err()); + assert!(gate.admit(true).is_none()); } #[tokio::test] async fn workspace_deletion_waits_for_session_attribution_and_fences_both_lanes() { - let gate = WorkspaceMutationGate::default(); - let attribution = gate.session_attributions.lock().await; + let gate = Arc::new(WorkspaceAdmission::default()); + let attribution = gate.admit(true).unwrap(); let mut deletion = Box::pin(gate.lock_deletion()); assert!(futures::poll!(&mut deletion).is_pending()); - assert!(gate.requests.try_lock().is_err()); + assert!(gate.admit(false).is_none()); + assert!(gate.admit(true).is_none()); drop(attribution); let deletion = deletion.await; - assert!(gate.requests.try_lock().is_err()); - assert!(gate.session_attributions.try_lock().is_err()); + assert!(gate.admit(false).is_none()); + assert!(gate.admit(true).is_none()); drop(deletion); - assert!(gate.requests.try_lock().is_ok()); - assert!(gate.session_attributions.try_lock().is_ok()); + assert!(gate.admit(false).is_some()); + assert!(gate.admit(true).is_some()); } #[tokio::test] -async fn only_exact_session_attribution_posts_can_reenter_worker_creation() { - let gate = WorkspaceMutationGate::default(); - let _create = gate.requests.lock().await; +async fn only_exact_authenticated_callbacks_can_reenter_deletion_drain() { + let gate = Arc::new(WorkspaceAdmission::default()); + let _create = gate.admit(false).unwrap(); + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); for (method, path) in [ (Method::POST, "/api/w/workspace-a/subjektiv/sessions"), (Method::POST, "/api/w/workspace-a/subjektiv/memory"), @@ -127,9 +126,8 @@ async fn only_exact_session_attribution_posts_can_reenter_worker_creation() { (Method::POST, "/api/w/workspace-a/workers"), ] { let available = gate - .request_lock(&method, path, "workspace-a") - .try_lock() - .is_ok(); + .admit(workspace_request_is_callback(&method, path, "workspace-a")) + .is_some(); assert_eq!( available, method == Method::POST && path == "/api/w/workspace-a/subjektiv/sessions", diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs new file mode 100644 index 000000000..e5c745503 --- /dev/null +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests.rs @@ -0,0 +1,1012 @@ +use super::*; +use sha2::Digest; +use std::sync::atomic::AtomicI64; +use tokio::sync::{mpsc, oneshot}; + +const START: i64 = 1_800_000_000; +const BOUND: std::time::Duration = std::time::Duration::from_secs(5); + +struct Fixture { + _root: tempfile::TempDir, + api: WorkspaceApi, + server: WorkspaceServerApi, + app: Router, + identity: RuntimeIdentityMaterial, + now: Arc, +} + +impl Fixture { + async fn new() -> Self { + let root = tempfile::tempdir().unwrap(); + let mut api = test_api(root.path()).await; + let identity = RuntimeIdentityMaterial::generate("workdirless-runtime").unwrap(); + configure_runtime_request_auth(&mut api, &identity, "workdirless-runtime"); + let runtime = WorkdirlessFixtureRuntime::default(); + for worker in ["1", "2"] { + seed_worker_source_member(&api, WorkdirlessFixtureRuntime::RUNTIME_ID, worker); + runtime.workers.lock().unwrap().push(InternalWorkerSummary { + restore_observation_token: None, + worker: RuntimeWorkerRef::new(WorkdirlessFixtureRuntime::RUNTIME_ID, worker), + host_id: "fixture-host".into(), + display_name: worker.into(), + label: worker.into(), + profile: None, + singleton_key: None, + tags: Vec::new(), + workspace: InternalWorkerWorkspaceSummary { + visibility: "workspace".into(), + identity: TEST_WORKSPACE_ID.into(), + workspace_id: Some(TEST_WORKSPACE_ID.into()), + }, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, + state: "idle".into(), + execution_reconstructed: false, + worker_state: None, + last_seen_at: None, + pinned: false, + retention_state: "normal".into(), + implementation: InternalWorkerImplementationSummary { + kind: "fixture".into(), + display_hint: "fixture".into(), + }, + workdir_attachments: Vec::new(), + diagnostics: Vec::new(), + }); + } + api.runtime.register_or_replace(runtime); + let now = Arc::new(AtomicI64::new(START)); + let clock = now.clone(); + let mut server = WorkspaceServerApi::new(api.config.clone(), api.store.clone()); + server.proof_clock = Arc::new(move || clock.load(Ordering::SeqCst)); + server + .apis + .lock() + .await + .insert(TEST_WORKSPACE_ID.into(), api.clone()); + server + .routers + .lock() + .await + .insert(TEST_WORKSPACE_ID.into(), build_inner_router(api.clone())); + let app = workspace_server_router(server.clone()); + Self { + _root: root, + api, + server, + app, + identity, + now, + } + } + + fn request( + &self, + worker: Option<&str>, + path: &str, + body: Vec, + issued: i64, + permission: &str, + ) -> Request { + let proof = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&self.identity) + .issue( + "server-test", + TEST_WORKSPACE_ID, + worker, + permission, + "POST", + path, + &body, + issued, + 60, + ) + .unwrap(); + Request::builder() + .method("POST") + .uri(path) + .header(CONTENT_TYPE, "application/json") + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof, + ) + .body(Body::from(body)) + .unwrap() + } + + fn operation(&self, worker: &str) -> Request { + let body = serde_json::to_vec(&WorkspaceWorkdirSessionOperationRequest { + target_workdir: "attachment".into(), + operation: WorkdirSessionOperation::Read(workdir::ReadRequest { + path: workdir::WorkdirPath::new("visible.txt").unwrap(), + offset: 0, + limit: 10, + max_bytes: 1024, + }), + }) + .unwrap(); + self.request( + Some(worker), + &format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations"), + body, + self.now.load(Ordering::SeqCst), + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ) + } + + fn attach(&self, worker: &str) -> mpsc::Receiver { + let id = format!("external-{worker}"); + let grant = ExternalWorkdirGrantRecord { + grant_id: id.clone(), + workspace_id: TEST_WORKSPACE_ID.into(), + workdir_id: id.clone(), + provider_instance_id: id.clone(), + display_name: id.clone(), + permissions: "read_only".into(), + created_by: "account-test".into(), + created_at: "1".into(), + expires_at: None, + generation: 1, + status: "online".into(), + updated_at: "1".into(), + }; + self.api + .store + .create_external_workdir_grant( + &grant, + &WorkdirRegistryRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + workdir_id: id.clone(), + display_name: Some(id.clone()), + source: WorkdirRegistrySource::ExternalGrant { + grant_id: id.clone(), + }, + 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: "present".into(), + cleanliness: "clean".into(), + created_at: "1".into(), + updated_at: "1".into(), + }, + ) + .unwrap(); + self.api + .store + .attach_worker_workdir(&WorkerWorkdirLinkRecord { + connection_id: String::new(), + workspace_id: TEST_WORKSPACE_ID.into(), + worker: RuntimeWorkerRef::new("workdirless-runtime", worker), + workdir_id: id.clone(), + alias: "attachment".into(), + capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY, + linked_at: "1".into(), + unlinked_at: None, + }) + .unwrap(); + let (sender, receiver) = mpsc::channel(16); + self.api + .external_workdir_providers + .lock() + .unwrap() + .connected(Arc::new(ExternalProviderConnection { + grant_id: id.clone(), + workdir_id: id.clone(), + provider_instance_id: id, + generation: 1, + expires_at: None, + capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY, + admission: Arc::new(tokio::sync::Semaphore::new(16)), + shutdown_confirmed: Arc::new(std::sync::atomic::AtomicBool::new(false)), + sender, + })); + receiver + } + + async fn start_read( + &self, + worker: &str, + receiver: &mut mpsc::Receiver, + ) -> ( + tokio::task::JoinHandle, + oneshot::Sender>, + ) { + let request = self.operation(worker); + let app = self.app.clone(); + let mut task = tokio::spawn(async move { app.oneshot(request).await.unwrap() }); + let command = loop { + let command = tokio::select! { + command = receiver.recv() => command.unwrap(), + result = &mut task => { + let result = result.unwrap(); + let status = result.status(); + let body = to_bytes(result.into_body(), usize::MAX).await.unwrap(); + panic!("read ended before provider: {status} {body:?}"); + } + _ = tokio::time::sleep(BOUND) => panic!("read did not reach provider"), + }; + if matches!(command, ExternalProviderCommand::Cancel { .. }) { + continue; + } + break command; + }; + let ExternalProviderCommand::Operation { + operation, + response, + .. + } = command + else { + panic!("expected read") + }; + assert!( + matches!(operation, WorkdirSessionOperation::Read(_)), + "{operation:?}" + ); + (task, response) + } +} + +fn read_result() -> WorkdirSessionOperationResult { + WorkdirSessionOperationResult::Read(workdir::ReadResult { + path: workdir::WorkdirPath::new("visible.txt").unwrap(), + bytes: b"visible\n".to_vec(), + start_line: 1, + total_lines: 1, + content_hash: sha2::Sha256::digest(b"visible\n").into(), + truncated: false, + }) +} + +async fn assert_json(response: Response, status: StatusCode) -> Value { + let actual = response.status(); + let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert_eq!(actual, status, "{bytes:?}"); + if bytes.is_empty() { + assert!(!status.is_success(), "successful route must return JSON"); + Value::Null // Existing non-repository dispatcher rejection is status-only. + } else { + serde_json::from_slice(&bytes).unwrap() + } +} + +#[tokio::test] +async fn long_worker_operation_does_not_block_other_worker_ticket_posts_or_workdir_read() { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let mut b = f.attach("2"); + let ticket = browser_ticket_backend(&f.api) + .unwrap() + .create(ticket::NewTicket::new("parallel ticket")) + .unwrap(); + let (long, response) = f.start_read("1", &mut a).await; + // A's proof will now be expired, but was already verified before entering + // its provider. B's freshly issued proof must not wait for A's operation. + f.now.store(START + 120, Ordering::SeqCst); + for (path, body) in [ + ( + format!("/api/w/{TEST_WORKSPACE_ID}/tickets/{}/show", ticket.id), + b"{}".to_vec(), + ), + ( + format!("/api/w/{TEST_WORKSPACE_ID}/tickets/query"), + b"{}".to_vec(), + ), + ] { + let request = f.request( + Some("2"), + &path, + body, + START + 120, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + let result = tokio::time::timeout(BOUND, f.app.clone().oneshot(request)) + .await + .expect("B ticket POST blocked by A") + .unwrap(); + let body = assert_json(result, StatusCode::OK).await; + assert!(body.is_object()); + } + let (read, reply) = f.start_read("2", &mut b).await; + reply.send(Ok(read_result())).unwrap(); + let body = assert_json( + tokio::time::timeout(BOUND, read).await.unwrap().unwrap(), + StatusCode::OK, + ) + .await; + assert!( + serde_json::to_string(&body) + .unwrap() + .contains("visible.txt") + ); + assert!(!long.is_finished(), "A was not controlled by the fixture"); + response.send(Ok(read_result())).unwrap(); + assert_json(long.await.unwrap(), StatusCode::OK).await; +} + +#[tokio::test] +async fn same_worker_session_stays_exclusive_but_proof_is_verified_before_waiting() { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let (first, reply) = f.start_read("1", &mut a).await; + let mut second = Box::pin(f.app.clone().oneshot(f.operation("1"))); + // Poll through dispatch/auth until the handler waits on its resource lock. + assert!(futures::poll!(&mut second).is_pending()); + assert!( + a.try_recv().is_err(), + "same session read escaped resource exclusion" + ); + f.now.store(START + 120, Ordering::SeqCst); + reply.send(Ok(read_result())).unwrap(); + assert_json(first.await.unwrap(), StatusCode::OK).await; + assert!(futures::poll!(&mut second).is_pending()); + let Some(ExternalProviderCommand::Operation { response, .. }) = + tokio::time::timeout(BOUND, a.recv()).await.unwrap() + else { + panic!("expected second read") + }; + response.send(Ok(read_result())).unwrap(); + assert_json( + tokio::time::timeout(BOUND, second).await.unwrap().unwrap(), + StatusCode::OK, + ) + .await; +} + +#[tokio::test] +async fn deletion_fences_new_requests_then_drains_without_callback_starvation() { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let (long, reply) = f.start_read("1", &mut a).await; + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); + let blocked = f.app.clone().oneshot(f.operation("2")).await.unwrap(); + assert_json(blocked, StatusCode::CONFLICT).await; + let callback = gate.admit(true).expect("parent may still need callback"); + reply.send(Ok(read_result())).unwrap(); + assert_json(long.await.unwrap(), StatusCode::OK).await; + // No new callbacks may prolong the drain once all parents have left. + assert!(gate.admit(true).is_none()); + assert!(futures::poll!(&mut deletion).is_pending()); + drop(callback); + let deletion = tokio::time::timeout(BOUND, deletion).await.unwrap(); + assert!(gate.admit(false).is_none()); + assert!(gate.admit(true).is_none()); + drop(deletion); + assert!(gate.admit(false).is_some()); +} + +#[tokio::test] +async fn cancelled_deletion_reopens_admission_and_cancelled_dispatch_releases_drain() { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let (long, _reply) = f.start_read("1", &mut a).await; + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); + drop(deletion); + assert!(gate.admit(false).is_some()); + long.abort(); + assert!(long.await.unwrap_err().is_cancelled()); + let deletion = tokio::time::timeout(BOUND, gate.lock_deletion()) + .await + .expect("cancel leaked admission"); + drop(deletion); + // Retry via the real dispatcher and resource lock, not only the gate. + let (retry, reply) = f.start_read("1", &mut a).await; + reply.send(Ok(read_result())).unwrap(); + assert_json(retry.await.unwrap(), StatusCode::OK).await; +} + +#[tokio::test] +async fn provider_error_and_disconnect_release_dispatch_admission() { + for disconnect in [false, true] { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let (long, reply) = f.start_read("1", &mut a).await; + if disconnect { + drop(reply); + } else { + reply + .send(Err(WorkdirTransportError::from_workdir_error( + &workdir::WorkdirError::Denied("fixture denied".into()), + ))) + .unwrap(); + } + let result = tokio::time::timeout(BOUND, long).await.unwrap().unwrap(); + assert_json( + result, + if disconnect { + StatusCode::SERVICE_UNAVAILABLE + } else { + StatusCode::FORBIDDEN + }, + ) + .await; + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + let deletion = tokio::time::timeout(BOUND, gate.lock_deletion()) + .await + .expect("error leaked admission"); + drop(deletion); + let (retry, reply) = f.start_read("1", &mut a).await; + reply.send(Ok(read_result())).unwrap(); + assert_json(retry.await.unwrap(), StatusCode::OK).await; + } +} + +#[tokio::test] +async fn runtime_proof_expiry_tamper_binding_replay_and_membership_are_not_relaxed() { + let f = Fixture::new().await; + let path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/query"); + for case in [ + "expired", + "future", + "body", + "path", + "method", + "signature", + "membership", + "replay", + ] { + let issued = match case { + "expired" => START - 61, + "future" => START + 61, + _ => START, + }; + let mut request = f.request( + Some(if case == "membership" { + "not-a-member" + } else { + "2" + }), + &path, + b"{}".to_vec(), + issued, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + match case { + "body" => *request.body_mut() = Body::from("{\"limit\":1}"), + "path" => *request.uri_mut() = format!("{path}?limit=1").parse().unwrap(), + "method" => *request.method_mut() = Method::GET, + "signature" => { + let key = worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER; + let proof = request.headers()[key].to_str().unwrap().to_owned(); + let mut bytes = proof.into_bytes(); + let index = bytes.iter().position(|b| *b == b'.').unwrap() + 1; + bytes[index] = if bytes[index] == b'A' { b'B' } else { b'A' }; + request + .headers_mut() + .insert(key, HeaderValue::from_bytes(&bytes).unwrap()); + } + "replay" => { + let proof = request.headers() + [worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER] + .clone(); + let mut first = f.request( + Some("2"), + &path, + b"{}".to_vec(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + first.headers_mut().insert( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof, + ); + assert_json(f.app.clone().oneshot(first).await.unwrap(), StatusCode::OK).await; + } + _ => {} + } + assert_json( + f.app.clone().oneshot(request).await.unwrap(), + StatusCode::UNAUTHORIZED, + ) + .await; + } + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + drop( + tokio::time::timeout(BOUND, gate.lock_deletion()) + .await + .unwrap(), + ); + assert_json( + f.app + .clone() + .oneshot(f.request( + Some("2"), + &path, + b"{}".to_vec(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + )) + .await + .unwrap(), + StatusCode::OK, + ) + .await; +} + +#[tokio::test] +async fn provider_timeout_releases_admission_without_wall_clock_wait() { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let (long, _reply) = f.start_read("1", &mut a).await; + tokio::time::pause(); + tokio::time::advance(std::time::Duration::from_secs(31)).await; + let body = assert_json(long.await.unwrap(), StatusCode::SERVICE_UNAVAILABLE).await; + assert_eq!(body["code"], "unavailable"); + tokio::time::resume(); + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + drop( + tokio::time::timeout(BOUND, gate.lock_deletion()) + .await + .unwrap(), + ); + let (retry, reply) = f.start_read("1", &mut a).await; + reply.send(Ok(read_result())).unwrap(); + assert_json(retry.await.unwrap(), StatusCode::OK).await; +} + +#[tokio::test] +async fn same_workdir_attachment_conflict_is_preserved_while_owner_operation_runs() { + let f = Fixture::new().await; + let mut a = f.attach("1"); + let (first, reply) = f.start_read("1", &mut a).await; + let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments"); + let request = f.request( + Some("2"), + &path, + serde_json::to_vec(&json!({"alias":"attachment", "working_directory_id":"external-1"})) + .unwrap(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + let body = assert_json( + tokio::time::timeout(BOUND, f.app.clone().oneshot(request)) + .await + .unwrap() + .unwrap(), + StatusCode::CONFLICT, + ) + .await; + assert!( + body["message"] + .as_str() + .unwrap() + .contains("already attached"), + "{body}" + ); + assert!(!first.is_finished()); + reply.send(Ok(read_result())).unwrap(); + assert_json(first.await.unwrap(), StatusCode::OK).await; +} + +#[tokio::test] +async fn runtime_callbacks_during_deletion_drain_still_require_bound_proofs() { + let f = Fixture::new().await; + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + let parent = gate.admit(false).unwrap(); + let mut deletion = Box::pin(gate.lock_deletion()); + assert!(futures::poll!(&mut deletion).is_pending()); + let path = format!("/api/runtime/v1/workspaces/{TEST_WORKSPACE_ID}/resources/fetch"); + let handle = f + .api + .resource_broker + .issue_repository_ssh_access_handle( + TEST_WORKSPACE_ID, + WorkdirlessFixtureRuntime::RUNTIME_ID, + "fixture-resource", + "1", + Utc::now().timestamp() + 60, + worker_runtime::resource::RepositorySshAccessSecret { + credential_candidates: Vec::new(), + known_hosts_entry: "fixture".into(), + }, + ) + .unwrap(); + let body = serde_json::to_vec(&worker_runtime::resource::BackendResourceFetchRequest { + runtime_id: WorkdirlessFixtureRuntime::RUNTIME_ID.into(), + worker_id: None, + audit_correlation_id: handle.audit_correlation_id.clone(), + handle, + }) + .unwrap(); + for (permission, expected) in [ + ( + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + StatusCode::UNAUTHORIZED, + ), + ( + worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION, + StatusCode::OK, + ), + ( + worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION, + StatusCode::NOT_FOUND, + ), + ] { + let request = f.request(None, &path, body.clone(), START, permission); + let response = tokio::time::timeout(BOUND, f.app.clone().oneshot(request)) + .await + .unwrap() + .unwrap(); + assert_json(response, expected).await; + } + for (path, permission) in [ + ( + format!("/api/w/{TEST_WORKSPACE_ID}/runtime-config?profile=builtin%3Acompanion"), + worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION, + ), + ( + format!("/api/w/{TEST_WORKSPACE_ID}/worker-discovery/workers?limit=1"), + worker_runtime::auth::WORKSPACE_WORKER_DISCOVERY_PERMISSION, + ), + ] { + let proof = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&f.identity) + .issue( + "server-test", + TEST_WORKSPACE_ID, + None, + permission, + "GET", + &path, + b"", + START, + 60, + ) + .unwrap(); + let request = Request::builder() + .uri(&path) + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof, + ) + .body(Body::empty()) + .unwrap(); + assert_json( + tokio::time::timeout(BOUND, f.app.clone().oneshot(request)) + .await + .unwrap() + .unwrap(), + StatusCode::OK, + ) + .await; + } + assert!(futures::poll!(&mut deletion).is_pending()); + drop(parent); + drop(tokio::time::timeout(BOUND, deletion).await.unwrap()); +} + +#[tokio::test] +async fn revoked_runtime_and_persisted_workspace_fence_reject_new_requests() { + for revoked in [true, false] { + let f = Fixture::new().await; + if revoked { + let mut binding = f + .api + .store + .get_workspace_runtime_binding( + TEST_WORKSPACE_ID, + WorkdirlessFixtureRuntime::RUNTIME_ID, + ) + .await + .unwrap() + .unwrap(); + binding.revoked_at = Some("revoked".into()); + binding.state = StoredRuntimeBindingState::Revoked; + f.api + .store + .upsert_workspace_runtime_binding_record(binding, true) + .await + .unwrap(); + } else { + let mut workspace = f + .api + .store + .get_workspace(TEST_WORKSPACE_ID) + .await + .unwrap() + .unwrap(); + workspace.state = "deleting".into(); + f.api.store.upsert_workspace(&workspace).await.unwrap(); + } + let path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/query"); + let request = f.request( + Some("2"), + &path, + b"{}".to_vec(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + assert_json( + f.app.clone().oneshot(request).await.unwrap(), + if revoked { + StatusCode::UNAUTHORIZED + } else { + StatusCode::CONFLICT + }, + ) + .await; + let path = format!("/api/w/{TEST_WORKSPACE_ID}/working-directories"); + let proof = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&f.identity) + .issue( + "server-test", + TEST_WORKSPACE_ID, + Some("2"), + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + "GET", + &path, + b"", + START, + 60, + ) + .unwrap(); + let request = Request::builder() + .uri(&path) + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof, + ) + .body(Body::empty()) + .unwrap(); + assert_json( + f.app.clone().oneshot(request).await.unwrap(), + if revoked { + StatusCode::UNAUTHORIZED + } else { + StatusCode::CONFLICT + }, + ) + .await; + drop( + tokio::time::timeout( + BOUND, + f.server + .admission_for_workspace(TEST_WORKSPACE_ID) + .await + .lock_deletion(), + ) + .await + .unwrap(), + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn create_and_restore_session_callbacks_finish_while_deletion_start_waits() { + let f = Fixture::new().await; + let subjektiv = open_subjektiv_store(&f.api).unwrap(); + let subject = subjektiv + .create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap()) + .unwrap(); + let (entered, mut entries) = mpsc::unbounded_channel(); + let (release, released) = std::sync::mpsc::channel(); + let released = Arc::new(Mutex::new(released)); + let runtime = WorkdirlessFixtureRuntime { + lifecycle_callback: Some(Arc::new(move |worker| { + entered.send(worker.to_owned()).unwrap(); + released + .lock() + .unwrap() + .recv_timeout(BOUND) + .expect("callback parent was not released"); + })), + ..Default::default() + }; + f.api.runtime.register_or_replace(runtime.clone()); + let token = seed_test_api_token(f.api.store.as_ref(), TEST_WORKSPACE_ID); + let owner_request = |path: &str, body: Value| { + Request::builder() + .method("POST") + .uri(path) + .header(CONTENT_TYPE, "application/json") + .header("authorization", format!("Bearer {token}")) + .body(Body::from(serde_json::to_vec(&body).unwrap())) + .unwrap() + }; + let mut worker_id = String::new(); + for restore in [false, true] { + let request = if !restore { + owner_request( + &format!("/api/w/{TEST_WORKSPACE_ID}/workers"), + json!({ + "runtime_id": WorkdirlessFixtureRuntime::RUNTIME_ID, "display_name": "callback parent", + "profile": "builtin:companion", "initial_submit": [], + "feature_connections": {"subjektiv":{"subject_id":subject.id}}, + }), + ) + } else { + let worker = RuntimeWorkerRef::new(WorkdirlessFixtureRuntime::RUNTIME_ID, &worker_id); + let observation = f + .api + .runtime + .worker(&worker) + .unwrap() + .restore_observation_token + .unwrap(); + owner_request( + &format!( + "/api/w/{TEST_WORKSPACE_ID}/runtimes/{}/workers/{worker_id}/restore", + WorkdirlessFixtureRuntime::RUNTIME_ID + ), + json!({"expected_observation_token":observation, "request_id":"restore-callback"}), + ) + }; + let app = f.app.clone(); + let parent = tokio::spawn(async move { app.oneshot(request).await.unwrap() }); + worker_id = tokio::time::timeout(BOUND, entries.recv()) + .await + .expect("create/restore did not reach Runtime") + .unwrap(); + // Use the actual deletion start route, not just the admission helper. + // Cancel it before preflight: this fixture keeps its Workspace assets. + let request = owner_request( + &format!("/api/workspaces/{TEST_WORKSPACE_ID}/deletion"), + json!({ + "operation_id": if restore {"delete-during-restore"} else {"delete-during-create"}, + "expected_revision": "fixture-revision", "confirmation":"fixture", + }), + ); + let mut deletion = Box::pin(f.app.clone().oneshot(request)); + assert!(futures::poll!(&mut deletion).is_pending()); + let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; + assert!( + gate.admit(false).is_none(), + "deletion start did not fence admission" + ); + let session_id = session_store::new_session_id().to_string(); + let callback_path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions"); + let callback = f.request( + Some(&worker_id), + &callback_path, + serde_json::to_vec(&json!({"session_id":session_id,"create_if_missing":true})).unwrap(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + let body = assert_json( + tokio::time::timeout(BOUND, f.app.clone().oneshot(callback)) + .await + .unwrap() + .unwrap(), + StatusCode::OK, + ) + .await; + assert_eq!(body["subject_id"], subject.id); + assert_eq!( + subjektiv + .session_attribution(&session_id) + .unwrap() + .unwrap() + .worker_id, + worker_id + ); + assert!(!parent.is_finished()); + assert!(futures::poll!(&mut deletion).is_pending()); + drop(deletion); + release.send(()).unwrap(); + let body = assert_json( + tokio::time::timeout(BOUND, parent).await.unwrap().unwrap(), + StatusCode::OK, + ) + .await; + if restore { + assert_eq!(body["result"]["state"], "accepted", "{body}"); + } + assert!(gate.admit(false).is_some()); + drop( + tokio::time::timeout(BOUND, gate.lock_deletion()) + .await + .unwrap(), + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn competing_memory_document_edits_are_atomic_without_workspace_exclusion() { + use crate::authority::MemoryAuthority; + let f = Fixture::new().await; + f.api.authority.update_memory_document("original").unwrap(); + let barrier = Arc::new(tokio::sync::Barrier::new(3)); + let mut edits = Vec::new(); + for worker in ["1", "2"] { + let request = f.request( + Some(worker), + &format!("/api/w/{TEST_WORKSPACE_ID}/memory/backend"), + serde_json::to_vec(&MemoryBackendOperation::UpdateDocument( + memory::backend::MemoryDocumentUpdateOperation { + old_string: "original".into(), + new_string: format!("winner-{worker}"), + replace_all: false, + }, + )) + .unwrap(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + let app = f.app.clone(); + let barrier = barrier.clone(); + edits.push(tokio::spawn(async move { + barrier.wait().await; + app.oneshot(request).await.unwrap() + })); + } + barrier.wait().await; + let mut winners = 0; + let mut rejected = 0; + for edit in edits { + let body = assert_json( + tokio::time::timeout(BOUND, edit).await.unwrap().unwrap(), + StatusCode::OK, + ) + .await; + match body["status"].as_str().unwrap() { + "ok" => winners += 1, + "error" => { + rejected += 1; + assert!( + body["message"].as_str().unwrap().contains("not found"), + "{body}" + ); + } + _ => panic!("unexpected memory response {body}"), + } + } + assert_eq!((winners, rejected), (1, 1)); + assert!( + f.api + .authority + .memory_document() + .unwrap() + .body_md + .starts_with("winner-") + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn proof_replay_consumption_is_atomic_across_concurrent_admissions() { + let f = Fixture::new().await; + let path = format!("/api/w/{TEST_WORKSPACE_ID}/tickets/query"); + let original = f.request( + Some("2"), + &path, + b"{}".to_vec(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + let proof = + original.headers()[worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER].clone(); + let barrier = Arc::new(tokio::sync::Barrier::new(3)); + let mut requests = Vec::new(); + for _ in 0..2 { + let mut request = f.request( + Some("2"), + &path, + b"{}".to_vec(), + START, + worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION, + ); + request.headers_mut().insert( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + proof.clone(), + ); + let app = f.app.clone(); + let barrier = barrier.clone(); + requests.push(tokio::spawn(async move { + barrier.wait().await; + app.oneshot(request).await.unwrap() + })); + } + barrier.wait().await; + let mut statuses = Vec::new(); + for request in requests { + let response = tokio::time::timeout(BOUND, request).await.unwrap().unwrap(); + let status = response.status(); + assert!(matches!(status, StatusCode::OK | StatusCode::UNAUTHORIZED)); + assert_json(response, status).await; + statuses.push(status.as_u16()); + } + statuses.sort(); + assert_eq!(statuses, [200, 401]); +} diff --git a/crates/workspace-server/src/server/workspace_admission.rs b/crates/workspace-server/src/server/workspace_admission.rs new file mode 100644 index 000000000..06c94b9a8 --- /dev/null +++ b/crates/workspace-server/src/server/workspace_admission.rs @@ -0,0 +1,104 @@ +//! Workspace lifecycle admission, not operation serialization. +//! +//! Resource owners retain their own locks/transactions. Deletion first stops +//! ordinary admission, then drains admitted requests. Synchronous Runtime +//! callbacks remain admissible while a parent request is still running; once +//! the last ordinary request leaves, callback admission also closes. This +//! avoids both callback/deletion cycles and an unbounded stream of new work +//! starving deletion. All leases, including the deletion fence, are RAII. +use std::sync::{Arc, Mutex}; + +use tokio::sync::{Mutex as AsyncMutex, Notify, OwnedMutexGuard}; + +#[derive(Default)] +struct AdmissionState { + deleting: bool, + requests: usize, + callbacks: usize, +} + +#[derive(Default)] +pub(super) struct WorkspaceAdmission { + state: Mutex, + changed: Notify, + deletion: Arc>, +} + +impl WorkspaceAdmission { + pub(super) fn admit(self: &Arc, callback: bool) -> Option { + let mut state = self.state.lock().expect("Workspace admission poisoned"); + if state.deleting && (!callback || state.requests == 0) { + return None; + } + if callback { + state.callbacks += 1; + } else { + state.requests += 1; + } + Some(RequestLease { + gate: self.clone(), + callback, + }) + } + + pub(super) async fn lock_deletion(self: &Arc) -> DeletionLease { + let serial = self.deletion.clone().lock_owned().await; + self.state + .lock() + .expect("Workspace admission poisoned") + .deleting = true; + // Construct the lease before awaiting: cancellation must reopen admission. + let lease = DeletionLease { + gate: self.clone(), + _serial: serial, + }; + loop { + let changed = self.changed.notified(); + tokio::pin!(changed); + changed.as_mut().enable(); + { + let state = self.state.lock().expect("Workspace admission poisoned"); + if state.requests == 0 && state.callbacks == 0 { + return lease; + } + } + changed.await; + } + } +} + +pub(super) struct RequestLease { + gate: Arc, + callback: bool, +} + +impl Drop for RequestLease { + fn drop(&mut self) { + let mut state = self + .gate + .state + .lock() + .expect("Workspace admission poisoned"); + if self.callback { + state.callbacks -= 1; + } else { + state.requests -= 1; + } + self.gate.changed.notify_waiters(); + } +} + +pub(super) struct DeletionLease { + gate: Arc, + _serial: OwnedMutexGuard<()>, +} + +impl Drop for DeletionLease { + fn drop(&mut self) { + self.gate + .state + .lock() + .expect("Workspace admission poisoned") + .deleting = false; + } +} diff --git a/crates/workspace-server/src/worker_source.rs b/crates/workspace-server/src/worker_source.rs index 7ce8040f8..3d51476d1 100644 --- a/crates/workspace-server/src/worker_source.rs +++ b/crates/workspace-server/src/worker_source.rs @@ -12,6 +12,7 @@ use worker_runtime::worker_source::InProcessWorkerMutationProof; use crate::server::{ServerConfig, WorkspaceApi}; mod runtime_request; +pub(crate) use runtime_request::verify_runtime_request_source_proof_with_clock; pub use runtime_request::{ RuntimeRequestProofError, VerifiedRuntimeRequestSource, diagnostic_id_hash, verify_runtime_request_source_proof, verify_runtime_request_source_proof_with_store, diff --git a/crates/workspace-server/src/worker_source/runtime_request.rs b/crates/workspace-server/src/worker_source/runtime_request.rs index 00c00bd49..36e4d6076 100644 --- a/crates/workspace-server/src/worker_source/runtime_request.rs +++ b/crates/workspace-server/src/worker_source/runtime_request.rs @@ -147,6 +147,32 @@ pub async fn verify_runtime_request_source_proof_with_store( method: &str, path: &str, body_digest: &str, +) -> Result { + verify_runtime_request_source_proof_with_clock( + store, + config, + proof, + workspace_id, + permission, + method, + path, + body_digest, + &|| i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX), + ) + .await +} + +#[allow(clippy::too_many_arguments)] +pub(crate) async fn verify_runtime_request_source_proof_with_clock( + store: &dyn ControlPlaneStore, + config: &ServerConfig, + proof: &str, + workspace_id: &str, + permission: &str, + method: &str, + path: &str, + body_digest: &str, + clock: &(dyn Fn() -> i64 + Send + Sync), ) -> Result { let unverified = decode_runtime_request_source_claims(proof) .map_err(|error| RuntimeRequestProofError::undecoded().auth_error(error))?; @@ -176,7 +202,7 @@ pub async fn verify_runtime_request_source_proof_with_store( method, path, body_digest, - now_unix: i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX), + now_unix: clock(), }; let claims = verify_runtime_request_source(proof, &trusted.public_key, &expected).map_err(|error| { diff --git a/docs/development/workspace-request-admission.md b/docs/development/workspace-request-admission.md new file mode 100644 index 000000000..cbbcb7cfa --- /dev/null +++ b/docs/development/workspace-request-admission.md @@ -0,0 +1,34 @@ +# Workspace request admission and deletion + +Workspace HTTP dispatch uses `WorkspaceAdmission` for **lifecycle admission**, not a Workspace-wide operation mutex. Each admitted handler owns a drop-safe lease. Independent requests (including read-oriented POSTs) may execute concurrently; proof validation occurs before any business/resource wait in the inner router. + +## Ordering and authority + +1. Acquire an immediate lifecycle lease. While deletion drains, new ordinary requests fail closed instead of queueing behind the deletion operation. +2. Validate the bounded request body, Runtime signature, audience/Workspace/permission/method/target/body binding, expiry, transactional JTI consumption, current Runtime trust, Worker membership/reservation and singleton ownership. Browser actor/origin checks are unchanged. No TTL extension or 401 retry policy is introduced. +3. Check the persisted Workspace state. Scoped requests, including GETs, reject a deleting Workspace, including after Server reconstruction. An absent Workspace remains 404. Server-owned deletion status/polling is outside scoped dispatch. +4. Execute the inner router. Resource owners retain their existing exclusion: per-Worker session/control/removal locks; transactional Workdir attachment exclusivity and removal fences; SQLite domain transactions and revision/CAS checks; the resource broker's one-shot handle consumption. Legacy memory document read/replace/write uses a short, resource-specific mutex because it spans multiple store calls. +5. Drop the lifecycle lease on response, error, or cancellation. An admitted long operation is not rechecked for proof expiry merely because its business execution lasts past the proof TTL. This does not grant subsequent requests or disable provider/attachment authority checks. + +The proof verifier's clock is injected into the Server instance for deterministic route tests, not obtained from test environment variables. Production uses the same wall-clock seconds and existing proof rules. Clock-skew policy remains a separate scope (T-728). + +## Deletion and callbacks + +Deletion start serializes only with other deletion starts. Before awaiting active requests it closes ordinary admission, so ongoing arrivals cannot starve the drain. The fence itself is RAII: a cancelled wait or a rejected preflight reopens admission. Successful reservation persists the deleting state before cleanup is scheduled; the persisted state remains the fence after the temporary drain lease is released. + +While at least one ordinary handler remains admitted, these exact callback paths may obtain their own tracked leases: + +- POST `/api/w/{workspace_id}/subjektiv/sessions` (create/restore session attribution) +- POST `/api/runtime/v1/workspaces/{workspace_id}/resources/fetch` (bound Runtime resource handles) +- GET `/api/w/{workspace_id}/runtime-config` +- GET `/api/w/{workspace_id}/worker-discovery/workers` + +They do not bypass authentication or authorization. Callback admission closes atomically when the last ordinary lease leaves; deletion then waits for already-admitted callbacks too. A callback stream cannot keep adding work once all parents have finished. There is no deletion write-lock queue in front of callbacks required by a parent. + +## Evidence boundary + +`server/tests/workspace_admission_tests.rs` exercises the real outer dispatcher and generated inner routes with isolated SQLite stores, authenticated Runtime proofs and controlled external-provider fixtures. The long-operation fixture stops A at its provider response and proves B's ShowTicket/QueryTicket POSTs and separate Workdir read finish before releasing A. It also covers same-session exclusion, conflicting Workdir attachments, memory edit and JTI atomicity, drain/fence/cancellation, real create/restore with session callbacks overlapping deletion start, resource/config/discovery callbacks, injected-clock proof rejections and provider failures/disconnect/timeout recovery. + +Timeout uses Tokio's test clock; proof tests advance an independent injected clock by 120 seconds. Bounds detect hangs, not guessed sleeps. Dispatch task cancellation represents cancellation of the HTTP handler future; provider disconnect is a dropped transport response. These are in-process boundary tests, not a real-socket disconnect or live browser/Runtime E2E test. Background command and websocket lifetimes retain their existing domain-specific cleanup; request admission covers the handler until its response, not the entire upgraded connection. + +Past production `401 expired` incidents have **not** been correlated with live revision/request traces. The structural failure path and deterministic regression evidence do not establish that all prior 401s had this cause. T-731 and its retained session/MR/source/Workdir are unrelated and untouched. WIP remains 0.2.0; no generated DTO/schema/route artifacts change. diff --git a/docs/report/t-732-subworker-checkout-attachment.md b/docs/report/t-732-subworker-checkout-attachment.md new file mode 100644 index 000000000..081b896c1 --- /dev/null +++ b/docs/report/t-732-subworker-checkout-attachment.md @@ -0,0 +1,9 @@ +# T-732: read-only SubWorker did not receive its delegated checkout + +During T-732, the parent spawned `concurrency-audit` with read-only scope `.` for a source-boundary audit. Spawn succeeded, but the child's committed session reported no Workdir attachment. Its Glob/Read/Grep calls could not inspect the checkout; WorkerList also exposed no alternative source-observation subject. The parent inspected the child's committed final entry and stopped it without treating the audit as completed. + +The parent could read and modify the assigned checkout through its own tools. This is an attachment/delegation surface problem, not a Rust-source finding. The parent continued the investigation itself; no child audit or review verdict is claimed. + +Expected behavior: a delegated read scope should be bound to a concrete parent Workdir alias, or spawn should reject it with an actionable error rather than start an inaccessible child. The UI/tool response should expose the attachment binding clearly. A trusted Reviewer also needs a usable source observation surface; a successful Spawn alone must not be mistaken for independent source review. + +No grant was broadened, no hidden paths were explored by the parent to repair the child's authority, and no live process/environment update or T-731 resume was attempted.