From e71177900e03a053c5cd25329d656afb757312ef Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 15 Oct 2026 18:12:07 +0900 Subject: [PATCH] feat: route backend job results through common notifications --- .../src/assignment_tracking_tests.rs | 3 +- .../src/job_notification_routing.rs | 443 ++++++++++++ .../src/job_notification_tests.rs | 639 ++++++++++++++++++ .../src/notification_delivery.rs | 209 +++++- .../src/notification_routing.rs | 14 +- .../src/notification_routing_tests.rs | 1 + crates/workspace-server/src/server.rs | 285 ++------ .../src/server_assignment_delivery_tests.rs | 2 + .../src/server_job_notification_tests.rs | 185 +++++ crates/workspace-server/src/store.rs | 140 +++- .../src/ticket_item_checker.rs | 45 +- .../src/workspace_deletion.rs | 4 + 12 files changed, 1700 insertions(+), 270 deletions(-) create mode 100644 crates/workspace-server/src/job_notification_routing.rs create mode 100644 crates/workspace-server/src/job_notification_tests.rs create mode 100644 crates/workspace-server/src/server_job_notification_tests.rs diff --git a/crates/workspace-server/src/assignment_tracking_tests.rs b/crates/workspace-server/src/assignment_tracking_tests.rs index b9392b4bd..160d25603 100644 --- a/crates/workspace-server/src/assignment_tracking_tests.rs +++ b/crates/workspace-server/src/assignment_tracking_tests.rs @@ -469,7 +469,8 @@ fn tracking_store_real_schema91_to92_retains_queue_cases_principals_audit_receip ( 95, "Common notification routing and durable Worker delivery".into() - ) + ), + (96,"Backend Job common notification routing".into()) ] ); for (table, rows) in retained.iter().zip(original.iter()) { diff --git a/crates/workspace-server/src/job_notification_routing.rs b/crates/workspace-server/src/job_notification_routing.rs new file mode 100644 index 000000000..60ceadd46 --- /dev/null +++ b/crates/workspace-server/src/job_notification_routing.rs @@ -0,0 +1,443 @@ +//! Backend Job adapter for the common notification ledger. Results, not Worker +//! lifecycle, are the durable source. Fixed Worker recipients never follow leases. +use super::*; +use crate::backend_job::{ + BackendJobAttemptState, BackendJobDeliveryRecord, BackendJobDeliveryState, + BackendJobResultAcceptance, BackendJobState, +}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobNotificationSource { + pub job_id: String, + pub attempt_id: String, + pub delivery_id: String, +} + +/// Domain facts, not a second notification policy. Rendering is performed by the +/// domain adapter; only this routing module chooses deliver/suppress/defer. +#[derive(Debug, Clone)] +pub enum JobNotificationFacts { + Result(String), + NoFindings, + StaleContent, + Failed(NotificationFailureCode), +} + +pub(in crate::store) fn install(conn: &Connection) -> Result<()> { + // Generalize the existing source parent without changing any sequence, request, + // or receipt. The migration caller disables FK enforcement for this rebuild. + let triggers = conn + .prepare("SELECT name,sql FROM sqlite_schema WHERE type='trigger'")? + .query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))? + .collect::>>()?; + let indexes = conn.prepare("SELECT sql FROM sqlite_schema WHERE type='index' AND tbl_name='notification_sources' AND sql IS NOT NULL")? + .query_map([], |r| r.get::<_,String>(0))?.collect::>>()?; + for (name, _) in &triggers { + conn.execute_batch(&format!("DROP TRIGGER \"{}\"", name.replace('"', "\"\"")))?; + } + conn.execute_batch( + "CREATE TABLE notification_sources_job_upgrade ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + workspace_id TEXT NOT NULL, ticket_id TEXT, input_event_id TEXT NOT NULL, + role TEXT NOT NULL, assignment_id TEXT NOT NULL, kind TEXT NOT NULL, + actor TEXT, event_index INTEGER, facts_json TEXT NOT NULL CHECK(json_valid(facts_json)), + CHECK(ticket_id IS NOT NULL OR role='backend_job'), + UNIQUE(workspace_id,ticket_id,input_event_id,role,assignment_id), + FOREIGN KEY(workspace_id,ticket_id) REFERENCES typed_tickets(workspace_id,ticket_id)); + INSERT INTO notification_sources_job_upgrade SELECT * FROM notification_sources; + DROP TABLE notification_sources; + ALTER TABLE notification_sources_job_upgrade RENAME TO notification_sources;", + )?; + for sql in indexes { + conn.execute_batch(&sql)?; + } + for (_, sql) in triggers { + conn.execute_batch(&sql)?; + } + conn.execute_batch("CREATE TRIGGER notification_job_source_guard BEFORE INSERT ON notification_sources WHEN NEW.role='backend_job' BEGIN + SELECT CASE WHEN NOT EXISTS(SELECT 1 FROM backend_jobs j JOIN backend_job_attempts a USING(workspace_id,job_id) + WHERE j.workspace_id=NEW.workspace_id AND j.job_id=json_extract(NEW.facts_json,'$.backend_job.job_id') AND a.attempt_id=json_extract(NEW.facts_json,'$.backend_job.attempt_id') + AND j.state='completed' AND a.state='completed' AND a.attempt=j.current_attempt AND j.result_digest=json_extract(NEW.facts_json,'$.result_digest')) + THEN RAISE(ABORT,'Job notification requires a committed current result') END; + END; + CREATE UNIQUE INDEX notification_job_identity ON notification_sources(workspace_id,input_event_id) WHERE role='backend_job'; + CREATE TABLE notification_job_targets ( + workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL, target_json TEXT NOT NULL CHECK(json_valid(target_json)), + PRIMARY KEY(workspace_id,delivery_id), + FOREIGN KEY(workspace_id,delivery_id) REFERENCES notification_requests(workspace_id,delivery_id)); + CREATE TRIGGER notification_job_targets_no_update BEFORE UPDATE ON notification_job_targets BEGIN SELECT RAISE(ABORT,'Job notification target is immutable'); END; + CREATE TRIGGER notification_job_targets_no_delete BEFORE DELETE ON notification_job_targets BEGIN SELECT RAISE(ABORT,'Job notification target is immutable'); END; + CREATE TABLE notification_job_legacy_receipts ( + workspace_id TEXT NOT NULL, delivery_id TEXT NOT NULL, record_json TEXT NOT NULL CHECK(json_valid(record_json)), + runtime_accepted INTEGER NOT NULL CHECK(runtime_accepted IN (0,1)), + PRIMARY KEY(workspace_id,delivery_id), + FOREIGN KEY(workspace_id,delivery_id) REFERENCES notification_decisions(workspace_id,delivery_id)); + CREATE TRIGGER notification_job_legacy_no_update BEFORE UPDATE ON notification_job_legacy_receipts BEGIN SELECT RAISE(ABORT,'Legacy notification evidence is immutable'); END; + CREATE TRIGGER notification_job_legacy_no_delete BEFORE DELETE ON notification_job_legacy_receipts BEGIN SELECT RAISE(ABORT,'Legacy notification evidence is immutable'); END;")?; + // Includes no-target results and the crash gap before the former reservation. + let identities = conn.prepare("SELECT a.workspace_id,a.job_id,a.attempt_id FROM backend_job_attempts a JOIN backend_jobs j USING(workspace_id,job_id) WHERE a.state='completed' AND j.state='completed' AND a.attempt=j.current_attempt")? + .query_map([], |r| Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,String>(2)?)))? + .collect::>>()?; + for (workspace, job, attempt) in identities { + capture(conn, &workspace, &job, &attempt)?; + import_legacy(conn, &workspace, &job, &attempt)?; + } + Ok(()) +} + +pub(in crate::store) fn capture( + conn: &Connection, + workspace: &str, + job: &str, + attempt: &str, +) -> Result<()> { + let id = retained_delivery_id(conn, workspace, job, attempt)?; + let source = JobNotificationSource { + job_id: job.into(), + attempt_id: attempt.into(), + delivery_id: id.clone(), + }; + let job = read_backend_job(conn, workspace, job)? + .ok_or_else(|| Error::Store("Job notification source missing".into()))?; + let target = &job.request.notification_target; + let facts = serde_json::json!({"principal_kind":if target.is_some(){Some("worker")}else{None}, "principal_id":null,"runtime_id":target.as_ref().map(|w|&w.runtime_id),"worker_id":target.as_ref().map(|w|&w.worker_id),"status":null,"case_state":"open","released":false,"backend_job":source,"recipient_kind":"fixed_worker","job_purpose":job.request.purpose,"result_digest":job.result_digest}); + conn.execute("INSERT OR IGNORE INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) VALUES(?1,NULL,?2,'backend_job','','backend_job_result',?3)",params![workspace,id,json(&facts)?])?; + Ok(()) +} + +pub(in crate::store) fn pending( + store: &SqliteWorkspaceStore, + workspace: &str, + limit: usize, +) -> Result> { + store.with_conn_mut(|conn| { + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let rows = tx.prepare("SELECT s.facts_json,s.sequence FROM notification_sources s LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND s.role='backend_job' AND (d.source_sequence IS NULL OR d.disposition='deliver') AND NOT EXISTS(SELECT 1 FROM notification_acknowledgements a WHERE a.workspace_id=?1 AND a.delivery_id=s.input_event_id) AND NOT EXISTS(SELECT 1 FROM notification_job_legacy_receipts l WHERE l.workspace_id=?1 AND l.delivery_id=s.input_event_id AND l.runtime_accepted=1) ORDER BY COALESCE(d.last_offered,0),s.sequence LIMIT ?2")? + .query_map(params![workspace,limit.clamp(1,100) as i64],|r|Ok((r.get::<_,String>(0)?,r.get::<_,i64>(1)?)))? + .collect::>>()?; + let tick:i64=tx.query_row("SELECT COALESCE(MAX(last_offered),0)+1 FROM notification_decisions WHERE workspace_id=?1",[workspace],|r|r.get(0))?; + let mut sources=Vec::new(); + for (facts,sequence) in rows { + let facts:serde_json::Value=decode(&facts)?; + sources.push(serde_json::from_value(facts["backend_job"].clone()).map_err(|e|Error::Store(e.to_string()))?); + tx.execute("UPDATE notification_decisions SET last_offered=?2 WHERE source_sequence=?1",params![sequence,tick])?; + } + tx.commit()?; Ok(sources) + }) +} + +fn policy(target: bool, facts: &JobNotificationFacts) -> (&'static str, NotificationReason) { + if !target { + return ("suppress", NotificationReason::NoRecipient); + } + match facts { + JobNotificationFacts::Result(_) => ("deliver", NotificationReason::Result), + JobNotificationFacts::NoFindings => ("suppress", NotificationReason::NoFindings), + JobNotificationFacts::StaleContent => ("suppress", NotificationReason::StaleContent), + JobNotificationFacts::Failed(_) => ("defer", NotificationReason::SourceUnavailable), + } +} + +pub(in crate::store) fn route( + store: &SqliteWorkspaceStore, + workspace: &str, + source: &JobNotificationSource, + facts: &JobNotificationFacts, +) -> Result { + let invalid; + let facts = match facts { + JobNotificationFacts::Result(content) + if content.len() > crate::backend_job::MAX_JOB_DELIVERY_BYTES => + { + invalid = JobNotificationFacts::Failed(NotificationFailureCode::ContentTooLarge); + &invalid + } + JobNotificationFacts::Result(content) if content.is_empty() || content.contains('\0') => { + invalid = JobNotificationFacts::Failed(NotificationFailureCode::RenderingFailed); + &invalid + } + other => other, + }; + store.with_conn_mut(|conn| { + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let job=read_backend_job(&tx,workspace,&source.job_id)?.ok_or_else(||Error::InvalidInput("Job notification source disappeared".into()))?; + let attempt=read_backend_job_attempt(&tx,workspace,&source.job_id,&source.attempt_id)?.ok_or_else(||Error::InvalidInput("Job notification attempt disappeared".into()))?; + if job.state!=BackendJobState::Completed || attempt.state!=BackendJobAttemptState::Completed || attempt.attempt!=job.current_attempt || source.delivery_id!=retained_delivery_id(&tx,workspace,&source.job_id,&source.attempt_id)? { + return Err(Error::InvalidInput("Job notification requires its current committed result".into())); + } + capture(&tx,workspace,&source.job_id,&source.attempt_id)?; + import_legacy(&tx, workspace, &source.job_id, &source.attempt_id)?; + let existing:Option=tx.query_row("SELECT disposition FROM notification_decisions WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,source.delivery_id],|r|r.get(0)).optional()?; + if let Some(disposition)=existing { + let ack:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM notification_acknowledgements WHERE workspace_id=?1 AND delivery_id=?2) OR EXISTS(SELECT 1 FROM notification_job_legacy_receipts WHERE workspace_id=?1 AND delivery_id=?2 AND runtime_accepted=1)",params![workspace,source.delivery_id],|r|r.get(0))?; + tx.commit()?; return Ok(disposition=="deliver" && !ack); + } + let (disposition,reason)=policy(job.request.notification_target.is_some(),facts); + save_decision(&tx, workspace, &source.delivery_id, disposition, reason)?; + if let JobNotificationFacts::Failed(failure)=facts { save_failure(&tx,workspace,&source.delivery_id,*failure)?; } + // Pin bytes with the decision, before any network call. A prepared body is + // never rerendered when content changes or an acknowledgement is lost. + if disposition=="deliver" { + if let (Some(target),JobNotificationFacts::Result(content))=(&job.request.notification_target,facts) { + tx.execute("INSERT OR IGNORE INTO notification_requests VALUES(?1,?2,?3,?4)",params![workspace,source.delivery_id,content,json(target)?])?; + } + } + tx.commit()?; Ok(disposition=="deliver") + }) +} + +pub(in crate::store) fn request( + store: &SqliteWorkspaceStore, + workspace: &str, + id: &str, +) -> Result> { + store.with_conn(|conn| { + let row:Option<(String,String)>=conn.query_row("SELECT content,target_json FROM notification_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?; + row.map(|(content,target)|Ok((content,decode(&target)?))).transpose() + }) +} + +pub(in crate::store) fn ack( + store: &SqliteWorkspaceStore, + workspace: &str, + id: &str, + target: &RuntimeWorkerRef, +) -> Result<()> { + store.with_conn_mut(|conn| { + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let saved:String=tx.query_row("SELECT target_json FROM notification_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0))?; + if saved!=json(target)? {return Err(Error::InvalidInput("Job notification target changed".into()));} + let authorized:String=tx.query_row("SELECT target_json FROM notification_job_targets WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0))?; + let authorized_target:TicketWorkerNotificationTarget=decode(&authorized)?; + if authorized_target.worker!=*target {return Err(Error::InvalidInput("Job acknowledgement target differs from authorized send".into()));} + tx.execute("INSERT OR IGNORE INTO notification_acknowledgements VALUES(?1,?2,?3)",params![workspace,id,authorized])?; + tx.commit()?; Ok(()) + }) +} + +fn save_decision( + conn: &Connection, + workspace: &str, + id: &str, + disposition: &str, + reason: NotificationReason, +) -> Result<()> { + conn.execute("INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose) SELECT sequence,workspace_id,?2,?3,?4,'result' FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2 AND role='backend_job'",params![workspace,id,disposition,code(reason)?])?; + Ok(()) +} + +// Old rows did not store request bytes. Only the original deterministic renderers +// may reconstruct an interrupted request; no current Ticket contents participate. +// Accepted legacy rows keep their actual evidence, never a fabricated new request. +fn import_legacy(conn: &Connection, workspace: &str, job_id: &str, attempt_id: &str) -> Result<()> { + let id = retained_delivery_id(conn, workspace, job_id, attempt_id)?; + let exists:bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM notification_decisions WHERE workspace_id=?1 AND delivery_id=?2)",params![workspace,id],|r|r.get(0))?; + if exists { + return Ok(()); + } + let Some(old) = read_backend_job_delivery(conn, workspace, &id)? else { + return Ok(()); + }; + let accepted = + old.state == BackendJobDeliveryState::Completed && old.failure_category.is_none(); + let count:i64=conn.query_row("SELECT COUNT(*) FROM backend_job_deliveries WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3",params![workspace,job_id,attempt_id],|r|r.get(0))?; + let mut content = None; + let mut failure = legacy_failure(&old); + let (disposition, reason) = match old.failure_category.as_deref() { + _ if accepted => ("deliver", NotificationReason::Result), + _ if count > 1 => { + failure = Some(NotificationFailureCode::PreparationFailed); + ("defer", NotificationReason::SourceUnavailable) + } + Some("notification_suppressed_no_findings") => ("suppress", NotificationReason::NoFindings), + Some("notification_suppressed_stale_ticket_content") => { + ("suppress", NotificationReason::StaleContent) + } + _ if old.state == BackendJobDeliveryState::Completed => { + ("defer", NotificationReason::SourceUnavailable) + } + Some( + "invalid_checker_input" + | "invalid_checker_result" + | "current_ticket_content_unavailable" + | "notification_too_large", + ) => ("defer", NotificationReason::SourceUnavailable), + _ => { + let job = read_backend_job(conn, workspace, job_id)? + .ok_or_else(|| Error::Store("Legacy Job result missing".into()))?; + let attempt = read_backend_job_attempt(conn, workspace, job_id, attempt_id)? + .ok_or_else(|| Error::Store("Legacy Job attempt missing".into()))?; + let acceptance = BackendJobResultAcceptance { + job, + attempt, + replayed: true, + }; + let facts = if acceptance.job.request.notification_target.as_ref() != Some(&old.target) + { + JobNotificationFacts::Failed(NotificationFailureCode::PreparationFailed) + } else { + legacy_result_facts(&acceptance) + }; + let decision = policy(acceptance.job.request.notification_target.is_some(), &facts); + match facts { + JobNotificationFacts::Result(body) => content = Some(body), + JobNotificationFacts::Failed(code) => failure = Some(code), + // A valid empty result could not have passed the old sender's + // findings gate even if it crashed before recording suppression. + JobNotificationFacts::NoFindings => failure = None, + JobNotificationFacts::StaleContent => {} + } + decision + } + }; + save_decision(conn, workspace, &id, disposition, reason)?; + conn.execute( + "INSERT INTO notification_job_legacy_receipts VALUES(?1,?2,?3,?4)", + params![workspace, id, json(&old)?, accepted], + )?; + if !accepted && disposition != "suppress" { + if let Some(failure) = failure { + save_failure(conn, workspace, &id, failure)?; + } + } + if disposition == "deliver" { + if let Some(content) = content { + conn.execute( + "INSERT INTO notification_requests VALUES(?1,?2,?3,?4)", + params![workspace, id, content, json(&old.target)?], + )?; + } + } + Ok(()) +} + +fn legacy_result_facts(acceptance: &BackendJobResultAcceptance) -> JobNotificationFacts { + let content = if acceptance.job.request.purpose == crate::ticket_item_checker::PURPOSE { + let Some(result) = &acceptance.job.result else { + return JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult); + }; + match crate::ticket_item_checker::validate_result(&acceptance.job.request.input, result) { + Ok(result) if result.findings.is_empty() => return JobNotificationFacts::NoFindings, + Ok(_) => {} + Err(_) => return JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult), + } + crate::ticket_item_checker::render_result(acceptance) + } else { + Ok(format!( + "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.", + acceptance.job.request.job_id, + acceptance.attempt.attempt_id, + acceptance.job.request.purpose, + acceptance + .job + .result_digest + .as_deref() + .unwrap_or("unavailable") + )) + }; + match content { + Ok(content) if content.len() > crate::backend_job::MAX_JOB_DELIVERY_BYTES => { + JobNotificationFacts::Failed(NotificationFailureCode::ContentTooLarge) + } + Ok(content) if !content.is_empty() && !content.contains('\0') => { + JobNotificationFacts::Result(content) + } + _ => JobNotificationFacts::Failed(NotificationFailureCode::RenderingFailed), + } +} + +fn legacy_failure(old: &BackendJobDeliveryRecord) -> Option { + use NotificationFailureCode as F; + match old.failure_category.as_deref() { + Some("notification_outcome_unknown") => Some(F::AcceptanceUnknown), + Some("notification_rejected") => Some(F::RuntimeRejected), + Some("notification_too_large") => Some(F::ContentTooLarge), + Some("invalid_checker_input" | "invalid_checker_result") => Some(F::InvalidResult), + Some("current_ticket_content_unavailable") => Some(F::SourceLookupFailed), + _ if matches!( + old.state, + BackendJobDeliveryState::Sending + | BackendJobDeliveryState::Unknown + | BackendJobDeliveryState::Completed + ) => + { + Some(F::AcceptanceUnknown) + } + _ => None, + } +} + +/// None authorizes this exact send target; a typed denial is not a DB failure. +pub(in crate::store) fn authorize( + store: &SqliteWorkspaceStore, + workspace: &str, + id: &str, + target: &TicketWorkerNotificationTarget, +) -> Result> { + store.with_conn_mut(|conn| { + let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let recipient:String=tx.query_row("SELECT target_json FROM notification_requests WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0))?; + if decode::(&recipient)?!=target.worker {return Ok(Some(NotificationFailureCode::PreparationFailed));} + let present:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM worker_registry w WHERE w.workspace_id=?1 AND w.runtime_id=?2 AND w.worker_id=?3 AND NOT EXISTS(SELECT 1 FROM worker_removal_operations r WHERE r.workspace_id=w.workspace_id AND r.runtime_id=w.runtime_id AND r.worker_id=w.worker_id AND r.state IN ('executing','failed','succeeded')))",params![workspace,target.worker.runtime_id,target.worker.worker_id],|r|r.get(0))?; + if !present {return Ok(Some(NotificationFailureCode::TargetUnavailable));} + let binding:Option<(String,Option)>=tx.query_row("SELECT binding_id,revoked_at FROM workspace_runtime_bindings WHERE workspace_id=?1 AND runtime_id=?2",params![workspace,target.worker.runtime_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?; + let Some((binding,revoked))=binding else {return Ok(Some(NotificationFailureCode::TargetUnavailable));}; + if revoked.is_some() || binding!=target.authentication_binding_id {return Ok(Some(NotificationFailureCode::BindingRevoked));} + let saved:Option=tx.query_row("SELECT target_json FROM notification_job_targets WHERE workspace_id=?1 AND delivery_id=?2",params![workspace,id],|r|r.get(0)).optional()?; + let serialized=json(target)?; + if saved.as_ref().is_some_and(|saved|*saved!=serialized) {return Ok(Some(NotificationFailureCode::BindingRevoked));} + tx.execute("INSERT OR IGNORE INTO notification_job_targets VALUES(?1,?2,?3)",params![workspace,id,serialized])?; + tx.commit()?; Ok(None) + }) +} + +pub(in crate::store) fn verify(conn: &Connection) -> Result<()> { + for name in [ + "notification_job_identity", + "notification_job_source_guard", + "notification_job_targets", + "notification_job_targets_no_update", + "notification_job_targets_no_delete", + "notification_job_legacy_receipts", + "notification_job_legacy_no_update", + "notification_job_legacy_no_delete", + ] { + let present: bool = conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name=?1)", + [name], + |r| r.get(0), + )?; + if !present { + return Err(Error::Store(format!( + "Job notification constraint {name} is missing" + ))); + } + } + Ok(()) +} + +fn retained_delivery_id( + conn: &Connection, + workspace: &str, + job: &str, + attempt: &str, +) -> Result { + // The production legacy sender used the derived ID, but retained receipts are + // authoritative even when imported history contains a non-derived identity. + // If history contains multiple identities, an actual ACK wins; otherwise the + // importer defers the ambiguity instead of manufacturing another send. + let saved:Option=conn.query_row("SELECT delivery_id FROM backend_job_deliveries WHERE workspace_id=?1 AND job_id=?2 AND attempt_id=?3 ORDER BY (state='completed' AND failure_category IS NULL) DESC,created_at,delivery_id LIMIT 1",params![workspace,job,attempt],|r|r.get(0)).optional()?; + Ok(saved.unwrap_or_else(|| crate::backend_job::delivery_id(job, attempt))) +} + +pub(in crate::store) fn source( + store: &SqliteWorkspaceStore, + workspace: &str, + job: &str, + attempt: &str, +) -> Result { + store.with_conn(|conn| { + let facts:String=conn.query_row("SELECT json_extract(facts_json,'$.backend_job') FROM notification_sources WHERE workspace_id=?1 AND role='backend_job' AND json_extract(facts_json,'$.backend_job.job_id')=?2 AND json_extract(facts_json,'$.backend_job.attempt_id')=?3",params![workspace,job,attempt],|r|r.get(0))?; + decode(&facts) + }) +} diff --git a/crates/workspace-server/src/job_notification_tests.rs b/crates/workspace-server/src/job_notification_tests.rs new file mode 100644 index 000000000..27cbc0779 --- /dev/null +++ b/crates/workspace-server/src/job_notification_tests.rs @@ -0,0 +1,639 @@ +// Real SQLite result transactions, migration, reopen and concurrent recovery. +fn job_notice_fixture( + store: &SqliteWorkspaceStore, + id: &str, + target: bool, +) -> (BackendJobResultAcceptance, JobNotificationSource) { + let mut request = resource_job_request(id, id, id); + request.purpose = "notification_test".into(); + request.grants = Default::default(); + request.profile = "builtin:backend-job".into(); + request.notification_target = target.then(|| RuntimeWorkerRef::new("r", "replacement")); + let worker = RuntimeWorkerRef::new("r", format!("job-worker-{id}")); + let mut registration = store + .get_worker_registry("w", &RuntimeWorkerRef::new("r", "replacement")) + .unwrap() + .unwrap(); + registration.worker = worker.clone(); + store.upsert_worker_registry(®istration).unwrap(); + let reservation = store + .reserve_backend_job("w", &request, RESOURCE_JOB_NOW) + .unwrap(); + store + .claim_backend_job_attempt_dispatch( + "w", + id, + &reservation.attempt.attempt_id, + RESOURCE_JOB_NOW, + ) + .unwrap(); + store + .bind_backend_job_attempt_worker( + "w", + id, + &reservation.attempt.attempt_id, + &worker, + Some("run"), + RESOURCE_JOB_NOW, + ) + .unwrap(); + let acceptance = store + .accept_backend_job_result( + "w", + &worker, + &BackendJobResultSubmission { + job_id: id.into(), + attempt_id: reservation.attempt.attempt_id.clone(), + input_digest: request.input_digest().unwrap(), + result: serde_json::json!({"valid":true}), + }, + RESOURCE_JOB_NOW, + ) + .unwrap(); + let source = JobNotificationSource { + job_id: id.into(), + attempt_id: acceptance.attempt.attempt_id.clone(), + delivery_id: crate::backend_job::delivery_id(id, &acceptance.attempt.attempt_id), + }; + (acceptance, source) +} + +fn job_notice_evidence( + store: &SqliteWorkspaceStore, + source: &JobNotificationSource, +) -> (String, String, bool, Option) { + store.with_conn(|conn|Ok(conn.query_row("SELECT disposition,reason,EXISTS(SELECT 1 FROM notification_acknowledgements WHERE workspace_id='w' AND delivery_id=?1),(SELECT code FROM notification_failures WHERE workspace_id='w' AND delivery_id=?1 ORDER BY sequence DESC LIMIT 1) FROM notification_decisions WHERE workspace_id='w' AND delivery_id=?1",[&source.delivery_id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?)))?)).unwrap() +} + +#[tokio::test] +async fn job_notification_routing_common_policy_distinguishes_suppression_failures_and_acceptance() +{ + let (_directory, store) = tracking_store_fixture().await; + for (name, target, facts, disposition, reason, failure) in [ + ( + "normal", + true, + JobNotificationFacts::Result("result notice".into()), + "deliver", + "result", + None, + ), + ( + "no-target", + false, + JobNotificationFacts::Result("unused".into()), + "suppress", + "no_recipient", + None, + ), + ( + "clean", + true, + JobNotificationFacts::NoFindings, + "suppress", + "no_findings", + None, + ), + ( + "stale", + true, + JobNotificationFacts::StaleContent, + "suppress", + "stale_content", + None, + ), + ( + "missing", + true, + JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed), + "defer", + "source_unavailable", + Some("source_lookup_failed"), + ), + ( + "invalid", + true, + JobNotificationFacts::Failed(NotificationFailureCode::InvalidResult), + "defer", + "source_unavailable", + Some("invalid_result"), + ), + ( + "oversize", + true, + JobNotificationFacts::Result( + "x".repeat(crate::backend_job::MAX_JOB_DELIVERY_BYTES + 1), + ), + "defer", + "source_unavailable", + Some("content_too_large"), + ), + ( + "render", + true, + JobNotificationFacts::Result(String::new()), + "defer", + "source_unavailable", + Some("rendering_failed"), + ), + ] { + let (acceptance, source) = job_notice_fixture(&store, name, target); + assert_eq!( + store.route_job_notification("w", &source, &facts).unwrap(), + disposition == "deliver" + ); + assert_eq!( + job_notice_evidence(&store, &source), + ( + disposition.into(), + reason.into(), + false, + failure.map(String::from) + ) + ); + assert_eq!( + store.get_backend_job("w", name).unwrap().unwrap().result, + acceptance.job.result + ); + assert_eq!( + store.get_backend_job("w", name).unwrap().unwrap().state, + BackendJobState::Completed + ); + if disposition != "deliver" { + assert!( + store + .job_notification_request("w", &source.delivery_id) + .unwrap() + .is_none() + ); + assert!( + !store + .route_job_notification( + "w", + &source, + &JobNotificationFacts::Result("new rendering".into()) + ) + .unwrap() + ); + } + } + assert_eq!( + store + .list_pending_job_notifications("w", 100) + .unwrap() + .len(), + 1 + ); +} + +#[tokio::test] +async fn job_notification_result_commit_reopen_parallel_recovery_and_fixed_target_receipt() { + let (directory, store) = tracking_store_fixture().await; + let (accepted, source) = job_notice_fixture(&store, "commit-gap", true); + // No delivery reservation or routing callback has run. The result transaction + // alone leaves one recoverable source, including after reopening SQLite. + assert!( + store + .job_notification_request("w", &source.delivery_id) + .unwrap() + .is_none() + ); + let submission = BackendJobResultSubmission { + job_id: source.job_id.clone(), + attempt_id: source.attempt_id.clone(), + input_digest: accepted.attempt.input_digest.clone(), + result: accepted.job.result.clone().unwrap(), + }; + assert!( + store + .accept_backend_job_result( + "w", + accepted.attempt.worker.as_ref().unwrap(), + &submission, + RESOURCE_JOB_NOW + ) + .unwrap() + .replayed + ); + assert!( + store + .accept_backend_job_result( + "other", + accepted.attempt.worker.as_ref().unwrap(), + &submission, + RESOURCE_JOB_NOW + ) + .is_err() + ); + let mut stale = submission.clone(); + stale.attempt_id = "old-attempt".into(); + assert!( + store + .accept_backend_job_result( + "w", + accepted.attempt.worker.as_ref().unwrap(), + &stale, + RESOURCE_JOB_NOW + ) + .is_err() + ); + drop(store); + let path = directory.path().join("workspace.db"); + let store = SqliteWorkspaceStore::open_for_test(&path).unwrap(); + assert_eq!( + store + .list_pending_job_notifications("w", 100) + .unwrap() + .len(), + 1 + ); + let barrier = std::sync::Arc::new(std::sync::Barrier::new(2)); + let threads = (0..2) + .map(|i| { + let path = path.clone(); + let source = source.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + let store = SqliteWorkspaceStore::open_for_test(path).unwrap(); + barrier.wait(); + store + .route_job_notification( + "w", + &source, + &JobNotificationFacts::Result(format!("immutable rendering {i}")), + ) + .unwrap(); + store + .job_notification_request("w", &source.delivery_id) + .unwrap() + .unwrap() + }) + }) + .collect::>(); + let results = threads + .into_iter() + .map(|thread| thread.join().unwrap()) + .collect::>(); + assert_eq!(results[0], results[1]); + assert!( + store + .route_job_notification("w", &source, &JobNotificationFacts::StaleContent) + .unwrap() + ); + assert_eq!( + store + .job_notification_request("w", &source.delivery_id) + .unwrap() + .unwrap(), + results[0] + ); + let target = TicketWorkerNotificationTarget { + worker: RuntimeWorkerRef::new("r", "replacement"), + authentication_binding_id: "tracking-binding".into(), + }; + assert_eq!( + store + .authorize_job_notification("w", &source.delivery_id, &target) + .unwrap(), + None + ); + let mut replacement = target.clone(); + replacement.worker.worker_id = "another-worker".into(); + assert!( + store + .authorize_job_notification("w", &source.delivery_id, &replacement) + .unwrap() + .is_some() + ); + replacement = target.clone(); + replacement.authentication_binding_id = "replacement-binding".into(); + assert!( + store + .authorize_job_notification("w", &source.delivery_id, &replacement) + .unwrap() + .is_some() + ); + store + .record_notification_delivery_failure( + "w", + &NotificationDeliveryKey::WorkerNotification { + delivery_id: source.delivery_id.clone(), + }, + NotificationFailureCode::AcceptanceUnknown, + ) + .unwrap(); + assert!(!job_notice_evidence(&store, &source).2); + assert_eq!( + store + .job_notification_request("w", &source.delivery_id) + .unwrap() + .unwrap(), + results[0] + ); + store.with_conn(|conn| {conn.execute("UPDATE workspace_runtime_bindings SET revoked_at=?1 WHERE workspace_id='w' AND runtime_id='r'",[RESOURCE_JOB_NOW])?;Ok(())}).unwrap(); + assert_eq!( + store + .authorize_job_notification("w", &source.delivery_id, &target) + .unwrap(), + Some(NotificationFailureCode::BindingRevoked) + ); + // A receipt about the already authorized send remains valid after revocation. + store + .ack_job_notification("w", &source.delivery_id, &target.worker) + .unwrap(); + store + .ack_job_notification("w", &source.delivery_id, &target.worker) + .unwrap(); + assert!(job_notice_evidence(&store, &source).2); + assert!( + store + .list_pending_job_notifications("w", 100) + .unwrap() + .is_empty() + ); + assert_eq!( + store + .get_backend_job("w", &source.job_id) + .unwrap() + .unwrap() + .current_attempt, + 1 + ); + store + .with_conn(|conn| { + assert_eq!( + conn.query_row( + "SELECT COUNT(*) FROM notification_sources WHERE role='backend_job'", + [], + |r| r.get::<_, i64>(0) + )?, + 1 + ); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM notification_requests", [], |r| r + .get::<_, i64>(0))?, + 1 + ); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r + .get::<_, i64>( + 0 + ))?, + 0 + ); + Ok(()) + }) + .unwrap(); +} + +fn job_notice_legacy_seed( + conn: &Connection, + name: &str, + state: Option<&str>, + category: Option<&str>, + target: bool, +) -> String { + let mut request = resource_job_request(name, name, name); + request.purpose = "legacy_notice".into(); + request.grants = Default::default(); + request.notification_target = + target.then(|| RuntimeWorkerRef::new("old-runtime", "old-worker")); + let attempt = attempt_id(name, 1); + let id = crate::backend_job::delivery_id(name, &attempt); + conn.execute("INSERT INTO backend_jobs(workspace_id,job_id,purpose,input_digest,input_ref,request_json,intent_fingerprint,state,current_attempt,result_json,result_digest,created_at,updated_at,completed_at) VALUES('w',?1,'legacy_notice',?2,'test://input',?3,'fingerprint','completed',1,'{}','digest',?4,?4,?4)",params![name,request.input_digest().unwrap(),serde_json::to_string(&request).unwrap(),RESOURCE_JOB_NOW]).unwrap(); + conn.execute("INSERT INTO backend_job_attempts(workspace_id,job_id,attempt_id,attempt,input_digest,state,deadline_at,result_json,result_digest,created_at,updated_at,completed_at) VALUES('w',?1,?2,1,?3,'completed',?4,'{}','digest',?4,?4,?4)",params![name,attempt,request.input_digest().unwrap(),RESOURCE_JOB_NOW]).unwrap(); + if let Some(state) = state { + conn.execute("INSERT INTO backend_job_deliveries(workspace_id,delivery_id,job_id,attempt_id,target_runtime_id,target_worker_id,state,failure_category,created_at,updated_at,delivered_at) VALUES('w',?1,?2,?3,'old-runtime','old-worker',?4,?5,?6,?6,CASE WHEN ?4='completed' THEN ?6 ELSE NULL END)",params![id,name,attempt,state,category,RESOURCE_JOB_NOW]).unwrap(); + } + id +} + +#[test] +fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and_commit_gaps() { + let directory = tempfile::tempdir().unwrap(); + let path = directory.path().join("workspace.db"); + let conn = tracking_schema91_fixture(&path); + migrate_ticket_tracking_v91_to_v92(&conn).unwrap(); + migrate_assignment_delivery_v92_to_v93(&conn).unwrap(); + migrate_ticket_posts_v93_to_v94(&conn).unwrap(); + migrate_notification_routing_v94_to_v95(&conn).unwrap(); + let accepted = job_notice_legacy_seed(&conn, "ack", Some("completed"), None, true); + let clean = job_notice_legacy_seed( + &conn, + "clean", + Some("completed"), + Some("notification_suppressed_no_findings"), + true, + ); + let stale = job_notice_legacy_seed( + &conn, + "stale", + Some("completed"), + Some("notification_suppressed_stale_ticket_content"), + true, + ); + let unknown = job_notice_legacy_seed( + &conn, + "unknown", + Some("unknown"), + Some("notification_outcome_unknown"), + true, + ); + let ambiguous = job_notice_legacy_seed( + &conn, + "ambiguous", + Some("completed"), + Some("unrecognized_legacy_reason"), + true, + ); + job_notice_legacy_seed(&conn, "no-target", None, None, false); + job_notice_legacy_seed(&conn, "commit-gap", None, None, true); + conn.execute("UPDATE backend_job_deliveries SET delivery_id='legacy-stable-request' WHERE delivery_id=?1",[&unknown]).unwrap(); + let unknown = "legacy-stable-request".to_string(); + let checker = job_notice_legacy_seed(&conn, "checker-unknown", Some("sending"), None, true); + let mut checker_request: BackendJobRequest = serde_json::from_str( + &conn + .query_row( + "SELECT request_json FROM backend_jobs WHERE job_id='checker-unknown'", + [], + |r| r.get::<_, String>(0), + ) + .unwrap(), + ) + .unwrap(); + checker_request.purpose = crate::ticket_item_checker::PURPOSE.into(); + checker_request.input = serde_json::json!({"ticket_id":"deleted-ticket","ticket_resource_key":"T-42","content_digest":"original-digest","title":"title","body":"Writer-only scope","edit":{"kind":"create"}}); + let checker_result = serde_json::json!({"findings":[{"category":"writer_scope","quote":"Writer-only scope","reason":"This is writer scope.","suggestion":"Remove it."}]}); + conn.execute( + "UPDATE backend_jobs SET request_json=?1,result_json=?2 WHERE job_id='checker-unknown'", + params![ + serde_json::to_string(&checker_request).unwrap(), + checker_result.to_string() + ], + ) + .unwrap(); + conn.execute( + "UPDATE backend_job_attempts SET result_json=?1 WHERE job_id='checker-unknown'", + [checker_result.to_string()], + ) + .unwrap(); + let before = stored_fixture_rows(&conn, "backend_job_deliveries"); + // Force migration failure after the source parent rebuild and verify rollback + // restores the old parent, rows, constraints, version and connection pragmas. + conn.execute_batch("CREATE TABLE notification_job_targets(collision TEXT)") + .unwrap(); + assert!(migrate_job_notifications_v95_to_v96(&conn).is_err()); + assert_eq!(current_schema_version(&conn).unwrap(), 95); + assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before); + assert_eq!( + conn.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0)) + .unwrap(), + 1 + ); + conn.execute_batch("DROP TABLE notification_job_targets") + .unwrap(); + migrate_job_notifications_v95_to_v96(&conn).unwrap(); + assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before); + for (id, disposition, reason, runtime_accepted) in [ + (&accepted, "deliver", "result", true), + (&clean, "suppress", "no_findings", false), + (&stale, "suppress", "stale_content", false), + (&unknown, "deliver", "result", false), + (&ambiguous, "defer", "source_unavailable", false), + ] { + let got=conn.query_row("SELECT d.disposition,d.reason,l.runtime_accepted FROM notification_decisions d JOIN notification_job_legacy_receipts l USING(workspace_id,delivery_id) WHERE d.workspace_id='w' AND d.delivery_id=?1",[id],|r|Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,bool>(2)?))).unwrap(); + assert_eq!(got, (disposition.into(), reason.into(), runtime_accepted)); + } + assert_eq!( + conn.query_row( + "SELECT COUNT(*) FROM notification_acknowledgements", + [], + |r| r.get::<_, i64>(0) + ) + .unwrap(), + 0, + "never fabricate modern ACKs or request bodies for old receipts" + ); + assert_eq!( + conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r + .get::<_, i64>( + 0 + )) + .unwrap(), + 0 + ); + let pinned: String = conn + .query_row( + "SELECT content FROM notification_requests WHERE delivery_id=?1", + [&unknown], + |r| r.get(0), + ) + .unwrap(); + assert!(pinned.contains("Durable result digest: digest.")); + drop(conn); + let store = SqliteWorkspaceStore::open_for_test(path).unwrap(); + let pending = store.list_pending_job_notifications("w", 100).unwrap(); + assert_eq!(pending.len(), 4); + let checker_body = store + .job_notification_request("w", &checker) + .unwrap() + .unwrap() + .0; + assert!(checker_body.contains("Content digest: original-digest")); + assert!(checker_body.contains("Quote: \"Writer-only scope\"")); + assert!(checker_body.contains("not a new user request, approval gate")); + assert!(pending.iter().all(|s| s.delivery_id != accepted + && s.delivery_id != clean + && s.delivery_id != stale + && s.delivery_id != ambiguous)); + assert_eq!( + store + .job_notification_request("w", &unknown) + .unwrap() + .unwrap() + .0, + pinned + ); +} + +#[tokio::test] +async fn job_notification_source_is_atomic_with_result_and_schema_guards_are_verified() { + let (directory, store) = tracking_store_fixture().await; + let mut request = resource_job_request("atomic-source", "atomic-source", "atomic-source"); + request.purpose = "notification_test".into(); + request.grants = Default::default(); + let worker = RuntimeWorkerRef::new("r", "replacement"); + let reservation = store + .reserve_backend_job("w", &request, RESOURCE_JOB_NOW) + .unwrap(); + store + .claim_backend_job_attempt_dispatch( + "w", + &request.job_id, + &reservation.attempt.attempt_id, + RESOURCE_JOB_NOW, + ) + .unwrap(); + store + .bind_backend_job_attempt_worker( + "w", + &request.job_id, + &reservation.attempt.attempt_id, + &worker, + Some("run"), + RESOURCE_JOB_NOW, + ) + .unwrap(); + let submission = BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: reservation.attempt.attempt_id.clone(), + input_digest: request.input_digest().unwrap(), + result: serde_json::json!({"saved":true}), + }; + store.with_conn(|conn| {conn.execute_batch("CREATE TRIGGER reject_job_source BEFORE INSERT ON notification_sources WHEN NEW.role='backend_job' BEGIN SELECT RAISE(ABORT,'injected source commit failure'); END;")?;Ok(())}).unwrap(); + assert!( + store + .accept_backend_job_result("w", &worker, &submission, RESOURCE_JOB_NOW) + .is_err() + ); + assert_eq!( + store + .get_backend_job("w", &request.job_id) + .unwrap() + .unwrap() + .state, + BackendJobState::Pending + ); + assert!( + store + .list_pending_job_notifications("w", 100) + .unwrap() + .is_empty() + ); + store + .with_conn(|conn| { + conn.execute_batch("DROP TRIGGER reject_job_source")?; + Ok(()) + }) + .unwrap(); + store + .accept_backend_job_result("w", &worker, &submission, RESOURCE_JOB_NOW) + .unwrap(); + assert_eq!( + store + .list_pending_job_notifications("w", 100) + .unwrap() + .len(), + 1 + ); + store + .with_conn(|conn| { + conn.execute_batch("DROP TRIGGER notification_job_targets_no_update")?; + Ok(()) + }) + .unwrap(); + drop(store); + assert!( + matches!(SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")),Err(error) if error.to_string().contains("notification_job_targets_no_update")) + ); +} diff --git a/crates/workspace-server/src/notification_delivery.rs b/crates/workspace-server/src/notification_delivery.rs index 0e72d430a..414527229 100644 --- a/crates/workspace-server/src/notification_delivery.rs +++ b/crates/workspace-server/src/notification_delivery.rs @@ -9,7 +9,8 @@ //! | --- | --- | --- | //! | Assign, Answer, content and post edits, case, tags, tracking, release | Ticket transaction captures facts; `notification_routing` decides both recipients and non-notification reasons | Saved assignment/Worker requests and ACKs; mutation wakeup and periodic rounds | //! | Workspace startup, Orchestrator create/restore/replacement, idle | Wakeup only; never a new request or a second policy | Same saved pending decisions | -//! | MR/handoff and Backend Job | Separate adapters to be connected by follow-up work | Not inferred from comments or display state | +//! | Backend Job result | Result transaction captures source; checker supplies domain facts; common routing records result/suppression/failure | Fixed Worker request bytes, binding, ACK and legacy receipt; same recovery rounds | +//! | MR/handoff | Separate adapter to be connected by follow-up work | Not inferred from comments or display state | //! | Explicit WorkerNotify/WorkerSendInput | Authenticated caller communication | Not automatic business routing | //! | Internal SubWorker result and Web subscription | Parent/child execution and UI synchronization respectively | Remain separate; no Ticket policy in Runtime | //! @@ -121,10 +122,17 @@ pub(super) fn record_notification_failure( pub(super) fn notification_target( api: &WorkspaceApi, input: &TicketWorkerNotification, +) -> std::result::Result { + fixed_worker_target(api, &input.worker) +} + +fn fixed_worker_target( + api: &WorkspaceApi, + recipient: &RuntimeWorkerRef, ) -> std::result::Result { let worker = api .runtime - .worker(&input.worker) + .worker(recipient) .map_err(|_| NotificationFailureCode::TargetUnavailable)?; if worker.workspace.workspace_id.as_deref() != Some(&api.config.workspace_id) { return Err(NotificationFailureCode::WorkspaceMismatch); @@ -134,14 +142,14 @@ pub(super) fn notification_target( } let binding = api .config_store - .get_workspace_runtime_binding(&api.config.workspace_id, &input.worker.runtime_id) + .get_workspace_runtime_binding(&api.config.workspace_id, &recipient.runtime_id) .map_err(|_| NotificationFailureCode::SourceLookupFailed)? .ok_or(NotificationFailureCode::TargetUnavailable)?; if binding.revoked_at.is_some() { return Err(NotificationFailureCode::BindingRevoked); } Ok(TicketWorkerNotificationTarget { - worker: input.worker.clone(), + worker: recipient.clone(), authentication_binding_id: binding.binding_id, }) } @@ -264,6 +272,7 @@ pub(super) fn recover_ticket_notifications(api: &WorkspaceApi) { let Some(_permit) = api.notification_delivery_lifecycle.begin() else { return; }; + recover_job_notifications(api); dispatch_ticket_worker_notifications(api); if let Some(orchestrator) = find_workspace_orchestrator(api) { dispatch_orchestrator_assignment_attention(api, &orchestrator.worker); @@ -339,3 +348,195 @@ mod tests { assert!(lifecycle.begin().is_none()); } } + +/// Recover notification sources, not Job execution. This same bounded round runs +/// on startup and periodically; no Worker restoration or assignment is implicit. +pub(super) fn recover_job_notifications(api: &WorkspaceApi) { + let Some(_permit) = api.notification_delivery_lifecycle.begin() else { + return; + }; + let sources = match api + .store + .list_pending_job_notifications(&api.config.workspace_id, 20) + { + Ok(sources) => sources, + Err(error) => { + tracing::warn!(%error,"Job notification scan failed"); + return; + } + }; + for source in sources { + if api.notification_delivery_lifecycle.stopped() { + break; + } + let acceptance = (|| -> Result { + let job = api + .store + .get_backend_job(&api.config.workspace_id, &source.job_id)? + .ok_or_else(|| Error::Store("Job notification result disappeared".into()))?; + let attempt = api + .store + .get_backend_job_attempt( + &api.config.workspace_id, + &source.job_id, + &source.attempt_id, + )? + .ok_or_else(|| Error::Store("Job notification attempt disappeared".into()))?; + Ok(BackendJobResultAcceptance { + job, + attempt, + replayed: true, + }) + })(); + match acceptance { + Ok(acceptance) => deliver_job_result(api, &acceptance), + Err(error) => tracing::warn!(%error,"Job notification source lookup failed"), + } + } +} + +fn job_result_facts( + api: &WorkspaceApi, + acceptance: &BackendJobResultAcceptance, +) -> crate::store::JobNotificationFacts { + use crate::store::JobNotificationFacts as F; + let content = if acceptance.job.request.purpose == ticket_item_checker::PURPOSE { + let input = match ticket_item_checker::parse_input(&acceptance.job.request.input) { + Ok(input) => input, + Err(_) => return F::Failed(NotificationFailureCode::InvalidResult), + }; + let ticket = match browser_ticket_backend(api) + .and_then(|backend| backend.show(input.ticket_id.into()).map_err(Error::from)) + { + Ok(ticket) => ticket, + Err(Error::Ticket(ticket::TicketError::NotFound(_))) => { + return F::Failed(NotificationFailureCode::SourceNotFound); + } + Err(_) => return F::Failed(NotificationFailureCode::SourceLookupFailed), + }; + match ticket_item_checker::result_facts(acceptance, &ticket) { + Ok(ticket_item_checker::TicketItemCheckFacts::Findings(content)) => content, + Ok(ticket_item_checker::TicketItemCheckFacts::NoFindings) => { + return F::NoFindings; + } + Ok(ticket_item_checker::TicketItemCheckFacts::Stale) => return F::StaleContent, + Err(_) => return F::Failed(NotificationFailureCode::InvalidResult), + } + } else { + format!( + "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.", + acceptance.job.request.job_id, + acceptance.attempt.attempt_id, + acceptance.job.request.purpose, + acceptance + .job + .result_digest + .as_deref() + .unwrap_or("unavailable") + ) + }; + if content.len() > MAX_JOB_DELIVERY_BYTES { + F::Failed(NotificationFailureCode::ContentTooLarge) + } else { + F::Result(content) + } +} + +pub(super) fn deliver_job_result(api: &WorkspaceApi, acceptance: &BackendJobResultAcceptance) { + use crate::store::JobNotificationFacts; + let Some(_permit) = api.notification_delivery_lifecycle.begin() else { + return; + }; + let source = match api.store.job_notification_source( + &api.config.workspace_id, + &acceptance.job.request.job_id, + &acceptance.attempt.attempt_id, + ) { + Ok(source) => source, + Err(error) => { + tracing::warn!(%error,"Job notification source lookup failed"); + return; + } + }; + let workspace = &api.config.workspace_id; + let key = NotificationDeliveryKey::WorkerNotification { + delivery_id: source.delivery_id.clone(), + }; + // The prepared body wins over all subsequent rendering and content changes. + let request = match api + .store + .job_notification_request(workspace, &source.delivery_id) + { + Ok(request) => request, + Err(error) => { + tracing::warn!(%error,"Job notification request lookup failed"); + return; + } + }; + let facts = match &request { + Some((content, _)) => JobNotificationFacts::Result(content.clone()), + None => job_result_facts(api, acceptance), + }; + match api.store.route_job_notification(workspace, &source, &facts) { + Ok(true) => {} + Ok(false) => return, + Err(error) => { + tracing::warn!(%error,"Job notification routing failed; committed source retained"); + return; + } + } + let request = match api + .store + .job_notification_request(workspace, &source.delivery_id) + { + Ok(Some(request)) => request, + _ => { + record_notification_failure(api, &key, NotificationFailureCode::PreparationFailed); + return; + } + }; + let (content, target) = request; + if acceptance.job.request.notification_target.as_ref() != Some(&target) { + record_notification_failure(api, &key, NotificationFailureCode::PreparationFailed); + return; + } + let authorized = match fixed_worker_target(api, &target) { + Ok(target) => target, + Err(code) => { + record_notification_failure(api, &key, code); + return; + } + }; + match api + .store + .authorize_job_notification(workspace, &source.delivery_id, &authorized) + { + Ok(None) => {} + Ok(Some(code)) => { + record_notification_failure(api, &key, code); + return; + } + Err(_) => { + record_notification_failure(api, &key, NotificationFailureCode::SourceLookupFailed); + return; + } + } + let result = api.runtime.send_input( + &target, + WorkerInputRequest { + kind: WorkerInputKind::Notify, + content, + submission_request_id: Some(source.delivery_id.clone()), + segments: None, + }, + ); + if let Some(failure) = notification_failure(&result) { + record_notification_failure(api, &key, failure); + } else if api + .store + .ack_job_notification(workspace, &source.delivery_id, &target) + .is_err() + { + record_notification_failure(api, &key, NotificationFailureCode::AcknowledgementFailed); + } +} diff --git a/crates/workspace-server/src/notification_routing.rs b/crates/workspace-server/src/notification_routing.rs index 46aeb6b22..bea3cddc4 100644 --- a/crates/workspace-server/src/notification_routing.rs +++ b/crates/workspace-server/src/notification_routing.rs @@ -2,6 +2,9 @@ //! Runtime acceptance, handling and execution authority remain independent. Explicit user //! communication, internal child notifications and Web subscriptions are not business routes. use super::*; +#[path = "job_notification_routing.rs"] +pub(super) mod jobs; +pub use jobs::{JobNotificationFacts, JobNotificationSource}; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -39,6 +42,7 @@ impl NotificationEventKind { "tags" | "tags_set" | "tags_changed" => Self::Tags, "release" => Self::Release, "assignment_changed" => Self::AssignmentChanged, + "backend_job_result" => Self::Result, _ => Self::Unknown, } } @@ -51,6 +55,9 @@ pub enum NotificationReason { Attention, Result, NoRecipient, + NoFindings, + StaleContent, + SourceUnavailable, UnsupportedRecipient, HistoricalOnly, CaseClosed, @@ -66,7 +73,10 @@ pub enum NotificationReason { pub enum NotificationFailureCode { ConfigurationUnavailable, RenderingFailed, + ContentTooLarge, + InvalidResult, SourceLookupFailed, + SourceNotFound, TargetUnavailable, TargetStopped, WorkspaceMismatch, @@ -371,7 +381,7 @@ fn read_source( Ok(( row.get(0)?, row.get(1)?, - row.get(2)?, + row.get::<_, Option>(2)?.unwrap_or_default(), row.get(3)?, row.get(4)?, row.get(5)?, @@ -469,7 +479,7 @@ fn decision_for_source( } pub(super) fn route_pending(conn: &Connection, workspace: &str) -> Result<()> { - let mut stmt = conn.prepare("SELECT s.* FROM notification_sources s LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND d.source_sequence IS NULL ORDER BY s.sequence LIMIT 500")?; + let mut stmt = conn.prepare("SELECT s.* FROM notification_sources s LEFT JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND s.role!='backend_job' AND d.source_sequence IS NULL ORDER BY s.sequence LIMIT 500")?; let mut rows = stmt.query([workspace])?; while let Some(row) = rows.next()? { let source = source_from_row(row)?; diff --git a/crates/workspace-server/src/notification_routing_tests.rs b/crates/workspace-server/src/notification_routing_tests.rs index ac62b6a32..e61700d65 100644 --- a/crates/workspace-server/src/notification_routing_tests.rs +++ b/crates/workspace-server/src/notification_routing_tests.rs @@ -843,6 +843,7 @@ async fn notification_routing_fresh_and_upgraded_schema_and_guards_match() { migrate_assignment_delivery_v92_to_v93(&conn).unwrap(); migrate_ticket_posts_v93_to_v94(&conn).unwrap(); migrate_notification_routing_v94_to_v95(&conn).unwrap(); + migrate_job_notifications_v95_to_v96(&conn).unwrap(); let snapshot = |conn: &Connection| -> Vec<(String, String)> { conn.prepare("SELECT name,sql FROM sqlite_schema WHERE name LIKE 'notification_%' OR name LIKE 'ticket_assignment_content%' OR name IN ('ticket_assignment_title_input','ticket_assignment_initial_post_input') ORDER BY name").unwrap().query_map([],|r|Ok((r.get(0)?,r.get(1)?))).unwrap().collect::>().unwrap() }; diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 941020e0d..57195f84a 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -181,7 +181,7 @@ use crate::authority::{ }; use crate::backend_job::{ ABSOLUTE_MAX_CONCURRENT_JOBS, BackendJobAttemptRecord, BackendJobAttemptState, - BackendJobDeliveryState, BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance, + BackendJobRequest, BackendJobReservation, BackendJobResultAcceptance, BackendJobResultSubmission, BackendJobState, BackendJobWorkerCleanupState, MAX_JOB_DELIVERY_BYTES, allocation_key, }; @@ -3973,38 +3973,7 @@ impl WorkspaceApi { api.host_executors .host()? .run(move || { - let recovery_now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); - let deliveries = api - .store - .recover_backend_job_deliveries(&api.config.workspace_id, &recovery_now)?; - for delivery in deliveries { - let Some(job) = api - .store - .get_backend_job(&api.config.workspace_id, &delivery.job_id)? - else { - continue; - }; - let Some(attempt) = api.store.get_backend_job_attempt( - &api.config.workspace_id, - &delivery.job_id, - &delivery.attempt_id, - )? - else { - continue; - }; - if job.state == BackendJobState::Completed - && attempt.state == BackendJobAttemptState::Completed - { - api.deliver_backend_job_result( - &BackendJobResultAcceptance { - job, - attempt, - replayed: true, - }, - &delivery.target, - ); - } - } + notification_delivery::recover_job_notifications(&api); api.recover_backend_jobs()?; if let Some(orchestrator) = find_online_workspace_orchestrator(&api) { dispatch_orchestrator_assignment_attention(&api, &orchestrator.worker); @@ -4892,175 +4861,19 @@ impl WorkspaceApi { &verified_submission, &accepted_at, )?; - if let Some(target) = acceptance.job.request.notification_target.as_ref() { - self.deliver_backend_job_result(&acceptance, target); - } + notification_delivery::deliver_job_result(self, &acceptance); let _ = self.schedule_backend_job_worker_cleanup(&acceptance.attempt); self.drain_reserved_backend_jobs(); Ok(acceptance) } + #[cfg(test)] fn deliver_backend_job_result( &self, acceptance: &BackendJobResultAcceptance, - target: &RuntimeWorkerRef, + _target: &RuntimeWorkerRef, ) { - let delivery_id = crate::backend_job::delivery_id( - &acceptance.job.request.job_id, - &acceptance.attempt.attempt_id, - ); - let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); - let Ok((delivery, claimed)) = self.store.reserve_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - &acceptance.job.request.job_id, - target, - &now, - ) else { - return; - }; - if !claimed || delivery.state != BackendJobDeliveryState::Sending { - return; - } - - let content = if acceptance.job.request.purpose == ticket_item_checker::PURPOSE { - let checker_input = - match ticket_item_checker::parse_input(&acceptance.job.request.input) { - Ok(input) => input, - Err(error) => { - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - BackendJobDeliveryState::Failed, - Some("invalid_checker_input"), - Some(&error.to_string()), - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); - return; - } - }; - let current_ticket = match browser_ticket_backend(self).and_then(|backend| { - backend - .show(checker_input.ticket_id.clone().into()) - .map_err(Error::from) - }) { - Ok(ticket) => ticket, - Err(error) => { - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - BackendJobDeliveryState::Failed, - Some("current_ticket_content_unavailable"), - Some(&error.to_string()), - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); - return; - } - }; - match ticket_item_checker::notification(acceptance, ¤t_ticket) { - Ok(ticket_item_checker::TicketItemCheckNotification::Deliver(content)) => content, - Ok(ticket_item_checker::TicketItemCheckNotification::NoFindings) => { - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - BackendJobDeliveryState::Completed, - Some("notification_suppressed_no_findings"), - None, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); - return; - } - Ok(ticket_item_checker::TicketItemCheckNotification::Stale) => { - let detail = format!( - "checked Ticket content digest {} is no longer current ({})", - checker_input.content_digest, - ticket_item_checker::content_digest(¤t_ticket) - ); - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - BackendJobDeliveryState::Completed, - Some("notification_suppressed_stale_ticket_content"), - Some(&detail), - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); - return; - } - Err(error) => { - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - BackendJobDeliveryState::Failed, - Some("invalid_checker_result"), - Some(&error.to_string()), - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); - return; - } - } - } else { - format!( - "Backend Job `{}` completed attempt `{}` for `{}`. Durable result digest: {}.", - acceptance.job.request.job_id, - acceptance.attempt.attempt_id, - acceptance.job.request.purpose, - acceptance - .job - .result_digest - .as_deref() - .unwrap_or("unavailable") - ) - }; - if content.len() > MAX_JOB_DELIVERY_BYTES { - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - BackendJobDeliveryState::Failed, - Some("notification_too_large"), - Some("Backend Job notification content exceeded its configured byte limit"), - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); - return; - } - let delivery_result = self.runtime.send_input( - target, - WorkerInputRequest { - kind: WorkerInputKind::Notify, - content, - submission_request_id: Some(delivery.delivery_id.clone()), - segments: None, - }, - ); - let (state, category, detail): (_, Option, Option) = match delivery_result { - Ok(result) if result.state == InternalWorkerOperationState::Accepted => { - (BackendJobDeliveryState::Completed, None, None) - } - Ok(result) if result.disposition == WorkerInputDisposition::Unknown => ( - BackendJobDeliveryState::Unknown, - Some("notification_outcome_unknown".to_string()), - result.diagnostics.first().map(|item| item.message.clone()), - ), - Ok(result) => ( - BackendJobDeliveryState::Failed, - Some("notification_rejected".to_string()), - result.diagnostics.first().map(|item| item.message.clone()), - ), - Err(error) => ( - BackendJobDeliveryState::Unknown, - Some("notification_outcome_unknown".to_string()), - Some(error.message()), - ), - }; - // Delivery is intentionally separate from result authority. A delivery - // bookkeeping failure must never roll back or re-run completed work. - let _ = self.store.finish_backend_job_delivery( - &self.config.workspace_id, - &delivery_id, - state, - category.as_deref(), - detail.as_deref(), - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ); + notification_delivery::deliver_job_result(self, acceptance); } fn register_workspace_runtime_binding(&self, binding: WorkspaceRuntimeBinding) -> Result<()> { @@ -46900,7 +46713,16 @@ mod tests { .unwrap(); } + fn job_notification_evidence( + api: &WorkspaceApi, + id: &str, + ) -> (String, String, bool, Option) { + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + conn.query_row("SELECT d.disposition,d.reason,EXISTS(SELECT 1 FROM notification_acknowledgements a WHERE a.workspace_id=d.workspace_id AND a.delivery_id=d.delivery_id),(SELECT code FROM notification_failures f WHERE f.workspace_id=d.workspace_id AND f.delivery_id=d.delivery_id ORDER BY sequence DESC LIMIT 1) FROM notification_decisions d WHERE d.workspace_id=?1 AND d.delivery_id=?2",rusqlite::params![TEST_WORKSPACE_ID,id],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).unwrap() + } + fn spawn_ticket_check_source(api: &WorkspaceApi, name: &str) -> RuntimeWorkerRef { + seed_test_delivery_binding(api); api.spawn_workspace_worker( EMBEDDED_WORKER_RUNTIME_ID, WorkerSpawnRequest { @@ -47578,21 +47400,9 @@ mod tests { ); let stale_delivery_id = crate::backend_job::delivery_id(&stale_job.request.job_id, &stale_attempt.attempt_id); - let (stale_delivery, claimed) = api - .store - .reserve_backend_job_delivery( - TEST_WORKSPACE_ID, - &stale_delivery_id, - &stale_job.request.job_id, - &source, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ) - .unwrap(); - assert!(!claimed); - assert_eq!(stale_delivery.state, BackendJobDeliveryState::Completed); assert_eq!( - stale_delivery.failure_category.as_deref(), - Some("notification_suppressed_stale_ticket_content") + job_notification_evidence(&api, &stale_delivery_id), + ("suppress".into(), "stale_content".into(), false, None) ); api.accept_backend_job_result( @@ -47610,20 +47420,9 @@ mod tests { ¤t_job.request.job_id, ¤t_attempt.attempt_id, ); - let (clean_delivery, claimed) = api - .store - .reserve_backend_job_delivery( - TEST_WORKSPACE_ID, - &clean_delivery_id, - ¤t_job.request.job_id, - &source, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ) - .unwrap(); - assert!(!claimed); assert_eq!( - clean_delivery.failure_category.as_deref(), - Some("notification_suppressed_no_findings") + job_notification_evidence(&api, &clean_delivery_id), + ("suppress".into(), "no_findings".into(), false, None) ); let Json(_) = execute_worker_ticket_test_operation( @@ -47939,21 +47738,15 @@ mod tests { ); let delivery_id = crate::backend_job::delivery_id(&request.job_id, &dispatched.attempt.attempt_id); - let (delivery, claimed) = api - .store - .reserve_backend_job_delivery( - TEST_WORKSPACE_ID, - &delivery_id, - &request.job_id, - ¬ification_target, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ) - .unwrap(); - assert!(!claimed); - assert_eq!(delivery.state, BackendJobDeliveryState::Unknown); - assert_eq!( - delivery.failure_category.as_deref(), - Some("notification_outcome_unknown") + let evidence = job_notification_evidence(&api, &delivery_id); + assert_eq!(evidence.0, "deliver"); + assert!(!evidence.2); + assert_eq!(evidence.3.as_deref(), Some("acceptance_unknown")); + assert!( + api.store + .job_notification_request(TEST_WORKSPACE_ID, &delivery_id) + .unwrap() + .is_some() ); } @@ -48421,6 +48214,7 @@ mod tests { ) .await .unwrap(); + seed_test_delivery_binding(&api); let target = api .spawn_workspace_worker( EMBEDDED_WORKER_RUNTIME_ID, @@ -48512,18 +48306,15 @@ mod tests { .any(|request_id| request_id.as_deref() == Some(delivery_id.as_str())), "recovery must replay the durable delivery id as the tracked notification id: {request_ids:?}" ); - let (delivery, claimed) = api - .store - .reserve_backend_job_delivery( - TEST_WORKSPACE_ID, - &delivery_id, - &request.job_id, - &target, - &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), - ) - .unwrap(); - assert!(!claimed); - assert_eq!(delivery.state, BackendJobDeliveryState::Completed); + assert_eq!(job_notification_evidence(&api, &delivery_id).2, true); + assert!( + api.store + .list_pending_job_notifications(TEST_WORKSPACE_ID, 100) + .unwrap() + .is_empty() + ); + notification_delivery::recover_job_notifications(&api); + assert!(execution.take_input_request_ids().is_empty()); } #[tokio::test] diff --git a/crates/workspace-server/src/server_assignment_delivery_tests.rs b/crates/workspace-server/src/server_assignment_delivery_tests.rs index bf33cb92b..649e1aea8 100644 --- a/crates/workspace-server/src/server_assignment_delivery_tests.rs +++ b/crates/workspace-server/src/server_assignment_delivery_tests.rs @@ -1765,3 +1765,5 @@ async fn catalog_shutdown_stops_notification_hook_and_fences_later_dispatch() { ); delivery_stop(&api, &orchestrator); } + +include!("server_job_notification_tests.rs"); diff --git a/crates/workspace-server/src/server_job_notification_tests.rs b/crates/workspace-server/src/server_job_notification_tests.rs new file mode 100644 index 000000000..3a4c8fb19 --- /dev/null +++ b/crates/workspace-server/src/server_job_notification_tests.rs @@ -0,0 +1,185 @@ +// Uses the real Controller/session fixtures above; only model output and the +// deliberately lost Runtime response are substituted. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn job_notification_real_worker_ack_loss_reopen_and_parallel_recovery_commits_once() { + let dir = tempfile::tempdir().unwrap(); + let (api, execution, _) = delivery_real_fixture(dir.path(), false).await; + let target = spawn_ticket_check_source(&api, "fixed-job-result-target"); + let request = BackendJobRequest { + job_id: "real-job-notification".into(), + purpose: "result_notification_fixture".into(), + input_ref: "test://job/result".into(), + input: json!({"fixed":true}), + instruction: "Return a result.".into(), + profile: "builtin:backend-job".into(), + grants: Default::default(), + serialization_key: None, + source_worker: None, + notification_target: Some(target.clone()), + limits: Default::default(), + }; + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let worker = dispatched.attempt.worker.as_ref().unwrap(); + let submission = BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: dispatched.attempt.attempt_id.clone(), + input_digest: request.input_digest().unwrap(), + result: json!({"result":"saved"}), + }; + let accepted = api + .store + .accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT) + .unwrap(); + let id = crate::backend_job::delivery_id(&request.job_id, &accepted.attempt.attempt_id); + // Result commit without its wakeup. No old reserve record is necessary. + assert!( + api.store + .job_notification_request(TEST_WORKSPACE_ID, &id) + .unwrap() + .is_none() + ); + execution + .lose_responses + .store(true, std::sync::atomic::Ordering::SeqCst); + notification_delivery::recover_job_notifications(&api); + delivery_wait_idle(&api, &target, 1).await; + let pinned = api + .store + .job_notification_request(TEST_WORKSPACE_ID, &id) + .unwrap() + .unwrap(); + let evidence = job_notification_evidence(&api, &id); + assert!(!evidence.2); + assert_eq!(evidence.3.as_deref(), Some("acceptance_unknown")); + assert_eq!(delivery_receipts(&api, &target).len(), 1); + // Reopen only Backend persistence, retaining the real Runtime receipt/session. + let mut reopened = api.clone(); + reopened.store = + Arc::new(SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap()); + execution + .lose_responses + .store(false, std::sync::atomic::Ordering::SeqCst); + let barrier = Arc::new(std::sync::Barrier::new(4)); + std::thread::scope(|scope| { + let joins = (0..4) + .map(|_| { + let api = reopened.clone(); + let barrier = barrier.clone(); + scope.spawn(move || { + barrier.wait(); + notification_delivery::recover_job_notifications(&api); + }) + }) + .collect::>(); + for join in joins { + join.join().unwrap(); + } + }); + delivery_wait_idle(&reopened, &target, 1).await; + assert!(job_notification_evidence(&reopened, &id).2); + assert_eq!(delivery_receipts(&reopened, &target).len(), 1); + assert_eq!(delivery_commits(&reopened, &target), vec![pinned.0.clone()]); + let attempts = execution.attempts.lock().unwrap(); + let retries = attempts + .iter() + .filter(|attempt| attempt.0 == id) + .collect::>(); + assert!(retries.len() >= 2); + assert!(retries.iter().all(|attempt| attempt.1 == pinned.0)); + drop(attempts); + assert!( + reopened + .store + .accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT) + .unwrap() + .replayed + ); + assert_eq!( + reopened + .store + .get_backend_job(TEST_WORKSPACE_ID, &request.job_id) + .unwrap() + .unwrap() + .current_attempt, + 1 + ); + assert!( + reopened + .store + .list_pending_job_notifications(TEST_WORKSPACE_ID, 100) + .unwrap() + .is_empty() + ); + delivery_stop(&reopened, &target); + delivery_stop(&reopened, worker); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn job_notification_stopped_fixed_recipient_is_not_replaced_or_restored() { + let dir = tempfile::tempdir().unwrap(); + let (api, execution, _) = delivery_real_fixture(dir.path(), false).await; + let target = spawn_ticket_check_source(&api, "stopped-job-target"); + let request = BackendJobRequest { + job_id: "stopped-fixed-job".into(), + purpose: "result_notification_fixture".into(), + input_ref: "test://job/stopped".into(), + input: json!({"fixed":true}), + instruction: "Return a result.".into(), + profile: "builtin:backend-job".into(), + grants: Default::default(), + serialization_key: None, + source_worker: None, + notification_target: Some(target.clone()), + limits: Default::default(), + }; + let dispatched = api.dispatch_backend_job(&request).unwrap(); + let worker = dispatched.attempt.worker.as_ref().unwrap(); + let accepted = api + .store + .accept_backend_job_result( + TEST_WORKSPACE_ID, + worker, + &BackendJobResultSubmission { + job_id: request.job_id.clone(), + attempt_id: dispatched.attempt.attempt_id.clone(), + input_digest: request.input_digest().unwrap(), + result: json!({"saved":true}), + }, + TEST_CREATED_AT, + ) + .unwrap(); + let id = crate::backend_job::delivery_id(&request.job_id, &accepted.attempt.attempt_id); + delivery_stop(&api, &target); + let replacement = spawn_ticket_check_source(&api, "different-job-target"); + notification_delivery::recover_job_notifications(&api); + let evidence = job_notification_evidence(&api, &id); + assert!(!evidence.2); + assert_eq!(evidence.3.as_deref(), Some("target_stopped")); + assert!( + !execution + .attempts + .lock() + .unwrap() + .iter() + .any(|attempt| attempt.0 == id) + ); + assert_eq!( + api.store + .job_notification_request(TEST_WORKSPACE_ID, &id) + .unwrap() + .unwrap() + .1, + target + ); + assert_eq!(api.runtime.worker(&target).unwrap().state, "stopped"); + assert_eq!( + api.store + .get_backend_job(TEST_WORKSPACE_ID, &request.job_id) + .unwrap() + .unwrap() + .result, + Some(json!({"saved":true})) + ); + delivery_stop(&api, &replacement); + delivery_stop(&api, worker); +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 442a3b09e..4ac0a151c 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -65,8 +65,9 @@ mod assignment_delivery; #[path = "notification_routing.rs"] mod notification_routing; pub use notification_routing::{ - NotificationDeliveryKey, NotificationEventKind, NotificationFailureCode, NotificationPurpose, - NotificationReason, TicketWorkerNotification, TicketWorkerNotificationTarget, + JobNotificationFacts, JobNotificationSource, NotificationDeliveryKey, NotificationEventKind, + NotificationFailureCode, NotificationPurpose, NotificationReason, TicketWorkerNotification, + TicketWorkerNotificationTarget, }; #[path = "assignment_tracking.rs"] mod assignment_tracking; @@ -79,7 +80,7 @@ pub use ticket::{ }; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 95; +const LATEST_SCHEMA_VERSION: i64 = 96; const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants"; const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs"; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; @@ -362,6 +363,11 @@ const MIGRATIONS: &[Migration] = &[ name: "Common notification routing and durable Worker delivery", apply: migrate_notification_routing_v94_to_v95, }, + Migration { + version: 96, + name: "Backend Job common notification routing", + apply: migrate_job_notifications_v95_to_v96, + }, ]; #[derive(Clone, Copy)] @@ -1702,6 +1708,40 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { failure_detail: Option<&str>, now: &str, ) -> Result; + fn job_notification_source( + &self, + workspace: &str, + job: &str, + attempt: &str, + ) -> Result; + fn list_pending_job_notifications( + &self, + workspace: &str, + limit: usize, + ) -> Result>; + fn route_job_notification( + &self, + workspace: &str, + source: &JobNotificationSource, + facts: &JobNotificationFacts, + ) -> Result; + fn job_notification_request( + &self, + workspace: &str, + id: &str, + ) -> Result>; + fn authorize_job_notification( + &self, + workspace: &str, + id: &str, + target: &TicketWorkerNotificationTarget, + ) -> Result>; + fn ack_job_notification( + &self, + workspace: &str, + id: &str, + target: &RuntimeWorkerRef, + ) -> Result<()>; fn reserve_backend_job_delivery( &self, workspace_id: &str, @@ -6966,6 +7006,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { .ok_or_else(|| Error::Store("accepted Backend Job disappeared".to_string()))?; let attempt = read_backend_job_attempt(&tx, workspace_id, &submission.job_id, &submission.attempt_id)? .ok_or_else(|| Error::Store("accepted Backend Job attempt disappeared".to_string()))?; + notification_routing::jobs::capture(&tx,workspace_id,&submission.job_id,&submission.attempt_id)?; tx.commit()?; Ok(BackendJobResultAcceptance { job, attempt, replayed: false }) }) @@ -7180,6 +7221,53 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn job_notification_source( + &self, + workspace: &str, + job: &str, + attempt: &str, + ) -> Result { + notification_routing::jobs::source(self, workspace, job, attempt) + } + fn list_pending_job_notifications( + &self, + workspace: &str, + limit: usize, + ) -> Result> { + notification_routing::jobs::pending(self, workspace, limit) + } + fn route_job_notification( + &self, + workspace: &str, + source: &JobNotificationSource, + facts: &JobNotificationFacts, + ) -> Result { + notification_routing::jobs::route(self, workspace, source, facts) + } + fn job_notification_request( + &self, + workspace: &str, + id: &str, + ) -> Result> { + notification_routing::jobs::request(self, workspace, id) + } + fn authorize_job_notification( + &self, + workspace: &str, + id: &str, + target: &TicketWorkerNotificationTarget, + ) -> Result> { + notification_routing::jobs::authorize(self, workspace, id, target) + } + fn ack_job_notification( + &self, + workspace: &str, + id: &str, + target: &RuntimeWorkerRef, + ) -> Result<()> { + notification_routing::jobs::ack(self, workspace, id, target) + } + fn reserve_backend_job_delivery( &self, workspace_id: &str, @@ -14123,11 +14211,48 @@ fn migrate_notification_routing_v94_to_v95(conn: &Connection) -> Result<()> { Ok(()) } +fn migrate_job_notifications_v95_to_v96(conn: &Connection) -> Result<()> { + if current_schema_version(conn)? != 95 { + return Err(Error::Store( + "Job notification migration requires schema 95".into(), + )); + } + let foreign_keys: bool = conn.query_row("PRAGMA foreign_keys", [], |r| r.get(0))?; + let legacy_alter: bool = conn.query_row("PRAGMA legacy_alter_table", [], |r| r.get(0))?; + conn.execute_batch("PRAGMA foreign_keys=OFF; PRAGMA legacy_alter_table=ON;")?; + let result = (|| -> Result<()> { + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + notification_routing::jobs::install(&tx)?; + let violations: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM pragma_foreign_key_check)", + [], + |r| r.get(0), + )?; + if violations { + return Err(Error::Store( + "Job notification migration foreign-key violation".into(), + )); + } + tx.execute("INSERT INTO __yoi_schema_migrations(version,name) VALUES(96,'Backend Job common notification routing')",[])?; + tx.commit()?; + Ok(()) + })(); + let restore = conn + .execute_batch(&format!( + "PRAGMA legacy_alter_table={}; PRAGMA foreign_keys={};", + i32::from(legacy_alter), + i32::from(foreign_keys) + )) + .map_err(Error::from); + result.and(restore) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { create_workspace_schema_structure(conn)?; assignment_delivery::install(conn)?; assignment_delivery::install_post_inputs(conn)?; notification_routing::install(conn)?; + notification_routing::jobs::install(conn)?; ticket::verify_sqlite_ticket_schema(conn).map_err(|error| { Error::Store(format!( "fresh Workspace Ticket schema verification failed: {error}" @@ -14386,6 +14511,9 @@ fn verify_workspace_resource_constraints(conn: &Connection) -> Result<()> { if current_schema_version(conn)? >= 95 { notification_routing::verify(conn)?; } + if current_schema_version(conn)? >= 96 { + notification_routing::jobs::verify(conn)?; + } Ok(()) } @@ -15141,6 +15269,7 @@ mod tests { include!("assignment_tracking_tests.rs"); include!("assignment_delivery_tests.rs"); include!("notification_routing_tests.rs"); + include!("job_notification_tests.rs"); include!("store_workspace_config_tests.rs"); fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse { @@ -17354,6 +17483,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx WorkspaceSchemaMigrationStep {version:93, name:"Durable assignment deliveries and source-bound startup fences".into()}, WorkspaceSchemaMigrationStep {version:94, name:"Canonical Ticket posts".into()}, WorkspaceSchemaMigrationStep {version:95, name:"Common notification routing and durable Worker delivery".into()}, + WorkspaceSchemaMigrationStep {version:96, name:"Backend Job common notification routing".into()}, ] ); @@ -17459,6 +17589,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx (93, "Durable assignment deliveries and source-bound startup fences".to_string()), (94, "Canonical Ticket posts".to_string()), (95, "Common notification routing and durable Worker delivery".to_string()), +(96, "Backend Job common notification routing".to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -20078,7 +20209,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx (92, "Ticket-owned assignment tracking".into()), (93, "Durable assignment deliveries and source-bound startup fences".into()), (94, "Canonical Ticket posts".into()), - (95, "Common notification routing and durable Worker delivery".into())]); + (95, "Common notification routing and durable Worker delivery".into()), + (96, "Backend Job common notification routing".into())]); for (table, rows) in unchanged.iter().zip(before.iter()) { assert_eq!(&stored_fixture_rows(conn, table), rows, "preserve {table}"); } diff --git a/crates/workspace-server/src/ticket_item_checker.rs b/crates/workspace-server/src/ticket_item_checker.rs index 4b1e394fc..8fcd21117 100644 --- a/crates/workspace-server/src/ticket_item_checker.rs +++ b/crates/workspace-server/src/ticket_item_checker.rs @@ -13,9 +13,10 @@ //! //! Results are checked for unknown fields, duplicates, count/string limits, and quotes that are //! exact substrings of the checked title/body. Invalid results end as invalid_result. -//! Delivery rereads the Ticket and notifies the original Worker only of findings matching the -//! saved digest. No findings and stale results end delivery with distinct suppression reasons, -//! without redelivery on recovery. Advice is neither an approval gate nor implementation guidance; +//! Before routing, the adapter rereads the Ticket and reports findings/content freshness. +//! Common routing records clean/stale suppression separately from Runtime acceptance. Once +//! a request is pinned, recovery reuses those bytes even if the Ticket later changes. +//! Advice is neither an approval gate nor implementation guidance; //! no findings do not prove the entire specification correct. use std::collections::BTreeSet; @@ -217,18 +218,18 @@ fn validate_finding_text(field: &str, value: &str, max_bytes: usize) -> Result<( } #[derive(Debug, Clone, PartialEq, Eq)] -pub enum TicketItemCheckNotification { - Deliver(String), +pub enum TicketItemCheckFacts { + Findings(String), NoFindings, Stale, } -/// Decide whether a completed check should produce an advisory. Clean and stale -/// results remain distinct so delivery recovery can record why no input was sent. -pub fn notification( +/// Domain facts about validated results and the current checked content. +/// The common routing policy owns whether these facts produce an advisory. +pub fn result_facts( acceptance: &BackendJobResultAcceptance, current_ticket: &ticket::Ticket, -) -> Result { +) -> Result { let input = parse_input(&acceptance.job.request.input)?; let result = validate_result( &acceptance.job.request.input, @@ -239,12 +240,32 @@ pub fn notification( if input.content_digest != content_digest(current_ticket) || input.ticket_id != current_ticket.meta.id { - return Ok(TicketItemCheckNotification::Stale); + return Ok(TicketItemCheckFacts::Stale); } if result.findings.is_empty() { - return Ok(TicketItemCheckNotification::NoFindings); + return Ok(TicketItemCheckFacts::NoFindings); } + Ok(TicketItemCheckFacts::Findings(render_result(acceptance)?)) +} + +/// Stable rendering from the original checked snapshot. Used to pin pre-upgrade +/// unacknowledged requests without reinterpreting them against newer Ticket text. +pub(crate) fn render_result(acceptance: &BackendJobResultAcceptance) -> Result { + let input = parse_input(&acceptance.job.request.input)?; + let result = validate_result( + &acceptance.job.request.input, + acceptance + .job + .result + .as_ref() + .ok_or_else(|| Error::InvalidInput("completed checker has no result".into()))?, + )?; + if result.findings.is_empty() { + return Err(Error::InvalidInput( + "clean checker result has no advisory body".into(), + )); + } let mut content = format!( "[Ticket item checker advisory]\nTicket: {}\nContent digest: {}\n\nThis is a post-save advisory based only on the inspected Ticket text. It is not a new user request, approval gate, workflow blocker, or implementation instruction.\n", input.ticket_resource_key, input.content_digest @@ -262,7 +283,7 @@ pub fn notification( finding.suggestion )); } - Ok(TicketItemCheckNotification::Deliver(content)) + Ok(content) } #[cfg(test)] diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index ad2462434..c5ef75ff9 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -45,6 +45,8 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "repository_ssh_host_trust_keys", "repository_ssh_host_trusts", "server_secret_objects", + "notification_job_targets", + "notification_job_legacy_receipts", "notification_assignment_targets", "notification_acknowledgements", "notification_requests", @@ -489,6 +491,8 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore { // commit; update guards remain installed throughout. let mut ticket_delete_guards = Vec::new(); for (table, trigger) in [ + ("notification_job_targets", "notification_job_targets_no_delete"), + ("notification_job_legacy_receipts", "notification_job_legacy_no_delete"), ("notification_assignment_targets", "notification_assignment_targets_no_delete"), ("notification_sources", "notification_sources_no_delete"), ("notification_decisions", "notification_decisions_no_delete"),