nxd_core/progress/

bulk.rs

use std::io::{IsTerminal, Write};

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum BulkState {
	#[default]
	Idle,
	BareCopying {
		count: usize,
	},
	Block {
		total: usize,
		count: usize,
		lines_seen: usize,
		bytes_seen: usize,
		size: Option<String>,
	},
}

pub struct TerminalBulkState {
	pub state: BulkState,
	pub active_counter_rendered: bool,
	pub suppressed_tail: Vec<String>,
	pub is_tty_override: Option<bool>,
}

impl Default for TerminalBulkState {
	fn default() -> Self {
		Self {
			state: BulkState::Idle,
			active_counter_rendered: false,
			suppressed_tail: Vec::new(),
			is_tty_override: None,
		}
	}
}

impl TerminalBulkState {
	pub fn is_tty(&self) -> bool {
		self.is_tty_override.unwrap_or_else(|| std::io::stdout().is_terminal())
	}

	pub fn clear_active_counter(&mut self) {
		let mut out = std::io::stdout();
		self.clear_active_counter_to_writer(&mut out);
	}

	pub fn clear_active_counter_to_writer<W: Write>(&mut self, writer: &mut W) {
		if self.active_counter_rendered {
			if self.is_tty() {
				let _ = write!(writer, "\r\x1b[K");
				let _ = writer.flush();
			}
			self.active_counter_rendered = false;
		}
	}

	pub fn flush_summary(&mut self) {
		let mut out = std::io::stdout();
		self.flush_summary_to_writer(&mut out);
	}

	pub fn flush_summary_to_writer<W: Write>(&mut self, writer: &mut W) {
		let summary_line = match &self.state {
			BulkState::BareCopying { count } if *count > 0 => Some(format!("copied {} paths", count)),
			BulkState::Block { count, total, size, .. } if *count > 0 => {
				if let Some(size_str) = size {
					Some(format!("copied {}/{} paths ({})", count, total, size_str))
				} else {
					Some(format!("copied {}/{} paths", count, total))
				}
			}
			_ => None,
		};

		if let Some(line) = summary_line {
			self.clear_active_counter_to_writer(writer);
			let formatted = crate::progress::stream::format_output_line(&line);
			let _ = writeln!(writer, "{}", formatted);
			let _ = writer.flush();
		}
		self.state = BulkState::Idle;
	}

	pub fn process_line(&mut self, line: &str) {
		let mut out = std::io::stdout();
		self.process_line_to_writer(line, false, &mut out);
	}

	pub fn process_line_to_writer<W: Write>(&mut self, line: &str, stderr: bool, writer: &mut W) {
		if self.suppressed_tail.len() >= 64 {
			self.suppressed_tail.remove(0);
		}
		self.suppressed_tail.push(line.to_string());

		if stderr || crate::config::get_runtime_options().verbose {
			self.clear_active_counter_to_writer(writer);
			self.flush_summary_to_writer(writer);
			let formatted = crate::progress::stream::format_output_line(line);
			let _ = writeln!(writer, "{}", formatted);
			let _ = writer.flush();
			return;
		}

		let is_tty = self.is_tty();

		if let Some((total, size)) = parse_block_header(line) {
			self.flush_summary_to_writer(writer);
			self.state =
				BulkState::Block { total, count: 0, lines_seen: 0, bytes_seen: line.len(), size };
			if is_tty {
				self.render_counter_in_place_to_writer(writer);
			}
			return;
		}

		if is_bare_copying_line(line) {
			match &mut self.state {
				BulkState::BareCopying { count } => {
					*count += 1;
				}
				// Nix lists a block's members once, then narrates the same paths
				// again as it transfers them. Counting both passes would report
				// more paths than the header promised.
				BulkState::Block { count, total, lines_seen, bytes_seen, .. } => {
					if *count < *total {
						*count += 1;
					}
					*lines_seen += 1;
					*bytes_seen += line.len();
				}
				BulkState::Idle => {
					self.state = BulkState::BareCopying { count: 1 };
				}
			}
			if is_tty {
				self.render_counter_in_place_to_writer(writer);
			}
			return;
		}

		let is_block_cont = is_block_continuation_line(line);
		let is_in_block = matches!(&self.state, BulkState::Block { .. });
		if is_block_cont && is_in_block {
			let BulkState::Block { count, total, lines_seen, bytes_seen, .. } = &mut self.state else {
				unreachable!()
			};
			if *count < *total {
				*count += 1;
			}
			*lines_seen += 1;
			*bytes_seen += line.len();

			if *lines_seen >= 4096 || *bytes_seen >= 256 * 1024 {
				self.flush_summary_to_writer(writer);
				self.clear_active_counter_to_writer(writer);
				let formatted = crate::progress::stream::format_output_line(line);
				let _ = writeln!(writer, "{}", formatted);
				let _ = writer.flush();
				return;
			}

			if is_tty {
				self.render_counter_in_place_to_writer(writer);
			}
			return;
		}

		self.flush_summary_to_writer(writer);
		self.clear_active_counter_to_writer(writer);
		let formatted = crate::progress::stream::format_output_line(line);
		let _ = writeln!(writer, "{}", formatted);
		let _ = writer.flush();
	}

