Skip to main content

file_engine/operations/
move_path.rs

1use std::collections::HashSet;
2use std::io;
3use std::path::{Path, PathBuf};
4use std::time::Instant;
5
6use tokio_util::sync::CancellationToken;
7
8use crate::error::{classify_io_error, Error, Result};
9use crate::planner::{
10    BatchConfig, CopyAction, EntryAction, ErrorStrategy, OperationOutcome, StopReason,
11};
12use crate::profiler::{Entry, DEFAULT_SMALL_FILE_THRESHOLD};
13use crate::progress::{Progress, ProgressReporter};
14
15use super::default_concurrency;
16use super::pipeline::run_copy_pipeline;
17
18pub struct MoveBuilder {
19    source: PathBuf,
20    dest: PathBuf,
21    overwrite: bool,
22    skip_if_identical: bool,
23    preserve_permissions: bool,
24    allow_filesystem_integrity_risk: bool,
25    small_file_threshold: Option<u64>,
26    batch_config: BatchConfig,
27    concurrency: Option<usize>,
28}
29
30impl MoveBuilder {
31    pub(crate) fn new(source: impl Into<PathBuf>, dest: impl Into<PathBuf>) -> Self {
32        Self {
33            source: source.into(),
34            dest: dest.into(),
35            overwrite: false,
36            skip_if_identical: false,
37            preserve_permissions: false,
38            allow_filesystem_integrity_risk: false,
39            small_file_threshold: None,
40            batch_config: BatchConfig::default(),
41            concurrency: None,
42        }
43    }
44
45    pub fn overwrite(mut self, overwrite: bool) -> Self {
46        self.overwrite = overwrite;
47        self
48    }
49
50    /// Only consulted when `.overwrite(false)` (the default) *and* the
51    /// destination already exists: instead of failing with
52    /// `Error::DestExists`, compares content and leaves an already-
53    /// identical destination alone — the source is still removed, since
54    /// that's still what "moved" means, it just skips redundantly
55    /// rewriting a destination that already matches. A differing
56    /// destination still fails exactly as without this. See
57    /// `CopyBuilder::skip_if_identical` for the full rationale; applies
58    /// equally to this builder's atomic-rename fast path and its
59    /// cross-device fallback.
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    /// Only meaningful for the cross-device fallback (which reuses
67    /// `CopyAction`) — the atomic-rename fast path already preserves
68    /// everything about the source, permissions included, for free.
69    #[cfg(all(unix, feature = "permissions"))]
70    pub fn preserve_permissions(mut self, preserve: bool) -> Self {
71        self.preserve_permissions = preserve;
72        self
73    }
74
75    /// Only meaningful for the cross-device fallback, for the same
76    /// reason `.preserve_permissions()` is — the atomic-rename fast path
77    /// never touches `dest`'s filesystem capabilities at all. See
78    /// `CopyBuilder::allow_filesystem_integrity_risk`.
79    pub fn allow_filesystem_integrity_risk(mut self, allow: bool) -> Self {
80        self.allow_filesystem_integrity_risk = allow;
81        self
82    }
83
84    pub fn small_file_threshold(mut self, bytes: u64) -> Self {
85        self.small_file_threshold = Some(bytes);
86        self
87    }
88
89    pub fn on_error(mut self, strategy: ErrorStrategy) -> Self {
90        self.batch_config.error_strategy = strategy;
91        self
92    }
93
94    pub fn batch_concurrency(mut self, n: usize) -> Self {
95        self.concurrency = Some(n);
96        self
97    }
98
99    pub fn start(self) -> Result<crate::handle::Handle<OperationOutcome>> {
100        let cancel = CancellationToken::new();
101        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
102        let reporter = ProgressReporter::new(tx);
103
104        let concurrency = self.concurrency.unwrap_or_else(default_concurrency);
105        let threshold = self
106            .small_file_threshold
107            .unwrap_or(DEFAULT_SMALL_FILE_THRESHOLD);
108        let cancel_for_task = cancel.clone();
109
110        let join_handle = tokio::spawn(async move {
111            let started = Instant::now();
112            let mut outcome = move_path(
113                &self.source,
114                &self.dest,
115                self.overwrite,
116                self.skip_if_identical,
117                self.preserve_permissions,
118                self.allow_filesystem_integrity_risk,
119                threshold,
120                &self.batch_config,
121                concurrency,
122                cancel_for_task,
123                reporter,
124                &TokioRenamer,
125            )
126            .await?;
127            outcome.duration = started.elapsed();
128            Ok(outcome)
129        });
130
131        Ok(crate::handle::Handle::new(join_handle, rx, cancel))
132    }
133}
134
135/// Pure classification, unit-testable with synthetic `io::Error` values —
136/// no real cross-device filesystem needed.
137pub(crate) fn is_cross_device(err: &io::Error) -> bool {
138    err.kind() == io::ErrorKind::CrossesDevices
139}
140
141/// Injectable rename seam: production code uses `TokioRenamer`, tests
142/// inject a fake that deterministically returns a synthetic cross-device
143/// error to exercise the fallback wiring without a second filesystem.
144pub(crate) trait Renamer {
145    async fn rename(&self, source: &Path, dest: &Path) -> io::Result<()>;
146}
147
148pub(crate) struct TokioRenamer;
149
150impl Renamer for TokioRenamer {
151    async fn rename(&self, source: &Path, dest: &Path) -> io::Result<()> {
152        tokio::fs::rename(source, dest).await
153    }
154}
155
156/// 1. If `dest` already exists and `source` is a single file, resolve
157///    the conflict up front (see `resolve_existing_dest_conflict`) —
158///    `rename(2)` would otherwise silently replace it regardless of
159///    `overwrite`. Directory sources are left alone here: `dest`
160///    legitimately pre-exists as the directory being moved *into*
161///    (matching `CopyAction`'s `dest_root.join(relative_path)`
162///    placement, which mirrors contents into an existing directory
163///    rather than nesting a new one under it), so a blanket
164///    "dest exists" pre-check would reject that normal case. Per-file
165///    overwrite conflicts inside a moved directory are still caught
166///    correctly, just per-entry, by `CopyAction` in the fallback path
167///    below.
168/// 2. Attempt a single atomic rename.
169/// 3. On cross-device failure, fall back to
170///    `pipeline::run_copy_pipeline` (unmodified — no dedicated
171///    `EntryAction` for move).
172/// 4. Any other rename error surfaces directly.
173/// 5. Once the copy phase resolves, run the deferred deletion sweep over
174///    `succeeded`, governed by the same `ErrorStrategy`.
175#[allow(clippy::too_many_arguments)]
176async fn move_path<R: Renamer>(
177    source: &Path,
178    dest: &Path,
179    overwrite: bool,
180    skip_if_identical: bool,
181    preserve_permissions: bool,
182    allow_filesystem_integrity_risk: bool,
183    small_file_threshold: u64,
184    config: &BatchConfig,
185    concurrency: usize,
186    cancel: CancellationToken,
187    reporter: ProgressReporter,
188    renamer: &R,
189) -> Result<OperationOutcome> {
190    // `rename(2)` fails with `NotFound` if any component of `dest`'s
191    // parent chain is missing — not just if `source` is missing — so
192    // without this, moving into a not-yet-created destination directory
193    // surfaces as a misleading `SourceNotFound` (see the `err` arm
194    // below, which blames `source` for every non-cross-device failure)
195    // and never reaches the copy-pipeline fallback, which *would* have
196    // created it. `create_dir_all` is a no-op when `parent` already
197    // exists, so this is safe to run unconditionally on every move.
198    if let Some(parent) = dest.parent() {
199        if let Err(err) = tokio::fs::create_dir_all(parent).await {
200            return Err(classify_io_error(err, dest.to_path_buf(), 0));
201        }
202    }
203
204    if !overwrite && resolve_existing_dest_conflict(source, dest, skip_if_identical).await? {
205        return Ok(OperationOutcome::default());
206    }
207
208    match renamer.rename(source, dest).await {
209        // Trivially "everything succeeded" without ever enumerating
210        // individual entries, so no progress events are emitted either.
211        Ok(()) => return Ok(OperationOutcome::default()),
212        Err(err) if is_cross_device(&err) => {}
213        Err(err) => return Err(classify_io_error(err, source.to_path_buf(), 0)),
214    }
215
216    let mut outcome = run_copy_pipeline(
217        source,
218        dest,
219        overwrite,
220        skip_if_identical,
221        preserve_permissions,
222        allow_filesystem_integrity_risk,
223        small_file_threshold,
224        config,
225        concurrency,
226        cancel,
227        reporter.clone(),
228    )
229    .await?;
230
231    sweep(&mut outcome, dest, config.error_strategy, reporter).await;
232
233    Ok(outcome)
234}
235
236/// Guards the atomic-rename fast path against `rename(2)`'s native
237/// overwrite semantics: on Unix (and Windows' `MoveFileEx` equivalent),
238/// `rename(source, dest)` happily replaces an existing destination
239/// *file* with no error, so without this check `overwrite=false` is
240/// silently unenforced here even though the cross-device fallback below
241/// (via `CopyAction`) enforces it correctly. Only applies when `source`
242/// is a single file — a directory `dest` pre-existing is the normal
243/// "move into" case (see the `move_path` doc comment), and there's no
244/// single-file checksum to compare a directory against anyway.
245///
246/// `Ok(false)` means "no conflict (or not applicable) — proceed with the
247/// rename exactly as before", the zero-added-cost path for the common
248/// case of moving to a location that doesn't exist yet. `Ok(true)` means
249/// the move is already fully resolved (dest was identical, so `source`
250/// was removed and nothing else needs to happen) without ever calling
251/// `rename`. `Err` is either the conflict itself (`Error::DestExists`)
252/// or a genuine I/O failure reaching either file's metadata.
253///
254/// `pub(crate)` (not private): `move_many.rs` runs this same per-source
255/// check ahead of its own rename attempt, for the same reason.
256pub(crate) async fn resolve_existing_dest_conflict(
257    source: &Path,
258    dest: &Path,
259    skip_if_identical: bool,
260) -> Result<bool> {
261    let dest_meta = match tokio::fs::metadata(dest).await {
262        Ok(meta) => meta,
263        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(false),
264        Err(e) => return Err(classify_io_error(e, dest.to_path_buf(), 0)),
265    };
266
267    let source_meta = tokio::fs::metadata(source)
268        .await
269        .map_err(|e| classify_io_error(e, source.to_path_buf(), 0))?;
270    if !source_meta.is_file() {
271        return Ok(false);
272    }
273
274    if identical_to_existing(
275        source,
276        dest,
277        source_meta.len(),
278        dest_meta.len(),
279        skip_if_identical,
280    )
281    .await?
282    {
283        tokio::fs::remove_file(source)
284            .await
285            .map_err(|e| classify_io_error(e, source.to_path_buf(), 0))?;
286        return Ok(true);
287    }
288
289    Err(Error::DestExists {
290        path: dest.to_path_buf(),
291    })
292}
293
294/// No-op fallback when `checksum` is disabled, so `enabled` (always
295/// `false` in that build, since the one builder method that can set it
296/// is itself `checksum`-gated) never needs its own `#[cfg]` at the call
297/// site — same pattern as `planner::action::CopyAction::identical_to_existing`.
298#[cfg(feature = "checksum")]
299async fn identical_to_existing(
300    source: &Path,
301    dest: &Path,
302    source_len: u64,
303    dest_len: u64,
304    enabled: bool,
305) -> Result<bool> {
306    if !enabled {
307        return Ok(false);
308    }
309    crate::checksum::files_identical(source, dest, source_len, dest_len).await
310}
311
312#[cfg(not(feature = "checksum"))]
313async fn identical_to_existing(
314    _source: &Path,
315    _dest: &Path,
316    _source_len: u64,
317    _dest_len: u64,
318    _enabled: bool,
319) -> Result<bool> {
320    Ok(false)
321}
322
323/// Deletes each `succeeded` entry's original source. Sequential — no
324/// batching/concurrency of its own, since deletions are cheap metadata
325/// operations, not data transfer.
326///
327/// `pub(crate)` (not private): `move_many.rs` reuses this verbatim over
328/// its own merged multi-source outcome — the same "delete every
329/// successfully-copied source, roll back on failure under `Undo`" logic
330/// applies regardless of whether the entries came from one source tree
331/// or several concatenated ones.
332pub(crate) async fn sweep(
333    outcome: &mut OperationOutcome,
334    dest_root: &Path,
335    error_strategy: ErrorStrategy,
336    reporter: ProgressReporter,
337) {
338    if outcome.succeeded.is_empty() {
339        return;
340    }
341
342    let entries = outcome.succeeded.clone();
343    let mut deleted_paths: HashSet<PathBuf> = HashSet::new();
344
345    reporter.send(Progress::Started {
346        bytes_total: None,
347        entries_total: entries.len(),
348    });
349
350    for entry in &entries {
351        reporter.send(Progress::EntryStarted {
352            entry: entry.clone(),
353        });
354
355        match remove_source(entry).await {
356            Ok(()) => {
357                reporter.send(Progress::EntryCompleted {
358                    entry: entry.clone(),
359                });
360                deleted_paths.insert(entry.path.clone());
361            }
362            Err(err) => {
363                reporter.send(Progress::EntryFailed {
364                    entry: entry.clone(),
365                });
366                let fatal = err.is_fatal();
367                let reason = if fatal {
368                    Some(StopReason::Fatal)
369                } else {
370                    match error_strategy {
371                        ErrorStrategy::ContinueAndCollect => None,
372                        ErrorStrategy::AbortOnError => Some(StopReason::AbortOnError),
373                        ErrorStrategy::Undo => Some(StopReason::Undo),
374                    }
375                };
376
377                outcome.cleanup_failed.push((entry.clone(), err));
378
379                if let Some(reason) = reason {
380                    if outcome.stopped_early.is_none() {
381                        outcome.stopped_early = Some(reason);
382                    }
383
384                    if matches!(error_strategy, ErrorStrategy::Undo) {
385                        rollback(&entries, &deleted_paths, dest_root).await;
386                        outcome.succeeded.clear();
387                        outcome.cleanup_failed.clear();
388                    }
389
390                    break;
391                }
392            }
393        }
394    }
395}
396
397/// Restores every entry to its pre-operation state: already-deleted
398/// sources are restored from the destination copy then that copy is
399/// removed; sources that were never touched just have their destination
400/// copy removed (identical to `CopyAction::undo`, reused directly rather
401/// than reimplemented).
402async fn rollback(entries: &[Entry], deleted_paths: &HashSet<PathBuf>, dest_root: &Path) {
403    let copy_action = CopyAction {
404        overwrite: true,
405        skip_if_identical: false,
406    };
407    for entry in entries.iter().rev() {
408        if deleted_paths.contains(&entry.path) {
409            let _ = restore_source(entry, dest_root).await;
410        } else {
411            let _ = copy_action.undo(entry, dest_root).await;
412        }
413    }
414}
415
416async fn remove_source(entry: &Entry) -> Result<()> {
417    match tokio::fs::remove_file(&entry.path).await {
418        Ok(()) => Ok(()),
419        Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
420        Err(e) => Err(classify_io_error(e, entry.path.clone(), 0)),
421    }
422}
423
424async fn restore_source(entry: &Entry, dest_root: &Path) -> Result<()> {
425    let dest_path = dest_root.join(&entry.relative_path);
426    tokio::fs::copy(&dest_path, &entry.path)
427        .await
428        .map_err(|e| classify_io_error(e, entry.path.clone(), 0))?;
429    tokio::fs::remove_file(&dest_path)
430        .await
431        .map_err(|e| classify_io_error(e, dest_path, 0))?;
432    Ok(())
433}
434
435#[cfg(test)]
436mod tests {
437    use std::fs;
438
439    use tempfile::tempdir;
440
441    use crate::error::Error;
442
443    use super::*;
444
445    #[test]
446    fn is_cross_device_true_for_crosses_devices_kind() {
447        let err = io::Error::from(io::ErrorKind::CrossesDevices);
448        assert!(is_cross_device(&err));
449    }
450
451    #[test]
452    fn is_cross_device_false_for_other_kinds() {
453        for kind in [
454            io::ErrorKind::PermissionDenied,
455            io::ErrorKind::NotFound,
456            io::ErrorKind::Other,
457        ] {
458            let err = io::Error::from(kind);
459            assert!(
460                !is_cross_device(&err),
461                "{kind:?} should not be classified as cross-device"
462            );
463        }
464    }
465
466    struct AlwaysCrossDevice;
467    impl Renamer for AlwaysCrossDevice {
468        async fn rename(&self, _source: &Path, _dest: &Path) -> io::Result<()> {
469            Err(io::Error::from(io::ErrorKind::CrossesDevices))
470        }
471    }
472
473    struct AlwaysPermissionDenied;
474    impl Renamer for AlwaysPermissionDenied {
475        async fn rename(&self, _source: &Path, _dest: &Path) -> io::Result<()> {
476            Err(io::Error::from(io::ErrorKind::PermissionDenied))
477        }
478    }
479
480    // Only referenced by the `#[cfg(unix)]` tests below (they need a
481    // deferred-deletion sweep to exercise, which only exists on the
482    // cross-device fallback path) — undetected until a Windows
483    // cross-compile of `--tests` was actually run.
484    #[cfg(unix)]
485    fn entry(path: PathBuf, relative_path: PathBuf, size: u64) -> Entry {
486        Entry {
487            path,
488            relative_path,
489            size,
490            modified: None,
491        }
492    }
493
494    #[tokio::test]
495    async fn same_filesystem_move_uses_rename_and_skips_pipeline() {
496        let root = tempdir().unwrap();
497        let source = root.path().join("src.txt");
498        let dest = root.path().join("dst.txt");
499        fs::write(&source, b"hello").unwrap();
500
501        let outcome = move_path(
502            &source,
503            &dest,
504            false,
505            false, // skip_if_identical
506            false,
507            false,
508            256,
509            &BatchConfig::default(),
510            2,
511            CancellationToken::new(),
512            ProgressReporter::noop(),
513            &TokioRenamer,
514        )
515        .await
516        .unwrap();
517
518        assert!(
519            outcome.succeeded.is_empty(),
520            "fast path doesn't enumerate entries"
521        );
522        assert!(!source.exists());
523        assert_eq!(fs::read(&dest).unwrap(), b"hello");
524    }
525
526    #[tokio::test]
527    async fn same_filesystem_move_creates_missing_dest_parent_dirs() {
528        let root = tempdir().unwrap();
529        let source = root.path().join("src.txt");
530        let dest = root
531            .path()
532            .join("does")
533            .join("not")
534            .join("exist")
535            .join("dst.txt");
536        fs::write(&source, b"hello").unwrap();
537
538        let outcome = move_path(
539            &source,
540            &dest,
541            false,
542            false, // skip_if_identical
543            false,
544            false,
545            256,
546            &BatchConfig::default(),
547            2,
548            CancellationToken::new(),
549            ProgressReporter::noop(),
550            &TokioRenamer,
551        )
552        .await
553        .unwrap();
554
555        assert!(outcome.succeeded.is_empty());
556        assert!(!source.exists());
557        assert_eq!(fs::read(&dest).unwrap(), b"hello");
558    }
559
560    #[cfg(feature = "checksum")]
561    #[tokio::test]
562    async fn same_filesystem_move_without_overwrite_fails_on_a_differing_destination() {
563        let root = tempdir().unwrap();
564        let source = root.path().join("src.txt");
565        let dest = root.path().join("dst.txt");
566        fs::write(&source, b"new content").unwrap();
567        fs::write(&dest, b"old content").unwrap();
568
569        let result = move_path(
570            &source,
571            &dest,
572            false,
573            false, // skip_if_identical
574            false,
575            false,
576            256,
577            &BatchConfig::default(),
578            2,
579            CancellationToken::new(),
580            ProgressReporter::noop(),
581            &TokioRenamer,
582        )
583        .await;
584
585        assert!(matches!(result, Err(Error::DestExists { .. })));
586        assert!(
587            source.exists(),
588            "a rejected move must leave source in place"
589        );
590        assert_eq!(fs::read(&dest).unwrap(), b"old content");
591    }
592
593    #[cfg(feature = "checksum")]
594    #[tokio::test]
595    async fn same_filesystem_move_skips_an_identical_destination_but_still_removes_source() {
596        let root = tempdir().unwrap();
597        let source = root.path().join("src.txt");
598        let dest = root.path().join("dst.txt");
599        fs::write(&source, b"same content").unwrap();
600        fs::write(&dest, b"same content").unwrap();
601
602        let outcome = move_path(
603            &source,
604            &dest,
605            false,
606            true, // skip_if_identical
607            false,
608            false,
609            256,
610            &BatchConfig::default(),
611            2,
612            CancellationToken::new(),
613            ProgressReporter::noop(),
614            &TokioRenamer,
615        )
616        .await
617        .unwrap();
618
619        assert!(outcome.succeeded.is_empty());
620        assert!(
621            outcome.skipped.is_empty(),
622            "the fast path never enumerates entries"
623        );
624        assert!(
625            !source.exists(),
626            "the move should still complete by removing the now-redundant source"
627        );
628        assert_eq!(fs::read(&dest).unwrap(), b"same content");
629    }
630
631    #[tokio::test]
632    async fn cross_device_fallback_copies_then_deletes_sources() {
633        let src_dir = tempdir().unwrap();
634        let dest_dir = tempdir().unwrap();
635        fs::write(src_dir.path().join("a.txt"), b"a").unwrap();
636        fs::write(src_dir.path().join("b.txt"), b"b").unwrap();
637
638        let outcome = move_path(
639            src_dir.path(),
640            dest_dir.path(),
641            false,
642            false, // skip_if_identical
643            false,
644            false,
645            256,
646            &BatchConfig::default(),
647            2,
648            CancellationToken::new(),
649            ProgressReporter::noop(),
650            &AlwaysCrossDevice,
651        )
652        .await
653        .unwrap();
654
655        assert_eq!(outcome.succeeded.len(), 2);
656        assert!(outcome.failed.is_empty());
657        assert!(outcome.cleanup_failed.is_empty());
658
659        assert!(!src_dir.path().join("a.txt").exists());
660        assert!(!src_dir.path().join("b.txt").exists());
661        assert_eq!(fs::read(dest_dir.path().join("a.txt")).unwrap(), b"a");
662        assert_eq!(fs::read(dest_dir.path().join("b.txt")).unwrap(), b"b");
663    }
664
665    #[tokio::test]
666    async fn non_cross_device_rename_error_surfaces_directly_without_fallback() {
667        let src_dir = tempdir().unwrap();
668        let dest_dir = tempdir().unwrap();
669        fs::write(src_dir.path().join("a.txt"), b"a").unwrap();
670
671        let result = move_path(
672            src_dir.path(),
673            dest_dir.path(),
674            false,
675            false, // skip_if_identical
676            false,
677            false,
678            256,
679            &BatchConfig::default(),
680            2,
681            CancellationToken::new(),
682            ProgressReporter::noop(),
683            &AlwaysPermissionDenied,
684        )
685        .await;
686
687        assert!(matches!(result, Err(Error::PermissionDenied { .. })));
688        assert!(!dest_dir.path().join("a.txt").exists());
689        assert!(src_dir.path().join("a.txt").exists());
690    }
691
692    #[cfg(unix)]
693    #[tokio::test]
694    async fn continue_and_collect_sweep_keeps_deleting_after_one_failure() {
695        use std::os::unix::fs::PermissionsExt;
696
697        let src_dir = tempdir().unwrap();
698        let dest_dir = tempdir().unwrap();
699
700        let locked_dir = src_dir.path().join("locked");
701        fs::create_dir(&locked_dir).unwrap();
702        let a = locked_dir.join("a.txt");
703        fs::write(&a, b"a").unwrap();
704
705        let b = src_dir.path().join("b.txt");
706        fs::write(&b, b"b").unwrap();
707
708        fs::create_dir_all(dest_dir.path().join("locked")).unwrap();
709        fs::write(dest_dir.path().join("locked").join("a.txt"), b"a").unwrap();
710        fs::write(dest_dir.path().join("b.txt"), b"b").unwrap();
711
712        let entry_a = entry(a.clone(), PathBuf::from("locked/a.txt"), 1);
713        let entry_b = entry(b.clone(), PathBuf::from("b.txt"), 1);
714
715        let mut outcome = OperationOutcome {
716            succeeded: vec![entry_a.clone(), entry_b.clone()],
717            ..OperationOutcome::default()
718        };
719
720        // unlink needs write+execute on the containing directory, so
721        // locking it down makes deleting `a` fail with permission denied.
722        fs::set_permissions(&locked_dir, fs::Permissions::from_mode(0o555)).unwrap();
723        sweep(
724            &mut outcome,
725            dest_dir.path(),
726            ErrorStrategy::ContinueAndCollect,
727            ProgressReporter::noop(),
728        )
729        .await;
730        fs::set_permissions(&locked_dir, fs::Permissions::from_mode(0o755)).unwrap();
731
732        assert_eq!(outcome.cleanup_failed.len(), 1);
733        assert_eq!(outcome.cleanup_failed[0].0.path, a);
734        assert!(
735            a.exists(),
736            "a's deletion should have failed, leaving it in place"
737        );
738        assert!(!b.exists(), "b's deletion should still have succeeded");
739        assert_eq!(outcome.stopped_early, None);
740    }
741
742    #[cfg(unix)]
743    #[tokio::test]
744    async fn abort_on_error_sweep_stops_after_first_deletion_failure() {
745        use std::os::unix::fs::PermissionsExt;
746
747        let src_dir = tempdir().unwrap();
748        let dest_dir = tempdir().unwrap();
749
750        let locked_dir = src_dir.path().join("locked");
751        fs::create_dir(&locked_dir).unwrap();
752        let a = locked_dir.join("a.txt");
753        fs::write(&a, b"a").unwrap();
754
755        let b = src_dir.path().join("b.txt");
756        fs::write(&b, b"b").unwrap();
757        let c = src_dir.path().join("c.txt");
758        fs::write(&c, b"c").unwrap();
759
760        fs::write(dest_dir.path().join("b.txt"), b"b").unwrap();
761        fs::write(dest_dir.path().join("c.txt"), b"c").unwrap();
762
763        let entry_a = entry(a.clone(), PathBuf::from("locked/a.txt"), 1);
764        let entry_b = entry(b.clone(), PathBuf::from("b.txt"), 1);
765        let entry_c = entry(c.clone(), PathBuf::from("c.txt"), 1);
766
767        let mut outcome = OperationOutcome {
768            succeeded: vec![entry_a, entry_b, entry_c],
769            ..OperationOutcome::default()
770        };
771
772        fs::set_permissions(&locked_dir, fs::Permissions::from_mode(0o555)).unwrap();
773        sweep(
774            &mut outcome,
775            dest_dir.path(),
776            ErrorStrategy::AbortOnError,
777            ProgressReporter::noop(),
778        )
779        .await;
780        fs::set_permissions(&locked_dir, fs::Permissions::from_mode(0o755)).unwrap();
781
782        assert_eq!(outcome.stopped_early, Some(StopReason::AbortOnError));
783        assert!(a.exists());
784        assert!(
785            b.exists(),
786            "b comes after the triggering failure, so it should never be attempted"
787        );
788        assert!(
789            c.exists(),
790            "c comes after the triggering failure, so it should never be attempted"
791        );
792    }
793
794    #[cfg(unix)]
795    #[tokio::test]
796    async fn undo_sweep_restores_everything_on_deletion_failure() {
797        use std::os::unix::fs::PermissionsExt;
798
799        let src_dir = tempdir().unwrap();
800        let dest_dir = tempdir().unwrap();
801
802        let locked_dir = src_dir.path().join("locked");
803        fs::create_dir(&locked_dir).unwrap();
804        let a = locked_dir.join("a.txt");
805        fs::write(&a, b"a").unwrap();
806
807        let b = src_dir.path().join("b.txt");
808        fs::write(&b, b"b").unwrap();
809
810        fs::create_dir_all(dest_dir.path().join("locked")).unwrap();
811        fs::write(dest_dir.path().join("locked").join("a.txt"), b"a").unwrap();
812        fs::write(dest_dir.path().join("b.txt"), b"b").unwrap();
813
814        let entry_a = entry(a.clone(), PathBuf::from("locked/a.txt"), 1);
815        let entry_b = entry(b.clone(), PathBuf::from("b.txt"), 1);
816
817        // b first (its deletion succeeds), then a (its deletion fails and
818        // triggers rollback of everything, including b).
819        let mut outcome = OperationOutcome {
820            succeeded: vec![entry_b, entry_a],
821            ..OperationOutcome::default()
822        };
823
824        fs::set_permissions(&locked_dir, fs::Permissions::from_mode(0o555)).unwrap();
825        sweep(
826            &mut outcome,
827            dest_dir.path(),
828            ErrorStrategy::Undo,
829            ProgressReporter::noop(),
830        )
831        .await;
832        fs::set_permissions(&locked_dir, fs::Permissions::from_mode(0o755)).unwrap();
833
834        assert!(outcome.succeeded.is_empty());
835        assert!(outcome.cleanup_failed.is_empty());
836        assert_eq!(outcome.stopped_early, Some(StopReason::Undo));
837
838        assert!(
839            a.exists(),
840            "a's source was never removed, since its deletion failed"
841        );
842        assert!(b.exists(), "b's source should have been restored from dest");
843        assert_eq!(fs::read(&b).unwrap(), b"b");
844
845        assert!(!dest_dir.path().join("locked").join("a.txt").exists());
846        assert!(!dest_dir.path().join("b.txt").exists());
847    }
848}