//! Progress reporting for pipeline runs: a full-screen dashboard when
//! stdout is a terminal, or line-oriented output when it is not (or when
//! `--plain` asks for it).
//!
//! Both reporters consume the pipeline's event channel until it
//! disconnects (which happens exactly when the pipeline finishes) and
//! let the user request suspension — with a key in the dashboard, or
//! SIGINT/SIGTERM in either mode.
//!
//! The dashboard answers three questions at once: what the run has
//! accomplished (the totals the pipeline itself keeps), what its workers
//! are doing this instant (the stages table), and what is queued ahead
//! of them (the table's waiting column for how deep each queue is now,
//! the backlog pane for which way it is going, and the pending pane for
//! the paths a run has yet to visit). Completed
//! items scroll past beside those, and skips — which are reports rather
//! than problems, and can be very numerous — are reduced to the most
//! recent one alongside their count.
use std::{
collections::{HashMap, VecDeque},
io::Write as _,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
time::{Duration, Instant},
};
use crossbeam_channel::Receiver;
use crossterm::event::{Event as TermEvent, KeyCode, KeyModifiers};
use ratatui::{
Frame,
layout::{Constraint, Layout},
style::{Modifier, Style},
text::Line,
widgets::{Block, Borders, Cell, List, ListItem, Paragraph, Row, Table},
};
use beeping::pipeline::{EarlyFinish, Event, Session, SkipReason, Stage, StoreLimit};
/// History kept for the activity pane; the visible tail is cut to the
/// pane's actual height at render time.
const HISTORY_LINES: usize = 128;
/// How much run time separates two backlog samples. The pipeline
/// reports several times a second, which would fill the backlog pane
/// with a few seconds of noise; a couple of seconds a column gives it
/// several minutes, which is the span over which a queue meaningfully
/// grows or drains.
const BACKLOG_SAMPLE: Duration = Duration::from_secs(2);
/// Backlog samples kept per stage, comfortably more than the widest
/// pane, so that history survives a narrow window being widened.
const BACKLOG_HISTORY: usize = 512;
/// The width the activity log gives its verb, so that the paths beside
/// it line up.
const VERB_WIDTH: u16 = 8;
/// How much recent history throughput is averaged over. Long enough to
/// ride out one slow file, short enough to still mean "right now".
const RATE_WINDOW: Duration = Duration::from_secs(5);
/// The shortest span of samples that yields a throughput figure; below
/// it the quotient is more noise than measurement.
const RATE_MIN_SPAN: Duration = Duration::from_millis(500);
/// One line of the activity log.
enum Activity {
/// Something that happened to a path. The verb and the path are
/// kept apart rather than formatted when the event arrives, because
/// how much room there is for the path is not known until the pane
/// is drawn — and what must survive the cut is its tail.
Item { verb: &'static str, path: String },
/// Something the run has to say about itself, which is prose rather
/// than a path and is left whole.
Note(String),
}
impl Activity {
fn render(&self, width: u16) -> String {
match self {
Activity::Item { verb, path } => format!(
"{verb:<label$} {}",
fit_path(path, width.saturating_sub(VERB_WIDTH + 1)),
label = VERB_WIDTH as usize
),
Activity::Note(note) => note.clone(),
}
}
}
/// One item in a worker's hands.
struct InHand {
/// When it was taken up, so the newest and the oldest can be told
/// apart.
sequence: u64,
path: String,
/// How far through it the worker has got, where the stage knows and
/// the item is big enough for it to matter.
progress: Option<(u64, u64)>,
}
impl InHand {
/// The item as a table cell: how far through it the worker is, then
/// as much of its path as the column can hold. The percentage keeps
/// its full width — it is the part that changes, and the part that
/// says whether anything is happening at all.
fn describe(&self, width: u16) -> String {
match self.progress {
Some((done, total)) if total > 0 => {
let percent = (done.saturating_mul(100) / total).min(100);
let shown = format!("{percent:>3}% ");
format!(
"{shown}{}",
fit_path(&self.path, width.saturating_sub(shown.len() as u16))
)
}
_ => fit_path(&self.path, width),
}
}
}
/// What one stage of the pipeline is doing, and how much is queued for
/// it.
struct StageView {
stage: Stage,
/// How many workers the stage runs, or `None` for the stages whose
/// work finalization does rather than a pool.
threads: Option<usize>,
/// The items in workers' hands, by worker index.
busy: HashMap<usize, InHand>,
waiting: u64,
/// How deep the queue has been, oldest sample first: what the
/// backlog pane draws.
history: VecDeque<u64>,
/// The head of the queue by name, in the order it will be taken up,
/// or `None` for a stage whose backlog is not a queue of paths.
upcoming: Option<Vec<String>>,
}
impl StageView {
fn new(stage: Stage) -> StageView {
StageView {
stage,
threads: None,
busy: HashMap::new(),
waiting: 0,
history: VecDeque::new(),
upcoming: None,
}
}
/// Takes one backlog sample, dropping the oldest once the history is
/// longer than any pane could use.
fn record(&mut self, waiting: u64) {
self.history.push_back(waiting);
if self.history.len() > BACKLOG_HISTORY {
self.history.pop_front();
}
}
/// The item the stage took up most recently and still holds: where
/// the run has got to.
fn newest(&self) -> Option<&InHand> {
self.busy.values().max_by_key(|item| item.sequence)
}
/// The item that has been in the stage's hands longest: the one
/// holding the stage up, which the newest item hides whenever the
/// other workers are churning past it. Named for what it measures —
/// how long it has been held — rather than for the slowness that is
/// usually, but not always, the reason.
///
/// Both are drawn from the items actually in hand, so neither ever
/// names something already finished, and a stage with one item names
/// it in both columns.
fn oldest(&self) -> Option<&InHand> {
self.busy.values().min_by_key(|item| item.sequence)
}
/// The "busy" column: workers occupied out of workers available.
fn occupancy(&self) -> String {
match self.threads {
Some(threads) => format!("{}/{}", self.busy.len(), threads),
None => "—".to_string(),
}
}
}
/// A throughput estimate in bytes per second, over the last few seconds
/// of totals samples.
///
/// Both quantities are the pipeline's own — the time it has spent
/// running and the bytes it has moved — so the estimate does not depend
/// on how often the display redraws.
#[derive(Default)]
struct Rate {
samples: VecDeque<(Duration, u64)>,
}
impl Rate {
fn observe(&mut self, elapsed: Duration, bytes: u64) {
// Samples only ever move forward within a run; anything else is
// a fresh pipeline reporting into the same display
if let Some((newest, _)) = self.samples.back()
&& elapsed < *newest
{
self.samples.clear();
}
self.samples.push_back((elapsed, bytes));
// Keep the shortest run of samples that still spans the window
while self.samples.len() > 2 && elapsed - self.samples[1].0 >= RATE_WINDOW {
self.samples.pop_front();
}
}
fn per_second(&self) -> Option<u64> {
let (oldest, then) = *self.samples.front()?;
let (newest, now) = *self.samples.back()?;
let span = newest - oldest;
if span < RATE_MIN_SPAN {
return None;
}
Some((now.saturating_sub(then) as f64 / span.as_secs_f64()) as u64)
}
}
/// What the run is doing, as far as its events have said.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
enum Phase {
#[default]
Working,
/// Storing a manifest for everything backed up so far, so that the
/// suspended run can be restored from.
Checkpointing,
/// All content is stored; the snapshot is being built.
Finalizing,
}
/// Everything the reporters display, fed by pipeline events.
#[derive(Default)]
struct Dashboard {
/// The run's totals, as the pipeline last reported them. It keeps
/// these counts for its own durable state, so taking them from it
/// rather than re-deriving them here keeps one set of numbers.
totals: Session,
phase: Phase,
recent: VecDeque<Activity>,
last_skip: Option<String>,
/// Stage views in the order the pipeline first mentioned them, which
/// is the order work flows through them.
stages: Vec<StageView>,
/// Throughput of finished content: the bytes of files stored or
/// restored.
content_rate: Rate,
/// Throughput to and from the repository. Distinct from the content
/// rate, and worth showing beside it: a run chewing through one huge
/// file finishes no content while transferring steadily, and a
/// re-backup of unchanged files does the reverse.
transfer_rate: Rate,
/// Bytes moved to or from the repository this session, accumulated
/// as they are reported and sampled whenever new totals arrive.
transferred: u64,
/// Ticks up with every item picked up anywhere, so each stage can
/// tell which of the items in its hands it took up last.
sequence: u64,
/// The pipeline's running time as of the last backlog sample, which
/// is what paces the next one.
backlog_sampled: Option<Duration>,
}
impl Dashboard {
fn apply(&mut self, event: Event) {
match event {
Event::Scanned { path } => self.push_recent(Activity::Item {
verb: "scanned",
path: path.display().to_string(),
}),
Event::Stored { path, .. } => self.push_recent(Activity::Item {
verb: "stored",
path: path.display().to_string(),
}),
Event::Restored { path, .. } => self.push_recent(Activity::Item {
verb: "restored",
path: path.display().to_string(),
}),
Event::Skipped { path, reason } => {
self.last_skip = Some(format!("{} ({})", path.display(), describe_skip(&reason)));
}
Event::Checkpointing => self.phase = Phase::Checkpointing,
Event::Working => self.phase = Phase::Working,
Event::FinishingEarly(reason) => self.push_recent(Activity::Note(format!(
"finishing early: {}",
describe_early_finish(reason)
))),
Event::LimitNotApplied(limit) => {
self.push_recent(Activity::Note(describe_unapplied_limit(limit)))
}
Event::Finalizing => self.phase = Phase::Finalizing,
Event::Resumed(session) => {
self.observe(session);
self.push_recent(Activity::Note(
"resumed: totals include previous sessions".to_string(),
));
}
Event::Totals(session) => self.observe(session),
Event::Staffing { stage, threads } => self.stage(stage).threads = Some(threads),
Event::Began {
stage,
worker,
path,
} => {
self.sequence += 1;
let sequence = self.sequence;
self.stage(stage).busy.insert(
worker,
InHand {
sequence,
path: path.display().to_string(),
progress: None,
},
);
}
Event::Progressed {
stage,
worker,
done,
total,
} => {
if let Some(item) = self.stage(stage).busy.get_mut(&worker) {
item.progress = Some((done, total));
}
}
Event::Idle { stage, worker } => {
self.stage(stage).busy.remove(&worker);
}
Event::Backlog(stages) => {
// Depths are taken as they arrive; history is sampled
// against the pipeline's own running time, as the
// throughput figures are, because the display is fed
// far more often than the history needs a column.
let elapsed = self.totals.elapsed;
// Time that has gone backwards is a fresh pipeline
// reporting into the same display, and its
// predecessor's queues do not continue into this one
if self.backlog_sampled.is_some_and(|last| elapsed < last) {
self.backlog_sampled = None;
for view in &mut self.stages {
view.history.clear();
}
}
let sampling = self
.backlog_sampled
.is_none_or(|last| elapsed - last >= BACKLOG_SAMPLE);
if sampling {
self.backlog_sampled = Some(elapsed);
}
for backlog in stages {
let view = self.stage(backlog.stage);
view.waiting = backlog.waiting;
view.upcoming = backlog.upcoming.map(|paths| {
paths
.iter()
.map(|path| path.display().to_string())
.collect()
});
if sampling {
view.record(backlog.waiting);
}
}
}
Event::Transferred { bytes } => self.transferred += bytes,
}
}
/// The view of one stage, created on first mention so the table
/// grows in pipeline order without either reporter knowing which
/// stages a given pipeline has.
fn stage(&mut self, stage: Stage) -> &mut StageView {
let position = self
.stages
.iter()
.position(|view| view.stage == stage)
.unwrap_or_else(|| {
self.stages.push(StageView::new(stage));
self.stages.len() - 1
});
&mut self.stages[position]
}
/// Takes a fresh set of totals, and with them a sampling point for
/// both throughput figures — the pipeline's own running time is the
/// clock they are measured against.
fn observe(&mut self, session: Session) {
self.totals = session;
self.content_rate.observe(
session.elapsed,
session.progress.stored_bytes + session.progress.restored_bytes,
);
self.transfer_rate
.observe(session.elapsed, self.transferred);
}
fn push_recent(&mut self, line: Activity) {
self.recent.push_back(line);
if self.recent.len() > HISTORY_LINES {
self.recent.pop_front();
}
}
fn summary(&self) -> String {
let progress = &self.totals.progress;
let mut parts = Vec::new();
if progress.scanned > 0 {
parts.push(format!("{} directories scanned", count(progress.scanned)));
}
if progress.stored_files > 0 {
parts.push(format!(
"{} files stored ({})",
count(progress.stored_files),
human_bytes(progress.stored_bytes)
));
}
if progress.restored_files > 0 {
parts.push(format!(
"{} files restored ({})",
count(progress.restored_files),
human_bytes(progress.restored_bytes)
));
}
if progress.skipped > 0 {
parts.push(format!("{} skipped", count(progress.skipped)));
}
if parts.is_empty() {
parts.push("no activity".to_string());
}
parts.join(", ")
}
/// What the run is doing right now, as one line: its state, the time
/// it has spent working (across every session, not just this one),
/// and its recent throughput.
fn status(&self, suspending: bool) -> String {
// A checkpoint happens *because* the run is suspending, and is
// the more informative of the two things to say: it can take a
// while, and it is what makes the interrupted backup restorable
let state = match (self.phase, suspending) {
(Phase::Checkpointing, _) => "checkpointing — making what is stored so far restorable",
(Phase::Finalizing, _) => "finalizing",
(Phase::Working, true) => "suspending — letting workers finish their current items",
(Phase::Working, false) => "running",
};
let mut parts = vec![state.to_string()];
if self.totals.elapsed >= Duration::from_secs(1) {
parts.push(format!("elapsed {}", human_duration(self.totals.elapsed)));
}
if let Some(rate) = self.content_rate.per_second() {
parts.push(format!("content {}/s", human_bytes(rate)));
}
if let Some(rate) = self.transfer_rate.per_second() {
parts.push(format!("repository {}/s", human_bytes(rate)));
}
parts.join(" · ")
}
/// The skip line: how many were skipped, and the most recent one.
/// Only the latest is worth screen space — they are reports, not
/// problems, and a long run produces a great many of them.
///
/// A resumed run counts the skips of earlier sessions without having
/// a recent one to name, so the two parts are reported separately.
fn skip_line(&self) -> String {
let skipped = self.totals.progress.skipped;
match (&self.last_skip, skipped) {
(Some(skip), _) => format!("skipped {} · last: {skip}", count(skipped)),
(None, 0) => "skipped none".to_string(),
(None, skipped) => format!("skipped {}", count(skipped)),
}
}
/// The stages table flattened onto one line, for the line-oriented
/// reporter: the same occupancy, backlog, and current item the
/// dashboard shows in columns.
fn stage_line(&self) -> Option<String> {
if self.stages.is_empty() {
return None;
}
let parts: Vec<String> = self
.stages
.iter()
.map(|view| {
let mut part = match view.threads {
Some(threads) => format!(
"{} {}/{} busy, {} waiting",
view.stage.label(),
view.busy.len(),
threads,
count(view.waiting)
),
None => format!("{} {} waiting", view.stage.label(), count(view.waiting)),
};
// The two named items, collapsed to one when a stage
// holds a single item and they are the same
match (view.newest(), view.oldest()) {
(Some(newest), Some(oldest)) if newest.sequence == oldest.sequence => {
part.push_str(&format!(" ({})", newest.describe(u16::MAX)));
}
(Some(newest), Some(oldest)) => {
part.push_str(&format!(
" ({}, oldest {})",
newest.describe(u16::MAX),
oldest.describe(u16::MAX)
));
}
_ => {}
}
part
})
.collect();
Some(parts.join(" · "))
}
}
/// Why a run stopped short of the tree, as the tail of a sentence.
pub fn describe_early_finish(reason: EarlyFinish) -> String {
match reason {
EarlyFinish::Requested => "the run was asked to publish what it had and stop".to_string(),
EarlyFinish::StoreNearlyFull { free, floor } => format!(
"the store has {} left, {} below the {} this backup was told to \
leave free",
human_bytes(free),
human_bytes(floor.saturating_sub(free)),
human_bytes(floor)
),
EarlyFinish::StoreAtMaximum { used, limit } => format!(
"the repository holds {}, at the {} this backup was told to keep \
it under",
human_bytes(used),
human_bytes(limit)
),
}
}
/// Why a limit the run was given is not being applied to it.
pub fn describe_unapplied_limit(limit: StoreLimit) -> String {
match limit {
StoreLimit::MinimumFreeSpace => {
"the store cannot say how much room is left on it, so no room is \
being left free"
.to_string()
}
StoreLimit::MaximumStoreSize => {
"the store cannot say how much it is holding, so the repository \
is not being kept to a size"
.to_string()
}
}
}
fn describe_skip(reason: &SkipReason) -> String {
match reason {
SkipReason::Excluded => "excluded".to_string(),
SkipReason::MountPoint => "mount point".to_string(),
SkipReason::Vanished => "vanished".to_string(),
SkipReason::Unreadable(detail) => format!("unreadable: {detail}"),
SkipReason::SpecialFile => "special file".to_string(),
SkipReason::UnsafePath => "unsafe path".to_string(),
SkipReason::Internal => "beeping's own working data".to_string(),
}
}
pub fn human_bytes(bytes: u64) -> String {
const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];
let mut value = bytes as f64;
let mut unit = 0;
while value >= 1024.0 && unit < UNITS.len() - 1 {
value /= 1024.0;
unit += 1;
}
if unit == 0 {
format!("{bytes} B")
} else {
format!("{value:.1} {}", UNITS[unit])
}
}
/// Whole-second durations, since sub-second precision in a figure that
/// counts hours is just noise.
fn human_duration(elapsed: Duration) -> String {
humantime::format_duration(Duration::from_secs(elapsed.as_secs())).to_string()
}
/// A count with thousands separators: these numbers reach the millions,
/// where an unpunctuated run of digits cannot be read at a glance.
pub fn count(value: u64) -> String {
let digits = value.to_string();
let mut out = String::with_capacity(digits.len() + digits.len() / 3);
for (position, digit) in digits.chars().enumerate() {
if position > 0 && (digits.len() - position).is_multiple_of(3) {
out.push(',');
}
out.push(digit);
}
out
}
/// The newest history lines that fit inside a bordered pane, oldest
/// first, so the pane fills its whole height and appears to scroll.
/// Each is cut to what the pane's width leaves it.
fn visible_tail(lines: &VecDeque<Activity>, area: ratatui::layout::Rect) -> Vec<ListItem<'static>> {
let height = area.height.saturating_sub(2) as usize; // borders
let width = area.width.saturating_sub(2);
lines
.iter()
.skip(lines.len().saturating_sub(height))
.map(|line| ListItem::new(line.render(width)))
.collect()
}
/// The narrow columns of the stages table, and the single space ratatui
/// puts between each pair of its five columns — what the two path
/// columns have to share the rest of the width with.
const STAGE_COLUMNS: [u16; 3] = [6, 7, 11];
const STAGE_GAPS: u16 = 4;
/// How wide each of the two path columns may be inside a table of this
/// total width, borders included.
fn path_columns(width: u16) -> (u16, u16) {
let paths = width
.saturating_sub(2)
.saturating_sub(STAGE_COLUMNS.iter().sum::<u16>() + STAGE_GAPS);
(paths / 2, paths - paths / 2)
}
/// Fits a path into a column, keeping its tail.
///
/// Clipping the other end would be worse than useless here: everything
/// on screen is being backed up from the same tree, so the leading
/// components are the part they all share, and the name at the end is
/// the part that says which item this is.
fn fit_path(path: &str, width: u16) -> String {
let width = width as usize;
if width == 0 {
return String::new();
}
let length = path.chars().count();
if length <= width {
return path.to_string();
}
let tail: String = path.chars().skip(length - width + 1).collect();
format!("…{tail}")
}
/// The stages table: one row per stage, in pipeline order, showing how
/// many of its workers are occupied, how much work is queued for it, and
/// the two items worth naming — the one it took up last and the one it
/// has held longest, which is the one holding it up. The stages without
/// workers of their own name their items too, once finalization reaches
/// them.
fn stages_table(dashboard: &Dashboard, width: u16) -> Table<'_> {
let dim = Style::default().add_modifier(Modifier::DIM);
let (newest, oldest) = path_columns(width);
let rows = dashboard.stages.iter().map(|view| {
Row::new([
Cell::from(view.stage.label()),
Cell::from(Line::from(view.occupancy()).right_aligned()),
Cell::from(Line::from(count(view.waiting)).right_aligned()),
Cell::from(
view.newest()
.map_or(String::new(), |item| item.describe(newest)),
),
Cell::from(
view.oldest()
.map_or(String::new(), |item| item.describe(oldest)),
),
])
});
Table::new(
rows,
[
Constraint::Length(STAGE_COLUMNS[0]),
Constraint::Length(STAGE_COLUMNS[1]),
Constraint::Length(STAGE_COLUMNS[2]),
Constraint::Length(newest),
Constraint::Fill(1),
],
)
.header(
Row::new([
Cell::from("stage"),
Cell::from(Line::from("busy").right_aligned()),
Cell::from(Line::from("waiting").right_aligned()),
Cell::from("newest"),
Cell::from("oldest"),
])
.style(dim),
)
.block(Block::default().borders(Borders::ALL).title("stages"))
}
/// The last `width` samples of a history: as much of it as a pane that
/// wide can draw.
fn window(history: &VecDeque<u64>, width: u16) -> Vec<u64> {
history
.iter()
.skip(history.len().saturating_sub(width as usize))
.copied()
.collect()
}
/// A queue's recent depth as one line of block characters, newest at the
/// right, scaled to the tallest sample shown.
///
/// A sample of zero is left blank rather than drawn at the lowest block,
/// so that a queue which ran dry is plainly empty rather than merely
/// nearly so.
fn spark(samples: &[u64], width: u16) -> String {
const BLOCKS: [char; 8] = ['▁', '▂', '▃', '▄', '▅', '▆', '▇', '█'];
let peak = samples.iter().copied().max().unwrap_or(0);
if peak == 0 {
return String::new();
}
// History grows leftward from the newest sample, so a history too
// short to fill the pane starts part-way across it
let mut line = " ".repeat((width as usize).saturating_sub(samples.len()));
for sample in samples {
line.push(match *sample {
0 => ' ',
value => BLOCKS[(value * 7 / peak) as usize],
});
}
line
}
/// The backlog pane: how deep each stage's queue has been over the last
/// few minutes.
///
/// The stages table says how much is waiting now; what it cannot say is
/// which way that number is going, which is the difference between a run
/// that is keeping up and one that is falling behind. Each stage is drawn
/// to its own scale — these queues differ by orders of magnitude — so the
/// peak beside it is what says how deep the tallest column is.
fn backlog_table(dashboard: &Dashboard, width: u16) -> Table<'_> {
let dim = Style::default().add_modifier(Modifier::DIM);
let room = width
.saturating_sub(2)
.saturating_sub(STAGE_COLUMNS[0] + STAGE_COLUMNS[2] + 2);
let rows = dashboard.stages.iter().map(move |view| {
let samples = window(&view.history, room);
let peak = samples.iter().copied().max().unwrap_or(0);
Row::new([
Cell::from(view.stage.label()),
Cell::from(spark(&samples, room)),
Cell::from(Line::from(count(peak)).right_aligned()),
])
});
Table::new(
rows,
[
Constraint::Length(STAGE_COLUMNS[0]),
Constraint::Fill(1),
Constraint::Length(STAGE_COLUMNS[2]),
],
)
.header(
Row::new([
Cell::from("stage"),
Cell::from("backlog"),
Cell::from(Line::from("peak").right_aligned()),
])
.style(dim),
)
.block(
Block::default()
.borders(Borders::ALL)
.title(backlog_span(dashboard, room)),
)
}
/// The backlog pane's title, which is where its unlabelled horizontal
/// axis gets its scale: the run time between the oldest sample drawn and
/// the newest, which grows as history accumulates and then holds at
/// whatever the pane's width covers.
fn backlog_span(dashboard: &Dashboard, room: u16) -> String {
let drawn = dashboard
.stages
.iter()
.map(|view| view.history.len())
.max()
.unwrap_or(0)
.min(room as usize);
// A single column spans nothing yet, and has no scale to give
match drawn {
0 | 1 => "backlog".to_string(),
columns => format!(
"backlog, last {}",
human_duration(BACKLOG_SAMPLE * (columns as u32 - 1))
),
}
}
/// What has yet to be visited: the paths each stage will take up next,
/// nearest first. A backup names the directories still to be walked
/// into; a restore names the entries still to be put back.
///
/// Most of a pipeline's backlog is deliberately absent. Files the scan
/// has already found, chunks in flight to the backend, and the record
/// of what is already stored are all work the run has in hand — naming
/// them would bury the frontier under thousands of lines that say
/// nothing about where the run is going. An empty pane means the stages
/// that do name their queues have nothing waiting.
///
/// The stages come in pipeline order, and a pane too short for all of
/// them lets them take turns claiming a line, so that a stage with a
/// short queue is not buried under one with a long one. The last line
/// accounts for everything waiting behind what is named — the whole
/// depth of those queues, not merely the part the pipeline sends by
/// name.
fn pending_lines(dashboard: &Dashboard, area: ratatui::layout::Rect) -> Vec<ListItem<'static>> {
let height = area.height.saturating_sub(2) as usize; // borders
let width = area.width.saturating_sub(2);
if height == 0 {
return Vec::new();
}
let queues: Vec<(&'static str, &[String])> = dashboard
.stages
.iter()
.filter_map(|view| {
view.upcoming
.as_deref()
.map(|paths| (view.stage.label(), paths))
})
.collect();
let waiting: u64 = dashboard
.stages
.iter()
.filter(|view| view.upcoming.is_some())
.map(|view| view.waiting)
.sum();
let named: usize = queues.iter().map(|(_, paths)| paths.len()).sum();
// Room for every name, or for all but the last line, which says how
// much is waiting behind them — except in a pane one line tall,
// where naming one path is worth more than counting the rest
let room = if waiting as usize > height && height > 1 {
named.min(height - 1)
} else {
named.min(height)
};
let mut share = vec![0usize; queues.len()];
let mut given = 0;
while given < room {
for (queue, (_, paths)) in queues.iter().enumerate() {
if given == room {
break;
}
if share[queue] < paths.len() {
share[queue] += 1;
given += 1;
}
}
}
let mut lines: Vec<ListItem<'static>> = Vec::with_capacity(height);
for (queue, (label, paths)) in queues.iter().enumerate() {
for path in &paths[..share[queue]] {
lines.push(ListItem::new(format!(
"{label:<column$} {}",
fit_path(path, width.saturating_sub(STAGE_COLUMNS[0] + 1)),
column = STAGE_COLUMNS[0] as usize
)));
}
}
if waiting > given as u64 && lines.len() < height {
lines.push(ListItem::new(format!(
"… and {} more",
count(waiting - given as u64)
)));
}
lines
}
/// Runs the full-screen dashboard until the pipeline finishes, returning
/// a final summary line. Pressing `s`, `q`, Esc, or Ctrl-C requests
/// suspension.
pub fn dashboard(
title: &str,
events: Receiver<Event>,
suspend: Arc<AtomicBool>,
) -> std::io::Result<String> {
let mut terminal = ratatui::init();
let result = dashboard_loop(title, &events, &suspend, &mut terminal);
ratatui::restore();
result
}
fn dashboard_loop(
title: &str,
events: &Receiver<Event>,
suspend: &AtomicBool,
terminal: &mut ratatui::DefaultTerminal,
) -> std::io::Result<String> {
let mut dashboard = Dashboard::default();
let mut finished = false;
while !finished {
loop {
match events.try_recv() {
Ok(event) => dashboard.apply(event),
Err(crossbeam_channel::TryRecvError::Empty) => break,
Err(crossbeam_channel::TryRecvError::Disconnected) => {
finished = true;
break;
}
}
}
let suspending = suspend.load(Ordering::Acquire);
terminal.draw(|frame| draw(frame, title, &dashboard, suspending))?;
if crossterm::event::poll(Duration::from_millis(100))?
&& let TermEvent::Key(key) = crossterm::event::read()?
{
let ctrl_c =
key.code == KeyCode::Char('c') && key.modifiers.contains(KeyModifiers::CONTROL);
if matches!(key.code, KeyCode::Char('s' | 'q') | KeyCode::Esc) || ctrl_c {
suspend.store(true, Ordering::Release);
}
}
}
Ok(dashboard.summary())
}
/// Draws one frame: the run's totals, its stages, and three panes — what
/// has been finished, how the queues ahead of it are trending, and what
/// its workers have in their hands.
fn draw(frame: &mut Frame, title: &str, dashboard: &Dashboard, suspending: bool) {
// Both stage tables are exactly as tall as they need to be
// — borders, a header, and one row per reported stage — so
// the activity log and the pending pane get what is left.
let stages_height = match dashboard.stages.len() {
0 => 0,
rows => rows as u16 + 3,
};
let [header, counters, stages, panes, skipped, footer] = Layout::vertical([
Constraint::Length(1),
Constraint::Length(4),
Constraint::Length(stages_height),
Constraint::Fill(1),
Constraint::Length(1),
Constraint::Length(1),
])
.areas(frame.area());
// What has happened on the left; what is happening and what
// is coming on the right
let [recent, current] =
Layout::horizontal([Constraint::Fill(1), Constraint::Fill(1)]).areas(panes);
// The backlog pane wants to be as tall as the stages table, and is
// what gives way when the window is too short to leave the pending
// pane anything worth showing
let [backlog, pending] =
Layout::vertical([Constraint::Max(stages_height), Constraint::Min(5)]).areas(current);
let dim = Style::default().add_modifier(Modifier::DIM);
frame.render_widget(
Paragraph::new(Line::from(title.to_string()))
.style(Style::default().add_modifier(Modifier::BOLD)),
header,
);
frame.render_widget(
Paragraph::new(format!(
"{}\nstatus: {}",
dashboard.summary(),
dashboard.status(suspending)
))
.block(Block::default().borders(Borders::ALL).title("progress")),
counters,
);
frame.render_widget(stages_table(dashboard, stages.width), stages);
frame.render_widget(
List::new(visible_tail(&dashboard.recent, recent))
.block(Block::default().borders(Borders::ALL).title("activity")),
recent,
);
frame.render_widget(backlog_table(dashboard, backlog.width), backlog);
frame.render_widget(
List::new(pending_lines(dashboard, pending))
.block(Block::default().borders(Borders::ALL).title("pending")),
pending,
);
frame.render_widget(Paragraph::new(dashboard.skip_line()).style(dim), skipped);
frame.render_widget(
Paragraph::new("s / q / Esc / Ctrl-C: suspend (resume with the same command)"),
footer,
);
}
/// Line-oriented reporter: skips are printed as they happen, progress
/// every couple of seconds, and the final summary at the end.
pub fn plain(events: Receiver<Event>) -> String {
let mut dashboard = Dashboard::default();
let mut last_progress = Instant::now();
let stderr = std::io::stderr();
for event in events {
if let Event::Skipped { path, reason } = &event {
let _ = writeln!(
stderr.lock(),
"skipped {} ({})",
path.display(),
describe_skip(reason)
);
}
match event {
Event::Checkpointing => {
let _ = writeln!(
stderr.lock(),
"checkpointing: making what is stored so far restorable..."
);
}
Event::Finalizing => {
let _ = writeln!(stderr.lock(), "finalizing...");
}
Event::FinishingEarly(reason) => {
let _ = writeln!(
stderr.lock(),
"finishing early: {}",
describe_early_finish(reason)
);
}
Event::LimitNotApplied(limit) => {
let _ = writeln!(
stderr.lock(),
"warning: {}",
describe_unapplied_limit(limit)
);
}
_ => {}
}
dashboard.apply(event);
if last_progress.elapsed() >= Duration::from_secs(2) {
let mut line = format!("{} · {}\n", dashboard.summary(), dashboard.status(false));
// The stages, indented under the totals they belong to
if let Some(stages) = dashboard.stage_line() {
line.push_str(" ");
line.push_str(&stages);
line.push('\n');
}
let _ = write!(stderr.lock(), "{line}");
last_progress = Instant::now();
}
}
dashboard.summary()
}
#[cfg(test)]
mod tests {
use super::*;
use beeping::pipeline::{Progress, StageBacklog};
use std::path::PathBuf;
fn lines(count: usize) -> VecDeque<Activity> {
(0..count)
.map(|n| Activity::Note(format!("line-{n}")))
.collect()
}
fn texts(items: &[ListItem<'_>]) -> Vec<String> {
// ListItem has no accessor; rebuild via Text conversion
items.iter().map(|item| format!("{:?}", item)).collect()
}
/// The paths the two columns would name, for comparing at a glance.
fn named(view: &StageView) -> (Option<&str>, Option<&str>) {
(
view.newest().map(|item| item.path.as_str()),
view.oldest().map(|item| item.path.as_str()),
)
}
/// A stage's backlog with nothing named: a queue of something other
/// than paths waiting for a worker.
fn backlog(stage: Stage, waiting: u64) -> StageBacklog {
StageBacklog {
stage,
waiting,
upcoming: None,
}
}
/// A stage's backlog with the head of its queue named.
fn queued(stage: Stage, waiting: u64, upcoming: &[&str]) -> StageBacklog {
StageBacklog {
stage,
waiting,
upcoming: Some(upcoming.iter().map(PathBuf::from).collect()),
}
}
fn began(stage: Stage, worker: usize, path: &str) -> Event {
Event::Began {
stage,
worker,
path: PathBuf::from(path),
}
}
#[test]
fn tall_panes_fill_with_the_newest_lines() {
let area = ratatui::layout::Rect::new(0, 0, 80, 22); // 20 inner rows
let history = lines(50);
let items = visible_tail(&history, area);
assert_eq!(items.len(), 20, "the pane's full height is used");
let rendered = texts(&items);
assert!(
rendered.first().unwrap().contains("line-30"),
"oldest visible"
);
assert!(rendered.last().unwrap().contains("line-49"), "newest last");
}
#[test]
fn short_histories_show_everything() {
let area = ratatui::layout::Rect::new(0, 0, 80, 22);
let history = lines(5);
let items = visible_tail(&history, area);
assert_eq!(items.len(), 5);
}
#[test]
fn degenerate_panes_render_nothing() {
let history = lines(10);
for height in [0, 1, 2] {
let area = ratatui::layout::Rect::new(0, 0, 80, height);
assert!(visible_tail(&history, area).len() <= height as usize);
}
}
/// The activity log's own pane is half the width it used to have, so
/// its lines are fitted like the table's are: the verb whole, and as
/// much of the path's tail as the rest of the line holds.
#[test]
fn activity_lines_keep_the_verb_and_the_end_of_the_path() {
let stored = Activity::Item {
verb: "stored",
path: "/home/djarb/models/big.safetensors".to_string(),
};
assert_eq!(
stored.render(60),
"stored /home/djarb/models/big.safetensors"
);
assert_eq!(stored.render(24), "stored …ig.safetensors");
// Prose is not a path, and front-truncating it would only hide
// the part that says what it is about
let note = Activity::Note("resumed: totals include previous sessions".to_string());
assert_eq!(note.render(20), "resumed: totals include previous sessions");
}
/// A run that stops short says why in the log, where it stays put:
/// nothing else is happening by then, and finalizing a large
/// manifest can take long enough for the question to come up.
#[test]
fn an_early_finish_says_why_in_the_activity_log() {
let mut dashboard = Dashboard::default();
dashboard.apply(Event::FinishingEarly(EarlyFinish::StoreNearlyFull {
free: 3 << 30,
floor: 20 << 30,
}));
dashboard.apply(Event::Finalizing);
let note = dashboard.recent.back().unwrap().render(80);
assert_eq!(
note,
"finishing early: the store has 3.0 GiB left, 17.0 GiB below the \
20.0 GiB this backup was told to leave free"
);
// and the run carries on to publish what it has
assert!(dashboard.status(false).starts_with("finalizing"));
// The other limit on the store reads from the other end
let mut dashboard = Dashboard::default();
dashboard.apply(Event::FinishingEarly(EarlyFinish::StoreAtMaximum {
used: 500 << 30,
limit: 500 << 30,
}));
assert_eq!(
dashboard.recent.back().unwrap().render(80),
"finishing early: the repository holds 500.0 GiB, at the 500.0 \
GiB this backup was told to keep it under"
);
}
/// A limit that cannot be applied is worth a line of its own: the
/// run looks exactly like one being held to it otherwise.
#[test]
fn a_limit_that_cannot_be_applied_says_so() {
let mut dashboard = Dashboard::default();
dashboard.apply(Event::LimitNotApplied(StoreLimit::MaximumStoreSize));
assert_eq!(
dashboard.recent.back().unwrap().render(80),
"the store cannot say how much it is holding, so the repository \
is not being kept to a size"
);
}
#[test]
fn counts_are_grouped_in_threes() {
assert_eq!(count(0), "0");
assert_eq!(count(999), "999");
assert_eq!(count(1000), "1,000");
assert_eq!(count(12345), "12,345");
assert_eq!(count(1234567890), "1,234,567,890");
}
#[test]
fn stages_appear_in_the_order_the_pipeline_mentions_them() {
let mut dashboard = Dashboard::default();
for (stage, threads) in [(Stage::Scan, 2), (Stage::Chunk, 4), (Stage::Upload, 3)] {
dashboard.apply(Event::Staffing { stage, threads });
}
dashboard.apply(Event::Backlog(vec![backlog(Stage::Record, 17)]));
let order: Vec<Stage> = dashboard.stages.iter().map(|view| view.stage).collect();
assert_eq!(
order,
[Stage::Scan, Stage::Chunk, Stage::Upload, Stage::Record]
);
// A stage nobody staffs is worked by finalization
assert_eq!(dashboard.stages[3].occupancy(), "—");
assert_eq!(dashboard.stages[3].waiting, 17);
}
#[test]
fn occupancy_follows_workers_in_and_out_of_work() {
let mut dashboard = Dashboard::default();
dashboard.apply(Event::Staffing {
stage: Stage::Chunk,
threads: 3,
});
dashboard.apply(began(Stage::Chunk, 0, "a.dat"));
dashboard.apply(began(Stage::Chunk, 1, "b.dat"));
assert_eq!(dashboard.stages[0].occupancy(), "2/3");
// Taking the next item replaces the worker's previous one
dashboard.apply(began(Stage::Chunk, 1, "c.dat"));
assert_eq!(dashboard.stages[0].occupancy(), "2/3");
dashboard.apply(Event::Idle {
stage: Stage::Chunk,
worker: 0,
});
assert_eq!(dashboard.stages[0].occupancy(), "1/3");
dashboard.apply(Event::Idle {
stage: Stage::Chunk,
worker: 1,
});
assert_eq!(dashboard.stages[0].occupancy(), "0/3");
assert_eq!(named(&dashboard.stages[0]), (None, None));
}
#[test]
fn a_stage_names_both_where_it_is_and_what_holds_it_up() {
let mut dashboard = Dashboard::default();
dashboard.apply(began(Stage::Chunk, 0, "slow.iso"));
dashboard.apply(began(Stage::Chunk, 1, "quick-a.txt"));
dashboard.apply(began(Stage::Chunk, 1, "quick-b.txt"));
assert_eq!(
named(&dashboard.stages[0]),
(Some("quick-b.txt"), Some("slow.iso")),
"where it has got to, and what is holding it up"
);
// A finished item is named by neither: with the fast worker
// idle, one item is in hand and both columns name it
dashboard.apply(Event::Idle {
stage: Stage::Chunk,
worker: 1,
});
assert_eq!(
named(&dashboard.stages[0]),
(Some("slow.iso"), Some("slow.iso"))
);
dashboard.apply(Event::Idle {
stage: Stage::Chunk,
worker: 0,
});
assert_eq!(
named(&dashboard.stages[0]),
(None, None),
"an idle stage names nothing rather than a stale item"
);
}
#[test]
fn progress_belongs_to_the_item_in_hand() {
let mut dashboard = Dashboard::default();
dashboard.apply(began(Stage::Chunk, 0, "big.iso"));
dashboard.apply(Event::Progressed {
stage: Stage::Chunk,
worker: 0,
done: 43,
total: 100,
});
let item = dashboard.stages[0].oldest().unwrap();
assert_eq!(item.describe(40), " 43% big.iso");
// The next item starts afresh rather than inheriting a
// percentage from the last one
dashboard.apply(began(Stage::Chunk, 0, "next.iso"));
assert_eq!(
dashboard.stages[0].oldest().unwrap().describe(40),
"next.iso"
);
// A percentage keeps its width; only the path gives way
dashboard.apply(Event::Progressed {
stage: Stage::Chunk,
worker: 0,
done: 7,
total: 1000,
});
let cell = dashboard.stages[0].oldest().unwrap().describe(10);
assert_eq!(cell, " 0% ….iso");
assert_eq!(cell.chars().count(), 10, "the column is not overrun");
// Progress for a worker holding nothing is simply dropped
dashboard.apply(Event::Idle {
stage: Stage::Chunk,
worker: 0,
});
dashboard.apply(Event::Progressed {
stage: Stage::Chunk,
worker: 0,
done: 1,
total: 2,
});
assert_eq!(named(&dashboard.stages[0]), (None, None));
}
#[test]
fn paths_are_cut_at_the_front_where_they_are_all_alike() {
assert_eq!(fit_path("sub/file.dat", 20), "sub/file.dat");
assert_eq!(fit_path("sub/file.dat", 12), "sub/file.dat");
assert_eq!(fit_path("sub/file.dat", 8), "…ile.dat");
assert_eq!(fit_path("sub/file.dat", 1), "…");
assert_eq!(fit_path("sub/file.dat", 0), "");
// Cut on char boundaries, not byte offsets
let accented = "café/ünïcøde.dat";
assert_eq!(accented.chars().count(), 16);
assert_eq!(fit_path(accented, 8), "…øde.dat");
assert_eq!(fit_path(accented, 8).chars().count(), 8);
}
#[test]
fn the_two_path_columns_share_what_the_narrow_ones_leave() {
let (newest, oldest) = path_columns(110);
assert_eq!(newest + oldest, 110 - 2 - 24 - 4);
assert!(oldest >= newest, "odd remainders go to the last column");
// A cramped table asks for no path width rather than underflowing
assert_eq!(path_columns(20), (0, 0));
}
#[test]
fn only_the_most_recent_skip_is_kept() {
let mut dashboard = Dashboard::default();
for name in ["one", "two", "three"] {
dashboard.apply(Event::Skipped {
path: PathBuf::from(name),
reason: SkipReason::Excluded,
});
}
dashboard.apply(Event::Totals(Session {
progress: Progress {
skipped: 3,
..Progress::default()
},
elapsed: Duration::from_secs(1),
}));
assert_eq!(dashboard.skip_line(), "skipped 3 · last: three (excluded)");
// A resumed run inherits the count without inheriting a path
let mut resumed = Dashboard::default();
assert_eq!(resumed.skip_line(), "skipped none");
resumed.apply(Event::Resumed(Session {
progress: Progress {
skipped: 1200,
..Progress::default()
},
elapsed: Duration::from_secs(30),
}));
assert_eq!(resumed.skip_line(), "skipped 1,200");
assert!(
dashboard.recent.is_empty(),
"skips do not crowd the activity log"
);
}
#[test]
fn the_line_reporter_says_what_every_stage_is_doing() {
let mut dashboard = Dashboard::default();
assert_eq!(dashboard.stage_line(), None, "before the pipeline speaks");
dashboard.apply(Event::Staffing {
stage: Stage::Chunk,
threads: 16,
});
dashboard.apply(began(Stage::Chunk, 4, "big.iso"));
dashboard.apply(Event::Backlog(vec![
backlog(Stage::Chunk, 0),
backlog(Stage::Record, 1489),
]));
assert_eq!(
dashboard.stage_line().unwrap(),
"chunk 1/16 busy, 0 waiting (big.iso) · record 1,489 waiting"
);
}
/// Each queue is drawn against its own peak, since a scan backlog of
/// half a million and an upload backlog of eight share a pane.
#[test]
fn a_backlog_is_drawn_against_its_own_peak() {
assert_eq!(spark(&[0, 1, 5, 10], 4), " ▁▄█");
assert_eq!(spark(&[0, 100, 500, 1000], 4), " ▁▄█");
// A queue that has been empty throughout has nothing to say
assert_eq!(spark(&[0, 0, 0], 8), "");
// History grows leftward from the newest sample
assert_eq!(spark(&[2, 1], 6), " █▄");
// A pane narrower than the history shows the recent end of it,
// which is what the caller's window hands over
let history: VecDeque<u64> = (1..=10).collect();
assert_eq!(window(&history, 3), vec![8, 9, 10]);
assert_eq!(window(&history, 40), (1..=10).collect::<Vec<u64>>());
}
/// The display is fed several times a second; the backlog history is
/// not, or the pane would hold a few seconds of noise.
#[test]
fn backlog_history_is_sampled_on_the_pipelines_clock() {
let mut dashboard = Dashboard::default();
let report = |dashboard: &mut Dashboard, millis: u64, waiting: u64| {
dashboard.apply(Event::Totals(Session {
progress: Progress::default(),
elapsed: Duration::from_millis(millis),
}));
dashboard.apply(Event::Backlog(vec![backlog(Stage::Chunk, waiting)]));
};
for tick in 0..=50 {
report(&mut dashboard, tick * 200, tick);
}
let view = &dashboard.stages[0];
// Ten seconds of reports, two seconds to the sample
assert_eq!(view.history.len(), 6);
assert_eq!(view.waiting, 50, "the depth itself is always current");
assert_eq!(*view.history.back().unwrap(), 50);
// A second pipeline reporting into the same display starts the
// history over rather than continuing the first one's
report(&mut dashboard, 0, 7);
let view = &dashboard.stages[0];
assert_eq!(view.history.len(), 1);
assert_eq!(*view.history.back().unwrap(), 7);
}
#[test]
fn transfers_and_finished_content_are_measured_separately() {
let mut dashboard = Dashboard::default();
let totals = |elapsed, stored_bytes| {
Event::Totals(Session {
progress: Progress {
stored_bytes,
..Progress::default()
},
elapsed: Duration::from_secs(elapsed),
})
};
// One huge file: chunks keep flowing, but nothing finishes
dashboard.apply(totals(0, 0));
for second in 1..=4 {
dashboard.apply(Event::Transferred { bytes: 3000 });
dashboard.apply(totals(second, 0));
}
assert_eq!(dashboard.content_rate.per_second(), Some(0));
assert_eq!(dashboard.transfer_rate.per_second(), Some(3000));
let status = dashboard.status(false);
assert!(status.contains("repository 2.9 KiB/s"), "{status}");
}
#[test]
fn throughput_is_measured_over_the_recent_window() {
let mut rate = Rate::default();
assert_eq!(rate.per_second(), None, "nothing sampled yet");
rate.observe(Duration::from_secs(0), 0);
assert_eq!(rate.per_second(), None, "one sample is not a rate");
for second in 1..=4 {
rate.observe(Duration::from_secs(second), second * 1000);
}
assert_eq!(rate.per_second(), Some(1000));
// Ten seconds of double speed pushes the old samples out of the
// window, leaving the current rate rather than the average
for second in 5..=14 {
rate.observe(Duration::from_secs(second), 4000 + (second - 4) * 2000);
}
assert_eq!(rate.per_second(), Some(2000));
}
#[test]
fn a_stalled_run_reports_no_throughput_rather_than_a_stale_one() {
let mut rate = Rate::default();
for second in 0..=3 {
rate.observe(Duration::from_secs(second), second * 1000);
}
// The bytes stop moving; the window fills with flat samples
for second in 4..=20 {
rate.observe(Duration::from_secs(second), 3000);
}
assert_eq!(rate.per_second(), Some(0));
}
/// The pending pane names what each queue will hand over next, and
/// accounts for everything waiting behind the names it has room for.
///
/// The dashboard serves both pipelines and knows neither, so what it
/// draws depends only on which stages name their queues — here, a
/// restore's two, beside a stage whose backlog is not a queue of
/// paths at all.
#[test]
fn the_pending_pane_names_what_the_queues_will_hand_out_next() {
let mut dashboard = Dashboard::default();
assert!(
pending_lines(&dashboard, ratatui::layout::Rect::new(0, 0, 40, 8)).is_empty(),
"nothing queued yet"
);
dashboard.apply(Event::Backlog(vec![
queued(Stage::Write, 2, &["/tree/src", "/tree/doc"]),
queued(
Stage::Apply,
400,
&["/tree/a.dat", "/tree/b.dat", "/tree/c.dat"],
),
backlog(Stage::Record, 9000),
]));
let pane = ratatui::layout::Rect::new(0, 0, 40, 8); // 6 inner rows
let rendered = texts(&pending_lines(&dashboard, pane));
assert_eq!(rendered.len(), 6);
assert!(rendered[0].contains("write"), "{:?}", rendered[0]);
assert!(rendered[0].contains("src"), "nearest first");
assert!(rendered[1].contains("doc"));
assert!(rendered[2].contains("apply"), "{:?}", rendered[2]);
assert!(rendered[2].contains("a.dat"), "then the next stage along");
assert!(
rendered[5].contains("and 397 more"),
"the rest of the two queues — and not the record queue, which \
holds what is already stored: {:?}",
rendered[5]
);
// Too short for all the names: every queue still gets a turn,
// and the last line grows to cover what is left out
let short = ratatui::layout::Rect::new(0, 0, 40, 4); // 2 inner rows
let rendered = texts(&pending_lines(&dashboard, short));
assert_eq!(rendered.len(), 2);
assert!(rendered[0].contains("src"), "{:?}", rendered[0]);
assert!(rendered[1].contains("and 401 more"), "{:?}", rendered[1]);
// One line is better spent naming something than counting
let sliver = ratatui::layout::Rect::new(0, 0, 40, 3); // 1 inner row
let rendered = texts(&pending_lines(&dashboard, sliver));
assert_eq!(rendered.len(), 1);
assert!(rendered[0].contains("src"), "{:?}", rendered[0]);
// A queue that drains takes its names off the pane with it
dashboard.apply(Event::Backlog(vec![
queued(Stage::Write, 0, &[]),
queued(Stage::Apply, 0, &[]),
]));
assert!(pending_lines(&dashboard, pane).is_empty());
}
/// The whole frame, as characters, for tests that care about how the
/// panes divide the screen rather than about one pane's contents.
fn rendered(dashboard: &Dashboard, width: u16, height: u16) -> Vec<String> {
let mut terminal =
ratatui::Terminal::new(ratatui::backend::TestBackend::new(width, height)).unwrap();
terminal
.draw(|frame| draw(frame, "beeping backup", dashboard, false))
.unwrap();
let buffer = terminal.backend().buffer().clone();
(0..height)
.map(|y| {
(0..width)
.map(|x| buffer[(x, y)].symbol())
.collect::<String>()
})
.collect()
}
/// The panes below the stages table divide the screen in two: what
/// has been finished on the left, and what the queues have been
/// doing and will do next on the right.
#[test]
fn the_lower_panes_split_the_width_between_history_and_what_is_ahead() {
let mut dashboard = Dashboard::default();
dashboard.apply(Event::Staffing {
stage: Stage::Scan,
threads: 2,
});
dashboard.apply(began(Stage::Scan, 0, "/tree/models"));
dashboard.apply(Event::Stored {
path: PathBuf::from("/tree/notes/a.txt"),
bytes: 12,
});
for tick in 0..=10 {
dashboard.apply(Event::Totals(Session {
progress: Progress {
scanned: 40 + tick,
stored_files: 300 + tick * 9,
stored_bytes: (12 + tick) * 1024 * 1024,
..Progress::default()
},
elapsed: Duration::from_secs(tick * 2),
}));
dashboard.apply(Event::Backlog(vec![queued(
Stage::Scan,
tick * 40,
&["/tree/notes", "/tree/models"],
)]));
}
let screen = rendered(&dashboard, 100, 20);
// A pane's title is where its top border opens, which is also
// the only place these words appear as titles
let titled = |name: &str| {
let opens = format!("┌{name}");
screen
.iter()
.position(|line| line.contains(&opens))
.unwrap_or_else(|| panic!("no {name} pane in\n{}", screen.join("\n")))
};
// activity and backlog share a row of titles; pending is under
// the backlog pane, which is as tall as the stages table
assert_eq!(titled("activity"), titled("backlog"));
assert!(titled("pending") > titled("backlog"));
let titles = &screen[titled("activity")];
assert!(
titles.contains("backlog, last 20s"),
"the pane says how much of the run it is drawing: {titles}"
);
let split = titles.split("┌backlog").next().unwrap().chars().count();
assert!((45..55).contains(&split), "{titles}");
// Each pane holds what belongs to it, and nothing has spilled
assert!(screen.iter().any(|line| line.contains("stored")));
assert!(
screen[titled("pending") + 1].contains("/tree/notes"),
"{}",
screen[titled("pending") + 1]
);
// Title, header, then the one stage this run has reported
assert!(
screen[titled("backlog") + 2].contains('█'),
"a queue that has been growing all run should show it: {}",
screen[titled("backlog") + 2]
);
}
#[test]
fn totals_come_from_the_pipeline_rather_than_being_re_derived() {
let mut dashboard = Dashboard::default();
// Per-item events feed the activity log, not the counters
dashboard.apply(Event::Stored {
path: PathBuf::from("a.dat"),
bytes: 10,
});
assert_eq!(dashboard.summary(), "no activity");
dashboard.apply(Event::Totals(Session {
progress: Progress {
scanned: 12,
stored_files: 3400,
stored_bytes: 5 * 1024 * 1024,
..Progress::default()
},
elapsed: Duration::from_secs(90),
}));
assert_eq!(
dashboard.summary(),
"12 directories scanned, 3,400 files stored (5.0 MiB)"
);
assert_eq!(
dashboard.status(false),
"running · elapsed 1m 30s",
"one sample in is too early for a throughput figure"
);
assert!(dashboard.status(true).starts_with("suspending"));
}
}