scope_cli/cli/

repeat.rs

//! The one repeat runner for commands that take `--every` and `--duration`
//! (#58).
//!
//! It owns the interval, the duration, the run count, the progress record,
//! the final-run rule and the rule that an error ends the repeat. A command
//! supplies one run (`step`). Each run's key values go to the CSV file and
//! to the screen's series (TUI); in the CLI the run's report prints as
//! before.

use crate::cli::output::{Output, RunRecord};
use crate::{note, out};
use scope::error::{Result, ScopeError};
use scope::metrics::{self as series, Metric, Sample};
use std::future::Future;
use std::io::Write as _;
use std::path::Path;
use std::time::{Duration, Instant};
// The schedule runs on tokio's clock, the one its sleep uses.
use tokio::time::Instant as Clock;

/// The timestamp format of a CSV row.
const CSV_TIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%SZ";

/// When a repeating command runs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Schedule {
    /// The time from one run's start to the next.
    pub every: Duration,
    /// How long the repeat lasts; a last run starts when it ends. `None`
    /// runs until the user cancels it (#96).
    pub duration: Option<Duration>,
}

impl Schedule {
    /// A schedule; the interval must be positive.
    pub fn new(every: Duration, duration: Duration) -> Result<Self> {
        Self::with(every, Some(duration))
    }

    /// A schedule that runs until the user cancels it.
    pub fn until_cancelled(every: Duration) -> Result<Self> {
        Self::with(every, None)
    }

    fn with(every: Duration, duration: Option<Duration>) -> Result<Self> {
        if every.is_zero() {
            return Err(ScopeError::InvalidInput(
                "Interval must be positive".to_string(),
            ));
        }
        Ok(Self { every, duration })
    }

    /// How many runs it makes: one every `every` over `duration`, and a last
    /// one at the end when the span is not a whole number of intervals (the
    /// final sleep is cut to the time left). `None` until cancelled.
    pub fn total(&self) -> Option<u64> {
        let every = self.every.as_millis().max(1);
        let total = self.duration?.as_millis().div_ceil(every).saturating_add(1);
        Some(u64::try_from(total).unwrap_or(u64::MAX))
    }
}

/// What one run gives the runner.
#[derive(Debug, Clone)]
pub struct RunOutput<T> {
    /// The run's key values, in the order of the runner's metric table.
    pub sample: Sample,
    /// The run's report, as the CLI prints it.
    pub snapshot: String,
    /// The command's own result of the run (for example for `--report`).
    pub value: T,
    /// The run's data, for a screen that draws its own view of it.
    pub view: Option<crate::cli::output::SummaryView>,
}

/// Runs `step` on `schedule` and returns the last run's value.
///
/// Before each run in the CLI, it notes `--- Run #N at HH:MM:SS ---`; after
/// it, the CLI prints the snapshot. With a screen series (TUI), the record
/// goes to the series instead, so the pane does not stack one report per
/// run. With `csv`, the file gets the header, then one row per run. An
/// error from `step` ends the repeat and is returned.
pub async fn run<T, F, Fut>(
    schedule: &Schedule,
    metrics: &[Metric],
    csv: Option<&Path>,
    out: &Output,
    mut step: F,
) -> Result<Option<T>>
where
    F: FnMut(u64) -> Fut,
    Fut: Future<Output = Result<RunOutput<T>>>,
{
    // The CSV columns, only for a CSV: a caller without one may pass an
    // empty table (a watchlist writes its own trail).
    let columns = match csv {
        Some(_) => series::csv_columns(metrics).map_err(|e| ScopeError::Other(e.to_string()))?,
        None => Vec::new(),
    };
    if let Some(path) = csv {
        std::fs::write(path, series::csv_header(&columns))
            .map_err(|e| ScopeError::io_context(e, format!("cannot write {}", path.display())))?;
    }
    out.update_progress(|p| {
        p.run = 0;
        p.total = schedule.total();
        p.next_at = None;
    });

    let tui = out.series_hook().is_some();
    let start = Clock::now();
    let end = schedule.duration.map(|d| start + d);
    let mut slot = start;
    let mut run: u64 = 1;
    loop {
        let began = Clock::now();
        out.update_progress(|p| {
            p.run = run;
            p.next_at = None;
        });
        if !tui {
            let ts = chrono::Utc::now().format("%H:%M:%S");
            note!(out, "--- Run #{} at {} ---\n", run, ts);
        }
        // An error ends the repeat: `?` returns it and no later run starts.
        let output = step(run).await?;
        let at = chrono::Utc::now();
        if let Some(path) = csv {
            let time = at.format(CSV_TIME_FORMAT).to_string();
            let row = series::csv_row(&columns, metrics, &time, run, &output.sample);
            append(path, &row)?;
        }
        let record = RunRecord {
            run,
            at,
            sample: output.sample,
            snapshot: output.snapshot.clone(),
            view: output.view.map(std::sync::Arc::new),
        };
        if !out.record_run(metrics, record) {
            out!(out, "{}", output.snapshot)?;
        }
        let last = output.value;

        // The repeat ends after the run that starts at the end, not after
        // any run that finishes past it: a run that crosses the end is
        // followed at once by the last run, so the count meets `total()`.
        if let (Some(duration), Some(end)) = (schedule.duration, end)
            && began >= end
        {
            note!(
                out,
                "\nCompleted {} run(s) over {}s.",
                run,
                duration.as_secs()
            );
            return Ok(Some(last));
        }
        run += 1;
        // Runs start on a fixed grid, so the time a run takes does not push
        // the later ones back. A run that overruns its slot moves the grid
        // instead of starting a burst of catch-up runs.
        slot = (slot + schedule.every).max(Clock::now());
        if let Some(end) = end {
            slot = slot.min(end);
        }
        let pause = slot.saturating_duration_since(Clock::now());
        out.update_progress(|p| p.next_at = Some(Instant::now() + pause));
        tokio::time::sleep_until(slot).await;
    }
}

