From ae3091e2a2e53fc32c08017994292833e5abc317 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Fri, 4 Sep 2026 09:19:59 +0100 Subject: [PATCH] fix(maestro): guard resume against kill and second resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The respawn runs with the session map unlocked, so two takes of the lock straddle it. A kill landing in that window deleted the sidecar before resume wrote it back, leaving a file for a session that was gone — the next recovery registered it as a dormant ghost. Resume now deletes the sidecar in the branch where it finds its own record gone. Two resumes both passed the `Running` check for the same reason. The record is marked in-flight under the first lock, so the second request is refused with `already_running` instead of losing a race for the FIFO and reporting a tmux failure. Every path out of resume clears the marker. Both races are driven deterministically in the tests: the manager's tmux spawn is parked mid-resume while the racing kill or resume runs. Closes #175 --- crates/maestro-daemon/src/session.rs | 45 ++++- crates/maestro-daemon/tests/resume.rs | 234 ++++++++++++++++++++++++++ docs/RESTART.md | 8 +- 3 files changed, 282 insertions(+), 5 deletions(-) diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index f66b72d..b58b020 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -87,6 +87,11 @@ pub struct SessionRecord { /// (`POST /.../focus`), `None` while unfocused. Daemon-local; never rides /// the wire. pub focused_since: Option, + /// Set for as long as a [`SessionManager::resume`] is between its two + /// takes of the map lock (#175). The spawn happens with the lock released, + /// so the `Running` check alone lets a second request through; this is the + /// state the second one is refused on. Daemon-local; never rides the wire. + pub resuming: bool, /// Sticky acknowledgement applied by [`crate::activity::apply_focus_suppression`]: /// once the pane has been focused long enough to clear an `AwaitingInput` /// glow, this holds through focus loss until the session does new work. @@ -595,6 +600,7 @@ impl SessionManager { naming: false, focused_since: None, awaiting_suppressed: false, + resuming: false, }, ); } @@ -646,15 +652,23 @@ impl SessionManager { id: Uuid, presets: &[maestro_protocol::SpawnPreset], ) -> Result { + // The spawn below runs with the map lock released, so the `Running` + // check alone is not enough to keep two resumes apart (#175): both + // would pass it, both would open a `PtyBridge` on the same FIFO, and + // the loser would report a tmux failure instead of the conflict it + // is. The record is marked in-flight under this first lock, and every + // path out of here clears it again. let mut session = { - let sessions = self.inner.sessions.lock().await; - let rec = sessions.get(&id).ok_or(SessionError::NotFound(id))?; - if rec.session.status == SessionStatus::Running { + let mut sessions = self.inner.sessions.lock().await; + let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?; + if rec.session.status == SessionStatus::Running || rec.resuming { return Err(SessionError::AlreadyRunning(id)); } + rec.resuming = true; rec.session.clone() }; if !session.cwd.exists() { + self.clear_resuming(id).await; return Err(SessionError::CwdMissing(session.cwd.clone())); } let command = crate::resume::resume_command(&session, presets); @@ -686,6 +700,7 @@ impl SessionManager { if let Err(e) = spawned { // `bridge` drops here, closing the FIFO reader. Nothing else was // touched, so the session stays exactly as dormant as it was. + self.clear_resuming(id).await; return Err(SessionError::Tmux(e)); } @@ -705,6 +720,7 @@ impl SessionManager { // sidecar does not describe. let tmux = self.inner.tmux.clone(); let _ = tmux_blocking(move || tmux.kill_session(id)).await; + self.clear_resuming(id).await; return Err(SessionError::Sidecar(e)); } @@ -712,9 +728,17 @@ impl SessionManager { let mut sessions = self.inner.sessions.lock().await; let Some(rec) = sessions.get_mut(&id) else { // Killed while we were spawning. Undo the pane instead of - // leaving one behind with no record of it. + // leaving one behind with no record of it — and delete the + // sidecar the write above just put back (#175). The kill + // deleted it before that write landed, so leaving it would + // have the next recovery register a dormant ghost for a + // session that is gone and whose worktree is already removed. + drop(sessions); let tmux = self.inner.tmux.clone(); let _ = tmux_blocking(move || tmux.kill_session(id)).await; + if let Err(e) = self.inner.sidecar.delete(id) { + tracing::warn!(error = ?e, %id, "resume: failed to delete the sidecar of a session killed mid-resume"); + } return Err(SessionError::NotFound(id)); }; rec.session = session.clone(); @@ -724,6 +748,7 @@ impl SessionManager { rec.settled = false; rec.focused_since = None; rec.awaiting_suppressed = false; + rec.resuming = false; } let _ = self.inner.events.send(CatalogEvent::SessionUpdated { session: session.clone(), @@ -731,6 +756,16 @@ impl SessionManager { Ok(session) } + /// Drop the in-flight marker [`Self::resume`] set under its first lock + /// (#175), so a refused or failed resume can be retried. A no-op when the + /// record is already gone: a kill that landed mid-resume took the marker + /// with it. + async fn clear_resuming(&self, id: Uuid) { + if let Some(rec) = self.inner.sessions.lock().await.get_mut(&id) { + rec.resuming = false; + } + } + /// Best-effort removal of a worktree created earlier in `create` when a /// later step fails. No-op when the request had no worktree. fn rollback_worktree(&self, base: &Option, path: &Option) { @@ -922,6 +957,7 @@ impl SessionManager { naming: false, focused_since: None, awaiting_suppressed: false, + resuming: false, }, ); let _ = self @@ -1048,6 +1084,7 @@ impl SessionManager { naming: false, focused_since: None, awaiting_suppressed: false, + resuming: false, }, ); registered.push(session); diff --git a/crates/maestro-daemon/tests/resume.rs b/crates/maestro-daemon/tests/resume.rs index f9fdcdc..0700ac5 100644 --- a/crates/maestro-daemon/tests/resume.rs +++ b/crates/maestro-daemon/tests/resume.rs @@ -357,3 +357,237 @@ async fn post_resume_on_an_unknown_session_is_404() { .unwrap(); assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND); } + +// --------------------------------------------------------------------------- +// Races (#175) +// --------------------------------------------------------------------------- + +/// Real tmux, with the *first* `new_session_piped` held inside the call until +/// the test releases it. Later calls pass straight through, so a second resume +/// racing the first never blocks on a gate the first one owns. +/// +/// This is what makes the two races below deterministic rather than timing +/// dependent: the resume under test is parked in exactly the window the bug +/// lives in, and the racing operation runs to completion before it moves. +#[derive(Debug)] +struct GatedSpawn { + inner: maestro_daemon::tmux::Tmux, + entered: tokio::sync::mpsc::UnboundedSender<()>, + release: std::sync::Mutex>, + gated: std::sync::atomic::AtomicBool, +} + +/// The test's end of a [`GatedSpawn`]. +struct Gate { + entered: tokio::sync::mpsc::UnboundedReceiver<()>, + release: std::sync::mpsc::SyncSender<()>, +} + +impl Gate { + /// Wait until the gated spawn is parked mid-resume. + async fn wait_entered(&mut self) { + tokio::time::timeout(Duration::from_secs(10), self.entered.recv()) + .await + .expect("resume reached the tmux spawn") + .expect("gate alive"); + } + + fn release(&self) { + self.release.send(()).expect("gate receiver alive"); + } +} + +fn gated_panes(inner: maestro_daemon::tmux::Tmux) -> (Arc, Gate) { + let (entered_tx, entered_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release_tx, release_rx) = std::sync::mpsc::sync_channel(1); + ( + Arc::new(GatedSpawn { + inner, + entered: entered_tx, + release: std::sync::Mutex::new(release_rx), + gated: std::sync::atomic::AtomicBool::new(true), + }), + Gate { + entered: entered_rx, + release: release_tx, + }, + ) +} + +impl maestro_daemon::panes::Panes for GatedSpawn { + fn new_session_piped( + &self, + id: Uuid, + cwd: &std::path::Path, + command: &str, + session_env: &std::collections::BTreeMap, + sink_command: &str, + ) -> Result<(), maestro_daemon::tmux::TmuxError> { + if self.gated.swap(false, std::sync::atomic::Ordering::SeqCst) { + let _ = self.entered.send(()); + let _ = self.release.lock().expect("gate mutex").recv(); + } + self.inner + .new_session_piped(id, cwd, command, session_env, sink_command) + } + + fn kill_session(&self, id: Uuid) -> Result<(), maestro_daemon::tmux::TmuxError> { + self.inner.kill_session(id) + } + + fn list_sessions(&self) -> Result, maestro_daemon::tmux::TmuxError> { + self.inner.list_sessions() + } + + fn pane_info( + &self, + id: Uuid, + ) -> Result, maestro_daemon::tmux::TmuxError> { + self.inner.pane_info(id) + } + + fn send_keys(&self, id: Uuid, bytes: &[u8]) -> Result<(), maestro_daemon::tmux::TmuxError> { + self.inner.send_keys(id, bytes) + } + + fn pipe_pane(&self, id: Uuid, sink: &str) -> Result<(), maestro_daemon::tmux::TmuxError> { + self.inner.pipe_pane(id, sink) + } + + fn resize_window( + &self, + id: Uuid, + cols: u16, + rows: u16, + ) -> Result<(), maestro_daemon::tmux::TmuxError> { + self.inner.resize_window(id, cols, rows) + } + + fn capture_pane(&self, id: Uuid) -> Result>, maestro_daemon::tmux::TmuxError> { + self.inner.capture_pane(id) + } +} + +/// A dormant session whose manager's tmux spawn can be parked by the test. +async fn dormant_gated(fx: &DaemonFixture, cwd: &TempDir) -> (SessionManager, Session, Gate) { + let session = fx + .manager + .create(CreateSessionRequest { + cwd: cwd.path().to_path_buf(), + command: "sleep 30".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .expect("create"); + fx.tmux.tmux.kill_server().expect("kill tmux server"); + + let (panes, gate) = gated_panes(fx.tmux.tmux.clone()); + let manager = SessionManager::new( + panes, + maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()), + maestro_daemon::worktree::WorktreeManager::new(None), + fx.fifo_dir(), + ); + manager.recover().await.expect("recover"); + assert_eq!( + manager.status(session.id).await, + Some(SessionStatus::Dormant) + ); + (manager, session, gate) +} + +/// A kill that lands while a resume is inside the tmux spawn must not leave +/// the sidecar behind. The kill deletes it; the resume writes it back a moment +/// later and then discovers its own record is gone. The sidecar it resurrected +/// would otherwise make the next recovery register a ghost dormant session for +/// a session that was killed. +#[tokio::test] +async fn a_kill_during_resume_leaves_no_sidecar_behind() { + let fx = DaemonFixture::new(); + let cwd = TempDir::new().unwrap(); + let (manager, session, mut gate) = dormant_gated(&fx, &cwd).await; + let id = session.id; + + let resuming = tokio::spawn({ + let manager = manager.clone(); + async move { manager.resume(id, &[]).await } + }); + gate.wait_entered().await; + + manager.kill(id, false).await.expect("kill mid-resume"); + gate.release(); + + let err = resuming.await.unwrap().expect_err("record is gone"); + assert!(matches!(err, SessionError::NotFound(got) if got == id)); + + assert!( + maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()) + .read(id) + .expect("read sidecar") + .is_none(), + "the killed session's sidecar must not be resurrected by the losing resume" + ); + + // And so the next recovery registers no ghost. + let after = revive(&fx); + after.recover().await.expect("recover"); + assert!( + after.snapshot().await.is_empty(), + "a killed session must not come back dormant" + ); +} + +/// Two resumes of the same session must not both proceed. The second is +/// refused on the state — `already_running` — not on its tmux spawn colliding +/// with the first one's. +#[tokio::test] +async fn a_second_resume_is_refused_while_the_first_is_in_flight() { + let fx = DaemonFixture::new(); + let cwd = TempDir::new().unwrap(); + let (manager, session, mut gate) = dormant_gated(&fx, &cwd).await; + let id = session.id; + + let resuming = tokio::spawn({ + let manager = manager.clone(); + async move { manager.resume(id, &[]).await } + }); + gate.wait_entered().await; + + let err = manager + .resume(id, &[]) + .await + .expect_err("a resume is already in flight"); + assert!(matches!(err, SessionError::AlreadyRunning(got) if got == id)); + assert_eq!(err.to_error_info().code, "already_running"); + + gate.release(); + let first = resuming.await.unwrap().expect("the first resume wins"); + assert_eq!(first.status, SessionStatus::Running); + assert_eq!( + fx.tmux.tmux.list_sessions().expect("list"), + vec![id], + "exactly one pane for the session" + ); +} + +/// A refused resume must not leave the session marked in-flight: once the +/// reason for the refusal is gone, resuming works. +#[tokio::test] +async fn a_refused_resume_can_be_retried() { + let fx = DaemonFixture::new(); + let cwd = TempDir::new().unwrap(); + let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "retried").await; + + std::fs::remove_dir_all(cwd.path()).unwrap(); + let err = manager + .resume(session.id, &[]) + .await + .expect_err("cwd is gone"); + assert_eq!(err.to_error_info().code, "cwd_missing"); + + std::fs::create_dir_all(cwd.path()).unwrap(); + let resumed = manager.resume(session.id, &[]).await.expect("resume"); + assert_eq!(resumed.status, SessionStatus::Running); +} diff --git a/docs/RESTART.md b/docs/RESTART.md index 60e46e5..d85c1cc 100644 --- a/docs/RESTART.md +++ b/docs/RESTART.md @@ -98,7 +98,13 @@ conversation id is respawned on the command it was created with. `Session::comma always keeps that original command, so a second resume resolves the same preset. Resume is refused with `409` for a session that is already running (`already_running`) and for one whose `cwd` has disappeared (`cwd_missing`) — -neither is something a respawn could fix. The other thing to do with a dormant +neither is something a respawn could fix. A session with a resume already in +flight is refused as `already_running` too (#175): the respawn happens with +the session map unlocked, so the record is marked in-flight before it starts +and the second request is answered on that state rather than on its tmux spawn +colliding with the first one's. A resume that loses a race with a kill takes +its own sidecar back off disk, so the killed session does not return as a +dormant ghost on the next recovery. The other thing to do with a dormant session is dismiss it, which removes the sidecar so the next recovery does not bring it back.