fix: preserve pending activation order on restore
This commit is contained in:
@@ -320,25 +320,6 @@ fn stage_oldest_passive_notification<St: Store + Clone>(
|
||||
})
|
||||
}
|
||||
|
||||
fn prepare_restored_auto_notification<St: Store + Clone>(
|
||||
pending_submissions: &crate::worker::PendingSubmissionHandle<St>,
|
||||
notify_buffer: &NotifyBuffer,
|
||||
) -> Option<PendingRun> {
|
||||
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<St: Store + Clone>(
|
||||
pending_submissions: &crate::worker::PendingSubmissionHandle<St>,
|
||||
notify_buffer: &NotifyBuffer,
|
||||
@@ -1435,7 +1416,16 @@ async fn controller_loop<C, St>(
|
||||
);
|
||||
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 {
|
||||
|
||||
@@ -1657,27 +1657,6 @@ where
|
||||
Some(notification)
|
||||
}
|
||||
|
||||
pub(crate) fn prepare_oldest_auto_notification(&self) -> Option<PendingNotification> {
|
||||
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)>,
|
||||
|
||||
Reference in New Issue
Block a user