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

fix(server): own residual Ticket MR and worker request stages

Browse Source
This commit is contained in:
Hare
2026-10-12 14:46:05 +09:00
parent ebce7ab7b1
commit 5873f25073
3 changed files with 1196 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+298 -170
View File
@@ -13555,8 +13555,8 @@ async fn scoped_get_ticket(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRecordPath>,
) -> ApiResult<Json<TicketDetail>> {
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<ScopedWorkspacePath>,
Json(query): Json<TicketQueryRequest>,
) -> ApiResult<Json<TicketQueryResponse>> {
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<ScopedRecordPath>,
Json(query): Json<TicketShowRequest>,
) -> ApiResult<Json<TicketDetail>> {
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<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRecordPath>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentsResponse>> {
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<crate::store::TicketOrchestratorAssignmentSourceContext>,
) -> ApiResult<Json<server_api::TicketRoleAssignmentMutationResponse>> {
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<server_api::CancelTicketImplementationRequest>,
) -> ApiResult<Json<TicketDetail>> {
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<ScopedRecordPath>,
Json(request): Json<server_api::BrowserEditTicketRequest>,
) -> ApiResult<Json<TicketDetail>> {
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<ScopedRecordPath>,
Json(request): Json<server_api::BrowserAppendTicketEventRequest>,
) -> ApiResult<Json<TicketDetail>> {
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<InternalTicketMarkReadyRequest>,
) -> ApiResult<Json<TicketDetail>> {
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<TicketBackendOperationResult> {
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<TicketBackendOperationResult> {
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<ticket::TicketAssignmentTrackingUpdate>,
) -> ApiResult<Json<ticket::TicketAssignmentTrackingProjection>> {
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<ticket::TicketCaseUpdate>,
) -> ApiResult<Json<ticket::Ticket>> {
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<WorkerMutationSource> {
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<server_api::RuntimeWorkerInputRequest>,
) -> ApiResult<Json<server_api::RuntimeWorkerInputResult>> {
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<server_api::RuntimeWorkerLifecycleRequest>,
) -> ApiResult<Json<server_api::RuntimeWorkerLifecycleResult>> {
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<server_api::RuntimeWorkerLifecycleRequest>,
) -> ApiResult<Json<server_api::RuntimeWorkerLifecycleResult>> {
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<Json<server_api::WorkerRestoreResponse>> {
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<server_api::WorkerObservationSubjectRef>,
) -> ApiResult<Json<server_api::WorkerObservationCaptureResponse>> {
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<WorkspaceApi>,
AxumPath(id): AxumPath<String>,
) -> ApiResult<Json<TicketDetail>> {
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<Json<server_api::WorkerRestoreResponse>> {
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<Json<server_api::WorkerRestoreResponse>> {
// 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<()> {
crates/workspace-server/src/server_db_execution_tests.rs
+55 -17
View File
@@ -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<ConfiguredRepository>) -> 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<String>) -> 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,
crates/workspace-server/src/server_residual_execution_tests.rs
+843
View File
@@ -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<mpsc::Sender<()>> {
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<server_api::RepairMergeRequestSelectorRequest> {
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(&registry_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(&registry_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"
);
}
}