From c82113f7c48ad62381aacb924080b7057c4fb388 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 00:53:31 +0900 Subject: [PATCH] fix: scope Ticket notification scans to each assigned Subject --- .../src/assignment_delivery.rs | 19 +- .../src/notification_routing.rs | 33 ++- crates/workspace-server/src/server.rs | 16 +- ...ver_subject_notification_recovery_tests.rs | 230 ++++++++++++++++++ crates/workspace-server/src/store.rs | 23 +- 5 files changed, 301 insertions(+), 20 deletions(-) create mode 100644 crates/workspace-server/src/server_subject_notification_recovery_tests.rs diff --git a/crates/workspace-server/src/assignment_delivery.rs b/crates/workspace-server/src/assignment_delivery.rs index 793408f72..c280a0303 100644 --- a/crates/workspace-server/src/assignment_delivery.rs +++ b/crates/workspace-server/src/assignment_delivery.rs @@ -141,16 +141,23 @@ pub(super) fn list_in_tx( conn: &Connection, workspace_id: &str, limit: usize, + subject_id: Option<&str>, ) -> Result> { if !(1..=100).contains(&limit) { return Err(Error::InvalidInput("Delivery limit must be 1..100".into())); } - notification_routing::pending_sources(conn, workspace_id, "orchestrator", limit)? - .into_iter() - .map(|(ticket, assignment, input, id)| { - delivery(conn, workspace_id, ticket, assignment, input, id) - }) - .collect() + notification_routing::pending_sources_for_subject( + conn, + workspace_id, + "orchestrator", + limit, + subject_id, + )? + .into_iter() + .map(|(ticket, assignment, input, id)| { + delivery(conn, workspace_id, ticket, assignment, input, id) + }) + .collect() } fn delivery( conn: &Connection, diff --git a/crates/workspace-server/src/notification_routing.rs b/crates/workspace-server/src/notification_routing.rs index 2a4c8fcc7..ebda0aac5 100644 --- a/crates/workspace-server/src/notification_routing.rs +++ b/crates/workspace-server/src/notification_routing.rs @@ -772,18 +772,33 @@ pub(super) fn pending_sources( workspace: &str, role: &str, limit: usize, +) -> Result> { + pending_sources_for_subject(conn, workspace, role, limit, None) +} + +// Scope before LIMIT and offer-order mutation: one Subject must not consume +// another Subject's batch while the Host resolves their current bodies separately. +pub(super) fn pending_sources_for_subject( + conn: &Connection, + workspace: &str, + role: &str, + limit: usize, + subject_id: Option<&str>, ) -> Result> { route_pending(conn, workspace)?; - let mut stmt=conn.prepare("SELECT s.ticket_id,s.assignment_id,s.input_event_id,d.delivery_id FROM notification_sources s JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND s.role=?2 AND d.disposition='deliver' AND NOT EXISTS(SELECT 1 FROM ticket_assignment_delivery_acknowledgements a WHERE a.workspace_id=?1 AND a.delivery_id=d.delivery_id) AND NOT EXISTS(SELECT 1 FROM notification_acknowledgements a WHERE a.workspace_id=?1 AND a.delivery_id=d.delivery_id) ORDER BY d.last_offered,s.sequence LIMIT ?3")?; + let mut stmt=conn.prepare("SELECT s.ticket_id,s.assignment_id,s.input_event_id,d.delivery_id FROM notification_sources s JOIN notification_decisions d ON d.source_sequence=s.sequence WHERE s.workspace_id=?1 AND s.role=?2 AND d.disposition='deliver' AND (?4 IS NULL OR EXISTS(SELECT 1 FROM ticket_worker_assignments a WHERE a.workspace_id=s.workspace_id AND a.ticket_id=s.ticket_id AND a.assignment_id=s.assignment_id AND a.role=s.role AND a.principal_kind='subject' AND a.principal_id=?4)) AND NOT EXISTS(SELECT 1 FROM ticket_assignment_delivery_acknowledgements a WHERE a.workspace_id=?1 AND a.delivery_id=d.delivery_id) AND NOT EXISTS(SELECT 1 FROM notification_acknowledgements a WHERE a.workspace_id=?1 AND a.delivery_id=d.delivery_id) ORDER BY d.last_offered,s.sequence LIMIT ?3")?; let candidates = stmt - .query_map(params![workspace, role, limit.clamp(1, 100) as i64], |r| { - Ok(( - r.get::<_, String>(0)?, - r.get::<_, String>(1)?, - r.get::<_, String>(2)?, - r.get::<_, String>(3)?, - )) - })? + .query_map( + params![workspace, role, limit.clamp(1, 100) as i64, subject_id], + |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, String>(1)?, + r.get::<_, String>(2)?, + r.get::<_, String>(3)?, + )) + }, + )? .collect::>>()?; let tick: i64 = conn.query_row( "SELECT COALESCE(MAX(last_offered),0)+1 FROM notification_decisions WHERE workspace_id=?1", diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 1fb784e83..0c7284053 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -22572,10 +22572,15 @@ fn check_orchestrator_assignment_attention( state: &Arc>, pending: Option, ) { - let inputs = match api - .store - .list_pending_ticket_assignment_deliveries(&api.config.workspace_id, 20) - { + let subject_id = match current_assignment_subject_source(api, worker) { + Ok(subject_id) => subject_id, + Err(_) => return, + }; + let inputs = match api.store.list_pending_subject_ticket_assignment_deliveries( + &api.config.workspace_id, + &subject_id, + 20, + ) { Ok(inputs) => inputs, Err(error) => { tracing::warn!(%error, "assignment notification scan failed"); @@ -59336,6 +59341,9 @@ mod tests { include!("server_assignment_test_support.rs"); include!("server_assignment_attention_tests.rs"); + mod subject_notification_recovery_tests { + include!("server_subject_notification_recovery_tests.rs"); + } #[tokio::test] async fn runtime_worker_create_rejects_reserved_singleton_key_squatting_and_authority_forgery() diff --git a/crates/workspace-server/src/server_subject_notification_recovery_tests.rs b/crates/workspace-server/src/server_subject_notification_recovery_tests.rs new file mode 100644 index 000000000..2b17c84b8 --- /dev/null +++ b/crates/workspace-server/src/server_subject_notification_recovery_tests.rs @@ -0,0 +1,230 @@ +// Included as a child module of server::tests by the parent change. +// Component regression: real Subject connections/singleton ownership, policy grants, +// Ticket transactions, routing, recovery, request persistence and ACK admission. +// DeterministicExecutionBackend records Runtime inputs instead of running a real +// Controller/model. Runtime acceptance here is NOT live-model processing evidence. +use super::*; +use std::collections::{BTreeMap, BTreeSet}; + +#[tokio::test] +async fn subject_notification_recovery_drains_two_recipients_beyond_global_scan_limit() { + let dir = tempfile::tempdir().unwrap(); + init_clean_git_workspace(dir.path()); + let (api, execution) = assignment_test_api_with_recording_backend(dir.path()).await; + seed_test_delivery_binding(&api); + let subject_store = open_subjektiv_store(&api).unwrap(); + let mut recipients = Vec::new(); + for role in ["recovery-alpha", "recovery-beta"] { + let subject = subject_store + .create_subject_with_behavior( + crate::subjektiv::SubjectRole::new(role).unwrap(), + "Handle assigned Tickets; delivery does not prove processing.".into(), + ) + .unwrap(); + // Normal Subject-connected creation, not a synthetic worker registry row + // or a monkey-patched assigned_subject_body/candidate selector. + let body = create_test_assignment_body(&api, EMBEDDED_WORKER_RUNTIME_ID, &subject.id).await; + bootstrap_test_assignment_session(&api, &body); + assert_eq!( + current_assignment_subject_source(&api, &body).unwrap(), + subject.id + ); + recipients.push((body, subject.id)); + } + // Recovery dispatches recipients in BTreeSet<(runtime_id, worker_id)> order. + // UUIDs need not have a predictable creation order: derive alpha/beta from + // the ACTUAL body keys, then commit beta's entire block before alpha's. + recipients.sort_by_key(|(body, _)| (body.runtime_id.clone(), body.worker_id.clone())); + let [(alpha, alpha_subject), (beta, beta_subject)] = recipients.as_slice() else { + unreachable!() + }; + assert_ne!(alpha_subject, beta_subject); + assert_ne!(alpha, beta); + assert!(execution.take_inputs().is_empty()); + assert!(execution.take_input_request_ids().is_empty()); + + let mut tickets = BTreeMap::new(); + for (name, body, subject) in [ + ("beta", beta, beta_subject), + ("alpha", alpha, alpha_subject), + ] { + for index in 0..20 { + let ticket = browser_ticket_backend(&api) + .unwrap() + .create(ticket::NewTicket::new(format!( + "Recovery {name} {index:02}" + ))) + .unwrap(); + grant_test_assignment_ticket(&api, subject, &ticket.id); + // Commit through the production store transaction, but do not fire a + // mutation wakeup: ONLY production recovery may drain this backlog. + api.store + .set_current_ticket_role_assignment( + &TicketRoleAssignmentRecord { + workspace_id: TEST_WORKSPACE_ID.into(), + ticket_id: ticket.id.clone(), + assignment_id: format!("recovery-{name}-{index:02}"), + role: TicketAssignmentRole::Orchestrator, + principal: TicketAssignmentPrincipal::Subject { + subject_id: subject.clone(), + }, + assigned_by: "fixture-owner".into(), + assigned_at: TEST_CREATED_AT.into(), + }, + None, + &format!("recovery-{name}-{index:02}-event"), + &format!("recovery-{name}-{index:02}-assign"), + false, + ) + .unwrap(); + tickets.insert(ticket.id.clone(), (body.clone(), subject.clone())); + } + } + let pending = api + .store + .list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100) + .unwrap(); + assert_eq!( + pending.len(), + 40, + "all forty inputs must be current and eligible" + ); + assert_eq!( + pending + .iter() + .filter(|input| tickets[&input.ticket_id].0 == *alpha) + .count(), + 20 + ); + assert_eq!( + pending + .iter() + .filter(|input| tickets[&input.ticket_id].0 == *beta) + .count(), + 20 + ); + assert!( + pending[..20] + .iter() + .all(|input| tickets[&input.ticket_id].0 == *beta), + "the global first page must belong to the recipient dispatched second" + ); + let expected = pending + .into_iter() + .map(|input| (input.request_id.clone(), input)) + .collect::>(); + assert_eq!(expected.len(), 40); + for input in expected.values() { + let (body, _) = &tickets[&input.ticket_id]; + assert_eq!( + assigned_subject_body(&api, &input.ticket_id, &input.assignment_id) + .unwrap() + .as_ref(), + Some(body) + ); + authorize_assignment_delivery(&api, body, &input.ticket_id).unwrap(); + } + assert!( + execution.take_inputs().is_empty(), + "fixture setup must not pre-deliver" + ); + + let mut observed = BTreeMap::new(); + let mut per_body = BTreeMap::::new(); + let mut progress = Vec::new(); + for _ in 0..5 { + // Exercise the actual global discovery -> recipient selection -> scoped + // attention -> durable reserve/render/admit -> Runtime -> ACK pipeline. + notification_delivery::recover_ticket_notifications(&api); + let inputs = execution.take_inputs(); + let request_ids = execution.take_input_request_ids(); + assert_eq!(inputs.len(), request_ids.len()); + for ((worker, content), request_id) in inputs.into_iter().zip(request_ids) { + let request_id = request_id.expect("recovery must supply a stable request identity"); + let input = expected + .get(&request_id) + .expect("unexpected Runtime request"); + let (body, _) = &tickets[&input.ticket_id]; + assert_eq!( + worker.worker_id.to_string(), + body.worker_id, + "cross-Subject delivery" + ); + assert!(content.contains(&input.assignment_id)); + assert!(content.contains(&input.input_event_id)); + *per_body.entry(body.worker_id.clone()).or_default() += 1; + assert!( + observed.insert(request_id, content).is_none(), + "ACKed input sent twice" + ); + } + progress.push(( + *per_body.get(&alpha.worker_id).unwrap_or(&0), + *per_body.get(&beta.worker_id).unwrap_or(&0), + )); + } + // Before the recipient-scoped scan fix, 26cc5e930's global scan offers all + // 40, alpha scans/offers beta's 20 then post-filters to empty; beta scans + // alpha's 20 and also filters to empty. Every recovery repeats (0, 0). + assert_eq!( + progress.last(), + Some(&(20, 20)), + "both connected Subjects must drain, progress by recovery round: {progress:?}" + ); + assert_eq!(observed.len(), 40); + assert!( + api.store + .list_pending_ticket_assignment_deliveries(TEST_WORKSPACE_ID, 100) + .unwrap() + .is_empty() + ); + + // Inspect actual persisted receipts, not a reimplementation of selection SQL. + let conn = rusqlite::Connection::open(&api.config.database_path).unwrap(); + let mut acknowledged = BTreeSet::new(); + for (request_id, input) in &expected { + let (body, subject) = &tickets[&input.ticket_id]; + let (content, source): (String, String) = conn.query_row( + "SELECT r.content,a.source_json FROM ticket_assignment_delivery_requests r JOIN ticket_assignment_delivery_acknowledgements a ON a.workspace_id=r.workspace_id AND a.delivery_id=r.delivery_id WHERE r.workspace_id=?1 AND r.delivery_id=?2", + rusqlite::params![TEST_WORKSPACE_ID, input.delivery_id], + |row| Ok((row.get(0)?, row.get(1)?)), + ).unwrap(); + let source: Value = serde_json::from_str(&source).unwrap(); + assert_eq!(source["subject_id"], *subject); + assert_eq!(source["worker_id"], body.worker_id); + assert_eq!(source["runtime_id"], body.runtime_id); + assert_eq!( + source["authentication_binding_id"], + orchestrator_delivery_target(&api, body) + .unwrap() + .authentication_binding_id + ); + assert_eq!( + &content, &observed[request_id], + "persisted request bytes differ from Runtime input" + ); + acknowledged.insert(input.delivery_id.clone()); + } + assert_eq!(acknowledged.len(), 40); + let ack_count: usize = conn.query_row( + "SELECT count(*) FROM ticket_assignment_delivery_acknowledgements WHERE workspace_id=?1", + [TEST_WORKSPACE_ID], |row| row.get(0), + ).unwrap(); + assert_eq!(ack_count, 40); + + // Lose the process-local attention cache, then recover repeatedly. SQLite + // ACKs, not local accepted/pending memory or a fake Runtime deduplicator, + // must close new-send admission for every delivered input. + for (body, _) in &recipients { + *orchestrator_attention_state(&api, body).lock().unwrap() = + OrchestratorAttentionState::default(); + } + for _ in 0..3 { + notification_delivery::recover_ticket_notifications(&api); + assert!( + execution.take_inputs().is_empty(), + "recovery resent ACKed history" + ); + assert!(execution.take_input_request_ids().is_empty()); + } +} diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 701950a1b..255ee9b20 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -2262,6 +2262,13 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { workspace_id: &str, limit: usize, ) -> Result>; + /// Recipient-scoped scan; other Subjects cannot consume this batch or its offer order. + fn list_pending_subject_ticket_assignment_deliveries( + &self, + workspace_id: &str, + subject_id: &str, + limit: usize, + ) -> Result>; /// Frozen payload/request identity for exact response-loss retry. fn reserve_ticket_assignment_delivery( &self, @@ -9455,7 +9462,21 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } self.with_conn_mut(|conn| { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; - let result = assignment_delivery::list_in_tx(&tx, workspace_id, limit)?; + let result = assignment_delivery::list_in_tx(&tx, workspace_id, limit, None)?; + tx.commit()?; + Ok(result) + }) + } + fn list_pending_subject_ticket_assignment_deliveries( + &self, + workspace_id: &str, + subject_id: &str, + limit: usize, + ) -> Result> { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let result = + assignment_delivery::list_in_tx(&tx, workspace_id, limit, Some(subject_id))?; tx.commit()?; Ok(result) })