dvs/files/
status.rs

1//! Local facts for machine-readable status, using cached hashes when available.
2use crate::cache::{self, FileStat};
3use crate::operation::{FileRow, Hashing, OperationContext, OperationPlan, OperationRow, RowState};
4use crate::{DvsPaths, ErrorCode, FileMetadata, Hashes, PathFilter, Status};
5use anyhow::Result;
6use fs_err as fs;
7use rayon::prelude::*;
8use serde::ser::SerializeStruct;
9use serde::{Serialize, Serializer};
10use std::path::PathBuf;
11use std::sync::Mutex;
12
13#[derive(Clone, Debug, Serialize)]
14pub struct Baseline {
15    pub blake3: String,
16    pub size: u64,
17}
18#[derive(Clone, Debug, Serialize)]
19pub struct WorkingCopy {
20    /// Whether the working file is present and available for inspection.
21    pub state: &'static str,
22    pub hashes: Option<Hashes>,
23    pub size: Option<u64>,
24    pub hash_source: Option<&'static str>,
25}
26#[derive(Clone, Debug)]
27pub struct StatusFacts {
28    pub path: PathBuf,
29    /// `None` on an error or cancelled row, which `code` and `error` describe.
30    pub status: Option<Status>,
31    pub metadata: Option<FileMetadata>,
32    pub working: WorkingCopy,
33    pub baseline: Option<Baseline>,
34    pub code: Option<ErrorCode>,
35    pub error: Option<String>,
36}
37
38impl StatusFacts {
39    /// The row's status as shown to people and in results: `error` on an
40    /// error row, `cancelled` on a row a cancel stopped.
41    pub fn status_name(&self) -> String {
42        match (self.status, self.code) {
43            (Some(status), _) => status.to_string(),
44            (None, Some(ErrorCode::Cancelled)) => "cancelled".to_string(),
45            (None, _) => "error".to_string(),
46        }
47    }
48}
49
50impl Serialize for StatusFacts {
51    fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
52        let mut row = s.serialize_struct("StatusFacts", 7)?;
53        row.serialize_field("path", &self.path)?;
54        row.serialize_field("status", &self.status_name())?;
55        row.serialize_field("metadata", &self.metadata)?;
56        row.serialize_field("working", &self.working)?;
57        row.serialize_field("baseline", &self.baseline)?;
58        row.serialize_field("code", &self.code)?;
59        row.serialize_field("error", &self.error)?;
60        row.end()
61    }
62}
63
64impl OperationRow for StatusFacts {
65    fn state(&self) -> RowState {
66        match (self.status, self.code) {
67            (Some(_), _) => RowState::Succeeded,
68            (None, Some(ErrorCode::Cancelled)) => RowState::Cancelled,
69            (None, _) => RowState::Failed,
70        }
71    }
72}
73
74/// The rows of a status run, sorted by path.
75#[derive(Clone, Debug, Default)]
76pub struct StatusReport {
77    pub files: Vec<StatusFacts>,
78    /// The cancel flag was set when the run finished
79    pub cancelled: bool,
80}
81
82/// How a present working file relates to its record and to the local baseline.
83fn classify(
84    hashes: &Hashes,
85    size: u64,
86    metadata: &FileMetadata,
87    baseline: Option<&Baseline>,
88) -> Status {
89    if *hashes == metadata.hashes && size == metadata.size {
90        Status::Current
91    } else {
92        match baseline {
93            Some(b) if b.blake3 == hashes.blake3 && b.size == size => Status::Stale,
94            Some(_) => Status::Modified,
95            None => Status::Unsynced,
96        }
97    }
98}
99
100/// Inspect tracked files in parallel. Each row goes to `ctx`'s observer as
101/// soon as it completes, so callers can stream rows; the returned rows are
102/// sorted by path. Hashing is reported only once a file is actually hashed.
103/// Once `ctx` is cancelled, files not yet started get a `cancelled` row, and a
104/// run that fails after a cancel still returns its (empty) cancelled report.
105///
106/// Paths in the result are relative to `paths.repo_root()`, independent of cwd.
107pub fn inspect_status(
108    paths: &DvsPaths,
109    filter: Option<&PathFilter>,
110    glob_pattern: Option<&str>,
111    ctx: &OperationContext,
112) -> Result<StatusReport> {
113    match inspect_all(paths, filter, glob_pattern, ctx) {
114        Ok(files) => Ok(StatusReport {
115            files,
116            cancelled: ctx.is_cancelled(),
117        }),
118        Err(_) if ctx.is_cancelled() => Ok(StatusReport {
119            files: Vec::new(),
120            cancelled: true,
121        }),
122        Err(e) => Err(e),
123    }
124}
125
126fn inspect_all(
127    paths: &DvsPaths,
128    filter: Option<&PathFilter>,
129    glob_pattern: Option<&str>,
130    ctx: &OperationContext,
131) -> Result<Vec<StatusFacts>> {
132    let observer = ctx.observer();
133    let (entries, cache, pool) = status_scan_inputs(paths, filter, glob_pattern)?;
134    // What hashing will read: the working files that are regular files. An
135    // extra stat per file, so only when someone hears the plan.
136    let bytes_total = if ctx.is_observed() {
137        pool.install(|| {
138            entries
139                .par_iter()
140                .filter_map(|path| fs::symlink_metadata(paths.file_path(path)).ok())
141                .filter(|meta| meta.is_file())
142                .map(|meta| meta.len())
143                .sum()
144        })
145    } else {
146        0
147    };
148    observer.planned(&OperationPlan {
149        files_total: entries.len() as u64,
150        bytes_total,
151        files_to_transfer: 0,
152        bytes_to_transfer: 0,
153    });
154    let hashing = Hashing::new(observer);
155    let mut results = pool.install(|| {
156        entries
157            .into_par_iter()
158            .map(|path| {
159                let row = if ctx.is_cancelled() {
160                    cancelled_row(path)
161                } else {
162                    inspect_path(path, paths, &cache, &hashing)
163                };
164                observer.file_finished(FileRow::Status(&row));
165                row
166            })
167            .collect::<Vec<_>>()
168    });
169    hashing.finish();
170    results.sort_by(|a, b| a.path.cmp(&b.path));
171    Ok(results)
172}
173
174/// A file the cancelled operation never looked at.
175fn cancelled_row(path: PathBuf) -> StatusFacts {
176    StatusFacts {
177        path,
178        status: None,
179        metadata: None,
180        working: WorkingCopy {
181            state: "unknown",
182            hashes: None,
183            size: None,
184            hash_source: None,
185        },
186        baseline: None,
187        code: Some(ErrorCode::Cancelled),
188        error: Some("cancelled before this file was checked".to_string()),
189    }
190}
191
192fn status_scan_inputs(
193    paths: &DvsPaths,
194    filter: Option<&PathFilter>,
195    glob_pattern: Option<&str>,
196) -> Result<(
197    Vec<PathBuf>,
198    Option<Mutex<cache::HashCache>>,
199    rayon::ThreadPool,
200)> {
201    let glob = crate::globbing::build_glob_matcher(glob_pattern)?;
202    let cwd_filter = if filter.is_none() && glob_pattern.is_some() {
203        Some(PathFilter::cwd_scoped(false, paths))
204    } else {
205        None
206    };
207    let filter = filter.or(cwd_filter.as_ref());
208    let cache = cache::try_open_cache(paths);
209    let entries: Vec<_> = paths
210        .tracked_paths()
211        .into_iter()
212        .filter(|path| !filter.is_some_and(|f| !f.matches(path, glob.as_ref())))
213        .collect();
214    let pool = crate::utils::get_threadpool(entries.len().max(1))?;
215    Ok((entries, cache, pool))
216}
217
218fn inspect_path(
219    path: PathBuf,
220    paths: &DvsPaths,
221    cache: &Option<Mutex<cache::HashCache>>,
222    hashing: &Hashing<'_>,
223) -> StatusFacts {
224    let mut row = StatusFacts {
225        path: path.clone(),
226        status: None,
227        metadata: None,
228        working: WorkingCopy {
229            state: "unknown",
230            hashes: None,
231            size: None,
232            hash_source: None,
233        },
234        baseline: None,
235        code: None,
236        error: None,
237    };
238    let facts = (|| -> Result<()> {
239        row.code = Some(ErrorCode::BadMetadata);
240        row.metadata = Some(serde_json::from_reader(fs::File::open(
241            paths.metadata_path(&path),
242        )?)?);
243        row.code = Some(ErrorCode::IoError);
244        let key = path.to_string_lossy();
245        row.baseline = cache
246            .as_ref()
247            .and_then(|c| c.lock().ok()?.baseline(&key).ok().flatten())
248            .map(|(blake3, size)| Baseline { blake3, size });
249        let full = paths.file_path(&path);
250        // Never hash a symlink or traverse a symlinked parent outside the project.
251        let mut component = paths.repo_root().to_path_buf();
252        for part in path.components() {
253            component.push(part.as_os_str());
254            match fs::symlink_metadata(&component) {
255                Ok(meta) if meta.file_type().is_symlink() => {
256                    anyhow::bail!("working path contains a symlink")
257                }
258                Ok(_) => (),
259                Err(e) if e.kind() == std::io::ErrorKind::NotFound => break,
260                Err(e) => return Err(e.into()),
261            }
262        }
263        let stat = match FileStat::from_path(&full) {
264            Ok(stat) => stat,
265            Err(e)
266                if e.downcast_ref::<std::io::Error>()
267                    .is_some_and(|e| e.kind() == std::io::ErrorKind::NotFound) =>
268            {
269                row.status = Some(Status::Absent);
270                row.working.state = "absent";
271                return Ok(());
272            }
273            Err(e) => return Err(e),
274        };
275        anyhow::ensure!(full.is_file(), "working path is not a regular file");
276        row.working.size = Some(stat.size);
277        // Hashing starts here: a cache lookup counts, while
278        // absent and failed rows never start hashing.
279        hashing.begin();
280        let hit = cache
281            .as_ref()
282            .and_then(|c| c.lock().ok()?.lookup(&key, &stat).ok().flatten());
283        let (hashes, size, source) = match hit {
284            Some(hashes) => (hashes, stat.size, "cache"),
285            None => {
286                let (hashes, size) = cache::hashes_for_file(&full, &key, cache.as_ref())?;
287                (hashes, size, "computed")
288            }
289        };
290        row.status = Some(classify(
291            &hashes,
292            size,
293            row.metadata.as_ref().unwrap(),
294            row.baseline.as_ref(),
295        ));
296        row.working = WorkingCopy {
297            state: "present",
298            hashes: Some(hashes),
299            size: Some(size),
300            hash_source: Some(source),
301        };
302        Ok(())
303    })();
304    match facts {
305        Ok(()) => row.code = None,
306        Err(e) => {
307            row.status = None;
308            row.error = Some(format!("{e:#}"));
309        }
310    }
311    row
312}
313
314#[cfg(test)]
315mod tests {
316    use super::*;
317    use crate::Compression;
318    use crate::operation::OperationObserver;
319    use crate::testutil::{create_file, create_temp_git_repo, init_dvs_repo};
320    use std::sync::mpsc;
321    use uuid::Uuid;
322    fn fixture() -> (tempfile::TempDir, DvsPaths) {
323        let (temp, root) = create_temp_git_repo();
324        let (config, _) = init_dvs_repo(&root);
325        let paths =
326            DvsPaths::new(root.clone(), root.clone(), config.metadata_folder_name()).unwrap();
327        let full = create_file(&root, "data.txt", b"original");
328        let (hashes, size) = Hashes::compute_from_path(&full, &[]).unwrap();
329        // Serialize via the actual metadata type so the fixture tracks the file format.
330        let metadata = FileMetadata::from_file(&full, crate::Compression::Zstd, None).unwrap();
331        fs::write(
332            paths.metadata_path(std::path::Path::new("data.txt")),
333            serde_json::to_vec(&metadata).unwrap(),
334        )
335        .unwrap();
336        assert_eq!(hashes.blake3, metadata.hashes.blake3);
337        assert_eq!(size, metadata.size);
338        (temp, paths)
339    }
340    #[test]
341    fn a_cancel_before_the_run_leaves_every_file_unchecked() {
342        let (_temp, paths) = fixture();
343        let token = crate::CancelToken::new();
344        token.cancel();
345        let ctx = OperationContext::new().with_cancel(token);
346        let report = inspect_status(&paths, None, None, &ctx).unwrap();
347        assert!(report.cancelled);
348        let row = &report.files[0];
349        assert_eq!((row.status, row.code), (None, Some(ErrorCode::Cancelled)));
350        assert_eq!(row.state(), RowState::Cancelled);
351        assert_eq!(serde_json::to_value(row).unwrap()["status"], "cancelled");
352        // A run that fails after a cancel (here an invalid glob) is still cancelled
353        let report = inspect_status(&paths, None, Some("["), &ctx).unwrap();
354        assert!(report.cancelled && report.files.is_empty());
355    }
356
357    #[test]
358    fn cache_miss_computes_status_facts() {
359        let (_temp, paths) = fixture();
360        let rows = inspect_status(&paths, None, None, &OperationContext::new())
361            .unwrap()
362            .files;
363        assert_eq!(rows[0].status, Some(Status::Current));
364        assert_eq!(rows[0].working.hash_source, Some("computed"));
365        #[cfg(feature = "cache")]
366        {
367            let rows = inspect_status(&paths, None, None, &OperationContext::new())
368                .unwrap()
369                .files;
370            assert_eq!(rows[0].status, Some(Status::Current));
371            assert_eq!(rows[0].working.hash_source, Some("cache"));
372            // A changed stat invalidates the cache without reading new file content.
373            fs::write(paths.repo_root().join("data.txt"), b"changed content").unwrap();
374            assert_eq!(
375                inspect_status(&paths, None, None, &OperationContext::new())
376                    .unwrap()
377                    .files[0]
378                    .status,
379                Some(Status::Unsynced)
380            );
381        }
382    }
383    #[test]
384    fn missing_and_bad_metadata_are_distinct() {
385        let (_temp, paths) = fixture();
386        fs::remove_file(paths.repo_root().join("data.txt")).unwrap();
387        let rows = inspect_status(&paths, None, None, &OperationContext::new())
388            .unwrap()
389            .files;
390        assert_eq!(rows[0].status, Some(Status::Absent));
391        assert!(rows[0].error.is_none());
392        fs::write(
393            paths.metadata_path(std::path::Path::new("data.txt")),
394            b"broken",
395        )
396        .unwrap();
397        let rows = inspect_status(&paths, None, None, &OperationContext::new())
398            .unwrap()
399            .files;
400        assert_eq!(rows[0].status, None);
401        assert_eq!(serde_json::to_value(&rows[0]).unwrap()["status"], "error");
402        assert_eq!(rows[0].code, Some(ErrorCode::BadMetadata));
403    }
404
405    #[test]
406    fn callback_receives_a_completed_row_before_the_scan_returns() {
407        let (_temp, paths) = fixture();
408        struct Pausing {
409            rows: Mutex<mpsc::Sender<StatusFacts>>,
410            resume: Mutex<mpsc::Receiver<()>>,
411        }
412        impl OperationObserver for Pausing {
413            fn file_finished(&self, row: FileRow<'_>) {
414                let FileRow::Status(row) = row else {
415                    panic!("status only reports status rows")
416                };
417                self.rows.lock().unwrap().send(row.clone()).unwrap();
418                self.resume.lock().unwrap().recv().unwrap();
419            }
420        }
421        let (row_sender, row_receiver) = mpsc::channel();
422        let (continue_sender, continue_receiver) = mpsc::channel();
423        let ctx = OperationContext::new().with_observer(std::sync::Arc::new(Pausing {
424            rows: Mutex::new(row_sender),
425            resume: Mutex::new(continue_receiver),
426        }));
427
428        std::thread::scope(|scope| {
429            let scan = scope.spawn(|| inspect_status(&paths, None, None, &ctx));
430            let row = row_receiver
431                .recv_timeout(std::time::Duration::from_secs(5))
432                .unwrap();
433            assert_eq!(row.status, Some(Status::Current));
434            assert!(!scan.is_finished());
435            continue_sender.send(()).unwrap();
436            scan.join().unwrap().unwrap();
437        });
438    }
439
440    #[test]
441    #[cfg(feature = "cache")]
442    fn baseline_distinguishes_stale_from_modified() {
443        let (_temp, paths) = fixture();
444        let cache = cache::open_cache(&paths).unwrap();
445        let rows = inspect_status(&paths, None, None, &OperationContext::new())
446            .unwrap()
447            .files;
448        let original = rows[0].working.hashes.as_ref().unwrap();
449        cache
450            .set_baseline(
451                "data.txt",
452                &original.blake3,
453                rows[0].metadata.as_ref().unwrap().size,
454            )
455            .unwrap();
456        let mut target = rows[0].metadata.clone().unwrap();
457        target.hashes.blake3 = "b".repeat(64);
458        fs::write(
459            paths.metadata_path(std::path::Path::new("data.txt")),
460            serde_json::to_vec(&target).unwrap(),
461        )
462        .unwrap();
463        let stale = inspect_status(&paths, None, None, &OperationContext::new())
464            .unwrap()
465            .files;
466        assert_eq!(stale[0].status, Some(Status::Stale));
467        fs::write(paths.repo_root().join("data.txt"), b"user edits").unwrap();
468        let modified = inspect_status(&paths, None, None, &OperationContext::new())
469            .unwrap()
470            .files;
471        assert_eq!(modified[0].status, Some(Status::Modified));
472    }
473
474    #[test]
475    fn status_returns_empty_vec_for_repo_with_no_tracked_files() {
476        let (_tmp, root) = create_temp_git_repo();
477        let (config, _dvs_dir) = init_dvs_repo(&root);
478        let paths =
479            DvsPaths::new(root.clone(), root.clone(), config.metadata_folder_name()).unwrap();
480
481        let statuses = inspect_status(&paths, None, None, &OperationContext::new())
482            .unwrap()
483            .files;
484        assert!(statuses.is_empty());
485    }
486
487    /// Helper to set up a repo with files at various directory depths.
488    /// Returns (TempDir, DvsPaths) with tracked files:
489    ///   "a.txt", "dir1/b.txt", "dir1/sub/c.txt", "dir2/d.txt"
490    fn setup_filtered_repo() -> (tempfile::TempDir, DvsPaths) {
491        let (tmp, root) = create_temp_git_repo();
492        let (config, _dvs_dir) = init_dvs_repo(&root);
493        let backend = config.backend();
494        let paths =
495            DvsPaths::new(root.clone(), root.clone(), config.metadata_folder_name()).unwrap();
496
497        for name in ["a.txt", "dir1/b.txt", "dir1/sub/c.txt", "dir2/d.txt"] {
498            let file_path = create_file(&root, name, name.as_bytes());
499            let mut metadata =
500                FileMetadata::from_file(&file_path, Compression::Zstd, None).unwrap();
501            metadata
502                .save(Uuid::new_v4(), &file_path, backend, &paths, name, None)
503                .unwrap();
504        }
505        (tmp, paths)
506    }
507
508    #[test]
509    fn status_with_filter() {
510        let (_tmp, paths) = setup_filtered_repo();
511
512        // Relative path filter
513        let filter = PathFilter {
514            paths: vec![PathBuf::from("dir1")],
515            recursive: false,
516        };
517        let statuses = inspect_status(&paths, Some(&filter), None, &OperationContext::new())
518            .unwrap()
519            .files;
520        assert_eq!(statuses.len(), 1);
521        assert_eq!(statuses[0].path, PathBuf::from("dir1/b.txt"));
522
523        // Absolute path filter via from_user_paths
524        let abs_path = paths.repo_root().join("dir1/b.txt");
525        let filter = PathFilter::from_user_paths(vec![abs_path], false, &paths);
526        let statuses = inspect_status(&paths, Some(&filter), None, &OperationContext::new())
527            .unwrap()
528            .files;
529        assert_eq!(statuses.len(), 1);
530        assert_eq!(statuses[0].path, PathBuf::from("dir1/b.txt"));
531    }
532
533    #[test]
534    fn status_whole_repo_applies_glob() {
535        let (_tmp, paths) = setup_filtered_repo();
536
537        // No filter (whole repo) + glob: the glob still applies, matched against
538        // the full repo-relative path. `*.txt` (literal separator) hits only
539        // root-level files.
540        let statuses = inspect_status(&paths, None, Some("*.txt"), &OperationContext::new())
541            .unwrap()
542            .files;
543        assert_eq!(statuses.len(), 1);
544        assert_eq!(statuses[0].path, PathBuf::from("a.txt"));
545
546        // `**/*.txt` reaches every tracked file.
547        let statuses = inspect_status(&paths, None, Some("**/*.txt"), &OperationContext::new())
548            .unwrap()
549            .files;
550        assert_eq!(statuses.len(), 4);
551    }
552
553    #[test]
554    fn status_bare_glob_anchors_to_cwd_subdir() {
555        let (_tmp, paths) = setup_filtered_repo();
556        let root = paths.repo_root().to_path_buf();
557        // cwd = dir1: a bare glob must scope to the cwd (like add/get), not the
558        // repo root. Pre-fix it globbed the whole repo from any directory.
559        let sub =
560            DvsPaths::new(fs::canonicalize(root.join("dir1")).unwrap(), root, ".dvs").unwrap();
561
562        // Literal separator: only the direct child dir1/b.txt, not a.txt/dir2.
563        let got: Vec<_> = inspect_status(&sub, None, Some("*.txt"), &OperationContext::new())
564            .unwrap()
565            .files
566            .into_iter()
567            .map(|f| f.path)
568            .collect();
569        assert_eq!(got, vec![PathBuf::from("dir1/b.txt")]);
570
571        // Recursive glob reaches dir1/sub/c.txt, still scoped under dir1.
572        let got: Vec<_> = inspect_status(&sub, None, Some("**/*.txt"), &OperationContext::new())
573            .unwrap()
574            .files
575            .into_iter()
576            .map(|f| f.path)
577            .collect();
578        assert_eq!(
579            got,
580            vec![PathBuf::from("dir1/b.txt"), PathBuf::from("dir1/sub/c.txt")]
581        );
582    }
583
584    #[test]
585    fn observed_status_plans_the_bytes_to_hash() {
586        let (_temp, paths) = fixture();
587        let recorder = std::sync::Arc::new(crate::testutil::Recorder::default());
588        let ctx = OperationContext::new().with_observer(recorder.clone());
589        inspect_status(&paths, None, None, &ctx).unwrap();
590        // The extra stat pass, done only when observed, sizes the plan
591        assert_eq!(
592            recorder.seen(),
593            ["plan 1 8 0 0", "hash start", "row data.txt", "hash end"]
594        );
595    }
596}
dvs/files/status.rs

dvs/files/
status.rs
