diff --git a/Cargo.lock b/Cargo.lock index 109cc4b..9e47eb6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -825,7 +825,7 @@ checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" [[package]] name = "piperack" -version = "0.1.4" +version = "0.2.0" dependencies = [ "anyhow", "arboard", diff --git a/docs/testing.md b/docs/testing.md index 417f64c..cae44d3 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -39,6 +39,9 @@ cargo llvm-cov - Verify shutdown UX: - `q` shows persistent "shutting down" status and exits cleanly. - `k` shows "sent SIGINT" immediately for a process. +- Verify high-volume output doesn't freeze the UI: + - `cargo run -- -- --name spam -- sh -c "yes | head -n 100000"` + - UI stays responsive; you can still select and kill the process. - Verify clipboard selection: - Mouse drag selects log lines. - Ctrl+C copies selection; if none, copies full selected process buffer. diff --git a/src/main.rs b/src/main.rs index 5858a8b..561f297 100644 --- a/src/main.rs +++ b/src/main.rs @@ -176,8 +176,15 @@ async fn main() -> Result<()> { } let (event_tx, mut event_rx) = mpsc::channel(256); + let (output_tx, mut output_rx) = mpsc::channel(256); let shutdown = ShutdownConfig::new(settings.shutdown_sigint_ms, settings.shutdown_sigterm_ms); - let mut manager = ProcessManager::new(specs.clone(), event_tx.clone(), shutdown); + let mut manager = ProcessManager::new( + specs.clone(), + event_tx.clone(), + output_tx.clone(), + shutdown, + !settings.no_ui, + ); let mut app = App::new( specs, settings.max_lines, @@ -194,6 +201,8 @@ async fn main() -> Result<()> { Some(tui::init_terminal()?) }; let tick_rate = Duration::from_millis(150); + let draw_interval = Duration::from_millis(33); + let mut last_draw = Instant::now() - draw_interval; if !settings.no_ui { spawn_input_listener(event_tx.clone()); @@ -215,301 +224,10 @@ async fn main() -> Result<()> { let mut last_signal_at: Option = None; loop { - tokio::select! { - Some(event) = event_rx.recv() => { - match event { - Event::ProcessStarting { id } => { - app.on_process_starting(id); - let name = app - .processes - .get(id) - .map(|p| p.spec.name.as_str()) - .unwrap_or("process"); - app.set_status_message(format!("Starting {}", name)); - if let Some(spec) = app.processes.get(id).map(|p| &p.spec) { - let cmd = format_command(spec); - emit_tool_message( - id, - format!("starting: {}", cmd), - &mut app, - &settings, - &mut output_state, - ); - } - } - Event::ProcessStarted { id, pid } => app.on_process_started(id, pid), - Event::ProcessReady { id } => { - app.on_process_ready(id); - let name = app - .processes - .get(id) - .map(|p| p.spec.name.as_str()) - .unwrap_or("process"); - app.set_status_message(format!("{} ready", name)); - emit_tool_message( - id, - "ready".to_string(), - &mut app, - &settings, - &mut output_state, - ); - if let Err(e) = manager.mark_ready(id).await { - app.on_process_failed(id, e.to_string()); - } - } - Event::ProcessWaiting { id, deps } => { - let name = app - .processes - .get(id) - .map(|p| p.spec.name.as_str()) - .unwrap_or("process"); - let waiting_on = if deps.is_empty() { - "dependencies".to_string() - } else { - deps.join(", ") - }; - app.set_status_message(format!("{} waiting for {}", name, waiting_on)); - emit_tool_message( - id, - format!("waiting for {}", waiting_on), - &mut app, - &settings, - &mut output_state, - ); - } - Event::ProcessOutput { id, line, stream } => { - let line_for_output = line.clone(); - app.on_process_output(id, line, stream); - if settings.no_ui { - output_state.handle_event( - &Event::ProcessOutput { id, line: line_for_output, stream }, - &app, - &settings, - ); - } else { - output_state.log_event(id, &line_for_output, &app, &settings); - } - } - Event::ProcessExited { id, code } => { - app.on_process_exited(id, code); - let name = app - .processes - .get(id) - .map(|p| p.spec.name.as_str()) - .unwrap_or("process"); - if !shutdown_in_progress { - let signal_recent = last_signal_at - .map(|at| at.elapsed() < MIN_SIGNAL_DISPLAY) - .unwrap_or(false); - if !signal_recent { - let message = match code { - Some(0) => format!("{} exited successfully", name), - Some(code) => format!("{} exited with code {}", name, code), - None => format!("{} exited", name), - }; - app.set_status_message(message); - } - } - let line = match code { - Some(0) => "process ended successfully".to_string(), - Some(code) => format!("process ended with code {}", code), - None => "process ended".to_string(), - }; - emit_tool_message(id, line, &mut app, &settings, &mut output_state); - let restart_info = if shutdown_in_progress { - None - } else { - handle_restart( - id, - code, - &app, - &settings, - &mut restart_attempts, - &event_tx, - ) - }; - if let Some(info) = restart_info { - emit_tool_message( - id, - format_restart_message(&info), - &mut app, - &settings, - &mut output_state, - ); - } - if shutdown_in_progress { - output_state.handle_exit(id, code); - let ready_to_exit = output_state.all_exited() - && shutdown_started_at - .map(|start| start.elapsed() >= MIN_SHUTDOWN_DISPLAY) - .unwrap_or(false); - if ready_to_exit { - app.should_quit = true; - } - } else { - handle_exit_policy( - id, - code, - &mut app, - &settings, - &mut output_state, - &mut manager, - &mut result, - ) - .await; - } - } - Event::ProcessFailed { id, error } => { - let error_message = error.clone(); - app.on_process_failed(id, error); - let name = app - .processes - .get(id) - .map(|p| p.spec.name.as_str()) - .unwrap_or("process"); - if !shutdown_in_progress { - let signal_recent = last_signal_at - .map(|at| at.elapsed() < MIN_SIGNAL_DISPLAY) - .unwrap_or(false); - if !signal_recent { - let message = format!("{} failed: {}", name, error_message); - app.set_status_message(message); - } - } - emit_tool_message( - id, - format!("process failed: {}", error_message), - &mut app, - &settings, - &mut output_state, - ); - let restart_info = if shutdown_in_progress { - None - } else { - handle_restart( - id, - Some(1), - &app, - &settings, - &mut restart_attempts, - &event_tx, - ) - }; - if let Some(info) = restart_info { - emit_tool_message( - id, - format_restart_message(&info), - &mut app, - &settings, - &mut output_state, - ); - } - if shutdown_in_progress { - output_state.handle_exit(id, Some(1)); - let ready_to_exit = output_state.all_exited() - && shutdown_started_at - .map(|start| start.elapsed() >= MIN_SHUTDOWN_DISPLAY) - .unwrap_or(false); - if ready_to_exit { - app.should_quit = true; - } - } else { - handle_exit_policy( - id, - Some(1), - &mut app, - &settings, - &mut output_state, - &mut manager, - &mut result, - ) - .await; - } - } - Event::ProcessSignal { id, signal } => { - let name = app - .processes - .get(id) - .map(|p| p.spec.name.as_str()) - .unwrap_or("process"); - let label = signal.label(); - if shutdown_in_progress || shutdown_pending.is_some() { - app.set_status_warning_persistent(format!( - "shutting down — sent {} to {}", - label, name - )); - } else { - last_signal_at = Some(Instant::now()); - app.set_status_warning_for( - format!("sent {} to {}", label, name), - MIN_SIGNAL_DISPLAY, - ); - } - emit_tool_message( - id, - format!("sent {}", label), - &mut app, - &settings, - &mut output_state, - ); - } - Event::Restart { id } => { - if let Err(err) = manager.restart_process(id).await { - app.on_process_failed(id, err.to_string()); - } - } - Event::Shutdown { signal } => { - if !shutdown_in_progress { - let label = signal.label(); - app.should_quit = false; - app.set_status_warning_persistent(format!( - "received {}, shutting down", - label - )); - shutdown_pending = Some(signal); - shutdown_dispatch_at = if settings.no_ui { - Some(Instant::now()) - } else { - None - }; - } - } - Event::Stdin(bytes) => { - if let Err(err) = manager.send_input_bytes_to_all(&bytes).await { - app.set_status_message(format!("Input failed: {}", err)); - } - } - Event::Key(key) => { - let action = app.handle_key(key); - handle_app_action( - action, - &mut app, - &mut manager, - &mut restart_attempts, - &event_tx, - ) - .await; - } - Event::Mouse(mouse) => { - let action = app.handle_mouse(mouse); - handle_app_action( - action, - &mut app, - &mut manager, - &mut restart_attempts, - &event_tx, - ) - .await; - } - Event::Resize { width, height } => { - let _ = (width, height); - if let Some(term) = terminal.as_mut() { - let _ = term.autoresize(); - } - } - } - - } + let event = tokio::select! { + biased; + Some(event) = event_rx.recv() => event, + Some(event) = output_rx.recv() => event, _ = ticker.tick() => { manager.poll_exits().await; if let Some(signal) = shutdown_pending { @@ -532,13 +250,309 @@ async fn main() -> Result<()> { { app.should_quit = true; } + continue; + } + }; + + match event { + Event::ProcessStarting { id } => { + app.on_process_starting(id); + let name = app + .processes + .get(id) + .map(|p| p.spec.name.as_str()) + .unwrap_or("process"); + app.set_status_message(format!("Starting {}", name)); + if let Some(spec) = app.processes.get(id).map(|p| &p.spec) { + let cmd = format_command(spec); + emit_tool_message( + id, + format!("starting: {}", cmd), + &mut app, + &settings, + &mut output_state, + ); + } + } + Event::ProcessStarted { id, pid } => app.on_process_started(id, pid), + Event::ProcessReady { id } => { + app.on_process_ready(id); + let name = app + .processes + .get(id) + .map(|p| p.spec.name.as_str()) + .unwrap_or("process"); + app.set_status_message(format!("{} ready", name)); + emit_tool_message( + id, + "ready".to_string(), + &mut app, + &settings, + &mut output_state, + ); + if let Err(e) = manager.mark_ready(id).await { + app.on_process_failed(id, e.to_string()); + } + } + Event::ProcessWaiting { id, deps } => { + let name = app + .processes + .get(id) + .map(|p| p.spec.name.as_str()) + .unwrap_or("process"); + let waiting_on = if deps.is_empty() { + "dependencies".to_string() + } else { + deps.join(", ") + }; + app.set_status_message(format!("{} waiting for {}", name, waiting_on)); + emit_tool_message( + id, + format!("waiting for {}", waiting_on), + &mut app, + &settings, + &mut output_state, + ); + } + Event::ProcessOutput { id, line, stream } => { + let line_for_output = line.clone(); + app.on_process_output(id, line, stream); + if settings.no_ui { + output_state.handle_event( + &Event::ProcessOutput { id, line: line_for_output, stream }, + &app, + &settings, + ); + } else { + output_state.log_event(id, &line_for_output, &app, &settings); + } + } + Event::ProcessExited { id, code } => { + app.on_process_exited(id, code); + let name = app + .processes + .get(id) + .map(|p| p.spec.name.as_str()) + .unwrap_or("process"); + if !shutdown_in_progress { + let signal_recent = last_signal_at + .map(|at| at.elapsed() < MIN_SIGNAL_DISPLAY) + .unwrap_or(false); + if !signal_recent { + let message = match code { + Some(0) => format!("{} exited successfully", name), + Some(code) => format!("{} exited with code {}", name, code), + None => format!("{} exited", name), + }; + app.set_status_message(message); + } + } + let line = match code { + Some(0) => "process ended successfully".to_string(), + Some(code) => format!("process ended with code {}", code), + None => "process ended".to_string(), + }; + emit_tool_message(id, line, &mut app, &settings, &mut output_state); + let restart_info = if shutdown_in_progress { + None + } else { + handle_restart( + id, + code, + &app, + &settings, + &mut restart_attempts, + &event_tx, + ) + }; + if let Some(info) = restart_info { + emit_tool_message( + id, + format_restart_message(&info), + &mut app, + &settings, + &mut output_state, + ); + } + if shutdown_in_progress { + output_state.handle_exit(id, code); + let ready_to_exit = output_state.all_exited() + && shutdown_started_at + .map(|start| start.elapsed() >= MIN_SHUTDOWN_DISPLAY) + .unwrap_or(false); + if ready_to_exit { + app.should_quit = true; + } + } else { + handle_exit_policy( + id, + code, + &mut app, + &settings, + &mut output_state, + &mut manager, + &mut result, + ) + .await; + } + } + Event::ProcessFailed { id, error } => { + let error_message = error.clone(); + app.on_process_failed(id, error); + let name = app + .processes + .get(id) + .map(|p| p.spec.name.as_str()) + .unwrap_or("process"); + if !shutdown_in_progress { + let signal_recent = last_signal_at + .map(|at| at.elapsed() < MIN_SIGNAL_DISPLAY) + .unwrap_or(false); + if !signal_recent { + let message = format!("{} failed: {}", name, error_message); + app.set_status_message(message); + } + } + emit_tool_message( + id, + format!("process failed: {}", error_message), + &mut app, + &settings, + &mut output_state, + ); + let restart_info = if shutdown_in_progress { + None + } else { + handle_restart( + id, + Some(1), + &app, + &settings, + &mut restart_attempts, + &event_tx, + ) + }; + if let Some(info) = restart_info { + emit_tool_message( + id, + format_restart_message(&info), + &mut app, + &settings, + &mut output_state, + ); + } + if shutdown_in_progress { + output_state.handle_exit(id, Some(1)); + let ready_to_exit = output_state.all_exited() + && shutdown_started_at + .map(|start| start.elapsed() >= MIN_SHUTDOWN_DISPLAY) + .unwrap_or(false); + if ready_to_exit { + app.should_quit = true; + } + } else { + handle_exit_policy( + id, + Some(1), + &mut app, + &settings, + &mut output_state, + &mut manager, + &mut result, + ) + .await; + } + } + Event::ProcessSignal { id, signal } => { + let name = app + .processes + .get(id) + .map(|p| p.spec.name.as_str()) + .unwrap_or("process"); + let label = signal.label(); + if shutdown_in_progress || shutdown_pending.is_some() { + app.set_status_warning_persistent(format!( + "shutting down — sent {} to {}", + label, name + )); + } else { + last_signal_at = Some(Instant::now()); + app.set_status_warning_for( + format!("sent {} to {}", label, name), + MIN_SIGNAL_DISPLAY, + ); + } + emit_tool_message( + id, + format!("sent {}", label), + &mut app, + &settings, + &mut output_state, + ); + } + Event::Restart { id } => { + if let Err(err) = manager.restart_process(id).await { + app.on_process_failed(id, err.to_string()); + } + } + Event::Shutdown { signal } => { + if !shutdown_in_progress { + let label = signal.label(); + app.should_quit = false; + app.set_status_warning_persistent(format!( + "received {}, shutting down", + label + )); + shutdown_pending = Some(signal); + shutdown_dispatch_at = if settings.no_ui { + Some(Instant::now()) + } else { + None + }; + } + } + Event::Stdin(bytes) => { + if let Err(err) = manager.send_input_bytes_to_all(&bytes).await { + app.set_status_message(format!("Input failed: {}", err)); + } + } + Event::Key(key) => { + let action = app.handle_key(key); + handle_app_action( + action, + &mut app, + &mut manager, + &mut restart_attempts, + &event_tx, + ) + .await; + } + Event::Mouse(mouse) => { + let action = app.handle_mouse(mouse); + handle_app_action( + action, + &mut app, + &mut manager, + &mut restart_attempts, + &event_tx, + ) + .await; + } + Event::Resize { width, height } => { + let _ = (width, height); + if let Some(term) = terminal.as_mut() { + let _ = term.autoresize(); + } } } if let Some(term) = terminal.as_mut() { - if let Err(err) = tui::draw(&mut app, term) { - result = Err(err.into()); - break; + if last_draw.elapsed() >= draw_interval { + if let Err(err) = tui::draw(&mut app, term) { + result = Err(err.into()); + break; + } + last_draw = Instant::now(); } } if shutdown_pending.is_some() && shutdown_dispatch_at.is_none() && !settings.no_ui { diff --git a/src/runner.rs b/src/runner.rs index 5a38e85..32422e4 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -23,7 +23,9 @@ use crate::process::ProcessSpec; pub struct ProcessManager { processes: Vec, event_tx: mpsc::Sender, + output_tx: mpsc::Sender, shutdown: ShutdownConfig, + lossy_output: bool, } struct ManagedProcess { @@ -85,7 +87,9 @@ impl ProcessManager { pub fn new( specs: Vec, event_tx: mpsc::Sender, + output_tx: mpsc::Sender, shutdown: ShutdownConfig, + lossy_output: bool, ) -> Self { let processes = specs .into_iter() @@ -102,7 +106,9 @@ impl ProcessManager { Self { processes, event_tx, + output_tx, shutdown, + lossy_output, } } @@ -231,14 +237,30 @@ impl ProcessManager { }; if let Some(stdout) = child.stdout.take() { - let tx = self.event_tx.clone(); + let tx = self.output_tx.clone(); let regex = log_ready_regex.clone(); - tokio::spawn(read_stream(id, StreamKind::Stdout, stdout, tx, regex)); + let lossy_output = self.lossy_output; + tokio::spawn(read_stream( + id, + StreamKind::Stdout, + stdout, + tx, + regex, + lossy_output, + )); } if let Some(stderr) = child.stderr.take() { - let tx = self.event_tx.clone(); + let tx = self.output_tx.clone(); let regex = log_ready_regex; // move last clone - tokio::spawn(read_stream(id, StreamKind::Stderr, stderr, tx, regex)); + let lossy_output = self.lossy_output; + tokio::spawn(read_stream( + id, + StreamKind::Stderr, + stderr, + tx, + regex, + lossy_output, + )); } if let Some(process) = self.processes.get_mut(id) { @@ -311,23 +333,25 @@ impl ProcessManager { } }; if let Some(stdout) = child.stdout.take() { - let tx = self.event_tx.clone(); + let tx = self.output_tx.clone(); tokio::spawn(read_stream_with_prefix( id, StreamKind::Stdout, "[pre] ", stdout, tx, + self.lossy_output, )); } if let Some(stderr) = child.stderr.take() { - let tx = self.event_tx.clone(); + let tx = self.output_tx.clone(); tokio::spawn(read_stream_with_prefix( id, StreamKind::Stderr, "[pre] ", stderr, tx, + self.lossy_output, )); } let status = child.wait().await?; @@ -800,7 +824,8 @@ mod tests { }; let (tx, _rx) = mpsc::channel(4); let shutdown = ShutdownConfig::new(10, 1000); - let mut manager = ProcessManager::new(vec![spec], tx, shutdown); + let (output_tx, _output_rx) = mpsc::channel(4); + let mut manager = ProcessManager::new(vec![spec], tx, output_tx, shutdown, false); let child = tokio::process::Command::new("sleep") .arg("5") .spawn() @@ -828,11 +853,13 @@ async fn read_stream( reader: R, tx: mpsc::Sender, readiness_regex: Option, + lossy_output: bool, ) where R: tokio::io::AsyncRead + Unpin, { let mut lines = BufReader::new(reader).lines(); let mut matched = false; + let mut dropped: u64 = 0; while let Ok(Some(line)) = lines.next_line().await { if !matched { if let Some(regex) = &readiness_regex { @@ -842,7 +869,39 @@ async fn read_stream( } } } - let _ = tx.send(Event::ProcessOutput { id, line, stream }).await; + if lossy_output { + if dropped > 0 { + if tx + .try_send(Event::ProcessOutput { + id, + line: format!("[piperack] dropped {} lines (output overflow)", dropped), + stream: StreamKind::Stderr, + }) + .is_ok() + { + dropped = 0; + } + } + match tx.try_send(Event::ProcessOutput { id, line, stream }) { + Ok(_) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + dropped = dropped.saturating_add(1); + } + Err(mpsc::error::TrySendError::Closed(_)) => break, + } + } else { + let _ = tx.send(Event::ProcessOutput { id, line, stream }).await; + } + } + + if lossy_output && dropped > 0 { + let _ = tx + .send(Event::ProcessOutput { + id, + line: format!("[piperack] dropped {} lines (output overflow)", dropped), + stream: StreamKind::Stderr, + }) + .await; } } @@ -853,16 +912,45 @@ async fn read_stream_with_prefix( prefix: &str, reader: R, tx: mpsc::Sender, + lossy_output: bool, ) where R: tokio::io::AsyncRead + Unpin, { let mut lines = BufReader::new(reader).lines(); + let mut dropped: u64 = 0; while let Ok(Some(line)) = lines.next_line().await { + let line = format!("{}{}", prefix, line); + if lossy_output { + if dropped > 0 { + if tx + .try_send(Event::ProcessOutput { + id, + line: format!("[piperack] dropped {} lines (output overflow)", dropped), + stream: StreamKind::Stderr, + }) + .is_ok() + { + dropped = 0; + } + } + match tx.try_send(Event::ProcessOutput { id, line, stream }) { + Ok(_) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + dropped = dropped.saturating_add(1); + } + Err(mpsc::error::TrySendError::Closed(_)) => break, + } + } else { + let _ = tx.send(Event::ProcessOutput { id, line, stream }).await; + } + } + + if lossy_output && dropped > 0 { let _ = tx .send(Event::ProcessOutput { id, - line: format!("{}{}", prefix, line), - stream, + line: format!("[piperack] dropped {} lines (output overflow)", dropped), + stream: StreamKind::Stderr, }) .await; }