alba 0.2.0

Alba, the Beamfile task runner: parallel beams, caching, watch mode, and an interactive interface
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
//! Renderers: the three ways a run's [`RunEvent`] stream is presented.
//!
//! A renderer is a plain [`Renderer::handle`] per event, called from a
//! single task that owns the receiving end of the engine's channel — so an
//! implementation never needs to synchronize anything, and events reach it
//! in exactly the order the engine emitted them.
//!
//! ## Which stream output goes to
//!
//! The two text renderers write *everything* they produce — line prefixes,
//! group headers, and beam output from both `stdout` and `stderr` — to the
//! process's stdout. Alba's own commentary (the run summary, the
//! `cancelling...` announcement, error diagnostics) goes to stderr. The
//! split is "the run's output" versus "Alba talking about the run", not
//! "the beam's stdout" versus "the beam's stderr": a prefix or a group
//! header has to stay contiguous with the lines it frames, which two
//! independently buffered streams cannot guarantee. The stream a line came
//! from is preserved losslessly by [`json::JsonRenderer`], which is the
//! mode built for a consumer that cares.
//!
//! ## Every line goes through a [`LineSink`]
//!
//! Nothing here uses `println!`/`eprintln!`. See [`LineSink`] for why: a
//! run whose reader walked away (`alba run x | head -1`) must still finish,
//! still report, and still exit with the code its beams earned.

use std::io::{self, Write};
use std::time::Duration;

use alba_engine::{BeamStatus, RunEvent, RunSummary};

mod grouped;
mod interleaved;
mod json;

pub use grouped::GroupedRenderer;
pub use interleaved::InterleavedRenderer;
pub use json::JsonRenderer;

/// A consumer of a run's event stream.
///
/// `Send` because the only caller drives it from a spawned task running
/// concurrently with the engine, so output is streamed while beams are
/// still running rather than buffered until the run ends.
pub trait Renderer: Send {
    fn handle(&mut self, event: &RunEvent);
}

/// A line-oriented writer that gives up quietly once its reader is gone.
///
/// `println!` and `eprintln!` *panic* when the write fails, and the write
/// fails as soon as the other end of a pipe closes. `alba run x | head -1`
/// is a pipeline users write without thinking about it, and under the
/// macros it panicked the renderer task mid-run: the summary was never
/// printed, and the process ended up reporting success regardless of what
/// the beams actually did.
///
/// So every line a renderer emits goes through here instead. The first
/// failed write latches the sink closed and every later write is a cheap
/// no-op — the run keeps going to completion with nobody left to read it,
/// and [`crate::commands::run`] still returns the code the run earned.
/// Any write error latches, not only `BrokenPipe`: a stdout that cannot be
/// written to is not a condition Alba can recover from or usefully report
/// (the report would go to the same broken stream), and retrying it once
/// per line for the rest of the run would be worse than silence.
pub struct LineSink<W: Write> {
    writer: W,
    open: bool,
}

impl<W: Write> LineSink<W> {
    pub fn new(writer: W) -> Self {
        Self { writer, open: true }
    }

    /// Writes `text` followed by a newline, or does nothing if this sink
    /// has already been closed by a failed write.
    pub fn line(&mut self, text: &str) {
        if !self.open {
            return;
        }
        if writeln!(self.writer, "{text}").is_err() {
            self.open = false;
        }
    }

    /// Writes `text` exactly as given — no newline — and flushes, or does
    /// nothing once the sink is closed. Exists for the clear-screen escape
    /// sequence, which must not be followed by a newline and must reach the
    /// terminal before the next run's first line.
    pub fn raw(&mut self, text: &str) {
        if !self.open {
            return;
        }
        if write!(self.writer, "{text}")
            .and_then(|()| self.writer.flush())
            .is_err()
        {
            self.open = false;
        }
    }
}

impl LineSink<io::Stdout> {
    pub fn stdout() -> Self {
        Self::new(io::stdout())
    }
}

impl LineSink<io::Stderr> {
    pub fn stderr() -> Self {
        Self::new(io::stderr())
    }
}

/// How long something took, as one decimal of a second (`4.1s`).
///
/// Deliberately the same shape at every magnitude — no minute/hour
/// breakdown — because the only thing that would buy is prettiness at
/// durations Alba's own tests never produce, and a second format is a
/// second thing to keep stable across platforms. Nothing in the test suite
/// asserts a duration's value, only that one is present.
pub fn format_duration(duration: Duration) -> String {
    format!("{:.1}s", duration.as_secs_f64())
}

/// How a beam ended, as it appears in a text renderer's status line.
///
/// `failed (allowed)` deliberately omits the exit code that `failed`
/// reports: an allowed failure does not affect the run, so the number is
/// noise, and the phrase is what the user greps for.
pub fn status_label(status: &BeamStatus) -> String {
    match status {
        BeamStatus::Succeeded => "ok".to_string(),
        BeamStatus::Cached => "cached".to_string(),
        BeamStatus::Failed { exit_code } => format!("failed (exit {exit_code})"),
        BeamStatus::FailedAllowed { .. } => "failed (allowed)".to_string(),
        BeamStatus::Cancelled => "cancelled".to_string(),
    }
}

/// The status line a watch event renders to, `None` for every other
/// event. Shared by the two text renderers so their phrasing cannot
/// drift; both print it to stderr — it is Alba talking about the run,
/// not the run's output (see the module doc comment).
pub(super) fn watch_line(event: &RunEvent) -> Option<String> {
    match event {
        RunEvent::WatchWaiting { files } => {
            let plural = if *files == 1 { "" } else { "s" };
            Some(format!(
                "watching — {files} file{plural}, waiting for changes"
            ))
        }
        RunEvent::WatchTriggered { paths } if paths.is_empty() => {
            Some("changes detected — running".to_string())
        }
        RunEvent::WatchTriggered { paths } => {
            let named = paths.iter().take(3).cloned().collect::<Vec<_>>().join(", ");
            let rest = paths.len().saturating_sub(3);
            if rest == 0 {
                Some(format!("change detected in {named} — running"))
            } else {
                Some(format!(
                    "change detected in {named} and {rest} more — running"
                ))
            }
        }
        _ => None,
    }
}

/// The two text renderers' shared handling of a watch event: prints
/// [`watch_line`] to `err` and, for a triggered run when `clear_between_runs`
/// is set, clears the screen on `out` first. Returns whether `event` was a
/// watch event at all, so a renderer's `handle` can `return` on `true` and
/// fall through to its own match otherwise.
///
/// Centralized here rather than repeated in `interleaved.rs` and
/// `grouped.rs`: both renderers happen to store their sinks under the same
/// `out`/`err` names, so the control flow and the clear-screen escape
/// sequence are exactly as duplicable as the phrasing `watch_line` already
/// guards against drifting.
pub(super) fn handle_watch_event(
    event: &RunEvent,
    out: &mut LineSink<io::Stdout>,
    err: &mut LineSink<io::Stderr>,
    clear_between_runs: bool,
) -> bool {
    let Some(line) = watch_line(event) else {
        return false;
    };
    if matches!(event, RunEvent::WatchTriggered { .. }) && clear_between_runs {
        // \x1b[2J clears the screen, \x1b[3J the scrollback, \x1b[H homes
        // the cursor: each triggered run starts on a clean page.
        out.raw("\u{1b}[2J\u{1b}[3J\u{1b}[H");
    }
    err.line(&line);
    true
}

/// Writes the one-line run summary to `sink` (stderr), e.g.
/// `✓ 3 succeeded · ✗ 1 failed · ⊘ 2 cancelled · 4.1s`.
///
/// Empty buckets are omitted rather than reported as zero: a run where
/// nothing failed should not spend half its summary saying so. The total
/// duration is always last, so the line is never empty even for a run
/// whose every bucket happens to be empty.
///
/// `affected_by` is the reference an `--affected` run targeted, `None` for
/// an ordinary run; see [`summary_line`] for what it changes.
///
/// Generic over the sink's writer rather than fixed to stderr: the
/// renderers all pass their own `LineSink<io::Stderr>`, while a test can
/// pass one over a buffer and read back the very line the user sees.
pub fn print_summary<W: Write>(
    sink: &mut LineSink<W>,
    summary: &RunSummary,
    affected_by: Option<&str>,
) {
    sink.line(&summary_line(summary, affected_by));
}

/// The exact `--affected` empty-run line, verbatim wherever a reference
/// found nothing to run: the headless short-circuit in
/// [`crate::commands::run`] prints it before a renderer even exists, and
/// [`summary_line`] prints the same text for a run that reaches its
/// summary with no beam in any bucket. One definition, so the two call
/// sites cannot drift apart.
pub fn nothing_affected_line(reference: &str) -> String {
    format!("\u{2713} nothing affected by {reference}")
}

/// The summary's text, split out from the writing so it can be asserted
/// directly rather than through a captured stream.
///
/// An `--affected` run with nothing to show for it — no bucket holds a
/// single beam — reports [`nothing_affected_line`] instead of an empty
/// count line: every bucket empty is what an ordinary run's summary would
/// print as a bare duration, which would read as "a run happened and
/// nothing came of it" rather than "there was nothing to run".
fn summary_line(summary: &RunSummary, affected_by: Option<&str>) -> String {
    if let Some(reference) = affected_by
        && summary.succeeded.is_empty()
        && summary.cached.is_empty()
        && summary.failed.is_empty()
        && summary.failed_allowed.is_empty()
        && summary.cancelled.is_empty()
    {
        return nothing_affected_line(reference);
    }

    let counts = [
        ("\u{2713}", summary.succeeded.len(), "succeeded"),
        ("\u{21ba}", summary.cached.len(), "cached"),
        ("\u{2717}", summary.failed.len(), "failed"),
        ("\u{26a0}", summary.failed_allowed.len(), "failed (allowed)"),
        ("\u{2298}", summary.cancelled.len(), "cancelled"),
    ];

    let mut parts: Vec<String> = counts
        .iter()
        .filter(|(_, count, _)| *count > 0)
        .map(|(mark, count, label)| format!("{mark} {count} {label}"))
        .collect();
    parts.push(format_duration(summary.duration));

    parts.join(" \u{b7} ")
}

#[cfg(test)]
mod tests {
    use super::*;
    use alba_core::BeamId;

    #[test]
    fn formats_durations_with_one_decimal() {
        assert_eq!(format_duration(Duration::from_millis(1234)), "1.2s");
        assert_eq!(format_duration(Duration::ZERO), "0.0s");
    }

    #[test]
    fn labels_a_failure_with_its_exit_code_but_an_allowed_one_without() {
        assert_eq!(
            status_label(&BeamStatus::Failed { exit_code: 7 }),
            "failed (exit 7)"
        );
        assert_eq!(
            status_label(&BeamStatus::FailedAllowed { exit_code: 1 }),
            "failed (allowed)"
        );
    }

    #[test]
    fn a_cached_beam_is_labelled_cached() {
        assert_eq!(status_label(&BeamStatus::Cached), "cached");
    }

    #[test]
    fn the_summary_counts_cached_beams() {
        let summary = RunSummary {
            succeeded: vec![BeamId("a".to_string())],
            cached: vec![BeamId("b".to_string()), BeamId("c".to_string())],
            duration: Duration::from_millis(4100),
            ..RunSummary::default()
        };

        assert_eq!(
            summary_line(&summary, None),
            "\u{2713} 1 succeeded \u{b7} \u{21ba} 2 cached \u{b7} 4.1s"
        );
    }

