wsb 0.255.6379

//! A run in a terminal, in the workspace's activity.
//!
//! # The asymmetry this removes
//!
//! Press a button in the workbench and the daemon starts `wsb <argv>` as a
//! child, captures every line it says, writes them to a log and shows the run
//! in the activity panel while it happens. Type the same command in a terminal
//! and none of that was true: the run existed only in the window it was typed
//! in. Two ways to do one thing, and only one of them was visible.
//!
//! Half of it was already fixed. Every wsb run writes an operation log —
//! [`crate::logs::op_log`], the same format and the same directory the daemon
//! writes — so a finished command line has always been readable afterwards.
//! What was missing was the LIVE half: nothing told the daemon a run was
//! happening, so nothing could show it.
//!
//! This is that telling. The run announces itself to the workspace's daemon,
//! streams the lines it is already writing to its own file, and says how it
//! ended. The daemon puts it in the same registry a button's operation goes
//! into, so the panel shows both, from one source, with one shape.
//!
//! # Why a run waits before announcing
//!
//! Most wsb invocations take milliseconds. `wsb daemon status`, `wsb secret
//! has`, the version verbs, the dozens of calls a hook makes — announcing
//! every one of them would fill the activity panel with rows that appeared and
//! vanished, and would put an HTTP round trip in front of commands that are
//! otherwise instant.
//!
//! So a run announces itself only once it is still going after
//! [`ANNOUNCE_AFTER`]. That is not a sampling heuristic: the activity panel is
//! for work in progress, and a run that is already over is not work in
//! progress. Its log is on disk either way, which is where a finished run is
//! read from.
//!
//! Nothing is lost by waiting. Every line said before the announcement is
//! carried in it, so the panel opens on the whole run rather than on its
//! middle.
//!
//! # What this never does
//!
//! Break the run. A daemon that is not there, that stops answering, or that
//! refuses is a reason to stop reporting and nothing else — the command being
//! reported on is the thing that matters, and it has already written its own
//! log.

use std::sync::mpsc::{Receiver, RecvTimeoutError, Sender};
use std::time::{Duration, Instant};

use crate::output::record::Line;

/// How long a run must last before it is worth showing as activity.
///
/// Long enough that the short commands a workspace runs constantly stay out of
/// the panel, short enough that anything a person would watch appears while
/// they are still looking at it.
pub const ANNOUNCE_AFTER: Duration = Duration::from_millis(750);

/// How often lines are sent once a run has been announced.
///
/// A build says thousands of lines a minute. One request per line would be one
/// request per millisecond; one request per batch is one every tick, and the
/// panel is not a terminal — nobody is reading it at a granularity this
/// interval is visible at.
const FLUSH_EVERY: Duration = Duration::from_millis(250);

/// The most lines carried in one request.
///
/// A cap on the request rather than on the run: a batch beyond this is split,
/// so a very loud second cannot build a body the daemon has to buffer whole.
const BATCH: usize = 500;

/// The most lines held while waiting to announce.
///
/// A run that says more than this before [`ANNOUNCE_AFTER`] has elapsed is one
/// whose opening is not worth the memory to keep intact. What is dropped is
/// counted and reported, so the panel says "N earlier lines not shown" rather
/// than presenting a truncated run as a whole one.
const PREFACE: usize = 2000;

/// Where a report goes.
///
/// A trait with one real implementation, for one reason: the interesting part
/// of this module is the DECIDING — whether a run is worth announcing, what an
/// announcement carries, when batches leave, what happens when the far end
/// refuses — and none of that can be tested against a live daemon without
/// starting one. The seam is where the deciding stops and the HTTP begins.
trait Report: Send + 'static {
    fn announce(&self, run: &Run, preface: &[Line], dropped: u64) -> Result<(), String>;
    fn lines(&self, run: &Run, lines: &[Line]) -> Result<(), String>;
    fn finished(&self, run: &Run, exit_code: i32) -> Result<(), String>;
}

