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

fix: queue ticket checks with exact snapshots

Browse Source
This commit is contained in:
Hare
2026-10-01 13:52:57 +09:00
parent ab8b4ab1ef
commit d101b4bacb
7 changed files with 799 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/backend_job.rs
+3
View File
@@ -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<Self> {
match value {
"reserved" => Ok(Self::Reserved),
"dispatching" => Ok(Self::Dispatching),
"dispatched" => Ok(Self::Dispatched),
"completed" => Ok(Self::Completed),
"failed" => Ok(Self::Failed),
crates/workspace-server/src/latest_schema.sql
+1 -1
View File
@@ -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,
crates/workspace-server/src/server.rs
+498 -22
View File
@@ -1394,6 +1394,7 @@ pub struct WorkspaceApi {
companion: Arc<CompanionConsole>,
orchestrator_spawn_lock: Arc<std::sync::Mutex<()>>,
orchestrator_attention_fingerprint: Arc<Mutex<Option<String>>>,
backend_job_drain_requests: Arc<AtomicU64>,
observation_proxy: BackendObservationProxy,
runtime_subscription_broker: RuntimeSubscriptionBroker,
pub(crate) worker_projection: Arc<crate::worker_projection::WorkerProjectionService>,
@@ -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::<Vec<_>>();
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,
crates/workspace-server/src/store.rs
+277 -27
View File
@@ -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<Vec<BackendJobAttemptRecord>>;
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<Vec<BackendJobAttemptRecord>>;
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<Vec<BackendJobAttemptRecord>> {
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::<rusqlite::Result<Vec<_>>>()?;
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::<Result<Vec<_>>>()?;
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<Vec<BackendJobAttemptRecord>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<String> {
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<_>>(),
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::<Vec<_>>();
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::<Vec<_>>();
handles
.into_iter()
.map(|handle| handle.join().unwrap())
.collect::<Vec<_>>()
});
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,