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 #[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"))]
70 pub fn preserve_permissions(mut self, preserve: bool) -> Self {
71 self.preserve_permissions = preserve;
72 self
73 }
74
75 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
135pub(crate) fn is_cross_device(err: &io::Error) -> bool {
138 err.kind() == io::ErrorKind::CrossesDevices
139}
140
141pub(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#[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 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 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
236pub(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#[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
323pub(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
397async 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 #[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, 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, 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, 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, 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, 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, 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 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 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}