fn append(path: &Path, row: &str) -> Result<()> {
    let context = || format!("cannot append to {}", path.display());
    let mut file = std::fs::OpenOptions::new()
        .append(true)
        .open(path)
        .map_err(|e| ScopeError::io_context(e, context()))?;
    file.write_all(row.as_bytes())
        .map_err(|e| ScopeError::io_context(e, context()))
}

#[cfg(test)]
mod tests {
    use super::*;
    use scope::market::HealthThresholds;
    use scope::metrics::summary_metrics;

    fn metrics() -> Vec<Metric> {
        summary_metrics(&HealthThresholds::default())
    }

    fn output(run: u64, metrics: &[Metric]) -> RunOutput<u64> {
        RunOutput {
            sample: Sample {
                values: vec![Some(run as f64); metrics.len()],
                healthy: true,
                failed_checks: vec![],
            },
            snapshot: format!("report {run}\n"),
            view: None,
            value: run,
        }
    }

    #[test]
    fn test_total_counts_the_final_run_at_the_boundary() {
        // 5s every 2s runs at 0, 2, 4 and then at 5 (the sleep is cut to
        // the time left): four runs, not 5/2+1.
        let s = |every, duration| {
            Schedule::new(Duration::from_secs(every), Duration::from_secs(duration)).unwrap()
        };
        assert_eq!(s(2, 5).total(), Some(4));
        assert_eq!(s(1, 2).total(), Some(3));
        assert_eq!(s(2, 1).total(), Some(2));
        // Until cancelled: there is no total to show.
        let open = Schedule::until_cancelled(Duration::from_secs(1)).unwrap();
        assert_eq!(open.total(), None);
    }

    #[tokio::test(start_paused = true)]
    async fn test_until_cancelled_runs_past_any_duration_with_flat_memory() {
        // #96: a run with no end must go on until the user stops it (here an
        // error at run 1001, as Esc would), past the 1 h default, and keep
        // only the newest MAX_SERIES_RUNS runs in memory.
        let m = metrics();
        let (out, _cap) = Output::capture_merged();
        let schedule = Schedule::until_cancelled(Duration::from_secs(30)).unwrap();
        let stop = 1_001;
        let result: Result<Option<u64>> = run(&schedule, &m, None, &out, |n| {
            let o = output(n, &m);
            async move {
                if n == stop {
                    Err(ScopeError::Other("cancelled".into()))
                } else {
                    Ok(o)
                }
            }
        })
        .await;
        assert!(result.is_err());
        let progress = out.progress_hook().unwrap().lock().unwrap().clone();
        assert_eq!(progress.run, stop, "it ran past 1 h (121 runs at 30 s)");
        assert_eq!(progress.total, None);
        let series = out.series_hook().unwrap();
        let series = series.lock().unwrap();
        assert_eq!(series.runs.len(), crate::cli::output::MAX_SERIES_RUNS);
        assert_eq!(
            series.dropped,
            stop - 1 - crate::cli::output::MAX_SERIES_RUNS as u64
        );
    }

