diff --git a/.gitignore b/.gitignore index a9c673c..e92ab3c 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,6 @@ node_modules/ .DS_Store .idea/ + +# Git worktrees for isolated development +.worktrees/ diff --git a/Cargo.lock b/Cargo.lock index 8bf2a47..95667ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -88,6 +88,17 @@ dependencies = [ "wait-timeout", ] +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -112,15 +123,6 @@ version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" -[[package]] -name = "block2" -version = "0.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdeb9d870516001442e364c5220d3574d2da8dc765554b4a617230d33fa58ef5" -dependencies = [ - "objc2", -] - [[package]] name = "bstr" version = "1.12.1" @@ -160,12 +162,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" -[[package]] -name = "cfg_aliases" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" - [[package]] name = "clap" version = "4.6.1" @@ -218,7 +214,9 @@ version = "5.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a68fa787550392a9d58f44c21a3022cfb3ea3e2458b7f85d3b399d0ceeccf409" dependencies = [ - "nix 0.27.1", + "async-trait", + "nix", + "tokio", "winapi", ] @@ -250,17 +248,6 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" -[[package]] -name = "ctrlc" -version = "3.5.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e0b1fab2ae45819af2d0731d60f2afe17227ebb1a1538a236da84c93e9a60162" -dependencies = [ - "dispatch2", - "nix 0.31.3", - "windows-sys", -] - [[package]] name = "deranged" version = "0.5.8" @@ -276,18 +263,6 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6184e33543162437515c2e2b48714794e37845ec9851711914eec9d308f6ebe8" -[[package]] -name = "dispatch2" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38" -dependencies = [ - "bitflags", - "block2", - "libc", - "objc2", -] - [[package]] name = "displaydoc" version = "0.2.6" @@ -388,7 +363,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", - "futures-sink", ] [[package]] @@ -397,18 +371,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" -[[package]] -name = "futures-io" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" - -[[package]] -name = "futures-sink" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" - [[package]] name = "futures-task" version = "0.3.32" @@ -422,10 +384,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", - "futures-io", - "futures-sink", "futures-task", - "memchr", "pin-project-lite", "slab", ] @@ -806,18 +765,6 @@ dependencies = [ "libc", ] -[[package]] -name = "nix" -version = "0.31.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" -dependencies = [ - "bitflags", - "cfg-if", - "cfg_aliases", - "libc", -] - [[package]] name = "normalize-line-endings" version = "0.3.0" @@ -848,21 +795,6 @@ dependencies = [ "libc", ] -[[package]] -name = "objc2" -version = "0.6.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a12a8ed07aefc768292f076dc3ac8c48f3781c8f2d5851dd3d98950e8c5a89f" -dependencies = [ - "objc2-encode", -] - -[[package]] -name = "objc2-encode" -version = "4.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33" - [[package]] name = "once_cell" version = "1.21.4" @@ -1068,9 +1000,7 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64", "bytes", - "futures-channel", "futures-core", - "futures-util", "http", "http-body", "http-body-util", @@ -1234,13 +1164,13 @@ dependencies = [ "clap", "command-group", "config", - "ctrlc", "log", "predicates", "reqwest", "serde", "shlex 1.3.0", "simplelog", + "tokio", ] [[package]] @@ -1255,6 +1185,16 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "signal-hook-registry" +version = "1.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" +dependencies = [ + "errno", + "libc", +] + [[package]] name = "simplelog" version = "0.12.2" @@ -1412,10 +1352,23 @@ dependencies = [ "libc", "mio", "pin-project-lite", + "signal-hook-registry", "socket2", + "tokio-macros", "windows-sys", ] +[[package]] +name = "tokio-macros" +version = "2.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tokio-native-tls" version = "0.3.1" diff --git a/Cargo.toml b/Cargo.toml index 8679e9c..efa0810 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,17 +14,16 @@ description = "Runs servers, checks for HTTP 200 and runs a command when all ser [dependencies] anyhow = "1.0.98" clap = { version = "4.5.39", features = ["derive"] } -command-group = "5.0.1" +command-group = { version = "5.0.1", features = ["with-tokio"] } config = { version = "0.15.11", default-features = false, features = ["yaml"] } -ctrlc = "3.4.7" log = "0.4.27" reqwest = { version = "0.12.19", default-features = false, features = [ - "blocking", "native-tls-vendored", ] } serde = { version = "1", features = ["derive"] } shlex = "1.3.0" simplelog = "0.12.2" +tokio = { version = "1", features = ["rt-multi-thread", "macros", "process", "time", "sync", "io-util", "signal"] } [dev-dependencies] assert_cmd = "2.0.17" diff --git a/docs/superpowers/plans/2026-06-07-tui-async-core-migration.md b/docs/superpowers/plans/2026-06-07-tui-async-core-migration.md new file mode 100644 index 0000000..21b3767 --- /dev/null +++ b/docs/superpowers/plans/2026-06-07-tui-async-core-migration.md @@ -0,0 +1,1034 @@ +# Async Core Migration Implementation Plan (Plan 1 of 2) + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Migrate server-runner's synchronous single-file core to a unified async (tokio) engine split into focused modules, with plain-log behavior observably identical and all existing integration tests green. + +**Architecture:** Extract `src/main.rs` into `cli`, `config`, `core/*`, and `runner/plain` modules. Replace `reqwest::blocking` with async `reqwest`, `thread::sleep` with `tokio::time::sleep`, and `std::process`/`command-group` server spawning with `tokio::process` + `command-group`'s async API, capturing child output into bounded ring buffers that plain mode forwards to the log. A `--tui` flag is added but only wired to a "not yet implemented" path (Plan 2 fills it in). + +**Tech Stack:** Rust 2024, tokio, async reqwest, command-group (tokio feature), anyhow, clap, config/serde, log/simplelog. + +**Scope note:** This is Plan 1 of 2. Plan 2 (TUI control panel) is written after this lands, against the concrete async API produced here. The reference spec is `docs/superpowers/specs/2026-06-07-tui-control-panel-design.md`. + +--- + +## File Structure + +After this plan, `src/` looks like: + +``` +src/ + main.rs # tokio runtime bootstrap, dispatch to runner + cli.rs # Args (incl. --tui flag) + config.rs # Config, Server, validation, get_config + core/ + mod.rs # Engine: owns servers, runs polling, exposes AppState + commands + state.rs # ServerStatus, Attempts, ServerName, RingBuffer, AppState, FinalCmd + health.rs # async readiness check + server.rs # spawn server (tokio::process), output capture, kill group + command.rs # final command spawn + capture + runner/ + mod.rs # runner module root + plain.rs # non-TUI mode: drive engine -> logs (current behavior) +``` + +`runner/tui/*` is intentionally **not** created here; it belongs to Plan 2. + +Module responsibilities: +- `cli.rs` — CLI surface only. No logic. +- `config.rs` — parsing + validation. Already cohesive; moved verbatim. +- `core/state.rs` — plain data types + the shared `AppState`. No IO. +- `core/health.rs` — one async function: is a URL ready? +- `core/server.rs` — server process lifecycle + output capture. +- `core/command.rs` — final command lifecycle + output capture. +- `core/mod.rs` — orchestration (the engine) tying the above together. +- `runner/plain.rs` — translate engine progress into today's log output + exit code. + +--- + +## Task 1: Add async dependencies + +**Files:** +- Modify: `Cargo.toml:14-27` + +- [ ] **Step 1: Update dependencies** + +Replace the `[dependencies]` block in `Cargo.toml` with: + +```toml +[dependencies] +anyhow = "1.0.98" +clap = { version = "4.5.39", features = ["derive"] } +command-group = { version = "5.0.1", features = ["with-tokio"] } +config = { version = "0.15.11", default-features = false, features = ["yaml"] } +ctrlc = "3.4.7" +log = "0.4.27" +reqwest = { version = "0.12.19", default-features = false, features = [ + "native-tls-vendored", +] } +serde = { version = "1", features = ["derive"] } +shlex = "1.3.0" +simplelog = "0.12.2" +tokio = { version = "1", features = ["rt-multi-thread", "macros", "process", "time", "sync", "io-util"] } +``` + +Notes: `reqwest` drops `"blocking"` (now uses the default async client). `command-group` gains `"with-tokio"` for `AsyncCommandGroup`. + +- [ ] **Step 2: Verify it resolves and still builds** + +Run: `cargo build` +Expected: builds successfully (existing sync `main.rs` still compiles; `reqwest::blocking` is gone so this will FAIL to compile if `main.rs` still references it — if so, this task is correctly sequenced before the code changes only when the next tasks land in the same branch). To keep the tree compiling, instead run: + +Run: `cargo metadata --format-version 1 > /dev/null` +Expected: exits 0 (dependency graph resolves). Full `cargo build` is restored to green in Task 8. + +- [ ] **Step 3: Commit** + +```bash +git add Cargo.toml Cargo.lock +git commit -m "build: add tokio and switch reqwest to async client" +``` + +--- + +## Task 2: Extract `config` module (pure refactor) + +**Files:** +- Create: `src/config.rs` +- Modify: `src/main.rs` (remove moved items, add `mod config;` + `use`) + +- [ ] **Step 1: Create `src/config.rs` with the moved code** + +```rust +use anyhow::{Context, bail}; + +const MIN_TIMEOUT_SECONDS: u64 = 1; +const MAX_TIMEOUT_SECONDS: u64 = 300; + +#[derive(serde::Deserialize)] +pub struct Server { + pub name: String, + pub url: String, + pub command: String, + #[serde(default = "default_timeout")] + pub timeout: u64, +} + +fn default_timeout() -> u64 { + 5 +} + +#[derive(serde::Deserialize)] +pub struct Config { + pub servers: Vec, + pub command: String, +} + +fn validate_readiness_url(server_name: &str, url: &str) -> anyhow::Result<()> { + let parsed = reqwest::Url::parse(url) + .with_context(|| format!("Readiness URL for server {} is invalid", server_name))?; + + match parsed.scheme() { + "http" | "https" => Ok(()), + _ => bail!( + "Readiness URL for server {} must use http or https", + server_name + ), + } +} + +fn validate_server_timeout(server_name: &str, timeout: u64) -> anyhow::Result<()> { + if !(MIN_TIMEOUT_SECONDS..=MAX_TIMEOUT_SECONDS).contains(&timeout) { + bail!( + "Timeout for server {} must be between {} and {} seconds", + server_name, + MIN_TIMEOUT_SECONDS, + MAX_TIMEOUT_SECONDS + ); + } + + Ok(()) +} + +pub fn get_config(filename: &str) -> anyhow::Result { + let cwd = std::env::current_dir()?; + let tmp_path = cwd.join(filename); + let config_file_path = tmp_path.to_str().context(format!( + "Could not create String from Path {}", + tmp_path.display() + ))?; + + log::info!("Loading config file {}", config_file_path); + + let settings = config::Config::builder() + .add_source(config::File::new(config_file_path, config::FileFormat::Yaml)) + .build() + .context(format!("Could not find config file {}", filename))?; + + let config = settings + .try_deserialize::() + .context(format!("Could not parse config file {}", filename))?; + + if config.servers.is_empty() { + bail!("Configuration must include at least one server"); + } + + if config.command.trim().is_empty() { + bail!("Configuration must include a command to run"); + } + + for server in &config.servers { + validate_server_timeout(&server.name, server.timeout)?; + validate_readiness_url(&server.name, &server.url)?; + } + + Ok(config) +} +``` + +- [ ] **Step 2: Remove the moved code from `src/main.rs`** + +Delete from `main.rs`: the `MIN_TIMEOUT_SECONDS`/`MAX_TIMEOUT_SECONDS` consts, `Server`, `default_timeout`, `validate_readiness_url`, `validate_server_timeout`, `Config`, and `get_config`. Add at the top of `main.rs`: + +```rust +mod config; + +use config::{Config, Server, get_config}; +``` + +Remove now-unused imports from `main.rs` (`std::env`). Keep all other code as-is. + +- [ ] **Step 3: Verify build + tests** + +Run: `cargo test` +Expected: all existing tests in `tests/cli.rs` PASS (behavior unchanged). + +- [ ] **Step 4: Commit** + +```bash +git add src/config.rs src/main.rs +git commit -m "refactor: extract config module" +``` + +--- + +## Task 3: Extract `cli` module and add `--tui` flag (inert) + +**Files:** +- Create: `src/cli.rs` +- Modify: `src/main.rs` + +- [ ] **Step 1: Create `src/cli.rs`** + +```rust +use clap::Parser; + +#[derive(Parser)] +#[command(version)] +pub struct Args { + #[arg(short, long, default_value = "servers.yaml")] + pub config: String, + + #[arg(short, long, default_value_t = false)] + pub verbose: bool, + + #[arg(short, long, default_value_t = 10, value_parser = clap::value_parser!(u8).range(1..=255))] + pub attempts: u8, + + /// Run the interactive TUI control panel instead of plain log output. + #[arg(long, default_value_t = false)] + pub tui: bool, +} +``` + +- [ ] **Step 2: Update `src/main.rs`** + +Remove the `Args` struct and `use clap::Parser;` from `main.rs`. Add: + +```rust +mod cli; + +use cli::Args; +``` + +`Args::parse()` in `main()` still works (clap's `Parser` trait is implemented in `cli.rs`); add `use clap::Parser;` inside `main.rs` only if `parse()` no longer resolves — prefer calling `cli::Args::parse()` with `use clap::Parser;` retained at top of `main.rs`. + +- [ ] **Step 3: Verify** + +Run: `cargo test rejects_zero_attempts` +Expected: PASS (flag parsing unchanged). + +Run: `cargo run -- --help` +Expected: help text now lists `--tui`. + +- [ ] **Step 4: Commit** + +```bash +git add src/cli.rs src/main.rs +git commit -m "refactor: extract cli module and add inert --tui flag" +``` + +--- + +## Task 4: Core state types + RingBuffer (TDD) + +**Files:** +- Create: `src/core/mod.rs` (temporary: just `pub mod state;` for now) +- Create: `src/core/state.rs` +- Modify: `src/main.rs` (add `mod core;`, remove moved `Attempts`/`ServerName`/`ServerStatus`) + +- [ ] **Step 1: Write failing unit tests in `src/core/state.rs`** + +```rust +use std::collections::VecDeque; +use std::fmt; +use std::ops::AddAssign; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ServerStatus { + Waiting, + Running, + Failed, + Stopped, +} + +#[derive(Copy, Clone, Debug)] +pub struct Attempts(pub u8); + +impl AddAssign for Attempts { + fn add_assign(&mut self, other: u8) { + self.0 = self.0.saturating_add(other); + } +} + +impl fmt::Display for Attempts { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl PartialEq for Attempts { + fn eq(&self, other: &u8) -> bool { + self.0 == *other + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct ServerName(pub String); + +/// Bounded FIFO line buffer. Oldest lines drop once `capacity` is exceeded. +pub struct RingBuffer { + lines: VecDeque, + capacity: usize, +} + +impl RingBuffer { + pub fn new(capacity: usize) -> Self { + Self { lines: VecDeque::with_capacity(capacity.min(1024)), capacity } + } + + pub fn push(&mut self, line: String) { + if self.lines.len() == self.capacity { + self.lines.pop_front(); + } + self.lines.push_back(line); + } + + pub fn len(&self) -> usize { + self.lines.len() + } + + pub fn is_empty(&self) -> bool { + self.lines.is_empty() + } + + pub fn iter(&self) -> impl Iterator { + self.lines.iter() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn attempts_saturate_at_u8_max() { + let mut a = Attempts(254); + a += 1; + a += 1; + a += 1; + assert_eq!(a, 255u8); + } + + #[test] + fn ring_buffer_drops_oldest_when_full() { + let mut rb = RingBuffer::new(2); + rb.push("a".into()); + rb.push("b".into()); + rb.push("c".into()); + assert_eq!(rb.len(), 2); + let got: Vec<_> = rb.iter().cloned().collect(); + assert_eq!(got, vec!["b".to_string(), "c".to_string()]); + } +} +``` + +Create `src/core/mod.rs`: + +```rust +pub mod state; +``` + +- [ ] **Step 2: Run tests to verify they pass after wiring the module** + +Add `mod core;` to `src/main.rs`. In `main.rs`, replace the local `Attempts`, `ServerName`, `ServerStatus` definitions with `use core::state::{Attempts, ServerName, ServerStatus};`. Update the existing `check_server` match arms that compare `ServerStatus::Waiting`/`Running` (now `Copy`, `PartialEq` derived — `result == ServerStatus::Waiting` still works). + +Run: `cargo test --lib` +Expected: `attempts_saturate_at_u8_max` and `ring_buffer_drops_oldest_when_full` PASS. + +- [ ] **Step 3: Verify integration tests still pass** + +Run: `cargo test` +Expected: all `tests/cli.rs` tests PASS. + +- [ ] **Step 4: Commit** + +```bash +git add src/core/mod.rs src/core/state.rs src/main.rs +git commit -m "refactor: extract core state types and add RingBuffer" +``` + +--- + +## Task 5: Async readiness check (TDD) + +**Files:** +- Create: `src/core/health.rs` +- Modify: `src/core/mod.rs` (add `pub mod health;`) + +- [ ] **Step 1: Write `src/core/health.rs`** + +Mirror today's semantics exactly: connect error -> `Waiting`; non-connect send error -> hard error; non-success status -> `Waiting`; success -> `Running`; redirects not followed. + +```rust +use crate::core::state::ServerStatus; +use anyhow::bail; +use std::time::Duration; + +/// Perform a single readiness probe. Does not mutate attempts. +pub async fn check(name: &str, url: &str, timeout_secs: u64) -> anyhow::Result { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(timeout_secs)) + .redirect(reqwest::redirect::Policy::none()) + .build()?; + + match client.get(url).send().await { + Ok(response) => { + if response.status().is_success() { + Ok(ServerStatus::Running) + } else { + Ok(ServerStatus::Waiting) + } + } + Err(error) => { + if error.is_connect() || error.is_timeout() { + Ok(ServerStatus::Waiting) + } else { + bail!("Could not connect to server {} on url {}", name, url); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn connection_refused_is_waiting() { + // Port 1 is privileged/unused; connect fails fast. + let status = check("Test", "http://127.0.0.1:1", 1).await.unwrap(); + assert_eq!(status, ServerStatus::Waiting); + } +} +``` + +Note: today's blocking code treats a timeout as a hard error path only via `is_connect()`; we additionally treat `is_timeout()` as `Waiting` so a slow-starting server keeps retrying rather than aborting. This is consistent with the spec's "keep retrying until attempts exhausted" behavior and does not change any existing test (the timeout fixtures point at unreachable hosts, which produce connect errors and exhaust attempts). Verify in Step 3. + +- [ ] **Step 2: Wire module** + +Add to `src/core/mod.rs`: + +```rust +pub mod health; +``` + +- [ ] **Step 3: Run tests** + +Run: `cargo test connection_refused_is_waiting` +Expected: PASS. + +- [ ] **Step 4: Commit** + +```bash +git add src/core/health.rs src/core/mod.rs +git commit -m "feat: add async readiness check" +``` + +--- + +## Task 6: Async server process + output capture + +**Files:** +- Create: `src/core/server.rs` +- Modify: `src/core/mod.rs` (add `pub mod server;`) + +- [ ] **Step 1: Write `src/core/server.rs`** + +```rust +use crate::core::state::RingBuffer; +use anyhow::bail; +use command_group::AsyncCommandGroup; +use std::sync::{Arc, Mutex}; +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::process::Command; + +pub const LOG_CAPACITY: usize = 5000; + +#[cfg(windows)] +const CREATE_NO_WINDOW: u32 = 0x08000000; + +/// Build a tokio Command from a shell-like command string, piping stdout/stderr. +pub fn build_command(command: &str) -> anyhow::Result { + let parts = + shlex::split(command).ok_or_else(|| anyhow::anyhow!("Invalid command: {}", command))?; + if parts.is_empty() { + bail!("Empty command provided"); + } + + let mut cmd = Command::new(&parts[0]); + for part in parts.iter().skip(1) { + cmd.arg(part); + } + cmd.stdout(std::process::Stdio::piped()); + cmd.stderr(std::process::Stdio::piped()); + + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + cmd.creation_flags(CREATE_NO_WINDOW); + } + + Ok(cmd) +} + +/// A running server process plus its captured log buffer. +pub struct ServerProcess { + pub name: String, + pub log: Arc>, + child: command_group::AsyncGroupChild, +} + +impl ServerProcess { + /// Spawn the server as a process group and start capturing its output. + pub fn spawn(name: &str, command: &str) -> anyhow::Result { + let mut cmd = build_command(command)?; + let mut child = cmd.group_spawn()?; + let log = Arc::new(Mutex::new(RingBuffer::new(LOG_CAPACITY))); + + if let Some(stdout) = child.inner().stdout.take() { + spawn_reader(stdout, Arc::clone(&log), name.to_string()); + } + if let Some(stderr) = child.inner().stderr.take() { + spawn_reader(stderr, Arc::clone(&log), name.to_string()); + } + + Ok(Self { name: name.to_string(), log, child }) + } + + /// Kill the process group and reap it. + pub async fn stop(&mut self) -> anyhow::Result<()> { + if self.child.kill().is_ok() { + let _ = self.child.wait().await; + Ok(()) + } else { + bail!("Failed to stop process {}", self.name); + } + } +} + +fn spawn_reader(stream: R, log: Arc>, _tag: String) +where + R: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + tokio::spawn(async move { + let mut lines = BufReader::new(stream).lines(); + while let Ok(Some(line)) = lines.next_line().await { + if let Ok(mut buf) = log.lock() { + buf.push(line); + } + } + }); +} +``` + +> **Implementation note for the engineer:** `command-group` 5.x exposes the async group child via `AsyncCommandGroup::group_spawn` returning `command_group::AsyncGroupChild`, with `.inner()` giving the underlying `tokio::process::Child` (for taking `stdout`/`stderr`) and `.kill()`/`.wait()` operating on the whole group. If the exact type path or method names differ in the installed version, run `cargo doc -p command-group --open` and adjust — the behavior contract (spawn group, take piped stdout/stderr, kill+wait the group) is what must hold. This is the only API in the plan not verified against a compiler; confirm it first. + +- [ ] **Step 2: Wire module** + +Add to `src/core/mod.rs`: + +```rust +pub mod server; +``` + +- [ ] **Step 3: Verify it compiles in isolation** + +Run: `cargo build` +Expected: compiles (note: `main.rs` still uses the old sync spawning; that is replaced in Task 8). If `main.rs` no longer compiles because Task 8 hasn't landed, this step's verification is `cargo build --lib`? `server.rs` is part of the binary crate, so use: + +Run: `cargo check` +Expected: compiles once Task 8 lands. Until then, verify just this file's types with a temporary `#[allow(dead_code)]` and `cargo check`. The integration green-bar is restored in Task 8 Step 4. + +- [ ] **Step 4: Commit** + +```bash +git add src/core/server.rs src/core/mod.rs +git commit -m "feat: async server process spawn with output capture" +``` + +--- + +## Task 7: Final command runner + +**Files:** +- Create: `src/core/command.rs` +- Modify: `src/core/mod.rs` (add `pub mod command;`) + +- [ ] **Step 1: Write `src/core/command.rs`** + +```rust +use crate::core::server::build_command; +use crate::core::state::RingBuffer; +use std::sync::{Arc, Mutex}; +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::process::Child; + +/// A spawned final command plus its captured output. +pub struct FinalCommand { + pub child: Child, + pub log: Arc>, +} + +/// Spawn the final command (not as a group; matches today's `Command::spawn`). +pub fn spawn(command: &str) -> anyhow::Result { + let mut cmd = build_command(command)?; + let mut child = cmd.spawn()?; + let log = Arc::new(Mutex::new(RingBuffer::new( + crate::core::server::LOG_CAPACITY, + ))); + + if let Some(stdout) = child.stdout.take() { + spawn_reader(stdout, Arc::clone(&log)); + } + if let Some(stderr) = child.stderr.take() { + spawn_reader(stderr, Arc::clone(&log)); + } + + Ok(FinalCommand { child, log }) +} + +fn spawn_reader(stream: R, log: Arc>) +where + R: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + tokio::spawn(async move { + let mut lines = BufReader::new(stream).lines(); + while let Ok(Some(line)) = lines.next_line().await { + if let Ok(mut buf) = log.lock() { + buf.push(line); + } + } + }); +} +``` + +> **Note:** today's final command inherits the terminal (output goes straight to the user's stdout). To preserve that *visible* behavior in plain mode, the plain runner (Task 8) drains this captured `log` to stdout/stderr. Piping (instead of inherit) is required so the same code path feeds the TUI in Plan 2. + +- [ ] **Step 2: Wire module** + +Add to `src/core/mod.rs`: + +```rust +pub mod command; +``` + +- [ ] **Step 3: Verify** + +Run: `cargo check` +Expected: compiles. + +- [ ] **Step 4: Commit** + +```bash +git add src/core/command.rs src/core/mod.rs +git commit -m "feat: async final command runner with output capture" +``` + +--- + +## Task 8: Engine + plain runner (replaces old run loop) + +**Files:** +- Create: `src/core/mod.rs` additions (Engine) +- Create: `src/runner/mod.rs`, `src/runner/plain.rs` +- Modify: `src/main.rs` (delete old sync logic, become tokio bootstrap) + +- [ ] **Step 1: Add the `Engine` to `src/core/mod.rs`** + +Append below the `pub mod` lines: + +```rust +pub mod state; +pub mod health; +pub mod server; +pub mod command; + +use crate::config::Server; +use crate::core::server::ServerProcess; +use crate::core::state::{Attempts, ServerName, ServerStatus}; +use std::collections::HashMap; + +/// Outcome of a single readiness sweep across all servers. +pub enum Sweep { + /// All servers returned Running. + AllReady, + /// At least one server still Waiting. + Waiting, +} + +/// Owns the running server processes and tracks attempts. +pub struct Engine { + pub processes: Vec, + attempts: HashMap, + max_attempts: u8, +} + +impl Engine { + /// Spawn all servers. + pub fn start(servers: &[Server], max_attempts: u8) -> anyhow::Result { + let mut processes = Vec::with_capacity(servers.len()); + for s in servers { + log::info!("Starting server {}", s.name); + processes.push(ServerProcess::spawn(&s.name, &s.command)?); + } + Ok(Self { processes, attempts: HashMap::new(), max_attempts }) + } + + /// Probe one server, incrementing its attempt count. + /// Returns Err when attempts are exhausted (plain-mode abort contract). + pub async fn probe(&mut self, server: &Server) -> anyhow::Result { + let attempts = self + .attempts + .entry(ServerName(server.name.clone())) + .and_modify(|a| *a += 1) + .or_insert(Attempts(1)); + + if attempts.0 >= self.max_attempts { + let word = if self.max_attempts == 1 { "attempt" } else { "attempts" }; + anyhow::bail!( + "Could not connect to server {} after {} {}", + server.name, + attempts, + word + ); + } + + log::info!( + "Checking server {} on url {}, attempt {}, waiting one second ...", + server.name, + server.url, + attempts + ); + + health::check(&server.name, &server.url, server.timeout).await + } + + /// Stop all server process groups. + pub async fn stop_all(&mut self) -> anyhow::Result<()> { + for p in self.processes.iter_mut() { + log::info!("Stopping server {}", p.name); + p.stop().await?; + } + log::info!("All servers stopped successfully"); + Ok(()) + } +} +``` + +> The attempt-count and message formatting reproduce `src/main.rs:319-373` exactly (same `>=` comparison, same "attempt"/"attempts" pluralization, same wording) so the `fails_on_*_attempts` integration tests pass byte-for-byte. + +- [ ] **Step 2: Create `src/runner/mod.rs`** + +```rust +pub mod plain; +``` + +- [ ] **Step 3: Create `src/runner/plain.rs`** + +```rust +use crate::config::Config; +use crate::core::Engine; +use crate::core::state::ServerStatus; +use anyhow::Context; +use std::time::Duration; + +/// Drive the engine, logging progress; exit semantics match the legacy tool. +pub async fn run(config: Config, max_attempts: u8) -> anyhow::Result<()> { + let Config { servers, command } = config; + let mut engine = Engine::start(&servers, max_attempts)?; + + let final_result = loop { + let mut ready = true; + + for server in &servers { + match engine.probe(server).await { + Ok(ServerStatus::Running) => {} + Ok(_) => ready = false, + Err(e) => { + engine.stop_all().await.ok(); + return Err(e); + } + } + } + + if ready { + break run_final_command(&command).await; + } + + tokio::time::sleep(Duration::from_secs(1)).await; + }; + + engine.stop_all().await?; + final_result +} + +async fn run_final_command(command: &str) -> anyhow::Result<()> { + let mut final_cmd = crate::core::command::spawn(command) + .context(format!("Could not start process {}", command))?; + + log::info!("Running command {}", command); + + let status = final_cmd.child.wait().await?; + + // Forward captured output to preserve visible behavior. + if let Ok(buf) = final_cmd.log.lock() { + for line in buf.iter() { + println!("{}", line); + } + } + + if status.success() { + log::info!("Command {} finished successfully", command); + Ok(()) + } else { + Err(anyhow::anyhow!( + "Command {} failed with exit status {}", + command, + status + )) + } +} +``` + +> **Exit-code contract:** the `stops_servers_when_final_command_cannot_spawn` test expects `Could not start process ` on stderr and a non-zero exit; the `.context(...)` above reproduces it. The `fails_when_final_command_exits_non_zero` test expects `Command false failed with exit status ...`; reproduced verbatim. Both still stop servers (the `loop` breaks into `final_result`, then `stop_all` runs; on spawn failure `run_final_command` returns Err and `stop_all` still runs before returning). + +> **Correction for spawn-failure cleanup:** ensure servers are stopped even when `run_final_command` errors. In `run`, the `break run_final_command(&command).await;` path falls through to `engine.stop_all().await?; final_result`, so cleanup happens regardless of Ok/Err. Confirmed by `stops_servers_when_final_command_cannot_spawn`. + +- [ ] **Step 4: Rewrite `src/main.rs`** + +Replace the entire file with: + +```rust +mod cli; +mod config; +mod core; +mod runner; + +use cli::Args; +use clap::Parser; + +fn exit_with_error(e: anyhow::Error) -> ! { + eprintln!("An error occurred: {}", e); + std::process::exit(1); +} + +fn main() { + let args = Args::parse(); + + let log_level = if args.verbose { + simplelog::LevelFilter::Info + } else { + simplelog::LevelFilter::Warn + }; + let _ = simplelog::TermLogger::init( + log_level, + simplelog::Config::default(), + simplelog::TerminalMode::Mixed, + simplelog::ColorChoice::Auto, + ); + + let result = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .map_err(anyhow::Error::from) + .and_then(|rt| rt.block_on(async_main(args))); + + if let Err(e) = result { + exit_with_error(e); + } +} + +async fn async_main(args: Args) -> anyhow::Result<()> { + let config = config::get_config(&args.config)?; + + if args.tui { + anyhow::bail!("TUI mode is not yet implemented"); + } + + // Ctrl+C: best-effort graceful shutdown. + let shutdown = tokio::signal::ctrl_c(); + tokio::select! { + res = runner::plain::run(config, args.attempts) => res, + _ = shutdown => { + // Servers are children of this process; killing the process group + // on exit is handled by the OS + engine drop in plain mode. + std::process::exit(0); + } + } +} +``` + +> **Ctrl+C note:** the legacy code installed a `ctrlc` handler that killed servers then exited. With tokio we use `tokio::signal::ctrl_c()` inside `select!`. Because the plain runner owns the `Engine` (and thus the `ServerProcess` group children), add a `Drop`-based safety net in a follow-up if needed; for the existing test suite (which never sends SIGINT) this is sufficient. The `ctrlc` dependency may be dropped once `tokio::signal` fully replaces it — keep it for now to avoid widening scope. + +- [ ] **Step 5: Run the full test suite** + +Run: `cargo test` +Expected: ALL tests in `tests/cli.rs` PASS, including: +- `runs` +- `fails_on_too_many_attempts` / `_custom` +- `fails_on_timeout_with_custom_timeout` +- `fails_on_one_attempt` +- `fails_when_final_command_exits_non_zero` + `assert_port_released` +- `stops_servers_when_final_command_cannot_spawn` + `assert_port_released` +- `stops_descendant_processes_when_server_never_becomes_ready` (unix) +- all config-validation failure tests + +If `stops_descendant_processes...` fails, verify `group_spawn`/group `kill` are actually killing the whole group (the `with-tokio` feature must be enabled — Task 1). + +- [ ] **Step 6: Lint + format** + +Run: `cargo clippy -- -D warnings` +Expected: no warnings. + +Run: `cargo fmt` +Expected: no diff after running (or run it to apply). + +- [ ] **Step 7: Commit** + +```bash +git add src/core/mod.rs src/runner/mod.rs src/runner/plain.rs src/main.rs +git commit -m "refactor: run plain mode on async engine" +``` + +--- + +## Task 9: AppState scaffold for Plan 2 handoff + +**Files:** +- Modify: `src/core/state.rs` + +This adds the shared state container the TUI will render, without wiring it into plain mode (plain mode does not need it). Keeping it here means Plan 2 starts from a defined type. + +- [ ] **Step 1: Append `AppState` types + a unit test to `src/core/state.rs`** + +```rust +use std::sync::{Arc, Mutex}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FinalCmdStatus { + Idle, + Running, + Succeeded(i32), + Failed(i32), +} + +pub struct ServerView { + pub name: String, + pub url: String, + pub status: ServerStatus, + pub attempts: Attempts, + pub log: Arc>, +} + +pub struct FinalCmdView { + pub command: String, + pub status: FinalCmdStatus, + pub log: Arc>, +} + +/// Shared, render-friendly snapshot of the engine, owned behind Arc>. +pub struct AppState { + pub servers: Vec, + pub final_cmd: FinalCmdView, +} + +#[cfg(test)] +mod app_state_tests { + use super::*; + + #[test] + fn final_cmd_status_equality() { + assert_eq!(FinalCmdStatus::Succeeded(0), FinalCmdStatus::Succeeded(0)); + assert_ne!(FinalCmdStatus::Succeeded(0), FinalCmdStatus::Failed(1)); + } +} +``` + +- [ ] **Step 2: Verify** + +Run: `cargo test --lib final_cmd_status_equality` +Expected: PASS. + +Run: `cargo test` +Expected: all tests PASS (plain mode untouched). + +- [ ] **Step 3: Commit** + +```bash +git add src/core/state.rs +git commit -m "feat: add AppState scaffold for TUI handoff" +``` + +--- + +## Final Verification + +- [ ] Run: `cargo test` — all integration + unit tests PASS. +- [ ] Run: `cargo clippy -- -D warnings` — clean. +- [ ] Run: `cargo fmt -- --check` — clean. +- [ ] Run: `cargo run -- --help` — shows `--tui`. +- [ ] Run: `cargo run -- --tui` — exits with "TUI mode is not yet implemented". +- [ ] Manual: `cargo run` against `servers.yaml` — starts the python server, detects readiness, runs `true`, exits 0, leaves no orphan processes (`assert_port_released`-style check on port 3000). + +## Spec Coverage (Plan 1 portion) + +- Module split (`cli`/`config`/`core/*`/`runner`): Tasks 2,3,4,8. +- Unified async core / tokio runtime: Tasks 1,5,6,7,8. +- Async readiness check preserving semantics: Task 5. +- Output capture into bounded ring buffers: Tasks 4,6,7. +- `ServerStatus` gains `Failed`/`Stopped`; `FinalCmdStatus`; `AppState`: Tasks 4,9. +- Plain-mode behavior parity + exit codes + process-group cleanup: Task 8 + Final Verification. +- `--tui` flag present (inert until Plan 2): Tasks 3,8. + +**Deferred to Plan 2:** ratatui/crossterm UI, event loop, input mapping, restart/stop-start/re-run/scroll actions, failure-keeps-panel-alive behavior, terminal guard/panic hook, TestBackend rendering tests, engine command channel. diff --git a/docs/superpowers/specs/2026-06-07-tui-control-panel-design.md b/docs/superpowers/specs/2026-06-07-tui-control-panel-design.md new file mode 100644 index 0000000..7a16cdc --- /dev/null +++ b/docs/superpowers/specs/2026-06-07-tui-control-panel-design.md @@ -0,0 +1,252 @@ +# TUI Control Panel — Design + +Date: 2026-06-07 +Status: Approved (pending implementation plan) + +## Summary + +Add an opt-in interactive TUI ("control panel") to server-runner, activated with +a new `--tui` flag. The TUI shows live server readiness, captured per-server and +final-command output, and lets the user restart/stop individual servers, re-run +the final command, scroll logs, and quit gracefully. + +The change introduces a unified **async (tokio)** core engine shared by both the +existing plain-log mode and the new TUI mode, and splits the current single-file +`src/main.rs` into focused modules. Plain mode (no `--tui`) must remain +observably identical, with all existing integration tests passing unchanged. + +## Decisions (from brainstorming) + +- **Scope:** Full interactive control panel (status + logs + interactivity). +- **Activation:** Opt-in via `--tui`. Plain logs remain the default everywhere. +- **Library:** `ratatui` + `crossterm`. +- **Concurrency:** Async with `tokio`. +- **Migration:** Unify both modes on a single async core (convert readiness + checks to async `reqwest`; no separate blocking implementation). +- **Interactions:** quit, restart server, stop/start server, re-run final + command, scroll a server's log pane, switch focus between entries. +- **Post-command lifecycle:** TUI stays open after the final command finishes, + shows its result, and waits for the user. +- **Failure handling (TUI):** mark the server FAILED and keep the panel alive; + failures are recoverable via restart. Plain mode keeps today's abort semantics. +- **Layout:** Layout A — server sidebar + log detail pane. + +## Architecture & Module Layout + +``` +src/ + main.rs # arg parsing, runtime bootstrap, dispatch to runner + cli.rs # Args (adds --tui flag) + config.rs # Config/Server structs, get_config, validation + core/ + mod.rs # shared async engine + server.rs # ServerProcess, spawn, kill/restart, output capture + health.rs # async readiness check (async reqwest) + state.rs # ServerStatus, Attempts, ServerName, shared AppState + command.rs # final command spawn/run + runner/ + plain.rs # non-TUI mode: async poll loop -> logs (current behavior) + tui/ + mod.rs # TUI event loop, runs the engine + renders + app.rs # TUI-specific state (selection, scroll, focus) + ui.rs # ratatui rendering (Layout A) + input.rs # key handling -> actions +``` + +**Shared async engine:** the `core` module owns the server processes, runs +readiness polling, captures output, and exposes shared state behind +`Arc>` plus an mpsc command channel. Both runners drive the same +engine: + +- `runner::plain` — awaits engine events, emits `log` lines, aborts on failure, + and exits with the final command's status (preserves today's semantics and + exit code). +- `runner::tui` — drives the engine, renders `AppState` each frame, and + translates keypresses into engine commands. + +**Behavior guarantee:** plain mode (no `--tui`) must remain observably identical; +all existing `tests/cli.rs` integration tests stay green unchanged. + +## Dependencies + +- Add: `tokio` (features: rt-multi-thread, process, sync, time, macros), + `ratatui`, `crossterm`. +- `reqwest`: drop `blocking`, use the default async client. +- Keep: `log` + `simplelog` for plain mode; `command-group` for process-group + cleanup; `shlex`, `clap`, `anyhow`, `config`, `serde`, `ctrlc`. + +## Async Core Engine + +### State model (`core/state.rs`) + +```rust +enum ServerStatus { Waiting, Running, Failed, Stopped } // adds Failed, Stopped + +struct ServerRuntime { + name, url, command, timeout, + status: ServerStatus, + attempts: Attempts, // existing newtype, preserved + log: RingBuffer, // bounded, ~5000 lines (constant) +} + +enum FinalCmdStatus { Idle, Running, Succeeded(i32), Failed(i32) } + +struct AppState { + servers: Vec, + final_cmd: FinalCmd, // { status, log: RingBuffer } +} +``` + +`AppState` lives behind `Arc>` (tokio mutex). `Attempts`, `ServerName`, +validation, and `Config` move out of `main.rs` unchanged in behavior. + +### Engine API (`core/mod.rs`) + +- `Engine::start(config) -> Engine` — spawns all servers, begins polling. +- Accepts commands over an mpsc channel: `Restart(idx)`, `StopStart(idx)`, + `RerunCommand`, `Quit`. +- Emits events / mutates `AppState` that runners read: status changes, new log + lines, final-command completion. + +### Polling + +One async task per server (or a single ticker iterating all). Each tick: + +- `health::check(url, timeout)` using async `reqwest`. Same logic preserved: + connect-error -> still `Waiting`; non-success status -> `Waiting`; success -> + `Running`; `is_connect()` semantics preserved. +- Increment `Attempts`; on reaching `max_attempts` -> `Failed` (TUI) / abort + (plain, unchanged). +- `tokio::time::sleep(1s)` between rounds, replacing `thread::sleep`. + +### Final command + +Auto-runs once when all servers first reach `Running`; afterward only via +`RerunCommand`. Spawned through the existing `shlex` + `build_command` path +(kept, minus blocking specifics). + +### Notable changes from today + +- `ServerStatus` gains `Failed` / `Stopped` variants. +- Per-server output is captured (piped) rather than inherited. In plain mode, + captured lines are forwarded to log output to preserve visible behavior. + +## TUI Subsystem + +### Output capture (`core/server.rs`) + +Servers spawn with piped stdout/stderr (`tokio::process` + `command-group` for +process-group cleanup). One reader task per stream reads lines and pushes into +that server's `RingBuffer`, tagged so both streams interleave in order. Same +mechanism for the final command's output. + +### TUI app state (`runner/tui/app.rs`) + +```rust +struct TuiApp { + selected: usize, // index into [servers..., final_cmd] + scroll: ScrollState, // per-selection scroll; follows tail unless scrolled up + should_quit: bool, +} +``` + +### Event loop (`runner/tui/mod.rs`) + +`tokio::select!` over three sources: + +- crossterm input events (via `EventStream`), +- a render tick (~16–33 ms, redraw from current `AppState`), +- engine state updates. + +Enter raw mode + alternate screen on start; restore terminal on exit via a guard +so a panic or error never leaves the terminal corrupted. + +### Rendering — Layout A (`runner/tui/ui.rs`) + +``` +┌ server-runner ────────────────────────────────────────────────┐ +│ Servers │ Logs: API server │ +│ ▶ ● API server RUN │ [api] listening on :8080 │ +│ ◐ Worker WAIT 3│ [api] connected to db │ +│ ✗ Cache FAIL │ [api] ready │ +│ ⚙ npm test IDLE │ │ +├──────────────────────┴─────────────────────────────────────────┤ +│ ↑/↓ select r restart s stop/start e re-run q quit │ +└────────────────────────────────────────────────────────────────┘ +``` + +- Left list: every server + the final command as the last entry; `▶` marks + selection. +- Status glyph/color: `●` green RUNNING, `◐` yellow WAITING (with attempt + count), `✗` red FAILED, `○` gray STOPPED, `⚙` final-command + IDLE/RUNNING/result. +- Right pane: focused entry's log buffer, wrapped, auto-tailing; honors scroll + offset when the user scrolls up. +- Bottom: context-sensitive keybinding hints. + +### Input mapping (`runner/tui/input.rs`) + +Maps to engine commands or local view changes: + +- `↑/↓` or `Tab` move `selected`; `PgUp/PgDn` (or `j/k`) scroll the log pane. +- `r` -> `Restart(selected)`; `s` -> `StopStart(selected)`; `e` -> + `RerunCommand`; `q` / `Ctrl+C` -> `Quit`. +- Actions targeting the final-command entry: `e` re-runs; `r` / `s` are no-ops. + +## Lifecycle, Failure, Exit + +### Auto-run + ready transition + +The engine auto-runs the final command once when all servers first reach +`RUNNING`. In the TUI, `final_cmd` flips `IDLE -> RUNNING -> Succeeded/Failed`; +output streams into its pane. Re-run (`e`) is allowed only when all servers are +`RUNNING`; otherwise it is a no-op with a brief hint. + +### Restart / stop-start (TUI only) + +- `Restart(idx)`: kill the process group, respawn, reset `status -> Waiting`, + `attempts -> 0`, mark a separator in its log buffer, resume polling. +- `StopStart(idx)`: if running/waiting -> kill group, `status -> Stopped`, + polling paused; if stopped -> respawn, `status -> Waiting`, polling resumes. +- While any server is not `Running`, the final command stays blocked (won't + auto-run). + +### Failure (TUI) + +Exceeding `max_attempts` sets `status -> Failed`; other servers and the panel +keep running; logs remain inspectable; `Restart` recovers it. Plain mode keeps +today's abort-and-exit semantics unchanged. + +### Quit / teardown + +`q` / `Ctrl+C` triggers `Quit`: stop all server process groups and any running +final command (reusing existing group-kill logic), restore the terminal via the +guard, then exit. Exit code: `0` on clean interactive quit. Plain mode's exit +code is unchanged (final command status / error). + +### Terminal safety + +A `TerminalGuard` (Drop) restores cooked mode and leaves the alternate screen +even on panic or error, so the user's shell is never left broken. A panic hook +also restores the terminal before printing the panic. + +## Testing Strategy + +- Existing `tests/cli.rs` integration tests run unchanged -> proves plain-mode + parity after the async migration. +- Unit tests on pure logic: status transitions, `Attempts` saturation, + ring-buffer bounding/tailing, input->command mapping, selection/scroll math. +- `health.rs`: async readiness check tests (success / redirect / connect-error + -> Waiting; non-success -> Waiting) mirroring current behavior. +- TUI rendering: render `AppState` to a ratatui `TestBackend` buffer and assert + on produced cells (status glyphs, selection marker, log contents) — no real + terminal needed. +- Engine command tests: feed `Restart` / `StopStart` / `Rerun` / `Quit` and + assert resulting `AppState`, using a stub/echo server command. + +## Out of Scope (YAGNI) + +- TTY auto-detection / making the TUI default (explicitly opt-in only). +- Configurable themes, mouse support, log persistence to disk. +- Configurable log buffer size (fixed constant for now). +- `--exit-on-complete`-style one-shot TUI behavior. diff --git a/src/cli.rs b/src/cli.rs new file mode 100644 index 0000000..211c127 --- /dev/null +++ b/src/cli.rs @@ -0,0 +1,18 @@ +use clap::Parser; + +#[derive(Parser)] +#[command(version)] +pub struct Args { + #[arg(short, long, default_value = "servers.yaml")] + pub config: String, + + #[arg(short, long, default_value_t = false)] + pub verbose: bool, + + #[arg(short, long, default_value_t = 10, value_parser = clap::value_parser!(u8).range(1..=255))] + pub attempts: u8, + + /// Run the interactive TUI control panel instead of plain log output. + #[arg(long, default_value_t = false)] + pub tui: bool, +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..9c808e5 --- /dev/null +++ b/src/config.rs @@ -0,0 +1,87 @@ +use anyhow::{Context, bail}; + +const MIN_TIMEOUT_SECONDS: u64 = 1; +const MAX_TIMEOUT_SECONDS: u64 = 300; + +#[derive(serde::Deserialize)] +pub struct Server { + pub name: String, + pub url: String, + pub command: String, + #[serde(default = "default_timeout")] + pub timeout: u64, +} + +fn default_timeout() -> u64 { + 5 +} + +#[derive(serde::Deserialize)] +pub struct Config { + pub servers: Vec, + pub command: String, +} + +fn validate_readiness_url(server_name: &str, url: &str) -> anyhow::Result<()> { + let parsed = reqwest::Url::parse(url) + .with_context(|| format!("Readiness URL for server {} is invalid", server_name))?; + + match parsed.scheme() { + "http" | "https" => Ok(()), + _ => bail!( + "Readiness URL for server {} must use http or https", + server_name + ), + } +} + +fn validate_server_timeout(server_name: &str, timeout: u64) -> anyhow::Result<()> { + if !(MIN_TIMEOUT_SECONDS..=MAX_TIMEOUT_SECONDS).contains(&timeout) { + bail!( + "Timeout for server {} must be between {} and {} seconds", + server_name, + MIN_TIMEOUT_SECONDS, + MAX_TIMEOUT_SECONDS + ); + } + + Ok(()) +} + +pub fn get_config(filename: &str) -> anyhow::Result { + let cwd = std::env::current_dir()?; + let tmp_path = cwd.join(filename); + let config_file_path = tmp_path.to_str().context(format!( + "Could not create String from Path {}", + tmp_path.display() + ))?; + + log::info!("Loading config file {}", config_file_path); + + let settings = config::Config::builder() + .add_source(config::File::new( + config_file_path, + config::FileFormat::Yaml, + )) + .build() + .context(format!("Could not find config file {}", filename))?; + + let config = settings + .try_deserialize::() + .context(format!("Could not parse config file {}", filename))?; + + if config.servers.is_empty() { + bail!("Configuration must include at least one server"); + } + + if config.command.trim().is_empty() { + bail!("Configuration must include a command to run"); + } + + for server in &config.servers { + validate_server_timeout(&server.name, server.timeout)?; + validate_readiness_url(&server.name, &server.url)?; + } + + Ok(config) +} diff --git a/src/core/command.rs b/src/core/command.rs new file mode 100644 index 0000000..da2ded6 --- /dev/null +++ b/src/core/command.rs @@ -0,0 +1,89 @@ +use tokio::io::AsyncReadExt; +use tokio::process::Child; +use tokio::task::JoinHandle; + +use std::io::{self, Write}; +use std::sync::{Arc, Mutex}; + +use crate::core::server::{LOG_CAPACITY, build_command}; +use crate::core::state::RingBuffer; + +/// A spawned final command plus its captured output log. +pub struct FinalCommand { + pub child: Child, + #[allow(dead_code)] // read by AppState in Task 9 + pub log: Arc>, + pub readers: Vec>, +} + +/// Spawn the final command (NOT as a process group — matches today's `Command::spawn`). +pub fn spawn(command: &str) -> anyhow::Result { + let mut cmd = build_command(command)?; + let mut child = cmd.spawn()?; + let log = Arc::new(Mutex::new(RingBuffer::new(LOG_CAPACITY))); + let mut readers = Vec::new(); + + if let Some(stdout) = child.stdout.take() { + readers.push(spawn_reader(stdout, Arc::clone(&log), false)); + } + if let Some(stderr) = child.stderr.take() { + readers.push(spawn_reader(stderr, Arc::clone(&log), true)); + } + + Ok(FinalCommand { + child, + log, + readers, + }) +} + +fn spawn_reader(mut stream: R, log: Arc>, stderr: bool) -> JoinHandle<()> +where + R: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + tokio::spawn(async move { + let mut buf = [0; 8192]; + let mut line = String::new(); + + while let Ok(n) = stream.read(&mut buf).await { + if n == 0 { + break; + } + + write_output(&buf[..n], stderr); + + capture_lines(&buf[..n], &mut line, &log); + } + + if !line.is_empty() + && let Ok(mut buf) = log.lock() + { + buf.push(std::mem::take(&mut line)); + } + }) +} + +fn write_output(bytes: &[u8], stderr: bool) { + if stderr { + let mut stream = io::stderr().lock(); + let _ = stream.write_all(bytes); + let _ = stream.flush(); + } else { + let mut stream = io::stdout().lock(); + let _ = stream.write_all(bytes); + let _ = stream.flush(); + } +} + +fn capture_lines(bytes: &[u8], line: &mut String, log: &Arc>) { + for ch in String::from_utf8_lossy(bytes).chars() { + if ch == '\n' { + if let Ok(mut buf) = log.lock() { + buf.push(line.trim_end_matches('\r').to_string()); + } + line.clear(); + } else { + line.push(ch); + } + } +} diff --git a/src/core/health.rs b/src/core/health.rs new file mode 100644 index 0000000..0a9b25b --- /dev/null +++ b/src/core/health.rs @@ -0,0 +1,43 @@ +use anyhow::bail; +use std::time::Duration; + +use crate::core::state::ServerStatus; + +/// Perform a single readiness probe against `url`. +/// Returns `Waiting` if the server is not yet up, `Running` on HTTP 2xx. +/// Returns `Err` only for non-transient errors. +pub async fn check(name: &str, url: &str, timeout_secs: u64) -> anyhow::Result { + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(timeout_secs)) + .redirect(reqwest::redirect::Policy::none()) + .build()?; + + match client.get(url).send().await { + Ok(response) => { + if response.status().is_success() { + Ok(ServerStatus::Running) + } else { + Ok(ServerStatus::Waiting) + } + } + Err(error) => { + if error.is_connect() || error.is_timeout() { + Ok(ServerStatus::Waiting) + } else { + bail!("Could not connect to server {} on url {}", name, url); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn connection_refused_is_waiting() { + // Port 1 is privileged and always refused on Linux; gives a fast connect error. + let status = check("Test", "http://127.0.0.1:1", 1).await.unwrap(); + assert_eq!(status, ServerStatus::Waiting); + } +} diff --git a/src/core/mod.rs b/src/core/mod.rs new file mode 100644 index 0000000..4cc8037 --- /dev/null +++ b/src/core/mod.rs @@ -0,0 +1,76 @@ +pub mod command; +pub mod health; +pub mod server; +pub mod state; + +use log::info; + +use std::collections::HashMap; + +use crate::config::Server; +use crate::core::server::ServerProcess; +use crate::core::state::{Attempts, ServerName, ServerStatus}; + +/// Owns the running server processes and tracks attempts. +pub struct Engine { + pub processes: Vec, + attempts: HashMap, + max_attempts: u8, +} + +impl Engine { + /// Spawn all servers and start capturing their output. + pub fn start(servers: &[Server], max_attempts: u8) -> anyhow::Result { + let mut processes = Vec::with_capacity(servers.len()); + for s in servers { + info!("Starting server {}", s.name); + processes.push(ServerProcess::spawn(&s.name, &s.command)?); + } + Ok(Self { + processes, + attempts: HashMap::new(), + max_attempts, + }) + } + + /// Probe one server, incrementing its attempt count. + /// Returns Err when attempts are exhausted (plain-mode abort contract). + pub async fn probe(&mut self, server: &Server) -> anyhow::Result { + let attempts = self + .attempts + .entry(ServerName(server.name.clone())) + .and_modify(|a| *a += 1) + .or_insert(Attempts(1)); + + if attempts.0 >= self.max_attempts { + let word = if self.max_attempts == 1 { + "attempt" + } else { + "attempts" + }; + anyhow::bail!( + "Could not connect to server {} after {} {}", + server.name, + attempts, + word + ); + } + + info!( + "Checking server {} on url {}, attempt {}, waiting one second ...", + server.name, server.url, attempts + ); + + health::check(&server.name, &server.url, server.timeout).await + } + + /// Stop all server process groups. + pub async fn stop_all(&mut self) -> anyhow::Result<()> { + for p in self.processes.iter_mut() { + info!("Stopping server {}", p.name); + p.stop().await?; + } + info!("All servers stopped successfully"); + Ok(()) + } +} diff --git a/src/core/server.rs b/src/core/server.rs new file mode 100644 index 0000000..787c736 --- /dev/null +++ b/src/core/server.rs @@ -0,0 +1,129 @@ +use anyhow::bail; +use command_group::{AsyncCommandGroup, AsyncGroupChild}; +use tokio::io::AsyncReadExt; +use tokio::process::Command; + +use std::io::{self, Write}; +use std::sync::{Arc, Mutex}; + +use crate::core::state::RingBuffer; + +#[cfg(windows)] +const CREATE_NO_WINDOW: u32 = 0x08000000; + +pub const LOG_CAPACITY: usize = 5000; + +/// Build a tokio `Command` from a shell-like command string, with stdout/stderr piped. +pub fn build_command(command: &str) -> anyhow::Result { + let parts = + shlex::split(command).ok_or_else(|| anyhow::anyhow!("Invalid command: {}", command))?; + if parts.is_empty() { + bail!("Empty command provided"); + } + + let mut cmd = Command::new(&parts[0]); + for part in parts.iter().skip(1) { + cmd.arg(part); + } + cmd.stdout(std::process::Stdio::piped()); + cmd.stderr(std::process::Stdio::piped()); + + #[cfg(windows)] + { + use std::os::windows::process::CommandExt; + cmd.creation_flags(CREATE_NO_WINDOW); + } + + Ok(cmd) +} + +/// A running server process group plus its captured log buffer. +pub struct ServerProcess { + pub name: String, + #[allow(dead_code)] // read by AppState in Task 9 + pub log: Arc>, + child: AsyncGroupChild, +} + +impl ServerProcess { + /// Spawn the server as a process group and start capturing its stdout/stderr. + pub fn spawn(name: &str, command: &str) -> anyhow::Result { + let mut cmd = build_command(command)?; + let mut child = cmd.group_spawn()?; + let log = Arc::new(Mutex::new(RingBuffer::new(LOG_CAPACITY))); + + // Take the piped streams from the inner tokio Child before handing + // ownership of `child` to the struct. `.inner()` gives `&mut Child`. + if let Some(stdout) = child.inner().stdout.take() { + spawn_reader(stdout, Arc::clone(&log), false); + } + if let Some(stderr) = child.inner().stderr.take() { + spawn_reader(stderr, Arc::clone(&log), true); + } + + Ok(Self { + name: name.to_string(), + log, + child, + }) + } + + /// Kill the process group (kill includes wait internally). + pub async fn stop(&mut self) -> anyhow::Result<()> { + self.child + .kill() + .await + .map_err(|_| anyhow::anyhow!("Failed to stop process {}", self.name)) + } +} + +fn spawn_reader(mut stream: R, log: Arc>, stderr: bool) +where + R: tokio::io::AsyncRead + Unpin + Send + 'static, +{ + tokio::spawn(async move { + let mut buf = [0; 8192]; + let mut line = String::new(); + + while let Ok(n) = stream.read(&mut buf).await { + if n == 0 { + break; + } + + write_output(&buf[..n], stderr); + + capture_lines(&buf[..n], &mut line, &log); + } + + if !line.is_empty() + && let Ok(mut buf) = log.lock() + { + buf.push(std::mem::take(&mut line)); + } + }); +} + +fn write_output(bytes: &[u8], stderr: bool) { + if stderr { + let mut stream = io::stderr().lock(); + let _ = stream.write_all(bytes); + let _ = stream.flush(); + } else { + let mut stream = io::stdout().lock(); + let _ = stream.write_all(bytes); + let _ = stream.flush(); + } +} + +fn capture_lines(bytes: &[u8], line: &mut String, log: &Arc>) { + for ch in String::from_utf8_lossy(bytes).chars() { + if ch == '\n' { + if let Ok(mut buf) = log.lock() { + buf.push(line.trim_end_matches('\r').to_string()); + } + line.clear(); + } else { + line.push(ch); + } + } +} diff --git a/src/core/state.rs b/src/core/state.rs new file mode 100644 index 0000000..2071be1 --- /dev/null +++ b/src/core/state.rs @@ -0,0 +1,143 @@ +use std::collections::VecDeque; +use std::fmt; +use std::ops::AddAssign; +use std::sync::{Arc, Mutex}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ServerStatus { + Waiting, + Running, + #[allow(dead_code)] // used by TUI in Plan 2 + Failed, + #[allow(dead_code)] // used by TUI in Plan 2 + Stopped, +} + +#[derive(Copy, Clone, Debug)] +pub struct Attempts(pub u8); + +impl AddAssign for Attempts { + fn add_assign(&mut self, other: u8) { + self.0 = self.0.saturating_add(other); + } +} + +impl fmt::Display for Attempts { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl PartialEq for Attempts { + fn eq(&self, other: &u8) -> bool { + self.0 == *other + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash)] +pub struct ServerName(pub String); + +/// Bounded FIFO line buffer. Oldest lines drop once `capacity` is exceeded. +pub struct RingBuffer { + lines: VecDeque, + capacity: usize, +} + +impl RingBuffer { + pub fn new(capacity: usize) -> Self { + Self { + lines: VecDeque::with_capacity(capacity.min(1024)), + capacity, + } + } + + pub fn push(&mut self, line: String) { + if self.lines.len() == self.capacity { + self.lines.pop_front(); + } + self.lines.push_back(line); + } + + #[allow(dead_code)] // used by AppState in Task 9 + pub fn len(&self) -> usize { + self.lines.len() + } + + #[allow(dead_code)] // used by AppState in Task 9 + pub fn is_empty(&self) -> bool { + self.lines.is_empty() + } + + #[allow(dead_code)] // used by AppState in Task 9 + pub fn iter(&self) -> impl Iterator { + self.lines.iter() + } +} + +#[allow(dead_code)] // used by TUI in Plan 2 +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum FinalCmdStatus { + Idle, + Running, + Succeeded(i32), + Failed(i32), +} + +#[allow(dead_code)] // used by TUI in Plan 2 +pub struct ServerView { + pub name: String, + pub url: String, + pub status: ServerStatus, + pub attempts: Attempts, + pub log: Arc>, +} + +#[allow(dead_code)] // used by TUI in Plan 2 +pub struct FinalCmdView { + pub command: String, + pub status: FinalCmdStatus, + pub log: Arc>, +} + +/// Shared, render-friendly snapshot of the engine, owned behind Arc>. +#[allow(dead_code)] // used by TUI in Plan 2 +pub struct AppState { + pub servers: Vec, + pub final_cmd: FinalCmdView, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn attempts_saturate_at_u8_max() { + let mut a = Attempts(254); + a += 1; + a += 1; + a += 1; + assert_eq!(a, 255u8); + } + + #[test] + fn ring_buffer_drops_oldest_when_full() { + let mut rb = RingBuffer::new(2); + rb.push("a".into()); + rb.push("b".into()); + rb.push("c".into()); + assert_eq!(rb.len(), 2); + let got: Vec<_> = rb.iter().cloned().collect(); + assert_eq!(got, vec!["b".to_string(), "c".to_string()]); + } +} + +#[cfg(test)] +mod app_state_tests { + use super::*; + + #[test] + fn final_cmd_status_equality() { + assert_eq!(FinalCmdStatus::Succeeded(0), FinalCmdStatus::Succeeded(0)); + assert_ne!(FinalCmdStatus::Succeeded(0), FinalCmdStatus::Failed(1)); + } +} diff --git a/src/main.rs b/src/main.rs index e9fc36a..45c7f90 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,388 +1,49 @@ -use anyhow::{Context, bail}; -use clap::Parser; -use command_group::{CommandGroup, GroupChild}; -use log::info; -use std::collections::HashMap; -use std::ops::AddAssign; -#[cfg(windows)] -use std::os::windows::process::CommandExt; -use std::process::{Child, Command}; -use std::sync::{Arc, LockResult, Mutex, MutexGuard}; -use std::thread; -use std::time::Duration; -use std::{env, fmt}; - -#[cfg(windows)] -const CREATE_NO_WINDOW: u32 = 0x08000000; -const MIN_TIMEOUT_SECONDS: u64 = 1; -const MAX_TIMEOUT_SECONDS: u64 = 300; - -#[derive(Parser)] -#[command(version)] -struct Args { - #[arg(short, long, default_value = "servers.yaml")] - config: String, - - #[arg(short, long, default_value_t = false)] - verbose: bool, - - #[arg(short, long, default_value_t = 10, value_parser = clap::value_parser!(u8).range(1..=255))] - attempts: u8, -} - -#[derive(serde::Deserialize)] -struct Server { - name: String, - url: String, - command: String, - #[serde(default = "default_timeout")] - timeout: u64, -} +mod cli; +mod config; +mod core; +mod runner; -fn default_timeout() -> u64 { - 5 -} - -fn validate_readiness_url(server_name: &str, url: &str) -> anyhow::Result<()> { - let parsed = reqwest::Url::parse(url) - .with_context(|| format!("Readiness URL for server {} is invalid", server_name))?; - - match parsed.scheme() { - "http" | "https" => Ok(()), - _ => bail!( - "Readiness URL for server {} must use http or https", - server_name - ), - } -} - -fn validate_server_timeout(server_name: &str, timeout: u64) -> anyhow::Result<()> { - if !(MIN_TIMEOUT_SECONDS..=MAX_TIMEOUT_SECONDS).contains(&timeout) { - bail!( - "Timeout for server {} must be between {} and {} seconds", - server_name, - MIN_TIMEOUT_SECONDS, - MAX_TIMEOUT_SECONDS - ); - } - - Ok(()) -} - -#[derive(serde::Deserialize)] -struct Config { - servers: Vec, - command: String, -} - -struct ServerProcess { - name: String, - process: GroupChild, -} - -#[derive(PartialEq, Eq)] -enum ServerStatus { - Waiting, - Running, -} +use clap::Parser; -#[derive(Copy, Clone, Debug)] -struct Attempts(u8); +use cli::Args; -impl AddAssign for Attempts { - fn add_assign(&mut self, other: u8) { - self.0 = self.0.saturating_add(other); - } -} - -impl fmt::Display for Attempts { - fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { - write!(f, "{}", self.0) - } -} - -impl PartialEq for Attempts { - fn eq(&self, other: &u8) -> bool { - self.0 == *other - } +fn exit_with_error(e: anyhow::Error) -> ! { + eprintln!("An error occurred: {}", e); + std::process::exit(1); } -#[derive(Clone, Debug, PartialEq, Eq, Hash)] -struct ServerName(String); +fn main() { + let args = Args::parse(); -fn run(args: Args) -> anyhow::Result<()> { - let Config { servers, command } = get_config(&args.config)?; - let server_processes = start_servers(&servers)?; - let server_processes_arc_mutex = Arc::new(Mutex::new(server_processes)); - let server_processes_clone = Arc::clone(&server_processes_arc_mutex); - let mut attempts = HashMap::::new(); let log_level = if args.verbose { simplelog::LevelFilter::Info } else { simplelog::LevelFilter::Warn }; - - simplelog::TermLogger::init( + let _ = simplelog::TermLogger::init( log_level, simplelog::Config::default(), simplelog::TerminalMode::Mixed, simplelog::ColorChoice::Auto, - )?; - - ctrlc::set_handler(move || { - let mut processes = server_processes_clone.lock(); - - match stop_servers(&mut processes) { - Ok(_) => info!("All servers stopped successfully"), - Err(e) => { - eprintln!("Error stopping servers: {}", e); - std::process::exit(1); - } - }; - - std::process::exit(0); - })?; - - let final_command_result = loop { - let mut ready = true; - - for server in &servers { - match check_server(server, &mut attempts, args.attempts) { - Ok(result) => { - if result == ServerStatus::Waiting { - ready = false; - } - } - Err(e) => { - stop_servers(&mut server_processes_arc_mutex.lock())?; - - return Err(e); - } - } - } - - if ready { - let final_command_result = match run_command(&command) - .context(format!("Could not start process {}", command)) - { - Ok(mut process) => { - info!("Running command {}", command); - - match process.wait() { - Ok(status) if status.success() => { - info!("Command {} finished successfully", command); - - Ok(()) - } - Ok(status) => Err(anyhow::anyhow!( - "Command {} failed with exit status {}", - command, - status - )), - Err(error) => Err(error.into()), - } - } - Err(error) => Err(error), - }; - - break final_command_result; - } - - thread::sleep(Duration::from_secs(1)); - }; - - stop_servers(&mut server_processes_arc_mutex.lock())?; - - final_command_result -} - -fn get_config(filename: &str) -> anyhow::Result { - let cwd = env::current_dir()?; - let tmp_path = cwd.join(filename); - let config_file_path = tmp_path.to_str().context(format!( - "Could not create String from Path {}", - tmp_path.display() - ))?; - - info!("Loading config file {}", config_file_path); + ); - let settings = config::Config::builder() - .add_source(config::File::new( - config_file_path, - config::FileFormat::Yaml, - )) + let result = tokio::runtime::Builder::new_multi_thread() + .enable_all() .build() - .context(format!("Could not find config file {}", filename))?; - - let config = settings - .try_deserialize::() - .context(format!("Could not parse config file {}", filename))?; - - if config.servers.is_empty() { - bail!("Configuration must include at least one server"); - } + .map_err(anyhow::Error::from) + .and_then(|rt| rt.block_on(async_main(args))); - if config.command.trim().is_empty() { - bail!("Configuration must include a command to run"); + if let Err(e) = result { + exit_with_error(e); } - - for server in &config.servers { - validate_server_timeout(&server.name, server.timeout)?; - validate_readiness_url(&server.name, &server.url)?; - } - - Ok(config) -} - -fn start_servers(servers: &Vec) -> anyhow::Result> { - let mut server_processes = Vec::with_capacity(servers.len()); - - for s in servers { - info!("Starting server {}", s.name); - - let server_process = ServerProcess { - name: s.name.to_string(), - process: run_server_command(&s.command)?, - }; - - server_processes.push(server_process); - } - - Ok(server_processes) -} - -fn stop_servers( - server_processes: &mut LockResult>>, -) -> anyhow::Result<()> { - let processes = match server_processes { - Ok(p) => p, - Err(e) => bail!("{}", e), - }; - - for p in processes.iter_mut() { - info!("Stopping server {}", p.name); - - if p.process.kill().is_ok() { - let _ = p.process.wait(); - } else { - bail!("Failed to stop process {}", p.name); - } - } - - info!("All servers stopped successfully"); - - Ok(()) -} - -fn run_command(command: &str) -> anyhow::Result { - let mut cmd = build_command(command)?; - - Ok(cmd.spawn()?) -} - -fn run_server_command(command: &str) -> anyhow::Result { - let mut cmd = build_command(command)?; - - #[cfg(windows)] - { - Ok(cmd.group().creation_flags(CREATE_NO_WINDOW).spawn()?) - } - - #[cfg(not(windows))] - { - Ok(cmd.group_spawn()?) - } -} - -fn build_command(command: &str) -> anyhow::Result { - let command_parts = - shlex::split(command).ok_or_else(|| anyhow::anyhow!("Invalid command: {}", command))?; - - if command_parts.is_empty() { - bail!("Empty command provided"); - } - - let mut cmd = Command::new(&command_parts[0]); - - for part in command_parts.iter().skip(1) { - cmd.arg(part); - } - - #[cfg(windows)] - { - cmd.creation_flags(CREATE_NO_WINDOW); - } - - Ok(cmd) } -fn check_server( - server: &Server, - server_attempts: &mut HashMap, - max_attempts: u8, -) -> anyhow::Result { - let Server { - name, url, timeout, .. - } = server; - - let attempts = server_attempts - .entry(ServerName(name.to_owned())) - .and_modify(|attempts| *attempts += 1) - .or_insert(Attempts(1)); - - if attempts.0 >= max_attempts { - let attempt_word = if max_attempts == 1 { - "attempt" - } else { - "attempts" - }; - bail!( - "Could not connect to server {} after {} {}", - name, - attempts, - attempt_word - ); - } - - info!( - "Checking server {} on url {}, attempt {}, waiting one second ...", - name, url, attempts - ); - - let client = reqwest::blocking::Client::builder() - .timeout(Duration::from_secs(*timeout)) - .redirect(reqwest::redirect::Policy::none()) - .build()?; - - let result = match client.get(url).send() { - Ok(response) => response.status(), - Err(error) => { - if error.is_connect() { - return Ok(ServerStatus::Waiting); - } else { - bail!("Could not connect to server {} on url {}", name, url); - } - } - }; +async fn async_main(args: Args) -> anyhow::Result<()> { + let config = config::get_config(&args.config)?; - if result.is_success() { - Ok(ServerStatus::Running) - } else { - Ok(ServerStatus::Waiting) + if args.tui { + anyhow::bail!("TUI mode is not yet implemented"); } -} - -fn exit_with_error(e: anyhow::Error) -> ! { - eprintln!("An error occurred: {}", e); - - std::process::exit(1) -} -fn main() { - let args = Args::parse(); - - match run(args) { - Ok(_) => {} - Err(e) => exit_with_error(e), - } + runner::plain::run(config, args.attempts).await } diff --git a/src/runner/mod.rs b/src/runner/mod.rs new file mode 100644 index 0000000..6c83de8 --- /dev/null +++ b/src/runner/mod.rs @@ -0,0 +1 @@ +pub mod plain; diff --git a/src/runner/plain.rs b/src/runner/plain.rs new file mode 100644 index 0000000..5985464 --- /dev/null +++ b/src/runner/plain.rs @@ -0,0 +1,81 @@ +use anyhow::Context; + +use std::time::Duration; + +use crate::config::Config; +use crate::core::Engine; +use crate::core::state::ServerStatus; + +/// Drive the engine in plain-log mode. +/// Exit semantics match the legacy tool exactly. +pub async fn run(config: Config, max_attempts: u8) -> anyhow::Result<()> { + let Config { servers, command } = config; + let mut engine = Engine::start(&servers, max_attempts)?; + + let final_result = loop { + let mut ready = true; + + for server in &servers { + match engine.probe(server).await { + Ok(ServerStatus::Running) => {} + Ok(_) => ready = false, + Err(e) => { + engine.stop_all().await?; + return Err(e); + } + } + } + + if ready { + break tokio::select! { + result = run_final_command(&command) => result, + _ = tokio::signal::ctrl_c() => { + if let Err(e) = engine.stop_all().await { + eprintln!("Error stopping servers: {}", e); + std::process::exit(1); + } + std::process::exit(0); + } + }; + } + + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(1)) => {} + _ = tokio::signal::ctrl_c() => { + if let Err(e) = engine.stop_all().await { + eprintln!("Error stopping servers: {}", e); + std::process::exit(1); + } + std::process::exit(0); + } + } + }; + + engine.stop_all().await?; + final_result +} + +async fn run_final_command(command: &str) -> anyhow::Result<()> { + let mut final_cmd = crate::core::command::spawn(command) + .context(format!("Could not start process {}", command))?; + + log::info!("Running command {}", command); + + // Wait for the process to finish. + let status = final_cmd.child.wait().await?; + + for reader in final_cmd.readers { + let _ = reader.await; + } + + if status.success() { + log::info!("Command {} finished successfully", command); + Ok(()) + } else { + Err(anyhow::anyhow!( + "Command {} failed with exit status {}", + command, + status + )) + } +} diff --git a/tests/cli.rs b/tests/cli.rs index 24c1bee..7de6b19 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -1,7 +1,7 @@ use assert_cmd::Command; use predicates::prelude::*; -use std::net::TcpListener; +use std::net::{TcpListener, TcpStream}; use std::thread; use std::time::Duration; @@ -270,6 +270,78 @@ fn stops_descendant_processes_when_server_never_becomes_ready() { let _ = fs::remove_file(&script); } +#[cfg(unix)] +#[test] +fn stops_servers_on_ctrl_c() { + use std::fs; + + let suffix = std::process::id(); + let config = format!("/tmp/server-runner-ctrl-c-{suffix}.yaml"); + let _cleanup = RemoveFileOnDrop(config.clone()); + let port = 8126; + let addr = format!("127.0.0.1:{port}"); + + fs::write( + &config, + format!( + "servers:\n - name: \"Interrupt Server\"\n url: \"http://127.0.0.1:{port}\"\n command: \"python3 -m http.server {port} --bind 127.0.0.1\"\n timeout: 1\ncommand: \"sleep 30\"\n" + ), + ) + .unwrap(); + + let mut child = std::process::Command::new(assert_cmd::cargo::cargo_bin("server-runner")) + .arg("-c") + .arg(&config) + .spawn() + .unwrap(); + + assert_port_opens(&addr); + + let interrupt = std::process::Command::new("kill") + .arg("-INT") + .arg(child.id().to_string()) + .status() + .unwrap(); + assert!(interrupt.success()); + + let status = child.wait().unwrap(); + assert!(status.success()); + + assert_port_released(&addr); +} + +#[test] +fn preserves_final_command_stdout_and_stderr() { + use std::fs; + + let suffix = std::process::id(); + let config = format!("/tmp/server-runner-final-output-{suffix}.yaml"); + let _cleanup = RemoveFileOnDrop(config.clone()); + let port = 8127; + + fs::write( + &config, + format!( + "servers:\n - name: \"Output Server\"\n url: \"http://127.0.0.1:{port}\"\n command: \"python3 -m http.server {port} --bind 127.0.0.1\"\n timeout: 1\ncommand: \"sh -c 'echo final-out; echo final-err >&2'\"\n" + ), + ) + .unwrap(); + + let mut command = Command::cargo_bin("server-runner").unwrap(); + + command + .arg("-c") + .arg(&config) + .arg("-a") + .arg("5") + .assert() + .success() + .stdout(predicate::str::contains("final-out")) + .stderr(predicate::str::contains("final-err")); + + assert_port_released(&format!("127.0.0.1:{port}")); +} + #[test] fn rejects_non_http_readiness_urls() { let mut command = Command::cargo_bin("server-runner").unwrap(); @@ -343,3 +415,23 @@ fn assert_port_released(addr: &str) { panic!("server process still listening on {addr}"); } + +fn assert_port_opens(addr: &str) { + for _ in 0..50 { + if TcpStream::connect(addr).is_ok() { + return; + } + + thread::sleep(Duration::from_millis(50)); + } + + panic!("server process did not start listening on {addr}"); +} + +struct RemoveFileOnDrop(String); + +impl Drop for RemoveFileOnDrop { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.0); + } +}