	fn render_counter_in_place_to_writer<W: Write>(&mut self, writer: &mut W) {
		let line = match &self.state {
			BulkState::BareCopying { count } => {
				format!("copying {} paths…", count)
			}
			BulkState::Block { count, total, size, .. } => {
				if let Some(size_str) = size {
					format!("copying {}/{} paths… ({})", count, total, size_str)
				} else {
					format!("copying {}/{} paths…", count, total)
				}
			}
			BulkState::Idle => return,
		};

		let formatted = crate::progress::stream::format_output_line(&line);
		let _ = write!(writer, "\r{}", formatted);
		let _ = writer.flush();
		self.active_counter_rendered = true;
	}

	pub fn take_suppressed_tail(&mut self) -> Vec<String> {
		std::mem::take(&mut self.suppressed_tail)
	}
}

pub fn parse_block_header(line: &str) -> Option<(usize, Option<String>)> {
	let trimmed = line.trim();
	// Nix announces a transfer as "copying N paths..." or "copying N paths" and a
	// realisation as "these N paths will be fetched" or "these N derivations will be built";
	// both open a listing. The token after the integer must be exactly "paths" (optionally
	// followed by "..."); any other token (e.g. "files") is not a bulk header.
	if let Some(rest) = trimmed.strip_prefix("copying ") {
		let mut parts = rest.split_whitespace();
		if let (Some(count_str), Some(next_token)) = (parts.next(), parts.next())
			&& parts.next().is_none()
			&& let Ok(count) = count_str.parse::<usize>()
			&& next_token.strip_suffix("...").unwrap_or(next_token) == "paths"
		{
			return Some((count, None));
		}
	}
	if !trimmed.starts_with("these ") {
		return None;
	}
	if !(trimmed.contains(" paths will be fetched") || trimmed.contains(" derivations will be built"))
	{
		return None;
	}
	let after_these = trimmed.strip_prefix("these ")?;
	let count_str = after_these.split_whitespace().next()?;
	let count: usize = count_str.parse().ok()?;

	let size = trimmed.find('(').and_then(|start| {
		trimmed[start..].find(')').map(|end| trimmed[start + 1..start + end].to_string())
	});

	Some((count, size))
}

pub fn is_bare_copying_line(line: &str) -> bool {
	let trimmed = line.trim();
	(trimmed.starts_with("copying path '") || trimmed.starts_with("copying path "))
		&& (trimmed.contains(" from ")
			|| trimmed.contains(" to ")
			|| trimmed.contains("' from '")
			|| trimmed.contains("' to '"))
}