    #[test]
    fn test_a_zero_interval_is_the_users_error() {
        let err = Schedule::new(Duration::ZERO, Duration::from_secs(1)).unwrap_err();
        assert!(matches!(err, ScopeError::InvalidInput(_)), "{err:?}");
    }

    #[tokio::test(start_paused = true)]
    async fn test_cli_prints_each_report_and_returns_the_last_value() {
        let m = metrics();
        let (out, cap) = Output::capture();
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(2)).unwrap();
        let last = run(&schedule, &m, None, &out, |n| {
            let o = output(n, &m);
            async move { Ok(o) }
        })
        .await
        .unwrap();
        assert_eq!(last, Some(3));
        assert_eq!(cap.out(), "report 1\nreport 2\nreport 3\n");
        assert!(cap.err().contains("--- Run #3 at"), "{}", cap.err());
        assert!(
            cap.err().contains("Completed 3 run(s) over 2s."),
            "{}",
            cap.err()
        );
    }

    #[tokio::test(start_paused = true)]
    async fn test_tui_gets_records_and_no_repeated_reports() {
        let m = metrics();
        let (out, cap) = Output::capture_merged();
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(1)).unwrap();
        run(&schedule, &m, None, &out, |n| {
            let o = output(n, &m);
            async move { Ok(o) }
        })
        .await
        .unwrap();
        assert!(!cap.out().contains("report 1"), "{}", cap.out());
        assert!(!cap.out().contains("--- Run #"), "{}", cap.out());
        let series = out.series_hook().unwrap();
        let series = series.lock().unwrap();
        assert_eq!(series.metrics, m);
        let runs: Vec<u64> = series.runs.iter().map(|r| r.run).collect();
        assert_eq!(runs, vec![1, 2]);
        assert_eq!(series.runs[1].snapshot, "report 2\n");
    }

    #[tokio::test(start_paused = true)]
    async fn test_an_error_ends_the_repeat() {
        let m = metrics();
        let (out, _cap) = Output::capture_merged();
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(30)).unwrap();
        let mut calls = 0;
        let result: Result<Option<u64>> = run(&schedule, &m, None, &out, |n| {
            calls += 1;
            let o = output(n, &m);
            async move {
                if n == 2 {
                    Err(ScopeError::Network("venue down".into()))
                } else {
                    Ok(o)
                }
            }
        })
        .await;
        assert!(result.is_err());
        assert_eq!(calls, 2, "no run after the error");
        let progress = out.progress_hook().unwrap().lock().unwrap().clone();
        assert_eq!(progress.run, 2);
        // The run before the error stays in the series.
        assert_eq!(out.series_hook().unwrap().lock().unwrap().runs.len(), 1);
    }

    #[tokio::test(start_paused = true)]
    async fn test_csv_gets_the_header_then_one_row_per_run() {
        let m = metrics();
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("s.csv");
        let (out, _cap) = Output::capture();
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(1)).unwrap();
        run(&schedule, &m, Some(&path), &out, |n| {
            let o = output(n, &m);
            async move { Ok(o) }
        })
        .await
        .unwrap();
        let body = std::fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = body.lines().collect();
        let columns = series::csv_columns(&m).unwrap();
        assert_eq!(format!("{}\n", lines[0]), series::csv_header(&columns));
        assert_eq!(lines.len(), 3, "{body}");
        assert_eq!(lines[2].split(',').nth(1), Some("2"));
    }

    /// Runs `schedule` with a step whose run `n` takes `cost(n)`, and
    /// returns each run's start time in milliseconds from the start.
    async fn start_times(schedule: Schedule, cost: impl Fn(u64) -> Duration) -> Vec<u128> {
        let m = metrics();
        let (out, _cap) = Output::capture_merged();
        let start = Clock::now();
        let starts = std::sync::Mutex::new(Vec::new());
        run(&schedule, &m, None, &out, |n| {
            starts.lock().unwrap().push(start.elapsed().as_millis());
            let o = output(n, &m);
            let cost = cost(n);
            async move {
                tokio::time::sleep(cost).await;
                Ok(o)
            }
        })
        .await
        .unwrap();
        starts.into_inner().unwrap()
    }

    #[tokio::test(start_paused = true)]
    async fn test_run_time_does_not_push_later_runs_back() {
        // A 0.75 s fetch every 30 s made "run 18 of 121" start at 523 s, and
        // the hour ended near run 118. Runs must stay on the 30 s grid.
        let schedule = Schedule::new(Duration::from_secs(30), Duration::from_secs(90)).unwrap();
        let starts = start_times(schedule, |_| Duration::from_millis(750)).await;
        assert_eq!(starts, vec![0, 30_000, 60_000, 90_000]);
        assert_eq!(Some(starts.len() as u64), schedule.total());
    }

    #[tokio::test(start_paused = true)]
    async fn test_an_overrun_moves_the_grid_without_a_burst() {
        // A slow first run (3.5 s, every 1 s) starts the next one at once,
        // and later runs keep the interval from there. A fixed grid would
        // fire the missed slots back to back at 3.5 s, 3.6 s and 3.7 s.
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(6)).unwrap();
        let cost = |n| Duration::from_millis(if n == 1 { 3_500 } else { 100 });
        let starts = start_times(schedule, cost).await;
        assert_eq!(starts, vec![0, 3_500, 4_500, 5_500, 6_000]);
    }

    #[tokio::test(start_paused = true)]
    async fn test_a_run_that_reaches_the_end_still_gets_the_last_run() {
        // Every 1 s for 1 s: a first run of 1 s or more used to end the
        // repeat after one run while `total()` said 2. The last run starts
        // at the end, or at once when run 1 finishes past it.
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(1)).unwrap();
        for (cost, second) in [(5, 1_000), (999, 1_000), (1_000, 1_000), (2_500, 2_500)] {
            let starts = start_times(schedule, |n| {
                Duration::from_millis(if n == 1 { cost } else { 10 })
            })
            .await;
            assert_eq!(starts, vec![0, second], "first run of {cost} ms");
            assert_eq!(Some(starts.len() as u64), schedule.total());
        }
    }

    #[tokio::test(start_paused = true)]
    async fn test_a_run_that_crosses_the_end_is_followed_by_one_last_run() {
        // Every 2 s for 5 s: runs at 0, 2 and 4; run 3 takes 1.5 s and ends
        // past 5 s, so the last run starts at once and nothing follows it.
        let schedule = Schedule::new(Duration::from_secs(2), Duration::from_secs(5)).unwrap();
        let cost = |n| Duration::from_millis(if n == 3 { 1_500 } else { 100 });
        let starts = start_times(schedule, cost).await;
        assert_eq!(starts, vec![0, 2_000, 4_000, 5_500]);
        assert_eq!(Some(starts.len() as u64), schedule.total());
    }

    #[tokio::test(start_paused = true)]
    async fn test_progress_reaches_the_total_when_a_run_crosses_the_end() {
        // `run N of M` must end at M = N, not stop one short.
        let m = metrics();
        let (out, _cap) = Output::capture_merged();
        let schedule = Schedule::new(Duration::from_secs(1), Duration::from_secs(1)).unwrap();
        run(&schedule, &m, None, &out, |n| {
            let o = output(n, &m);
            async move {
                tokio::time::sleep(Duration::from_millis(1_200)).await;
                Ok(o)
            }
        })
        .await
        .unwrap();
        let progress = out.progress_hook().unwrap().lock().unwrap().clone();
        assert_eq!(progress.total, Some(2));
        assert_eq!(progress.run, 2);
    }

    #[test]
    fn test_the_series_ring_keeps_the_newest_runs() {
        let m = metrics();
        let (out, _cap) = Output::capture_merged();
        let extra = 3;
        for n in 1..=(crate::cli::output::MAX_SERIES_RUNS as u64 + extra) {
            let o = output(n, &m);
            out.record_run(
                &m,
                RunRecord {
                    run: n,
                    at: chrono::Utc::now(),
                    sample: o.sample,
                    snapshot: o.snapshot,
                    view: None,
                },
            );
        }
        let series = out.series_hook().unwrap();
        let series = series.lock().unwrap();
        assert_eq!(series.runs.len(), crate::cli::output::MAX_SERIES_RUNS);
        assert_eq!(series.dropped, extra);
        assert_eq!(series.runs.front().unwrap().run, extra + 1);
    }
}