diff --git a/crates/workspace-server/src/job_notification_routing.rs b/crates/workspace-server/src/job_notification_routing.rs index 60ceadd46..650a654ab 100644 --- a/crates/workspace-server/src/job_notification_routing.rs +++ b/crates/workspace-server/src/job_notification_routing.rs @@ -20,6 +20,7 @@ pub enum JobNotificationFacts { Result(String), NoFindings, StaleContent, + /// SourceLookupFailed leaves routing undecided; other failures are terminal. Failed(NotificationFailureCode), } @@ -74,6 +75,23 @@ pub(in crate::store) fn install(conn: &Connection) -> Result<()> { 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;")?; + // Preparation may fail before a routing decision exists. Keep observations + // against the common source, never invent a terminal decision to hold them. + conn.execute_batch("CREATE TABLE notification_source_preparation_failures ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + source_sequence INTEGER NOT NULL REFERENCES notification_sources(sequence), + code TEXT NOT NULL, + recorded_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))); + CREATE INDEX notification_source_preparation_failures_source ON notification_source_preparation_failures(source_sequence,sequence); + CREATE TRIGGER notification_source_preparation_failures_no_update BEFORE UPDATE ON notification_source_preparation_failures BEGIN SELECT RAISE(ABORT,'Source preparation evidence is immutable'); END; + CREATE TRIGGER notification_source_preparation_failures_no_delete BEFORE DELETE ON notification_source_preparation_failures BEGIN SELECT RAISE(ABORT,'Source preparation evidence is immutable'); END;")?; + // Mutable scheduling only, independent of immutable source/decision evidence. + // Both undecided preparation retries and pinned deliveries use this clock. + conn.execute_batch( + "CREATE TABLE notification_source_offers ( + source_sequence INTEGER PRIMARY KEY REFERENCES notification_sources(sequence), + last_offered INTEGER NOT NULL CHECK(last_offered > 0));", + )?; // 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)?)))? @@ -112,15 +130,17 @@ pub(in crate::store) fn pending( ) -> 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")? + 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 LEFT JOIN notification_source_offers o ON o.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(o.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))?; + // Advance every offered source under the scan's write lock, even if no + // routing decision exists yet. Failure observation IDs are not a clock. + let tick:i64=tx.query_row("SELECT COALESCE(MAX(o.last_offered),0)+1 FROM notification_source_offers o JOIN notification_sources s ON s.sequence=o.source_sequence WHERE s.workspace_id=?1 AND s.role='backend_job'",[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.execute("INSERT INTO notification_source_offers(source_sequence,last_offered) VALUES(?1,?2) ON CONFLICT(source_sequence) DO UPDATE SET last_offered=excluded.last_offered",params![sequence,tick])?; } tx.commit()?; Ok(sources) }) @@ -172,6 +192,13 @@ pub(in crate::store) fn route( 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); } + if let JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)=facts { + // A transient checker read is not suppression or an immutable defer. + // With no decision, the existing pending scan retries the same source. + // Check existing decisions first: never replace pinned request evidence. + tx.execute("INSERT INTO notification_source_preparation_failures(source_sequence,code) SELECT sequence,?3 FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2 AND role='backend_job'",params![workspace,source.delivery_id,code(NotificationFailureCode::SourceLookupFailed)?])?; + tx.commit()?; return Ok(false); + } 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)?; } @@ -393,6 +420,11 @@ pub(in crate::store) fn authorize( pub(in crate::store) fn verify(conn: &Connection) -> Result<()> { for name in [ + "notification_source_offers", + "notification_source_preparation_failures", + "notification_source_preparation_failures_source", + "notification_source_preparation_failures_no_update", + "notification_source_preparation_failures_no_delete", "notification_job_identity", "notification_job_source_guard", "notification_job_targets", diff --git a/crates/workspace-server/src/job_notification_tests.rs b/crates/workspace-server/src/job_notification_tests.rs index b32ff9ec4..8c2975a63 100644 --- a/crates/workspace-server/src/job_notification_tests.rs +++ b/crates/workspace-server/src/job_notification_tests.rs @@ -102,14 +102,6 @@ async fn job_notification_routing_common_policy_distinguishes_suppression_failur "stale_content", None, ), - ( - "missing", - true, - JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed), - "defer", - "source_unavailable", - Some("source_lookup_failed"), - ), ( "invalid", true, @@ -201,6 +193,153 @@ async fn job_notification_routing_common_policy_distinguishes_suppression_failur ); } +#[tokio::test] +async fn job_notification_preparation_failures_remain_pending_and_append_only() { + let (_directory, store) = tracking_store_fixture().await; + for target in [true, false] { + let (_, source) = job_notice_fixture(&store, if target { "retry" } else { "retry-no-target" }, target); + for _ in 0..2 { + assert!(!store.route_job_notification("w", &source, &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)).unwrap()); + } + store.with_conn(|conn| { + let sequence: i64 = conn.query_row("SELECT sequence FROM notification_sources WHERE input_event_id=?1", [&source.delivery_id], |r| r.get(0))?; + assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_source_preparation_failures WHERE source_sequence=?1 AND code='source_lookup_failed'", [sequence], |r| r.get::<_, i64>(0))?, 2); + assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_decisions WHERE source_sequence=?1", [sequence], |r| r.get::<_, i64>(0))?, 0); + assert!(conn.execute("UPDATE notification_source_preparation_failures SET code='invalid_result' WHERE source_sequence=?1", [sequence]).is_err()); + assert!(conn.execute("DELETE FROM notification_source_preparation_failures WHERE source_sequence=?1", [sequence]).is_err()); + assert!(conn.execute("INSERT INTO notification_source_preparation_failures(source_sequence,code) VALUES(-1,'source_lookup_failed')", []).is_err()); + Ok(()) + }).unwrap(); + assert!(store.list_pending_job_notifications("w", 100).unwrap().iter().any(|s| s.delivery_id == source.delivery_id)); + assert_eq!(store.route_job_notification("w", &source, &JobNotificationFacts::Result("recovered body".into())).unwrap(), target); + let pinned = store.job_notification_request("w", &source.delivery_id).unwrap(); + // Later failed reads cannot overwrite the pinned decision/body/recipient + // or fabricate a new pre-decision observation after routing has finished. + assert_eq!(store.route_job_notification("w", &source, &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed)).unwrap(), target); + assert_eq!(store.job_notification_request("w", &source.delivery_id).unwrap(), pinned); + store.with_conn(|conn| { + assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_source_preparation_failures f JOIN notification_sources s ON s.sequence=f.source_sequence WHERE s.input_event_id=?1", [&source.delivery_id], |r| r.get::<_, i64>(0))?, 2); + Ok(()) + }).unwrap(); + } +} + +#[tokio::test] +async fn job_notification_fair_offers_survive_reopen_and_bound_preparation_and_ack_retries() { + let (directory, mut store) = tracking_store_fixture().await; + let mut failing = Vec::new(); + for i in 0..25 { + let (_, source) = job_notice_fixture(&store, &format!("failing-{i}"), true); + assert!(!store.route_job_notification( + "w", &source, + &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed), + ).unwrap()); + failing.push(source); + } + let (_, healthy) = job_notice_fixture(&store, "later-healthy", true); + let (_, retry) = job_notice_fixture(&store, "later-pinned-retry", true); + let target = TicketWorkerNotificationTarget { + worker: RuntimeWorkerRef::new("r", "replacement"), + authentication_binding_id: "tracking-binding".into(), + }; + assert!(store.route_job_notification( + "w", &retry, &JobNotificationFacts::Result("original pinned bytes".into()), + ).unwrap()); + assert_eq!(store.authorize_job_notification("w", &retry.delivery_id, &target).unwrap(), None); + let retry_key = NotificationDeliveryKey::WorkerNotification { + delivery_id: retry.delivery_id.clone(), + }; + store.record_notification_delivery_failure( + "w", &retry_key, NotificationFailureCode::AcceptanceUnknown, + ).unwrap(); + let pinned = store.job_notification_request("w", &retry.delivery_id).unwrap(); + // Start with an already-offered pinned retry, not only a never-offered result. + // All sources share the same offer clock; 25 failure observations are not ticks. + assert_eq!(store.list_pending_job_notifications("w", 100).unwrap().len(), 27); + let snapshot = |store: &SqliteWorkspaceStore, table: &str| { + store.with_conn(|conn| Ok(stored_fixture_rows(conn, table))).unwrap() + }; + let original_sources = snapshot(&store, "notification_sources"); + let original_decisions = snapshot(&store, "notification_decisions"); + let original_preparation_failures = snapshot(&store, "notification_source_preparation_failures"); + let original_delivery_failures = snapshot(&store, "notification_failures"); + let original_jobs = snapshot(&store, "backend_jobs"); + let mut failure_offers = vec![0; failing.len()]; + let mut healthy_round = None; + let mut retry_rounds = Vec::new(); + + for round in 1..=4 { + // Reopen before every production-sized round, including the first retry. + let saved_offers = snapshot(&store, "notification_source_offers"); + drop(store); + store = SqliteWorkspaceStore::open_for_test(directory.path().join("workspace.db")).unwrap(); + assert_eq!(snapshot(&store, "notification_source_offers"), saved_offers); + let pending = store.list_pending_job_notifications("w", 20).unwrap(); + assert_eq!(pending.len(), 20); + if round == 1 { + assert!(pending.iter().all(|s| s.job_id.starts_with("failing-"))); + } + for source in pending { + if let Some(index) = failing.iter().position(|s| s.delivery_id == source.delivery_id) { + failure_offers[index] += 1; + assert!(!store.route_job_notification( + "w", &source, + &JobNotificationFacts::Failed(NotificationFailureCode::SourceLookupFailed), + ).unwrap()); + } else if source.delivery_id == healthy.delivery_id { + assert!(healthy_round.replace(round).is_none()); + assert!(store.route_job_notification( + "w", &source, &JobNotificationFacts::Result("healthy body".into()), + ).unwrap()); + assert_eq!(store.authorize_job_notification("w", &source.delivery_id, &target).unwrap(), None); + store.ack_job_notification("w", &source.delivery_id, &target.worker).unwrap(); + } else { + assert_eq!(source.delivery_id, retry.delivery_id); + retry_rounds.push(round); + // Neither scheduling nor a new rendering may replace the request. + assert!(store.route_job_notification( + "w", &source, &JobNotificationFacts::Result("must not replace pinned bytes".into()), + ).unwrap()); + assert_eq!(store.job_notification_request("w", &source.delivery_id).unwrap(), pinned); + if retry_rounds.len() == 1 { + store.record_notification_delivery_failure( + "w", &retry_key, NotificationFailureCode::AcceptanceUnknown, + ).unwrap(); + } else { + store.ack_job_notification("w", &source.delivery_id, &target.worker).unwrap(); + } + } + } + if round == 2 { + assert!(failure_offers.iter().all(|count| *count >= 1)); + assert_eq!(healthy_round, Some(2)); + assert_eq!(retry_rounds, vec![2]); + } + } + // With 26 sources still pending after the healthy ACK, even the last source + // in an offer-time tie returns within ceil(26 / 20) = 2 further rounds. + assert_eq!(retry_rounds, vec![2, 4], "a pinned retry must not starve behind undecided sources"); + assert!(job_notice_evidence(&store, &healthy).2); + assert!(job_notice_evidence(&store, &retry).2); + assert_eq!(snapshot(&store, "notification_sources"), original_sources); + assert_eq!(snapshot(&store, "backend_jobs"), original_jobs); + let decisions = snapshot(&store, "notification_decisions"); + assert!(original_decisions.iter().all(|row| decisions.contains(row))); + let preparation_failures = snapshot(&store, "notification_source_preparation_failures"); + assert!(preparation_failures.starts_with(&original_preparation_failures)); + assert_eq!(preparation_failures.len(), 25 + failure_offers.iter().sum::()); + assert!(snapshot(&store, "notification_failures").starts_with(&original_delivery_failures)); + assert_eq!(store.job_notification_request("w", &retry.delivery_id).unwrap(), pinned); + assert!(store.list_pending_job_notifications("w", 20).unwrap().iter().all(|s| s.job_id.starts_with("failing-"))); + store.with_conn(|conn| { + assert!(conn.execute("INSERT INTO notification_source_offers VALUES(-1,1)", []).is_err()); + assert!(conn.execute("UPDATE notification_source_offers SET last_offered=0", []).is_err()); + assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_sources s JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.role='backend_job'", [], |r| r.get::<_, i64>(0))?, 2); + assert_eq!(conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r.get::<_, i64>(0))?, 0); + Ok(()) + }).unwrap(); +} + #[tokio::test] async fn job_notification_result_commit_reopen_parallel_recovery_and_fixed_target_receipt() { let (directory, store) = tracking_store_fixture().await; @@ -432,6 +571,46 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and migrate_assignment_delivery_v92_to_v93(&conn).unwrap(); migrate_ticket_posts_v93_to_v94(&conn).unwrap(); migrate_notification_routing_v94_to_v95(&conn).unwrap(); + // Populate the v95 common graph before rebuilding its source parent. Include + // both accepted and retryable Ticket requests, with retained failure history. + for id in ["ticket-ack", "ticket-pending"] { + conn.execute("INSERT INTO notification_sources(workspace_id,ticket_id,input_event_id,role,assignment_id,kind,facts_json) VALUES('w','pending',?1,'worker','ticket-worker-assignment','comment','{\"principal_kind\":\"worker\",\"principal_id\":null,\"runtime_id\":\"r\",\"worker_id\":\"worker\",\"status\":null,\"case_state\":\"open\",\"released\":false}')", [id]).unwrap(); + conn.execute("INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose,last_offered) SELECT sequence,'w',?1,'deliver','attention','attention',17 FROM notification_sources WHERE input_event_id=?1", [id]).unwrap(); + conn.execute("INSERT INTO notification_requests VALUES('w',?1,'original Ticket bytes','{\"worker\":{\"runtime_id\":\"r\",\"worker_id\":\"worker\"},\"authentication_binding_id\":\"original-binding\"}')", [id]).unwrap(); + conn.execute("INSERT INTO notification_failures(workspace_id,delivery_id,code,recorded_at) VALUES('w',?1,'acceptance_unknown','original-failure-time')", [id]).unwrap(); + } + conn.execute("INSERT INTO notification_acknowledgements SELECT workspace_id,delivery_id,target_json FROM notification_requests WHERE delivery_id='ticket-ack'", []).unwrap(); + let ticket_graph = |conn: &Connection| { + [ + "SELECT * FROM notification_sources WHERE role!='backend_job' ORDER BY sequence".to_string(), + "SELECT d.* FROM notification_decisions d JOIN notification_sources s ON s.sequence=d.source_sequence WHERE s.role!='backend_job' ORDER BY d.rowid".to_string(), + ].into_iter().chain(["notification_requests", "notification_acknowledgements", "notification_failures"].map(|table| format!("SELECT f.* FROM {table} f JOIN notification_decisions d USING(workspace_id,delivery_id) JOIN notification_sources s ON s.sequence=d.source_sequence WHERE s.role!='backend_job' ORDER BY f.rowid"))) + .map(|sql| fixture_query_rows(conn, &sql)).collect::>() + }; + let ticket_before = ticket_graph(&conn); + assert!(ticket_before.iter().all(|rows| !rows.is_empty())); + let schema_before = fixture_query_rows(&conn, "SELECT type,name,sql FROM sqlite_schema WHERE type IN ('trigger','index') AND sql IS NOT NULL ORDER BY name"); + let check_guards = |conn: &Connection| { + for sql in [ + "UPDATE notification_sources SET kind='changed' WHERE input_event_id='ticket-ack'", + "DELETE FROM notification_sources WHERE input_event_id='ticket-ack'", + "UPDATE notification_decisions SET reason='changed' WHERE delivery_id='ticket-ack'", + "DELETE FROM notification_decisions WHERE delivery_id='ticket-ack'", + "UPDATE notification_requests SET content='changed' WHERE delivery_id='ticket-ack'", + "DELETE FROM notification_acknowledgements WHERE delivery_id='ticket-ack'", + "UPDATE notification_failures SET code='changed' WHERE delivery_id='ticket-ack'", + "INSERT INTO notification_decisions(source_sequence,workspace_id,delivery_id,disposition,reason,purpose) VALUES(-1,'w','orphan','deliver','result','result')", + "INSERT INTO notification_requests VALUES('w','orphan','content','{}')", + "INSERT INTO notification_acknowledgements VALUES('w','orphan','{}')", + "INSERT INTO notification_failures(workspace_id,delivery_id,code) VALUES('w','orphan','acceptance_unknown')", + ] { + assert!(conn.execute_batch(sql).is_err(), "guard lost: {sql}"); + } + assert_eq!(conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r.get::<_, i64>(0)).unwrap(), 0); + assert_eq!(conn.query_row("PRAGMA legacy_alter_table", [], |r| r.get::<_, i64>(0)).unwrap(), 0); + assert_eq!(conn.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0)).unwrap(), 1); + }; + check_guards(&conn); let accepted = job_notice_legacy_seed(&conn, "ack", Some("completed"), None, true); let clean = job_notice_legacy_seed( &conn, @@ -492,6 +671,15 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and [checker_result.to_string()], ) .unwrap(); + // Multiple retained identities must never create a fresh send: an ACK wins + // over unknown history; without an ACK the ambiguity stays terminal. + let multi_ack = job_notice_legacy_seed(&conn, "multi-ack", Some("completed"), None, true); + let multi_unknown = job_notice_legacy_seed(&conn, "multi-unknown", Some("unknown"), Some("notification_outcome_unknown"), true); + for id in [&multi_ack, &multi_unknown] { + 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) SELECT workspace_id,'another-' || delivery_id,job_id,attempt_id,target_runtime_id,target_worker_id,'unknown','notification_outcome_unknown','0','0' FROM backend_job_deliveries WHERE delivery_id=?1", [id]).unwrap(); + } + // Earlier unknown beats the derived identity only when there is no real ACK. + let multi_unknown_retained = format!("another-{multi_unknown}"); 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. @@ -499,6 +687,9 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and .unwrap(); assert!(migrate_job_notifications_v95_to_v96(&conn).is_err()); assert_eq!(current_schema_version(&conn).unwrap(), 95); + assert_eq!(ticket_graph(&conn), ticket_before); + assert_eq!(fixture_query_rows(&conn, "SELECT type,name,sql FROM sqlite_schema WHERE type IN ('trigger','index') AND sql IS NOT NULL ORDER BY name"), schema_before); + check_guards(&conn); assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before); assert_eq!( conn.query_row("PRAGMA foreign_keys", [], |r| r.get::<_, i64>(0)) @@ -508,6 +699,11 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and conn.execute_batch("DROP TABLE notification_job_targets") .unwrap(); migrate_job_notifications_v95_to_v96(&conn).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 96); + assert_eq!(ticket_graph(&conn), ticket_before); + check_guards(&conn); + let schema_after = fixture_query_rows(&conn, "SELECT type,name,sql FROM sqlite_schema WHERE type IN ('trigger','index') AND sql IS NOT NULL ORDER BY name"); + assert!(schema_before.iter().all(|row| schema_after.contains(row))); assert_eq!(stored_fixture_rows(&conn, "backend_job_deliveries"), before); for (id, disposition, reason, runtime_accepted) in [ (&accepted, "deliver", "result", true), @@ -515,6 +711,8 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and (&stale, "suppress", "stale_content", false), (&unknown, "deliver", "result", false), (&ambiguous, "defer", "source_unavailable", false), + (&multi_ack, "deliver", "result", true), + (&multi_unknown_retained, "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)); @@ -526,9 +724,13 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and |r| r.get::<_, i64>(0) ) .unwrap(), - 0, - "never fabricate modern ACKs or request bodies for old receipts" + 1, + "retain the Ticket ACK, never fabricate modern ACKs for legacy Job receipts" ); + for id in [&multi_ack, &multi_unknown_retained] { + assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_requests WHERE delivery_id=?1", [id], |r| r.get::<_, i64>(0)).unwrap(), 0); + } + assert_eq!(conn.query_row("SELECT code FROM notification_failures WHERE delivery_id=?1", [&multi_unknown_retained], |r| r.get::<_, String>(0)).unwrap(), "preparation_failed"); assert_eq!( conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r .get::<_, i64>( @@ -547,6 +749,12 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and assert!(pinned.contains("Durable result digest: digest.")); drop(conn); let store = SqliteWorkspaceStore::open_for_test(path).unwrap(); + store.with_conn(|conn| { + assert_eq!(ticket_graph(conn), ticket_before); + assert_eq!(stored_fixture_rows(conn, "backend_job_deliveries"), before); + check_guards(conn); + Ok(()) + }).unwrap(); let pending = store.list_pending_job_notifications("w", 100).unwrap(); assert_eq!(pending.len(), 4); for (id,stage) in [(&accepted,server_api::NotificationDiagnosticStage::RuntimeAccepted),(&clean,server_api::NotificationDiagnosticStage::Suppressed),(&stale,server_api::NotificationDiagnosticStage::Suppressed),(&unknown,server_api::NotificationDiagnosticStage::AcceptanceUnknown),(&ambiguous,server_api::NotificationDiagnosticStage::Deferred)] { @@ -573,7 +781,10 @@ fn job_notification_migration_retains_legacy_acks_suppressions_unknown_bytes_and assert!(pending.iter().all(|s| s.delivery_id != accepted && s.delivery_id != clean && s.delivery_id != stale - && s.delivery_id != ambiguous)); + && s.delivery_id != ambiguous + && s.delivery_id != multi_ack + && s.delivery_id != multi_unknown_retained + && s.delivery_id != multi_unknown)); assert_eq!( store .job_notification_request("w", &unknown) diff --git a/crates/workspace-server/src/server_job_notification_tests.rs b/crates/workspace-server/src/server_job_notification_tests.rs index 0bc7eb637..acb19e2a5 100644 --- a/crates/workspace-server/src/server_job_notification_tests.rs +++ b/crates/workspace-server/src/server_job_notification_tests.rs @@ -123,6 +123,85 @@ async fn job_notification_real_worker_ack_loss_reopen_and_parallel_recovery_comm delivery_stop(&reopened, worker); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn job_notification_checker_read_failure_reopen_recovers_same_source_to_real_receipt() { + let dir = tempfile::tempdir().unwrap(); + let (api, execution, _) = delivery_real_fixture(dir.path(), false).await; + let target = spawn_ticket_check_source(&api, "checker-retry-target"); + let backend = browser_ticket_backend(&api).unwrap(); + let ticket_ref = backend.create(ticket::NewTicket::new("Writer-only scope in requirements")).unwrap(); + let ticket = backend.show(ticket_ref.id.into()).unwrap(); + let request = ticket_item_checker::request( + &ticket, + ticket_item_checker::TicketItemCheckEdit::Create, + target.clone(), + "Check saved Ticket content.".into(), + ).unwrap(); + 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!({"findings":[{"category":"writer_scope","quote":"Writer-only scope","reason":"Writer scope is mixed into requirements.","suggestion":"Remove the writer-specific sentence."}]}), + }; + let accepted = api.store.accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT).unwrap(); + let source = api.store.job_notification_source(TEST_WORKSPACE_ID, &request.job_id, &accepted.attempt.attempt_id).unwrap(); + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + let source_snapshot = || conn.query_row("SELECT sequence,facts_json FROM notification_sources WHERE workspace_id=?1 AND input_event_id=?2", rusqlite::params![TEST_WORKSPACE_ID,source.delivery_id], |r| Ok((r.get::<_,i64>(0)?,r.get::<_,String>(1)?))).unwrap(); + let original_source = source_snapshot(); + // Fail exactly one checker read through the real adapter's error branch. + // Its independent Ticket connection cannot open a directory as SQLite, while + // the already-open result store remains available to persist failure evidence. + let mut unavailable = api.clone(); + unavailable.config.database_path = dir.path().to_path_buf(); + notification_delivery::recover_job_notifications(&unavailable); + let failures = || { + let mut query = conn.prepare("SELECT sequence,source_sequence,code,recorded_at FROM notification_source_preparation_failures WHERE source_sequence=?1 ORDER BY sequence").unwrap(); + query.query_map([original_source.0], |r| Ok((r.get::<_,i64>(0)?,r.get::<_,i64>(1)?,r.get::<_,String>(2)?,r.get::<_,String>(3)?))).unwrap().collect::>>().unwrap() + }; + let failed_once = failures(); + assert_eq!(failed_once.len(), 1); + assert_eq!(failed_once[0].2, "source_lookup_failed"); + assert_eq!(conn.query_row("SELECT COUNT(*) FROM notification_decisions WHERE source_sequence=?1", [original_source.0], |r| r.get::<_,i64>(0)).unwrap(), 0); + assert!(api.store.job_notification_request(TEST_WORKSPACE_ID, &source.delivery_id).unwrap().is_none()); + assert!(delivery_receipts(&api, &target).is_empty()); + assert!(!execution.attempts.lock().unwrap().iter().any(|attempt| attempt.0 == source.delivery_id)); + drop(unavailable); + + // Reopen Backend persistence and run the unchanged pending-source scan. It + // rereads the real Ticket, obtains findings, then reaches the actual Runtime. + let mut reopened = api.clone(); + reopened.store = Arc::new(SqliteWorkspaceStore::open_for_test(&api.config.database_path).unwrap()); + let pending = reopened.store.list_pending_job_notifications(TEST_WORKSPACE_ID, 100).unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].delivery_id, source.delivery_id); + assert_eq!(pending[0].job_id, source.job_id); + assert_eq!(pending[0].attempt_id, source.attempt_id); + notification_delivery::recover_job_notifications(&reopened); + delivery_wait_idle(&reopened, &target, 1).await; + assert_eq!(job_notification_evidence(&reopened, &source.delivery_id), ("deliver".into(), "result".into(), true, None)); + let pinned = reopened.store.job_notification_request(TEST_WORKSPACE_ID, &source.delivery_id).unwrap().unwrap(); + assert_eq!(pinned.1, target); + assert!(pinned.0.contains("Quote: \"Writer-only scope\"")); + assert_eq!(delivery_commits(&reopened, &target), vec![pinned.0.clone()]); + assert_eq!(delivery_receipts(&reopened, &target).len(), 1); + assert!(execution.attempts.lock().unwrap().iter().any(|attempt| attempt.0 == source.delivery_id && attempt.1 == pinned.0)); + assert_eq!(source_snapshot(), original_source); + assert_eq!(failures(), failed_once, "recovery must retain the original failure observation"); + let job = reopened.store.get_backend_job(TEST_WORKSPACE_ID, &request.job_id).unwrap().unwrap(); + assert_eq!(job.state, BackendJobState::Completed); + assert_eq!(job.current_attempt, accepted.job.current_attempt); + assert_eq!(job.result, accepted.job.result); + assert_eq!(job.result_digest, accepted.job.result_digest); + assert!(reopened.store.accept_backend_job_result(TEST_WORKSPACE_ID, worker, &submission, TEST_CREATED_AT).unwrap().replayed); + notification_delivery::recover_job_notifications(&reopened); + assert_eq!(delivery_receipts(&reopened, &target).len(), 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();