fxcp_core/
versioning.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
3//
4// fxcp-core/src/versioning.rs  --  MARS versioning  --  reflink snapshots, epoch management
5
6//! Mirror & Archive Recovery System (MARS)  --  zero-cost reflink snapshots
7//! with epoch-based version management and retention policies.
8
9use std::path::{Path, PathBuf};
10use crate::error::{Result, FxcpError as MirrorError};
11use std::fs;
12use std::io::{Read, Seek, SeekFrom, copy};
13use crate::sidecar;
14use tracing::{warn, info};
15use chrono::{DateTime, Utc};
16use comfy_table::{Table, Row, Cell, CellAlignment};
17use std::os::unix::fs::MetadataExt;
18use walkdir::WalkDir;
19use rayon::prelude::*;
20use crate::security;
21use std::convert::TryInto;
22use dashmap::DashMap;
23/// A single file version entry in the MARS versioning index.
24#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone)]
25pub struct FileVersion {
26    pub inode: u64,
27    pub epoch_seq: u64,
28    pub timestamp: i64,
29    pub path: PathBuf,
30    pub size: u64,
31    pub mtime: i64,
32    pub content_hash: Option<u64>,
33}
34/// In-memory index mapping inodes and content hashes to version snapshots.
35#[derive(Debug)]
36pub struct VersionLookup {
37    inode_index: DashMap<u64, Vec<FileVersion>>,
38    hash_index: DashMap<u64, Vec<PathBuf>>,
39    root: PathBuf,
40    ready: std::sync::atomic::AtomicBool,
41}
42impl VersionLookup {
43    /// Creates a new version index rooted at the given directory.
44    pub fn new(root: PathBuf) -> Self {
45        Self {
46            inode_index: DashMap::new(),
47            hash_index: DashMap::new(),
48            root,
49            ready: std::sync::atomic::AtomicBool::new(false),
50        }
51    }
52    /// Scans the versions directory and builds the inode and hash indexes.
53    pub fn index_directory(&self) {
54        let versions_dir = self.root.join(".foxing_versions").join("live");
55        // Backward compat: fall back to legacy path
56        let versions_dir = if versions_dir.exists() {
57            versions_dir
58        } else {
59            let legacy = self.root.join(".mirror").join(".versions");
60            if !legacy.exists() {
61                self.ready.store(true, std::sync::atomic::Ordering::SeqCst);
62                return;
63            }
64            legacy
65        };
66        info!("VersionLookup: Starting background scan of {:?}", versions_dir);
67        let start = std::time::Instant::now();
68        
69        // Collect entries sequentially (WalkDir is serial)
70        let entries: Vec<_> = WalkDir::new(&versions_dir)
71            .follow_links(false)
72            .min_depth(1)
73            .max_depth(1)
74            .into_iter()
75            .filter_map(|e| e.ok())
76            .filter(|e| e.file_type().is_file())
77            .collect();
78
79        // Process parsing and metadata in parallel
80        let processed_versions: Vec<FileVersion> = entries.par_iter().filter_map(|entry| {
81            let path = entry.path();
82            let name = path.file_name()?.to_str()?;
83            let parts: Vec<&str> = name.split('_').collect();
84            if parts.len() < 3 { return None; }
85            let inode = parts[0].parse::<u64>().ok()?;
86            let epoch_seq = parts[1].parse::<u64>().ok()?;
87            let timestamp = parts[2].parse::<i64>().ok()?;
88            
89            // Metadata call might block, but parallel threads handle latency
90            let meta = fs::metadata(path).ok()?;
91            let hash_bytes = sidecar::get_metadata(path, "user.foxing.content_hash");
92            let content_hash = hash_bytes.map(|b| {
93                if b.len() >= 8 {
94                    u64::from_le_bytes(b[0..8].try_into().unwrap_or([0;8]))
95                } else {
96                    0
97                }
98            });
99            Some(FileVersion {
100                inode,
101                epoch_seq,
102                timestamp,
103                path: path.to_path_buf(),
104                size: meta.len(),
105                mtime: meta.mtime(),
106                content_hash,
107            })
108        }).collect();
109
110        let mut version_count = 0;
111        for v in processed_versions {
112            let mut list = self.inode_index.entry(v.inode).or_default();
113            if let Some(hash) = v.content_hash
114                && hash != 0 {
115                    self.hash_index.entry(hash).or_default().push(v.path.clone());
116                }
117            if !list.iter().any(|existing| existing.path == v.path) {
118                list.push(v);
119            }
120        }
121        
122        // Sort individual lists (fast enough sequentially per inode usually, or could use par_iter on map)
123        for mut entry in self.inode_index.iter_mut() {
124            entry.value_mut().sort_by(|a, b| {
125                a.epoch_seq.cmp(&b.epoch_seq).then_with(|| a.timestamp.cmp(&b.timestamp))
126            });
127            version_count += entry.value().len();
128        }
129        
130        self.ready.store(true, std::sync::atomic::Ordering::SeqCst);
131        info!("VersionLookup: Indexing complete. Loaded {} versions in {:?}.",
132              version_count, start.elapsed());
133    }
134    /// Registers a new version entry, inserting it in sorted order.
135    pub fn register(&self, v: FileVersion) {
136        let mut list = self.inode_index.entry(v.inode).or_default();
137        if !list.iter().any(|existing| existing.path == v.path) {
138            list.push(v.clone());
139            list.sort_by(|a, b| {
140                a.epoch_seq.cmp(&b.epoch_seq).then_with(|| a.timestamp.cmp(&b.timestamp))
141            });
142            if let Some(hash) = v.content_hash
143                && hash != 0 {
144                    self.hash_index.entry(hash).or_default().push(v.path);
145                }
146        }
147    }
148    /// Blocks the calling thread until the background index scan completes.
149    pub fn wait_for_scan(&self) {
150        while !self.ready.load(std::sync::atomic::Ordering::SeqCst) {
151             std::thread::sleep(std::time::Duration::from_millis(crate::constants::VERSION_SCAN_POLL_INTERVAL_MS));
152        }
153    }
154    /// Finds a version file by content hash, size, and optional byte-level verification.
155    pub fn find_by_hash(&self, hash: u64, size: u64, verification_sample: &[u8]) -> Option<PathBuf> {
156        if !self.ready.load(std::sync::atomic::Ordering::SeqCst) { return None; }
157        if let Some(candidates) = self.hash_index.get(&hash).map(|map| map.value().clone()) {
158            for path in candidates {
159                if let Ok(mut f) = fs::File::open(&path)
160                    && let Ok(m) = f.metadata()
161                        && m.len() == size {
162                            if !verification_sample.is_empty() {
163                                let mut buf = vec![0u8; verification_sample.len()];
164                                if f.read_exact(&mut buf).is_ok() && buf == verification_sample {
165                                    return Some(path.clone());
166                                }
167                            } else {
168                                return Some(path.clone());
169                            }
170                        }
171            }
172        }
173        None
174    }
175    /// Finds a version file by inode number, matching on size and mtime.
176    pub fn find_by_inode(&self, inode: u64, size: u64, mtime: i64) -> Option<PathBuf> {
177        if !self.ready.load(std::sync::atomic::Ordering::SeqCst) { return None; }
178        if let Some(versions) = self.inode_index.get(&inode).map(|map| map.value().clone()) {
179            for v in versions.iter().rev() {
180                if v.size == size && v.mtime == mtime {
181                    return Some(v.path.clone());
182                }
183            }
184        }
185        None
186    }
187    /// Returns all versions for the given inode, sorted by epoch and timestamp.
188    pub fn list(&self, inode: u64) -> Vec<FileVersion> {
189        if !self.ready.load(std::sync::atomic::Ordering::SeqCst) { return vec![]; }
190        self.inode_index.get(&inode).map(|r| r.value().clone()).unwrap_or_default()
191    }
192}
193/// Compares two files by sampling the first and last 64KB of content.
194pub fn verify_content_match(p1: &Path, p2: &Path) -> Result<bool> {
195    let mut f1 = fs::File::open(p1).map_err(MirrorError::Io)?;
196    let mut f2 = fs::File::open(p2).map_err(MirrorError::Io)?;
197    let len = f1.metadata().map_err(MirrorError::Io)?.len();
198    if len != f2.metadata().map_err(MirrorError::Io)?.len() { return Ok(false); }
199    let mut buf1 = [0u8; crate::constants::CONTENT_VERIFY_SAMPLE_SIZE];
200    let mut buf2 = [0u8; crate::constants::CONTENT_VERIFY_SAMPLE_SIZE];
201    let n1 = f1.read(&mut buf1).map_err(MirrorError::Io)?;
202    let n2 = f2.read(&mut buf2).map_err(MirrorError::Io)?;
203    if n1 != n2 || buf1[..n1] != buf2[..n1] { return Ok(false); }
204    if len > crate::constants::CONTENT_VERIFY_TAIL_THRESHOLD as u64 {
205        f1.seek(SeekFrom::End(-(crate::constants::CONTENT_VERIFY_SAMPLE_SIZE as i64))).map_err(MirrorError::Io)?;
206        f2.seek(SeekFrom::End(-(crate::constants::CONTENT_VERIFY_SAMPLE_SIZE as i64))).map_err(MirrorError::Io)?;
207        let n1 = f1.read(&mut buf1).map_err(MirrorError::Io)?;
208        let n2 = f2.read(&mut buf2).map_err(MirrorError::Io)?;
209        if n1 != n2 || buf1[..n1] != buf2[..n1] { return Ok(false); }
210    }
211    Ok(true)
212}
213fn find_target_root(live_file: &Path) -> Result<PathBuf> {
214    live_file.ancestors()
215        .find(|p| p.join(".mirror").join(".versions").exists())
216        .map(|p| p.to_path_buf())
217        .ok_or_else(|| MirrorError::Versioning(format!("Could not determine target root for {:?}.", live_file)))
218}
219/// Lists all MARS versions for a live file by scanning the target root.
220pub fn list_versions(live_file: &Path) -> Result<Vec<FileVersion>> {
221    let target_root = find_target_root(live_file)?;
222    let index = VersionLookup::new(target_root);
223    index.index_directory();
224    index.wait_for_scan();
225    if let Ok(meta) = fs::metadata(live_file) {
226        Ok(index.list(meta.ino()))
227    } else {
228        Ok(vec![])
229    }
230}
231/// Removes old versions exceeding count or size limits, preserving the newest.
232pub fn cleanup_versions(live_path: &Path, _root_path: &Path, max_count: usize, max_size_mb: u64) -> Result<()> {
233    let versions = list_versions(live_path)?;
234    if versions.is_empty() { return Ok(()); }
235    let max_size_bytes = max_size_mb * 1024 * 1024;
236    let mut versions_to_delete: Vec<FileVersion> = Vec::new();
237    let mut sorted_versions = versions;
238    sorted_versions.sort_by_key(|v| v.timestamp);
239    let keep_start = sorted_versions.len().saturating_sub(max_count);
240    let mut keep_versions = sorted_versions.split_off(keep_start);
241    versions_to_delete.extend(sorted_versions);
242    let mut current_size: u64 = keep_versions.iter().map(|v| v.size).sum();
243    let mut indices_to_move = Vec::new();
244    for (i, version) in keep_versions.iter().enumerate() {
245        if current_size > max_size_bytes {
246             current_size = current_size.saturating_sub(version.size);
247             indices_to_move.push(i);
248        } else {
249             break;
250        }
251    }
252    for &i in indices_to_move.iter().rev() {
253        versions_to_delete.push(keep_versions.remove(i));
254    }
255    let now_ts = Utc::now().timestamp();
256    const DELETION_GUARD_SECS: i64 = 5;
257    for v in versions_to_delete {
258        if (now_ts - v.timestamp).abs() < DELETION_GUARD_SECS {
259            continue;
260        }
261        if v.path.exists() {
262            let _ = fs::remove_file(&v.path);
263        }
264    }
265    Ok(())
266}
267/// Deletes oldest version snapshots globally until `bytes_to_free` bytes are reclaimed.
268pub fn prune_global_history(target_root: &Path, bytes_to_free: u64) -> Result<u64> {
269    let versions_dir = target_root.join(".foxing_versions").join("live");
270    let versions_dir = if versions_dir.exists() {
271        versions_dir
272    } else {
273        let legacy = target_root.join(".mirror").join(".versions");
274        if !legacy.exists() { return Ok(0); }
275        legacy
276    };
277    let mut all_versions = Vec::new();
278    let entries = fs::read_dir(&versions_dir).map_err(MirrorError::Io)?;
279    for e in entries.flatten() {
280        let path = e.path();
281        if path.is_file()
282            && let Ok(m) = e.metadata() {
283                all_versions.push((path, m.len(), m.mtime()));
284            }
285    }
286    all_versions.sort_by_key(|v| v.2);
287    let mut freed = 0u64;
288    let mut count = 0;
289    for (path, size, _) in all_versions {
290        if freed >= bytes_to_free { break; }
291        if fs::remove_file(&path).is_ok() {
292            freed += size;
293            count += 1;
294        }
295    }
296    if count > 0 {
297        warn!("EMERGENCY: Global Pruner deleted {} historical snapshots to free {} bytes.", count, freed);
298    }
299    Ok(freed)
300}
301/// Prints a formatted table of file versions to stdout, showing the last `limit` entries.
302pub fn print_versions_table(versions: Vec<FileVersion>, limit: usize) {
303    let mut table = Table::new();
304    table.set_header(vec![
305        Cell::new("Epoch Seq").set_alignment(CellAlignment::Left),
306        Cell::new("Timestamp (UTC)").set_alignment(CellAlignment::Left),
307        Cell::new("Size").set_alignment(CellAlignment::Right),
308        Cell::new("Hash (Partial)").set_alignment(CellAlignment::Left),
309    ]);
310    let num_versions = versions.len();
311    let start_index = num_versions.saturating_sub(limit);
312    for v in versions.iter().skip(start_index) {
313        let dt = DateTime::<Utc>::from_timestamp(v.timestamp, 0).map(|dt| dt.to_string()).unwrap_or_else(|| "Invalid Date".to_string());
314        let size_mb = v.size as f64 / (1024.0 * 1024.0);
315        let hash_str = v.content_hash.map(|h| format!("{:016x}", h)).unwrap_or_else(|| "None".to_string());
316        table.add_row(Row::from(vec![
317            v.epoch_seq.to_string(),
318            dt,
319            format!("{:.2} MB", size_mb),
320            hash_str,
321        ]));
322    }
323    if num_versions > limit {
324        println!("\n... Showing last {} of {} total versions.", limit, num_versions);
325    } else {
326        println!("\nShowing {} total versions.", num_versions);
327    }
328}
329fn find_version_path(versions: &[FileVersion], epoch: u64) -> Result<&PathBuf> {
330    versions.iter()
331        .find(|v| v.epoch_seq == epoch)
332        .map(|v| &v.path)
333        .ok_or_else(|| MirrorError::Versioning(format!("Version with epoch {} not found.", epoch)))
334}
335/// Extracts a specific version epoch to a destination path.
336pub fn copy_version_to_path(live_file: &Path, epoch: u64, destination: &Path) -> Result<()> {
337    let versions = list_versions(live_file)?;
338    let version_path = find_version_path(&versions, epoch)?;
339    let mut src = fs::File::open(version_path).map_err(MirrorError::Io)?;
340    let mut dst = fs::File::create(destination).map_err(MirrorError::Io)?;
341    copy(&mut src, &mut dst).map_err(MirrorError::Io)?;
342    Ok(())
343}
344/// Atomically reverts a live file to the specified version epoch via reflink.
345pub fn revert_file(live_file: &Path, epoch: u64) -> Result<()> {
346    let target_root = find_target_root(live_file)?;
347    let index = VersionLookup::new(target_root);
348    index.index_directory();
349    index.wait_for_scan();
350    if let Ok(meta) = fs::metadata(live_file) {
351        let versions = index.list(meta.ino());
352        let version_path = find_version_path(&versions, epoch)?;
353        security::revert_snapshot(version_path, live_file)
354    } else {
355        Err(MirrorError::Versioning("Live file not found for revert.".into()))
356    }
357}
358/// CLI entry point for version cleanup with optional dry-run mode.
359pub async fn cleanup_cli(path: &Path, dry_run: bool) -> Result<()> {
360    let path = path.canonicalize().map_err(MirrorError::Io)?;
361    info!("Starting cleanup for path: {:?}", path);
362    if path.is_file() {
363        let root = find_target_root(&path)?;
364        if dry_run {
365            info!("Dry run: Would cleanup versions for file {:?}", path);
366        } else {
367            cleanup_versions(&path, &root, 5, 500)?;
368            info!("Cleaned up versions for {:?}", path);
369        }
370    } else if path.is_dir() {
371        if dry_run {
372            info!("Dry run: Would prune global history in {:?}", path);
373        } else {
374            let freed = prune_global_history(&path, 1024 * 1024 * 1024)?;
375            info!("Pruned {} bytes from global history in {:?}", freed, path);
376        }
377    }
378    Ok(())
379}
380/// CLI entry point for forcing a tagged version snapshot (not yet implemented).
381pub async fn force_version_cli(path: &Path, tag: &str) -> Result<()> {
382    let _path = path.canonicalize().map_err(MirrorError::Io)?;
383    Err(MirrorError::Versioning(format!("force_version with tag '{}' is not yet implemented", tag)))
384}
385
386#[cfg(test)]
387mod tests {
388    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
389    use super::*;
390    use tempfile::TempDir;
391    use std::io::Write;
392
393    fn write_fake_snapshot(dir: &Path, name: &str, size: usize) {
394        let path = dir.join(name);
395        let mut f = fs::File::create(&path).unwrap();
396        f.write_all(&vec![0xABu8; size]).unwrap();
397    }
398
399    #[test]
400    fn prune_finds_snapshots_in_foxing_versions_live() {
401        let tmp = TempDir::new().unwrap();
402        let live_dir = tmp.path().join(".foxing_versions").join("live");
403        fs::create_dir_all(&live_dir).unwrap();
404
405        write_fake_snapshot(&live_dir, "1001_1_1000000", 100);
406        write_fake_snapshot(&live_dir, "1002_2_1000001", 100);
407        write_fake_snapshot(&live_dir, "1003_3_1000002", 100);
408
409        let freed = prune_global_history(tmp.path(), 1).unwrap();
410        assert!(freed > 0, "expected freed > 0, got {}", freed);
411
412        let remaining: Vec<_> = fs::read_dir(&live_dir)
413            .unwrap()
414            .filter_map(|e| e.ok())
415            .collect();
416        assert!(remaining.len() < 3, "expected at least 1 file deleted, {} remain", remaining.len());
417    }
418
419    #[test]
420    fn prune_handles_empty_versions_dir() {
421        let tmp = TempDir::new().unwrap();
422        let live_dir = tmp.path().join(".foxing_versions").join("live");
423        fs::create_dir_all(&live_dir).unwrap();
424
425        let freed = prune_global_history(tmp.path(), 1024).unwrap();
426        assert_eq!(freed, 0);
427    }
428
429    #[test]
430    fn prune_falls_back_to_legacy_path() {
431        let tmp = TempDir::new().unwrap();
432        // No .foxing_versions dir  --  only legacy path
433        let legacy_dir = tmp.path().join(".mirror").join(".versions");
434        fs::create_dir_all(&legacy_dir).unwrap();
435
436        write_fake_snapshot(&legacy_dir, "2001_1_2000000", 100);
437        write_fake_snapshot(&legacy_dir, "2002_2_2000001", 100);
438
439        let freed = prune_global_history(tmp.path(), 1).unwrap();
440        assert!(freed > 0, "expected freed > 0 from legacy path, got {}", freed);
441    }
442}
fxcp_core/versioning.rs

fxcp_core/
versioning.rs
