1use std::collections::{BTreeSet, HashMap};
20use std::path::{Path, PathBuf};
21use std::time::UNIX_EPOCH;
22
23use bon::bon;
24use futures::StreamExt;
25use globset::{Glob, GlobMatcher};
26
27use crate::buckets::{BucketTreeEntry, BucketUpload, HFBucket};
28use crate::error::{HFError, HFResult};
29use crate::progress::{DownloadEvent, EmitEvent, Progress};
30
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
33pub enum BucketSyncDirection {
34 Upload,
36 Download,
38}
39
40#[derive(Clone)]
43pub(crate) struct BucketSyncParams {
44 pub local_path: PathBuf,
45 pub direction: BucketSyncDirection,
46 pub prefix: Option<String>,
47 pub delete: bool,
48 pub ignore_times: bool,
49 pub ignore_sizes: bool,
50 pub existing: bool,
51 pub ignore_existing: bool,
52 pub include: Vec<String>,
53 pub exclude: Vec<String>,
54 pub verbose: bool,
55 pub progress: Option<Progress>,
56}
57
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
60pub enum BucketSyncAction {
61 Upload,
63 Download,
65 Delete,
67 Skip,
69}
70
71#[derive(Debug, Clone)]
73pub struct BucketSyncOperation {
74 pub action: BucketSyncAction,
76 pub path: String,
78 pub size: Option<u64>,
80 pub reason: String,
82}
83
84#[derive(Debug, Clone)]
88pub struct BucketSyncPlan {
89 pub direction: BucketSyncDirection,
91 pub operations: Vec<BucketSyncOperation>,
93 pub(crate) download_entries: HashMap<String, BucketTreeEntry>,
96}
97
98impl BucketSyncPlan {
99 pub fn uploads(&self) -> usize {
101 self.operations
102 .iter()
103 .filter(|op| op.action == BucketSyncAction::Upload)
104 .count()
105 }
106
107 pub fn downloads(&self) -> usize {
109 self.operations
110 .iter()
111 .filter(|op| op.action == BucketSyncAction::Download)
112 .count()
113 }
114
115 pub fn deletes(&self) -> usize {
117 self.operations
118 .iter()
119 .filter(|op| op.action == BucketSyncAction::Delete)
120 .count()
121 }
122
123 pub fn skips(&self) -> usize {
128 self.operations.iter().filter(|op| op.action == BucketSyncAction::Skip).count()
129 }
130
131 pub fn transfer_bytes(&self) -> u64 {
133 self.operations
134 .iter()
135 .filter(|op| op.action == BucketSyncAction::Upload || op.action == BucketSyncAction::Download)
136 .filter_map(|op| op.size)
137 .sum()
138 }
139}
140
141const SYNC_TIME_WINDOW_MS: f64 = 1000.0;
142
143fn validate_params(params: &BucketSyncParams) -> HFResult<()> {
144 if params.ignore_times && params.ignore_sizes {
145 return Err(HFError::InvalidParameter("cannot use both --ignore-times and --ignore-sizes".to_string()));
146 }
147 if params.existing && params.ignore_existing {
148 return Err(HFError::InvalidParameter("cannot use both --existing and --ignore-existing".to_string()));
149 }
150 if params.direction == BucketSyncDirection::Upload && !params.local_path.is_dir() {
151 return Err(HFError::InvalidParameter(format!(
152 "local path must be an existing directory for upload: {}",
153 params.local_path.display()
154 )));
155 }
156 Ok(())
157}
158
159fn compile_patterns(patterns: &[String]) -> HFResult<Vec<GlobMatcher>> {
160 patterns
161 .iter()
162 .map(|p| {
163 Glob::new(p)
164 .map(|g| g.compile_matcher())
165 .map_err(|e| HFError::InvalidParameter(format!("invalid glob pattern '{p}': {e}")))
166 })
167 .collect()
168}
169
170fn matches_filters(path: &str, include: &[GlobMatcher], exclude: &[GlobMatcher]) -> bool {
171 for pat in exclude {
172 if pat.is_match(path) {
173 return false;
174 }
175 }
176 if include.is_empty() {
177 return true;
178 }
179 for pat in include {
180 if pat.is_match(path) {
181 return true;
182 }
183 }
184 false
185}
186
187fn list_local_files(root: &Path) -> HFResult<HashMap<String, (u64, f64)>> {
188 let mut result = HashMap::new();
189 let mut stack = vec![root.to_path_buf()];
190
191 while let Some(dir) = stack.pop() {
192 let entries = std::fs::read_dir(&dir)?;
193 for entry in entries {
194 let entry = entry?;
195 let ft = entry.file_type()?;
196 if ft.is_dir() {
197 stack.push(entry.path());
198 } else if ft.is_file() {
199 let path = entry.path();
200 let rel = path
201 .strip_prefix(root)
202 .map_err(|e| HFError::InvalidParameter(format!("failed to strip prefix: {e}")))?;
203 let rel_str = rel
204 .components()
205 .map(|c| c.as_os_str().to_string_lossy())
206 .collect::<Vec<_>>()
207 .join("/");
208 let metadata = std::fs::metadata(&path)?;
209 let size = metadata.len();
210 let mtime_ms = metadata
211 .modified()
212 .ok()
213 .and_then(|t| t.duration_since(UNIX_EPOCH).ok())
214 .map(|d| d.as_millis() as f64)
215 .unwrap_or(0.0);
216 result.insert(rel_str, (size, mtime_ms));
217 }
218 }
219 }
220
221 Ok(result)
222}
223
224fn strip_prefix<'a>(path: &'a str, prefix: &str) -> Option<&'a str> {
225 if prefix.is_empty() {
226 return Some(path);
227 }
228 let with_slash = if prefix.ends_with('/') {
229 prefix.to_string()
230 } else {
231 format!("{prefix}/")
232 };
233 path.strip_prefix(&with_slash)
234}
235
236fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
237 let y = if m <= 2 { y - 1 } else { y };
238 let m = if m <= 2 { m + 9 } else { m - 3 };
239 let era = if y >= 0 { y } else { y - 399 } / 400;
240 let yoe = (y - era * 400) as u64;
241 let doy = (153 * m as u64 + 2) / 5 + d as u64 - 1;
242 let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
243 era * 146097 + doe as i64 - 719468
244}
245
246fn parse_iso_mtime(s: &str) -> f64 {
247 let s = s.trim();
250 if s.len() < 19 {
251 return 0.0;
252 }
253
254 let parse = || -> Option<f64> {
255 let year: i64 = s[0..4].parse().ok()?;
256 if s.as_bytes().get(4)? != &b'-' {
257 return None;
258 }
259 let month: i64 = s[5..7].parse().ok()?;
260 if s.as_bytes().get(7)? != &b'-' {
261 return None;
262 }
263 let day: i64 = s[8..10].parse().ok()?;
264 if s.as_bytes().get(10)? != &b'T' {
265 return None;
266 }
267 let hour: i64 = s[11..13].parse().ok()?;
268 if s.as_bytes().get(13)? != &b':' {
269 return None;
270 }
271 let minute: i64 = s[14..16].parse().ok()?;
272 if s.as_bytes().get(16)? != &b':' {
273 return None;
274 }
275 let second: i64 = s[17..19].parse().ok()?;
276
277 let mut frac_ms: f64 = 0.0;
278 let mut rest_start = 19;
279 if s.as_bytes().get(19) == Some(&b'.') {
280 let frac_start = 20;
281 let mut frac_end = frac_start;
282 while frac_end < s.len() && s.as_bytes()[frac_end].is_ascii_digit() {
283 frac_end += 1;
284 }
285 if frac_end > frac_start {
286 let frac_str = &s[frac_start..frac_end];
287 let frac_val: f64 = frac_str.parse().ok()?;
288 let divisor = 10f64.powi(frac_str.len() as i32);
289 frac_ms = (frac_val / divisor) * 1000.0;
290 }
291 rest_start = frac_end;
292 }
293
294 let _ = rest_start;
296
297 let days = days_from_civil(year, month, day);
298 let total_ms = days as f64 * 86_400_000.0
299 + hour as f64 * 3_600_000.0
300 + minute as f64 * 60_000.0
301 + second as f64 * 1000.0
302 + frac_ms;
303
304 Some(total_ms)
305 };
306
307 parse().unwrap_or(0.0)
308}
309
310#[derive(Debug, Clone, Copy)]
311enum CompareRole {
312 Upload,
313 Download,
314}
315
316fn compare_files(
317 path: String,
318 role: CompareRole,
319 source_size: u64,
320 source_mtime: f64,
321 dest_size: u64,
322 dest_mtime: f64,
323 params: &BucketSyncParams,
324) -> Option<BucketSyncOperation> {
325 let action = match role {
326 CompareRole::Upload => BucketSyncAction::Upload,
327 CompareRole::Download => BucketSyncAction::Download,
328 };
329
330 let source_newer = (source_mtime - dest_mtime) > SYNC_TIME_WINDOW_MS;
331 let dest_newer = (dest_mtime - source_mtime) > SYNC_TIME_WINDOW_MS;
332 let size_differs = source_size != dest_size;
333
334 let source_label = match role {
335 CompareRole::Upload => "local newer",
336 CompareRole::Download => "remote newer",
337 };
338 let dest_newer_label = match role {
339 CompareRole::Upload => "remote newer",
340 CompareRole::Download => "local newer",
341 };
342
343 if params.ignore_existing {
344 if params.verbose {
345 return Some(BucketSyncOperation {
346 action: BucketSyncAction::Skip,
347 path,
348 size: Some(source_size),
349 reason: "exists on receiver (--ignore-existing)".to_string(),
350 });
351 }
352 return None;
353 }
354
355 if params.ignore_sizes {
356 if source_newer {
357 return Some(BucketSyncOperation {
358 action,
359 path,
360 size: Some(source_size),
361 reason: source_label.to_string(),
362 });
363 }
364 let reason = if dest_newer {
365 dest_newer_label.to_string()
366 } else {
367 "same mtime".to_string()
368 };
369 if params.verbose {
370 return Some(BucketSyncOperation {
371 action: BucketSyncAction::Skip,
372 path,
373 size: Some(source_size),
374 reason,
375 });
376 }
377 return None;
378 }
379
380 if params.ignore_times {
381 if size_differs {
382 return Some(BucketSyncOperation {
383 action,
384 path,
385 size: Some(source_size),
386 reason: "size differs".to_string(),
387 });
388 }
389 if params.verbose {
390 return Some(BucketSyncOperation {
391 action: BucketSyncAction::Skip,
392 path,
393 size: Some(source_size),
394 reason: "same size".to_string(),
395 });
396 }
397 return None;
398 }
399
400 if size_differs || source_newer {
401 let reason = if size_differs {
402 "size differs".to_string()
403 } else {
404 source_label.to_string()
405 };
406 return Some(BucketSyncOperation {
407 action,
408 path,
409 size: Some(source_size),
410 reason,
411 });
412 }
413
414 if params.verbose {
415 return Some(BucketSyncOperation {
416 action: BucketSyncAction::Skip,
417 path,
418 size: Some(source_size),
419 reason: "identical".to_string(),
420 });
421 }
422 None
423}
424
425impl HFBucket {
426 async fn list_remote_files(
427 &self,
428 prefix: &Option<String>,
429 include: &[GlobMatcher],
430 exclude: &[GlobMatcher],
431 ) -> HFResult<(HashMap<String, (u64, f64)>, HashMap<String, BucketTreeEntry>)> {
432 let stream = self.list_tree().maybe_prefix(prefix.clone()).recursive(true).send()?;
433 futures::pin_mut!(stream);
434
435 let mut files: HashMap<String, (u64, f64)> = HashMap::new();
436 let mut entries: HashMap<String, BucketTreeEntry> = HashMap::new();
437
438 let prefix_str = prefix.as_deref().unwrap_or("");
439
440 while let Some(item) = stream.next().await {
441 let entry = item?;
442 match &entry {
443 BucketTreeEntry::File { path, size, mtime, .. } => {
444 let rel = match strip_prefix(path, prefix_str) {
445 Some(r) => r.to_string(),
446 None => continue,
447 };
448 if !matches_filters(&rel, include, exclude) {
449 continue;
450 }
451 let mtime_ms = mtime.as_deref().map(parse_iso_mtime).unwrap_or(0.0);
452 files.insert(rel.clone(), (*size, mtime_ms));
453 entries.insert(rel, entry);
454 },
455 BucketTreeEntry::Directory { .. } => continue,
456 }
457 }
458
459 Ok((files, entries))
460 }
461
462 fn compute_upload_plan(
463 &self,
464 local_files: &HashMap<String, (u64, f64)>,
465 remote_files: &HashMap<String, (u64, f64)>,
466 params: &BucketSyncParams,
467 ) -> BucketSyncPlan {
468 let mut all_keys = BTreeSet::new();
469 all_keys.extend(local_files.keys().cloned());
470 all_keys.extend(remote_files.keys().cloned());
471
472 let mut operations = Vec::new();
473
474 for key in &all_keys {
475 let in_local = local_files.contains_key(key);
476 let in_remote = remote_files.contains_key(key);
477
478 if in_local && !in_remote {
479 if params.existing {
480 if params.verbose {
481 operations.push(BucketSyncOperation {
482 action: BucketSyncAction::Skip,
483 path: key.clone(),
484 size: local_files.get(key).map(|(s, _)| *s),
485 reason: "not on receiver (--existing)".to_string(),
486 });
487 }
488 } else {
489 let (size, _) = local_files[key];
490 operations.push(BucketSyncOperation {
491 action: BucketSyncAction::Upload,
492 path: key.clone(),
493 size: Some(size),
494 reason: "new file".to_string(),
495 });
496 }
497 } else if in_local && in_remote {
498 let (local_size, local_mtime) = local_files[key];
499 let (remote_size, remote_mtime) = remote_files[key];
500 if let Some(op) = compare_files(
501 key.clone(),
502 CompareRole::Upload,
503 local_size,
504 local_mtime,
505 remote_size,
506 remote_mtime,
507 params,
508 ) {
509 operations.push(op);
510 }
511 } else if !in_local && in_remote && params.delete {
512 let (size, _) = remote_files[key];
513 operations.push(BucketSyncOperation {
514 action: BucketSyncAction::Delete,
515 path: key.clone(),
516 size: Some(size),
517 reason: "not in source (--delete)".to_string(),
518 });
519 }
520 }
521
522 BucketSyncPlan {
523 direction: BucketSyncDirection::Upload,
524 operations,
525 download_entries: HashMap::new(),
526 }
527 }
528
529 fn compute_download_plan(
530 &self,
531 local_files: &HashMap<String, (u64, f64)>,
532 remote_files: &HashMap<String, (u64, f64)>,
533 remote_entries: &HashMap<String, BucketTreeEntry>,
534 params: &BucketSyncParams,
535 ) -> BucketSyncPlan {
536 let mut all_keys = BTreeSet::new();
537 all_keys.extend(local_files.keys().cloned());
538 all_keys.extend(remote_files.keys().cloned());
539
540 let mut operations = Vec::new();
541 let mut download_entries = HashMap::new();
542
543 for key in &all_keys {
544 let in_local = local_files.contains_key(key);
545 let in_remote = remote_files.contains_key(key);
546
547 if in_remote && !in_local {
548 if params.existing {
549 if params.verbose {
550 operations.push(BucketSyncOperation {
551 action: BucketSyncAction::Skip,
552 path: key.clone(),
553 size: remote_files.get(key).map(|(s, _)| *s),
554 reason: "not on receiver (--existing)".to_string(),
555 });
556 }
557 } else {
558 let (size, _) = remote_files[key];
559 operations.push(BucketSyncOperation {
560 action: BucketSyncAction::Download,
561 path: key.clone(),
562 size: Some(size),
563 reason: "new file".to_string(),
564 });
565 if let Some(entry) = remote_entries.get(key) {
566 download_entries.insert(key.clone(), entry.clone());
567 }
568 }
569 } else if in_local && in_remote {
570 let (remote_size, remote_mtime) = remote_files[key];
571 let (local_size, local_mtime) = local_files[key];
572 if let Some(op) = compare_files(
573 key.clone(),
574 CompareRole::Download,
575 remote_size,
576 remote_mtime,
577 local_size,
578 local_mtime,
579 params,
580 ) {
581 if op.action == BucketSyncAction::Download
582 && let Some(entry) = remote_entries.get(key)
583 {
584 download_entries.insert(key.clone(), entry.clone());
585 }
586 operations.push(op);
587 }
588 } else if in_local && !in_remote && params.delete {
589 let (size, _) = local_files[key];
590 operations.push(BucketSyncOperation {
591 action: BucketSyncAction::Delete,
592 path: key.clone(),
593 size: Some(size),
594 reason: "not in source (--delete)".to_string(),
595 });
596 }
597 }
598
599 BucketSyncPlan {
600 direction: BucketSyncDirection::Download,
601 operations,
602 download_entries,
603 }
604 }
605
606 async fn execute_upload_plan(&self, plan: &BucketSyncPlan, params: &BucketSyncParams) -> HFResult<()> {
607 let upload_files: Vec<BucketUpload> = plan
608 .operations
609 .iter()
610 .filter(|op| op.action == BucketSyncAction::Upload)
611 .map(|op| {
612 let local_path = params.local_path.join(&op.path);
613 let remote_path = match ¶ms.prefix {
614 Some(prefix) => format!("{prefix}/{}", op.path),
615 None => op.path.clone(),
616 };
617 BucketUpload::new(local_path, remote_path)
618 })
619 .collect();
620
621 let delete_paths: Vec<String> = plan
622 .operations
623 .iter()
624 .filter(|op| op.action == BucketSyncAction::Delete)
625 .map(|op| match ¶ms.prefix {
626 Some(prefix) => format!("{prefix}/{}", op.path),
627 None => op.path.clone(),
628 })
629 .collect();
630
631 if !upload_files.is_empty() {
632 self.upload_files()
633 .files(upload_files)
634 .maybe_progress(params.progress.clone())
635 .send()
636 .await?;
637 }
638 if !delete_paths.is_empty() {
639 self.delete_files().paths(delete_paths).send().await?;
640 }
641
642 Ok(())
643 }
644
645 async fn execute_download_plan(&self, plan: &BucketSyncPlan, params: &BucketSyncParams) -> HFResult<()> {
646 let mut xet_batch_files = Vec::new();
647 let mut total_bytes: u64 = 0;
648
649 for op in plan.operations.iter().filter(|op| op.action == BucketSyncAction::Download) {
650 let entry = plan.download_entries.get(&op.path).ok_or_else(|| HFError::EntryNotFound {
651 path: op.path.clone(),
652 repo_id: self.bucket_id(),
653 context: None,
654 })?;
655
656 let local_path = params.local_path.join(&op.path);
657 if let Some(parent) = local_path.parent() {
658 std::fs::create_dir_all(parent)?;
659 }
660
661 match entry {
662 BucketTreeEntry::File { xet_hash, size, .. } => {
663 let remote_path = match ¶ms.prefix {
664 Some(prefix) => format!("{prefix}/{}", op.path),
665 None => op.path.clone(),
666 };
667 total_bytes += size;
668 xet_batch_files.push(crate::xet::XetBatchFile {
669 hash: xet_hash.clone(),
670 file_size: *size,
671 path: local_path,
672 filename: remote_path,
673 });
674 },
675 BucketTreeEntry::Directory { path, .. } => {
676 return Err(HFError::InvalidParameter(format!("Cannot download directory entry: {path}")));
677 },
678 }
679 }
680
681 if !xet_batch_files.is_empty() {
682 params.progress.emit(DownloadEvent::Start {
683 total_files: xet_batch_files.len(),
684 total_bytes,
685 });
686 self.xet_download_batch(&xet_batch_files, ¶ms.progress).await?;
687 params.progress.emit(DownloadEvent::Complete);
688 }
689
690 let delete_paths: Vec<&str> = plan
691 .operations
692 .iter()
693 .filter(|op| op.action == BucketSyncAction::Delete)
694 .map(|op| op.path.as_str())
695 .collect();
696
697 for rel_path in &delete_paths {
698 let local_path = params.local_path.join(rel_path);
699 if local_path.is_file() {
700 std::fs::remove_file(&local_path)?;
701 }
702 }
703
704 let mut dirs_to_check: BTreeSet<PathBuf> = BTreeSet::new();
706 for rel_path in &delete_paths {
707 let local_path = params.local_path.join(rel_path);
708 let mut parent = local_path.parent().map(Path::to_path_buf);
709 while let Some(p) = parent {
710 if p == params.local_path || !p.starts_with(¶ms.local_path) {
711 break;
712 }
713 dirs_to_check.insert(p.clone());
714 parent = p.parent().map(Path::to_path_buf);
715 }
716 }
717
718 let mut dirs_sorted: Vec<PathBuf> = dirs_to_check.into_iter().collect();
719 dirs_sorted.sort_by_key(|b| std::cmp::Reverse(b.components().count()));
720 for dir in dirs_sorted {
721 if dir.is_dir() {
722 let is_empty = std::fs::read_dir(&dir).map(|mut d| d.next().is_none()).unwrap_or(false);
723 if is_empty {
724 let _ = std::fs::remove_dir(&dir);
725 }
726 }
727 }
728
729 Ok(())
730 }
731
732 async fn sync_impl(&self, params: BucketSyncParams) -> HFResult<BucketSyncPlan> {
733 validate_params(¶ms)?;
734
735 let include = compile_patterns(¶ms.include)?;
736 let exclude = compile_patterns(¶ms.exclude)?;
737
738 let (remote_files, remote_entries) = self.list_remote_files(¶ms.prefix, &include, &exclude).await?;
739
740 match params.direction {
741 BucketSyncDirection::Upload => {
742 let all_local = list_local_files(¶ms.local_path)?;
743 let local_files: HashMap<String, (u64, f64)> = all_local
744 .into_iter()
745 .filter(|(k, _)| matches_filters(k, &include, &exclude))
746 .collect();
747
748 let plan = self.compute_upload_plan(&local_files, &remote_files, ¶ms);
749 self.execute_upload_plan(&plan, ¶ms).await?;
750 Ok(plan)
751 },
752 BucketSyncDirection::Download => {
753 std::fs::create_dir_all(¶ms.local_path)?;
754
755 let all_local = list_local_files(¶ms.local_path)?;
756 let local_files: HashMap<String, (u64, f64)> = all_local
757 .into_iter()
758 .filter(|(k, _)| matches_filters(k, &include, &exclude))
759 .collect();
760
761 let plan = self.compute_download_plan(&local_files, &remote_files, &remote_entries, ¶ms);
762 self.execute_download_plan(&plan, ¶ms).await?;
763 Ok(plan)
764 },
765 }
766 }
767}
768
769#[bon]
770impl HFBucket {
771 #[builder(finish_fn = send, derive(Debug, Clone))]
808 pub async fn sync(
809 &self,
810 #[builder(into)]
812 local_path: PathBuf,
813 direction: BucketSyncDirection,
815 #[builder(into)]
817 prefix: Option<String>,
818 #[builder(default)]
820 delete: bool,
821 #[builder(default)]
823 ignore_times: bool,
824 #[builder(default)]
826 ignore_sizes: bool,
827 #[builder(default)]
829 existing: bool,
830 #[builder(default)]
832 ignore_existing: bool,
833 #[builder(default)]
835 include: Vec<String>,
836 #[builder(default)]
838 exclude: Vec<String>,
839 #[builder(default)]
841 verbose: bool,
842 #[builder(into)]
844 progress: Option<Progress>,
845 ) -> HFResult<BucketSyncPlan> {
846 Box::pin(self.sync_impl(BucketSyncParams {
847 local_path,
848 direction,
849 prefix,
850 delete,
851 ignore_times,
852 ignore_sizes,
853 existing,
854 ignore_existing,
855 include,
856 exclude,
857 verbose,
858 progress,
859 }))
860 .await
861 }
862}
863
864#[cfg(feature = "blocking")]
865#[bon]
866impl crate::blocking::HFBucketSync {
867 #[builder(finish_fn = send, derive(Debug, Clone))]
870 pub fn sync(
871 &self,
872 #[builder(into)] local_path: PathBuf,
873 direction: BucketSyncDirection,
874 #[builder(into)] prefix: Option<String>,
875 #[builder(default)] delete: bool,
876 #[builder(default)] ignore_times: bool,
877 #[builder(default)] ignore_sizes: bool,
878 #[builder(default)] existing: bool,
879 #[builder(default)] ignore_existing: bool,
880 #[builder(default)] include: Vec<String>,
881 #[builder(default)] exclude: Vec<String>,
882 #[builder(default)] verbose: bool,
883 #[builder(into)] progress: Option<Progress>,
884 ) -> HFResult<BucketSyncPlan> {
885 self.runtime.block_on(
886 self.inner
887 .sync()
888 .local_path(local_path)
889 .direction(direction)
890 .maybe_prefix(prefix)
891 .delete(delete)
892 .ignore_times(ignore_times)
893 .ignore_sizes(ignore_sizes)
894 .existing(existing)
895 .ignore_existing(ignore_existing)
896 .include(include)
897 .exclude(exclude)
898 .verbose(verbose)
899 .maybe_progress(progress)
900 .send(),
901 )
902 }
903}
904
905#[cfg(test)]
906mod tests {
907 use super::*;
908
909 fn make_params(direction: BucketSyncDirection) -> BucketSyncParams {
910 BucketSyncParams {
911 local_path: PathBuf::from("/tmp"),
912 direction,
913 prefix: None,
914 delete: false,
915 ignore_times: false,
916 ignore_sizes: false,
917 existing: false,
918 ignore_existing: false,
919 include: Vec::new(),
920 exclude: Vec::new(),
921 verbose: false,
922 progress: None,
923 }
924 }
925
926 fn make_plan(ops: Vec<(BucketSyncAction, Option<u64>)>) -> BucketSyncPlan {
927 BucketSyncPlan {
928 direction: BucketSyncDirection::Upload,
929 operations: ops
930 .into_iter()
931 .enumerate()
932 .map(|(i, (action, size))| BucketSyncOperation {
933 action,
934 path: format!("file_{i}.txt"),
935 size,
936 reason: "test".to_string(),
937 })
938 .collect(),
939 download_entries: HashMap::new(),
940 }
941 }
942
943 #[test]
944 fn test_plan_counts() {
945 let plan = make_plan(vec![
946 (BucketSyncAction::Upload, Some(100)),
947 (BucketSyncAction::Upload, Some(200)),
948 (BucketSyncAction::Download, Some(300)),
949 (BucketSyncAction::Delete, Some(50)),
950 (BucketSyncAction::Skip, None),
951 ]);
952 assert_eq!(plan.uploads(), 2);
953 assert_eq!(plan.downloads(), 1);
954 assert_eq!(plan.deletes(), 1);
955 assert_eq!(plan.skips(), 1);
956 }
957
958 #[test]
959 fn test_transfer_bytes() {
960 let plan = make_plan(vec![
961 (BucketSyncAction::Upload, Some(100)),
962 (BucketSyncAction::Download, Some(300)),
963 (BucketSyncAction::Delete, Some(50)),
964 (BucketSyncAction::Skip, None),
965 ]);
966 assert_eq!(plan.transfer_bytes(), 400);
967 }
968
969 #[test]
970 fn test_empty_plan() {
971 let plan = make_plan(vec![]);
972 assert_eq!(plan.uploads(), 0);
973 assert_eq!(plan.downloads(), 0);
974 assert_eq!(plan.deletes(), 0);
975 assert_eq!(plan.skips(), 0);
976 assert_eq!(plan.transfer_bytes(), 0);
977 }
978
979 #[test]
980 fn test_parse_iso_mtime_basic() {
981 let ms = parse_iso_mtime("2024-01-15T10:30:00Z");
982 assert!(ms > 0.0);
983
984 let expected_days = days_from_civil(2024, 1, 15);
985 let expected_ms = expected_days as f64 * 86_400_000.0 + 10.0 * 3_600_000.0 + 30.0 * 60_000.0;
986 assert!((ms - expected_ms).abs() < 1.0);
987 }
988
989 #[test]
990 fn test_parse_iso_mtime_with_frac() {
991 let ms = parse_iso_mtime("2024-01-15T10:30:00.500Z");
992 let expected_days = days_from_civil(2024, 1, 15);
993 let expected_ms = expected_days as f64 * 86_400_000.0 + 10.0 * 3_600_000.0 + 30.0 * 60_000.0 + 500.0;
994 assert!((ms - expected_ms).abs() < 1.0);
995 }
996
997 #[test]
998 fn test_parse_iso_mtime_invalid() {
999 assert_eq!(parse_iso_mtime(""), 0.0);
1000 assert_eq!(parse_iso_mtime("not a date"), 0.0);
1001 assert_eq!(parse_iso_mtime("short"), 0.0);
1002 }
1003
1004 #[test]
1005 fn test_days_from_civil_epoch() {
1006 assert_eq!(days_from_civil(1970, 1, 1), 0);
1007 }
1008
1009 #[test]
1010 fn test_days_from_civil_known_date() {
1011 assert_eq!(days_from_civil(2024, 1, 1), 19723);
1013 }
1014
1015 #[test]
1016 fn test_strip_prefix_empty() {
1017 assert_eq!(strip_prefix("foo/bar.txt", ""), Some("foo/bar.txt"));
1018 }
1019
1020 #[test]
1021 fn test_strip_prefix_match() {
1022 assert_eq!(strip_prefix("data/foo/bar.txt", "data"), Some("foo/bar.txt"));
1023 assert_eq!(strip_prefix("data/foo/bar.txt", "data/"), Some("foo/bar.txt"));
1024 }
1025
1026 #[test]
1027 fn test_strip_prefix_no_match() {
1028 assert_eq!(strip_prefix("other/bar.txt", "data"), None);
1029 }
1030
1031 #[test]
1032 fn test_strip_prefix_partial_no_match() {
1033 assert_eq!(strip_prefix("submarine.txt", "sub"), None);
1034 }
1035
1036 #[test]
1037 fn test_strip_prefix_exact_file() {
1038 assert_eq!(strip_prefix("data/file.txt", "data"), Some("file.txt"));
1039 }
1040
1041 #[test]
1042 fn test_matches_filters_no_patterns() {
1043 assert!(matches_filters("foo.txt", &[], &[]));
1044 }
1045
1046 #[test]
1047 fn test_matches_filters_include() {
1048 let include = compile_patterns(&["*.txt".to_string()]).unwrap();
1049 assert!(matches_filters("foo.txt", &include, &[]));
1050 assert!(!matches_filters("foo.bin", &include, &[]));
1051 }
1052
1053 #[test]
1054 fn test_matches_filters_exclude() {
1055 let exclude = compile_patterns(&["*.log".to_string()]).unwrap();
1056 assert!(matches_filters("foo.txt", &[], &exclude));
1057 assert!(!matches_filters("foo.log", &[], &exclude));
1058 }
1059
1060 #[test]
1061 fn test_matches_filters_exclude_takes_priority() {
1062 let include = compile_patterns(&["*.txt".to_string()]).unwrap();
1063 let exclude = compile_patterns(&["secret*".to_string()]).unwrap();
1064 assert!(matches_filters("foo.txt", &include, &exclude));
1065 assert!(!matches_filters("secret.txt", &include, &exclude));
1066 }
1067
1068 #[test]
1069 fn test_validate_params_conflicting_times_sizes() {
1070 let params = BucketSyncParams {
1071 ignore_times: true,
1072 ignore_sizes: true,
1073 ..make_params(BucketSyncDirection::Download)
1074 };
1075 assert!(validate_params(¶ms).is_err());
1076 }
1077
1078 #[test]
1079 fn test_validate_params_conflicting_existing() {
1080 let params = BucketSyncParams {
1081 existing: true,
1082 ignore_existing: true,
1083 ..make_params(BucketSyncDirection::Download)
1084 };
1085 assert!(validate_params(¶ms).is_err());
1086 }
1087
1088 #[test]
1089 fn test_compare_files_identical() {
1090 let params = BucketSyncParams {
1091 verbose: true,
1092 ..make_params(BucketSyncDirection::Upload)
1093 };
1094
1095 let op = compare_files(String::new(), CompareRole::Upload, 100, 5000.0, 100, 5000.0, ¶ms).unwrap();
1096 assert_eq!(op.action, BucketSyncAction::Skip);
1097 assert_eq!(op.reason, "identical");
1098 }
1099
1100 #[test]
1101 fn test_compare_files_identical_not_verbose() {
1102 let params = make_params(BucketSyncDirection::Upload);
1103
1104 let op = compare_files(String::new(), CompareRole::Upload, 100, 5000.0, 100, 5000.0, ¶ms);
1105 assert!(op.is_none());
1106 }
1107
1108 #[test]
1109 fn test_compare_files_size_differs() {
1110 let params = make_params(BucketSyncDirection::Upload);
1111
1112 let op = compare_files(String::new(), CompareRole::Upload, 200, 5000.0, 100, 5000.0, ¶ms).unwrap();
1113 assert_eq!(op.action, BucketSyncAction::Upload);
1114 assert_eq!(op.reason, "size differs");
1115 }
1116
1117 #[test]
1118 fn test_compare_files_source_newer() {
1119 let params = make_params(BucketSyncDirection::Upload);
1120
1121 let op = compare_files(String::new(), CompareRole::Upload, 100, 7000.0, 100, 5000.0, ¶ms).unwrap();
1122 assert_eq!(op.action, BucketSyncAction::Upload);
1123 assert_eq!(op.reason, "local newer");
1124 }
1125
1126 #[test]
1127 fn test_compare_files_download_source_newer() {
1128 let params = make_params(BucketSyncDirection::Download);
1129
1130 let op = compare_files(String::new(), CompareRole::Download, 100, 7000.0, 100, 5000.0, ¶ms).unwrap();
1131 assert_eq!(op.action, BucketSyncAction::Download);
1132 assert_eq!(op.reason, "remote newer");
1133 }
1134
1135 #[test]
1136 fn test_compare_files_within_safety_window() {
1137 let params = BucketSyncParams {
1138 verbose: true,
1139 ..make_params(BucketSyncDirection::Upload)
1140 };
1141
1142 let op = compare_files(String::new(), CompareRole::Upload, 100, 5500.0, 100, 5000.0, ¶ms).unwrap();
1143 assert_eq!(op.action, BucketSyncAction::Skip);
1144 assert_eq!(op.reason, "identical");
1145 }
1146
1147 #[test]
1148 fn test_compare_files_ignore_times() {
1149 let params = BucketSyncParams {
1150 ignore_times: true,
1151 verbose: true,
1152 ..make_params(BucketSyncDirection::Upload)
1153 };
1154
1155 let op = compare_files(String::new(), CompareRole::Upload, 100, 9000.0, 100, 5000.0, ¶ms).unwrap();
1156 assert_eq!(op.action, BucketSyncAction::Skip);
1157 assert_eq!(op.reason, "same size");
1158
1159 let op = compare_files(String::new(), CompareRole::Upload, 200, 5000.0, 100, 5000.0, ¶ms).unwrap();
1160 assert_eq!(op.action, BucketSyncAction::Upload);
1161 assert_eq!(op.reason, "size differs");
1162 }
1163
1164 #[test]
1165 fn test_compare_files_ignore_sizes() {
1166 let params = BucketSyncParams {
1167 ignore_sizes: true,
1168 verbose: true,
1169 ..make_params(BucketSyncDirection::Upload)
1170 };
1171
1172 let op = compare_files(String::new(), CompareRole::Upload, 200, 5000.0, 100, 5000.0, ¶ms).unwrap();
1173 assert_eq!(op.action, BucketSyncAction::Skip);
1174 assert_eq!(op.reason, "same mtime");
1175
1176 let op = compare_files(String::new(), CompareRole::Upload, 100, 3000.0, 100, 5000.0, ¶ms).unwrap();
1177 assert_eq!(op.action, BucketSyncAction::Skip);
1178 assert_eq!(op.reason, "remote newer");
1179 }
1180
1181 #[test]
1182 fn test_compare_files_ignore_sizes_download() {
1183 let params = BucketSyncParams {
1184 ignore_sizes: true,
1185 verbose: true,
1186 ..make_params(BucketSyncDirection::Download)
1187 };
1188
1189 let op = compare_files(String::new(), CompareRole::Download, 100, 3000.0, 100, 5000.0, ¶ms).unwrap();
1190 assert_eq!(op.action, BucketSyncAction::Skip);
1191 assert_eq!(op.reason, "local newer");
1192 }
1193
1194 #[test]
1195 fn test_compare_files_ignore_existing() {
1196 let params = BucketSyncParams {
1197 ignore_existing: true,
1198 verbose: true,
1199 ..make_params(BucketSyncDirection::Upload)
1200 };
1201
1202 let op = compare_files(String::new(), CompareRole::Upload, 200, 9000.0, 100, 5000.0, ¶ms).unwrap();
1203 assert_eq!(op.action, BucketSyncAction::Skip);
1204 assert_eq!(op.reason, "exists on receiver (--ignore-existing)");
1205 }
1206
1207 #[test]
1208 fn test_compare_files_ignore_existing_not_verbose() {
1209 let params = BucketSyncParams {
1210 ignore_existing: true,
1211 ..make_params(BucketSyncDirection::Upload)
1212 };
1213
1214 let op = compare_files(String::new(), CompareRole::Upload, 200, 9000.0, 100, 5000.0, ¶ms);
1215 assert!(op.is_none());
1216 }
1217}