diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index e39c337d..4de1743f 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -320,25 +320,6 @@ fn stage_oldest_passive_notification( }) } -fn prepare_restored_auto_notification( - pending_submissions: &crate::worker::PendingSubmissionHandle, - notify_buffer: &NotifyBuffer, -) -> Option { - let notification = pending_submissions.prepare_oldest_auto_notification()?; - let extension = pending_submissions.notification_activation_extension(); - let notification_request_id = notification.notification_request_id.clone(); - notify_buffer.push_durable_notify( - notification.message, - true, - notification.provenance, - extension, - ); - Some(PendingRun::RunForNotification { - invoke_kind: protocol::InvokeKind::Notify, - notification_request_id: Some(notification_request_id), - }) -} - fn prepare_pending_run( pending_submissions: &crate::worker::PendingSubmissionHandle, notify_buffer: &NotifyBuffer, @@ -1435,7 +1416,16 @@ async fn controller_loop( ); let pending_submissions = worker.pending_submission_handle(); stage_oldest_passive_notification(&pending_submissions, ¬ify_buffer); - let mut pending = prepare_restored_auto_notification(&pending_submissions, ¬ify_buffer); + let mut pending = match prepare_pending_run(&pending_submissions, ¬ify_buffer, None) { + Ok(pending) => pending, + Err(error) => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::Internal, + message: error.to_string(), + }); + None + } + }; loop { // Top-of-iteration: if an event handler staged a run, fire it @@ -2663,7 +2653,7 @@ mod tests { assert!(snapshot.head_id.is_some()); let notify_buffer = NotifyBuffer::new(); assert!(matches!( - prepare_restored_auto_notification(&pending, ¬ify_buffer), + prepare_pending_run(&pending, ¬ify_buffer, None).unwrap(), Some(PendingRun::RunForNotification { notification_request_id: Some(_), .. @@ -2672,6 +2662,55 @@ mod tests { assert!(notify_buffer.has_auto_run_pending()); } + #[test] + fn restored_mixed_activations_preserve_global_fifo_order() { + let temp = TempDir::new().unwrap(); + let pending = + crate::worker::PendingSubmissionHandle::for_test(&temp.path().join("submit-first")); + pending + .accept( + "submit-first".into(), + vec![protocol::Segment::Text { + content: "queued submit".into(), + }], + false, + ) + .unwrap(); + pending + .accept_notification("notify-second".into(), "newer notification".into(), true) + .unwrap(); + let notify_buffer = NotifyBuffer::new(); + + assert!(matches!( + prepare_pending_run(&pending, ¬ify_buffer, None).unwrap(), + Some(PendingRun::Submit(_)) + )); + + let pending = + crate::worker::PendingSubmissionHandle::for_test(&temp.path().join("notify-first")); + pending + .accept_notification("notify-first".into(), "older notification".into(), true) + .unwrap(); + pending + .accept( + "submit-second".into(), + vec![protocol::Segment::Text { + content: "newer submit".into(), + }], + false, + ) + .unwrap(); + let notify_buffer = NotifyBuffer::new(); + + assert!(matches!( + prepare_pending_run(&pending, ¬ify_buffer, None).unwrap(), + Some(PendingRun::RunForNotification { + notification_request_id: Some(request_id), + .. + }) if request_id == "notify-first" + )); + } + #[test] fn image_attachment_gate_requires_vision_and_supported_openai_scheme() { let openai = manifest::ModelManifest { diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 67c7f7ed..2a0347c3 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -1657,27 +1657,6 @@ where Some(notification) } - pub(crate) fn prepare_oldest_auto_notification(&self) -> Option { - let mut state = self - .state - .lock() - .expect("pending activation state poisoned"); - if state.activating.is_some() || state.activating_notification.is_some() { - return None; - } - let index = state - .pending_notifications - .iter() - .position(|notification| notification.auto_run)?; - let notification = state - .pending_notifications - .remove(index) - .expect("located auto-run notification must exist"); - state.activating_notification = Some(notification.clone()); - state.revision = state.revision.saturating_add(1); - Some(notification) - } - pub(crate) fn prepare_next_activation( &self, fence: Option<(u64, &str)>,