worker: persist child completion before wakeup
This commit is contained in:
@@ -272,17 +272,18 @@ impl ParentNotificationTarget {
|
|||||||
};
|
};
|
||||||
match self {
|
match self {
|
||||||
Self::Controller { sender, fallback } => {
|
Self::Controller { sender, fallback } => {
|
||||||
|
// Persist through the Worker-owned pending authority before the
|
||||||
|
// child completion path returns. The Controller delivery is only
|
||||||
|
// a wake-up hint; replay is idempotent by notification_request_id.
|
||||||
|
fallback(method.clone());
|
||||||
let Some(parent_method_tx) = sender.upgrade() else {
|
let Some(parent_method_tx) = sender.upgrade() else {
|
||||||
fallback(method);
|
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
let fallback = fallback.clone();
|
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
if let Err(error) = parent_method_tx.send(method).await {
|
if parent_method_tx.send(method).await.is_err() {
|
||||||
tracing::warn!(
|
tracing::warn!(
|
||||||
"failed to notify parent Controller; using durable pending authority"
|
"failed to wake parent Controller after durable notification acceptance"
|
||||||
);
|
);
|
||||||
fallback(error.0);
|
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -1201,6 +1202,30 @@ enabled = false
|
|||||||
assert!(*captured.lock().unwrap());
|
assert!(*captured.lock().unwrap());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn parent_controller_notification_is_durable_before_wakeup_delivery() {
|
||||||
|
let (parent_method_tx, mut parent_method_rx) = mpsc::channel(1);
|
||||||
|
let durable = Arc::new(AtomicBool::new(false));
|
||||||
|
let durable_for_fallback = Arc::clone(&durable);
|
||||||
|
let target = ParentNotificationTarget::with_controller_fallback(
|
||||||
|
parent_method_tx.downgrade(),
|
||||||
|
ParentNotificationTarget::Durable(Arc::new(move |_| {
|
||||||
|
durable_for_fallback.store(true, Ordering::Release);
|
||||||
|
})),
|
||||||
|
);
|
||||||
|
|
||||||
|
target.notify("child-session".into(), "completed".into());
|
||||||
|
let wakeup = parent_method_rx.recv().await.unwrap();
|
||||||
|
assert!(durable.load(Ordering::Acquire));
|
||||||
|
assert!(matches!(
|
||||||
|
wakeup,
|
||||||
|
Method::NotifyTracked {
|
||||||
|
source: protocol::AuthenticatedInputSource::SubWorker { session_id },
|
||||||
|
..
|
||||||
|
} if session_id == "child-session"
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn durable_parent_notification_target_preserves_child_source() {
|
fn durable_parent_notification_target_preserves_child_source() {
|
||||||
let captured = Arc::new(std::sync::Mutex::new(None));
|
let captured = Arc::new(std::sync::Mutex::new(None));
|
||||||
|
|||||||
Reference in New Issue
Block a user