fix(maestro): classify exited ACP children
This commit is contained in:
@@ -158,6 +158,7 @@ enum Command {
|
||||
|
||||
enum ChildCommand {
|
||||
Signal(libc::c_int),
|
||||
Reap(oneshot::Sender<bool>),
|
||||
}
|
||||
|
||||
/// Why the handshake did not produce a session id.
|
||||
@@ -206,6 +207,18 @@ impl AcpSession {
|
||||
self.exit.borrow().is_some()
|
||||
}
|
||||
|
||||
/// Reap the child if it has already exited.
|
||||
pub async fn reap_if_exited(&self) -> bool {
|
||||
if self.has_exited() {
|
||||
return true;
|
||||
}
|
||||
let (sent, received) = oneshot::channel();
|
||||
if self.child_commands.send(ChildCommand::Reap(sent)).is_err() {
|
||||
return self.has_exited();
|
||||
}
|
||||
received.await.unwrap_or_else(|_| self.has_exited())
|
||||
}
|
||||
|
||||
/// Wait for the child to exit.
|
||||
pub async fn wait(&self) -> Option<ExitStatus> {
|
||||
wait_exit(&self.exit).await
|
||||
@@ -392,6 +405,21 @@ async fn supervise_child(
|
||||
// so its PID cannot be reused in that window.
|
||||
unsafe { libc::kill(pid as libc::pid_t, sig) };
|
||||
}
|
||||
Some(ChildCommand::Reap(done)) => match child.try_wait() {
|
||||
Ok(Some(status)) => {
|
||||
let _ = done.send(true);
|
||||
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
|
||||
let _ = exit_tx.send(Some(status));
|
||||
return;
|
||||
}
|
||||
Ok(None) => {
|
||||
let _ = done.send(false);
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::warn!(pid, %error, "could not reap {ADAPTER_BIN}");
|
||||
let _ = done.send(false);
|
||||
}
|
||||
},
|
||||
None => {
|
||||
// The session vanished without a command receiver. This
|
||||
// task still owns the unreaped child, so this cannot hit
|
||||
|
||||
@@ -379,8 +379,26 @@ impl SessionManager {
|
||||
if children.is_empty() {
|
||||
return;
|
||||
}
|
||||
tracing::info!(count = children.len(), "shutdown: ending adapter children");
|
||||
let terminations = children
|
||||
let reaped =
|
||||
futures_util::future::join_all(children.iter().map(|(_, acp, _)| acp.reap_if_exited()))
|
||||
.await;
|
||||
let mut natural_exits = Vec::new();
|
||||
let mut live_children = Vec::new();
|
||||
for (child, exited) in children.iter().cloned().zip(reaped) {
|
||||
if exited {
|
||||
natural_exits.push(child.0);
|
||||
} else {
|
||||
live_children.push(child);
|
||||
}
|
||||
}
|
||||
for id in natural_exits {
|
||||
mark_session_dead_with_worktree_settlement(&self.inner, id, true).await;
|
||||
}
|
||||
tracing::info!(
|
||||
count = live_children.len(),
|
||||
"shutdown: ending adapter children"
|
||||
);
|
||||
let terminations = live_children
|
||||
.iter()
|
||||
.map(|(_, acp, _)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
|
||||
futures_util::future::join_all(terminations).await;
|
||||
@@ -389,7 +407,7 @@ impl SessionManager {
|
||||
.filter_map(|(_, _, done)| done.clone())
|
||||
.map(wait_for_event_pump);
|
||||
futures_util::future::join_all(drains).await;
|
||||
for (id, _, _) in children {
|
||||
for (id, _, _) in live_children {
|
||||
mark_session_dead_after_shutdown(&self.inner, id).await;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -504,6 +504,44 @@ async fn shutdown_reaps_every_adapter_child() {
|
||||
assert_eq!(sidecar.status, SessionStatus::Dead);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shutdown_does_not_recover_an_adapter_that_already_exited() {
|
||||
let fx = DaemonFixture::new();
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let mut env = fake.env();
|
||||
env.insert("FAKE_ACP_EXIT_AFTER_NEW".into(), "1".into());
|
||||
|
||||
let session = fx
|
||||
.manager
|
||||
.create(structured(cwd.path(), env))
|
||||
.await
|
||||
.expect("create");
|
||||
wait_for_exit(fake.pid()).await;
|
||||
|
||||
fx.manager.shutdown().await;
|
||||
assert!(
|
||||
!fx.sidecar_dir()
|
||||
.join(format!("{}.recover", session.id))
|
||||
.exists(),
|
||||
"a naturally exited adapter is not recoverable"
|
||||
);
|
||||
|
||||
let revived = SessionManager::new(
|
||||
Arc::new(fx.tmux.tmux.clone()),
|
||||
SidecarStore::new(fx.sidecar_dir()),
|
||||
WorktreeManager::new(None),
|
||||
fx.fifo_dir(),
|
||||
);
|
||||
assert_eq!(revived.recover().await.expect("recover"), 1);
|
||||
assert_eq!(revived.status(session.id).await, Some(SessionStatus::Dead));
|
||||
assert!(
|
||||
!fake.log().contains("recv session/load"),
|
||||
"a naturally dead adapter must not be resumed: {}",
|
||||
fake.log()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
|
||||
let fx = DaemonFixture::new();
|
||||
|
||||
Reference in New Issue
Block a user