/// The real one: this workspace's daemon.
struct ToDaemon(crate::daemon::Client);

impl Report for ToDaemon {
    fn announce(&self, run: &Run, preface: &[Line], dropped: u64) -> Result<(), String> {
        let body = serde_json::json!({
            "id": run.id,
            "kind": run.kind,
            "label": run.label,
            "argv": run.argv,
            "started_at": run.started_at,
            "log_path": run.log_path,
            "lines": preface,
            "dropped": dropped,
        });
        let _: serde_json::Value = self.0.worktree_post("/ops/adopt", &body)?;
        Ok(())
    }

    fn lines(&self, run: &Run, lines: &[Line]) -> Result<(), String> {
        let body = serde_json::json!({ "lines": lines });
        let _: serde_json::Value =
            self.0.worktree_post(&format!("/ops/{}/lines", run.id), &body)?;
        Ok(())
    }

    fn finished(&self, run: &Run, exit_code: i32) -> Result<(), String> {
        let body = serde_json::json!({ "exit_code": exit_code });
        let _: serde_json::Value = self
            .0
            .worktree_post(&format!("/ops/{}/finished", run.id), &body)?;
        Ok(())
    }
}

/// What the run tells its reporter.
enum Message {
    Line(Box<Line>),
    /// The run is over, with the status it ended on and a way to say the
    /// reporting is done.
    Finished(i32, Sender<()>),
}

/// What a run reports about itself, decided once by whoever attached the
/// transcript.
#[derive(Debug, Clone)]
pub struct Run {
    pub id: uuid::Uuid,
    /// The verb, as the log directory names it.
    pub kind: String,
    /// The command line as a person would say it.
    pub label: String,
    pub argv: Vec<String>,
    pub started_at: chrono::DateTime<chrono::Utc>,
    /// The file this run is writing itself to. The daemon shows it rather than
    /// keeping a second copy: there is one log per run, and it is this one.
    pub log_path: String,
}

/// A running report. Dropping it without [`Reporter::finished`] stops the
/// thread without saying how the run ended, which is what a panic leaves
/// behind — and the daemon shows such a run as one that stopped talking,
/// because that is what happened.
pub struct Reporter {
    tx: Sender<Message>,
}

impl Reporter {
    /// Start reporting this run to a daemon.
    ///
    /// Returns immediately. Everything — the wait, the announcement, the
    /// batching, the HTTP — happens on the thread this starts, so a command
    /// line pays nothing for being watched but the cost of cloning a line.
    pub fn start(
        client: crate::daemon::Client,
        run: Run,
        preface: Vec<Line>,
        dropped: u64,
    ) -> Result<Self, String> {
        Self::watching(ToDaemon(client), run, preface, dropped, ANNOUNCE_AFTER)
    }

    /// The same, against any report, with any threshold. What the tests drive.
    fn watching(
        sink: impl Report,
        run: Run,
        preface: Vec<Line>,
        dropped: u64,
        announce_after: Duration,
    ) -> Result<Self, String> {
        let (tx, rx) = std::sync::mpsc::channel();
        std::thread::Builder::new()
            .name("wsb-activity".to_string())
            .spawn(move || report(sink, run, preface, dropped, announce_after, rx))
            .map_err(|e| format!("cannot start the activity reporter: {e}"))?;
        Ok(Self { tx })
    }

    /// One line the run just said.
    pub fn line(&self, line: &Line) {
        // A closed channel is a reporter that gave up; the run carries on.
        let _ = self.tx.send(Message::Line(Box::new(line.clone())));
    }

    /// The run is over. Waits briefly for the reporter to say so.
    ///
    /// Briefly, and never longer: a command's exit must not be held up by a
    /// daemon that has stopped answering. What is lost by giving up is one
    /// row in a panel showing a run as still going until the daemon notices
    /// otherwise, which is a smaller cost than a terminal that will not
    /// return.
    pub fn finished(self, exit_code: i32) {
        let (done, wait) = std::sync::mpsc::channel();
        if self.tx.send(Message::Finished(exit_code, done)).is_err() {
            return;
        }
        let _ = wait.recv_timeout(Duration::from_millis(1500));
    }
}

