file_engine/operations/
copy.rs

1use std::path::PathBuf;
2use std::time::Instant;
3
4use tokio_util::sync::CancellationToken;
5
6use crate::error::Result;
7use crate::handle::Handle;
8use crate::planner::{BatchConfig, ErrorStrategy, OperationOutcome, SortOrder};
9use crate::profiler::DEFAULT_SMALL_FILE_THRESHOLD;
10use crate::progress::ProgressReporter;
11
12use super::default_concurrency;
13use super::pipeline::run_copy_pipeline;
14
15pub struct CopyBuilder {
16    source: PathBuf,
17    dest: PathBuf,
18    overwrite: bool,
19    skip_if_identical: bool,
20    /// Unconditional field even though only `.preserve_permissions()`
21    /// (Unix-only) can ever set it.
22    preserve_permissions: bool,
23    allow_filesystem_integrity_risk: bool,
24    small_file_threshold: Option<u64>,
25    batch_config: BatchConfig,
26    concurrency: Option<usize>,
27}
28
29impl CopyBuilder {
30    pub(crate) fn new(source: impl Into<PathBuf>, dest: impl Into<PathBuf>) -> Self {
31        Self {
32            source: source.into(),
33            dest: dest.into(),
34            overwrite: false,
35            skip_if_identical: false,
36            preserve_permissions: false,
37            allow_filesystem_integrity_risk: false,
38            small_file_threshold: None,
39            batch_config: BatchConfig::default(),
40            concurrency: None,
41        }
42    }
43
44    pub fn overwrite(mut self, overwrite: bool) -> Self {
45        self.overwrite = overwrite;
46        self
47    }
48
49    /// Only consulted when `.overwrite(false)` (the default) *and* the
50    /// destination already exists: instead of failing with
51    /// `Error::DestExists`, compares content (size first, then a blake3
52    /// hash of both files — see `checksum::files_identical`) and leaves
53    /// an already-identical destination alone rather than re-copying it
54    /// or failing on it. A destination that exists but differs still
55    /// fails with `Error::DestExists` exactly as without this — a
56    /// library has no way to interactively ask whether to replace it;
57    /// that decision belongs to whatever's built on top of this crate,
58    /// which can catch `DestExists` and retry with `.overwrite(true)` if
59    /// the user says yes. Requires the `checksum` feature.
60    #[cfg(feature = "checksum")]
61    pub fn skip_if_identical(mut self, skip: bool) -> Self {
62        self.skip_if_identical = skip;
63        self
64    }
65
66    #[cfg(all(unix, feature = "permissions"))]
67    pub fn preserve_permissions(mut self, preserve: bool) -> Self {
68        self.preserve_permissions = preserve;
69        self
70    }
71
72    /// Proceed even when the destination filesystem has a known
73    /// write-integrity risk on this platform (currently: exFAT on
74    /// macOS) — without this, `.start()`'s `Handle` resolves to
75    /// `Err(Error::FilesystemIntegrityRisk)` before any data is written.
76    /// See `docs/guide/filesystem-safety.md`.
77    pub fn allow_filesystem_integrity_risk(mut self, allow: bool) -> Self {
78        self.allow_filesystem_integrity_risk = allow;
79        self
80    }
81
82    pub fn small_file_threshold(mut self, bytes: u64) -> Self {
83        self.small_file_threshold = Some(bytes);
84        self
85    }
86
87    pub fn max_bytes_per_batch(mut self, bytes: u64) -> Self {
88        self.batch_config.max_bytes_per_batch = bytes;
89        self
90    }
91
92    pub fn max_files_per_batch(mut self, n: usize) -> Self {
93        self.batch_config.max_files_per_batch = Some(n);
94        self
95    }
96
97    pub fn batch_sort_order(mut self, order: SortOrder) -> Self {
98        self.batch_config.sort_order = order;
99        self
100    }
101
102    pub fn on_error(mut self, strategy: ErrorStrategy) -> Self {
103        self.batch_config.error_strategy = strategy;
104        self
105    }
106
107    pub fn batch_concurrency(mut self, n: usize) -> Self {
108        self.concurrency = Some(n);
109        self
110    }
111
112    pub fn start(self) -> Result<Handle<OperationOutcome>> {
113        let cancel = CancellationToken::new();
114        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
115        let reporter = ProgressReporter::new(tx);
116
117        let concurrency = self.concurrency.unwrap_or_else(default_concurrency);
118        let threshold = self
119            .small_file_threshold
120            .unwrap_or(DEFAULT_SMALL_FILE_THRESHOLD);
121        let cancel_for_task = cancel.clone();
122
123        let join_handle = tokio::spawn(async move {
124            let started = Instant::now();
125            let mut outcome = run_copy_pipeline(
126                &self.source,
127                &self.dest,
128                self.overwrite,
129                self.skip_if_identical,
130                self.preserve_permissions,
131                self.allow_filesystem_integrity_risk,
132                threshold,
133                &self.batch_config,
134                concurrency,
135                cancel_for_task,
136                reporter,
137            )
138            .await?;
139            outcome.duration = started.elapsed();
140            Ok(outcome)
141        });
142
143        Ok(Handle::new(join_handle, rx, cancel))
144    }
145}
file_engine/operations/copy.rs

file_engine/operations/
copy.rs
