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

fix(maestro): close ACP replay seams

Browse Source 
Serialize replay reads with event appends, acquire the live receiver under that lock, and drain the event pump during shutdown. Recovery now removes its FIFO when opening the log fails.
This commit is contained in:
Miguel Palhas
2026-09-04 23:25:46 +01:00
parent d11098a478
commit e9142bb027
3 changed files with 183 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+105 -45
View File
@@ -17,7 +17,7 @@ use maestro_protocol::{
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
};
use tokio::sync::{broadcast, Mutex};
use tokio::sync::{broadcast, watch, Mutex};
use uuid::Uuid;
use crate::panes::Panes;
@@ -94,9 +94,18 @@ pub struct SessionRecord {
/// adapter never came up. Shared so a kill or shutdown can terminate it
/// outside the sessions lock. Daemon-local; never rides the wire.
pub acp: Option<Arc<AcpSession>>,
/// Live notifications after they have been appended to the event log.
/// `None` for terminal sessions and failed structured spawns.
pub acp_events: Option<broadcast::Sender<LoggedEvent>>,
/// Open log plus its post-append live sender. Their shared lock is the
/// replay/live seam used by the structured WebSocket in #138.
acp_events: Option<Arc<Mutex<AcpEventStream>>>,
/// Completion signal for the task draining ACP notifications into the
/// event log. Shutdown waits for it before returning.
acp_event_pump: Option<watch::Receiver<bool>>,
}
#[derive(Debug)]
struct AcpEventStream {
log: EventLog,
live: broadcast::Sender<LoggedEvent>,
}
#[derive(Debug, thiserror::Error)]
@@ -297,11 +306,28 @@ impl SessionManager {
sessions.values().map(|r| r.session.clone()).collect()
}
/// Subscribe to live structured-session events. Every event returned here
/// has already been appended to the session's durable log.
pub async fn subscribe_acp_events(&self, id: Uuid) -> Option<broadcast::Receiver<LoggedEvent>> {
let sessions = self.inner.sessions.lock().await;
Some(sessions.get(&id)?.acp_events.as_ref()?.subscribe())
/// Read replay and acquire its live continuation under the same lock.
/// Events are therefore present exactly once across the returned seam.
pub async fn subscribe_acp_events_with_replay(
&self,
id: Uuid,
cursor: EventCursor,
limit: usize,
) -> Result<Option<(Vec<LoggedEvent>, broadcast::Receiver<LoggedEvent>)>, SessionError> {
let stream = {
let sessions = self.inner.sessions.lock().await;
let Some(record) = sessions.get(&id) else {
return Ok(None);
};
let Some(stream) = &record.acp_events else {
return Ok(None);
};
stream.clone()
};
let stream = stream.lock().await;
let live = stream.live.subscribe();
let replay = self.inner.sidecar.read_events(id, cursor, limit)?;
Ok(Some((replay, live)))
}
/// Read a cursor-bounded slice of the durable structured-session log.
@@ -323,12 +349,15 @@ 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) {
let children: Vec<(Uuid, Arc<AcpSession>)> = {
let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
let sessions = self.inner.sessions.lock().await;
sessions
.iter()
.filter(|(_, r)| r.session.status == SessionStatus::Running)
.filter_map(|(id, r)| r.acp.clone().map(|acp| (*id, acp)))
.filter_map(|(id, r)| {
r.acp
.clone()
.map(|acp| (*id, acp, r.acp_event_pump.clone()))
})
.collect()
};
if children.is_empty() {
@@ -337,9 +366,14 @@ impl SessionManager {
tracing::info!(count = children.len(), "shutdown: ending adapter children");
let terminations = children
.iter()
.map(|(_, acp)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
.map(|(_, acp, _)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
futures_util::future::join_all(terminations).await;
for (id, _) in children {
let drains = children
.iter()
.filter_map(|(_, _, done)| done.clone())
.map(wait_for_event_pump);
futures_util::future::join_all(drains).await;
for (id, _, _) in children {
mark_session_dead(&self.inner, id).await;
}
}
@@ -527,11 +561,11 @@ impl SessionManager {
}
};
let (acp_events, event_log) = if let Some(acp) = &acp {
let acp_events = if let Some(acp) = &acp {
match self.inner.sidecar.open_events(id) {
Ok(log) => {
let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
(Some(tx), Some(log))
Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx })))
}
Err(e) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
@@ -542,7 +576,7 @@ impl SessionManager {
}
}
} else {
(None, None)
None
};
// Capture spawn-history inputs before `req` is partially moved into
@@ -615,8 +649,7 @@ impl SessionManager {
let event_pump = acp
.as_ref()
.zip(acp_events.as_ref())
.zip(event_log)
.map(|((acp, events), log)| (acp.clone(), events.clone(), log));
.map(|(acp, events)| (acp.clone(), events.clone()));
{
let mut sessions = self.inner.sessions.lock().await;
sessions.insert(
@@ -635,11 +668,15 @@ impl SessionManager {
awaiting_suppressed: false,
acp,
acp_events,
acp_event_pump: None,
},
);
}
if let Some((acp, events, log)) = event_pump {
start_acp_event_pump(&self.inner, id, &acp, events, log);
if let Some((acp, events)) = event_pump {
let done = start_acp_event_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(),
@@ -856,6 +893,7 @@ impl SessionManager {
// daemon; only tmux-backed sessions can be adopted.
acp: None,
acp_events: None,
acp_event_pump: None,
},
);
let _ = self
@@ -880,10 +918,10 @@ impl SessionManager {
}
let id = session.id;
let mut session_env = sanitize_env(&session.env)?;
let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
let bridge =
PtyBridge::start_with_replay(fifo_path.clone(), self.inner.replay_bytes).await?;
let mut session_env = sanitize_env(&session.env)?;
session_env.insert("MAESTRO_SESSION_ID".into(), id.to_string());
if let Some(url) = self
.inner
@@ -930,20 +968,21 @@ impl SessionManager {
};
session.activity_detail = failure;
let (acp_events, event_log) = if let Some(acp) = &acp {
let acp_events = if let Some(acp) = &acp {
match self.inner.sidecar.open_events(id) {
Ok(log) => {
let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
(Some(tx), Some(log))
Some(Arc::new(Mutex::new(AcpEventStream { log, live: tx })))
}
Err(error) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
let _ = std::fs::remove_file(&fifo_path);
return Err(SessionError::Sidecar(error));
}
}
} else {
(None, None)
None
};
if let Err(error) = self.inner.sidecar.write(&session) {
if let Some(acp) = &acp {
@@ -956,8 +995,7 @@ impl SessionManager {
let event_pump = acp
.as_ref()
.zip(acp_events.as_ref())
.zip(event_log)
.map(|((acp, events), log)| (acp.clone(), events.clone(), log));
.map(|(acp, events)| (acp.clone(), events.clone()));
self.inner.sessions.lock().await.insert(
id,
@@ -981,10 +1019,14 @@ impl SessionManager {
awaiting_suppressed: false,
acp,
acp_events,
acp_event_pump: None,
},
);
if let Some((acp, events, log)) = event_pump {
start_acp_event_pump(&self.inner, id, &acp, events, log);
if let Some((acp, events)) = event_pump {
let done = start_acp_event_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
@@ -1321,34 +1363,49 @@ fn start_acp_event_pump(
inner: &Arc<Inner>,
id: Uuid,
acp: &Arc<AcpSession>,
events: broadcast::Sender<LoggedEvent>,
mut log: EventLog,
) {
events: Arc<Mutex<AcpEventStream>>,
) -> watch::Receiver<bool> {
let (done_tx, done_rx) = watch::channel(false);
let Some(mut updates) = acp.take_updates() else {
tracing::error!(%id, "structured session notification stream was already taken");
return;
let _ = done_tx.send(true);
return done_rx;
};
let weak_inner = Arc::downgrade(inner);
let weak_acp = Arc::downgrade(acp);
tokio::spawn(async move {
while let Some(notification) = updates.recv().await {
let seq = match log.append(&notification) {
Ok(seq) => seq,
Err(error) => {
tracing::error!(%id, %error, "structured event log append failed");
if let Some(acp) = weak_acp.upgrade() {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
let append_error = {
let mut events = events.lock().await;
match events.log.append(&notification) {
Ok(seq) => {
let _ = events.live.send(LoggedEvent { seq, notification });
None
}
if let Some(inner) = weak_inner.upgrade() {
mark_session_dead(&inner, id).await;
}
return;
Err(error) => Some(error),
}
};
let _ = events.send(LoggedEvent { seq, notification });
if let Some(error) = append_error {
tracing::error!(%id, %error, "structured event log append failed");
if let Some(acp) = weak_acp.upgrade() {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
}
if let Some(inner) = weak_inner.upgrade() {
mark_session_dead(&inner, id).await;
}
break;
}
}
let _ = done_tx.send(true);
});
done_rx
}
async fn wait_for_event_pump(mut done: watch::Receiver<bool>) {
if !*done.borrow() {
let _ = done.wait_for(|finished| *finished).await;
}
}
async fn tmux_blocking<T, F>(f: F) -> T
@@ -1742,6 +1799,9 @@ async fn cleanup_record(
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
}
if let Some(done) = record.acp_event_pump.clone() {
wait_for_event_pump(done).await;
}
let _ = std::fs::remove_file(&record.fifo_path);
if let Err(e) = inner.sidecar.delete(id) {
tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar");