pito_tui/
command.rs

1use std::io::{self, BufRead, BufReader, Read};
2use std::mem;
3use std::process::{Child, Command, ExitStatus, Stdio};
4use std::sync::atomic::{AtomicUsize, Ordering};
5use std::sync::mpsc::Sender;
6use std::sync::{Arc, Mutex, MutexGuard};
7use std::thread;
8use std::time::{Duration, Instant};
9
10use ratatui::text::Line;
11use serde_json::Value;
12
13use crate::activity::{Activity, State};
14use crate::wake::Wake;
15
16pub const GRACE: Duration = Duration::from_secs(2);
17const POST: Duration = Duration::from_millis(100);
18const REAP: Duration = Duration::from_millis(1);
19
20#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
21#[non_exhaustive]
22pub enum Stream {
23    #[default]
24    Stdout,
25    Stderr,
26    Both,
27}
28
29impl Stream {
30    fn stdout(self) -> bool {
31        matches!(self, Stream::Stdout | Stream::Both)
32    }
33
34    fn stderr(self) -> bool {
35        matches!(self, Stream::Stderr | Stream::Both)
36    }
37}
38
39#[derive(Debug, Clone, PartialEq)]
40#[non_exhaustive]
41pub struct Progress {
42    pub state: String,
43    pub stage: Option<String>,
44    pub fraction: Option<f64>,
45    pub message: Option<String>,
46    pub ok: Option<bool>,
47}
48
49#[derive(Debug, Clone, PartialEq)]
50#[non_exhaustive]
51pub enum Heard {
52    #[non_exhaustive]
53    Line {
54        stream: Stream,
55        text: String,
56        progress: Option<Progress>,
57    },
58    Exit(Exit),
59}
60
61#[derive(Debug, Clone, PartialEq, Eq)]
62#[non_exhaustive]
63pub enum Exit {
64    Code(i32),
65    Signal(i32),
66    Stopped,
67    Error(String),
68}
69
70impl Exit {
71    pub fn state(&self) -> State {
72        match self {
73            Exit::Code(0) => State::Done,
74            Exit::Stopped => State::Stopped,
75            _ => State::Failed,
76        }
77    }
78
79    fn of(status: io::Result<ExitStatus>) -> Self {
80        match status {
81            Ok(status) => status
82                .code()
83                .map_or_else(|| Exit::Signal(killed(status)), Exit::Code),
84            Err(error) => Exit::Error(error.to_string()),
85        }
86    }
87}
88
89#[cfg(unix)]
90fn killed(status: ExitStatus) -> i32 {
91    std::os::unix::process::ExitStatusExt::signal(&status).unwrap_or_default()
92}
93
94#[cfg(not(unix))]
95fn killed(_status: ExitStatus) -> i32 {
96    0
97}
98
99#[derive(Debug, Clone)]
100pub struct Report {
101    pub(crate) activity: Activity,
102    pub(crate) heard: Vec<Heard>,
103}
104
105pub fn read(line: &str) -> Option<Progress> {
106    let value: Value = serde_json::from_str(line.trim()).ok()?;
107    let object = value.as_object()?;
108    if object.get("v").is_some_and(|v| v.as_u64() != Some(1)) {
109        return None;
110    }
111    let text = |key: &str| {
112        object
113            .get(key)
114            .and_then(Value::as_str)
115            .filter(|text| !text.is_empty())
116            .map(str::to_string)
117    };
118    Some(Progress {
119        state: text("state")?,
120        stage: text("stage"),
121        fraction: object.get("fraction").and_then(Value::as_f64),
122        message: text("msg"),
123        ok: object.get("ok").and_then(Value::as_bool),
124    })
125}
126
127fn note(progress: &Progress) -> String {
128    let mut parts: Vec<String> = Vec::new();
129    parts.extend(progress.stage.clone());
130    parts.push(progress.state.clone());
131    if let Some(fraction) = progress.fraction {
132        parts.push(crate::progress::share(fraction, false));
133    }
134    parts.extend(progress.message.clone());
135    parts.join("  ")
136}
137
138pub(crate) fn follow(activity: &mut Activity, progress: &Progress, now: Instant) {
139    if activity.state.finished() {
140        return;
141    }
142    if let Some(text) = progress.message.clone().or_else(|| progress.stage.clone()) {
143        activity.status = text;
144    }
145    if progress.state == "end" {
146        activity.state = if progress.ok == Some(true) {
147            State::Done
148        } else {
149            State::Failed
150        };
151        activity.progress = None;
152        activity.ended = Some(now);
153        return;
154    }
155    activity.state = State::Running;
156    activity.progress = progress.fraction;
157}
158
159pub(crate) fn halt(activity: &mut Activity, now: Instant) {
160    activity.state = State::Stopped;
161    activity.progress = None;
162    activity.ended = Some(now);
163}
164
165pub(crate) struct Run {
166    pub(crate) screen: usize,
167    pub(crate) activity: Activity,
168    pub(crate) command: Command,
169    pub(crate) progress: Stream,
170    pub(crate) keep: bool,
171}
172
173pub(crate) struct Handle {
174    pub(crate) id: u64,
175    child: Arc<Mutex<Child>>,
176    shared: Arc<Mutex<Shared>>,
177}
178
179impl Handle {
180    pub(crate) fn over(&self) -> bool {
181        lock(&self.shared).done
182    }
183
184    pub(crate) fn stop(&self, now: Instant) -> Option<Activity> {
185        let activity = self.stopping(now)?;
186        signal(&self.child, false);
187        let child = Arc::clone(&self.child);
188        thread::spawn(move || {
189            thread::sleep(GRACE);
190            signal(&child, true);
191        });
192        Some(activity)
193    }
194
195    fn stopping(&self, now: Instant) -> Option<Activity> {
196        let mut state = lock(&self.shared);
197        if state.activity.state.finished() {
198            return None;
199        }
200        halt(&mut state.activity, now);
201        state.stopped = true;
202        state.dirty = true;
203        Some(state.snapshot())
204    }
205
206    fn alive(&self) -> bool {
207        matches!(lock(&self.child).try_wait(), Ok(None))
208    }
209
210    pub(crate) fn forget(&self) {
211        let mut state = lock(&self.shared);
212        state.keep = false;
213        state.lines = Vec::new();
214    }
215}
216
217pub(crate) fn end(handles: &[Handle], now: Instant) -> Vec<Activity> {
218    let stopped = handles
219        .iter()
220        .filter_map(|handle| handle.stopping(now))
221        .collect();
222    for handle in handles {
223        signal(&handle.child, false);
224    }
225    let limit = Instant::now() + GRACE;
226    let mut pause = REAP;
227    while handles.iter().any(Handle::alive) {
228        let left = limit.saturating_duration_since(Instant::now());
229        if left.is_zero() {
230            break;
231        }
232        thread::sleep(pause.min(left));
233        pause = (pause * 2).min(POST);
234    }
235    for handle in handles {
236        signal(&handle.child, true);
237    }
238    stopped
239}
240
241fn signal(child: &Mutex<Child>, hard: bool) {
242    let mut child = lock(child);
243    if !matches!(child.try_wait(), Ok(None)) {
244        return;
245    }
246    #[cfg(unix)]
247    {
248        use rustix::process::{Pid, Signal, kill_process_group};
249        let signal = if hard { Signal::KILL } else { Signal::TERM };
250        let _ = kill_process_group(Pid::from_child(&child), signal);
251    }
252    #[cfg(not(unix))]
253    {
254        let _ = hard;
255        let _ = child.kill();
256    }
257}
258
259fn reap(child: &Mutex<Child>) -> io::Result<ExitStatus> {
260    let mut pause = REAP;
261    loop {
262        if let Some(status) = lock(child).try_wait()? {
263            return Ok(status);
264        }
265        thread::sleep(pause);
266        pause = (pause * 2).min(POST);
267    }
268}
269
270struct Shared {
271    activity: Activity,
272    lines: Vec<Line<'static>>,
273    heard: Vec<Heard>,
274    keep: bool,
275    stopped: bool,
276    dirty: bool,
277    done: bool,
278}
279
280impl Shared {
281    fn push(&mut self, line: impl FnOnce() -> String) {
282        if self.keep {
283            self.lines.push(Line::from(line()));
284        }
285        self.dirty = true;
286    }
287
288    fn snapshot(&self) -> Activity {
289        let mut activity = self.activity.clone();
290        activity.detail = Arc::new(self.lines.clone());
291        activity
292    }
293
294    fn post(&mut self) -> Report {
295        self.dirty = false;
296        Report {
297            activity: self.snapshot(),
298            heard: mem::take(&mut self.heard),
299        }
300    }
301
302    fn end(&mut self, exit: Exit) {
303        self.heard.push(Heard::Exit(exit));
304        self.done = true;
305        self.dirty = true;
306    }
307}
308
309fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
310    mutex
311        .lock()
312        .unwrap_or_else(|poisoned| poisoned.into_inner())
313}
314
315fn drain(stream: impl Read, shared: &Mutex<Shared>, from: Stream, progress: bool) {
316    let mut reader = BufReader::new(stream);
317    let mut bytes = Vec::new();
318    loop {
319        bytes.clear();
320        match reader.read_until(b'\n', &mut bytes) {
321            Ok(0) | Err(_) => return,
322            Ok(_) => {}
323        }
324        let line = bytes.strip_suffix(b"\n").unwrap_or(&bytes);
325        let line = line.strip_suffix(b"\r").unwrap_or(line);
326        let line = String::from_utf8_lossy(line).into_owned();
327        let parsed = read(&line).filter(|_| progress);
328        let mut state = lock(shared);
329        match &parsed {
330            Some(parsed) => {
331                follow(&mut state.activity, parsed, Instant::now());
332                state.push(|| note(parsed));
333            }
334            None => state.push(|| line.clone()),
335        }
336        state.heard.push(Heard::Line {
337            stream: from,
338            text: line,
339            progress: parsed,
340        });
341    }
342}
343
344pub(crate) fn spawn<E: Send + 'static>(
345    run: Run,
346    sender: Sender<Wake<E>>,
347    running: Arc<AtomicUsize>,
348) -> Option<Handle> {
349    let Run {
350        screen,
351        activity,
352        mut command,
353        progress,
354        keep,
355    } = run;
356    command
357        .stdin(Stdio::null())
358        .stdout(Stdio::piped())
359        .stderr(Stdio::piped());
360    #[cfg(unix)]
361    std::os::unix::process::CommandExt::process_group(&mut command, 0);
362    let id = activity.id;
363    let shared = Arc::new(Mutex::new(Shared {
364        activity,
365        lines: Vec::new(),
366        heard: Vec::new(),
367        keep,
368        stopped: false,
369        dirty: true,
370        done: false,
371    }));
372    let mut child = match command.spawn() {
373        Ok(child) => child,
374        Err(error) => {
375            let mut state = lock(&shared);
376            state.activity.state = State::Failed;
377            state.activity.status = error.to_string();
378            state.activity.ended = Some(Instant::now());
379            state.end(Exit::Error(error.to_string()));
380            let _ = sender.send(Wake::Command(screen, state.post()));
381            return None;
382        }
383    };
384    running.fetch_add(1, Ordering::SeqCst);
385    let stdout = child.stdout.take();
386    let stderr = child.stderr.take();
387    let child = Arc::new(Mutex::new(child));
388    let errors = Arc::clone(&shared);
389    let quiet = thread::spawn(move || {
390        if let Some(stderr) = stderr {
391            drain(stderr, &errors, Stream::Stderr, progress.stderr());
392        }
393    });
394    let reader = Arc::clone(&shared);
395    let reaped = Arc::clone(&child);
396    thread::spawn(move || {
397        if let Some(stdout) = stdout {
398            drain(stdout, &reader, Stream::Stdout, progress.stdout());
399        }
400        let _ = quiet.join();
401        let status = reap(&reaped);
402        let mut state = lock(&reader);
403        if !state.activity.state.finished() {
404            let ok = status.as_ref().is_ok_and(|status| status.success());
405            state.activity.state = if ok { State::Done } else { State::Failed };
406            state.activity.progress = None;
407            state.activity.ended = Some(Instant::now());
408        }
409        let exit = if state.stopped {
410            Exit::Stopped
411        } else {
412            Exit::of(status)
413        };
414        state.end(exit);
415    });
416    let handle = Handle {
417        id,
418        child,
419        shared: Arc::clone(&shared),
420    };
421    thread::spawn(move || {
422        loop {
423            thread::sleep(POST);
424            let mut state = lock(&shared);
425            if state.dirty && sender.send(Wake::Command(screen, state.post())).is_err() {
426                break;
427            }
428            if state.done {
429                break;
430            }
431        }
432        running.fetch_sub(1, Ordering::SeqCst);
433    });
434    Some(handle)
435}
436
437#[cfg(test)]
438mod tests {
439    #[cfg(unix)]
440    use std::sync::mpsc;
441
442    use super::*;
443
444    const START: &str = r#"{"v":1,"op":"update","app":"dispatch","ref":"v0.9.1","stage":"clone","state":"start","fraction":null,"msg":"cloning pito-dispatch at v0.9.1","at":1790950000415}"#;
445    const STEP: &str = r#"{"v":1,"op":"update","app":"dispatch","ref":null,"stage":"build","state":"progress","fraction":0.5,"msg":null,"at":1790950000900,"id":"7"}"#;
446    const END: &str = r#"{"v":1,"op":"update","app":"dispatch","ref":"v0.9.1","state":"end","ok":true,"msg":"pdispatch 0.9.1 is installed","at":1790950057026,"store":{"kept":3}}"#;
447    const FAILED: &str = r#"{"v":1,"op":"update","app":"dispatch","ref":null,"state":"end","ok":false,"msg":"","at":1790950057026}"#;
448
449    #[test]
450    fn progress_lines_carry_an_activity_from_its_start_to_its_end() {
451        let now = Instant::now();
452        let mut activity = Activity::new(1, "update", now);
453        follow(&mut activity, &read(START).unwrap(), now);
454        assert_eq!(activity.status, "cloning pito-dispatch at v0.9.1");
455        assert_eq!(activity.progress, None);
456        follow(&mut activity, &read(STEP).unwrap(), now);
457        assert_eq!(
458            (activity.state, activity.progress),
459            (State::Running, Some(0.5))
460        );
461        assert_eq!(activity.status, "build");
462        follow(&mut activity, &read(END).unwrap(), now);
463        assert_eq!(activity.state, State::Done);
464        assert_eq!(activity.status, "pdispatch 0.9.1 is installed");
465        assert_eq!(activity.ended, Some(now));
466
467        let mut failed = Activity::new(2, "update", now);
468        follow(&mut failed, &read(FAILED).unwrap(), now);
469        assert_eq!(failed.state, State::Failed);
470        assert!(read(r#"{"v":2,"state":"start"}"#).is_none());
471        assert!(read("cloning…").is_none());
472    }
473
474    #[cfg(unix)]
475    fn started(
476        script: &str,
477        progress: Stream,
478    ) -> (Handle, mpsc::Receiver<Wake<()>>, Arc<AtomicUsize>) {
479        let mut command = Command::new("sh");
480        command.args(["-c", script]);
481        let run = Run {
482            screen: 0,
483            activity: Activity::new(1, "sim", Instant::now()),
484            command,
485            progress,
486            keep: true,
487        };
488        let (sender, inbox) = mpsc::channel();
489        let running = Arc::new(AtomicUsize::new(0));
490        let handle = spawn(run, sender, Arc::clone(&running)).unwrap();
491        (handle, inbox, running)
492    }
493
494    #[cfg(unix)]
495    fn last(
496        inbox: &mpsc::Receiver<Wake<()>>,
497        until: impl Fn(&Activity) -> bool,
498    ) -> (Activity, Vec<Heard>) {
499        let limit = Instant::now() + Duration::from_secs(10);
500        let mut seen = None;
501        let mut heard = Vec::new();
502        while let Ok(wake) = inbox.recv_timeout(limit.saturating_duration_since(Instant::now())) {
503            if let Wake::Command(_, report) = wake {
504                heard.extend(report.heard);
505                let done = until(&report.activity);
506                seen = Some(report.activity);
507                if done {
508                    break;
509                }
510            }
511        }
512        (seen.unwrap(), heard)
513    }
514
515    #[cfg(unix)]
516    #[test]
517    fn progress_lines_are_followed_on_the_stream_the_app_picks() {
518        let script = format!("echo '{FAILED}' >&2; echo plain");
519        let (_, inbox, _) = started(&script, Stream::Stderr);
520        let (ended, _) = last(&inbox, |_| false);
521        assert_eq!(ended.state, State::Failed);
522        assert!(ended.detail.iter().any(|line| line.to_string() == "plain"));
523        let (_, inbox, _) = started(&script, Stream::Stdout);
524        let (ended, _) = last(&inbox, |_| false);
525        assert_eq!(ended.state, State::Done);
526        assert!(ended.detail.iter().any(|line| line.to_string() == FAILED));
527    }
528
529    #[cfg(unix)]
530    #[test]
531    fn a_stopped_command_ends_at_once_as_stopped_and_keeps_its_detail() {
532        let script = format!("echo '{STEP}'; echo started; sleep 30; echo '{END}'");
533        let (handle, inbox, running) = started(&script, Stream::Stdout);
534        let said = |activity: &Activity| {
535            activity
536                .detail
537                .iter()
538                .any(|line| line.to_string() == "started")
539        };
540        assert!(said(&last(&inbox, said).0));
541        let asked = Instant::now();
542        let stopped = handle.stop(asked).unwrap();
543        assert_eq!(
544            (stopped.state, stopped.ended, stopped.progress),
545            (State::Stopped, Some(asked), None)
546        );
547        assert!(handle.stop(asked).is_none());
548        let (ended, heard) = last(&inbox, |_| false);
549        assert!(asked.elapsed() < GRACE);
550        assert_eq!(ended.state, State::Stopped);
551        assert_eq!(heard.last(), Some(&Heard::Exit(Exit::Stopped)));
552        assert!(said(&ended));
553        assert!(handle.over());
554        assert_eq!(running.load(Ordering::SeqCst), 0);
555    }
556
557    #[cfg(unix)]
558    #[test]
559    fn the_end_of_the_loop_stops_every_command_and_kills_one_that_ignores_the_signal() {
560        let said = |activity: &Activity| {
561            activity
562                .detail
563                .iter()
564                .any(|line| line.to_string() == "started")
565        };
566        for (script, stubborn) in [
567            ("echo started; sleep 30", false),
568            ("trap '' TERM; echo started; sleep 30", true),
569        ] {
570            let (handle, inbox, running) = started(script, Stream::Stdout);
571            assert!(said(&last(&inbox, said).0));
572            let asked = Instant::now();
573            let stopped = end(std::slice::from_ref(&handle), asked);
574            assert_eq!(stopped.len(), 1);
575            assert_eq!(asked.elapsed() >= GRACE, stubborn);
576            let (ended, heard) = last(&inbox, |_| false);
577            assert!(asked.elapsed() < GRACE + Duration::from_secs(1));
578            assert_eq!(ended.state, State::Stopped);
579            assert_eq!(heard.last(), Some(&Heard::Exit(Exit::Stopped)));
580            assert!(handle.over());
581            assert_eq!(running.load(Ordering::SeqCst), 0);
582        }
583    }
584
585    #[cfg(unix)]
586    #[test]
587    fn a_forgotten_command_keeps_no_lines_and_its_screen_still_hears_them() {
588        let (handle, inbox, _) = started("echo one; sleep 0.3; echo two", Stream::Stdout);
589        let one =
590            |activity: &Activity| activity.detail.iter().any(|line| line.to_string() == "one");
591        let (_, mut heard) = last(&inbox, one);
592        handle.forget();
593        let (ended, rest) = last(&inbox, |_| false);
594        heard.extend(rest);
595        assert!(ended.detail.is_empty());
596        let texts: Vec<&str> = heard
597            .iter()
598            .filter_map(|heard| match heard {
599                Heard::Line { text, .. } => Some(text.as_str()),
600                _ => None,
601            })
602            .collect();
603        assert_eq!(texts, ["one", "two"]);
604    }
605
606    #[cfg(unix)]
607    #[test]
608    fn the_screen_hears_every_line_with_its_stream_then_the_exit() {
609        let script = format!("echo '{STEP}' >&2; printf 'one\\377\\n'; echo two; exit 3");
610        let (_, inbox, _) = started(&script, Stream::Stderr);
611        let (_, heard) = last(&inbox, |_| false);
612        let lines = |from: Stream| -> Vec<(String, bool)> {
613            heard
614                .iter()
615                .filter_map(|heard| match heard {
616                    Heard::Line {
617                        stream,
618                        text,
619                        progress,
620                    } if *stream == from => Some((text.clone(), progress.is_some())),
621                    _ => None,
622                })
623                .collect()
624        };
625        assert_eq!(
626            lines(Stream::Stdout),
627            [
628                ("one\u{FFFD}".to_string(), false),
629                ("two".to_string(), false)
630            ]
631        );
632        assert_eq!(lines(Stream::Stderr), [(STEP.to_string(), true)]);
633        assert_eq!(heard.last(), Some(&Heard::Exit(Exit::Code(3))));
634    }
635}
pito_tui/command.rs

pito_tui/
command.rs
