dvs/
config.rs

1use std::io;
2use std::io::Write;
3use std::path::Path;
4
5use crate::backends::Backend;
6use crate::backends::local::LocalBackend;
7use crate::backends::server::ServerBackend;
8use crate::paths::{CONFIG_FILE_NAME, DEFAULT_FOLDER_NAME, find_repo_root};
9use crate::progress::ProgressReader;
10use crate::utils::parse_size;
11use anyhow::Result;
12use fs_err as fs;
13use serde::{Deserialize, Deserializer, Serialize};
14use url::Url;
15
16const DEFAULT_PROGRESS_BYTE_SIZE_THRESHOLD: u64 = 524_288_000;
17
18fn deserialize_size_option<'de, D: Deserializer<'de>>(
19    deserializer: D,
20) -> Result<Option<u64>, D::Error> {
21    let s: Option<String> = Option::deserialize(deserializer)?;
22    match s {
23        None => Ok(None),
24        Some(s) => parse_size(&s).map(Some).map_err(serde::de::Error::custom),
25    }
26}
27
28#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Default)]
29#[serde(rename_all = "lowercase")]
30pub enum Compression {
31    None,
32    #[default]
33    Zstd,
34}
35
36impl std::fmt::Display for Compression {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        match self {
39            Compression::None => write!(f, "none"),
40            Compression::Zstd => write!(f, "zstd"),
41        }
42    }
43}
44
45impl Compression {
46    pub fn compress(
47        &self,
48        source: &Path,
49        dest: &Path,
50        on_bytes: Option<&(dyn Fn(u64) + Send + Sync)>,
51    ) -> Result<u64> {
52        match self {
53            Compression::None => {
54                if let Some(cb) = on_bytes {
55                    let input = fs::File::open(source)?;
56                    let output = fs::File::create(dest)?;
57                    let mut reader = ProgressReader::new(input, cb);
58                    let mut writer = io::BufWriter::new(output);
59                    let bytes = io::copy(&mut reader, &mut writer)?;
60                    writer.flush()?;
61                    Ok(bytes)
62                } else {
63                    let bytes = fs::copy(source, dest)?;
64                    Ok(bytes)
65                }
66            }
67            Compression::Zstd => {
68                let input = fs::File::open(source)?;
69                let output = fs::File::create(dest)?;
70
71                if let Some(cb) = on_bytes {
72                    let tracked = ProgressReader::new(input, cb);
73                    let mut encoder = zstd::stream::read::Encoder::new(tracked, 0)?;
74                    let mut writer = io::BufWriter::new(output);
75                    let bytes = io::copy(&mut encoder, &mut writer)?;
76                    writer.flush()?;
77                    Ok(bytes)
78                } else {
79                    let mut encoder = zstd::stream::read::Encoder::new(input, 0)?;
80                    let mut writer = io::BufWriter::new(output);
81                    let bytes = io::copy(&mut encoder, &mut writer)?;
82                    writer.flush()?;
83                    Ok(bytes)
84                }
85            }
86        }
87    }
88
89    pub fn decompress(
90        &self,
91        source: &Path,
92        dest: &Path,
93        on_bytes: Option<&(dyn Fn(u64) + Send + Sync)>,
94    ) -> Result<()> {
95        match self {
96            Compression::None => {
97                if let Some(cb) = on_bytes {
98                    let input = fs::File::open(source)?;
99                    let output = fs::File::create(dest)?;
100                    let mut reader = ProgressReader::new(input, cb);
101                    let mut writer = io::BufWriter::new(output);
102                    io::copy(&mut reader, &mut writer)?;
103                    writer.flush()?;
104                } else {
105                    let mut input = fs::File::open(source)?;
106                    let mut output = io::BufWriter::new(fs::File::create(dest)?);
107                    io::copy(&mut input, &mut output)?;
108                    output.flush()?;
109                }
110                Ok(())
111            }
112            Compression::Zstd => {
113                let input = fs::File::open(source)?;
114                let output = fs::File::create(dest)?;
115
116                let mut decoder = zstd::stream::read::Decoder::new(input)?;
117                let mut writer = io::BufWriter::new(output);
118                if let Some(cb) = on_bytes {
119                    let mut reader = ProgressReader::new(&mut decoder, cb);
120                    io::copy(&mut reader, &mut writer)?;
121                } else {
122                    io::copy(&mut decoder, &mut writer)?;
123                }
124                writer.flush()?;
125                Ok(())
126            }
127        }
128    }
129}
130
131#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
132pub struct CliConfig {
133    /// Defaults to 500MB if not set in the config file
134    #[serde(
135        default,
136        skip_serializing_if = "Option::is_none",
137        deserialize_with = "deserialize_size_option"
138    )]
139    progress_threshold: Option<u64>,
140}
141
142/// What `get` does with a working file that differs from its target and that
143/// dvs cannot show it left there itself (local changes, or no baseline).
144#[derive(Debug, Serialize, Deserialize, Clone, Copy, PartialEq, Eq, Default)]
145#[serde(rename_all = "lowercase")]
146pub enum OnConflict {
147    /// Change no files if any requested file has a conflict or another known problem
148    #[default]
149    Abort,
150    /// Leave conflicting files as they are and get the rest
151    Skip,
152    /// Replace conflicting files; their local changes are lost
153    Overwrite,
154}
155
156impl OnConflict {
157    pub const CHOICES: [&'static str; 3] = ["abort", "skip", "overwrite"];
158
159    pub fn as_str(&self) -> &'static str {
160        match self {
161            OnConflict::Abort => "abort",
162            OnConflict::Skip => "skip",
163            OnConflict::Overwrite => "overwrite",
164        }
165    }
166}
167
168impl std::fmt::Display for OnConflict {
169    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
170        f.write_str(self.as_str())
171    }
172}
173
174impl std::str::FromStr for OnConflict {
175    type Err = String;
176
177    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
178        match s {
179            "abort" => Ok(OnConflict::Abort),
180            "skip" => Ok(OnConflict::Skip),
181            "overwrite" => Ok(OnConflict::Overwrite),
182            other => Err(format!(
183                "invalid on_conflict value {other:?}: expected one of abort, skip, overwrite"
184            )),
185        }
186    }
187}
188
189/// Where an effective setting came from, reported with every result.
190#[derive(Debug, Serialize, Clone, Copy, PartialEq, Eq)]
191#[serde(rename_all = "lowercase")]
192pub enum SettingSource {
193    Argument,
194    Env,
195    Config,
196    Default,
197}
198
199impl SettingSource {
200    /// The stable name used in JSON and R results.
201    pub fn as_str(&self) -> &'static str {
202        match self {
203            SettingSource::Argument => "argument",
204            SettingSource::Env => "env",
205            SettingSource::Config => "config",
206            SettingSource::Default => "default",
207        }
208    }
209}
210
211/// For people: names where the setting was made rather than "config".
212impl std::fmt::Display for SettingSource {
213    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
214        f.write_str(match self {
215            SettingSource::Argument => "command line",
216            SettingSource::Env => GET_ON_CONFLICT_ENV,
217            SettingSource::Config => "dvs.toml",
218            SettingSource::Default => "default",
219        })
220    }
221}
222
223/// A problem with dvs.toml or a DVS_* environment variable. Frontends report it
224/// distinctly (the CLI exits 2) because no operation was attempted.
225#[derive(Debug)]
226pub struct ConfigError(pub String);
227
228impl std::fmt::Display for ConfigError {
229    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
230        f.write_str(&self.0)
231    }
232}
233
234impl std::error::Error for ConfigError {}
235
236pub const GET_ON_CONFLICT_ENV: &str = "DVS_GET_ON_CONFLICT";
237
238#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Default)]
239#[serde(deny_unknown_fields)]
240pub struct GetConfig {
241    #[serde(default, skip_serializing_if = "Option::is_none")]
242    on_conflict: Option<OnConflict>,
243}
244
245#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
246pub struct Config {
247    /// Compression algorithm to use for files in the storage directory
248    #[serde(default, skip_serializing_if = "Option::is_none")]
249    compression: Option<Compression>,
250    /// By default, all the metadata files (the .dvs files) will be stored in a `.dvs` folder
251    /// at the root of the repository
252    /// If this option is set, dvs will use that folder name instead of `.dvs`
253    metadata_folder_name: Option<String>,
254    pub(crate) backend: Backend,
255    #[serde(default, skip_serializing_if = "Option::is_none")]
256    cli: Option<CliConfig>,
257    #[serde(default, skip_serializing_if = "Option::is_none")]
258    get: Option<GetConfig>,
259}
260
261impl Config {
262    pub fn new_local(path: impl AsRef<Path>, group: Option<String>) -> Result<Config> {
263        let backend = LocalBackend::new(path.as_ref(), group)?;
264        Ok(Config {
265            compression: Some(Compression::Zstd),
266            metadata_folder_name: None,
267            backend: Backend::Local(backend),
268            cli: None,
269            get: None,
270        })
271    }
272
273    pub fn new_server(name: String, url: Url, group: Option<String>) -> Config {
274        let backend = ServerBackend::new(name, group, url);
275        Config {
276            compression: Some(Compression::Zstd),
277            metadata_folder_name: None,
278            backend: Backend::Server(backend),
279            cli: None,
280            get: None,
281        }
282    }
283
284    pub fn save(&self, directory: impl AsRef<Path>) -> Result<()> {
285        let config_path = directory.as_ref().join(CONFIG_FILE_NAME);
286        let content = toml::to_string_pretty(&self)?;
287        fs::write(&config_path, content)?;
288        log::info!("Configuration saved to {}", config_path.display());
289        Ok(())
290    }
291
292    pub fn find(current_directory: impl AsRef<Path>) -> Option<Result<Self>> {
293        let repo_root = find_repo_root(current_directory);
294        let config_path = repo_root.join(CONFIG_FILE_NAME);
295        log::debug!("Looking for config at {}", config_path.display());
296        if config_path.exists() {
297            let content = match fs::read_to_string(&config_path) {
298                Ok(c) => c,
299                Err(e) => return Some(Err(e.into())),
300            };
301            Some(toml::from_str(&content).map_err(|e| {
302                ConfigError(format!("Failed to parse {}: {e}", config_path.display())).into()
303            }))
304        } else {
305            log::debug!("No config file found at {}", config_path.display());
306            None
307        }
308    }
309
310    pub fn set_metadata_folder_name(&mut self, name: String) {
311        self.metadata_folder_name = Some(name);
312    }
313
314    pub fn metadata_folder_name(&self) -> &str {
315        if let Some(name) = &self.metadata_folder_name {
316            name.as_str()
317        } else {
318            DEFAULT_FOLDER_NAME
319        }
320    }
321
322    pub fn compression(&self) -> Compression {
323        self.compression.unwrap_or_default()
324    }
325
326    pub fn set_compression(&mut self, compression: Option<Compression>) {
327        self.compression = compression;
328    }
329
330    pub fn backend(&self) -> &Backend {
331        &self.backend
332    }
333
334    pub fn server_url(&self) -> Option<&Url> {
335        match &self.backend {
336            Backend::Local(_) => None,
337            Backend::Server(s) => Some(&s.url),
338        }
339    }
340
341    /// The `on_conflict` policy for `get`: an explicit argument, else
342    /// `DVS_GET_ON_CONFLICT`, else `[get] on_conflict` in dvs.toml, else abort.
343    pub fn get_on_conflict(
344        &self,
345        explicit: Option<OnConflict>,
346    ) -> std::result::Result<(OnConflict, SettingSource), ConfigError> {
347        resolve_on_conflict(
348            explicit,
349            std::env::var(GET_ON_CONFLICT_ENV).ok().as_deref(),
350            self.get.as_ref().and_then(|g| g.on_conflict),
351        )
352    }
353
354    pub fn progress_bytes_threshold(&self) -> u64 {
355        self.cli
356            .as_ref()
357            .and_then(|x| x.progress_threshold)
358            .unwrap_or(DEFAULT_PROGRESS_BYTE_SIZE_THRESHOLD)
359    }
360}
361
362#[cfg(test)]
363mod tests {
364    use super::*;
365    use crate::testutil::create_temp_git_repo;
366
367    #[test]
368    fn config_save_and_find_roundtrip() {
369        let (_tmp, root) = create_temp_git_repo();
370        let storage = root.join(".storage");
371
372        let original = Config::new_local(&storage, None).unwrap();
373        original.save(&root).unwrap();
374
375        let loaded = Config::find(&root).unwrap().unwrap();
376        assert_eq!(original, loaded);
377    }
378
379    #[test]
380    fn config_find_returns_none_without_config_file() {
381        let (_tmp, root) = create_temp_git_repo();
382        assert!(Config::find(&root).is_none());
383    }
384
385    #[cfg(unix)]
386    #[test]
387    fn new_local_validates_group_exists() {
388        let tmp = tempfile::tempdir().unwrap();
389        let storage = tmp.path().join(".storage");
390
391        // Non-existent group should fail
392        let result = Config::new_local(&storage, Some("nonexistent_group_12345".to_string()));
393        assert!(result.is_err());
394        assert!(result.unwrap_err().to_string().contains("not found"));
395    }
396
397    #[test]
398    fn config_with_custom_metadata_folder() {
399        let (_tmp, root) = create_temp_git_repo();
400        let storage = root.join(".storage");
401
402        let mut config = Config::new_local(&storage, None).unwrap();
403        config.set_metadata_folder_name(".custom_dvs".to_string());
404        config.save(&root).unwrap();
405
406        let loaded = Config::find(&root).unwrap().unwrap();
407        assert_eq!(loaded.metadata_folder_name(), ".custom_dvs");
408    }
409}
410
411/// Precedence for `on_conflict`, validating every layer whichever wins, so a
412/// bad setting is reported to everyone, not only to those who don't override it.
413///
414/// dvs.toml is shared by everyone working on the project, so it may not say
415/// `overwrite`: that would discard collaborators' unsaved work without them
416/// asking. It is a configuration error, never silently replaced.
417fn resolve_on_conflict(
418    explicit: Option<OnConflict>,
419    env: Option<&str>,
420    project: Option<OnConflict>,
421) -> std::result::Result<(OnConflict, SettingSource), ConfigError> {
422    let env = env
423        .map(|v| {
424            v.parse::<OnConflict>()
425                .map_err(|e| ConfigError(format!("{GET_ON_CONFLICT_ENV}: {e}")))
426        })
427        .transpose()?;
428    if project == Some(OnConflict::Overwrite) {
429        return Err(ConfigError(format!(
430            "{CONFIG_FILE_NAME}: [get] on_conflict = \"overwrite\" is not allowed in project \
431             configuration, because it would discard collaborators' local changes. Use \
432             \"abort\" or \"skip\" here; pass --force (or on_conflict = \"overwrite\" in R, \
433             or {GET_ON_CONFLICT_ENV}=overwrite) when you mean it."
434        )));
435    }
436    Ok(explicit
437        .map(|p| (p, SettingSource::Argument))
438        .or(env.map(|p| (p, SettingSource::Env)))
439        .or(project.map(|p| (p, SettingSource::Config)))
440        .unwrap_or((OnConflict::default(), SettingSource::Default)))
441}
442
443#[cfg(test)]
444mod get_config_tests {
445    use super::*;
446    use crate::testutil::create_temp_git_repo;
447    use OnConflict::*;
448
449    fn config_with(get_section: &str) -> std::result::Result<Config, String> {
450        let (_tmp, root) = create_temp_git_repo();
451        let storage = root.parent().unwrap().join(".storage");
452        Config::new_local(&storage, None)
453            .unwrap()
454            .save(&root)
455            .unwrap();
456        let path = root.join(CONFIG_FILE_NAME);
457        let mut text = fs::read_to_string(&path).unwrap();
458        text.push_str(get_section);
459        fs::write(&path, text).unwrap();
460        Config::find(&root).unwrap().map_err(|e| e.to_string())
461    }
462
463    #[test]
464    fn precedence_is_argument_env_project_default() {
465        use SettingSource as S;
466        let r = resolve_on_conflict;
467        assert_eq!(
468            r(Some(Overwrite), Some("skip"), Some(Skip)).unwrap(),
469            (Overwrite, S::Argument)
470        );
471        assert_eq!(
472            r(None, Some("overwrite"), Some(Skip)).unwrap(),
473            (Overwrite, S::Env)
474        );
475        assert_eq!(r(None, None, Some(Skip)).unwrap(), (Skip, S::Config));
476        assert_eq!(r(None, None, None).unwrap(), (Abort, S::Default));
477    }
478
479    #[test]
480    fn bad_settings_are_errors_even_when_overridden() {
481        let err = resolve_on_conflict(Some(Skip), Some("sometimes"), None).unwrap_err();
482        assert!(err.0.contains(GET_ON_CONFLICT_ENV), "{err}");
483        let err = resolve_on_conflict(Some(Abort), Some("overwrite"), Some(Overwrite)).unwrap_err();
484        assert!(err.0.contains("not allowed"), "{err}");
485    }
486
487    #[test]
488    fn dvs_toml_get_section_is_read_and_checked() {
489        let config = config_with("\n[get]\non_conflict = \"skip\"\n").unwrap();
490        assert_eq!(config.get.unwrap().on_conflict, Some(Skip));
491        let err = config_with("\n[get]\non_conflicts = \"skip\"\n").unwrap_err();
492        assert!(err.contains("unknown field"), "{err}");
493        let err = config_with("\n[get]\non_conflict = \"sometimes\"\n").unwrap_err();
494        assert!(err.contains("sometimes"), "{err}");
495    }
496}
dvs/config.rs

dvs/
config.rs
