fix(maestro): serialize shutdown finalization

This commit is contained in:
Miguel Palhas
2026-09-05 00:07:19 +01:00
parent 88a40c1ea7
commit 9d7ed8c071
+29 -3
View File
@@ -7,7 +7,7 @@
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -195,6 +195,9 @@ struct Inner {
/// client that reconnects never collides with the registration its
/// previous socket has not finished releasing.
next_viewport: AtomicU64,
/// Once daemon shutdown starts, it owns structured-session finalization.
/// Background passes must not classify those deliberate exits as natural.
shutting_down: AtomicBool,
}
impl SessionManager {
@@ -237,6 +240,7 @@ impl SessionManager {
events,
viewports: Mutex::new(Viewports::new()),
next_viewport: AtomicU64::new(0),
shutting_down: AtomicBool::new(false),
}),
}
}
@@ -349,13 +353,17 @@ impl SessionManager {
/// is on disk after the restart is truthful. tmux sessions are left
/// alone — surviving a restart is what they are for.
pub async fn shutdown(&self) {
if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
return;
}
let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
let sessions = self.inner.sessions.lock().await;
sessions
.iter()
.filter_map(|(id, r)| {
r.acp
.clone()
(r.session.status == SessionStatus::Running)
.then_some(())
.and_then(|()| r.acp.clone())
.map(|acp| (*id, acp, r.acp_event_pump.clone()))
})
.collect()
@@ -1438,6 +1446,9 @@ async fn probe_worktree(
/// One pass of the liveness watcher: diff the in-memory session set against
/// the tmux-reported one, and reap anything we still track but tmux has lost.
async fn run_liveness_pass(inner: &Arc<Inner>) {
if inner.shutting_down.load(Ordering::Acquire) {
return;
}
// Snapshot the in-memory ids first; if we own no sessions there is
// nothing to reap and no reason to spawn a tmux subprocess (this also
// avoids racing with the very first `new-session` that would otherwise
@@ -1497,6 +1508,9 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
/// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
/// steady state produces no diffs. Runs on the existing liveness tick.
async fn run_activity_pass(inner: &Arc<Inner>) {
if inner.shutting_down.load(Ordering::Acquire) {
return;
}
let ids: Vec<Uuid> = {
let sessions = inner.sessions.lock().await;
if sessions.is_empty() {
@@ -1539,6 +1553,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
let updated = {
let mut sessions = inner.sessions.lock().await;
if inner.shutting_down.load(Ordering::Acquire) {
return;
}
let Some(rec) = sessions.get_mut(&id) else {
continue;
};
@@ -1613,6 +1630,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
/// error must not read as "nothing blocks". Like the activity pass, this only
/// emits `SessionUpdated` when the value actually changed.
async fn run_blocked_work_pass(inner: &Arc<Inner>) {
if inner.shutting_down.load(Ordering::Acquire) {
return;
}
let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
let sessions = inner.sessions.lock().await;
sessions
@@ -1640,6 +1660,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
};
let updated = {
let mut sessions = inner.sessions.lock().await;
if inner.shutting_down.load(Ordering::Acquire) {
return;
}
let Some(rec) = sessions.get_mut(&id) else {
continue;
};
@@ -1672,6 +1695,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
/// [`SessionManager::recover`] adopts only what tmux still holds. Structured
/// sessions stopped for daemon shutdown retain their sidecars for ACP recovery.
async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
if inner.shutting_down.load(Ordering::Acquire) {
return;
}
mark_session_dead_with_worktree_settlement(inner, id, true).await;
}