pub fn is_block_continuation_line(line: &str) -> bool {
	// Only a store-path entry continues a block. Indentation alone is not enough:
	// Nix indents error context too, and treating that as bulk would hide a real
	// failure reported in the middle of a fetch listing.
	let trimmed = line.trim();
	trimmed.starts_with("/nix/store/")
}

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

	#[test]
	fn a_copying_header_opens_a_block_instead_of_printing_beside_its_counter() {
		assert_eq!(parse_block_header("copying 2 paths..."), Some((2, None)));
		assert_eq!(parse_block_header("copying 2 paths"), Some((2, None)));
		assert_eq!(parse_block_header("copying 12 paths"), Some((12, None)));
		assert_eq!(parse_block_header("copying 12 paths..."), Some((12, None)));
		assert_eq!(parse_block_header("copying 0 paths..."), Some((0, None)));
		// "files" is not "paths" — must be rejected so the line stays visible.
		assert_eq!(parse_block_header("copying 12 files to /var"), None);
		// A per-path line is not a header; it increments the counter.
		assert_eq!(parse_block_header("copying path '/nix/store/x' from 'https://c'"), None);
		assert!(is_bare_copying_line("copying path '/nix/store/x' from 'https://c'"));
	}

	#[test]
	fn rejected_copying_files_line_and_ordinary_lines_remain_visible() {
		// "copying N files" must not be classified as bulk — it must appear in output.
		let mut bulk = TerminalBulkState { is_tty_override: Some(false), ..Default::default() };
		let mut out = Vec::new();
		bulk.process_line_to_writer("copying 12 files to /var", false, &mut out);
		let rendered = String::from_utf8(out).unwrap();
		assert!(
			rendered.contains("copying 12 files to /var"),
			"rejected line must remain visible: {rendered:?}"
		);

		// Ordinary non-bulk output must remain visible.
		let mut bulk2 = TerminalBulkState { is_tty_override: Some(false), ..Default::default() };
		let mut out2 = Vec::new();
		bulk2.process_line_to_writer("Activating configuration...", false, &mut out2);
		let rendered2 = String::from_utf8(out2).unwrap();
		assert!(
			rendered2.contains("Activating configuration..."),
			"ordinary non-bulk line must remain visible: {rendered2:?}"
		);
	}

	#[test]
	fn indented_error_inside_a_block_is_never_suppressed() {
		// Nix indents error context. If a block swallows every indented line, a real
		// failure reported mid-block disappears from the terminal.
		assert!(
			!is_block_continuation_line("       error: build of '/nix/store/x.drv' failed"),
			"indented error must not be treated as block continuation"
		);
	}

	#[test]
	fn test_logp15_bare_copying_stream_pinned_fixture() {
		// Pinned real Nix output from `nix copy --to file:///tmp/bulk-fixture`
		let lines = [
			"copying path '/nix/store/jspv3c5l2zx4kiwzhq0zgxcwp34cqifz-libiconv-115.100.1' to 'file:///tmp/bulk-fixture'...",
			"copying path '/nix/store/v4f0jj9sz97ckskvacf40llz4nfr19jf-hello-2.12.3' to 'file:///tmp/bulk-fixture'...",
			"setting up pam...",
		];

		let mut bulk = TerminalBulkState { is_tty_override: Some(true), ..Default::default() };
		let mut out = Vec::new();

		for line in lines {
			bulk.process_line_to_writer(line, false, &mut out);
		}

		let rendered = String::from_utf8(out).unwrap();
		assert!(rendered.contains("\rcopying 1 paths…"));
		assert!(rendered.contains("\rcopying 2 paths…"));
		assert!(rendered.contains("copied 2 paths\n"));
		assert!(rendered.contains("setting up pam...\n"));
	}

	#[test]
	fn test_logp15_these_n_paths_fetched_block_pinned_fixture() {
		// Pinned real Nix fetch block output
		let lines = [
			"these 2 paths will be fetched (0.1 MiB download, 0.4 MiB unpacked):",
			"  /nix/store/jspv3c5l2zx4kiwzhq0zgxcwp34cqifz-libiconv-115.100.1",
			"  /nix/store/v4f0jj9sz97ckskvacf40llz4nfr19jf-hello-2.12.3",
			"Activating home-manager...",
		];

		let mut bulk = TerminalBulkState { is_tty_override: Some(true), ..Default::default() };
		let mut out = Vec::new();

		for line in lines {
			bulk.process_line_to_writer(line, false, &mut out);
		}

		let rendered = String::from_utf8(out).unwrap();
		assert!(rendered.contains("\rcopying 0/2 paths… (0.1 MiB download, 0.4 MiB unpacked)"));
		assert!(rendered.contains("\rcopying 1/2 paths… (0.1 MiB download, 0.4 MiB unpacked)"));
		assert!(rendered.contains("\rcopying 2/2 paths… (0.1 MiB download, 0.4 MiB unpacked)"));
		assert!(rendered.contains("copied 2/2 paths (0.1 MiB download, 0.4 MiB unpacked)\n"));
		assert!(rendered.contains("Activating home-manager...\n"));
	}

	#[test]
	fn a_copying_header_counts_its_own_paths_once() {
		// Pinned from a real `nix copy --to ssh://…` run: the header promises N,
		// then every path is narrated. The counter must track the header, not
		// restart or overshoot it.
		let lines = [
			"copying 3 paths...",
			"copying path '/nix/store/aaa-source' to 'ssh://deploy@host'...",
			"copying path '/nix/store/bbb-source' to 'ssh://deploy@host'...",
			"copying path '/nix/store/ccc-source' to 'ssh://deploy@host'...",
			"building '/nix/store/x.drv'...",
		];

		let mut bulk = TerminalBulkState { is_tty_override: Some(true), ..Default::default() };
		let mut out = Vec::new();

		for line in lines {
			bulk.process_line_to_writer(line, false, &mut out);
		}

		let rendered = String::from_utf8(out).unwrap();
		assert!(rendered.contains("\rcopying 1/3 paths…"));
		assert!(rendered.contains("\rcopying 3/3 paths…"));
		assert!(!rendered.contains("copying 1 paths…"), "must not restart a bare tally mid-block");
		assert!(rendered.contains("copied 3/3 paths\n"));
	}

	#[test]
	fn a_block_never_counts_past_the_total_it_announced() {
		// Nix lists a fetch block's members, then narrates the same paths again
		// while transferring them. Both passes reaching the counter would report
		// more paths than the header promised.
		let lines = [
			"these 2 paths will be fetched (0.1 MiB):",
			"  /nix/store/aaa-hello",
			"  /nix/store/bbb-world",
			"copying path '/nix/store/aaa-hello' from 'https://cache.nixos.org'...",
			"copying path '/nix/store/bbb-world' from 'https://cache.nixos.org'...",
			"setting up pam...",
		];

		let mut bulk = TerminalBulkState { is_tty_override: Some(true), ..Default::default() };
		let mut out = Vec::new();

		for line in lines {
			bulk.process_line_to_writer(line, false, &mut out);
		}

		let rendered = String::from_utf8(out).unwrap();
		assert!(rendered.contains("copied 2/2 paths (0.1 MiB)\n"));
		assert!(!rendered.contains("3/2"), "counter must not exceed the announced total");
		assert!(!rendered.contains("4/2"), "counter must not exceed the announced total");
	}

	#[test]
	fn test_logp15_unrecognised_line_passes_through() {
		let line = "Activating home-manager for user lamt...";
		let mut bulk = TerminalBulkState { is_tty_override: Some(true), ..Default::default() };
		let mut out = Vec::new();

		bulk.process_line_to_writer(line, false, &mut out);

		let rendered = String::from_utf8(out).unwrap();
		// Unrecognised line must pass through immediately without carriage returns
		assert_eq!(rendered, "Activating home-manager for user lamt...\n");
	}

	#[test]
	fn test_logp15_stderr_never_suppressed() {
		let line = "warning: Git tree '/Users/lamt/lamt-nixconfig' is dirty";
		let mut bulk = TerminalBulkState { is_tty_override: Some(true), ..Default::default() };
		let mut out = Vec::new();

		// Start a bulk block
		bulk.process_line_to_writer("copying path '/nix/store/abc' to 'store'...", false, &mut out);
		// Stderr line arrives
		bulk.process_line_to_writer(line, true, &mut out);

		let rendered = String::from_utf8(out).unwrap();
		assert!(rendered.contains("warning: Git tree '/Users/lamt/lamt-nixconfig' is dirty\n"));
	}

	#[test]
	fn test_logp15_non_tty_sink_receives_no_carriage_returns() {
		let lines = [
			"these 2 paths will be fetched (0.1 MiB):",
			"  /nix/store/jspv3c5l2zx4kiwzhq0zgxcwp34cqifz-libiconv-115.100.1",
			"  /nix/store/v4f0jj9sz97ckskvacf40llz4nfr19jf-hello-2.12.3",
			"setting up pam...",
		];

		let mut bulk = TerminalBulkState {
			is_tty_override: Some(false), // Non-TTY!
			..Default::default()
		};
		let mut out = Vec::new();

		for line in lines {
			bulk.process_line_to_writer(line, false, &mut out);
		}

		let rendered = String::from_utf8(out).unwrap();
		// Must not contain any '\r' carriage returns!
		assert!(!rendered.contains('\r'), "non-TTY sink must receive no carriage returns");
		// Must contain durable summary line and unrecognised line
		assert!(rendered.contains("copied 2/2 paths (0.1 MiB)\n"));
		assert!(rendered.contains("setting up pam...\n"));
	}
}