refactor(maestro): single-source broadcast capacities + extract cleanup_record (NAP-273 D/E)
D: the four independent `= 256` broadcast-channel constants (session events, local event bus, PTY fan-out, registry events) now derive from one shared `maestro_config::BroadcastTuning::DEFAULT`, so the coincidence is explicit and there's a home for per-deployment tuning later. E: `kill()` and `reap_dead_session()` hand-rolled the same FIFO/sidecar/ worktree/broadcast teardown, differing only in whether tmux still had a session to kill. Extracted into `cleanup_record(inner, id, record, kill_tmux)`. kill's sidecar-delete is now best-effort (the session is already out of the authoritative map), matching the reaper. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -22,9 +22,11 @@ use tokio::sync::broadcast;
|
||||
|
||||
use crate::session::{SessionEvent, SessionManager};
|
||||
|
||||
/// Capacity of the per-bus broadcast channel. Matches the registry's
|
||||
/// `EVENT_CHANNEL_CAPACITY` so a slow client behaves the same on either side.
|
||||
const BUS_CAPACITY: usize = 256;
|
||||
/// Capacity of the per-bus broadcast channel. Sourced from the shared
|
||||
/// [`maestro_config::BroadcastTuning`], which the registry's
|
||||
/// `EVENT_CHANNEL_CAPACITY` also tracks, so a slow client behaves the same on
|
||||
/// either side (NAP-273).
|
||||
const BUS_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.events_capacity;
|
||||
|
||||
/// Cheap-to-clone handle around a `broadcast::Sender<RegistryEvent>`.
|
||||
#[derive(Debug, Clone)]
|
||||
|
||||
@@ -38,8 +38,9 @@ use tokio::task::JoinHandle;
|
||||
|
||||
/// How many recent PTY frames each new subscriber inherits before starting to
|
||||
/// receive live frames. A small buffer is fine — clients render a fresh
|
||||
/// terminal anyway.
|
||||
pub const PTY_BROADCAST_CAPACITY: usize = 256;
|
||||
/// terminal anyway. Sourced from the shared [`maestro_config::BroadcastTuning`]
|
||||
/// (NAP-273).
|
||||
pub const PTY_BROADCAST_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.pty_capacity;
|
||||
|
||||
/// Default replay-buffer size — roughly one screenful for a typical 24×80
|
||||
/// terminal plus reasonable scrollback. Override via
|
||||
|
||||
@@ -22,8 +22,10 @@ use crate::worktree::{WorktreeError, WorktreeManager};
|
||||
|
||||
/// Capacity of the session-events broadcast channel. The registry client
|
||||
/// listens; if it lags by this much we just drop frames (the registry already
|
||||
/// reconciles on `Hello` via the snapshot).
|
||||
pub const EVENTS_CAPACITY: usize = 256;
|
||||
/// reconciles on `Hello` via the snapshot). Sourced from the shared
|
||||
/// [`maestro_config::BroadcastTuning`] so all broadcast channels track one
|
||||
/// definition (NAP-273).
|
||||
pub const EVENTS_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.events_capacity;
|
||||
|
||||
/// How often the liveness watcher polls tmux for sessions that disappeared.
|
||||
pub const DEFAULT_LIVENESS_INTERVAL: Duration = Duration::from_secs(2);
|
||||
@@ -299,20 +301,7 @@ impl SessionManager {
|
||||
sessions.remove(&id)
|
||||
};
|
||||
let record = removed.ok_or(SessionError::NotFound(id))?;
|
||||
|
||||
let tmux = self.inner.tmux.clone();
|
||||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||||
let _ = std::fs::remove_file(&record.fifo_path);
|
||||
self.inner
|
||||
.sidecar
|
||||
.delete(id)
|
||||
.map_err(SessionError::Sidecar)?;
|
||||
if record.cleanup_worktree_on_kill {
|
||||
if let (Some(base), Some(path)) = (&record.worktree_base_repo, &record.worktree_path) {
|
||||
let _ = self.inner.worktree_mgr.remove(base, path);
|
||||
}
|
||||
}
|
||||
let _ = self.inner.events.send(SessionEvent::Removed(id));
|
||||
cleanup_record(&self.inner, id, record, true).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -494,9 +483,26 @@ async fn reap_dead_session(inner: &Arc<Inner>, id: Uuid) {
|
||||
return;
|
||||
};
|
||||
tracing::info!(%id, "session vanished from tmux — reaping");
|
||||
cleanup_record(inner, id, record, false).await;
|
||||
}
|
||||
|
||||
/// Tear down a session's artifacts after it has already been removed from the
|
||||
/// live map. Shared by [`SessionManager::kill`] and [`reap_dead_session`],
|
||||
/// which previously hand-rolled the same five-step choreography and only
|
||||
/// differed in whether tmux still had a session to kill (NAP-273).
|
||||
///
|
||||
/// Best-effort: every step is fire-and-forget because the session is already
|
||||
/// gone from the authoritative in-memory map, so a stray FIFO or sidecar can't
|
||||
/// resurrect it. `kill_tmux` is `true` for an explicit kill and `false` for the
|
||||
/// liveness reaper (tmux already lost the session).
|
||||
async fn cleanup_record(inner: &Arc<Inner>, id: Uuid, record: SessionRecord, kill_tmux: bool) {
|
||||
if kill_tmux {
|
||||
let tmux = inner.tmux.clone();
|
||||
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
|
||||
}
|
||||
let _ = std::fs::remove_file(&record.fifo_path);
|
||||
if let Err(e) = inner.sidecar.delete(id) {
|
||||
tracing::warn!(error = ?e, %id, "reap: failed to delete sidecar");
|
||||
tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar");
|
||||
}
|
||||
if record.cleanup_worktree_on_kill {
|
||||
if let (Some(base), Some(path)) = (&record.worktree_base_repo, &record.worktree_path) {
|
||||
|
||||
@@ -45,8 +45,10 @@ impl<K: Eq + Hash, V> Drop for PendingGuard<'_, K, V> {
|
||||
|
||||
/// Capacity of the registry-wide broadcast channel that feeds
|
||||
/// `WS /api/events` subscribers. Subscribers that fall this far behind get a
|
||||
/// `Lagged` event and are expected to resync via `GET /api/sessions`.
|
||||
const EVENT_CHANNEL_CAPACITY: usize = 256;
|
||||
/// `Lagged` event and are expected to resync via `GET /api/sessions`. Sourced
|
||||
/// from the shared [`maestro_config::BroadcastTuning`] so the daemon-side buses
|
||||
/// track the same value (NAP-273).
|
||||
const EVENT_CHANNEL_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.events_capacity;
|
||||
|
||||
/// How long a daemon can go without sending us *any* message before it is
|
||||
/// considered offline and its sessions are dropped from the catalog.
|
||||
|
||||
Reference in New Issue
Block a user