fix(maestro): preserve post-load events
Switch replay suppression off inside the ordered session/load response callback, before the dispatcher can deliver the next live notification.
This commit is contained in:
@@ -456,15 +456,34 @@ async fn run_connection<T>(
|
||||
return Ok(());
|
||||
}
|
||||
replaying.store(true, Ordering::SeqCst);
|
||||
let loaded = cx
|
||||
let (loaded_tx, loaded_rx) = oneshot::channel();
|
||||
let replaying_done = replaying.clone();
|
||||
if let Err(e) = cx
|
||||
.send_request(LoadSessionRequest::new(existing.clone(), cwd))
|
||||
.block_task()
|
||||
.await;
|
||||
replaying.store(false, Ordering::SeqCst);
|
||||
if let Err(e) = loaded {
|
||||
.on_receiving_result(move |result| {
|
||||
replaying_done.store(false, Ordering::SeqCst);
|
||||
async move {
|
||||
let _ = loaded_tx.send(result);
|
||||
Ok(())
|
||||
}
|
||||
})
|
||||
{
|
||||
let _ = ready_tx.send(Err(ready_error(e)));
|
||||
return Ok(());
|
||||
}
|
||||
match loaded_rx.await {
|
||||
Ok(Ok(_)) => {}
|
||||
Ok(Err(e)) => {
|
||||
let _ = ready_tx.send(Err(ready_error(e)));
|
||||
return Ok(());
|
||||
}
|
||||
Err(_) => {
|
||||
let _ = ready_tx.send(Err(ReadyError::Handshake(
|
||||
"session/load response was not delivered".into(),
|
||||
)));
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
existing
|
||||
}
|
||||
None => match cx
|
||||
|
||||
@@ -125,6 +125,7 @@ for raw in sys.stdin:
|
||||
for i in range(int(os.environ.get("FAKE_ACP_LOAD_REPLAY", "0"))):
|
||||
send(chunk(f"replayed {i + 1}"))
|
||||
send({"jsonrpc": "2.0", "id": rid, "result": {}})
|
||||
send(chunk("hello after load"))
|
||||
elif method == "session/cancel":
|
||||
if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
|
||||
sys.exit(0)
|
||||
|
||||
@@ -301,16 +301,13 @@ async fn load_session_resumes_the_stored_id_and_drops_the_replay() {
|
||||
assert!(!log.contains("recv session/new"), "{log}");
|
||||
|
||||
// The three replayed chunks never reach the receiver: the event log
|
||||
// already holds them. A notification emitted after the load does.
|
||||
// already holds them. The first notification after the response does.
|
||||
let mut updates = session.take_updates().expect("updates");
|
||||
unsafe {
|
||||
libc::kill(session.pid() as libc::pid_t, libc::SIGUSR1);
|
||||
}
|
||||
let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
|
||||
.await
|
||||
.expect("update within 5s")
|
||||
.expect("channel open");
|
||||
assert_eq!(chunk_text(&first), "tick 1");
|
||||
assert_eq!(chunk_text(&first), "hello after load");
|
||||
// Nothing queued behind it either.
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
assert!(updates.try_recv().is_err(), "replay leaked through");
|
||||
|
||||
Reference in New Issue
Block a user