dvs/files/
add.rs

1use std::path::{Path, PathBuf};
2use std::sync::Mutex;
3
4use anyhow::{Result, bail};
5use fs_err as fs;
6use rayon::prelude::*;
7use serde::{Deserialize, Serialize};
8use uuid::Uuid;
9
10use crate::backends::Backend;
11use crate::cache::{HashCache, try_open_cache};
12use crate::config::Compression;
13use crate::files::metadata::FileMetadata;
14use crate::gitignore::add_to_gitignore;
15use crate::paths::DvsPaths;
16use crate::progress::OnFileStart;
17use crate::utils::get_threadpool;
18use crate::{Outcome, cache};
19
20/// Result of adding a single file.
21#[derive(Debug, Clone, Serialize, Deserialize)]
22pub struct AddResult {
23    pub path: PathBuf,
24    #[serde(flatten)]
25    pub detail: AddDetail,
26}
27
28#[derive(Debug, Clone, Serialize, Deserialize)]
29#[serde(untagged)]
30pub enum AddDetail {
31    Success {
32        outcome: Outcome,
33        hash: String,
34        size: u64,
35        stored_size: u64,
36    },
37    Error {
38        error: String,
39    },
40}
41
42#[allow(clippy::too_many_arguments)]
43fn add_file(
44    relative_path: &Path,
45    paths: &DvsPaths,
46    backend: &dyn Backend,
47    cache: Option<&Mutex<HashCache>>,
48    operation_id: Uuid,
49    message: Option<String>,
50    compression: Compression,
51    dry_run: bool,
52    on_bytes: Option<&(dyn Fn(u64) + Send + Sync)>,
53) -> Result<(Outcome, FileMetadata, Option<u64>)> {
54    let full_path = paths.file_path(relative_path);
55    let rel_str = relative_path.to_string_lossy();
56    let (hashes, size) = cache::hashes_for_file(&full_path, &rel_str, cache)?;
57    let metadata = FileMetadata::from_hashes(hashes, size, compression, message);
58    if dry_run {
59        let dvs_file_path = paths.metadata_path(relative_path);
60        let dvs_file_exists = dvs_file_path.is_file();
61        let storage_exists = backend.exists(&metadata.hashes)?;
62        let outcome = if dvs_file_exists && storage_exists {
63            let existing: FileMetadata =
64                serde_json::from_reader(fs_err::File::open(&dvs_file_path)?)?;
65            if existing == metadata {
66                Outcome::Present
67            } else {
68                Outcome::Copied
69            }
70        } else {
71            Outcome::Copied
72        };
73        Ok((outcome, metadata, None))
74    } else {
75        let (outcome, stored_size) = metadata.save(
76            operation_id,
77            &full_path,
78            backend,
79            paths,
80            relative_path,
81            on_bytes,
82        )?;
83        Ok((outcome, metadata, stored_size))
84    }
85}
86
87/// Adds files matching a glob pattern to DVS.
88///
89/// The pattern is matched against files relative to cwd.
90/// Files are stored with paths relative to repo_root.
91pub fn add_files(
92    files: Vec<PathBuf>,
93    paths: &DvsPaths,
94    backend: &dyn Backend,
95    message: Option<String>,
96    compression: Compression,
97    dry_run: bool,
98    on_file_start: Option<&OnFileStart>,
99) -> Result<Vec<AddResult>> {
100    let matched_paths = paths.validate_for_add(&files);
101    if matched_paths.is_empty() {
102        return Ok(Vec::new());
103    }
104
105    // If any input is invalid we just bail instead of doing partial execution
106    let invalid = matched_paths
107        .iter()
108        .filter_map(|(path, status)| {
109            status
110                .reason()
111                .map(|reason| format!("  {}: {reason}", path.display()))
112        })
113        .collect::<Vec<_>>();
114    if !invalid.is_empty() {
115        bail!(
116            "Refusing to add, the following paths are invalid:\n{}",
117            invalid.join("\n")
118        );
119    }
120
121    let valid_paths = matched_paths
122        .into_iter()
123        .map(|(path, _)| path)
124        .collect::<Vec<_>>();
125
126    let pool = get_threadpool(valid_paths.len())?;
127    let cache = try_open_cache(paths);
128    let operation_id = Uuid::new_v4();
129
130    let mut results: Vec<AddResult> = pool.install(|| {
131        valid_paths
132            .into_par_iter()
133            .map(|relative_path| {
134                let full_path = paths.file_path(&relative_path);
135                let file_size = fs::metadata(&full_path).map(|m| m.len()).unwrap_or(0);
136                let file_progress = on_file_start.map(|f| f(&relative_path, file_size));
137                let on_bytes = file_progress.as_ref().map(|fp| &*fp.on_bytes);
138                let result = match add_file(
139                    &relative_path,
140                    paths,
141                    backend,
142                    cache.as_ref(),
143                    operation_id,
144                    message.clone(),
145                    compression,
146                    dry_run,
147                    on_bytes,
148                ) {
149                    Ok((outcome, metadata, stored_size)) => {
150                        log::info!(
151                            "Successfully added {} ({:?})",
152                            relative_path.display(),
153                            outcome
154                        );
155                        let stored_size = stored_size.unwrap_or(metadata.size);
156                        AddResult {
157                            path: relative_path,
158                            detail: AddDetail::Success {
159                                outcome,
160                                hash: metadata.hashes.blake3,
161                                size: metadata.size,
162                                stored_size,
163                            },
164                        }
165                    }
166                    Err(e) => {
167                        log::warn!("Failed to add {}: {e}", relative_path.display());
168                        AddResult {
169                            path: relative_path,
170                            detail: AddDetail::Error {
171                                error: e.to_string(),
172                            },
173                        }
174                    }
175                };
176                if let Some(fp) = &file_progress {
177                    (fp.on_done)(matches!(result.detail, AddDetail::Success { .. }));
178                }
179                result
180            })
181            .collect()
182    });
183    results.sort_by(|a, b| a.path.cmp(&b.path));
184
185    let successful_paths: Vec<_> = results
186        .iter()
187        .filter(|r| matches!(r.detail, AddDetail::Success { .. }))
188        .map(|r| r.path.clone())
189        .collect();
190    if !dry_run && !successful_paths.is_empty() {
191        if let Err(e) = add_to_gitignore(paths.repo_root(), &successful_paths) {
192            log::warn!("Failed to update .gitignore: {e}");
193        }
194    }
195
196    Ok(results)
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202    use crate::progress::FileProgress;
203    use crate::testutil::{create_file, create_temp_git_repo, init_dvs_repo};
204    use fs_err as fs;
205    use std::path::Path;
206    use std::sync::Arc;
207    use std::sync::atomic::{AtomicUsize, Ordering};
208
209    fn make_paths(root: &Path, config: &crate::config::Config) -> DvsPaths {
210        DvsPaths::new(
211            root.to_path_buf(),
212            root.to_path_buf(),
213            config.metadata_folder_name(),
214        )
215        .unwrap()
216    }
217
218    #[test]
219    fn add_files_aborts_batch_when_any_file_missing() {
220        let (_tmp, root) = create_temp_git_repo();
221        let (config, _dvs_dir) = init_dvs_repo(&root);
222        let backend = config.backend();
223        let paths = make_paths(&root, &config);
224
225        // One valid file alongside a missing one.
226        create_file(&root, "a.txt", b"a");
227
228        let err = add_files(
229            vec!["a.txt".into(), "missing.csv".into()],
230            &paths,
231            backend,
232            None,
233            Compression::Zstd,
234            false,
235            None,
236        )
237        .unwrap_err()
238        .to_string();
239        assert!(err.contains("missing.csv"), "unexpected error: {err}");
240
241        // No partial success: the valid file must not have been added.
242        assert!(
243            !paths.metadata_path(Path::new("a.txt")).exists(),
244            "no files should be added when one is missing"
245        );
246    }
247
248    #[test]
249    fn add_files_aborts_when_path_outside_or_directory() {
250        let (_tmp, root) = create_temp_git_repo();
251        let (config, _dvs_dir) = init_dvs_repo(&root);
252        let backend = config.backend();
253        let paths = make_paths(&root, &config);
254
255        // Valid file
256        create_file(&root, "a.txt", b"a");
257
258        // Outside-project file: a sibling of the repo, outside its root.
259        let outside_file = root.parent().unwrap().join("outside.txt");
260        fs::write(&outside_file, b"outside").unwrap();
261        let outside_relative = PathBuf::from("..").join("outside.txt");
262
263        // Directory inside the repo
264        fs::create_dir(root.join("subdir")).unwrap();
265
266        // An outside-project path or a directory is invalid up front, so the
267        // whole batch is refused rather than partially added.
268        let err = add_files(
269            vec!["a.txt".into(), outside_relative, "subdir".into()],
270            &paths,
271            backend,
272            None,
273            Compression::Zstd,
274            false,
275            None,
276        )
277        .unwrap_err()
278        .to_string();
279        assert!(err.contains("outside project"), "unexpected error: {err}");
280        assert!(err.contains("directory"), "unexpected error: {err}");
281
282        // No partial success: the valid file must not have been added.
283        assert!(
284            !paths.metadata_path(Path::new("a.txt")).exists(),
285            "no files should be added when any path is invalid"
286        );
287    }
288
289    #[test]
290    fn add_files_invokes_on_done_for_each_file() {
291        let (_tmp, root) = create_temp_git_repo();
292        let (config, _dvs_dir) = init_dvs_repo(&root);
293        let backend = config.backend();
294        let paths = make_paths(&root, &config);
295
296        create_file(&root, "a.txt", b"a");
297        create_file(&root, "b.txt", b"b");
298        create_file(&root, "c.txt", b"c");
299
300        let done_calls = Arc::new(AtomicUsize::new(0));
301        let success_calls = Arc::new(AtomicUsize::new(0));
302        let on_file_start = {
303            let done_calls = Arc::clone(&done_calls);
304            let success_calls = Arc::clone(&success_calls);
305            move |_path: &Path, _size: u64| {
306                let done_calls = Arc::clone(&done_calls);
307                let success_calls = Arc::clone(&success_calls);
308                FileProgress {
309                    on_bytes: Box::new(|_| {}),
310                    on_done: Box::new(move |success| {
311                        done_calls.fetch_add(1, Ordering::SeqCst);
312                        if success {
313                            success_calls.fetch_add(1, Ordering::SeqCst);
314                        }
315                    }),
316                }
317            }
318        };
319
320        let results = add_files(
321            vec!["a.txt".into(), "b.txt".into(), "c.txt".into()],
322            &paths,
323            backend,
324            None,
325            Compression::Zstd,
326            false,
327            Some(&on_file_start),
328        )
329        .unwrap();
330
331        assert_eq!(results.len(), 3);
332        assert!(
333            results
334                .iter()
335                .all(|r| matches!(r.detail, AddDetail::Success { .. }))
336        );
337        assert_eq!(done_calls.load(Ordering::SeqCst), 3);
338        assert_eq!(success_calls.load(Ordering::SeqCst), 3);
339    }
340}
dvs/files/add.rs

dvs/files/
add.rs
