fix: compose passive notify with queued submit

crates/worker/src/controller.rs

@@ -325,8 +325,13 @@ fn prepare_pending_run<St: Store + Clone>(
     notify_buffer: &NotifyBuffer,
     fence: Option<(u64, &str)>,
 ) -> Result<Option<PendingRun>, crate::worker::PendingSubmissionError> {
+    let staged_passive_notification = pending_submissions.activating_passive_notification_id();
     Ok(match pending_submissions.prepare_next_activation(fence)? {
         Some(crate::worker::PendingActivation::Submission(submission)) => {
+            if staged_passive_notification.is_some() {
+                let extension = pending_submissions.notification_activation_extension();
+                debug_assert!(notify_buffer.replace_durable_notification_extension(extension));
+            }
             Some(PendingRun::Submit(submission))
         }
         Some(crate::worker::PendingActivation::Notification(notification)) => {
@@ -2709,6 +2714,34 @@ mod tests {
                 ..
             }) if request_id == "notify-first"
         ));
+
+        let pending =
+            crate::worker::PendingSubmissionHandle::for_test(&temp.path().join("passive-first"));
+        pending
+            .accept_notification("passive-first".into(), "passive notification".into(), false)
+            .unwrap();
+        pending
+            .accept(
+                "submit-after-passive".into(),
+                vec![protocol::Segment::Text {
+                    content: "queued after passive".into(),
+                }],
+                false,
+            )
+            .unwrap();
+        let snapshot = pending.snapshot();
+        let head_id = snapshot.head_id.clone().expect("queued Submit is the head");
+        let notify_buffer = NotifyBuffer::new();
+        assert!(stage_oldest_passive_notification(&pending, &notify_buffer));
+        assert!(matches!(
+            prepare_pending_run(
+                &pending,
+                &notify_buffer,
+                Some((snapshot.revision + 1, &head_id)),
+            )
+            .unwrap(),
+            Some(PendingRun::Submit(_))
+        ));
     }
 
     #[test]

crates/worker/src/ipc/notify_buffer.rs

@@ -110,6 +110,25 @@ impl NotifyBuffer {
         });
     }
 
+    pub(crate) fn replace_durable_notification_extension(
+        &self,
+        extension: SessionExtension,
+    ) -> bool {
+        let mut queue = self.inner.lock().expect("notify buffer poisoned");
+        let Some(extensions) = queue.iter_mut().rev().find_map(|pending| match pending {
+            PendingNotify::Notify {
+                auto_run: false,
+                extensions,
+                ..
+            } if !extensions.is_empty() => Some(extensions),
+            _ => None,
+        }) else {
+            return false;
+        };
+        *extensions = vec![extension];
+        true
+    }
+
     /// Push a typed worker-event entry onto the queue.
     pub fn push_worker_event(&self, event: WorkerEvent) {
         self.push_entry(PendingNotify::WorkerEvent { event });

crates/worker/src/worker.rs

@@ -1674,14 +1674,24 @@ where
         if let Some((expected_revision, expected_head_id)) = fence {
             Self::validate_fence(&state, expected_revision, Some(expected_head_id))?;
         }
-        if state.activating.is_some() || state.activating_notification.is_some() {
+        if state.activating.is_some()
+            || state
+                .activating_notification
+                .as_ref()
+                .is_some_and(|notification| notification.auto_run)
+        {
             return Ok(None);
         }
+        let has_staged_passive_notification = state.activating_notification.is_some();
         let submission_sequence = state.pending.front().map(|item| item.activation_sequence);
-        let notification_index = state
+        let notification_index = if has_staged_passive_notification {
-            .pending_notifications
+            None
-            .iter()
+        } else {
-            .position(|item| item.auto_run);
+            state
+                .pending_notifications
+                .iter()
+                .position(|item| item.auto_run)
+        };
         let notification_sequence = notification_index
             .and_then(|index| state.pending_notifications.get(index))
             .map(|item| item.activation_sequence);
@@ -1785,6 +1795,7 @@ where
             .lock()
             .expect("pending activation state poisoned");
         let mut committed = state.clone();
+        committed.activating = None;
         committed.activating_notification = None;
         committed.revision = committed.revision.saturating_add(1);
         pending_activation_extension(&committed)