Handle closed stdout, report align's dropped input, and tidy CLI edge cases

README.md

@@ -4,8 +4,9 @@ Small Unix-style tools for exploring the shape of timestamped data — waveforms
 distributions, co-occurrence, and window coverage in logs and time series.
 Written in Rust, born out of incident triage.
 
-Each tool reads from stdin, does one thing, and writes to stdout — so they
-compose naturally with `grep` / `jq` / `awk` and with each other.
+Each tool does one thing and writes to stdout — so they compose naturally with
+`grep` / `jq` / `awk` and with each other. Most read stdin; the ones that take
+several inputs (`ts-correlate`, `ts-align`, `ts-timeline`) read files instead.
 
 ## Tools
 
@@ -263,8 +264,10 @@ ts-align req.txt lat.txt tasks.txt \
 ```
 
 where req / lat / tasks are CloudWatch RequestCount (Sum), TargetResponseTime
-(Average) and LiveTaskCount (Minimum). A `common N / per-file totals` line goes to
-stderr so dropped (non-common) timestamps are visible.
+(Average) and LiveTaskCount (Minimum). A
+`N timestamps common to all K series (file=count, …)` line goes to stderr so
+dropped (non-common) timestamps are visible. Unparseable lines and duplicate
+timestamps (last value kept) are reported per file on stderr too.
 
 ## ts-timeline
 

crates/align/src/main.rs

@@ -14,6 +14,7 @@
 //!       | ts-series
 
 use std::collections::BTreeMap;
+use std::io::{self, Write};
 use std::process::ExitCode;
 
 use align::inner_join;
@@ -80,15 +81,26 @@ fn run() -> Result<ExitCode, String> {
 
     let rows = inner_join(&maps);
 
-    for (instant, values) in &rows {
-        // render in the input's display offset; values go out at full precision
-        // (align feeds downstream math — rounding here would corrupt it)
-        let mut line = instant.with_timezone(&offset).to_rfc3339();
-        for v in values {
-            line.push('\t');
-            line.push_str(&v.to_string());
+    let mut out = io::BufWriter::new(io::stdout().lock());
+    let written = rows
+        .iter()
+        .try_for_each(|(instant, values)| {
+            // render in the input's display offset; values go out at full precision
+            // (align feeds downstream math — rounding here would corrupt it)
+            let mut line = instant.with_timezone(&offset).to_rfc3339();
+            for v in values {
+                line.push('\t');
+                line.push_str(&v.to_string());
+            }
+            writeln!(out, "{line}")
+        })
+        .and_then(|()| out.flush());
+    if let Err(e) = written {
+        // Downstream closed the pipe (e.g. `| head`): stop quietly.
+        if e.kind() == io::ErrorKind::BrokenPipe {
+            return Ok(ExitCode::SUCCESS);
         }
-        println!("{line}");
+        return Err(format!("write error: {e}"));
     }
 
     // report what the inner join dropped, so non-common timestamps aren't silent
@@ -109,18 +121,38 @@ fn run() -> Result<ExitCode, String> {
 /// Read a file of `timestamp value` points into a map keyed by UTC instant
 /// (last value wins on a duplicate timestamp), plus the input's offset and point
 /// count. Errors if the file holds no points.
+///
+/// Unparseable lines and duplicate timestamps are reported on stderr: the
+/// per-file count is of distinct timestamps, so without this both would vanish
+/// from the summary unnoticed.
 fn load(path: &str) -> Result<(Series, FixedOffset, usize), String> {
     let text = std::fs::read_to_string(path).map_err(|e| format!("{path}: {e}"))?;
     let mut map = BTreeMap::new();
     let mut offset = None;
+    let (mut skipped, mut duplicates) = (0u64, 0u64);
     for line in text.lines() {
-        if let Some((dt, value)) = parse_point(line) {
-            offset.get_or_insert_with(|| *dt.offset());
-            // key by the UTC instant (full sub-second precision): truncating to
-            // seconds would collide distinct points and join non-equal instants
-            map.insert(dt.to_utc(), value);
+        match parse_point(line) {
+            Some((dt, value)) => {
+                offset.get_or_insert_with(|| *dt.offset());
+                // key by the UTC instant (full sub-second precision): truncating to
+                // seconds would collide distinct points and join non-equal instants
+                if map.insert(dt.to_utc(), value).is_some() {
+                    duplicates += 1;
+                }
+            }
+            None => {
+                if !line.trim().is_empty() {
+                    skipped += 1;
+                }
+            }
         }
     }
+    if skipped > 0 {
+        eprintln!("ts-align: {path}: skipped {skipped} unparseable line(s)");
+    }
+    if duplicates > 0 {
+        eprintln!("ts-align: {path}: {duplicates} duplicate timestamp(s), last value kept");
+    }
     match offset {
         Some(off) => {
             let n = map.len();

crates/align/tests/cli.rs

@@ -1,7 +1,7 @@
 //! End-to-end smoke tests: run the built binary against fixture files.
 //! Uses `CARGO_BIN_EXE_ts-align` and `CARGO_MANIFEST_DIR` (set by Cargo).
 
-use std::process::{Command, Output};
+use std::process::{Command, Output, Stdio};
 
 fn run(args: &[&str]) -> Output {
     Command::new(env!("CARGO_BIN_EXE_ts-align"))
@@ -76,3 +76,42 @@ fn needs_two_files() {
     let stderr = String::from_utf8(out.stderr).unwrap();
     assert!(stderr.contains("at least two files"), "stderr:\n{stderr}");
 }
+
+#[test]
+fn reports_skipped_lines_and_duplicates() {
+    let a = fixture("dup.txt");
+    let out = run(&[a.as_str(), a.as_str()]);
+    assert!(out.status.success());
+    let stdout = String::from_utf8(out.stdout).unwrap();
+    // last value wins on the duplicated 10:20:00
+    assert!(
+        stdout.starts_with("2026-09-03T10:20:00+09:00\t2\t2\n"),
+        "stdout:\n{stdout}"
+    );
+    let stderr = String::from_utf8(out.stderr).unwrap();
+    assert!(
+        stderr.contains("dup.txt: skipped 1 unparseable line(s)"),
+        "stderr:\n{stderr}"
+    );
+    assert!(
+        stderr.contains("dup.txt: 1 duplicate timestamp(s), last value kept"),
+        "stderr:\n{stderr}"
+    );
+}
+
+#[test]
+fn closed_stdout_exits_quietly() {
+    // reader dropped before spawn, so every write hits a closed pipe (`| head`)
+    let (reader, writer) = std::io::pipe().unwrap();
+    drop(reader);
+    let req = fixture("req.txt");
+    let out = Command::new(env!("CARGO_BIN_EXE_ts-align"))
+        .args([req.as_str(), req.as_str()])
+        .stdout(writer)
+        .stderr(Stdio::piped())
+        .output()
+        .expect("run align");
+    let stderr = String::from_utf8(out.stderr).unwrap();
+    assert!(out.status.success(), "stderr:\n{stderr}");
+    assert!(!stderr.contains("Broken pipe"), "stderr:\n{stderr}");
+}

crates/align/tests/fixtures/dup.txt

@@ -0,0 +1,5 @@
+2026-09-03T10:20:00+09:00	1
+2026-09-03T10:20:00+09:00	2
+not a point
+
+2026-09-03T10:21:00+09:00	3

crates/burst/src/main.rs

@@ -14,6 +14,7 @@
 
 use std::collections::BTreeMap;
 use std::io::{self, BufRead, Write};
+use std::process::ExitCode;
 
 use burst::{Summary, counts_by_second, summarize};
 use chrono::{FixedOffset, TimeZone};
@@ -39,16 +40,16 @@ plus total / peak / first / last / span. All lines must share one UTC offset.
 example:
     jq -r 'select(.status==502) | .time' alb.jsonl | ts-burst";
 
-fn main() -> io::Result<()> {
+fn main() -> ExitCode {
     if let Some(arg) = std::env::args().nth(1) {
         match arg.as_str() {
             "-h" | "--help" => {
                 println!("{HELP}");
-                return Ok(());
+                return ExitCode::SUCCESS;
             }
             other => {
                 eprintln!("ts-burst: unexpected argument: {other}\n\n{HELP}");
-                std::process::exit(1);
+                return ExitCode::FAILURE;
             }
         }
     }
@@ -58,7 +59,13 @@ fn main() -> io::Result<()> {
 
     let stdin = io::stdin();
     for line in stdin.lock().lines() {
-        let line = line?;
+        let line = match line {
+            Ok(line) => line,
+            Err(e) => {
+                eprintln!("ts-burst: read error: {e}");
+                return ExitCode::FAILURE;
+            }
+        };
         match parse_line(&line) {
             Some(dt) => {
                 let this = *dt.offset();
@@ -68,7 +75,7 @@ fn main() -> io::Result<()> {
                     // times stay aligned with the source log.
                     Some(first) if first != this => {
                         eprintln!("ts-burst: mixed UTC offsets: {first} and {this}");
-                        std::process::exit(1);
+                        return ExitCode::FAILURE;
                     }
                     Some(_) => {}
                 }
@@ -92,12 +99,20 @@ fn main() -> io::Result<()> {
             "ts-burst: no parseable timestamps on stdin \
              (expected RFC3339 with offset, e.g. 2026-09-03T10:21:31.124+09:00)"
         );
-        std::process::exit(1);
+        return ExitCode::FAILURE;
     };
     // summarize returned Some, so at least one line parsed and set the offset.
     let offset = offset.expect("offset is set whenever a timestamp parses");
 
-    render(&counts, &summary, offset)
+    if let Err(e) = render(&counts, &summary, offset) {
+        // Downstream closed the pipe (e.g. `| head`): stop quietly.
+        if e.kind() == io::ErrorKind::BrokenPipe {
+            return ExitCode::SUCCESS;
+        }
+        eprintln!("ts-burst: write error: {e}");
+        return ExitCode::FAILURE;
+    }
+    ExitCode::SUCCESS
 }
 
 fn render(counts: &BTreeMap<i64, u64>, summary: &Summary, offset: FixedOffset) -> io::Result<()> {
@@ -109,8 +124,7 @@ fn render(counts: &BTreeMap<i64, u64>, summary: &Summary, offset: FixedOffset) -
         );
     }
 
-    let stdout = io::stdout();
-    let mut out = stdout.lock();
+    let mut out = io::BufWriter::new(io::stdout().lock());
 
     let scale = summary.peak_count.max(1);
     if fill {
@@ -140,7 +154,7 @@ fn render(counts: &BTreeMap<i64, u64>, summary: &Summary, offset: FixedOffset) -
     writeln!(out, "first    {}", fmt_time(offset, summary.first_sec))?;
     writeln!(out, "last     {}", fmt_time(offset, summary.last_sec))?;
     writeln!(out, "span     {span}s (first→last)")?;
-    Ok(())
+    out.flush()
 }
 
 fn write_row(

crates/burst/tests/cli.rs

@@ -54,3 +54,25 @@ fn empty_input_fails_cleanly() {
         "stderr:\n{stderr}"
     );
 }
+
+#[test]
+fn closed_stdout_exits_quietly() {
+    // reader dropped before spawn, so every write hits a closed pipe (`| head`)
+    let (reader, writer) = std::io::pipe().unwrap();
+    drop(reader);
+    let mut child = Command::new(env!("CARGO_BIN_EXE_ts-burst"))
+        .stdin(Stdio::piped())
+        .stdout(writer)
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("spawn burst");
+    let _ = child
+        .stdin
+        .take()
+        .unwrap()
+        .write_all(include_str!("fixtures/simple.txt").as_bytes());
+    let out = child.wait_with_output().expect("wait burst");
+    let stderr = String::from_utf8(out.stderr).unwrap();
+    assert!(out.status.success(), "stderr:\n{stderr}");
+    assert!(!stderr.contains("Broken pipe"), "stderr:\n{stderr}");
+}

crates/correlate/src/main.rs

@@ -43,7 +43,12 @@ fn run() -> Result<ExitCode, String> {
     let mut args = std::env::args().skip(1);
     while let Some(arg) = args.next() {
         match arg.as_str() {
-            "--window" => window_arg = Some(args.next().ok_or("--window requires a value")?),
+            "--window" => {
+                if window_arg.is_some() {
+                    return Err("--window given more than once".into());
+                }
+                window_arg = Some(args.next().ok_or("--window requires a value")?);
+            }
             "-h" | "--help" => {
                 println!("{HELP}");
                 return Ok(ExitCode::SUCCESS);

crates/correlate/tests/cli.rs

@@ -68,3 +68,16 @@ fn rejects_bad_window() {
     let stderr = String::from_utf8(out.stderr).unwrap();
     assert!(stderr.contains("invalid duration"), "stderr:\n{stderr}");
 }
+
+#[test]
+fn rejects_repeated_window() {
+    let a = fixture("a.txt");
+    let b = fixture("b.txt");
+    let out = run(&[a.as_str(), b.as_str(), "--window", "5s", "--window", "1s"]);
+    assert!(!out.status.success());
+    let stderr = String::from_utf8(out.stderr).unwrap();
+    assert!(
+        stderr.contains("--window given more than once"),
+        "stderr:\n{stderr}"
+    );
+}

crates/dist/src/main.rs

@@ -8,10 +8,10 @@
 //!     jq -r .target_processing_time alb.jsonl | ts-dist
 //!     grep responseTime app.log | awk '{print $NF}' | ts-dist
 
-use std::io::{self, BufRead};
+use std::io::{self, BufRead, Write};
 use std::process::ExitCode;
 
-use dist::{parse_number, summarize};
+use dist::{Stats, parse_number, summarize};
 
 const HELP: &str = "\
 ts-dist — summarize the distribution of a column of numbers
@@ -69,14 +69,26 @@ fn main() -> ExitCode {
         return ExitCode::FAILURE;
     };
 
-    println!("count    {:>10}", s.count);
-    println!("min      {:>10.3}", s.min);
-    println!("mean     {:>10.3}", s.mean);
-    println!("p50      {:>10.3}", s.p50);
-    println!("p90      {:>10.3}", s.p90);
-    println!("p95      {:>10.3}", s.p95);
-    println!("p99      {:>10.3}", s.p99);
-    println!("max      {:>10.3}", s.max);
-    println!("stddev   {:>10.3}", s.stddev);
+    let mut out = io::BufWriter::new(io::stdout().lock());
+    if let Err(e) = render(&mut out, &s).and_then(|()| out.flush()) {
+        // Downstream closed the pipe (e.g. `| head`): stop quietly.
+        if e.kind() == io::ErrorKind::BrokenPipe {
+            return ExitCode::SUCCESS;
+        }
+        eprintln!("ts-dist: write error: {e}");
+        return ExitCode::FAILURE;
+    }
     ExitCode::SUCCESS
 }
+
+fn render(out: &mut impl Write, s: &Stats) -> io::Result<()> {
+    writeln!(out, "count    {:>10}", s.count)?;
+    writeln!(out, "min      {:>10.3}", s.min)?;
+    writeln!(out, "mean     {:>10.3}", s.mean)?;
+    writeln!(out, "p50      {:>10.3}", s.p50)?;
+    writeln!(out, "p90      {:>10.3}", s.p90)?;
+    writeln!(out, "p95      {:>10.3}", s.p95)?;
+    writeln!(out, "p99      {:>10.3}", s.p99)?;
+    writeln!(out, "max      {:>10.3}", s.max)?;
+    writeln!(out, "stddev   {:>10.3}", s.stddev)
+}

crates/dist/tests/cli.rs

@@ -40,3 +40,25 @@ fn empty_input_fails_cleanly() {
     let stderr = String::from_utf8(out.stderr).unwrap();
     assert!(stderr.contains("no numbers"), "stderr:\n{stderr}");
 }
+
+#[test]
+fn closed_stdout_exits_quietly() {
+    // reader dropped before spawn, so every write hits a closed pipe (`| head`)
+    let (reader, writer) = std::io::pipe().unwrap();
+    drop(reader);
+    let mut child = Command::new(env!("CARGO_BIN_EXE_ts-dist"))
+        .stdin(Stdio::piped())
+        .stdout(writer)
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("spawn dist");
+    let _ = child
+        .stdin
+        .take()
+        .unwrap()
+        .write_all(include_str!("fixtures/numbers.txt").as_bytes());
+    let out = child.wait_with_output().expect("wait dist");
+    let stderr = String::from_utf8(out.stderr).unwrap();
+    assert!(out.status.success(), "stderr:\n{stderr}");
+    assert!(!stderr.contains("Broken pipe"), "stderr:\n{stderr}");
+}