From d101b4bacbda2406d09fcec057184c729c591fe7 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 1 Oct 2026 13:52:57 +0900 Subject: [PATCH] fix: queue ticket checks with exact snapshots --- crates/ticket/src/lib.rs | 18 +- crates/ticket/src/tool.rs | 4 + crates/workspace-server/src/backend_job.rs | 3 + crates/workspace-server/src/latest_schema.sql | 2 +- crates/workspace-server/src/server.rs | 520 +++++++++++++++++- crates/workspace-server/src/store.rs | 304 +++++++++- docs/design/ticket-item-checker.md | 2 +- 7 files changed, 799 insertions(+), 54 deletions(-) diff --git a/crates/ticket/src/lib.rs b/crates/ticket/src/lib.rs index 3e5199dc7..0f66c9b20 100644 --- a/crates/ticket/src/lib.rs +++ b/crates/ticket/src/lib.rs @@ -1909,6 +1909,11 @@ pub trait TicketBackend { fn list(&self, filter: TicketListQuery) -> Result>; fn show(&self, id: TicketIdOrSlug) -> Result; fn create(&self, input: NewTicket) -> Result; + fn create_with_snapshot(&self, input: NewTicket) -> Result<(TicketRef, Ticket)> { + let reference = self.create(input)?; + let ticket = self.show(reference.id.clone().into())?; + Ok((reference, ticket)) + } fn edit_item(&self, id: TicketIdOrSlug, edit: TicketItemEdit) -> Result; fn dependency_check(&self, id: TicketIdOrSlug) -> Result; fn add_event(&self, id: TicketIdOrSlug, event: NewTicketEvent) -> Result<()>; @@ -3087,6 +3092,11 @@ impl TicketBackend for SqliteTicketBackend { } fn create(&self, input: NewTicket) -> Result { + self.create_with_snapshot(input) + .map(|(reference, _)| reference) + } + + fn create_with_snapshot(&self, input: NewTicket) -> Result<(TicketRef, Ticket)> { self.with_write(|conn| { if input.title.trim().is_empty() { return Err(TicketError::Conflict( @@ -3167,12 +3177,14 @@ impl TicketBackend for SqliteTicketBackend { }; let resource_key = Self::allocate_resource_key(conn, &self.workspace_id, &id, &now)?; self.insert_ticket(conn, &ticket)?; - Ok(TicketRef { + let reference = TicketRef { id: id.clone(), resource_key: Some(resource_key), - slug: id, + slug: id.clone(), status: TicketStatus::Open, - }) + }; + let persisted = self.load_ticket(conn, &id)?; + Ok((reference, persisted)) }) } diff --git a/crates/ticket/src/tool.rs b/crates/ticket/src/tool.rs index 42ae82bd2..f1d58daa5 100644 --- a/crates/ticket/src/tool.rs +++ b/crates/ticket/src/tool.rs @@ -268,6 +268,10 @@ impl TicketBackend for TicketToolBackend { self.backend.create(input) } + fn create_with_snapshot(&self, input: NewTicket) -> TicketResult<(TicketRef, Ticket)> { + self.backend.create_with_snapshot(input) + } + fn edit_item(&self, id: TicketIdOrSlug, edit: crate::TicketItemEdit) -> TicketResult { self.backend.edit_item(id, edit) } diff --git a/crates/workspace-server/src/backend_job.rs b/crates/workspace-server/src/backend_job.rs index c1c83e50c..e7f7bce77 100644 --- a/crates/workspace-server/src/backend_job.rs +++ b/crates/workspace-server/src/backend_job.rs @@ -62,6 +62,7 @@ impl BackendJobState { #[serde(rename_all = "snake_case")] pub enum BackendJobAttemptState { Reserved, + Dispatching, Dispatched, Completed, Failed, @@ -72,6 +73,7 @@ impl BackendJobAttemptState { pub(crate) fn as_str(self) -> &'static str { match self { Self::Reserved => "reserved", + Self::Dispatching => "dispatching", Self::Dispatched => "dispatched", Self::Completed => "completed", Self::Failed => "failed", @@ -82,6 +84,7 @@ impl BackendJobAttemptState { pub(crate) fn parse(value: &str) -> Result { match value { "reserved" => Ok(Self::Reserved), + "dispatching" => Ok(Self::Dispatching), "dispatched" => Ok(Self::Dispatched), "completed" => Ok(Self::Completed), "failed" => Ok(Self::Failed), diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index 11b68f143..faa3d7d30 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -836,7 +836,7 @@ CREATE TABLE backend_job_attempts ( attempt_id TEXT NOT NULL, attempt INTEGER NOT NULL CHECK (attempt > 0 AND attempt <= 3), input_revision TEXT NOT NULL, - state TEXT NOT NULL CHECK (state IN ('reserved', 'dispatched', 'completed', 'failed', 'unknown')), + state TEXT NOT NULL CHECK (state IN ('reserved', 'dispatching', 'dispatched', 'completed', 'failed', 'unknown')), runtime_id TEXT, worker_id TEXT, runtime_run_id TEXT, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index c1ff1eb56..f4d5dbf59 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1394,6 +1394,7 @@ pub struct WorkspaceApi { companion: Arc, orchestrator_spawn_lock: Arc>, orchestrator_attention_fingerprint: Arc>>, + backend_job_drain_requests: Arc, observation_proxy: BackendObservationProxy, runtime_subscription_broker: RuntimeSubscriptionBroker, pub(crate) worker_projection: Arc, @@ -3208,6 +3209,7 @@ impl WorkspaceApi { companion, orchestrator_spawn_lock: Arc::new(std::sync::Mutex::new(())), orchestrator_attention_fingerprint: Arc::new(Mutex::new(None)), + backend_job_drain_requests: Arc::new(AtomicU64::new(0)), observation_proxy, runtime_subscription_broker, worker_projection, @@ -3297,7 +3299,11 @@ impl WorkspaceApi { let reservation = self.store .reserve_backend_job(&self.config.workspace_id, request, &now)?; - self.dispatch_backend_job_reservation(reservation) + let result = self.dispatch_backend_job_reservation(reservation); + if result.is_err() { + self.drain_reserved_backend_jobs(); + } + result } /// Explicit re-evaluation creates a new bounded attempt only after a @@ -3310,7 +3316,11 @@ impl WorkspaceApi { let reservation = self.store .reserve_backend_job_retry(&self.config.workspace_id, job_id, &now)?; - self.dispatch_backend_job_reservation(reservation) + let result = self.dispatch_backend_job_reservation(reservation); + if result.is_err() { + self.drain_reserved_backend_jobs(); + } + result } fn dispatch_backend_job_reservation( @@ -3323,6 +3333,15 @@ impl WorkspaceApi { { return Ok(reservation); } + let (reservation, claimed) = self.store.claim_backend_job_attempt_dispatch( + &self.config.workspace_id, + &reservation.job.request.job_id, + &reservation.attempt.attempt_id, + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + if !claimed { + return Ok(reservation); + } let request = &reservation.job.request; let worker_input = request.worker_input(&reservation.attempt.attempt_id)?; let control_operation = WorkerControlOperation { @@ -3523,6 +3542,94 @@ impl WorkspaceApi { }) } + fn drain_reserved_backend_jobs(&self) { + if self + .backend_job_drain_requests + .fetch_add(1, Ordering::AcqRel) + != 0 + { + return; + } + loop { + self.drain_reserved_backend_jobs_inner(); + let requests = self.backend_job_drain_requests.swap(0, Ordering::AcqRel); + if requests <= 1 { + return; + } + // A synchronous fast result can request another drain while this + // runner is dispatching. Re-run without recursion, unless a caller + // arriving after the swap has already become the next runner. + if self + .backend_job_drain_requests + .fetch_add(1, Ordering::AcqRel) + != 0 + { + return; + } + } + } + + fn drain_reserved_backend_jobs_inner(&self) { + let page_limit = usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS); + loop { + let attempts = match self + .store + .list_reserved_backend_job_attempts(&self.config.workspace_id, page_limit) + { + Ok(attempts) => attempts, + Err(error) => { + tracing::warn!(%error, "failed to load queued Backend Job attempts"); + return; + } + }; + if attempts.is_empty() { + return; + } + let mut made_progress = false; + for attempt in &attempts { + let job = match self + .store + .get_backend_job(&self.config.workspace_id, &attempt.job_id) + { + Ok(Some(job)) => job, + Ok(None) => continue, + Err(error) => { + tracing::warn!(%error, job_id = %attempt.job_id, "failed to load queued Backend Job"); + continue; + } + }; + match self.dispatch_backend_job_reservation(BackendJobReservation { + job, + attempt: attempt.clone(), + replayed: true, + }) { + Ok(reservation) => { + made_progress |= + reservation.attempt.state != BackendJobAttemptState::Reserved; + } + Err(error) => { + tracing::warn!(%error, "queued Backend Job dispatch failed"); + made_progress |= self + .store + .get_backend_job_attempt( + &self.config.workspace_id, + &attempt.job_id, + &attempt.attempt_id, + ) + .ok() + .flatten() + .is_some_and(|current| { + current.state != BackendJobAttemptState::Reserved + }); + } + } + } + if attempts.len() < page_limit || !made_progress { + return; + } + } + } + /// Reconcile durable non-terminal attempts after Backend restart. Reserved /// attempts are replay-dispatched through their deterministic allocation; /// dispatched attempts keep running only while their bound Worker remains @@ -3551,6 +3658,17 @@ impl WorkspaceApi { tracing::warn!(%error, "Backend Job reserved-attempt recovery failed"); } } + BackendJobAttemptState::Dispatching => { + self.store.finish_backend_job_attempt( + &self.config.workspace_id, + &attempt.job_id, + &attempt.attempt_id, + BackendJobAttemptState::Unknown, + "recovery_dispatch_claim_unknown", + "a Backend Job dispatch claim could not be safely replayed after restart", + &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), + )?; + } BackendJobAttemptState::Dispatched => { if attempt.runtime_run_id.is_none() { self.store.finish_backend_job_attempt( @@ -3591,6 +3709,7 @@ impl WorkspaceApi { | BackendJobAttemptState::Unknown => {} } } + self.drain_reserved_backend_jobs(); Ok(()) } @@ -3604,9 +3723,10 @@ impl WorkspaceApi { let Ok(handle) = tokio::runtime::Handle::try_current() else { return; }; - let store = self.store.clone(); - let runtime = self.runtime.clone(); - let workspace_id = self.config.workspace_id.clone(); + let api = self.clone(); + let store = api.store.clone(); + let runtime = api.runtime.clone(); + let workspace_id = api.config.workspace_id.clone(); let job_id = job_id.to_string(); let attempt_id = attempt_id.to_string(); handle.spawn(async move { @@ -3642,6 +3762,7 @@ impl WorkspaceApi { }, ); } + api.drain_reserved_backend_jobs(); } }); } @@ -3694,6 +3815,7 @@ impl WorkspaceApi { &detail, &accepted_at, )?; + self.drain_reserved_backend_jobs(); } return Err(error); } @@ -3707,6 +3829,7 @@ impl WorkspaceApi { if let Some(target) = acceptance.job.request.notification_target.as_ref() { self.deliver_backend_job_result(&acceptance, target); } + self.drain_reserved_backend_jobs(); Ok(acceptance) } @@ -12338,6 +12461,7 @@ fn schedule_ticket_item_check( %error, "asynchronous Ticket item checker dispatch failed" ); + api.drain_reserved_backend_jobs(); } }); } @@ -12455,7 +12579,19 @@ async fn execute_ticket_rest_operation( backend = backend.with_event_attributes(event_attributes); } - let result = execute_ticket_backend_operation(&backend, operation).map_err(Error::from)?; + let (result, created_ticket) = match operation { + TicketBackendOperation::Create { input } => { + let (reference, ticket) = backend.create_with_snapshot(input).map_err(Error::from)?; + ( + TicketBackendOperationResult::TicketRef(reference), + Some(ticket), + ) + } + operation => ( + execute_ticket_backend_operation(&backend, operation).map_err(Error::from)?, + None, + ), + }; if is_mutation { if let TicketBackendOperationResult::QueueOutcome(outcome) = &result { for ticket_id in &outcome.queued_tickets { @@ -12482,20 +12618,12 @@ async fn execute_ticket_rest_operation( } } if let Some((source_worker, edit)) = ticket_item_check { - let checked_ticket_id = match &result { - TicketBackendOperationResult::TicketRef(ticket) => Some(ticket.id.clone()), - TicketBackendOperationResult::Ticket(ticket) => Some(ticket.meta.id.clone()), + let checked_ticket = created_ticket.or_else(|| match &result { + TicketBackendOperationResult::Ticket(ticket) => Some(ticket.clone()), _ => None, - }; - if let Some(ticket_id) = checked_ticket_id { - match backend.show(ticket_id.clone().into()) { - Ok(ticket) => schedule_ticket_item_check(api, &ticket, edit, source_worker), - Err(error) => tracing::warn!( - %ticket_id, - %error, - "Ticket item was saved but could not be reloaded for asynchronous checking" - ), - } + }); + if let Some(ticket) = checked_ticket { + schedule_ticket_item_check(api, &ticket, edit, source_worker); } } Ok(result) @@ -34335,13 +34463,16 @@ mod tests { job: &crate::backend_job::BackendJobRecord, ) -> crate::backend_job::BackendJobAttemptRecord { let attempt_id = crate::backend_job::attempt_id(&job.request.job_id, job.current_attempt); - for _ in 0..100 { + for _ in 0..500 { let attempt = api .store .get_backend_job_attempt(TEST_WORKSPACE_ID, &job.request.job_id, &attempt_id) .unwrap() .unwrap(); - if attempt.state != BackendJobAttemptState::Reserved { + if !matches!( + attempt.state, + BackendJobAttemptState::Reserved | BackendJobAttemptState::Dispatching + ) { return attempt; } tokio::time::sleep(std::time::Duration::from_millis(10)).await; @@ -34383,7 +34514,7 @@ mod tests { panic!("unexpected Ticket create result") }; entered_rx - .recv_timeout(std::time::Duration::from_secs(1)) + .recv_timeout(std::time::Duration::from_secs(5)) .expect("checker dispatch reached the held fake model"); let job = ticket_check_job(&api, &ticket_ref.id); let checker_input = ticket_item_checker::parse_input(&job.request.input).unwrap(); @@ -34445,6 +34576,291 @@ mod tests { assert!(execution.take_inputs().is_empty()); } + #[tokio::test] + async fn ticket_checker_fake_model_regressions_preserve_classification_boundaries() { + struct Case { + name: &'static str, + body: &'static str, + fake_model_result: Value, + advisory_quote: Option<&'static str>, + } + let cases = vec![ + Case { + name: "writer-scope-positive", + body: "This time I only create the Ticket; I will not implement it.", + fake_model_result: serde_json::json!({"findings": [{ + "category": "writer_scope", + "quote": "I will not implement it", + "reason": "The active requirement records the writer's own implementation boundary.", + "suggestion": "Remove the writer-specific action boundary." + }]}), + advisory_quote: Some("I will not implement it"), + }, + Case { + name: "production-approval-negative", + body: "Production deployment requires approval.", + fake_model_result: serde_json::json!({"findings": []}), + advisory_quote: None, + }, + Case { + name: "quoted-background-negative", + body: "Background bug example: a generated Ticket once said “I will not implement it.” Do not reproduce that sentence in generated requirements.", + fake_model_result: serde_json::json!({"findings": []}), + advisory_quote: None, + }, + Case { + name: "unresolved-versus-settled-positive", + body: "The rollout region is undecided. The rollout region is us-east-1.", + fake_model_result: serde_json::json!({"findings": [{ + "category": "internal_inconsistency", + "quote": "The rollout region is undecided.", + "reason": "The same active text also settles the rollout region as us-east-1.", + "suggestion": "Choose either the unresolved or settled rollout-region statement." + }]}), + advisory_quote: Some("The rollout region is undecided."), + }, + Case { + name: "ambiguous-boundary-confirmation", + body: "Implementation stops after Ticket creation.", + fake_model_result: serde_json::json!({"findings": [{ + "category": "ambiguous_boundary", + "quote": "Implementation stops after Ticket creation.", + "reason": "The text alone does not establish whether this is a delivery constraint or the writer's current action boundary.", + "suggestion": "Confirm whether this limits the product workflow or only the current writer." + }]}), + advisory_quote: Some("Implementation stops after Ticket creation."), + }, + ]; + + let temp = tempfile::tempdir().unwrap(); + let (api, execution) = test_api_with_recording_backend(temp.path()).await; + let source = spawn_ticket_check_source(&api, "ticket-check-boundary-author"); + for case in cases { + let mut input = ticket::NewTicket::new(format!("Boundary case: {}", case.name)); + input.body = MarkdownText::new(case.body); + let Json(TicketBackendOperationResult::TicketRef(reference)) = + execute_worker_ticket_test_operation( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ticket_check_headers(&source), + Json(TicketBackendOperation::Create { input }), + ) + .await + .unwrap() + else { + panic!("{}: unexpected Ticket create result", case.name); + }; + let job = ticket_check_job(&api, &reference.id); + let attempt = wait_for_ticket_check_attempt(&api, &job).await; + execution.take_inputs(); + api.accept_backend_job_result( + attempt.worker.as_ref().unwrap(), + &BackendJobResultSubmission { + job_id: job.request.job_id.clone(), + attempt_id: attempt.attempt_id, + input_revision: job.request.input_revision, + result: case.fake_model_result, + }, + ) + .unwrap_or_else(|error| panic!("{}: {error}", case.name)); + let notifications = execution.take_inputs(); + let advisories = notifications + .iter() + .filter(|(_, content)| content.contains("[Ticket item checker advisory]")) + .collect::>(); + match case.advisory_quote { + Some(quote) => { + assert_eq!(advisories.len(), 1, "{}", case.name); + assert!(advisories[0].1.contains(quote), "{}", case.name); + } + None => assert!(advisories.is_empty(), "{}", case.name), + } + } + } + + #[tokio::test] + async fn ticket_checker_burst_keeps_excess_revisions_durable_and_drains_them() { + let temp = tempfile::tempdir().unwrap(); + let (api, _execution) = test_api_with_recording_backend(temp.path()).await; + let source = spawn_ticket_check_source(&api, "ticket-check-burst-author"); + let mut jobs = Vec::new(); + for index in 0..6 { + let mut input = ticket::NewTicket::new(format!("Burst check {index}")); + input.body = MarkdownText::new(format!("Durable requirement {index}.")); + let Json(result) = execute_worker_ticket_test_operation( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ticket_check_headers(&source), + Json(TicketBackendOperation::Create { input }), + ) + .await + .unwrap(); + let TicketBackendOperationResult::TicketRef(reference) = result else { + panic!("unexpected Ticket create result"); + }; + jobs.push(ticket_check_job(&api, &reference.id)); + } + + let mut initial = Vec::new(); + for _ in 0..200 { + initial = jobs + .iter() + .map(|job| { + api.store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &job.request.job_id, + &crate::backend_job::attempt_id(&job.request.job_id, 1), + ) + .unwrap() + .unwrap() + }) + .collect(); + let dispatched = initial + .iter() + .filter(|attempt| attempt.state == BackendJobAttemptState::Dispatched) + .count(); + let reserved = initial + .iter() + .filter(|attempt| attempt.state == BackendJobAttemptState::Reserved) + .count(); + if dispatched == 4 && reserved == 2 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert_eq!( + initial + .iter() + .filter(|attempt| attempt.state == BackendJobAttemptState::Dispatched) + .count(), + 4 + ); + assert_eq!( + initial + .iter() + .filter(|attempt| attempt.state == BackendJobAttemptState::Reserved) + .count(), + 2, + "capacity must leave excess successful saves durably queued" + ); + + // Restart reconciliation is idempotent for running attempts and retains + // the queued revisions until a slot becomes available. + api.recover_backend_jobs().unwrap(); + let mut completed = std::collections::BTreeSet::new(); + for _ in 0..300 { + for job in &jobs { + let attempt = api + .store + .get_backend_job_attempt( + TEST_WORKSPACE_ID, + &job.request.job_id, + &crate::backend_job::attempt_id(&job.request.job_id, 1), + ) + .unwrap() + .unwrap(); + if attempt.state == BackendJobAttemptState::Dispatched + && !completed.contains(&job.request.job_id) + { + api.accept_backend_job_result( + attempt.worker.as_ref().unwrap(), + &BackendJobResultSubmission { + job_id: job.request.job_id.clone(), + attempt_id: attempt.attempt_id, + input_revision: job.request.input_revision.clone(), + result: serde_json::json!({"findings": []}), + }, + ) + .unwrap(); + completed.insert(job.request.job_id.clone()); + } + } + if completed.len() == jobs.len() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert_eq!(completed.len(), jobs.len()); + assert!(jobs.iter().all(|job| { + api.store + .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id) + .unwrap() + .is_some_and(|job| job.state == BackendJobState::Completed) + })); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn concurrent_worker_edits_bind_exact_saved_snapshots_to_their_sources() { + let temp = tempfile::tempdir().unwrap(); + let (api, _execution) = test_api_with_recording_backend(temp.path()).await; + let source_a = spawn_ticket_check_source(&api, "ticket-check-editor-a"); + let source_b = spawn_ticket_check_source(&api, "ticket-check-editor-b"); + let ticket_ref = browser_ticket_backend(&api) + .unwrap() + .create(ticket::NewTicket::new("Concurrent checker snapshots")) + .unwrap(); + let barrier = Arc::new(std::sync::Barrier::new(2)); + let run_edit = |source: RuntimeWorkerRef, body: &'static str| { + let api = api.clone(); + let ticket_id = ticket_ref.id.clone(); + let barrier = barrier.clone(); + tokio::task::spawn_blocking(move || { + barrier.wait(); + tokio::runtime::Handle::current().block_on(execute_worker_ticket_test_operation( + State(api), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + ticket_check_headers(&source), + Json(TicketBackendOperation::EditItem { + id: ticket_id.into(), + edit: TicketItemEdit { + body: Some(MarkdownText::new(body)), + ..Default::default() + }, + }), + )) + }) + }; + let (result_a, result_b) = tokio::join!( + run_edit(source_a.clone(), "Exact body from worker A."), + run_edit(source_b.clone(), "Exact body from worker B.") + ); + let Json(TicketBackendOperationResult::Ticket(ticket_a)) = result_a.unwrap().unwrap() + else { + panic!("unexpected first edit result"); + }; + let Json(TicketBackendOperationResult::Ticket(ticket_b)) = result_b.unwrap().unwrap() + else { + panic!("unexpected second edit result"); + }; + let snapshots = [ + (ticket_a, source_a, "Exact body from worker A."), + (ticket_b, source_b, "Exact body from worker B."), + ]; + for (ticket, source, expected_body) in snapshots { + let revision = ticket_item_checker::item_revision(&ticket); + let job = api + .store + .get_backend_job( + TEST_WORKSPACE_ID, + &format!("ticket-item-check:{}:{revision}", ticket.meta.id), + ) + .unwrap() + .expect("each successful edit has its own checker Job"); + let input = ticket_item_checker::parse_input(&job.request.input).unwrap(); + assert_eq!(input.revision, revision); + assert_eq!(input.body, expected_body); + assert_eq!(job.request.source_worker.as_ref(), Some(&source)); + assert_eq!(job.request.notification_target.as_ref(), Some(&source)); + } + } + #[tokio::test] async fn ticket_checker_suppresses_stale_and_clean_results_and_diagnoses_invalid_output() { let temp = tempfile::tempdir().unwrap(); @@ -35012,6 +35428,56 @@ mod tests { ); } + #[tokio::test] + async fn backend_job_recovery_drains_more_than_one_reserved_page() { + let temp = tempfile::tempdir().unwrap(); + let api = test_api(temp.path()).await; + api.runtime + .unregister_if_idle(EMBEDDED_WORKER_RUNTIME_ID, 10) + .unwrap(); + let mut jobs = Vec::new(); + for index in 0..(usize::from(ABSOLUTE_MAX_CONCURRENT_JOBS) + 6) { + let request = BackendJobRequest { + job_id: format!("restart-page-{index:03}"), + purpose: "restart_page_contract_check".to_string(), + input_revision: "1".to_string(), + input_ref: format!("fixture://restart-page/{index}"), + input: serde_json::json!({"index": index}), + instruction: "Return a structured fixture result.".to_string(), + profile: "builtin:backend-job".to_string(), + source_worker: None, + notification_target: None, + limits: crate::backend_job::BackendJobLimits { + max_concurrent_jobs: ABSOLUTE_MAX_CONCURRENT_JOBS, + ..Default::default() + }, + }; + api.store + .reserve_backend_job( + TEST_WORKSPACE_ID, + &request, + &format!("2026-09-01T00:{:02}:{:02}Z", index / 60, index % 60), + ) + .unwrap(); + jobs.push(request.job_id); + } + + api.recover_backend_jobs().unwrap(); + assert!( + api.store + .list_reserved_backend_job_attempts(TEST_WORKSPACE_ID, jobs.len()) + .unwrap() + .is_empty(), + "restart recovery must continue beyond its first bounded page when dispatches terminalize" + ); + assert!(jobs.iter().all(|job_id| { + api.store + .get_backend_job(TEST_WORKSPACE_ID, job_id) + .unwrap() + .is_some_and(|job| job.state == BackendJobState::Failed) + })); + } + #[tokio::test] async fn backend_job_recovery_dispatches_reserved_attempt_without_duplicate_worker() { let temp = tempfile::tempdir().unwrap(); @@ -35098,6 +35564,16 @@ mod tests { updated_at: "2026-09-01T00:00:00Z".to_string(), }) .unwrap(); + let (_, claimed) = api + .store + .claim_backend_job_attempt_dispatch( + TEST_WORKSPACE_ID, + &request.job_id, + &reserved.attempt.attempt_id, + "2026-09-01T00:00:00Z", + ) + .unwrap(); + assert!(claimed); api.store .bind_backend_job_attempt_worker( TEST_WORKSPACE_ID, diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index dd6b3d720..cea5ba56d 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -27,7 +27,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 71; +const LATEST_SCHEMA_VERSION: i64 = 72; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -64,6 +64,8 @@ const BACKEND_JOB_RUNNER_MIGRATION_NAME: &str = "Backend-owned Runtime Worker jobs and structured results"; const BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME: &str = "atomic Backend Job notification delivery claims"; +const BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME: &str = + "durable bounded Backend Job dispatch queue"; const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7; const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline"; const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#" @@ -216,6 +218,11 @@ const MIGRATIONS: &[Migration] = &[ name: BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME, apply: migrate_backend_job_delivery_claims_v70_to_v71, }, + Migration { + version: 72, + name: BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME, + apply: migrate_backend_job_dispatch_queue_v71_to_v72, + }, ]; #[derive(Clone, Copy)] @@ -1489,6 +1496,18 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { workspace_id: &str, limit: usize, ) -> Result>; + fn claim_backend_job_attempt_dispatch( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + now: &str, + ) -> Result<(BackendJobReservation, bool)>; + fn list_reserved_backend_job_attempts( + &self, + workspace_id: &str, + limit: usize, + ) -> Result>; fn bind_backend_job_attempt_worker( &self, workspace_id: &str, @@ -5886,7 +5905,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ))); } } - ensure_backend_job_capacity(&tx, workspace_id, request.limits.max_concurrent_jobs)?; let current_attempt = 1_u8; let attempt_id = attempt_id(&request.job_id, current_attempt); tx.execute( @@ -5930,7 +5948,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore { "Backend Job `{job_id}` reached its attempt limit" ))); } - ensure_backend_job_capacity(&tx, workspace_id, job.request.limits.max_concurrent_jobs)?; let next = job.current_attempt + 1; let next_id = attempt_id(job_id, next); let deadline = backend_job_deadline(now, job.request.limits.timeout_seconds)?; @@ -5975,17 +5992,88 @@ impl ControlPlaneStore for SqliteWorkspaceStore { &self, workspace_id: &str, limit: usize, + ) -> Result> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + // A dispatch claim has no externally meaningful identity until the + // deterministic Worker allocation is bound. Startup can therefore + // replay an interrupted claim through the same allocation key. + tx.execute( + "UPDATE backend_job_attempts SET state = 'reserved' WHERE workspace_id = ?1 AND state = 'dispatching' AND worker_id IS NULL", + params![workspace_id], + )?; + let mut statement = tx.prepare( + "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('reserved', 'dispatched') ORDER BY created_at LIMIT ?2", + )?; + let identities = statement.query_map(params![workspace_id, limit as i64], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + })?.collect::>>()?; + drop(statement); + let attempts = identities.into_iter().map(|(job_id, attempt_id)| { + read_backend_job_attempt(&tx, workspace_id, &job_id, &attempt_id)? + .ok_or_else(|| Error::Store("recoverable Backend Job attempt disappeared".to_string())) + }).collect::>>()?; + tx.commit()?; + Ok(attempts) + }) + } + + fn claim_backend_job_attempt_dispatch( + &self, + workspace_id: &str, + job_id: &str, + attempt_id: &str, + now: &str, + ) -> Result<(BackendJobReservation, bool)> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let job = read_backend_job(&tx, workspace_id, job_id)? + .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job `{job_id}`")))?; + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| Error::InvalidInput(format!("unknown Backend Job attempt `{attempt_id}`")))?; + if attempt.state != BackendJobAttemptState::Reserved { + tx.commit()?; + return Ok((BackendJobReservation { job, attempt, replayed: true }, false)); + } + let active: i64 = tx.query_row( + "SELECT COUNT(*) FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('dispatching', 'dispatched')", + params![workspace_id], + |row| row.get(0), + )?; + if active >= i64::from(job.request.limits.max_concurrent_jobs) { + tx.commit()?; + return Ok((BackendJobReservation { job, attempt, replayed: true }, false)); + } + let deadline = backend_job_deadline(now, job.request.limits.timeout_seconds)?; + let changed = tx.execute( + "UPDATE backend_job_attempts SET state = 'dispatching', deadline_at = ?4, updated_at = ?5 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'reserved' AND worker_id IS NULL", + params![workspace_id, job_id, attempt_id, deadline, now], + )?; + if changed != 1 { + return Err(Error::Store("Backend Job dispatch claim compare-and-set failed".to_string())); + } + let attempt = read_backend_job_attempt(&tx, workspace_id, job_id, attempt_id)? + .ok_or_else(|| Error::Store("claimed Backend Job attempt disappeared".to_string()))?; + tx.commit()?; + Ok((BackendJobReservation { job, attempt, replayed: false }, true)) + }) + } + + fn list_reserved_backend_job_attempts( + &self, + workspace_id: &str, + limit: usize, ) -> Result> { self.with_conn(|conn| { let mut statement = conn.prepare( - "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('reserved', 'dispatched') ORDER BY created_at LIMIT ?2", + "SELECT job_id, attempt_id FROM backend_job_attempts WHERE workspace_id = ?1 AND state = 'reserved' ORDER BY created_at, attempt_id LIMIT ?2", )?; let identities = statement.query_map(params![workspace_id, limit as i64], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) })?.collect::>>()?; identities.into_iter().map(|(job_id, attempt_id)| { read_backend_job_attempt(conn, workspace_id, &job_id, &attempt_id)? - .ok_or_else(|| Error::Store("recoverable Backend Job attempt disappeared".to_string())) + .ok_or_else(|| Error::Store("reserved Backend Job attempt disappeared".to_string())) }).collect() }) } @@ -6031,7 +6119,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { )); } let changed = tx.execute( - "UPDATE backend_job_attempts SET state = 'dispatched', runtime_id = ?4, worker_id = ?5, runtime_run_id = ?6, dispatched_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'reserved' AND worker_id IS NULL", + "UPDATE backend_job_attempts SET state = 'dispatched', runtime_id = ?4, worker_id = ?5, runtime_run_id = ?6, dispatched_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state = 'dispatching' AND worker_id IS NULL", params![workspace_id, job_id, attempt_id, worker.runtime_id, worker.worker_id, runtime_run_id, now], )?; if changed != 1 { @@ -6211,7 +6299,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { return Ok(attempt); } tx.execute( - "UPDATE backend_job_attempts SET state = ?4, failure_category = ?5, failure_detail = ?6, completed_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state IN ('reserved', 'dispatched')", + "UPDATE backend_job_attempts SET state = ?4, failure_category = ?5, failure_detail = ?6, completed_at = ?7, updated_at = ?7 WHERE workspace_id = ?1 AND job_id = ?2 AND attempt_id = ?3 AND state IN ('reserved', 'dispatching', 'dispatched')", params![workspace_id, job_id, attempt_id, state.as_str(), failure_category, detail, now], )?; if attempt.attempt == job.current_attempt && job.state == BackendJobState::Pending { @@ -9914,24 +10002,6 @@ fn backend_job_deadline(now: &str, timeout_seconds: u32) -> Result { Ok((now + chrono::Duration::seconds(i64::from(timeout_seconds))).to_rfc3339()) } -fn ensure_backend_job_capacity( - conn: &Connection, - workspace_id: &str, - max_concurrent: u16, -) -> Result<()> { - let active: i64 = conn.query_row( - "SELECT COUNT(*) FROM backend_job_attempts WHERE workspace_id = ?1 AND state IN ('reserved', 'dispatched')", - params![workspace_id], - |row| row.get(0), - )?; - if active >= i64::from(max_concurrent) { - return Err(Error::InvalidInput(format!( - "Backend Job concurrency limit {max_concurrent} is reached" - ))); - } - Ok(()) -} - fn read_backend_job( conn: &Connection, workspace_id: &str, @@ -13297,6 +13367,83 @@ fn migrate_backend_job_delivery_claims_v70_to_v71(conn: &Connection) -> Result<( Ok(()) } +fn migrate_backend_job_dispatch_queue_v71_to_v72(conn: &Connection) -> Result<()> { + let current = current_schema_version(conn)?; + if current != 71 { + return Err(Error::Store(format!( + "expected schema version 71 before {BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME} migration, found {current}" + ))); + } + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + tx.execute_batch( + r#" + DROP INDEX backend_job_deliveries_pending; + DROP INDEX backend_job_attempts_recovery; + ALTER TABLE backend_job_deliveries RENAME TO backend_job_deliveries_v71; + ALTER TABLE backend_job_attempts RENAME TO backend_job_attempts_v71; + CREATE TABLE backend_job_attempts ( + workspace_id TEXT NOT NULL, + job_id TEXT NOT NULL, + attempt_id TEXT NOT NULL, + attempt INTEGER NOT NULL CHECK (attempt > 0 AND attempt <= 3), + input_revision TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('reserved', 'dispatching', 'dispatched', 'completed', 'failed', 'unknown')), + runtime_id TEXT, + worker_id TEXT, + runtime_run_id TEXT, + dispatched_at TEXT, + deadline_at TEXT NOT NULL, + result_json TEXT, + result_digest TEXT, + failure_category TEXT, + failure_detail TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT, + PRIMARY KEY (workspace_id, job_id, attempt_id), + UNIQUE (workspace_id, job_id, attempt), + UNIQUE (workspace_id, runtime_id, worker_id), + FOREIGN KEY (workspace_id, job_id) + REFERENCES backend_jobs(workspace_id, job_id) ON DELETE CASCADE, + CHECK ((runtime_id IS NULL) = (worker_id IS NULL)), + CHECK ((result_json IS NULL) = (result_digest IS NULL)), + CHECK ((state = 'completed') = (result_json IS NOT NULL)) + ); + INSERT INTO backend_job_attempts SELECT * FROM backend_job_attempts_v71; + CREATE INDEX backend_job_attempts_recovery + ON backend_job_attempts(workspace_id, state, deadline_at); + CREATE TABLE backend_job_deliveries ( + workspace_id TEXT NOT NULL, + delivery_id TEXT NOT NULL, + job_id TEXT NOT NULL, + attempt_id TEXT NOT NULL, + target_runtime_id TEXT NOT NULL, + target_worker_id TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'sending', 'completed', 'failed', 'unknown')), + failure_category TEXT, + failure_detail TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + delivered_at TEXT, + PRIMARY KEY (workspace_id, delivery_id), + FOREIGN KEY (workspace_id, job_id, attempt_id) + REFERENCES backend_job_attempts(workspace_id, job_id, attempt_id) ON DELETE CASCADE + ); + INSERT INTO backend_job_deliveries SELECT * FROM backend_job_deliveries_v71; + CREATE INDEX backend_job_deliveries_pending + ON backend_job_deliveries(workspace_id, state, updated_at); + DROP TABLE backend_job_deliveries_v71; + DROP TABLE backend_job_attempts_v71; + "#, + )?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![72_i64, BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?; conn.execute_batch(include_str!("latest_schema.sql"))?; @@ -15476,6 +15623,10 @@ mod tests { version: 71, name: BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 72, + name: BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME.to_string(), + }, ] ); @@ -15551,6 +15702,7 @@ mod tests { 71, BACKEND_JOB_DELIVERY_CLAIM_MIGRATION_NAME.to_string(), ), + (72, BACKEND_JOB_DISPATCH_QUEUE_MIGRATION_NAME.to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -15836,7 +15988,7 @@ mod tests { .map(|migration| migration.version) .collect::>(), vec![ - 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71 + 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72 ] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); @@ -15845,7 +15997,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 22); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 23); assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap()); assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap()); assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap()); @@ -19528,6 +19680,15 @@ INSERT INTO worker_registry ( .unwrap(); assert!(replay.replayed); assert_eq!(replay.attempt.attempt_id, first.attempt.attempt_id); + let (_, claimed) = store + .claim_backend_job_attempt_dispatch( + workspace_id, + &request.job_id, + &first.attempt.attempt_id, + "2026-09-01T00:00:02Z", + ) + .unwrap(); + assert!(claimed); store .bind_backend_job_attempt_worker( workspace_id, @@ -19762,12 +19923,101 @@ INSERT INTO worker_registry ( "Worker removal must not discard the durable Job result" ); + // Dispatch admission is a transactional claim: concurrent claimers may + // leave durable work queued, but they cannot exceed the configured cap. + let queued = (0..6) + .map(|index| { + let mut queued_request = request.clone(); + queued_request.job_id = format!("concurrent-claim-{index}"); + queued_request.source_worker = None; + queued_request.notification_target = None; + queued_request.limits.max_concurrent_jobs = 4; + store + .reserve_backend_job( + workspace_id, + &queued_request, + &format!("2026-09-01T00:00:{:02}Z", 20 + index), + ) + .unwrap() + }) + .collect::>(); + let barrier = Arc::new(std::sync::Barrier::new(queued.len())); + let claims = std::thread::scope(|scope| { + let handles = queued + .iter() + .map(|reservation| { + let store = store.clone(); + let job_id = reservation.job.request.job_id.clone(); + let attempt_id = reservation.attempt.attempt_id.clone(); + let barrier = barrier.clone(); + scope.spawn(move || { + barrier.wait(); + store + .claim_backend_job_attempt_dispatch( + workspace_id, + &job_id, + &attempt_id, + "2026-09-01T00:00:30Z", + ) + .unwrap() + }) + }) + .collect::>(); + handles + .into_iter() + .map(|handle| handle.join().unwrap()) + .collect::>() + }); + assert_eq!(claims.iter().filter(|(_, claimed)| *claimed).count(), 4); + assert_eq!( + claims + .iter() + .filter(|(reservation, _)| { + reservation.attempt.state == BackendJobAttemptState::Dispatching + }) + .count(), + 4 + ); + assert_eq!( + store + .list_reserved_backend_job_attempts(workspace_id, 10) + .unwrap() + .len(), + 2 + ); + for reservation in claims + .iter() + .filter(|(_, claimed)| *claimed) + .map(|(reservation, _)| reservation) + { + store + .finish_backend_job_attempt( + workspace_id, + &reservation.job.request.job_id, + &reservation.attempt.attempt_id, + BackendJobAttemptState::Failed, + "test_cleanup", + "release test dispatch slot", + "2026-09-01T00:00:31Z", + ) + .unwrap(); + } + let mut retry_request = request; retry_request.job_id = "bounded-retry".to_string(); retry_request.notification_target = None; let failed = store .reserve_backend_job(workspace_id, &retry_request, "2026-09-01T00:01:00Z") .unwrap(); + let (_, claimed) = store + .claim_backend_job_attempt_dispatch( + workspace_id, + &retry_request.job_id, + &failed.attempt.attempt_id, + "2026-09-01T00:01:01Z", + ) + .unwrap(); + assert!(claimed); store .bind_backend_job_attempt_worker( workspace_id, diff --git a/docs/design/ticket-item-checker.md b/docs/design/ticket-item-checker.md index bea013be3..ebab424ff 100644 --- a/docs/design/ticket-item-checker.md +++ b/docs/design/ticket-item-checker.md @@ -22,7 +22,7 @@ The revision is the latest `create` or `item_edit` event ID, matching `TicketDet ## Execution and bounds -The trigger reserves durable intent synchronously after save, then dispatches without waiting for model completion. It uses T-674's Backend-owned Job runner and an ordinary registered embedded Runtime Worker with `builtin:backend-job`. That profile selects the existing lightweight `codex-oauth/gpt-5.6-luna` model through normal model configuration and authentication. The checker has only `SubmitBackendJobResult`; it has no Ticket mutation, Queue, Worker management, conversation discovery, Web, Memory, or Workdir tools. +The trigger reserves durable intent synchronously after save, then dispatches without waiting for model completion. Concurrency pressure never rejects that intent: attempts above the four-Job cap remain durably `reserved`, and terminal attempt transitions plus restart recovery drain queued attempts through an atomic `reserved` → `dispatching` slot claim. It uses T-674's Backend-owned Job runner and an ordinary registered embedded Runtime Worker with `builtin:backend-job`. That profile selects the existing lightweight `codex-oauth/gpt-5.6-luna` model through normal model configuration and authentication. The checker has only `SubmitBackendJobResult`; it has no Ticket mutation, Queue, Worker management, conversation discovery, Web, Memory, or Workdir tools. Each check is bounded to 4 concurrent Backend Jobs, 45 seconds, one attempt, an 8 KiB result, at most 8 findings, bounded finding fields, and the Backend Job runner's fixed input/instruction/delivery limits. The deterministic Ticket/revision Job ID suppresses duplicate execution. Generic Job and attempt records retain completed results and categorized dispatch, timeout, invalid-output, and other failures.