From 5873f250736fb06c35176db71097e6168064cbc7 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 12 Oct 2026 14:46:05 +0900 Subject: [PATCH] fix(server): own residual Ticket MR and worker request stages --- crates/workspace-server/src/server.rs | 468 ++++++---- .../src/server_db_execution_tests.rs | 72 +- .../src/server_residual_execution_tests.rs | 843 ++++++++++++++++++ 3 files changed, 1196 insertions(+), 187 deletions(-) create mode 100644 crates/workspace-server/src/server_residual_execution_tests.rs diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 3283f13da..4d75d5985 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -13555,8 +13555,8 @@ async fn scoped_get_ticket( State(api): State, AxumPath(path): AxumPath, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; browser_ticket_detail(&api, &path.id) @@ -13569,8 +13569,8 @@ async fn scoped_query_tickets( AxumPath(path): AxumPath, Json(query): Json, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; Ok(Json(api.authority.query_tickets(query)?)) @@ -13583,8 +13583,8 @@ async fn scoped_show_ticket( AxumPath(path): AxumPath, Json(query): Json, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let backend = browser_ticket_backend(&api)?; @@ -13668,8 +13668,8 @@ async fn scoped_list_ticket_assignments( State(api): State, AxumPath(path): AxumPath, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let ticket = api.authority.ticket(&path.id)?; @@ -14337,8 +14337,8 @@ async fn apply_set_ticket_assignment( let prep_api = api.clone(); let prep_workspace = workspace_id.clone(); let (ticket, role, operation_id, expected_assignment_id, record, workdir_bindings) = api - .store - .db_executor() + .host_executors + .host()? .run(move || { let api = &prep_api; let workspace_id = prep_workspace; @@ -14488,8 +14488,8 @@ async fn apply_clear_ticket_assignment( author: String, source: Option, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &workspace_id)?; let ticket = api.authority.ticket(&id)?; @@ -14560,60 +14560,65 @@ async fn scoped_cancel_ticket_implementation( Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let ticket = api.authority.ticket(&path.id)?; - let operation_id = require_ticket_assignment_value("operation_id", request.operation_id)?; - let assignment_id = require_ticket_assignment_value("assignment_id", request.assignment_id)?; - let reason = require_ticket_assignment_value("reason", request.reason)?; - if reason.len() > 512 { - return Err(Error::InvalidInput( - "implementation cancellation reason must be at most 512 bytes".to_string(), - ) - .into()); - } + // Runtime cancel and session cleanup are mixed work. One accepted Host + // owner retains authority through the whole cancellation transaction/detail. + run_host_workflow(api.host_executors.host()?, async move { + let ticket = api.authority.ticket(&path.id)?; + let operation_id = require_ticket_assignment_value("operation_id", request.operation_id)?; + let assignment_id = require_ticket_assignment_value("assignment_id", request.assignment_id)?; + let reason = require_ticket_assignment_value("reason", request.reason)?; + if reason.len() > 512 { + return Err(Error::InvalidInput( + "implementation cancellation reason must be at most 512 bytes".to_string(), + ) + .into()); + } - let current = api.store.get_active_ticket_role_assignment( - &path.workspace_id, - &ticket.id, - TicketAssignmentRole::Worker, - )?; - if let Some(assignment) = current.filter(|value| value.assignment_id == assignment_id) - && let TicketAssignmentPrincipal::Worker { - runtime_id, - worker_id, - } = assignment.principal - { - if api - .store - .get_ticket_assignment_operation(&path.workspace_id, &operation_id)? - .is_some() + let current = api.store.get_active_ticket_role_assignment( + &path.workspace_id, + &ticket.id, + TicketAssignmentRole::Worker, + )?; + if let Some(assignment) = current.filter(|value| value.assignment_id == assignment_id) + && let TicketAssignmentPrincipal::Worker { + runtime_id, + worker_id, + } = assignment.principal { + if api + .store + .get_ticket_assignment_operation(&path.workspace_id, &operation_id)? + .is_some() + { + return Err(Error::TicketAssignmentConflict(format!( + "operation `{operation_id}` was already used for another Ticket assignment mutation" + )) + .into()); + } + let worker = RuntimeWorkerRef::new(runtime_id, worker_id); + cancel_ticket_worker(&api, &worker, &reason).await?; + } + + let cancelled = api.store.cancel_current_ticket_worker_assignment( + &path.workspace_id, + &ticket.id, + &assignment_id, + &new_id("tasev"), + &new_id("tev"), + &operation_id, + "workspace-web", + &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), + &reason, + )?; + if !cancelled { return Err(Error::TicketAssignmentConflict(format!( - "operation `{operation_id}` was already used for another Ticket assignment mutation" + "assignment `{assignment_id}` is not the current Coder implementation" )) .into()); } - let worker = RuntimeWorkerRef::new(runtime_id, worker_id); - cancel_ticket_worker(&api, &worker, &reason).await?; - } - - let cancelled = api.store.cancel_current_ticket_worker_assignment( - &path.workspace_id, - &ticket.id, - &assignment_id, - &new_id("tasev"), - &new_id("tev"), - &operation_id, - "workspace-web", - &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), - &reason, - )?; - if !cancelled { - return Err(Error::TicketAssignmentConflict(format!( - "assignment `{assignment_id}` is not the current Coder implementation" - )) - .into()); - } - browser_ticket_detail(&api, &ticket.id) + browser_ticket_detail(&api, &ticket.id) + }) + .await } async fn cancel_ticket_worker( @@ -15070,8 +15075,8 @@ async fn scoped_edit_ticket_item( AxumPath(path): AxumPath, Json(request): Json, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let mut targets = request.target.map(|target| match target { @@ -15125,8 +15130,8 @@ async fn scoped_append_ticket_event( AxumPath(path): AxumPath, Json(request): Json, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || { validate_workspace_scope(&api, &path.workspace_id)?; let role = match request.role { @@ -15154,14 +15159,32 @@ async fn scoped_mark_ticket_ready_from_browser( Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let _ = scoped_mark_ticket_ready( - State(api.clone()), - AxumPath((path.workspace_id.clone(), path.id.clone())), - headers, - Json(request), - ) - .await?; - browser_ticket_detail(&api, &path.id) + api.host_executors + .host()? + .run(move || { + let result = execute_ticket_rest_operation_on_host( + &api, + &path.workspace_id, + headers, + None, + TicketBackendOperation::MarkReady { + id: TicketIdOrSlug::Query(path.id.clone()), + request: ticket::TicketMarkReady { + operation_key: request.operation_key, + reason: request.reason, + author: None, + intake_summary: request.intake_summary, + }, + }, + )?; + // Preserve result validation before projecting the committed detail. + let _ = ticket_rest_result(result, |result| match result { + TicketBackendOperationResult::Ticket(ticket) => Some(ticket), + _ => None, + })?; + browser_ticket_detail(&api, &path.id) + }) + .await } async fn scoped_queue_ticket( @@ -15357,7 +15380,7 @@ async fn execute_ticket_rest_operation_with_context( workspace_id: &str, headers: HeaderMap, context: Option<&server_api::ServerRequestContext>, - mut operation: TicketBackendOperation, + operation: TicketBackendOperation, ) -> ApiResult { validate_workspace_scope(api, workspace_id)?; if let TicketBackendOperation::UpdateCase { ticket, request } = operation { @@ -15379,10 +15402,25 @@ async fn execute_ticket_rest_operation_with_context( let job_api = api.clone(); let workspace_id = workspace_id.to_owned(); let context = context.cloned(); - api.host_executors.host()?.run(move || { - let api = &job_api; - let workspace_id = workspace_id.as_str(); - let context = context.as_ref(); + api.host_executors + .host()? + .run(move || { + let api = &job_api; + let workspace_id = workspace_id.as_str(); + let context = context.as_ref(); + execute_ticket_rest_operation_on_host(api, workspace_id, headers, context, operation) + }) + .await +} + +// Already-owned callers must not re-admit to the occupied Host lane. +fn execute_ticket_rest_operation_on_host( + api: &WorkspaceApi, + workspace_id: &str, + headers: HeaderMap, + context: Option<&server_api::ServerRequestContext>, + mut operation: TicketBackendOperation, +) -> ApiResult { validate_ticket_operation_repository_keys(api, workspace_id, &operation)?; let mut backend = SqliteTicketBackend::open_verified( api.config.database_path.clone(), @@ -15580,7 +15618,6 @@ async fn execute_ticket_rest_operation_with_context( } } Ok(result) - }).await } #[cfg(test)] @@ -17044,37 +17081,46 @@ async fn scoped_repair_merge_request_selector( if !input.explicit_confirmation { return Err(Error::BrowserReopenConfirmationRequired.into()); } - let store = merge_request_store(&api, &workspace_id)?; - let mr = store.get_by_id(&workspace_id, &merge_request_id)?; - let ticket_id = merge_request_ticket_id(&mr)?; - let assignment = api - .store - .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? - .ok_or_else(|| { - Error::TicketAssignmentConflict("Ticket has no active Coder work assignment".into()) - })?; - let resolved_subject_ref = observe_published_source_ref( - &api, - &workspace_id, - &assignment.worker.runtime_id, - &mr.repository_id, - &input.selector_from, - )? - .resolved_ref; - let repaired = store.repair_selector_from(merge_request::RepairSelectorFrom { - workspace_id: workspace_id.clone(), - merge_request_id, - ticket_id, - selector_from: input.selector_from, - resolved_subject_ref, - repaired_by: merge_request::WorkerIdentity { - runtime_id: "browser".into(), - worker_id: "authenticated-user".into(), - }, - reason: input.reason, - now: Utc::now(), - })?; - Ok(Json(public_merge_request(&api, &workspace_id, repaired)?)) + // Provider observation and intact repair/audit SQL share an accepted owner; + // authentication's earlier DB await does not offload this mixed stage. + api.host_executors + .host()? + .run(move || { + let store = merge_request_store(&api, &workspace_id)?; + let mr = store.get_by_id(&workspace_id, &merge_request_id)?; + let ticket_id = merge_request_ticket_id(&mr)?; + let assignment = api + .store + .get_active_ticket_worker_assignment(&workspace_id, &ticket_id)? + .ok_or_else(|| { + Error::TicketAssignmentConflict( + "Ticket has no active Coder work assignment".into(), + ) + })?; + let resolved_subject_ref = observe_published_source_ref( + &api, + &workspace_id, + &assignment.worker.runtime_id, + &mr.repository_id, + &input.selector_from, + )? + .resolved_ref; + let repaired = store.repair_selector_from(merge_request::RepairSelectorFrom { + workspace_id: workspace_id.clone(), + merge_request_id, + ticket_id, + selector_from: input.selector_from, + resolved_subject_ref, + repaired_by: merge_request::WorkerIdentity { + runtime_id: "browser".into(), + worker_id: "authenticated-user".into(), + }, + reason: input.reason, + now: Utc::now(), + })?; + Ok(Json(public_merge_request(&api, &workspace_id, repaired)?)) + }) + .await } async fn scoped_register_reviewer_child_session( @@ -17473,8 +17519,8 @@ async fn scoped_ticket_assignment_tracking_update_with_context( Json(update): Json, ) -> ApiResult> { let context = context.clone(); - api.store - .db_executor() + api.host_executors + .host()? .run(move || { let context = &context; @@ -17623,7 +17669,7 @@ async fn apply_ticket_case_update( Json(mut request): Json, ) -> ApiResult> { let actor = actor.cloned(); - api.store.db_executor().run(move || { + api.host_executors.host()?.run(move || { let actor = actor.as_ref(); validate_workspace_scope(&api, &workspace_id)?; @@ -18224,6 +18270,20 @@ fn authenticate_worker_mutation_source( Ok(worker) } +async fn authenticate_worker_mutation_source_owned( + api: &WorkspaceApi, + workspace_id: &str, + headers: &HeaderMap, +) -> Result { + let api = api.clone(); + let workspace_id = workspace_id.to_owned(); + let headers = headers.clone(); + api.host_executors + .host()? + .run(move || authenticate_worker_mutation_source(&api, &workspace_id, &headers)) + .await +} + fn current_worker_identity( api: &WorkspaceApi, workspace_id: &str, @@ -24031,7 +24091,8 @@ async fn send_known_worker_input( Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; + let source = + authenticate_worker_mutation_source_owned(&api, &path.workspace_id, &headers).await?; let context = WorkerOperationContext::WorkerControl { controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), }; @@ -24069,7 +24130,8 @@ async fn cancel_known_worker( Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; + let source = + authenticate_worker_mutation_source_owned(&api, &path.workspace_id, &headers).await?; let context = WorkerOperationContext::WorkerControl { controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), }; @@ -24090,7 +24152,8 @@ async fn stop_known_worker( Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; + let source = + authenticate_worker_mutation_source_owned(&api, &path.workspace_id, &headers).await?; let context = WorkerOperationContext::WorkerControl { controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), }; @@ -24111,7 +24174,8 @@ async fn restore_known_worker( request: server_api::WorkerRestoreRequest, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; + let source = + authenticate_worker_mutation_source_owned(&api, &path.workspace_id, &headers).await?; let context = WorkerOperationContext::WorkerControl { controller: RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id), }; @@ -24179,7 +24243,8 @@ async fn scoped_capture_worker_observation_session( Json(subject): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let source = authenticate_worker_mutation_source(&api, &path.workspace_id, &headers)?; + let source = + authenticate_worker_mutation_source_owned(&api, &path.workspace_id, &headers).await?; let server_api::WorkerObservationSubjectRef::RuntimeWorker { runtime_id, worker_id, @@ -24191,20 +24256,47 @@ async fn scoped_capture_worker_observation_session( }; let target = RuntimeWorkerRef::new(runtime_id, worker_id); let controller = RuntimeWorkerRef::new(&source.runtime_id, &source.worker_id); - let grant = authorize_known_worker_permission( - &api, - &path.workspace_id, - &controller, - &target, - "observe", - )?; + let grant_api = api.clone(); + let grant_controller = controller.clone(); + let grant_target = target.clone(); + let workspace_id = path.workspace_id.clone(); + let grant = api + .store + .db_executor() + .run(move || { + authorize_known_worker_permission( + &grant_api, + &workspace_id, + &grant_controller, + &grant_target, + "observe", + ) + }) + .await?; let lock = worker_control_lock(&api, &grant.grant_id); - let _guard = lock.lock().await; - authorize_known_worker_permission(&api, &path.workspace_id, &controller, &target, "observe")?; - let target_summary = api - .runtime - .worker(&target) - .map_err(|error| error.into_error())?; + let guard = lock.lock_owned().await; + let lookup_api = api.clone(); + let lookup_target = target.clone(); + let (_guard, target_summary) = api + .host_executors + .host()? + .run(move || { + // Retain the control guard during the actual re-fence and live lookup, + // even if the observation caller cancels its wait. + authorize_known_worker_permission( + &lookup_api, + &path.workspace_id, + &controller, + &lookup_target, + "observe", + )?; + let target_summary = lookup_api + .runtime + .worker(&lookup_target) + .map_err(|error| error.into_error())?; + Ok::<_, Error>((guard, target_summary)) + }) + .await?; if matches!( target_summary.state.as_str(), "stopped" | "failed" | "rejected" | "disconnected" @@ -28973,8 +29065,8 @@ async fn get_ticket( State(api): State, AxumPath(id): AxumPath, ) -> ApiResult> { - api.store - .db_executor() + api.host_executors + .host()? .run(move || Ok(Json(api.authority.ticket(&id)?))) .await } @@ -30705,45 +30797,81 @@ async fn restore_runtime_worker_with_context( request: server_api::WorkerRestoreRequest, ) -> ApiResult> { let worker = WorkspaceWorker::resolve_async(&api, &runtime_id, &worker_id).await?; - let mut result = worker.restore(&context, request).await?; - let projected_worker = if let Some(worker) = result.worker.as_ref() { - // A replayed terminal receipt is historical evidence, not a new Runtime observation. - let record = api - .store - .get_worker_registry(api.workspace_id(), &worker.worker)? - .ok_or_else(|| Error::UnknownWorker { - worker: worker.worker.clone(), - })?; - let links = api - .store - .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?; - let workdirs = api - .store - .list_workdir_registry(&api.config.workspace_id, 500)?; - match project_worker_registry_projection(&api, Some(worker), &record, links, &workdirs) { - Ok(worker) => Some(worker), - Err(error) => { - result.diagnostics.push(RuntimeDiagnostic::new( - "worker_restore_current_projection_unavailable", - "warning", - sanitize_backend_error(&error.error.to_string()), - )); + let result = worker.restore(&context, request).await?; + project_worker_restore_response_owned(api, runtime_id, worker_id, result).await +} + +// Post-restore stored projection is its own whole DB stage. Historical terminal +// receipts are never turned into a fresh Runtime observation or registry write. +async fn project_worker_restore_response_owned( + api: WorkspaceApi, + runtime_id: String, + worker_id: String, + mut result: crate::hosts::InternalWorkerRestoreResult, +) -> ApiResult> { + // A terminal result without a Worker has no stored projection to read. + // Do not turn that pure response into a new availability failure on DB overload. + if result.worker.is_none() { + return Ok(Json(server_api::WorkerRestoreResponse { + workspace_id: api.workspace_id().to_string(), + runtime_id, + worker_id, + result: server_api::WorkerRestoreResult { + state: result.state, + worker: None, + diagnostics: result.diagnostics.into_iter().map(Into::into).collect(), + }, + })); + } + api.store + .db_executor() + .run(move || { + let projected_worker = if let Some(worker) = result.worker.as_ref() { + // A replayed terminal receipt is historical evidence, not a new Runtime observation. + let record = api + .store + .get_worker_registry(api.workspace_id(), &worker.worker)? + .ok_or_else(|| Error::UnknownWorker { + worker: worker.worker.clone(), + })?; + let links = api + .store + .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?; + let workdirs = api + .store + .list_workdir_registry(&api.config.workspace_id, 500)?; + match project_worker_registry_projection( + &api, + Some(worker), + &record, + links, + &workdirs, + ) { + Ok(worker) => Some(worker), + Err(error) => { + result.diagnostics.push(RuntimeDiagnostic::new( + "worker_restore_current_projection_unavailable", + "warning", + sanitize_backend_error(&error.error.to_string()), + )); + None + } + } + } else { None - } - } - } else { - None - }; - Ok(Json(server_api::WorkerRestoreResponse { - workspace_id: api.workspace_id().to_string(), - runtime_id: runtime_id.clone(), - worker_id: worker_id.clone(), - result: server_api::WorkerRestoreResult { - state: result.state, - worker: projected_worker, - diagnostics: result.diagnostics.into_iter().map(Into::into).collect(), - }, - })) + }; + Ok(Json(server_api::WorkerRestoreResponse { + workspace_id: api.workspace_id().to_string(), + runtime_id: runtime_id.clone(), + worker_id: worker_id.clone(), + result: server_api::WorkerRestoreResult { + state: result.state, + worker: projected_worker, + diagnostics: result.diagnostics.into_iter().map(Into::into).collect(), + }, + })) + }) + .await } fn reject_workdir_for_embedded_runtime(runtime_id: &str, has_workdir: bool) -> ApiResult<()> { diff --git a/crates/workspace-server/src/server_db_execution_tests.rs b/crates/workspace-server/src/server_db_execution_tests.rs index 6cfa24529..d52ffacd6 100644 --- a/crates/workspace-server/src/server_db_execution_tests.rs +++ b/crates/workspace-server/src/server_db_execution_tests.rs @@ -1,4 +1,6 @@ //! Real SQLite waits at the HTTP/authority boundary (not mock executor jobs). +#[path = "server_residual_execution_tests.rs"] +mod residual; use super::*; use crate::SqliteWorkspaceStore; use crate::db_executor::{DbExecutor, DbExecutorLimits, DbOperationLifetime}; @@ -36,6 +38,10 @@ struct Fixture { impl Fixture { async fn new() -> Self { + Self::with_repositories(Vec::new()).await + } + + async fn with_repositories(repositories: Vec) -> Self { let dir = tempfile::tempdir().unwrap(); let executor = DbExecutor::new(DbExecutorLimits { workers: NonZeroUsize::new(1).unwrap(), @@ -59,7 +65,7 @@ impl Fixture { cookie_name: "db_execution_session".into(), }, max_records: 100, - repositories: Vec::new(), + repositories, runtime_event_sources: Vec::new(), remote_runtime_sources: Vec::new(), backend_base_url: Some("http://localhost:8787".into()), @@ -72,6 +78,24 @@ impl Fixture { seed_test_registered_workspace(store.as_ref(), &config) .await .unwrap(); + for repository in &config.repositories { + store + .upsert_repository(&crate::store::RepositoryRecord { + workspace_id: WORKSPACE.into(), + repository_id: repository.id.clone(), + repository_key: repository.repository_key.clone(), + kind: "git".into(), + provider: Some(repository.provider.clone()), + source: repository.source.clone(), + default_ref: repository.default_selector.clone(), + source_fingerprint: repository.source_fingerprint.clone(), + observed_status: repository.observed_status.clone(), + observed_at: repository.observed_at.clone(), + created_at: config.workspace_created_at.clone(), + updated_at: config.workspace_created_at.clone(), + }) + .unwrap(); + } store .upsert_user(&UserRecord { user_id: "fixture-user".into(), @@ -924,6 +948,12 @@ struct HeldRuntimeReply { } impl HeldRuntimeReply { fn new() -> Self { + Self::with_held_path(None) + } + fn for_path(path: &str) -> Self { + Self::with_held_path(Some(path.to_owned())) + } + fn with_held_path(path: Option) -> Self { use std::io::{Read, Write}; let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let address = listener.local_addr().unwrap(); @@ -932,25 +962,33 @@ impl HeldRuntimeReply { let upstream_released = Arc::new(AtomicBool::new(false)); let flag = upstream_released.clone(); let thread = std::thread::spawn(move || { - let Ok((mut stream, _)) = listener.accept() else { - return; - }; - stream.set_read_timeout(Some(WATCHDOG)).unwrap(); - let mut buf = [0; 4096]; - let Ok(n) = stream.read(&mut buf) else { return }; - if n == 0 { - return; - } - let _ = entered.send( - String::from_utf8_lossy(&buf[..n]) + loop { + let Ok((mut stream, _)) = listener.accept() else { + return; + }; + stream.set_read_timeout(Some(WATCHDOG)).unwrap(); + let mut buf = [0; 4096]; + let Ok(n) = stream.read(&mut buf) else { return }; + if n == 0 { + return; + } + let request = String::from_utf8_lossy(&buf[..n]) .lines() .next() .unwrap() - .to_owned(), - ); - let _ = permit.recv_timeout(WATCHDOG); - flag.store(true, Ordering::SeqCst); - let _ = stream.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + .to_owned(); + if path.as_ref().is_some_and(|path| !request.contains(path)) { + // Ready's best-effort notification lookup precedes its detail + // source observation. Deny it promptly; hold only the named stage. + let _ = stream.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + continue; + } + let _ = entered.send(request); + let _ = permit.recv_timeout(WATCHDOG); + flag.store(true, Ordering::SeqCst); + let _ = stream.write_all(b"HTTP/1.1 503 Service Unavailable\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"); + break; + } }); Self { address, diff --git a/crates/workspace-server/src/server_residual_execution_tests.rs b/crates/workspace-server/src/server_residual_execution_tests.rs new file mode 100644 index 000000000..8b0e3620a --- /dev/null +++ b/crates/workspace-server/src/server_residual_execution_tests.rs @@ -0,0 +1,843 @@ +use super::*; + +fn registry_record(worker: RuntimeWorkerRef) -> WorkerRegistryRecord { + WorkerRegistryRecord { + workspace_id: WORKSPACE.into(), + worker, + display_name: "current stored label".into(), + profile: Some("builtin:coder".into()), + retention_state: "normal".into(), + transcript_ref: None, + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: "2026-01-01T00:00:00Z".into(), + updated_at: "2026-01-01T00:00:00Z".into(), + } +} + +fn assigned_ticket(fixture: &Fixture) -> String { + let record = registry_record(RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + Uuid::now_v7().to_string(), + )); + fixture.store.upsert_worker_registry(&record).unwrap(); + let mut input = ticket::NewTicket::new("cancel this implementation"); + input.workflow_state = Some(TicketState::Open); + let ticket = browser_ticket_backend(&fixture.api) + .unwrap() + .create(input) + .unwrap(); + fixture + .store + .set_current_ticket_worker_assignment( + &TicketWorkerAssignmentRecord { + workspace_id: WORKSPACE.into(), + ticket_id: ticket.id.clone(), + assignment_id: "current-coder".into(), + worker: record.worker, + assigned_by: "fixture-user".into(), + assigned_at: "2026-01-01T00:00:00Z".into(), + }, + None, + "assign-event", + "assign-operation", + false, + ) + .unwrap(); + ticket.id +} + +async fn fixture() -> Fixture { + let fixture = Fixture::new().await; + fixture.api.worker_projection.shutdown_async().await; + observed(fixture.executor.drain()).await.unwrap(); + fixture +} + +fn cancel_request() -> serde_json::Value { + serde_json::json!({ + "operation_id": "cancel-operation", "assignment_id": "current-coder", "reason": "use the corrected design" + }) +} + +#[tokio::test(flavor = "current_thread")] +async fn cancelled_ticket_implementation_keeps_assignment_commit_owned() { + let fixture = fixture().await; + let ticket = assigned_ticket(&fixture); + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut response = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + fixture.inner_router().oneshot(Fixture::request( + "POST", + &format!("/api/w/{WORKSPACE}/tickets/{ticket}/implementation-cancellations"), + cancel_request(), + )), + )); + assert!( + futures::poll!(&mut response).is_pending(), + "cancellation must admit an owned stage, not run SQL on Tokio" + ); + observed(waiting).await.unwrap(); // Actual cancellation transaction, after Runtime/session cleanup. + runtime_progress_while_locked(&lock).await; + drop(response); + assert_eq!( + leases.load(Ordering::SeqCst), + 1, + "actual mixed cancellation must keep its authority after the waiter drops" + ); + lock.release(); + busy.resume(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + assert!( + fixture + .store + .get_active_ticket_worker_assignment(WORKSPACE, &ticket) + .unwrap() + .is_none() + ); + assert!( + fixture + .store + .get_ticket_assignment_operation(WORKSPACE, "cancel-operation") + .unwrap() + .is_some() + ); + let detail = browser_ticket_detail(&fixture.api, &ticket).unwrap().0; + assert!(detail.current_worker.is_none()); + assert_eq!(detail.state.as_str(), TicketState::Open.as_str()); +} + +async fn saturated_host(fixture: &Fixture) -> Vec> { + let mut releases = Vec::new(); + let (started, running) = oneshot::channel(); + let (release, permit) = mpsc::channel::<()>(); + releases.push(release); + fixture + .host_executor + .dispatch(move || { + let _ = started.send(()); + let _ = permit.recv_timeout(WATCHDOG); + }) + .unwrap(); + observed(running).await.unwrap(); + let (release, permit) = mpsc::channel::<()>(); + releases.push(release); + fixture + .host_executor + .dispatch(move || { + let _ = permit.recv_timeout(WATCHDOG); + }) + .unwrap(); + releases +} + +#[tokio::test(flavor = "current_thread")] +async fn cancellation_host_overload_leaves_the_coder_assignment_unchanged() { + let fixture = fixture().await; + let ticket = assigned_ticket(&fixture); + let before = fixture + .store + .get_active_ticket_worker_assignment(WORKSPACE, &ticket) + .unwrap() + .unwrap(); + let releases = saturated_host(&fixture).await; + let body = response_json( + observed(fixture.inner_router().oneshot(Fixture::request( + "POST", + &format!("/api/w/{WORKSPACE}/tickets/{ticket}/implementation-cancellations"), + cancel_request(), + ))) + .await + .unwrap(), + StatusCode::SERVICE_UNAVAILABLE, + ) + .await; + assert!(body.to_string().contains("db_capacity_exceeded"), "{body}"); + assert_eq!( + fixture + .store + .get_active_ticket_worker_assignment(WORKSPACE, &ticket) + .unwrap() + .unwrap(), + before + ); + assert!( + fixture + .store + .get_ticket_assignment_operation(WORKSPACE, "cancel-operation") + .unwrap() + .is_none() + ); + drop(releases); + observed(fixture.host_executor.drain()).await.unwrap(); +} + +fn seed_unrepaired_mr(fixture: &Fixture) { + let ticket = browser_ticket_backend(&fixture.api) + .unwrap() + .create(ticket::NewTicket::new("repair source selector")) + .unwrap(); + fixture + .store + .upsert_repository(&crate::store::RepositoryRecord { + workspace_id: WORKSPACE.into(), + repository_id: "fixture-repo".into(), + repository_key: "fixture-repo".into(), + kind: "git".into(), + provider: Some("git".into()), + source: server_api::RepositorySource { + kind: server_api::RepositorySourceKind::LocalPath, + uri: fixture + ._dir + .path() + .join("unused-repository") + .to_string_lossy() + .into_owned(), + }, + default_ref: Some("develop".into()), + source_fingerprint: "sha256:fixture".into(), + observed_status: server_api::RepositoryObservedStatus::Unverified, + observed_at: None, + created_at: "2026-01-01T00:00:00Z".into(), + updated_at: "2026-01-01T00:00:00Z".into(), + }) + .unwrap(); + // Persist the external unrepaired-selector shape. No active assignment: + // the real handler must reach its assignment read, then reject without provider dispatch. + let conn = Connection::open(&fixture.api.config.database_path).unwrap(); + conn.execute("INSERT INTO merge_requests VALUES (?1,'repair-fixture','fixture-repo','open',NULL,'develop',?2,?2)", rusqlite::params![WORKSPACE, "2026-01-01T00:00:00Z"]).unwrap(); + conn.execute("INSERT INTO merge_request_ticket_relations VALUES (?1,'repair-fixture',?2,'implements',?3)", rusqlite::params![WORKSPACE, ticket.id, "2026-01-01T00:00:00Z"]).unwrap(); +} + +fn repair_headers() -> HeaderMap { + let mut headers = HeaderMap::new(); + headers.insert( + "cookie", + "db_execution_session=fixture-browser-token" + .parse() + .unwrap(), + ); + headers +} + +fn repair_input() -> Json { + Json(server_api::RepairMergeRequestSelectorRequest { + selector_from: "work/fixed-selector".into(), + reason: "repair the imported selector".into(), + explicit_confirmation: true, + }) +} + +#[tokio::test(flavor = "current_thread")] +async fn authenticated_selector_repair_waits_off_tokio_and_keeps_cancelled_host_ownership() { + let fixture = fixture().await; + seed_unrepaired_mr(&fixture); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut repair = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + scoped_repair_merge_request_selector( + State(fixture.api.clone()), + repair_headers(), + AxumPath((WORKSPACE.into(), "repair-fixture".into())), + repair_input(), + ), + )); + assert!(futures::poll!(&mut repair).is_pending()); // Browser authentication DB stage only. + observed(fixture.executor.drain()).await.unwrap(); + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let writer = fixture.store.clone(); + fixture + .executor + .dispatch(move || { + writer + .mark_api_token_used("missing", "2026-01-01T00:00:00Z") + .unwrap(); + }) + .unwrap(); + observed(waiting).await.unwrap(); + assert!( + futures::poll!(&mut repair).is_pending(), + "post-auth selector repair must not wait on the connection mutex on Tokio" + ); + runtime_progress_while_locked(&lock).await; + drop(repair); + assert_eq!( + leases.load(Ordering::SeqCst), + 1, + "actual repair stage must retain authority after cancellation" + ); + lock.release(); + busy.resume(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + let mr = merge_request_store(&fixture.api, WORKSPACE) + .unwrap() + .get_by_id(WORKSPACE, "repair-fixture") + .unwrap(); + assert!(mr.selector_from.is_none()); + assert!( + merge_request_store(&fixture.api, WORKSPACE) + .unwrap() + .thread_page_by_id(WORKSPACE, "repair-fixture", None, 10) + .unwrap() + .is_empty() + ); +} + +#[tokio::test(flavor = "current_thread")] +async fn selector_repair_host_overload_preserves_confirmation_and_no_side_effects() { + let fixture = fixture().await; + seed_unrepaired_mr(&fixture); + let releases = saturated_host(&fixture).await; + let error = scoped_repair_merge_request_selector( + State(fixture.api.clone()), + repair_headers(), + AxumPath((WORKSPACE.into(), "repair-fixture".into())), + repair_input(), + ) + .await + .unwrap_err(); + let body = response_json(error.into_response(), StatusCode::SERVICE_UNAVAILABLE).await; + assert!(body.to_string().contains("db_capacity_exceeded"), "{body}"); + let mut input = repair_input(); + input.0.explicit_confirmation = false; + let error = scoped_repair_merge_request_selector( + State(fixture.api.clone()), + repair_headers(), + AxumPath((WORKSPACE.into(), "repair-fixture".into())), + input, + ) + .await + .unwrap_err(); + assert!( + !matches!(error.error, Error::DbCapacityExceeded), + "browser confirmation denial must precede Host admission" + ); + let mr = merge_request_store(&fixture.api, WORKSPACE) + .unwrap() + .get_by_id(WORKSPACE, "repair-fixture") + .unwrap(); + assert!(mr.selector_from.is_none()); + drop(releases); + observed(fixture.host_executor.drain()).await.unwrap(); +} + +#[tokio::test(flavor = "current_thread")] +async fn post_restore_projection_waits_off_tokio_and_preserves_historical_results() { + let fixture = fixture().await; + let record = registry_record(RuntimeWorkerRef::new( + EMBEDDED_WORKER_RUNTIME_ID, + Uuid::now_v7().to_string(), + )); + fixture.store.upsert_worker_registry(&record).unwrap(); + let mut historical = worker_summary_from_registry(&record); + historical.label = "historical label".into(); + historical.state = "paused".into(); + let result = || crate::hosts::InternalWorkerRestoreResult { + state: server_api::WorkerRestoreState::Rejected, + worker: Some(historical.clone()), + diagnostics: Vec::new(), + }; + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let writer = fixture.store.clone(); + fixture + .executor + .dispatch(move || { + writer + .mark_api_token_used("missing", "2026-01-01T00:00:00Z") + .unwrap(); + }) + .unwrap(); + observed(waiting).await.unwrap(); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let mut projection = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + project_worker_restore_response_owned( + fixture.api.clone(), + record.worker.runtime_id.clone(), + record.worker.worker_id.clone(), + result(), + ), + )); + assert!( + futures::poll!(&mut projection).is_pending(), + "stored projection must await a whole DB stage after Host restore returns" + ); + runtime_progress_while_locked(&lock).await; + assert_eq!( + observed(fixture.host_executor.run(|| Ok::<_, Error>("control"))) + .await + .unwrap(), + "control" + ); + let error = project_worker_restore_response_owned( + fixture.api.clone(), + record.worker.runtime_id.clone(), + record.worker.worker_id.clone(), + result(), + ) + .await + .unwrap_err(); + response_json(error.into_response(), StatusCode::SERVICE_UNAVAILABLE).await; + let Json(no_worker) = project_worker_restore_response_owned( + fixture.api.clone(), + record.worker.runtime_id.clone(), + record.worker.worker_id.clone(), + crate::hosts::InternalWorkerRestoreResult { + state: server_api::WorkerRestoreState::Rejected, + worker: None, + diagnostics: Vec::new(), + }, + ) + .await + .unwrap(); + assert_eq!( + no_worker.result.state, + server_api::WorkerRestoreState::Rejected + ); + drop(projection); + assert_eq!(leases.load(Ordering::SeqCst), 1); + lock.release(); + busy.resume(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + let Json(response) = project_worker_restore_response_owned( + fixture.api.clone(), + record.worker.runtime_id.clone(), + record.worker.worker_id.clone(), + result(), + ) + .await + .unwrap(); + assert_eq!( + response.result.state, + server_api::WorkerRestoreState::Rejected + ); + let worker = response.result.worker.unwrap(); + assert_eq!(worker.label, record.display_name); + assert_eq!( + worker.state, "paused", + "a historical receipt is not a new Runtime observation" + ); + assert_eq!( + fixture + .store + .get_worker_registry(WORKSPACE, &record.worker) + .unwrap() + .unwrap(), + record, + "response projection must not write a new observation" + ); + observed(fixture.executor.drain()).await.unwrap(); +} + +#[tokio::test(flavor = "current_thread")] +async fn ticket_source_ref_reads_and_ready_publication_leave_sql_and_tokio_free() { + for operation in ["detail", "query", "ready"] { + let repository_dir = tempfile::tempdir().unwrap(); + let source = server_api::RepositorySource { + kind: server_api::RepositorySourceKind::LocalPath, + uri: repository_dir.path().to_string_lossy().into_owned(), + }; + let fixture = Fixture::with_repositories(vec![ConfiguredRepository { + id: "fixture-repo".into(), + repository_key: "fixture-repo".into(), + provider: "git".into(), + source_fingerprint: crate::repository_source::repository_source_fingerprint(&source), + source, + observed_status: server_api::RepositoryObservedStatus::Unverified, + observed_at: None, + path: Some(repository_dir.path().into()), + default_selector: Some("develop".into()), + }]) + .await; + fixture.api.worker_projection.shutdown_async().await; + observed(fixture.executor.drain()).await.unwrap(); + let mut reply = HeldRuntimeReply::for_path("/v1/repository-refs/observe"); + // Narrow Runtime HTTP/provider placement fixture, not a signed authority + // or complete Git/Runtime E2E test. Existing signed routes test that fence. + install_remote_fixture( + &fixture, + "ref-runtime", + &format!("http://{}", reply.address), + false, + ); + let worker = RuntimeWorkerRef::new("ref-runtime", Uuid::now_v7().to_string()); + fixture + .store + .upsert_worker_registry(®istry_record(worker.clone())) + .unwrap(); + let mut input = ticket::NewTicket::new("source ref projection"); + input.workflow_state = Some(TicketState::Open); + let ticket = browser_ticket_backend(&fixture.api) + .unwrap() + .create(input) + .unwrap(); + fixture + .store + .set_current_ticket_worker_assignment( + &TicketWorkerAssignmentRecord { + workspace_id: WORKSPACE.into(), + ticket_id: ticket.id.clone(), + assignment_id: "ref-assignment".into(), + worker: worker.clone(), + assigned_by: "fixture-user".into(), + assigned_at: "2026-01-01T00:00:00Z".into(), + }, + None, + "ref-event", + "ref-operation", + false, + ) + .unwrap(); + merge_request_store(&fixture.api, WORKSPACE) + .unwrap() + .open_merge_request(merge_request::OpenMergeRequest { + merge_request_id: "ref-mr".into(), + ticket_id: ticket.id.clone(), + repository_id: "fixture-repo".into(), + selector_from: "work/source".into(), + selector_to: "develop".into(), + summary: String::new(), + auth: merge_request::MergeRequestAuth { + workspace_id: WORKSPACE.into(), + repository_id: "fixture-repo".into(), + runtime_id: worker.runtime_id, + worker_id: worker.worker_id, + assignment_id: "ref-assignment".into(), + }, + now: Utc::now(), + }) + .unwrap(); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let api = fixture.api.clone(); + let ticket_id = ticket.id.clone(); + let mut response = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + async move { + match operation { + "detail" => scoped_get_ticket( + State(api), + AxumPath(ScopedRecordPath { + workspace_id: WORKSPACE.into(), + id: ticket_id, + }), + ) + .await + .map(|_| ()), + "query" => scoped_query_tickets( + State(api), + AxumPath(ScopedWorkspacePath { + workspace_id: WORKSPACE.into(), + }), + Json(TicketQueryRequest::default()), + ) + .await + .map(|_| ()), + _ => scoped_mark_ticket_ready_from_browser( + State(api), + HeaderMap::new(), + AxumPath(ScopedRecordPath { + workspace_id: WORKSPACE.into(), + id: ticket_id, + }), + Json(InternalTicketMarkReadyRequest { + operation_key: "ready-operation".into(), + reason: None, + intake_summary: None, + }), + ) + .await + .map(|_| ()), + } + }, + )); + assert!(futures::poll!(&mut response).is_pending()); + let request = observed(reply.entered.take().unwrap()).await.unwrap(); + assert!( + request.contains("/v1/repository-refs/observe"), + "{operation}: {request}" + ); + let store = fixture.store.clone(); + observed( + fixture + .executor + .run(move || store.get_workspace_memory_settings(WORKSPACE)), + ) + .await + .unwrap(); + assert_eq!( + observed(tokio::spawn(async { "control" })).await.unwrap(), + "control" + ); + observed(tokio::time::sleep(StdDuration::from_millis(1))).await; + assert!( + !reply.upstream_released.load(Ordering::SeqCst), + "watchdog release is not provider progress" + ); + if operation == "ready" { + let saved = browser_ticket_backend(&fixture.api) + .unwrap() + .show(TicketIdOrSlug::Id(ticket.id)) + .unwrap(); + assert_eq!( + saved.meta.readiness.as_deref(), + Some("ready"), + "Ready mutation must be durable before the held detail source observation" + ); + assert_eq!( + saved + .events + .iter() + .filter( + |event| event.attributes.get("operation_key").map(String::as_str) + == Some("ready-operation") + ) + .count(), + 1 + ); + assert_eq!( + saved.meta.workflow_state, + TicketState::Open, + "Ready must commit before detail publication" + ); + } + drop(response); + assert_eq!( + leases.load(Ordering::SeqCst), + 1, + "{operation}: cancelled provider stage lost actual ownership" + ); + fixture.host_executor.dispatch(|| {}).unwrap(); // Remaining finite queued slot. + let error = scoped_get_ticket( + State(fixture.api.clone()), + AxumPath(ScopedRecordPath { + workspace_id: WORKSPACE.into(), + id: "unused".into(), + }), + ) + .await + .unwrap_err(); + response_json(error.into_response(), StatusCode::SERVICE_UNAVAILABLE).await; + reply.finish(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + } +} + +#[tokio::test(flavor = "current_thread")] +async fn worker_control_source_lookup_waits_off_tokio_and_retains_finite_host_ownership() { + let fixture = fixture().await; + let mut reply = HeldRuntimeReply::new(); + install_remote_fixture( + &fixture, + "source-runtime", + &format!("http://{}", reply.address), + false, + ); + let source = RuntimeWorkerRef::new("source-runtime", Uuid::now_v7().to_string()); + let mut headers = HeaderMap::new(); + headers.insert("x-yoi-runtime-id", source.runtime_id.parse().unwrap()); + headers.insert("x-yoi-worker-id", source.worker_id.parse().unwrap()); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let request = || { + Json(server_api::RuntimeWorkerLifecycleRequest { + reason: None, + ticket_assignment: None, + }) + }; + let path = || { + AxumPath(ScopedRuntimeWorkerPath { + workspace_id: WORKSPACE.into(), + worker: source.clone(), + }) + }; + let mut control = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + cancel_known_worker( + State(fixture.api.clone()), + path(), + headers.clone(), + request(), + ), + )); + assert!( + futures::poll!(&mut control).is_pending(), + "control source lookup must not synchronously join Runtime HTTP on Tokio" + ); + let request_line = observed(reply.entered.take().unwrap()).await.unwrap(); + assert!(request_line.contains("/v1/workers/"), "{request_line}"); + let store = fixture.store.clone(); + observed( + fixture + .executor + .run(move || store.get_workspace_memory_settings(WORKSPACE)), + ) + .await + .unwrap(); + assert_eq!( + observed(tokio::spawn(async { "control" })).await.unwrap(), + "control" + ); + observed(tokio::time::sleep(StdDuration::from_millis(1))).await; + assert!(!reply.upstream_released.load(Ordering::SeqCst)); + drop(control); + assert_eq!(leases.load(Ordering::SeqCst), 1); + fixture.host_executor.dispatch(|| {}).unwrap(); + let error = cancel_known_worker(State(fixture.api.clone()), path(), headers, request()) + .await + .unwrap_err(); + let body = response_json(error.into_response(), StatusCode::SERVICE_UNAVAILABLE).await; + assert!( + body.to_string().contains("db_capacity_exceeded"), + "Host overload is availability, not identity denial: {body}" + ); + reply.finish(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); +} + +#[tokio::test(flavor = "current_thread")] +async fn ticket_tracking_and_case_authority_runtime_lookup_leave_db_and_tokio_free() { + for operation in ["tracking", "case"] { + let fixture = fixture().await; + let ticket = browser_ticket_backend(&fixture.api) + .unwrap() + .create(ticket::NewTicket::new("decision authority")) + .unwrap(); + let mut reply = HeldRuntimeReply::new(); + install_remote_fixture( + &fixture, + "orchestrator-runtime", + &format!("http://{}", reply.address), + false, + ); + let memory = fixture + .store + .get_workspace_memory_settings(WORKSPACE) + .unwrap(); + let reservation = fixture + .store + .reserve_worker_create( + WORKSPACE, + "orchestrator-runtime", + "orch-create", + "orch-create-fingerprint", + Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY), + &memory, + ) + .unwrap(); + let source = + RuntimeWorkerRef::new("orchestrator-runtime", reservation.worker_id.to_string()); + fixture + .store + .upsert_worker_registry(®istry_record(source.clone())) + .unwrap(); + fixture + .store + .complete_worker_create_reservation(WORKSPACE, reservation.worker_id) + .unwrap(); + let context = server_api::ServerRequestContext { + actor: None, + worker_source: None, + runtime_source: Some(server_api::ServerRuntimeSource { + authentication_binding_id: "verified-fixture-binding".into(), + runtime_id: source.runtime_id.clone(), + worker_id: Some(source.worker_id.clone()), + }), + origin: None, + transport_headers: Vec::new(), + }; + let backend = browser_ticket_backend(&fixture.api).unwrap(); + let before = backend.show(TicketIdOrSlug::Id(ticket.id.clone())).unwrap(); + let leases = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let api = fixture.api.clone(); + let ticket_id = ticket.id.clone(); + let mut decision = Box::pin(crate::db_executor::with_operation_lifetime( + Arc::new(CountJobLifetime(leases.clone())), + async { + if operation == "tracking" { + scoped_ticket_assignment_tracking_update_with_context( + State(api), + HeaderMap::new(), + &context, + AxumPath((WORKSPACE.into(), ticket_id)), + Json(ticket::TicketAssignmentTrackingUpdate { + assignment_id: "no-delegation".into(), + expected_last_event_id: "baseline".into(), + operation_key: "track-operation".into(), + tracking: ticket::TicketAssignmentTracking { + status: ticket::TicketAssignmentTrackingStatus::Inprogress, + waiting_reason: None, + progress: None, + next_action: None, + result: None, + }, + }), + ) + .await + .map(|_| ()) + } else { + apply_ticket_case_update( + State(api), + Some(source), + None, + AxumPath((WORKSPACE.into(), ticket_id)), + Json(ticket::TicketCaseUpdate { + expected_content_digest: ticket::ticket_content_digest(&before), + expected_state: before.meta.workflow_state, + expected_closure_operation_key: None, + operation_key: "case-operation".into(), + action: ticket::TicketCaseAction::Close { + reason: ticket::TicketClosureReason::Completed, + }, + explanation: "check current authority".into(), + references: Vec::new(), + author: None, + }), + ) + .await + .map(|_| ()) + } + }, + )); + assert!(futures::poll!(&mut decision).is_pending()); + let request = observed(reply.entered.take().unwrap()).await.unwrap(); + assert!(request.contains("/v1/workers/"), "{operation}: {request}"); + let store = fixture.store.clone(); + observed( + fixture + .executor + .run(move || store.get_workspace_memory_settings(WORKSPACE)), + ) + .await + .unwrap(); + assert_eq!( + observed(tokio::spawn(async { "control" })).await.unwrap(), + "control" + ); + observed(tokio::time::sleep(StdDuration::from_millis(1))).await; + assert!(!reply.upstream_released.load(Ordering::SeqCst)); + drop(decision); + assert_eq!(leases.load(Ordering::SeqCst), 1); + reply.finish(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + assert_eq!(leases.load(Ordering::SeqCst), 0); + assert_eq!( + backend.show(TicketIdOrSlug::Id(ticket.id)).unwrap(), + before, + "failed live authority must not make a decision" + ); + } +}