diff --git a/crates/outrig-cli/src/cli/clean.rs b/crates/outrig-cli/src/cli/clean.rs index 0382f15..37f6b69 100644 --- a/crates/outrig-cli/src/cli/clean.rs +++ b/crates/outrig-cli/src/cli/clean.rs @@ -1,12 +1,14 @@ //! `outrig clean` -- bulk-remove old session records and stray containers. //! -//! Two sweeps run together. The session-store walk removes old -//! metadata/log directories, refusing to touch sessions whose container is -//! still running. The label sweep catches *stray* containers -- ones -//! carrying `org.outrig.session` whose session record is gone (lost record, -//! SIGKILLed outrig, failed `--rm`) -- and `podman rm -f`s the stopped ones -//! older than the cutoff. Running containers are never removed, labeled or -//! not. +//! Two sweeps run together, both answered from a single unfiltered `podman ps +//! -a`. The session-store walk removes old metadata/log directories, refusing +//! to touch a session whose recorded container name is among the running +//! containers -- including an attach session's borrowed, outrig-unlabeled +//! container, which a label-filtered listing would miss. The label sweep +//! catches *stray* containers -- ones carrying `org.outrig.session` whose +//! session record is gone (lost record, SIGKILLed outrig, failed `--rm`) -- +//! and `podman rm -f`s the stopped ones older than the cutoff in one +//! invocation. Running containers are never removed, labeled or not. use std::collections::BTreeSet; use std::future::Future; @@ -19,7 +21,7 @@ use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite, AsyncWriteExt}; use crate::error::{OutrigError, Result}; use crate::session::{self, Session, SessionStore}; -use outrig::container::{Container, LABEL_SESSION, LABEL_SIDECAR}; +use outrig::container::{LABEL_SESSION, LABEL_SIDECAR}; const DAY: u64 = 24 * 60 * 60; pub const DEFAULT_OLDER_THAN: Duration = Duration::from_secs(30 * DAY); @@ -64,7 +66,7 @@ pub async fn execute( cwd, )?; let store = SessionStore::new(root); - let labeled = list_labeled_containers().await?; + let (labeled, running) = list_all_containers().await?; let stdin = tokio::io::BufReader::new(tokio::io::stdin()); let mut stderr = tokio::io::stderr(); execute_with( @@ -73,30 +75,28 @@ pub async fn execute( &store, args, SystemTime::now(), - podman_is_running, + running, labeled, - podman_remove_force, + podman_remove_force_batch, ) .await } #[allow(clippy::too_many_arguments)] -pub async fn execute_with( +pub async fn execute_with( stderr: &mut E, stdin: R, store: &SessionStore, args: &CleanArgs, now: SystemTime, - mut is_running: F, + running: BTreeSet, labeled: Vec, - mut remove_container: D, + mut remove_containers: D, ) -> Result where E: AsyncWrite + Unpin, R: AsyncBufRead + Unpin, - F: FnMut(String) -> Fut, - Fut: Future>, - D: FnMut(String) -> DFut, + D: FnMut(Vec) -> DFut, DFut: Future>, { let sessions = store.list()?; @@ -107,7 +107,7 @@ where if !older_than(session, args.older_than, now) { continue; } - if is_running(session.container_name.clone()).await? { + if running.contains(&session.container_name) { skipped_running.push(session.clone()); continue; } @@ -154,12 +154,16 @@ where } } - let mut strays_removed = 0usize; - for stray in &stray_targets { - remove_container(stray.name.clone()).await?; - strays_removed += 1; - let msg = format!("[outrig] removed container {}\n", stray.name); - stderr.write_all(msg.as_bytes()).await?; + // One `podman rm -f` for every stray. Guard the empty case -- `podman rm + // -f` with no names is an error -- and print the same per-container line + // for each on success, so the happy path stays byte-identical. + let strays_removed = stray_targets.len(); + if !stray_targets.is_empty() { + remove_containers(stray_targets.iter().map(|s| s.name.clone()).collect()).await?; + for stray in &stray_targets { + let msg = format!("[outrig] removed container {}\n", stray.name); + stderr.write_all(msg.as_bytes()).await?; + } } let summary = format!( @@ -392,21 +396,21 @@ fn format_retention(duration: Duration) -> String { } } -async fn podman_is_running(name: String) -> Result { - Ok(Container::is_running(&name).await?) -} - -/// `podman rm -f `, propagating failure -- clean should report a -/// container it could not remove rather than claiming success. -async fn podman_remove_force(name: String) -> Result<()> { +/// `podman rm -f ...` for every stray in one invocation, propagating +/// failure -- clean should report a container it could not remove rather than +/// claiming success. Callers guard the empty case (`podman rm -f` with no +/// names is an error). +async fn podman_remove_force_batch(names: Vec) -> Result<()> { let output = tokio::process::Command::new("podman") - .args(["rm", "-f", &name]) + .args(["rm", "-f"]) + .args(&names) .stdin(Stdio::null()) .output() .await?; if !output.status.success() { return Err(OutrigError::Configuration(format!( - "podman rm -f {name} failed: {}", + "podman rm -f {} failed: {}", + names.join(" "), String::from_utf8_lossy(&output.stderr).trim() )) .into()); @@ -414,37 +418,33 @@ async fn podman_remove_force(name: String) -> Result<()> { Ok(()) } -/// `podman ps -a --filter label=org.outrig.session --format json`, decoded -/// into [`LabeledContainer`]s. Containers missing the expected fields are -/// skipped rather than failing the sweep. -async fn list_labeled_containers() -> Result> { +/// `podman ps -a --format json`, decoded into the labeled-container rows (the +/// stray-sweep input) plus the set of *all* running container names (the +/// record-walk running check). Unfiltered so an attach session's borrowed, +/// outrig-unlabeled primary still shows up in the running set. Rows missing the +/// expected fields are skipped rather than failing the sweep. +async fn list_all_containers() -> Result<(Vec, BTreeSet)> { let output = tokio::process::Command::new("podman") - .args([ - "ps", - "-a", - "--filter", - &format!("label={LABEL_SESSION}"), - "--format", - "json", - ]) + .args(["ps", "-a", "--format", "json"]) .stdin(Stdio::null()) .output() .await?; if !output.status.success() { return Err(OutrigError::Configuration(format!( - "podman ps for labeled containers failed: {}", + "podman ps failed: {}", String::from_utf8_lossy(&output.stderr).trim() )) .into()); } - parse_labeled_containers(&output.stdout) + parse_all_containers(&output.stdout) } -fn parse_labeled_containers(stdout: &[u8]) -> Result> { +fn parse_all_containers(stdout: &[u8]) -> Result<(Vec, BTreeSet)> { let rows: Vec = serde_json::from_slice(stdout).map_err(|source| { OutrigError::Configuration(format!("podman ps --format json: invalid JSON: {source}")) })?; - let mut out = Vec::new(); + let mut labeled = Vec::new(); + let mut running = BTreeSet::new(); for row in rows { let Some(name) = row .get("Names") @@ -454,6 +454,13 @@ fn parse_labeled_containers(stdout: &[u8]) -> Result> { else { continue; }; + let is_running = row + .get("State") + .and_then(|state| state.as_str()) + .is_some_and(|state| state.eq_ignore_ascii_case("running")); + if is_running { + running.insert(name.to_string()); + } let labels = row.get("Labels").and_then(|labels| labels.as_object()); let Some(session_label) = labels .and_then(|labels| labels.get(LABEL_SESSION)) @@ -465,24 +472,20 @@ fn parse_labeled_containers(stdout: &[u8]) -> Result> { .and_then(|labels| labels.get(LABEL_SIDECAR)) .and_then(|value| value.as_str()) .map(str::to_string); - let running = row - .get("State") - .and_then(|state| state.as_str()) - .is_some_and(|state| state.eq_ignore_ascii_case("running")); let created = row .get("Created") .and_then(|created| created.as_i64()) .map(|secs| SystemTime::UNIX_EPOCH + Duration::from_secs(secs.max(0) as u64)) .unwrap_or(SystemTime::UNIX_EPOCH); - out.push(LabeledContainer { + labeled.push(LabeledContainer { name: name.to_string(), session_label: session_label.to_string(), sidecar_label, - running, + running: is_running, created, }); } - Ok(out) + Ok((labeled, running)) } #[cfg(test)] @@ -490,7 +493,7 @@ mod tests { use super::*; #[test] - fn parse_labeled_containers_decodes_podman_ps_json() { + fn parse_all_containers_splits_labeled_rows_and_running_names() { let stdout = br#"[ { "Names": ["outrig-abc"], @@ -505,32 +508,46 @@ mod tests { "Created": 1700000100 }, { - "Names": ["unlabeled"], + "Names": ["unlabeled-stopped"], "Labels": {}, "State": "exited", "Created": 1700000200 + }, + { + "Names": ["borrowed-attach-container"], + "Labels": {}, + "State": "running", + "Created": 1700000300 } ]"#; - let parsed = parse_labeled_containers(stdout).expect("parse"); - assert_eq!(parsed.len(), 2, "unlabeled containers are skipped"); + let (labeled, running) = parse_all_containers(stdout).expect("parse"); - assert_eq!(parsed[0].name, "outrig-abc"); - assert_eq!(parsed[0].session_label, "abc"); - assert_eq!(parsed[0].sidecar_label, None); - assert!(parsed[0].running); + // Only the two `org.outrig.session` rows are stray-sweep input. + assert_eq!(labeled.len(), 2, "unlabeled containers are not stray input"); + assert_eq!(labeled[0].name, "outrig-abc"); + assert_eq!(labeled[0].session_label, "abc"); + assert_eq!(labeled[0].sidecar_label, None); + assert!(labeled[0].running); assert_eq!( - parsed[0].created, + labeled[0].created, SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000) ); - - assert_eq!(parsed[1].name, "outrig-abc-tools"); - assert_eq!(parsed[1].sidecar_label.as_deref(), Some("tools")); - assert!(!parsed[1].running); + assert_eq!(labeled[1].name, "outrig-abc-tools"); + assert_eq!(labeled[1].sidecar_label.as_deref(), Some("tools")); + assert!(!labeled[1].running); + + // The running set spans every running container, labeled or not, so an + // attach session's unlabeled primary is still seen as alive. + assert_eq!(running.len(), 2); + assert!(running.contains("outrig-abc")); + assert!(running.contains("borrowed-attach-container")); + assert!(!running.contains("outrig-abc-tools"), "exited sidecar"); + assert!(!running.contains("unlabeled-stopped")); } #[test] - fn parse_labeled_containers_rejects_bad_json() { - assert!(parse_labeled_containers(b"not json").is_err()); + fn parse_all_containers_rejects_bad_json() { + assert!(parse_all_containers(b"not json").is_err()); } } diff --git a/crates/outrig-cli/src/cli/session_setup.rs b/crates/outrig-cli/src/cli/session_setup.rs index e8b01e8..dc3fdaa 100644 --- a/crates/outrig-cli/src/cli/session_setup.rs +++ b/crates/outrig-cli/src/cli/session_setup.rs @@ -24,7 +24,7 @@ //! //! [`run`]: crate::cli::run -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::fs; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -33,12 +33,12 @@ use std::time::{Duration, Instant, SystemTime}; use crate::cli::env_arg::CliEnvEntries; use crate::cli::volume_arg::CliVolume; use crate::cli::watcher::SessionWatcher; -use crate::error::{OutrigError, Result}; +use crate::error::{CliError, OutrigError, Result}; use crate::llm; use crate::paths::{default_session_root, repo_root_from_config_path}; use crate::session::{self, Session, SessionId, SessionStore}; use outrig::config::{ - Config, ImageConfig, MistralrsDeviceSpec, MountAccess, MountConfig, NetworkMode, + Config, ImageConfig, McpServerSpec, MistralrsDeviceSpec, MountAccess, MountConfig, NetworkMode, SidecarOnFailure, SidecarStart, }; use outrig::container::{ @@ -611,6 +611,8 @@ pub async fn setup(args: SessionSetupArgs<'_>) -> Result { Some(SessionWatcher::spawn( containers.primary.name().to_string(), session.sidecar_container_names.clone(), + sid.0.clone(), + session.started_at, )) } else { None @@ -709,27 +711,43 @@ async fn setup_sidecars_and_network( let image_mcp = embedded::read_embedded_mcp(args.image_tag, args.transcript).await?; sidecar::merge_primary_labels(&mut plan, image_mcp); - let sidecar_names: Vec = plan.sidecars.keys().cloned().collect(); - for name in &sidecar_names { - let sc = plan.sidecars[name].clone(); - - let tag = match ensure_sidecar_image(args.cfg, args.repo_root, &sc.image, args.transcript) - .await - { - Ok(tag) => tag, - Err(e) => { - warn_or_bail(&plan, &sc, e)?; + // Phase A -- ensure each distinct sidecar image (and read its labels when a + // non-anonymous sidecar needs them) concurrently. Sidecars are independent, + // so their podman-heavy ensure/inspect work overlaps; the results are + // consumed below in deterministic name order. + let resolutions = + resolve_sidecar_images(&plan, args.cfg, args.repo_root, args.transcript).await; + + // Phase B -- merge sidecar labels and decide which sidecars start, serially + // in name order. Keeping this ordered preserves the deterministic + // label-collision errors that the concurrent Phase A must not perturb. + let mut to_start: Vec<(String, ImageTag, SidecarPlan)> = Vec::new(); + for name in plan.sidecars.keys().cloned().collect::>() { + let sc = plan.sidecars[&name].clone(); + let resolution = &resolutions[&sc.image]; + + let tag = match &resolution.tag { + Ok(tag) => tag.clone(), + Err(message) => { + warn_or_bail(&plan, &sc, shared_image_error(message))?; continue; } }; if !sc.anonymous { - let label_mcp = embedded::read_embedded_mcp(&tag, args.transcript).await?; - sidecar::merge_sidecar_labels(&mut plan, name, label_mcp)?; + let label_mcp = match resolution + .labels + .as_ref() + .expect("non-anonymous sidecar image is label-read in Phase A") + { + Ok(label_mcp) => label_mcp.clone(), + Err(message) => return Err(shared_image_error(message)), + }; + sidecar::merge_sidecar_labels(&mut plan, &name, label_mcp)?; } if !args.start_sidecars || sc.start != SidecarStart::Auto { - if args.start_sidecars && plan.servers_in(name).next().is_some() { + if args.start_sidecars && plan.servers_in(&name).next().is_some() { eprintln!( "[outrig] sidecar {name} is start = \"manual\"; skipping its MCP servers" ); @@ -737,23 +755,12 @@ async fn setup_sidecars_and_network( continue; } - let started = match plan.entrypoint_server_in(&sc) { - Some((server_name, placed)) => { - create_one_entrypoint_sidecar(&args, &tag, &sc, server_name, &placed.spec).await - } - None => { - let needs_bootstrap = plan.sidecar_needs_bootstrap(&sc); - start_one_sidecar(&args.start_ctx(), &tag, &sc, needs_bootstrap).await - } - }; - match started { - Ok(container) => { - containers.sidecars.insert(name.clone(), container); - } - Err(e) => warn_or_bail(&plan, &sc, e)?, - } + to_start.push((name, tag, sc)); } + // Phase C -- start the auto sidecars concurrently, inserting in name order. + start_auto_sidecars(&plan, &args, to_start, containers).await?; + let network = match args.network_mode { NetworkMode::Default => None, NetworkMode::Audit => Some(attach_interceptor("audit", &plan, containers, &args).await?), @@ -763,6 +770,114 @@ async fn setup_sidecars_and_network( Ok((plan, network)) } +/// One distinct sidecar image resolved once and shared by every sidecar that +/// references it (Phase A of [`setup_sidecars_and_network`]). `tag` is the +/// image-ensure result; `labels` is the `org.outrig.mcp` read, present only +/// when at least one *non-anonymous* sidecar uses the image -- anonymous +/// sidecars never read labels, so an anonymous-only image is never inspected +/// (inspecting it could newly fail a session on a malformed label). A failure +/// is kept as its `Display` text -- one image feeds many sidecars but +/// [`CliError`] is not `Clone` -- and [`shared_image_error`] re-wraps it +/// verbatim per consumer. +struct ImageResolution { + tag: std::result::Result, + labels: Option, String>>, +} + +/// Ensure every distinct sidecar image (and read its labels when needed) +/// concurrently, deduped by image ref so two sidecars sharing an image run the +/// tag-compute + ensure + label-inspect once. `join_all` runs the futures on +/// the current task, so the borrowed inputs need no `'static` and nothing is +/// cloned beyond the image key. +async fn resolve_sidecar_images( + plan: &SessionMcpPlan, + cfg: &Config, + repo_root: &Path, + transcript: Option<&Transcript>, +) -> HashMap { + // Distinct images, each flagged for a label read iff a non-anonymous + // sidecar uses it (anonymous-only images are never inspected). + let mut images: BTreeMap<&str, bool> = BTreeMap::new(); + for sc in plan.sidecars.values() { + *images.entry(sc.image.as_str()).or_default() |= !sc.anonymous; + } + + futures_util::future::join_all(images.into_iter().map(|(image, read_labels)| async move { + let resolution = match ensure_sidecar_image(cfg, repo_root, image, transcript).await { + Ok(tag) => { + let labels = if read_labels { + Some( + embedded::read_embedded_mcp(&tag, transcript) + .await + .map_err(|e| e.to_string()), + ) + } else { + None + }; + ImageResolution { + tag: Ok(tag), + labels, + } + } + Err(e) => ImageResolution { + tag: Err(e.to_string()), + labels: None, + }, + }; + (image.to_string(), resolution) + })) + .await + .into_iter() + .collect() +} + +/// Start the `start = "auto"` sidecars concurrently, then insert them into +/// `containers` in name order. The container starts (and any bootstrap) are +/// independent; only the ordered insert and `warn`/`abort` routing run after +/// the join, so a `warn` sidecar drops exactly as it did serially. +async fn start_auto_sidecars( + plan: &SessionMcpPlan, + args: &SidecarPhaseArgs<'_>, + to_start: Vec<(String, ImageTag, SidecarPlan)>, + containers: &mut SessionContainers, +) -> Result<()> { + let started = + futures_util::future::join_all(to_start.into_iter().map(|(name, tag, sc)| async move { + let result = match plan.entrypoint_server_in(&sc) { + Some((server_name, placed)) => { + create_one_entrypoint_sidecar(args, &tag, &sc, server_name, &placed.spec).await + } + None => { + let needs_bootstrap = plan.sidecar_needs_bootstrap(&sc); + start_one_sidecar(&args.start_ctx(), &tag, &sc, needs_bootstrap).await + } + }; + (name, sc, result) + })) + .await; + + for (name, sc, result) in started { + match result { + Ok(container) => { + containers.sidecars.insert(name, container); + } + Err(e) => warn_or_bail(plan, &sc, e)?, + } + } + Ok(()) +} + +/// Rebuild an owned error from an [`ImageResolution`]'s shared failure text. +/// Only the error's `Display` reaches the user (`app.rs` prints `error: {e}` and +/// always exits 1), so preserving the exact message -- not the original variant +/// -- is what keeps this observably identical to the serial path. `io::Error` +/// displays its payload verbatim (`OutrigError::Io` and `CliError::Outrig` are +/// both transparent); `OutrigError::Configuration` would instead prepend +/// "configuration: ". Revisit if exit codes ever become variant-dependent. +fn shared_image_error(message: &str) -> CliError { + CliError::from(std::io::Error::other(message.to_string())) +} + /// Start the network interceptor on the primary and attach every running /// sidecar. A sidecar attach failure follows its `on-failure`; `warn` stops /// and drops that sidecar. @@ -1178,6 +1293,16 @@ pub async fn teardown( mod tests { use super::*; + #[test] + fn shared_image_error_preserves_message_without_a_configuration_prefix() { + // The serial path surfaced an ensure/label failure verbatim; deduped + // Phase A keeps only its `Display` text, so re-wrapping must not add a + // variant prefix (`OutrigError::Configuration` would prepend + // "configuration: "). + let message = "process `buildah` exited with code 1"; + assert_eq!(shared_image_error(message).to_string(), message); + } + fn config_image(image_ref: &str) -> ImageConfig { ImageConfig { image_name: Some(image_ref.to_string()), diff --git a/crates/outrig-cli/src/cli/watcher.rs b/crates/outrig-cli/src/cli/watcher.rs index 4f9fd6f..084d9bb 100644 --- a/crates/outrig-cli/src/cli/watcher.rs +++ b/crates/outrig-cli/src/cli/watcher.rs @@ -2,94 +2,74 @@ //! //! Lifecycle coupling between the primary container and its sidecars is //! entirely outrig-managed (no pods, no `--requires`, no shared PID -//! namespaces). This module supplies the last coupling layer: a `podman wait` -//! on the primary for the session's lifetime, so a primary that dies out from -//! under outrig (manual `podman kill`, OOM) still reaps its sidecars and ends -//! the session with an error. Companion `podman wait`s per sidecar log -//! sidecar death promptly; nothing restarts. +//! namespaces). This module supplies the last coupling layer: a single +//! `podman events` stream, filtered to this session's containers, running for +//! the session's lifetime. A primary that dies out from under outrig (manual +//! `podman kill`, OOM) reaps its sidecars and ends the session with an error; +//! a sidecar death is logged; nothing restarts. One events process replaces +//! the former 1-plus-N `podman wait` children (`podman wait` given several +//! names waits for *all* of them, so it cannot batch a session's containers). //! //! Orderly teardown must call [`SessionWatcher::shutdown`] *before* stopping //! any container, so its own stops are never mistaken for external death. use std::process::Stdio; use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; +use serde::Deserialize; +use tokio::io::AsyncBufReadExt; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use crate::error::{CliError, Result}; -use outrig::container::force_remove_detached; +use outrig::container::{LABEL_SESSION, force_remove_detached}; /// Watches a session's containers. Spawned only for sessions that declare /// sidecars (started or `start = "manual"`) -- a single-container session -/// keeps today's behavior. The sidecar list is shared with the primary -/// reaper so a `/sidecar add` mid-session is covered from the moment it is -/// [registered](SessionWatcher::register_sidecar). +/// keeps today's behavior. The sidecar list is shared with the events reader +/// so a `/sidecar add` mid-session is covered from the moment it is +/// [registered](SessionWatcher::register_sidecar); the label-filtered stream +/// already delivers the new container's death, so registration only has to +/// record the name for the primary-death reap. #[derive(Debug)] pub struct SessionWatcher { died: CancellationToken, sidecars: Arc>>, - tasks: Vec>, + task: JoinHandle<()>, } impl SessionWatcher { - /// Arm the watcher: one `podman wait` on the primary (reaps every - /// registered sidecar and cancels [`SessionWatcher::primary_died`] when - /// it fires) plus one per sidecar (log-only). - pub fn spawn(primary: String, sidecars: Vec) -> Self { + /// Arm the watcher: one `podman events` reader for the whole session. It + /// reaps every registered sidecar and cancels [`SessionWatcher::primary_died`] + /// when the primary dies, and logs any sidecar death. `since` (the session + /// start) is replayed so a container that died between start and arm is not + /// missed. + pub fn spawn(primary: String, sidecars: Vec, sid: String, since: SystemTime) -> Self { let died = CancellationToken::new(); - let shared: Arc>> = Arc::new(Mutex::new(Vec::new())); - - let primary_task = { - let died = died.clone(); - let shared = Arc::clone(&shared); - tokio::spawn(async move { - wait_for_container_exit(&primary).await; - // Read the list when the wait fires, not when the watcher - // was armed, so dynamically added sidecars are reaped too. - let names = shared.lock().expect("sidecar name list lock").clone(); - for name in &names { - force_remove_detached(name); - } - eprintln!( - "[outrig] primary container {primary} exited unexpectedly; \ - reaping {} sidecar container(s)", - names.len() - ); - died.cancel(); - }) - }; - - let mut watcher = Self { + let sidecars: Arc>> = Arc::new(Mutex::new(sidecars)); + let task = tokio::spawn(watch_events( + primary, + sid, + since, + died.clone(), + Arc::clone(&sidecars), + )); + Self { died, - sidecars: shared, - tasks: vec![primary_task], - }; - for name in sidecars { - watcher.register_sidecar(name); + sidecars, + task, } - watcher } - /// Cover one more sidecar container: reaped when the primary dies, plus - /// a log-only `podman wait` announcing its own death. Called at arm time - /// for session-start sidecars and mid-session by `/sidecar add`. + /// Cover one more sidecar container: record its name so the primary-death + /// reap includes it. The session-filtered events stream already reports its + /// death, so no new process is needed. Called mid-session by `/sidecar add`. pub fn register_sidecar(&mut self, name: String) { self.sidecars .lock() .expect("sidecar name list lock") - .push(name.clone()); - self.tasks.push(tokio::spawn(async move { - wait_for_container_exit(&name).await; - eprintln!( - "[outrig] sidecar container {name} exited; \ - its MCP tools will return errors until the session ends" - ); - tracing::warn!( - target: "outrig::cli::watcher", - "sidecar container {name} exited mid-session" - ); - })); + .push(name); } /// Token cancelled when the primary dies externally. Clone into a @@ -98,12 +78,154 @@ impl SessionWatcher { self.died.clone() } - /// Disarm before orderly teardown. Aborting the tasks drops their - /// `podman wait` children (spawned `kill_on_drop`), so no watcher fires - /// for stops outrig performs itself. + /// Disarm before orderly teardown. Aborting the task drops its `podman + /// events` child (spawned `kill_on_drop`), so no watcher fires for stops + /// outrig performs itself. pub fn shutdown(self) { - for task in self.tasks { - task.abort(); + self.task.abort(); + } +} + +/// Where a `died` event should be routed. Pure classification; the membership +/// and reaping *policy* stays in [`watch_events`] so this stays testable. +#[derive(Debug, PartialEq, Eq)] +enum EventRoute { + PrimaryDied, + SidecarDied(String), + Ignore, +} + +/// The subset of a `podman events --format json` object we read. Unknown fields +/// (exit code, id, image, time, attributes) are ignored. +#[derive(Deserialize)] +struct PodmanEvent { + #[serde(rename = "Type")] + event_type: String, + #[serde(rename = "Status")] + status: String, + #[serde(rename = "Name")] + name: String, +} + +/// Classify one NDJSON event line. Defensive on `Type`/`Status` even though the +/// command already filters `event=died`; a malformed or unrelated line is +/// [`EventRoute::Ignore`]. +fn route_event(line: &str, primary: &str) -> EventRoute { + let Ok(event) = serde_json::from_str::(line) else { + return EventRoute::Ignore; + }; + if event.event_type != "container" || event.status != "died" { + return EventRoute::Ignore; + } + if event.name == primary { + EventRoute::PrimaryDied + } else { + EventRoute::SidecarDied(event.name) + } +} + +/// Read this session's `died` events until the primary dies or the stream ends. +/// Filtered by `event=died` + the session label, and replayed from `since` so a +/// death during setup is caught. A sidecar-death line is emitted only for a +/// name in the shared list: `--since` replays setup-time deaths of warn-path +/// sidecars that were created then dropped and never registered, which must +/// stay silent. +async fn watch_events( + primary: String, + sid: String, + since: SystemTime, + died: CancellationToken, + sidecars: Arc>>, +) { + let since_secs = since + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) + .to_string(); + let label_filter = format!("label={LABEL_SESSION}={sid}"); + let mut child = match tokio::process::Command::new("podman") + .args(["events", "--since", &since_secs]) + .args(["--filter", "event=died", "--filter", &label_filter]) + .args(["--format", "json"]) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .spawn() + { + Ok(child) => child, + Err(e) => { + tracing::warn!( + target: "outrig::cli::watcher", + "podman events for session {sid} failed to spawn: {e}; auto-reap disabled" + ); + return; + } + }; + let Some(stdout) = child.stdout.take() else { + tracing::warn!( + target: "outrig::cli::watcher", + "podman events for session {sid} produced no stdout; auto-reap disabled" + ); + return; + }; + + let mut lines = tokio::io::BufReader::new(stdout).lines(); + loop { + match lines.next_line().await { + Ok(Some(line)) => match route_event(&line, &primary) { + EventRoute::PrimaryDied => { + // Snapshot when the death fires, not at arm time, so + // dynamically added sidecars are reaped too. + let names = sidecars.lock().expect("sidecar name list lock").clone(); + for name in &names { + force_remove_detached(name); + } + eprintln!( + "[outrig] primary container {primary} exited unexpectedly; \ + reaping {} sidecar container(s)", + names.len() + ); + died.cancel(); + // Stop reading: the reap generates further `died` events we + // do not want to log as spontaneous sidecar deaths. + return; + } + EventRoute::SidecarDied(name) => { + let known = sidecars + .lock() + .expect("sidecar name list lock") + .contains(&name); + if known { + eprintln!( + "[outrig] sidecar container {name} exited; \ + its MCP tools will return errors until the session ends" + ); + tracing::warn!( + target: "outrig::cli::watcher", + "sidecar container {name} exited mid-session" + ); + } + } + EventRoute::Ignore => {} + }, + // Stream ended without a primary death. Degrade to today's + // single-container "no auto-reap"; never cancel -- the session is + // still alive. + Ok(None) => { + tracing::warn!( + target: "outrig::cli::watcher", + "podman events stream for session {sid} ended; auto-reap disabled" + ); + return; + } + Err(e) => { + tracing::warn!( + target: "outrig::cli::watcher", + "reading podman events for session {sid} failed: {e}; auto-reap disabled" + ); + return; + } } } } @@ -129,7 +251,8 @@ pub fn exit_if_monitor_stopped(outcome: &Result, final_exit: i32) { } /// Block until the named container exits. `podman wait` returning an error -/// (e.g. no such container) is treated as "already gone". +/// (e.g. no such container) is treated as "already gone". Used by attach-mode +/// `outrig mcp`, whose single borrowed container needs no events stream. pub(crate) async fn wait_for_container_exit(name: &str) { let child = tokio::process::Command::new("podman") .args(["wait", name]) @@ -150,3 +273,41 @@ pub(crate) async fn wait_for_container_exit(name: &str) { } } } + +#[cfg(test)] +mod tests { + use super::*; + + const PRIMARY: &str = "outrig-20260713T173119-fa97"; + + #[test] + fn route_event_matches_primary_by_name() { + let line = r#"{"ContainerExitCode":137,"ID":"7d3c","Image":"localhost/x:latest","Name":"outrig-20260713T173119-fa97","Status":"died","Time":"2026-07-13T11:31:49-06:00","Type":"container","Attributes":{"io.buildah.version":"1.33.7"}}"#; + assert_eq!(route_event(line, PRIMARY), EventRoute::PrimaryDied); + } + + #[test] + fn route_event_routes_other_names_to_sidecar() { + let line = r#"{"ContainerExitCode":137,"ID":"4ccc","Image":"localhost/x:latest","Name":"outrig-20260713T173119-fa97-tools","Status":"died","Time":"2026-07-13T11:31:47-06:00","Type":"container","Attributes":{"org.outrig.session":"20260713T173119-fa97","org.outrig.sidecar":"tools"}}"#; + assert_eq!( + route_event(line, PRIMARY), + EventRoute::SidecarDied("outrig-20260713T173119-fa97-tools".to_string()) + ); + } + + #[test] + fn route_event_ignores_malformed_json() { + assert_eq!(route_event("not json", PRIMARY), EventRoute::Ignore); + assert_eq!(route_event("", PRIMARY), EventRoute::Ignore); + } + + #[test] + fn route_event_ignores_non_died_and_non_container_events() { + let started = + r#"{"Name":"outrig-20260713T173119-fa97","Status":"start","Type":"container"}"#; + assert_eq!(route_event(started, PRIMARY), EventRoute::Ignore); + let image_event = + r#"{"Name":"outrig-20260713T173119-fa97","Status":"died","Type":"image"}"#; + assert_eq!(route_event(image_event, PRIMARY), EventRoute::Ignore); + } +} diff --git a/crates/outrig-cli/tests/session_cli.rs b/crates/outrig-cli/tests/session_cli.rs index 5a50f6b..84ebf7e 100644 --- a/crates/outrig-cli/tests/session_cli.rs +++ b/crates/outrig-cli/tests/session_cli.rs @@ -6,6 +6,7 @@ //! No podman dependency: the discard tests inject their own running-check //! closure. +use std::collections::BTreeSet; use std::path::PathBuf; use std::time::{Duration, SystemTime}; @@ -634,7 +635,7 @@ async fn clean_default_30d_removes_only_old_finished_sessions() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), Vec::new(), |_| async { Ok(()) }, ) @@ -692,7 +693,7 @@ async fn clean_custom_older_than_uses_requested_cutoff() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), Vec::new(), |_| async { Ok(()) }, ) @@ -729,7 +730,7 @@ async fn clean_without_yes_aborts_on_n() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), Vec::new(), |_| async { Ok(()) }, ) @@ -767,7 +768,7 @@ async fn clean_accepts_yes_at_prompt() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), Vec::new(), |_| async { Ok(()) }, ) @@ -798,10 +799,7 @@ async fn clean_skips_running_sessions() { &store, &args, now, - move |name| { - let is_running = name == running.as_str(); - async move { Ok(is_running) } - }, + BTreeSet::from([running]), Vec::new(), |_| async { Ok(()) }, ) @@ -837,7 +835,7 @@ async fn clean_removes_stale_unfinalized_non_running_sessions() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), Vec::new(), |_| async { Ok(()) }, ) @@ -880,7 +878,7 @@ async fn clean_removes_symlinked_session_target_and_link() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), Vec::new(), |_| async { Ok(()) }, ) @@ -970,8 +968,8 @@ async fn clean_removes_old_stopped_stray_containers() { labeled("outrig-young", "young", None, false, days(1), now), ]; - let removed: std::sync::Arc>> = Default::default(); - let removed_ref = removed.clone(); + let batches: std::sync::Arc>>> = Default::default(); + let batches_ref = batches.clone(); let (mut ew, stderr_r) = duplex(8192); let stdin = tokio::io::BufReader::new(tokio::io::empty()); let args = clean_args(clean::DEFAULT_OLDER_THAN, true); @@ -981,12 +979,12 @@ async fn clean_removes_old_stopped_stray_containers() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), strays, - move |name| { - let removed = removed_ref.clone(); + move |names: Vec| { + let batches = batches_ref.clone(); async move { - removed.lock().unwrap().push(name); + batches.lock().unwrap().push(names); Ok(()) } }, @@ -996,14 +994,14 @@ async fn clean_removes_old_stopped_stray_containers() { drop(ew); assert_eq!(rc, 0); - let removed = removed.lock().unwrap().clone(); + let batches = batches.lock().unwrap().clone(); assert_eq!( - removed, - vec![ + batches, + vec![vec![ "outrig-lost-1".to_string(), "outrig-lost-1-tools".to_string() - ], - "only the old stopped strays should be removed" + ]], + "both old stopped strays should be removed in a single batched podman rm -f" ); let err = drain(stderr_r).await; assert!( @@ -1050,8 +1048,8 @@ async fn clean_never_removes_running_or_record_backed_containers() { labeled("outrig-orphan", "orphan-sid", None, true, days(40), now), ]; - let removed: std::sync::Arc>> = Default::default(); - let removed_ref = removed.clone(); + let batches: std::sync::Arc>>> = Default::default(); + let batches_ref = batches.clone(); let (mut ew, stderr_r) = duplex(8192); let stdin = tokio::io::BufReader::new(tokio::io::empty()); let args = clean_args(clean::DEFAULT_OLDER_THAN, true); @@ -1061,12 +1059,15 @@ async fn clean_never_removes_running_or_record_backed_containers() { &store, &args, now, - |_| async { Ok(true) }, + BTreeSet::from([ + "outrig-20260501T134412-3f2a".to_string(), + "outrig-orphan".to_string(), + ]), strays, - move |name| { - let removed = removed_ref.clone(); + move |names: Vec| { + let batches = batches_ref.clone(); async move { - removed.lock().unwrap().push(name); + batches.lock().unwrap().push(names); Ok(()) } }, @@ -1076,7 +1077,7 @@ async fn clean_never_removes_running_or_record_backed_containers() { drop(ew); assert!( - removed.lock().unwrap().is_empty(), + batches.lock().unwrap().is_empty(), "no container should be removed" ); let err = drain(stderr_r).await; @@ -1117,8 +1118,8 @@ async fn clean_sweeps_strays_of_records_removed_in_same_run() { now, )]; - let removed: std::sync::Arc>> = Default::default(); - let removed_ref = removed.clone(); + let batches: std::sync::Arc>>> = Default::default(); + let batches_ref = batches.clone(); let (mut ew, stderr_r) = duplex(8192); let stdin = tokio::io::BufReader::new(tokio::io::empty()); let args = clean_args(clean::DEFAULT_OLDER_THAN, true); @@ -1128,12 +1129,12 @@ async fn clean_sweeps_strays_of_records_removed_in_same_run() { &store, &args, now, - |_| async { Ok(false) }, + BTreeSet::new(), strays, - move |name| { - let removed = removed_ref.clone(); + move |names: Vec| { + let batches = batches_ref.clone(); async move { - removed.lock().unwrap().push(name); + batches.lock().unwrap().push(names); Ok(()) } }, @@ -1143,8 +1144,8 @@ async fn clean_sweeps_strays_of_records_removed_in_same_run() { drop(ew); assert_eq!( - removed.lock().unwrap().clone(), - vec!["outrig-20260501T134412-3f2a".to_string()], + batches.lock().unwrap().clone(), + vec![vec!["outrig-20260501T134412-3f2a".to_string()]], "the leftover container of a removed record should be swept" ); let err = drain(stderr_r).await; diff --git a/plan/done/0086-sidecar-startup-performance.md b/plan/done/0086-sidecar-startup-performance.md new file mode 100644 index 0000000..d635c8d --- /dev/null +++ b/plan/done/0086-sidecar-startup-performance.md @@ -0,0 +1,82 @@ +# 0086 -- Sidecar startup and clean-sweep performance follow-ups + +## Goal + +Land the deferred efficiency items from task 0079's review pass; none block correctness. + +## Deliverables + +- **Concurrent sidecar bring-up.** `setup_sidecars_and_network` runs image-ensure -> + label read -> `podman run` -> bootstrap fully serially per sidecar. Sidecars are + independent; fan the ensure/label phase out (`JoinSet`), apply `merge_sidecar_labels` + over the collected results in name order (keeps deterministic collision errors), then + start containers concurrently. The ensure phase could also overlap the primary's + bootstrap. +- **Memoize duplicate sidecar images.** Two sidecars sharing one `image` re-run tag + compute + ensure + label inspect. A per-session `image -> (tag, labels)` map fixes it. +- **One watcher process per session.** The watcher holds 1 + N idle `podman wait` + children. `podman events --filter event=died --filter label=org.outrig.session=` + watches every session container with a single process (`podman wait` with multiple + args waits for *all*, so it cannot batch). +- **`outrig clean` running-checks.** The record walk still spawns one + `podman ps`-equivalent per aged session while the new label sweep already fetched + every labeled container's state in one `podman ps -a` call; answer record-backed + running-state from that listing and keep per-name probes only for pre-label sessions. + Batch the stray `podman rm -f` calls into one invocation while there. + +## Acceptance + +- Observable behavior is unchanged: same tools, same deterministic label-collision + errors, same session teardown semantics. +- Sidecar bring-up overlaps across sidecars rather than running serially. +- One watcher process per session instead of 1 + N `podman wait` children. +- `outrig clean` answers record-backed running-state from the single `podman ps -a` + listing and batches its `podman rm -f` calls. + +## Dependencies + +None (follows up on completed task 0079; sequenced after 0085 since both touch +`setup_sidecars_and_network`). + +## Decisions + +Made during planning/execution (2026-07-13): + +- **`join_all`, not `JoinSet`.** The codebase's only fan-out idiom is + `futures_util::future::join_all` (`network.rs`); there is no `JoinSet`. `join_all` + preserves input order (so the name-ordered label merge stays deterministic) and runs on + the current task, so the borrowed `&Config`/`&Path`/`&Transcript`/`&plan` need no + `'static`/`Send` and nothing is cloned to satisfy a spawn bound. +- **Three-phase bring-up.** `setup_sidecars_and_network` fans out image-ensure + label-read + (Phase A, deduped by image ref -> memoization), then merges labels + decides starts + serially in name order (Phase B, preserves deterministic collision errors), then starts + containers concurrently and inserts in name order (Phase C). Network attach stays serial. +- **Memoized label read gated on `need_labels`.** An image is label-inspected only if a + *non-anonymous* sidecar uses it. This is a correctness constraint, not just perf: eagerly + inspecting an anonymous-only image would run a `podman image inspect` the serial path + never ran and could newly fail a session on a malformed `org.outrig.mcp` label. +- **Shared image failures kept as `Display` text, re-wrapped via `io::Error::other`.** + `CliError` is not `Clone`, so a deduped image's failure is stored as its message string + and rebuilt per consumer. `OutrigError::Configuration` is *not* transparent (it prepends + `"configuration: "`), so reconstructing through it would alter the error text; + `io::Error` (via transparent `OutrigError::Io`/`CliError::Outrig`) round-trips the message + verbatim. Only `Display` + exit-1 are observable (`app.rs`), so the variant change is not. +- **One `podman events` watcher per session** replaces the 1+N `podman wait` children: + `podman events --since --filter event=died --filter + label=org.outrig.session= --format json`. `--since` replays a death that landed + during the arm window; a **shared-list membership gate** on the sidecar-death line + suppresses the replay of setup-time deaths of warn-dropped sidecars (which were never + registered). The reader stops after the primary dies (so the reap's own sidecar deaths + are not logged as spontaneous) and, on unexpected stream EOF, degrades (warns, never + cancels, no respawn) to today's single-container "no auto-reap". `register_sidecar` + becomes push-only; the label-filtered stream already covers a `/sidecar add`. + `wait_for_container_exit` stays for attach-mode `outrig mcp`. +- **`outrig clean` running-state from one unfiltered `podman ps -a` (Option 1).** The single + listing yields both the labeled stray rows (unchanged `classify_strays` input) and the set + of *all* running container names. The record walk answers running-state by name from that + set instead of a `podman inspect` per aged session. Unfiltered because a record's primary + can be outrig-**unlabeled** -- not only pre-0079 sessions but every `outrig mcp --attach` + session (it records the borrowed container's name) -- and a label-filtered sweep would + miss those. Stray `podman rm -f`s are batched into one invocation. +- **Primary-bootstrap overlap deferred** to `plan/next/sidecar-primary-bootstrap-overlap.md` + (bounded ~100ms win against restructuring the primary abort path). diff --git a/plan/next/sidecar-primary-bootstrap-overlap.md b/plan/next/sidecar-primary-bootstrap-overlap.md new file mode 100644 index 0000000..5c79263 --- /dev/null +++ b/plan/next/sidecar-primary-bootstrap-overlap.md @@ -0,0 +1,21 @@ +# Overlap sidecar image-ensure with primary bootstrap + +Task 0086 made sidecar bring-up concurrent *across sidecars* (`setup_sidecars_and_network` +now fans out image-ensure/label-read in Phase A, then starts containers in Phase C). It +deliberately left one overlap on the table: the sidecar Phase-A ensure work still begins +only *after* the primary is started and `bootstrap_user`'d (`setup`, around the +`Container::start_named` -> `bootstrap_user` sequence for the primary), even though the two +are independent -- sidecar image-ensure/label-read depend only on `image_cfg`/config, not +on the primary container existing. + +The win is bounded: `bootstrap_user` is a short chain of `podman exec` round-trips (~100ms), +so overlapping it with sidecar ensure saves at most ~that, and only for sessions that +declare sidecars (single-container sessions have nothing to overlap). The cost is +restructuring the primary path's delicate abort tail (stop primary + finalize row + return +on bootstrap failure). + +If pursued: split Phase A into a pure ensure/label-read future (no `plan` mutation, no +container starts) and run it as `tokio::join!(container.bootstrap_user(), phase_a_ensure)`, +keeping the label-merge (Phase B) and container starts (Phase C) after the join. Preserve +the primary-bootstrap abort semantics exactly: a primary `bootstrap_user` failure must still +stop the primary and finalize the row with exit 1, regardless of the sidecar future's state. diff --git a/plan/todo/0086-sidecar-startup-performance.md b/plan/todo/0086-sidecar-startup-performance.md deleted file mode 100644 index faf6350..0000000 --- a/plan/todo/0086-sidecar-startup-performance.md +++ /dev/null @@ -1,39 +0,0 @@ -# 0086 -- Sidecar startup and clean-sweep performance follow-ups - -## Goal - -Land the deferred efficiency items from task 0079's review pass; none block correctness. - -## Deliverables - -- **Concurrent sidecar bring-up.** `setup_sidecars_and_network` runs image-ensure -> - label read -> `podman run` -> bootstrap fully serially per sidecar. Sidecars are - independent; fan the ensure/label phase out (`JoinSet`), apply `merge_sidecar_labels` - over the collected results in name order (keeps deterministic collision errors), then - start containers concurrently. The ensure phase could also overlap the primary's - bootstrap. -- **Memoize duplicate sidecar images.** Two sidecars sharing one `image` re-run tag - compute + ensure + label inspect. A per-session `image -> (tag, labels)` map fixes it. -- **One watcher process per session.** The watcher holds 1 + N idle `podman wait` - children. `podman events --filter event=died --filter label=org.outrig.session=` - watches every session container with a single process (`podman wait` with multiple - args waits for *all*, so it cannot batch). -- **`outrig clean` running-checks.** The record walk still spawns one - `podman ps`-equivalent per aged session while the new label sweep already fetched - every labeled container's state in one `podman ps -a` call; answer record-backed - running-state from that listing and keep per-name probes only for pre-label sessions. - Batch the stray `podman rm -f` calls into one invocation while there. - -## Acceptance - -- Observable behavior is unchanged: same tools, same deterministic label-collision - errors, same session teardown semantics. -- Sidecar bring-up overlaps across sidecars rather than running serially. -- One watcher process per session instead of 1 + N `podman wait` children. -- `outrig clean` answers record-backed running-state from the single `podman ps -a` - listing and batches its `podman rm -f` calls. - -## Dependencies - -None (follows up on completed task 0079; sequenced after 0085 since both touch -`setup_sidecars_and_network`). diff --git a/plan/todo/README.md b/plan/todo/README.md index 8a49baa..7c8c5cb 100644 --- a/plan/todo/README.md +++ b/plan/todo/README.md @@ -8,4 +8,3 @@ See [`.claude/CLAUDE.md`](../../.claude/CLAUDE.md) for the workflow conventions. | Task | Title | Dependencies | |------|----------------------------------------------------|--------------| -| 0086 | Sidecar startup and clean-sweep performance | -- |