naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(maestro): allow concurrent ACP handshakes

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 11:29:44 +01:00
parent 1576380080
commit ac2eb01bda
3 changed files with 325 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+135 -57
View File
@@ -6,9 +6,12 @@
//! handler interact with tmux.
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use chrono::Utc;
@@ -18,7 +21,7 @@ use maestro_protocol::{
SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, WorktreeSpec, NAME_KEY,
NAME_SOURCE_KEY,
};
use tokio::sync::{broadcast, oneshot, watch, Mutex};
use tokio::sync::{broadcast, oneshot, watch, Mutex, Notify};
use uuid::Uuid;
use crate::panes::Panes;
@@ -263,9 +266,99 @@ struct Inner {
/// Once daemon shutdown starts, it owns structured-session finalization.
/// Background passes must not classify those deliberate exits as natural.
shutting_down: AtomicBool,
/// Serializes structured creation with shutdown so every adapter child is
/// either rejected or included in shutdown's final drain.
structured_lifecycle: Mutex<()>,
/// Admission is serialized with shutdown, while admitted handshakes run
/// independently and are counted until registration or rollback finishes.
structured_lifecycle: StructuredLifecycle,
}
#[derive(Debug, Default)]
struct StructuredLifecycle {
gate: std::sync::Mutex<()>,
in_flight: AtomicUsize,
settled: Notify,
}
impl StructuredLifecycle {
fn admit<'a>(
&'a self,
shutting_down: &AtomicBool,
) -> Result<StructuredCreate<'a>, SessionError> {
let _gate = self.gate.lock().expect("structured lifecycle gate");
if shutting_down.load(Ordering::Acquire) {
return Err(SessionError::ShuttingDown);
}
self.in_flight.fetch_add(1, Ordering::AcqRel);
Ok(StructuredCreate { lifecycle: self })
}
async fn wait_until_settled(&self) {
loop {
let settled = self.settled.notified();
if self.in_flight.load(Ordering::Acquire) == 0 {
return;
}
settled.await;
}
}
}
struct StructuredCreate<'a> {
lifecycle: &'a StructuredLifecycle,
}
impl Drop for StructuredCreate<'_> {
fn drop(&mut self) {
if self.lifecycle.in_flight.fetch_sub(1, Ordering::AcqRel) == 1 {
self.lifecycle.settled.notify_one();
}
}
}
struct AdmittedStructuredCreate<'a> {
create: Pin<Box<dyn Future<Output = Result<Session, SessionError>> + Send + 'a>>,
/// Field order matters: dropping a cancelled create rolls back its inner
/// future before this permit releases shutdown's in-flight barrier.
_permit: StructuredCreate<'a>,
}
impl Future for AdmittedStructuredCreate<'_> {
type Output = Result<Session, SessionError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.create.as_mut().poll(cx)
}
}
struct PendingCreate<'a> {
inner: &'a Inner,
id: Uuid,
fifo_path: PathBuf,
worktree_path: Option<PathBuf>,
worktree_base_repo: Option<PathBuf>,
committed: bool,
}
impl PendingCreate<'_> {
fn commit(&mut self) {
self.committed = true;
}
}
impl Drop for PendingCreate<'_> {
fn drop(&mut self) {
if self.committed {
return;
}
let _ = std::fs::remove_file(&self.fifo_path);
if let Err(error) = self.inner.sidecar.delete(self.id) {
tracing::warn!(%error, id = %self.id, "rollback: failed to delete create artifacts");
}
if let (Some(base), Some(path)) = (&self.worktree_base_repo, &self.worktree_path) {
if let Err(error) = self.inner.worktree_mgr.remove(base, path) {
tracing::warn!(%error, ?path, "rollback: failed to remove worktree after partial create");
}
}
}
}
impl SessionManager {
@@ -309,7 +402,7 @@ impl SessionManager {
viewports: Mutex::new(Viewports::new()),
next_viewport: AtomicU64::new(0),
shutting_down: AtomicBool::new(false),
structured_lifecycle: Mutex::new(()),
structured_lifecycle: StructuredLifecycle::default(),
}),
}
}
@@ -827,10 +920,18 @@ 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 _gate = self
.inner
.structured_lifecycle
.gate
.lock()
.expect("structured lifecycle gate");
if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
return;
}
}
let _lifecycle = self.inner.structured_lifecycle.lock().await;
self.inner.structured_lifecycle.wait_until_settled().await;
let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
let sessions = self.inner.sessions.lock().await;
sessions
@@ -980,11 +1081,15 @@ impl SessionManager {
/// Spawn a new session. On success returns the freshly-created `Session`.
pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
if req.view == SessionView::Structured {
let _lifecycle = self.inner.structured_lifecycle.lock().await;
if self.inner.shutting_down.load(Ordering::Acquire) {
return Err(SessionError::ShuttingDown);
let permit = self
.inner
.structured_lifecycle
.admit(&self.inner.shutting_down)?;
return AdmittedStructuredCreate {
create: Box::pin(self.create_inner(req)),
_permit: permit,
}
return self.create_inner(req).await;
.await;
}
self.create_inner(req).await
}
@@ -1022,22 +1127,17 @@ impl SessionManager {
// Each side effect below unwinds the ones before it on failure, so a
// partial create never leaks a worktree or — worse — a ghost tmux
// session with no in-memory record (NAP-260).
let bridge =
match PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await {
Ok(b) => b,
Err(e) => {
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(e.into());
}
};
let mut session_env = match sanitize_env(&req.env) {
Ok(env) => env,
Err(e) => {
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(e);
}
let mut pending = PendingCreate {
inner: &self.inner,
id,
fifo_path: fifo_path.clone(),
worktree_path: worktree_path.clone(),
worktree_base_repo: worktree_base_repo.clone(),
committed: false,
};
let bridge =
PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?;
let mut session_env = sanitize_env(&req.env)?;
// Identity + reporting target for the in-session hook (#106). Ours
// win over whatever the request carried, so a client cannot spoof
// another session's identity (#103).
@@ -1084,10 +1184,6 @@ impl SessionManager {
.await
};
if let Err(e) = spawned {
// `bridge` drops at the end of this block, closing the
// FIFO reader.
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Tmux(e));
}
(None, None)
@@ -1105,8 +1201,6 @@ impl SessionManager {
Err(error) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Sidecar(error));
}
};
@@ -1131,8 +1225,6 @@ impl SessionManager {
Err(e) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Sidecar(e));
}
}
@@ -1201,9 +1293,6 @@ impl SessionManager {
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
}
}
let _ = std::fs::remove_file(&fifo_path);
let _ = self.inner.sidecar.delete(id);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Sidecar(e));
}
@@ -1213,6 +1302,11 @@ impl SessionManager {
.map(|(acp, events)| (acp.clone(), events.clone()));
{
let mut sessions = self.inner.sessions.lock().await;
let event_pump = event_pump.map(|(acp, events)| {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&self.inner, id, &acp, events);
done
});
sessions.insert(
id,
SessionRecord {
@@ -1229,17 +1323,10 @@ impl SessionManager {
awaiting_suppressed: false,
acp,
acp_events,
acp_event_pump: None,
acp_event_pump: event_pump,
},
);
}
if let Some((acp, events)) = event_pump {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&self.inner, id, &acp, events);
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
record.acp_event_pump = Some(done);
}
}
let _ = self.inner.events.send(CatalogEvent::SessionAdded {
session: session.clone(),
});
@@ -1264,19 +1351,10 @@ impl SessionManager {
}
}
pending.commit();
Ok(session)
}
/// 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<PathBuf>, path: &Option<PathBuf>) {
if let (Some(base), Some(path)) = (base, path) {
if let Err(e) = self.inner.worktree_mgr.remove(base, path) {
tracing::warn!(error = %e, ?path, "rollback: failed to remove worktree after partial create");
}
}
}
/// Kill a session and remove its on-disk artifacts.
///
/// Refused (#79) when the session's worktree holds unsaved work —