    #[test]
    fn the_summary_omits_empty_buckets_and_always_ends_with_the_duration() {
        let summary = RunSummary {
            succeeded: vec![BeamId("a".to_string())],
            cancelled: vec![BeamId("b".to_string()), BeamId("c".to_string())],
            duration: Duration::from_millis(4100),
            ..RunSummary::default()
        };

        assert_eq!(
            summary_line(&summary, None),
            "\u{2713} 1 succeeded \u{b7} \u{2298} 2 cancelled \u{b7} 4.1s"
        );
    }

    /// The empty-run line, verbatim: `--affected <ref>` with nothing to
    /// show for it reports why, rather than a bare duration.
    #[test]
    fn an_affected_run_with_no_targets_says_nothing_was_affected() {
        assert_eq!(
            summary_line(&RunSummary::default(), Some("HEAD")),
            "\u{2713} nothing affected by HEAD"
        );
    }

    /// `affected_by` only changes the empty case: a run that did affect
    /// something reports its counts as usual.
    #[test]
    fn an_affected_run_with_targets_reports_its_counts_as_usual() {
        let summary = RunSummary {
            succeeded: vec![BeamId("a".to_string())],
            duration: Duration::from_millis(4100),
            ..RunSummary::default()
        };

        assert_eq!(
            summary_line(&summary, Some("HEAD")),
            "\u{2713} 1 succeeded \u{b7} 4.1s"
        );
    }

    /// A writer whose every write fails, standing in for the closed end of
    /// a pipe: the sink must swallow it rather than panic, and must not
    /// keep trying afterwards.
    struct BrokenWriter {
        attempts: usize,
    }

    impl Write for BrokenWriter {
        fn write(&mut self, _: &[u8]) -> io::Result<usize> {
            self.attempts += 1;
            Err(io::Error::new(io::ErrorKind::BrokenPipe, "broken pipe"))
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    #[test]
    fn a_broken_sink_swallows_the_failure_and_stops_writing() {
        let mut sink = LineSink::new(BrokenWriter { attempts: 0 });

        sink.line("first");
        sink.line("second");
        sink.line("third");

        assert_eq!(
            sink.writer.attempts, 1,
            "the sink must latch closed after the first failure"
        );
    }

    #[test]
    fn an_open_sink_writes_every_line() {
        let mut sink = LineSink::new(Vec::new());

        sink.line("first");
        sink.line("second");

        assert_eq!(String::from_utf8(sink.writer).unwrap(), "first\nsecond\n");
    }

    #[test]
    fn watch_lines_read_as_status_not_as_beam_output() {
        assert_eq!(
            watch_line(&RunEvent::WatchWaiting { files: 3 }),
            Some("watching — 3 files, waiting for changes".to_string())
        );
        // A one-file session is an ordinary one, not a plural.
        assert_eq!(
            watch_line(&RunEvent::WatchWaiting { files: 1 }),
            Some("watching — 1 file, waiting for changes".to_string())
        );
        // A session parked on a project that will not load resolves
        // nothing, and says so rather than claiming to be running.
        assert_eq!(
            watch_line(&RunEvent::WatchWaiting { files: 0 }),
            Some("watching — 0 files, waiting for changes".to_string())
        );
        assert_eq!(
            watch_line(&RunEvent::WatchTriggered {
                paths: vec!["a.rs".into(), "b.rs".into()],
            }),
            Some("change detected in a.rs, b.rs — running".to_string())
        );
        // Beyond three paths, name the first three and count the rest.
        assert_eq!(
            watch_line(&RunEvent::WatchTriggered {
                paths: vec!["a".into(), "b".into(), "c".into(), "d".into(), "e".into()],
            }),
            Some("change detected in a, b, c and 2 more — running".to_string())
        );
        // A rescan has no paths to name.
        assert_eq!(
            watch_line(&RunEvent::WatchTriggered { paths: vec![] }),
            Some("changes detected — running".to_string())
        );
        assert_eq!(
            watch_line(&RunEvent::RunFinished {
                summary: RunSummary::default()
            }),
            None
        );
    }
}