/// The reporter thread.
fn report(
    sink: impl Report,
    run: Run,
    mut held: Vec<Line>,
    mut dropped: u64,
    announce_after: Duration,
    rx: Receiver<Message>,
) {
    // ── waiting to see whether this run is worth showing ──
    let announce_at = Instant::now() + announce_after;
    loop {
        let left = announce_at.saturating_duration_since(Instant::now());
        if left.is_zero() {
            break;
        }
        match rx.recv_timeout(left) {
            Ok(Message::Line(line)) => {
                held.push(*line);
                while held.len() > PREFACE {
                    held.remove(0);
                    dropped += 1;
                }
            }
            // Over before it was worth announcing. Nothing was said to the
            // daemon, so there is nothing to correct.
            Ok(Message::Finished(_, done)) => {
                let _ = done.send(());
                return;
            }
            Err(RecvTimeoutError::Timeout) => break,
            // The run went away without finishing — a panic, a kill. Nothing
            // was announced, so nothing is left claiming to be running.
            Err(RecvTimeoutError::Disconnected) => return,
        }
    }

    // ── announcing ──
    if sink.announce(&run, &held, dropped).is_err() {
        // Said at debug and no louder. A daemon that will not take an
        // announcement is not this command's problem to report on: the run is
        // recorded in its own file, and a warning on every command line of a
        // machine whose daemon is restarting would be noise about nothing.
        tracing::debug!("this run is not being shown as activity");
        drain(rx);
        return;
    }

    // ── streaming ──
    let mut pending: Vec<Line> = Vec::new();
    let mut next_flush = Instant::now() + FLUSH_EVERY;
    loop {
        let left = next_flush.saturating_duration_since(Instant::now());
        match rx.recv_timeout(left) {
            Ok(Message::Line(line)) => {
                pending.push(*line);
                if pending.len() >= BATCH {
                    if flush(&sink, &run, &mut pending).is_err() {
                        drain(rx);
                        return;
                    }
                    next_flush = Instant::now() + FLUSH_EVERY;
                }
            }
            Ok(Message::Finished(exit_code, done)) => {
                let _ = flush(&sink, &run, &mut pending);
                let _ = sink.finished(&run, exit_code);
                let _ = done.send(());
                return;
            }
            Err(RecvTimeoutError::Timeout) => {
                if flush(&sink, &run, &mut pending).is_err() {
                    drain(rx);
                    return;
                }
                next_flush = Instant::now() + FLUSH_EVERY;
            }
            // The run ended without saying so. The op is left as it stands:
            // the daemon sweeps a run whose reporter stopped talking, and
            // inventing an outcome here would be inventing one.
            Err(RecvTimeoutError::Disconnected) => return,
        }
    }
}

/// Keep receiving until the run lets go, without sending anything.
///
/// The alternative is a run whose `line()` calls start failing the moment the
/// reporter gives up — harmless, since they are already ignored, but this also
/// releases `finished()` promptly instead of leaving it to time out.
fn drain(rx: Receiver<Message>) {
    while let Ok(message) = rx.recv() {
        if let Message::Finished(_, done) = message {
            let _ = done.send(());
            return;
        }
    }
}

/// Send what has piled up, in requests no larger than [`BATCH`].
fn flush(sink: &impl Report, run: &Run, pending: &mut Vec<Line>) -> Result<(), String> {
    if pending.is_empty() {
        return Ok(());
    }
    for chunk in std::mem::take(pending).chunks(BATCH) {
        sink.lines(run, chunk)?;
    }
    Ok(())
}

#[cfg(test)]
#[path = "activity_tests.rs"]
mod tests;