Fix snowflake recent and archive files (7b0aa317) · Commits · juga / collector-fixes

src/fixes/fix_snowflake.rs

0 → 100644

+691 −0

Original line number	Diff line number	Diff line
		//! `fix-snowflake` - trim bloated snowflake-stats files down to a single
		//! reporting period, both in `out/recent/snowflakes/` and in the monthly
		//! `out/archive/snowflakes/snowflake-stats-*.tar.xz` tarballs.
		//!
		//! ## Background
		//!
		//! The snowflake broker's `/metrics` endpoint is meant to return a single
		//! `snowflake-stats-end <date> <time> (86400 s)` reporting block per fetch.
		//! Starting around collector-rs's deployment in 2026, the broker began
		//! returning the entire cumulative history since 2024-09-19 on every
		//! fetch instead: every previous day's block concatenated, with the new
		//! day's block appended at the end. Every file collector-rs wrote during
		//! that window inherited that bloat verbatim - both the daily files in
		//! `out/recent/snowflakes/` and the entries written into the monthly
		//! archive tarballs.
		//!
		//! `collector-rs` itself was fixed (see `snowflake_stats_latest_block` in
		//! `src/models/file.rs`) to trim future fetches down to the most recent
		//! block before they are ever written to disk. This fix repairs the
		//! files that were already written before that change shipped.
		//!
		//! ## What the fix does
		//!
		//! For each affected file (whether a loose file in `recent/` or an entry
		//! inside an archive tarball):
		//!
		//! 1. Split the content on every `snowflake-stats-end ` line.
		//! 2. If there is 0 or 1 block, the file is already correct - skip it.
		//! 3. If there are 2+ blocks, keep only the last one (the block whose
		//! date matches what the filename / tar entry date implies) and
		//! discard everything before it.
		//!
		//! `recent/` files are rewritten in place via an atomic `.tmp` + rename.
		//! Archive tarballs are read entirely into memory, trimmed entry-by-entry,
		//! and the whole tarball is rewritten via an atomic `.tmp` + rename - the
		//! same approach used by `fix_votes`.
		//!
		//! Dry-run mode prints every file/entry that would be trimmed, and by how
		//! many bytes, without touching anything on disk.

		use liblzma::{read::XzDecoder, write::XzEncoder};
		use std::{
		fs,
		path::{Path, PathBuf},
		};
		use tar::{Archive, Builder, Header};

		use collector_rs::defaults::FsData;

		// public entry point

		/// Run the fix over both `out/recent/snowflakes/` and every
		/// `snowflake-stats-*.tar.xz` under `out/archive/snowflakes/`.
		///
		/// With `dry_run = true` the function logs what would be trimmed but does
		/// not modify anything on disk.
		pub fn run(dry_run: bool) -> Result<(), Box<dyn std::error::Error>> {
		let data_dir = FsData::default().path;

		let recent_result = fix_recent(&data_dir.join("out/recent/snowflakes"), dry_run);
		let archive_result =
		fix_archive(&data_dir.join("out/archive/snowflakes"), dry_run);

		let (recent_trimmed, recent_errors) = recent_result?;
		let (archive_trimmed, archive_errors) = archive_result?;

		let total_errors = recent_errors + archive_errors;

		eprintln!(
		"fix-snowflake: done - recent_trimmed={recent_trimmed} \
		archive_entries_trimmed={archive_trimmed} errors={total_errors}{}",
		if dry_run { " (dry-run)" } else { "" },
		);

		if total_errors > 0 {
		Err(format!("{total_errors} error(s); see output above").into())
		} else {
		Ok(())
		}
		}

		// recent/ handling

		/// Trim every loose file in `recent_dir` down to its most recent
		/// `snowflake-stats-end` block.
		///
		/// Returns `(files_trimmed, error_count)`.
		fn fix_recent(
		recent_dir: &Path,
		dry_run: bool,
		) -> Result<(usize, usize), Box<dyn std::error::Error>> {
		eprintln!("fix-snowflake: scanning {}", recent_dir.display());

		if !recent_dir.exists() {
		eprintln!(
		"fix-snowflake: recent directory does not exist: {}",
		recent_dir.display()
		);
		return Ok((0, 0));
		}

		let mut files: Vec<PathBuf> = fs::read_dir(recent_dir)?
		.filter_map(\|e\| e.ok())
		.map(\|e\| e.path())
		.filter(\|p\| p.is_file())
		.collect();
		files.sort();

		if files.is_empty() {
		eprintln!("fix-snowflake: no files found in {}", recent_dir.display());
		return Ok((0, 0));
		}

		let mut trimmed = 0usize;
		let mut errors = 0usize;

		for path in &files {
		let file_name = path
		.file_name()
		.and_then(\|n\| n.to_str())
		.unwrap_or("?")
		.to_owned();

		let content = match fs::read(path) {
		Ok(c) => c,
		Err(e) => {
		eprintln!("fix-snowflake: could not read {}: {e}", path.display());
		errors += 1;
		continue;
		}
		};

		let block_count = count_blocks(&content);
		if block_count <= 1 {
		continue;
		}

		let kept = latest_block(&content);

		println!(
		"{}recent/{file_name}: {block_count} blocks -> 1 \
		({} bytes -> {} bytes)",
		if dry_run { "[dry-run] " } else { "" },
		content.len(),
		kept.len(),
		);

		if dry_run {
		trimmed += 1;
		continue;
		}

		if let Err(e) = write_atomic(path, &kept) {
		eprintln!("fix-snowflake: failed to rewrite {}: {e}", path.display());
		errors += 1;
		continue;
		}

		trimmed += 1;
		}

		Ok((trimmed, errors))
		}

		/// Atomically rewrite `path` with `content` via a `.tmp` sibling + rename.
		fn write_atomic(path: &Path, content: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
		let tmp = path.with_extension("tmp");
		fs::write(&tmp, content)?;
		fs::rename(&tmp, path)?;
		Ok(())
		}

		// archive/ handling

		/// An entry read from a tarball: its tar path and raw bytes.
		struct TarEntry {
		path: PathBuf,
		content: Vec<u8>,
		mode: u32,
		}

		/// Trim every `snowflake-stats-*.tar.xz` entry under `archive_dir` down to
		/// its most recent `snowflake-stats-end` block.
		///
		/// Returns `(entries_trimmed, error_count)`.
		fn fix_archive(
		archive_dir: &Path,
		dry_run: bool,
		) -> Result<(usize, usize), Box<dyn std::error::Error>> {
		eprintln!("fix-snowflake: scanning {}", archive_dir.display());

		if !archive_dir.exists() {
		eprintln!(
		"fix-snowflake: archive directory does not exist: {}",
		archive_dir.display()
		);
		return Ok((0, 0));
		}

		let mut tarballs: Vec<PathBuf> = fs::read_dir(archive_dir)?
		.filter_map(\|e\| e.ok())
		.map(\|e\| e.path())
		.filter(\|p\| {
		p.file_name()
		.and_then(\|n\| n.to_str())
		.map(\|n\| n.starts_with("snowflake-stats-") && n.ends_with(".tar.xz"))
		.unwrap_or(false)
		})
		.collect();
		tarballs.sort();

		if tarballs.is_empty() {
		eprintln!(
		"fix-snowflake: no snowflake-stats-*.tar.xz files found in {}",
		archive_dir.display()
		);
		return Ok((0, 0));
		}

		let mut total_trimmed = 0usize;
		let mut total_errors = 0usize;

		for tarball in &tarballs {
		match fix_tarball(tarball, dry_run) {
		Ok((trimmed, errors)) => {
		total_trimmed += trimmed;
		total_errors += errors;
		}
		Err(e) => {
		eprintln!(
		"fix-snowflake: failed to process {}: {e}",
		tarball.display()
		);
		total_errors += 1;
		}
		}
		}

		Ok((total_trimmed, total_errors))
		}

		/// Process one archive tarball entirely in memory, trimming every entry
		/// that contains more than one `snowflake-stats-end` block.
		///
		/// Returns `(entries_trimmed, error_count)`.
		fn fix_tarball(
		tarball: &Path,
		dry_run: bool,
		) -> Result<(usize, usize), Box<dyn std::error::Error>> {
		let tarball_name = tarball
		.file_name()
		.and_then(\|n\| n.to_str())
		.unwrap_or("?");

		let mut entries = read_tar_entries(tarball)?;
		entries.sort_by(\|a, b\| a.path.cmp(&b.path));

		let mut trimmed_indices: Vec<usize> = Vec::new();
		let mut errors = 0usize;

		for (idx, entry) in entries.iter().enumerate() {
		let entry_name = entry.path.to_string_lossy().into_owned();

		let block_count = count_blocks(&entry.content);
		if block_count <= 1 {
		continue;
		}

		let kept = latest_block(&entry.content);

		println!(
		"{}{tarball_name}: {entry_name}: {block_count} blocks -> 1 \
		({} bytes -> {} bytes)",
		if dry_run { "[dry-run] " } else { "" },
		entry.content.len(),
		kept.len(),
		);

		trimmed_indices.push(idx);
		}

		if trimmed_indices.is_empty() {
		return Ok((0, errors));
		}

		if dry_run {
		return Ok((trimmed_indices.len(), errors));
		}

		for idx in &trimmed_indices {
		entries[idx].content = latest_block(&entries[idx].content);
		}

		let tmp_out = tarball.with_extension("tar.xz.tmp");
		if let Err(e) = write_tar_entries(&entries, &tmp_out) {
		eprintln!("fix-snowflake: [{tarball_name}] failed to repack: {e}");
		errors += 1;
		let _ = fs::remove_file(&tmp_out);
		return Ok((0, errors));
		}
		fs::rename(&tmp_out, tarball)?;

		eprintln!(
		"fix-snowflake: [{tarball_name}] replaced ({} entries trimmed)",
		trimmed_indices.len()
		);

		Ok((trimmed_indices.len(), errors))
		}

		// tar helpers (same approach as fix_votes)

		/// Read every file entry from a `.tar.xz` into memory.
		fn read_tar_entries(
		tarball: &Path,
		) -> Result<Vec<TarEntry>, Box<dyn std::error::Error>> {
		use std::io::Read as _;

		let file = fs::File::open(tarball)?;
		let decoder = XzDecoder::new(file);
		let mut ar = Archive::new(decoder);

		let mut out = Vec::new();
		for entry in ar.entries()? {
		let mut entry = entry?;
		let path = entry.path()?.into_owned();
		let mode = entry.header().mode().unwrap_or(0o644);
		let mut content = Vec::new();
		entry.read_to_end(&mut content)?;
		out.push(TarEntry { path, content, mode });
		}
		Ok(out)
		}

		/// Write a list of in-memory entries into a new `.tar.xz` file.
		fn write_tar_entries(
		entries: &[TarEntry],
		dest: &Path,
		) -> Result<(), Box<dyn std::error::Error>> {
		let out_file = fs::File::create(dest)?;
		let encoder = XzEncoder::new(out_file, 6);
		let mut ar = Builder::new(encoder);

		for entry in entries {
		let mut header = Header::new_gnu();
		header.set_size(entry.content.len() as u64);
		header.set_mode(entry.mode);
		header.set_cksum();
		ar.append_data(&mut header, &entry.path, entry.content.as_slice())
		.map_err(\|e\| format!("tar append {}: {e}", entry.path.display()))?;
		}

		ar.into_inner()
		.map_err(\|e\| format!("tar finish: {e}"))?
		.finish()
		.map_err(\|e\| format!("xz finish: {e}"))?;

		Ok(())
		}

		// content helpers

		const MARKER: &str = "snowflake-stats-end ";

		/// Count how many `snowflake-stats-end ` blocks `content` contains.
		///
		/// Returns 0 if the marker is absent or content is not valid UTF-8 (in
		/// which case the caller should leave the content untouched).
		fn count_blocks(content: &[u8]) -> usize {
		let Ok(text) = std::str::from_utf8(content) else {
		return 0;
		};
		text.lines().filter(\|l\| l.starts_with(MARKER)).count()
		}

		/// Return only the most recent `snowflake-stats-end` block in `content`.
		///
		/// Mirrors `snowflake_stats_latest_block` in `collector-rs`'s
		/// `src/models/file.rs`: find every line starting with the marker, and
		/// keep everything from the last such line onward. Any preamble before
		/// the first marker (e.g. an `@type` annotation) is dropped along with
		/// the earlier blocks - `annotate_content` already re-adds the
		/// annotation elsewhere in `collector-rs`'s pipeline, and tar entries
		/// written before the original fix never carried one anyway.
		///
		/// If content is not valid UTF-8, or the marker is absent or appears
		/// only once, the content is returned unchanged.
		fn latest_block(content: &[u8]) -> Vec<u8> {
		let Ok(text) = std::str::from_utf8(content) else {
		return content.to_vec();
		};

		let mut block_starts: Vec<usize> = Vec::new();
		let mut offset = 0usize;
		for line in text.split_inclusive('\n') {
		if line.starts_with(MARKER) {
		block_starts.push(offset);
		}
		offset += line.len();
		}

		match block_starts.len() {
		0 \| 1 => content.to_vec(),
		_ => {
		let last_start = *block_starts.last().unwrap();
		text[last_start..].as_bytes().to_vec()
		}
		}
		}

		// tests

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

		fn block(date: &str, ips: &str) -> String {
		format!(
		"snowflake-stats-end {date} 13:40:04 (86400 s)\n\
		snowflake-ips {ips}\n\
		snowflake-idle-count 1\n"
		)
		}

		fn make_tar_xz(entries: &[(&str, &[u8])]) -> Vec<u8> {
		let mut buf = Vec::new();
		{
		let enc = XzEncoder::new(&mut buf, 1);
		let mut ar = Builder::new(enc);
		for (name, data) in entries {
		let mut header = Header::new_gnu();
		header.set_size(data.len() as u64);
		header.set_mode(0o644);
		header.set_cksum();
		ar.append_data(&mut header, name, *data).unwrap();
		}
		ar.into_inner().unwrap().finish().unwrap();
		}
		buf
		}

		// content helper unit tests

		#[test]
		fn count_blocks_zero_for_no_marker() {
		let content = b"no marker here\njust text\n";
		assert_eq!(count_blocks(content), 0);
		}

		#[test]
		fn count_blocks_one_for_single_block() {
		let content = block("2026-05-03", "us=300").into_bytes();
		assert_eq!(count_blocks(&content), 1);
		}

		#[test]
		fn count_blocks_counts_multiple() {
		let content = format!(
		"{}{}{}",
		block("2026-05-01", "us=100"),
		block("2026-05-02", "us=200"),
		block("2026-05-03", "us=300"),
		)
		.into_bytes();
		assert_eq!(count_blocks(&content), 3);
		}

		#[test]
		fn latest_block_no_op_for_single_block() {
		let content = block("2026-05-03", "us=300").into_bytes();
		assert_eq!(latest_block(&content), content);
		}

		#[test]
		fn latest_block_keeps_only_last_of_many() {
		let b1 = block("2026-05-01", "us=100");
		let b2 = block("2026-05-02", "us=200");
		let b3 = block("2026-05-03", "us=300");
		let content = format!("{b1}{b2}{b3}").into_bytes();

		let result = latest_block(&content);
		assert_eq!(result, b3.into_bytes());

		let result_str = String::from_utf8(result).unwrap();
		assert!(!result_str.contains("2026-05-01"));
		assert!(!result_str.contains("2026-05-02"));
		assert!(result_str.contains("2026-05-03"));
		}

		#[test]
		fn latest_block_handles_real_broker_sample() {
		// Verbatim two-day excerpt from an actual bloated broker fetch.
		let content = "\
		snowflake-stats-end 2024-09-19 13:40:04 (86400 s)
		snowflake-ips ??=1
		snowflake-ips-total 1
		snowflake-idle-count 216
		snowflake-ips-nat-unknown 1
		snowflake-stats-end 2024-09-20 13:40:04 (86400 s)
		snowflake-ips
		snowflake-ips-total 0
		snowflake-idle-count 0
		snowflake-ips-nat-unknown 0
		"
		.as_bytes();

		let result = latest_block(content);
		let result_str = String::from_utf8(result).unwrap();

		assert!(result_str.starts_with(
		"snowflake-stats-end 2024-09-20 13:40:04 (86400 s)"
		));
		assert!(!result_str.contains("2024-09-19"));
		assert!(result_str.contains("snowflake-ips-nat-unknown 0"));
		}

		// fix_recent integration tests

		#[test]
		fn fix_recent_trims_bloated_file_in_place() {
		let dir = TempDir::new().unwrap();
		let recent_dir = dir.path().join("snowflakes");
		fs::create_dir_all(&recent_dir).unwrap();

		let bloated = format!(
		"{}{}",
		block("2026-05-01", "us=100"),
		block("2026-05-02", "us=200"),
		);
		let path = recent_dir.join("2026-05-02-00-00-00-snowflake-stats");
		fs::write(&path, &bloated).unwrap();

		let (trimmed, errors) = fix_recent(&recent_dir, false).unwrap();
		assert_eq!(trimmed, 1);
		assert_eq!(errors, 0);

		let result = fs::read_to_string(&path).unwrap();
		assert!(!result.contains("2026-05-01"));
		assert!(result.contains("2026-05-02"));
		}

		#[test]
		fn fix_recent_skips_already_correct_file() {
		let dir = TempDir::new().unwrap();
		let recent_dir = dir.path().join("snowflakes");
		fs::create_dir_all(&recent_dir).unwrap();

		let good = block("2026-05-02", "us=200");
		let path = recent_dir.join("2026-05-02-00-00-00-snowflake-stats");
		fs::write(&path, &good).unwrap();
		let original_len = good.len();

		let (trimmed, errors) = fix_recent(&recent_dir, false).unwrap();
		assert_eq!(trimmed, 0);
		assert_eq!(errors, 0);
		assert_eq!(fs::read(&path).unwrap().len(), original_len);
		}

		#[test]
		fn fix_recent_dry_run_does_not_modify_file() {
		let dir = TempDir::new().unwrap();
		let recent_dir = dir.path().join("snowflakes");
		fs::create_dir_all(&recent_dir).unwrap();

		let bloated = format!(
		"{}{}",
		block("2026-05-01", "us=100"),
		block("2026-05-02", "us=200"),
		);
		let path = recent_dir.join("2026-05-02-00-00-00-snowflake-stats");
		fs::write(&path, &bloated).unwrap();
		let original_len = bloated.len();

		let (trimmed, errors) = fix_recent(&recent_dir, true).unwrap();
		assert_eq!(trimmed, 1);
		assert_eq!(errors, 0);
		assert_eq!(fs::read(&path).unwrap().len(), original_len);
		}

		// fix_tarball integration tests

		#[test]
		fn fix_tarball_trims_bloated_entry_and_repacks() {
		let b1 = block("2026-05-01", "us=100");
		let b2 = block("2026-05-02", "us=200");
		let bloated_entry = format!("{b1}{b2}");

		let entry_path = "snowflake-stats-2026-05/02/2026-05-02-snowflake-stats";
		let tar_bytes = make_tar_xz(&[(entry_path, bloated_entry.as_bytes())]);

		let dir = TempDir::new().unwrap();
		let tarball = dir.path().join("snowflake-stats-2026-05.tar.xz");
		fs::write(&tarball, &tar_bytes).unwrap();

		let (trimmed, errors) = fix_tarball(&tarball, false).unwrap();
		assert_eq!(trimmed, 1);
		assert_eq!(errors, 0);

		let entries = read_tar_entries(&tarball).unwrap();
		assert_eq!(entries.len(), 1);
		let content = String::from_utf8(entries[0].content.clone()).unwrap();
		assert!(!content.contains("2026-05-01"));
		assert!(content.contains("2026-05-02"));
		}

		#[test]
		fn fix_tarball_skips_already_correct_entries() {
		let good_entry = block("2026-05-02", "us=200");
		let entry_path = "snowflake-stats-2026-05/02/2026-05-02-snowflake-stats";
		let tar_bytes = make_tar_xz(&[(entry_path, good_entry.as_bytes())]);

		let dir = TempDir::new().unwrap();
		let tarball = dir.path().join("snowflake-stats-2026-05.tar.xz");
		fs::write(&tarball, &tar_bytes).unwrap();
		let original_len = tar_bytes.len();

		let (trimmed, errors) = fix_tarball(&tarball, false).unwrap();
		assert_eq!(trimmed, 0);
		assert_eq!(errors, 0);
		assert_eq!(fs::read(&tarball).unwrap().len(), original_len);
		}

		#[test]
		fn fix_tarball_dry_run_does_not_modify_tarball() {
		let b1 = block("2026-05-01", "us=100");
		let b2 = block("2026-05-02", "us=200");
		let bloated_entry = format!("{b1}{b2}");

		let entry_path = "snowflake-stats-2026-05/02/2026-05-02-snowflake-stats";
		let tar_bytes = make_tar_xz(&[(entry_path, bloated_entry.as_bytes())]);

		let dir = TempDir::new().unwrap();
		let tarball = dir.path().join("snowflake-stats-2026-05.tar.xz");
		fs::write(&tarball, &tar_bytes).unwrap();
		let original_len = tar_bytes.len();

		let (trimmed, errors) = fix_tarball(&tarball, true).unwrap();
		assert_eq!(trimmed, 1);
		assert_eq!(errors, 0);
		assert_eq!(fs::read(&tarball).unwrap().len(), original_len);
		}

		#[test]
		fn fix_tarball_trims_multiple_bloated_entries_independently() {
		// Two daily entries, each bloated by a different amount, in the
		// same tarball - both must be trimmed to their own latest block.
		let day1_bloated = format!(
		"{}{}",
		block("2026-05-01", "us=100"),
		block("2026-05-01", "us=150"), // pretend duplicate header, still 2 blocks
		);
		let day2_bloated = format!(
		"{}{}{}",
		block("2026-05-01", "us=100"),
		block("2026-05-02", "us=200"),
		block("2026-05-03", "us=300"),
		);

		let entries = [
		(
		"snowflake-stats-2026-05/01/2026-05-01-snowflake-stats",
		day1_bloated.as_bytes(),
		),
		(
		"snowflake-stats-2026-05/03/2026-05-03-snowflake-stats",
		day2_bloated.as_bytes(),
		),
		];
		let tar_bytes = make_tar_xz(&entries);

		let dir = TempDir::new().unwrap();
		let tarball = dir.path().join("snowflake-stats-2026-05.tar.xz");
		fs::write(&tarball, &tar_bytes).unwrap();

		let (trimmed, errors) = fix_tarball(&tarball, false).unwrap();
		assert_eq!(trimmed, 2);
		assert_eq!(errors, 0);

		let result_entries = read_tar_entries(&tarball).unwrap();
		assert_eq!(result_entries.len(), 2);
		for e in &result_entries {
		assert_eq!(
		count_blocks(&e.content),
		1,
		"entry {} still has multiple blocks",
		e.path.display(),
		);
		}
		}
		}

src/fixes/mod.rs

+9 −2

Original line number	Diff line number	Diff line
		@@ -17,6 +17,7 @@
		//! Modules are named `fix_<noun>` (eg. `fix_votes`). The CLI name is the
		//! noun without the prefix (eg. `votes`).

		pub mod fix_snowflake;
		pub mod fix_votes;

		// registry
		@@ -27,6 +28,11 @@ pub const FIXES: &[(&str, &str)] = &[
		"votes",
		"Repair vote filenames where the SHA1 suffix was truncated.",
		),
		(
		"snowflake",
		"Trim bloated snowflake-stats files (recent and archive) down \
		to a single reporting period each.",
		),
		];

		/// Dispatch to the fix named `name`.
		@@ -38,6 +44,7 @@ pub fn run(
		) -> Result<(), Box<dyn std::error::Error>> {
		match name {
		"votes" => fix_votes::run(dry_run),
		"snowflake" => fix_snowflake::run(dry_run),
		other => Err(format!("unknown fix: '{other}'").into()),
		}
		}