diff --git a/README.md b/README.md index 2a790dc1..430fb700 100644 --- a/README.md +++ b/README.md @@ -107,8 +107,8 @@ rc rb local/my-bucket # Recursively copy directory rc cp -r ./local-dir/ local/bucket/remote-dir/ -# Mirror between S3 locations -rc mirror local/bucket1/ local/bucket2/ +# Mirror between the local filesystem and RustFS +rc mirror ./local-dir/ local/bucket/backup/ # Find objects rc find local/bucket --name "*.txt" --newer 1d @@ -363,7 +363,7 @@ For full command documentation, see the [`rc` command reference](docs/reference/ | `find` | Find objects | | `anonymous` | Manage anonymous access to buckets and objects | | `diff` | Compare two locations | -| `mirror` | Mirror sync between S3 locations | +| `mirror` | Mirror local and S3-compatible directory trees | | `tree` | Tree view display | | `share` | Generate presigned URLs | | `event` | Manage bucket event notifications | diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index f485f41e..7f96e1d0 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -23,6 +23,7 @@ rc-s3 = { workspace = true } # Async runtime tokio.workspace = true futures.workspace = true +async-trait.workspace = true # CLI clap.workspace = true @@ -51,6 +52,7 @@ jiff.workspace = true humansize.workspace = true humantime.workspace = true mime_guess.workspace = true +tempfile.workspace = true glob.workspace = true shlex.workspace = true urlencoding.workspace = true diff --git a/crates/cli/src/commands/cp.rs b/crates/cli/src/commands/cp.rs index 8a5451db..ecdd29a9 100644 --- a/crates/cli/src/commands/cp.rs +++ b/crates/cli/src/commands/cp.rs @@ -724,7 +724,7 @@ fn build_transfer_controls(args: &CpArgs) -> Result { Ok(controls) } -fn parse_age_cutoff(value: &str, now: Timestamp) -> Result { +pub(super) fn parse_age_cutoff(value: &str, now: Timestamp) -> Result { let value = value.trim(); if value.is_empty() { return Err("Transfer age must not be empty".to_string()); @@ -762,7 +762,7 @@ fn parse_rewind_cutoff(value: &str, now: Timestamp) -> Result .map_err(|error| format!("Invalid rewind value '{value}': {error}")) } -fn parse_byte_rate(value: &str) -> Result { +pub(super) fn parse_byte_rate(value: &str) -> Result { let normalized = value.trim().to_ascii_lowercase(); let normalized = normalized .strip_suffix("/s") diff --git a/crates/cli/src/commands/mirror.rs b/crates/cli/src/commands/mirror.rs index 167b0bb5..c5716b97 100644 --- a/crates/cli/src/commands/mirror.rs +++ b/crates/cli/src/commands/mirror.rs @@ -1,46 +1,98 @@ -//! mirror command - Synchronize objects between two locations -//! -//! Mirrors objects from source to destination, optionally removing extra files. +//! mirror command - Synchronize trees between local filesystems and S3-compatible storage. + +use std::collections::BTreeMap; +use std::path::{Component, Path, PathBuf}; +use std::sync::Arc; use clap::Args; -use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; -use rc_core::{AliasManager, ListOptions, ObjectStore as _, ParsedPath, RemotePath, parse_path}; +use jiff::Timestamp; +use rc_core::alias::RetryConfig; +use rc_core::{ + AliasManager, Error, ListOptions, ObjectInfo, ObjectStore as _, ParsedPath, RemotePath, + TransferCandidate, TransferControls, TransferExecutor, TransferOutcomeState, TransferPlan, + TransferReport, TransferSelection, TransferSummary, parse_path, +}; use rc_s3::S3Client; use serde::Serialize; -use std::collections::HashMap; -use std::sync::Arc; -use tokio::task::{JoinError, JoinSet}; +use tokio::io::AsyncWriteExt as _; -use crate::commands::diff::{DiffEntry, DiffStatus}; +use super::cp::{parse_age_cutoff, parse_byte_rate}; use crate::exit_code::ExitCode; use crate::output::{Formatter, OutputConfig}; -/// Synchronize objects between two locations -#[derive(Args, Debug)] +const MIRROR_AFTER_HELP: &str = "\ +Examples: + rc mirror ./site/ local/web/site/ --overwrite + rc mirror local/archive/ ./restore/ --remove --dry-run + rc mirror stage/data/ prod/data/ --include '**/*.json' --summary"; + +/// Synchronize objects between two directory-like roots. +#[derive(Args, Clone, Debug)] +#[command(after_help = MIRROR_AFTER_HELP)] pub struct MirrorArgs { - /// Source path (alias/bucket/prefix) + /// Source directory or remote bucket/prefix pub source: String, - /// Destination path (alias/bucket/prefix) + /// Destination directory or remote bucket/prefix pub target: String, - /// Remove extra objects at destination + /// Remove destination files that are absent from the selected source tree #[arg(long)] pub remove: bool, - /// Overwrite existing objects + /// Overwrite changed destination files #[arg(long)] pub overwrite: bool, - /// Dry run (show what would be done without doing it) + /// Include source-relative paths matching this glob (repeatable) + #[arg(long)] + pub include: Vec, + + /// Exclude source-relative paths matching this glob; exclusions always win (repeatable) + #[arg(long)] + pub exclude: Vec, + + /// Select files modified more recently than this age (for example 1h or 7d) + #[arg(long)] + pub newer_than: Option, + + /// Select files modified less recently than this age (for example 1h or 7d) + #[arg(long)] + pub older_than: Option, + + /// Continue after per-file failures + #[arg(long, visible_alias = "skip-errors")] + pub continue_on_error: bool, + + /// Only show the deterministic plan without writing or deleting #[arg(short = 'n', long)] pub dry_run: bool, - /// Number of parallel operations - #[arg(short = 'P', long, default_value = "4")] - pub parallel: usize, + /// Maximum number of operations in flight across the command + #[arg(short = 'P', long, visible_alias = "parallel", default_value_t = 4)] + pub concurrency: usize, + + /// Aggregate transfer start rate in bytes per second (for example 10MiB/s) + #[arg(long)] + pub rate_limit: Option, + + /// Maximum attempts for transient transfer failures + #[arg(long, default_value_t = 3)] + pub retry_attempts: u32, - /// Disable progress bar + /// Initial transient retry backoff in milliseconds + #[arg(long, default_value_t = 100)] + pub retry_initial_backoff_ms: u64, + + /// Maximum transient retry backoff in milliseconds + #[arg(long, default_value_t = 10_000)] + pub retry_max_backoff_ms: u64, + + /// Print deterministic aggregate transfer counters + #[arg(long)] + pub summary: bool, + + /// Suppress non-error mirror output (legacy command-local alias) #[arg(long)] pub quiet: bool, } @@ -56,1064 +108,1674 @@ struct MirrorOutput { dry_run: bool, } -#[derive(Debug, Clone)] -struct FileInfo { - size: Option, - modified: Option, +#[derive(Debug, Clone, PartialEq, Eq)] +struct MirrorSnapshot { + size_bytes: Option, + modified: Option, etag: Option, } -struct MirrorTaskContext<'a> { - success_marker: &'a str, - operation: &'a str, - quiet: bool, - formatter: &'a Formatter, - progress: Option<&'a ProgressBar>, +#[derive(Debug, Clone, PartialEq, Eq)] +enum MirrorLocation { + Local(PathBuf), + Remote(RemotePath), } -trait MirrorCopySource { - async fn head_object_for_mirror( - &self, - path: &RemotePath, - ) -> rc_core::Result; +impl std::fmt::Display for MirrorLocation { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Local(path) => write!(formatter, "{}", path.display()), + Self::Remote(path) => write!(formatter, "{path}"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct MirrorEntry { + relative_path: String, + location: MirrorLocation, + snapshot: MirrorSnapshot, +} - async fn get_object_for_mirror(&self, path: &RemotePath) -> rc_core::Result>; +#[derive(Debug, Default)] +struct MirrorManifest { + entries: BTreeMap, + // A skipped source symlink protects the same destination path (and descendants) from removal. + protected_paths: Vec, + ignored: usize, } -trait MirrorCopyTarget { - async fn put_object_for_mirror( +#[derive(Debug, Clone)] +enum MirrorEndpointSpec { + Local(PathBuf), + Remote(RemotePath), +} + +impl MirrorEndpointSpec { + fn location_for(&self, relative_path: &str) -> rc_core::Result { + let relative_path = normalize_relative_path(relative_path)?; + match self { + Self::Local(root) => { + let mut target = root.clone(); + for component in relative_path.split('/') { + target.push(component); + } + Ok(MirrorLocation::Local(target)) + } + Self::Remote(root) => { + let prefix = normalized_remote_root_prefix(&root.key)?; + Ok(MirrorLocation::Remote(RemotePath::new( + &root.alias, + &root.bucket, + format!("{prefix}{relative_path}"), + ))) + } + } + } +} + +#[derive(Debug, Clone)] +struct MirrorCopyOperation { + relative_path: String, + source: MirrorEntry, + target: MirrorLocation, + target_before: Option, +} + +#[derive(Debug, Clone)] +struct MirrorRemoveOperation { + relative_path: String, + target: MirrorEntry, +} + +#[async_trait::async_trait] +trait MirrorIo: Send + Sync + 'static { + async fn copy(&self, operation: MirrorCopyOperation) -> rc_core::Result; + async fn remove(&self, operation: MirrorRemoveOperation) -> rc_core::Result; +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum RemoteWriteCondition { + IfAbsent, + IfMatch(String), +} + +#[async_trait::async_trait] +trait MirrorRemoteTransfer: Send + Sync { + async fn mirror_head(&self, path: &RemotePath) -> rc_core::Result; + async fn mirror_download(&self, path: &RemotePath, destination: &Path) -> rc_core::Result; + async fn mirror_upload( &self, path: &RemotePath, - data: Vec, + source: &Path, content_type: Option<&str>, - if_absent: bool, - ) -> rc_core::Result; + condition: RemoteWriteCondition, + ) -> rc_core::Result; } -impl MirrorCopySource for S3Client { - async fn head_object_for_mirror( - &self, - path: &RemotePath, - ) -> rc_core::Result { +#[async_trait::async_trait] +impl MirrorRemoteTransfer for S3Client { + async fn mirror_head(&self, path: &RemotePath) -> rc_core::Result { rc_core::ObjectStore::head_object(self, path).await } - async fn get_object_for_mirror(&self, path: &RemotePath) -> rc_core::Result> { - rc_core::ObjectStore::get_object(self, path).await + async fn mirror_download(&self, path: &RemotePath, destination: &Path) -> rc_core::Result { + self.download_object_to_path(path, destination, |_, _| {}) + .await } -} -impl MirrorCopyTarget for S3Client { - async fn put_object_for_mirror( + async fn mirror_upload( &self, path: &RemotePath, - data: Vec, + source: &Path, content_type: Option<&str>, - if_absent: bool, - ) -> rc_core::Result { - if if_absent { - self.put_object_if_absent(path, data, content_type).await - } else { - rc_core::ObjectStore::put_object(self, path, data, content_type, None).await + condition: RemoteWriteCondition, + ) -> rc_core::Result { + match condition { + RemoteWriteCondition::IfAbsent => { + self.put_object_from_path_if_absent(path, source, content_type, None, |_| {}) + .await + } + RemoteWriteCondition::IfMatch(etag) => { + self.put_object_from_path_if_match(path, source, content_type, None, &etag, |_| {}) + .await + } } } } -/// Execute the mirror command -pub async fn execute(args: MirrorArgs, output_config: OutputConfig) -> ExitCode { - let formatter = Formatter::new(output_config); - - if !(1..=256).contains(&args.parallel) { - formatter.error("--parallel must be between 1 and 256"); - return ExitCode::UsageError; - } - - // Parse both paths - let source_parsed = parse_path(&args.source); - let target_parsed = parse_path(&args.target); - - // Both must be remote for now - let (source_path, target_path) = match (&source_parsed, &target_parsed) { - (Ok(ParsedPath::Remote(s)), Ok(ParsedPath::Remote(t))) => (s.clone(), t.clone()), - (Ok(ParsedPath::Local(_)), _) | (_, Ok(ParsedPath::Local(_))) => { - formatter.error("Local paths are not yet supported in mirror command"); - return ExitCode::UsageError; - } - (Err(e), _) => { - formatter.error(&format!("Invalid source path: {e}")); - return ExitCode::UsageError; - } - (_, Err(e)) => { - formatter.error(&format!("Invalid target path: {e}")); - return ExitCode::UsageError; - } - }; - - // Load aliases - let alias_manager = match AliasManager::new() { - Ok(am) => am, - Err(e) => { - formatter.error(&format!("Failed to load aliases: {e}")); - return ExitCode::GeneralError; - } - }; +struct MirrorOperationReports { + copy: TransferReport, + remove: Option>, + blocked_removals: usize, +} - // Create clients - let source_alias = match alias_manager.get(&source_path.alias) { - Ok(a) => a, - Err(_) => { - formatter.error(&format!("Alias '{}' not found", source_path.alias)); - return ExitCode::NotFound; - } - }; +#[derive(Clone)] +enum RuntimeEndpoint { + Local { + root: PathBuf, + }, + Remote { + root: RemotePath, + client: Arc, + }, +} - let target_alias = match alias_manager.get(&target_path.alias) { - Ok(a) => a, - Err(_) => { - formatter.error(&format!("Alias '{}' not found", target_path.alias)); - return ExitCode::NotFound; +impl RuntimeEndpoint { + fn spec(&self) -> MirrorEndpointSpec { + match self { + Self::Local { root } => MirrorEndpointSpec::Local(root.clone()), + Self::Remote { root, .. } => MirrorEndpointSpec::Remote(root.clone()), } - }; - - if mirror_locations_overlap( - &source_path, - &target_path, - &source_alias.endpoint, - &target_alias.endpoint, - ) { - formatter.error("Mirror source and destination prefixes must not overlap"); - return ExitCode::UsageError; } - let source_client = Arc::new(match S3Client::new(source_alias).await { - Ok(c) => c, - Err(e) => { - formatter.error(&format!("Failed to create source client: {e}")); - return ExitCode::NetworkError; + async fn current_entry(&self, relative_path: &str) -> rc_core::Result> { + let location = self.spec().location_for(relative_path)?; + match (&location, self) { + (MirrorLocation::Local(path), Self::Local { root }) => { + inspect_local_entry(root, relative_path, path).await + } + (MirrorLocation::Remote(path), Self::Remote { client, .. }) => { + match client.head_object(path).await { + Ok(info) => Ok(Some(MirrorEntry { + relative_path: relative_path.to_string(), + location, + snapshot: snapshot_from_object(&info)?, + })), + Err(Error::NotFound(_)) => Ok(None), + Err(error) => Err(error), + } + } + _ => Err(Error::General( + "Mirror runtime endpoint does not match its planned location".to_string(), + )), } - }); + } +} - let target_client = Arc::new(match S3Client::new(target_alias).await { - Ok(c) => c, - Err(e) => { - formatter.error(&format!("Failed to create target client: {e}")); - return ExitCode::NetworkError; - } - }); +#[derive(Clone)] +struct LiveMirrorIo { + source: RuntimeEndpoint, + target: RuntimeEndpoint, +} - // List objects from both paths - let source_objects = match list_objects_map(&source_client, &source_path).await { - Ok(o) => o, - Err(e) => { - formatter.error(&format!("Failed to list source: {e}")); - return ExitCode::NetworkError; +#[async_trait::async_trait] +impl MirrorIo for LiveMirrorIo { + async fn copy(&self, operation: MirrorCopyOperation) -> rc_core::Result { + match (&operation.source.location, &operation.target) { + (MirrorLocation::Local(source), MirrorLocation::Remote(target)) => { + self.copy_local_to_remote(source, target, &operation).await + } + (MirrorLocation::Remote(source), MirrorLocation::Local(target)) => { + self.copy_remote_to_local(source, target, &operation).await + } + (MirrorLocation::Remote(source), MirrorLocation::Remote(target)) => { + self.copy_remote_to_remote(source, target, &operation).await + } + (MirrorLocation::Local(_), MirrorLocation::Local(_)) => Err(Error::UnsupportedFeature( + "Local-to-local mirror is out of scope; use a filesystem synchronization tool" + .to_string(), + )), } - }; + } - let target_objects = match list_objects_map(&target_client, &target_path).await { - Ok(o) => o, - Err(e) => { - formatter.error(&format!("Failed to list target: {e}")); - return ExitCode::NetworkError; + async fn remove(&self, operation: MirrorRemoveOperation) -> rc_core::Result { + let Some(current) = self.target.current_entry(&operation.relative_path).await? else { + // A previous attempt may have removed the entry before its response was lost. + return Ok(0); + }; + if current.snapshot != operation.target.snapshot { + return Err(Error::Conflict(format!( + "Destination changed before removal: {}", + operation.relative_path + ))); } - }; - - // Compare and determine operations - let diff_entries = compare_objects_internal(&source_objects, &target_objects); - let mut to_copy: Vec<(&str, &FileInfo, bool)> = Vec::new(); - let mut to_remove: Vec<(&str, &FileInfo)> = Vec::new(); - let mut skipped = 0; - - for entry in &diff_entries { - match entry.status { - DiffStatus::OnlyFirst => { - // New object, copy it - if let Some(info) = source_objects.get(&entry.key) { - to_copy.push((&entry.key, info, true)); - } - } - DiffStatus::Different => { - if args.overwrite { - // Different and overwrite enabled, copy it - if let Some(info) = source_objects.get(&entry.key) { - to_copy.push((&entry.key, info, false)); - } - } else { - skipped += 1; - } - } - DiffStatus::OnlySecond => { - if args.remove { - // Extra object at destination, remove it - if let Some(info) = target_objects.get(&entry.key) { - to_remove.push((&entry.key, info)); - } + match (¤t.location, &self.target) { + (MirrorLocation::Local(path), RuntimeEndpoint::Local { root }) => { + let safe_path = secure_local_path(root, &operation.relative_path, false).await?; + if &safe_path != path { + return Err(Error::InvalidPath(format!( + "Removal target escaped mirror root: {}", + path.display() + ))); } + tokio::fs::remove_file(path).await?; + Ok(0) } - DiffStatus::Same => { - skipped += 1; + (MirrorLocation::Remote(path), RuntimeEndpoint::Remote { client, .. }) => { + let etag = current.snapshot.etag.as_deref().ok_or_else(|| { + Error::Conflict(format!( + "Refusing to remove {} because its ETag is unavailable", + operation.relative_path + )) + })?; + client.delete_object_if_match(path, etag).await?; + Ok(0) } + _ => Err(Error::General( + "Mirror removal target does not match the runtime endpoint".to_string(), + )), } } +} - // Dry run output - if args.dry_run { - if !formatter.is_json() { - formatter.println("Dry run mode - no changes will be made:"); - formatter.println(""); - - if !to_copy.is_empty() { - formatter.println(&format!("Would copy {} object(s):", to_copy.len())); - for (key, info, _) in &to_copy { - let size = info - .size - .map(|s| humansize::format_size(s as u64, humansize::BINARY)) - .unwrap_or_default(); - formatter.println(&format!(" + {} ({size})", formatter.sanitize_text(key))); - } - formatter.println(""); +impl LiveMirrorIo { + async fn copy_local_to_remote( + &self, + source_path: &Path, + target_path: &RemotePath, + operation: &MirrorCopyOperation, + ) -> rc_core::Result { + validate_local_entry(&operation.source)?; + let staged = stage_local_source(source_path).await?; + async { + validate_local_entry(&operation.source)?; + let staged_size = tokio::fs::metadata(&staged).await?.len(); + if operation + .source + .snapshot + .size_bytes + .is_some_and(|expected| expected != staged_size) + { + return Err(Error::Conflict(format!( + "Source changed during mirror: {}", + operation.relative_path + ))); } - - if !to_remove.is_empty() { - formatter.println(&format!("Would remove {} object(s):", to_remove.len())); - for (key, _) in &to_remove { - formatter.println(&format!(" - {}", formatter.sanitize_text(key))); + match self.target_disposition(operation).await? { + TargetDisposition::AlreadyComplete => { + return Ok(operation.source.snapshot.size_bytes.unwrap_or_default()); } - formatter.println(""); + TargetDisposition::Ready => {} } - formatter.println(&format!( - "Summary: {} to copy, {} to remove, {} skipped", - to_copy.len(), - to_remove.len(), - skipped - )); - } else { - let output = MirrorOutput { - source: args.source.clone(), - target: args.target.clone(), - copied: to_copy.len(), - removed: to_remove.len(), - skipped, - errors: 0, - dry_run: true, + let RuntimeEndpoint::Remote { client, .. } = &self.target else { + return Err(Error::General( + "Remote mirror target client is unavailable".to_string(), + )); }; - formatter.json(&output); + let content_type = mime_guess::from_path(source_path) + .first() + .map(|value| value.essence_str().to_string()); + let size = operation.source.snapshot.size_bytes.unwrap_or_default(); + let condition = remote_write_condition(operation)?; + let info = client + .mirror_upload(target_path, &staged, content_type.as_deref(), condition) + .await?; + Ok(object_size(&info).unwrap_or(size)) } - return ExitCode::Success; + .await } - // Progress bar setup - let multi_progress = if !args.quiet && !formatter.is_json() { - Some(MultiProgress::new()) - } else { - None - }; - - let overall_pb = multi_progress.as_ref().map(|mp| { - let pb = mp.add(ProgressBar::new((to_copy.len() + to_remove.len()) as u64)); - pb.set_style( - ProgressStyle::default_bar() - .template("{spinner:.green} [{bar:40.cyan/blue}] {pos}/{len} {msg}") - .expect("Valid template") - .progress_chars("#>-"), - ); - pb.set_message("Syncing..."); - pb - }); - - // Perform copy operations - let mut copied = 0; - let mut errors = 0; - - let parallel_limit = args.parallel.max(1); - let mut copy_tasks: JoinSet<(String, Result<(), String>)> = JoinSet::new(); - let copy_context = MirrorTaskContext { - success_marker: "+", - operation: "copy", - quiet: args.quiet, - formatter: &formatter, - progress: overall_pb.as_ref(), - }; - - for (key, _, if_absent) in to_copy { - let source_sep = if source_path.key.is_empty() || source_path.key.ends_with('/') { - "" - } else { - "/" + async fn copy_remote_to_remote( + &self, + source_path: &RemotePath, + target_path: &RemotePath, + operation: &MirrorCopyOperation, + ) -> rc_core::Result { + let RuntimeEndpoint::Remote { + client: source_client, + .. + } = &self.source + else { + return Err(Error::General( + "Remote mirror source client is unavailable".to_string(), + )); }; - let source_full = RemotePath::new( - &source_path.alias, - &source_path.bucket, - format!("{}{source_sep}{key}", source_path.key), - ); - - let target_sep = if target_path.key.is_empty() || target_path.key.ends_with('/') { - "" - } else { - "/" + let RuntimeEndpoint::Remote { + client: target_client, + .. + } = &self.target + else { + return Err(Error::General( + "Remote mirror target client is unavailable".to_string(), + )); }; - let target_full = RemotePath::new( - &target_path.alias, - &target_path.bucket, - format!("{}{target_sep}{key}", target_path.key), - ); - - let key = key.to_string(); - let source_client = Arc::clone(&source_client); - let target_client = Arc::clone(&target_client); - copy_tasks.spawn(async move { - let result = copy_object_with_metadata( - source_client.as_ref(), - target_client.as_ref(), - &source_full, - &target_full, - if_absent, - ) - .await - .map_err(|e| format!("Failed to copy {key}: {e}")); - (key, result) - }); - if copy_tasks.len() >= parallel_limit - && let Some(task_result) = copy_tasks.join_next().await - { - record_mirror_task(task_result, ©_context, &mut copied, &mut errors); + match self.target_disposition(operation).await? { + TargetDisposition::AlreadyComplete => { + return Ok(operation.source.snapshot.size_bytes.unwrap_or_default()); + } + TargetDisposition::Ready => {} } + transfer_remote_to_remote( + source_client.as_ref(), + target_client.as_ref(), + source_path, + target_path, + operation, + remote_write_condition(operation)?, + ) + .await } - while let Some(task_result) = copy_tasks.join_next().await { - record_mirror_task(task_result, ©_context, &mut copied, &mut errors); - } - - // Perform remove operations - let mut removed = 0; - - if args.remove && errors == 0 { - let mut remove_tasks: JoinSet<(String, Result<(), String>)> = JoinSet::new(); - let remove_context = MirrorTaskContext { - success_marker: "-", - operation: "remove", - quiet: args.quiet, - formatter: &formatter, - progress: overall_pb.as_ref(), + async fn copy_remote_to_local( + &self, + source_path: &RemotePath, + target_path: &Path, + operation: &MirrorCopyOperation, + ) -> rc_core::Result { + let RuntimeEndpoint::Remote { + client: source_client, + .. + } = &self.source + else { + return Err(Error::General( + "Remote mirror source client is unavailable".to_string(), + )); + }; + let RuntimeEndpoint::Local { root } = &self.target else { + return Err(Error::General( + "Local mirror target root is unavailable".to_string(), + )); }; - for (key, info) in to_remove { - let sep = if target_path.key.is_empty() || target_path.key.ends_with('/') { - "" - } else { - "/" - }; - let target_full = RemotePath::new( - &target_path.alias, - &target_path.bucket, - format!("{}{sep}{key}", target_path.key), - ); - - let key = key.to_string(); - let etag = info.etag.clone(); - let target_client = Arc::clone(&target_client); - remove_tasks.spawn(async move { - let result = match etag { - Some(etag) => { - target_client - .delete_object_if_match(&target_full, &etag) - .await - } - None => Err(rc_core::Error::Conflict(format!( - "Refusing to remove {key} because its ETag is unavailable" - ))), - } - .map_err(|e| format!("Failed to remove {key}: {e}")); - (key, result) - }); - - if remove_tasks.len() >= parallel_limit - && let Some(task_result) = remove_tasks.join_next().await - { - record_mirror_task(task_result, &remove_context, &mut removed, &mut errors); + let before = source_client.head_object(source_path).await?; + ensure_snapshot_matches(&operation.source, &before)?; + match self.target_disposition(operation).await? { + TargetDisposition::AlreadyComplete => { + return Ok(operation.source.snapshot.size_bytes.unwrap_or_default()); } + TargetDisposition::Ready => {} } - while let Some(task_result) = remove_tasks.join_next().await { - record_mirror_task(task_result, &remove_context, &mut removed, &mut errors); + let destination = secure_local_path(root, &operation.relative_path, true).await?; + if destination != target_path { + return Err(Error::InvalidPath(format!( + "Download target escaped mirror root: {}", + target_path.display() + ))); } - } else if args.remove && errors > 0 && !formatter.is_json() { - formatter.error("Skipping removals because one or more copy operations failed"); - } - - if let Some(pb) = overall_pb { - pb.finish_with_message("Done"); - } - - // Output results - if formatter.is_json() { - let output = MirrorOutput { - source: args.source.clone(), - target: args.target.clone(), - copied, - removed, - skipped, - errors, - dry_run: false, + let staging = mirror_staging_path(target_path)?; + let downloaded = source_client + .download_object_to_path(source_path, &staging, |_, _| {}) + .await; + let validation = match downloaded { + Ok(bytes) => { + if operation + .source + .snapshot + .size_bytes + .is_some_and(|expected| expected != bytes) + { + Err(Error::Conflict(format!( + "Source changed during mirror: {}", + operation.relative_path + ))) + } else { + let after = source_client.head_object(source_path).await; + match after { + Ok(info) => { + ensure_snapshot_matches(&operation.source, &info).map(|()| bytes) + } + Err(error) => Err(error), + } + } + } + Err(error) => Err(error), }; - formatter.json(&output); - } else { - formatter.println(""); - formatter.println(&format!( - "Mirror complete: {copied} copied, {removed} removed, {skipped} skipped, {errors} errors" - )); - } - - if errors > 0 { - ExitCode::GeneralError - } else { - ExitCode::Success - } -} - -fn record_mirror_task( - task_result: Result<(String, Result<(), String>), JoinError>, - context: &MirrorTaskContext<'_>, - succeeded: &mut usize, - errors: &mut usize, -) { - match task_result { - Ok((key, Ok(()))) => { - *succeeded += 1; - if !context.quiet && !context.formatter.is_json() { - context.formatter.println(&format!( - "{} {}", - context.success_marker, - context.formatter.sanitize_text(&key) - )); + if validation.is_ok() { + match self.target_disposition(operation).await { + Ok(TargetDisposition::Ready) => {} + Ok(TargetDisposition::AlreadyComplete) => { + return Ok(operation.source.snapshot.size_bytes.unwrap_or_default()); + } + Err(error) => { + return Err(error); + } } } - Ok((_, Err(message))) => { - *errors += 1; - if !context.formatter.is_json() { - context.formatter.error(&message); - } + finish_staged_download( + staging, + target_path, + validation, + operation.source.snapshot.modified, + operation.target_before.is_some(), + ) + .await + } + + async fn target_disposition( + &self, + operation: &MirrorCopyOperation, + ) -> rc_core::Result { + let current = self.target.current_entry(&operation.relative_path).await?; + if let Some(current) = ¤t + && source_matches_target(&operation.source, current) + { + return Ok(TargetDisposition::AlreadyComplete); } - Err(join_error) => { - *errors += 1; - if !context.formatter.is_json() { - context.formatter.error(&format!( - "Mirror {} worker failed: {join_error}", - context.operation - )); + + match (&operation.target_before, current) { + (None, None) => Ok(TargetDisposition::Ready), + (Some(expected), Some(current)) if expected.snapshot == current.snapshot => { + Ok(TargetDisposition::Ready) } + _ => Err(Error::Conflict(format!( + "Destination changed after mirror planning: {}", + operation.relative_path + ))), } } +} + +enum TargetDisposition { + Ready, + AlreadyComplete, +} - if let Some(progress) = context.progress { - progress.inc(1); +fn remote_write_condition( + operation: &MirrorCopyOperation, +) -> rc_core::Result { + match &operation.target_before { + None => Ok(RemoteWriteCondition::IfAbsent), + Some(target) => target + .snapshot + .etag + .as_ref() + .map(|etag| RemoteWriteCondition::IfMatch(etag.clone())) + .ok_or_else(|| { + Error::Conflict(format!( + "Refusing to overwrite {} because its ETag is unavailable", + operation.relative_path + )) + }), } } -// Mirror should preserve source content type when metadata is available, but -// still fall back to a plain byte copy if metadata lookup fails for any reason. -async fn copy_object_with_metadata( +async fn transfer_remote_to_remote( source_client: &S, target_client: &T, source_path: &RemotePath, target_path: &RemotePath, - if_absent: bool, -) -> rc_core::Result<()> + operation: &MirrorCopyOperation, + condition: RemoteWriteCondition, +) -> rc_core::Result where - S: MirrorCopySource, - T: MirrorCopyTarget, + S: MirrorRemoteTransfer, + T: MirrorRemoteTransfer, { - let content_type = match source_client.head_object_for_mirror(source_path).await { - Ok(source_info) => source_info.content_type, - Err(error) => { - tracing::warn!( - source_alias = %source_path.alias, - source_bucket = %source_path.bucket, - source_key = %source_path.key, - error = %error, - "Falling back to mirror upload without source content type" - ); - None + let before = source_client.mirror_head(source_path).await?; + ensure_snapshot_matches(&operation.source, &before)?; + let staging = temporary_mirror_path("remote-source")?; + async { + let bytes = source_client.mirror_download(source_path, &staging).await?; + if operation + .source + .snapshot + .size_bytes + .is_some_and(|expected| expected != bytes) + { + return Err(Error::Conflict(format!( + "Source changed during mirror: {}", + operation.relative_path + ))); } - }; - - let data = source_client.get_object_for_mirror(source_path).await?; - target_client - .put_object_for_mirror(target_path, data, content_type.as_deref(), if_absent) - .await?; - Ok(()) + let after = source_client.mirror_head(source_path).await?; + ensure_snapshot_matches(&operation.source, &after)?; + let content_type = after + .content_type + .as_deref() + .or(before.content_type.as_deref()); + let info = target_client + .mirror_upload(target_path, &staging, content_type, condition) + .await?; + Ok(object_size(&info) + .or(operation.source.snapshot.size_bytes) + .unwrap_or(bytes)) + } + .await } -async fn list_objects_map( - client: &S3Client, - path: &RemotePath, -) -> Result, rc_core::Error> { - let mut objects = HashMap::new(); - let mut continuation_token: Option = None; - let base_prefix = &path.key; - - loop { - let options = ListOptions { - recursive: true, - max_keys: Some(1000), - continuation_token: continuation_token.clone(), - ..Default::default() - }; +/// Execute the mirror command. +pub async fn execute(args: MirrorArgs, mut output_config: OutputConfig) -> ExitCode { + output_config.quiet |= args.quiet; + let formatter = Formatter::new(output_config); - let result = client.list_objects(path, options).await?; + let selection = match build_selection(&args, Timestamp::now()) { + Ok(selection) => selection, + Err(error) => return formatter.fail(ExitCode::UsageError, &error), + }; + let controls = match build_controls(&args) { + Ok(controls) => controls, + Err(error) => return formatter.fail(ExitCode::UsageError, &error), + }; + let alias_manager = match AliasManager::new() { + Ok(manager) => manager, + Err(error) => { + return formatter.fail( + ExitCode::GeneralError, + &format!("Failed to load aliases: {error}"), + ); + } + }; + let source = match parse_mirror_path(&args.source, &alias_manager) { + Ok(path) => path, + Err(error) => { + return formatter.fail( + ExitCode::UsageError, + &format!("Invalid source path: {error}"), + ); + } + }; + let target = match parse_mirror_path(&args.target, &alias_manager) { + Ok(path) => path, + Err(error) => { + return formatter.fail( + ExitCode::UsageError, + &format!("Invalid target path: {error}"), + ); + } + }; + if matches!( + (&source, &target), + (ParsedPath::Local(_), ParsedPath::Local(_)) + ) { + return formatter.fail( + ExitCode::UnsupportedFeature, + "Local-to-local mirror is out of scope; use a filesystem synchronization tool", + ); + } + if let Err(error) = validate_remote_overlap(&source, &target, &alias_manager) { + return formatter.fail(exit_code_for_error(&error), &error.to_string()); + } - for item in result.items { - if item.is_dir { - continue; + let (source_runtime, source_manifest) = + match prepare_endpoint(&source, MissingRootPolicy::Error, &alias_manager).await { + Ok(prepared) => prepared, + Err(error) => { + return formatter.fail( + exit_code_for_error(&error), + &format!("Failed to enumerate mirror source: {error}"), + ); } - - // Get relative key (remove base prefix) - let relative_key = item.key.strip_prefix(base_prefix).unwrap_or(&item.key); - let relative_key = relative_key.trim_start_matches('/').to_string(); - - if relative_key.is_empty() { - continue; + }; + let (target_runtime, target_manifest) = + match prepare_endpoint(&target, MissingRootPolicy::Empty, &alias_manager).await { + Ok(prepared) => prepared, + Err(error) => { + return formatter.fail( + exit_code_for_error(&error), + &format!("Failed to enumerate mirror destination: {error}"), + ); } - - objects.insert( - relative_key, - FileInfo { - size: item.size_bytes, - modified: item.last_modified.map(|t| t.to_string()), - etag: item.etag, - }, - ); + }; + let copy_plan = match build_copy_plan( + &source_manifest, + &target_manifest, + &target_runtime.spec(), + &selection, + args.overwrite, + ) { + Ok(plan) => plan, + Err(error) => return formatter.fail(exit_code_for_error(&error), &error.to_string()), + }; + let remove_plan = if args.remove { + match build_remove_plan(&source_manifest, &target_manifest, &selection) { + Ok(plan) => plan, + Err(error) => return formatter.fail(exit_code_for_error(&error), &error.to_string()), } + } else { + empty_transfer_plan() + }; - if result.truncated { - continuation_token = result.continuation_token; - } else { - break; - } + if args.dry_run { + output_dry_run(&formatter, &args, ©_plan, &remove_plan); + return ExitCode::Success; } - Ok(objects) + let io = Arc::new(LiveMirrorIo { + source: source_runtime, + target: target_runtime, + }); + let reports = + match execute_operation_plans(io, copy_plan, remove_plan, controls, args.remove).await { + Ok(reports) => reports, + Err(error) => return formatter.fail(ExitCode::UsageError, &error.to_string()), + }; + output_reports(&formatter, &args, &reports); + + reports + .copy + .first_failure() + .or_else(|| { + reports + .remove + .as_ref() + .and_then(TransferReport::first_failure) + }) + .map_or(ExitCode::Success, exit_code_for_error) } -fn compare_objects_internal( - source: &HashMap, - target: &HashMap, -) -> Vec { - let mut entries = Vec::new(); - - // Check objects in source - for (key, source_info) in source { - if let Some(target_info) = target.get(key) { - // Object exists in both - let is_same = source_info.size == target_info.size - && matches!( - (&source_info.etag, &target_info.etag), - (Some(source_etag), Some(target_etag)) if source_etag == target_etag - ); +fn build_selection(args: &MirrorArgs, now: Timestamp) -> Result { + let newer_than = args + .newer_than + .as_deref() + .map(|value| parse_age_cutoff(value, now)) + .transpose()?; + let older_than = args + .older_than + .as_deref() + .map(|value| parse_age_cutoff(value, now)) + .transpose()?; + TransferSelection::new(&args.include, &args.exclude, newer_than, older_than, None) + .map_err(|error| error.to_string()) +} - let status = if is_same { - DiffStatus::Same - } else { - DiffStatus::Different - }; +fn build_controls(args: &MirrorArgs) -> Result { + let bytes_per_second = args + .rate_limit + .as_deref() + .map(parse_byte_rate) + .transpose()?; + let controls = TransferControls { + concurrency: args.concurrency, + bytes_per_second, + retry: RetryConfig { + max_attempts: args.retry_attempts, + initial_backoff_ms: args.retry_initial_backoff_ms, + max_backoff_ms: args.retry_max_backoff_ms, + }, + continue_on_error: args.continue_on_error, + }; + controls.validate().map_err(|error| error.to_string())?; + Ok(controls) +} - entries.push(DiffEntry { - key: key.clone(), - status, - first_size: source_info.size, - second_size: target_info.size, - first_modified: source_info.modified.clone(), - second_modified: target_info.modified.clone(), - }); - } else { - // Only in source - entries.push(DiffEntry { - key: key.clone(), - status: DiffStatus::OnlyFirst, - first_size: source_info.size, - second_size: None, - first_modified: source_info.modified.clone(), - second_modified: None, - }); +fn parse_mirror_path(raw: &str, alias_manager: &AliasManager) -> rc_core::Result { + let parsed = match parse_path(raw) { + Ok(parsed) => parsed, + Err(_) if !raw.is_empty() && Path::new(raw).exists() => { + return Ok(ParsedPath::Local(PathBuf::from(raw))); } + Err(error) => return Err(error), + }; + let ParsedPath::Remote(remote) = &parsed else { + return Ok(parsed); + }; + if matches!(alias_manager.exists(&remote.alias), Ok(true)) { + return Ok(parsed); + } + if Path::new(raw).exists() { + return Ok(ParsedPath::Local(PathBuf::from(raw))); } + Ok(parsed) +} - // Check objects only in target - for (key, target_info) in target { - if !source.contains_key(key) { - entries.push(DiffEntry { - key: key.clone(), - status: DiffStatus::OnlySecond, - first_size: None, - second_size: target_info.size, - first_modified: None, - second_modified: target_info.modified.clone(), +fn validate_remote_overlap( + source: &ParsedPath, + target: &ParsedPath, + alias_manager: &AliasManager, +) -> rc_core::Result<()> { + let (ParsedPath::Remote(source), ParsedPath::Remote(target)) = (source, target) else { + return Ok(()); + }; + let source_alias = alias_manager + .get(&source.alias) + .map_err(|_| Error::AliasNotFound(source.alias.clone()))?; + let target_alias = alias_manager + .get(&target.alias) + .map_err(|_| Error::AliasNotFound(target.alias.clone()))?; + if mirror_locations_overlap( + source, + target, + &source_alias.endpoint, + &target_alias.endpoint, + )? { + return Err(Error::InvalidPath( + "Mirror source and destination roots must not overlap".to_string(), + )); + } + Ok(()) +} + +async fn prepare_endpoint( + parsed: &ParsedPath, + missing_root: MissingRootPolicy, + alias_manager: &AliasManager, +) -> rc_core::Result<(RuntimeEndpoint, MirrorManifest)> { + match parsed { + ParsedPath::Local(root) => { + let manifest = enumerate_local_manifest(root, missing_root)?; + Ok((RuntimeEndpoint::Local { root: root.clone() }, manifest)) + } + ParsedPath::Remote(root) => { + let alias = alias_manager + .get(&root.alias) + .map_err(|_| Error::AliasNotFound(root.alias.clone()))?; + let planning_client = S3Client::new(alias.clone()).await?; + let manifest = enumerate_remote_manifest(&planning_client, root).await?; + let mut execution_alias = alias; + execution_alias.retry = Some(RetryConfig { + max_attempts: 1, + initial_backoff_ms: 1, + max_backoff_ms: 1, }); + let execution_client = Arc::new(S3Client::new(execution_alias).await?); + Ok(( + RuntimeEndpoint::Remote { + root: root.clone(), + client: execution_client, + }, + manifest, + )) } } +} - entries.sort_by(|a, b| a.key.cmp(&b.key)); - entries +#[derive(Debug, Clone, Copy)] +enum MissingRootPolicy { + Error, + Empty, } -fn mirror_locations_overlap( - source: &RemotePath, - target: &RemotePath, - source_endpoint: &str, - target_endpoint: &str, -) -> bool { - if source.bucket != target.bucket || !same_endpoint(source_endpoint, target_endpoint) { - return false; +fn enumerate_local_manifest( + root: &Path, + missing_root: MissingRootPolicy, +) -> rc_core::Result { + let metadata = match std::fs::symlink_metadata(root) { + Ok(metadata) => metadata, + Err(error) + if error.kind() == std::io::ErrorKind::NotFound + && matches!(missing_root, MissingRootPolicy::Empty) => + { + return Ok(MirrorManifest::default()); + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return Err(Error::NotFound(format!( + "Local mirror root not found: {}", + root.display() + ))); + } + Err(error) => return Err(Error::Io(error)), + }; + if metadata.file_type().is_symlink() { + return Err(Error::InvalidPath(format!( + "Local mirror root must not be a symbolic link: {}", + root.display() + ))); } - - let source = source.key.trim_matches('/'); - let target = target.key.trim_matches('/'); - if source.is_empty() || target.is_empty() || source == target { - return true; + if !metadata.is_dir() { + return Err(Error::InvalidPath(format!( + "Local mirror root must be a directory: {}", + root.display() + ))); } - target - .strip_prefix(source) - .is_some_and(|rest| rest.starts_with('/')) - || source - .strip_prefix(target) - .is_some_and(|rest| rest.starts_with('/')) + let mut manifest = MirrorManifest::default(); + enumerate_local_directory(root, root, &mut manifest)?; + manifest.protected_paths.sort(); + manifest.protected_paths.dedup(); + Ok(manifest) } -fn same_endpoint(first: &str, second: &str) -> bool { - match (url::Url::parse(first), url::Url::parse(second)) { - (Ok(first), Ok(second)) => { - first.scheme().eq_ignore_ascii_case(second.scheme()) - && first.host_str().zip(second.host_str()).is_some_and( - |(first_host, second_host)| first_host.eq_ignore_ascii_case(second_host), - ) - && first.port_or_known_default() == second.port_or_known_default() +fn enumerate_local_directory( + directory: &Path, + root: &Path, + manifest: &mut MirrorManifest, +) -> rc_core::Result<()> { + let mut entries = std::fs::read_dir(directory)?.collect::>>()?; + entries.sort_by_key(std::fs::DirEntry::file_name); + for entry in entries { + let path = entry.path(); + let relative = local_relative_path(root, &path)?; + let file_type = entry.file_type()?; + if file_type.is_symlink() { + manifest.protected_paths.push(relative); + manifest.ignored += 1; + continue; } - _ => first - .trim_end_matches('/') - .eq_ignore_ascii_case(second.trim_end_matches('/')), + if file_type.is_dir() { + enumerate_local_directory(&path, root, manifest)?; + continue; + } + if !file_type.is_file() { + manifest.protected_paths.push(relative); + manifest.ignored += 1; + continue; + } + let metadata = entry.metadata()?; + insert_manifest_entry( + &mut manifest.entries, + MirrorEntry { + relative_path: relative, + location: MirrorLocation::Local(path), + snapshot: snapshot_from_metadata(&metadata), + }, + )?; } + Ok(()) } -#[cfg(test)] -mod tests { - use super::*; - use rc_core::ObjectInfo; - use std::sync::Mutex; - - #[derive(Debug)] - struct TestMirrorSource { - content_type: Option, - data: Vec, - head_error: Option, - head_calls: Mutex>, - get_calls: Mutex>, - } - - impl TestMirrorSource { - fn new(content_type: Option<&str>, data: &[u8]) -> Self { - Self { - content_type: content_type.map(ToOwned::to_owned), - data: data.to_vec(), - head_error: None, - head_calls: Mutex::new(Vec::new()), - get_calls: Mutex::new(Vec::new()), +async fn enumerate_remote_manifest( + client: &S3Client, + root: &RemotePath, +) -> rc_core::Result { + let prefix = normalized_remote_root_prefix(&root.key)?; + let list_root = RemotePath::new(&root.alias, &root.bucket, &prefix); + let mut continuation_token = None; + let mut manifest = MirrorManifest::default(); + loop { + let result = client + .list_objects( + &list_root, + ListOptions { + recursive: true, + max_keys: Some(1_000), + continuation_token: continuation_token.clone(), + ..ListOptions::default() + }, + ) + .await?; + for object in result + .items + .into_iter() + .filter(|object| !object.is_dir && !object.key.ends_with('/')) + { + let raw_relative = object.key.strip_prefix(&prefix).ok_or_else(|| { + Error::InvalidPath(format!( + "Remote key '{}' is outside mirror root '{}'", + object.key, prefix + )) + })?; + if raw_relative.is_empty() { + continue; } + let relative_path = normalize_relative_path(raw_relative)?; + let snapshot = snapshot_from_object(&object)?; + insert_manifest_entry( + &mut manifest.entries, + MirrorEntry { + relative_path, + location: MirrorLocation::Remote(RemotePath::new( + &root.alias, + &root.bucket, + object.key, + )), + snapshot, + }, + )?; } - - fn with_head_error(error: rc_core::Error, data: &[u8]) -> Self { - Self { - content_type: None, - data: data.to_vec(), - head_error: Some(error), - head_calls: Mutex::new(Vec::new()), - get_calls: Mutex::new(Vec::new()), - } + if !result.truncated { + break; } + continuation_token = result.continuation_token; } + Ok(manifest) +} - impl MirrorCopySource for TestMirrorSource { - async fn head_object_for_mirror(&self, path: &RemotePath) -> rc_core::Result { - self.head_calls - .lock() - .expect("head call mutex should not be poisoned") - .push(path.key.clone()); - if let Some(error) = &self.head_error { - return Err(rc_core::Error::General(error.to_string())); - } - Ok(ObjectInfo { - key: path.key.clone(), - size_bytes: Some(self.data.len() as i64), - size_human: None, - last_modified: None, - etag: None, - storage_class: None, - content_type: self.content_type.clone(), - metadata: None, - is_dir: false, - }) - } +fn insert_manifest_entry( + entries: &mut BTreeMap, + entry: MirrorEntry, +) -> rc_core::Result<()> { + let relative_path = entry.relative_path.clone(); + if entries.insert(relative_path.clone(), entry).is_some() { + return Err(Error::Conflict(format!( + "Multiple mirror entries normalize to '{}'", + relative_path + ))); + } + Ok(()) +} - async fn get_object_for_mirror(&self, path: &RemotePath) -> rc_core::Result> { - self.get_calls - .lock() - .expect("get call mutex should not be poisoned") - .push(path.key.clone()); - Ok(self.data.clone()) +fn build_copy_plan( + source: &MirrorManifest, + target: &MirrorManifest, + target_endpoint: &MirrorEndpointSpec, + selection: &TransferSelection, + overwrite: bool, +) -> rc_core::Result> { + let mut candidates = Vec::with_capacity(source.entries.len()); + for entry in source.entries.values() { + if path_is_protected(&entry.relative_path, &target.protected_paths) { + return Err(Error::Conflict(format!( + "Destination path crosses a symbolic link or special file: {}", + entry.relative_path + ))); } + let target_before = target.entries.get(&entry.relative_path).cloned(); + let target_location = target_before.as_ref().map_or_else( + || target_endpoint.location_for(&entry.relative_path), + |target| Ok(target.location.clone()), + )?; + candidates.push(TransferCandidate { + payload: MirrorCopyOperation { + relative_path: entry.relative_path.clone(), + source: entry.clone(), + target: target_location.clone(), + target_before, + }, + source: entry.location.to_string(), + target: target_location.to_string(), + relative_path: entry.relative_path.clone(), + modified: entry.snapshot.modified, + size_bytes: entry.snapshot.size_bytes, + }); } - #[derive(Debug, PartialEq, Eq)] - struct PutCall { - key: String, - content_type: Option, - data: Vec, - if_absent: bool, - } - - #[derive(Debug, Default)] - struct TestMirrorTarget { - put_calls: Mutex>, - } - - impl MirrorCopyTarget for TestMirrorTarget { - async fn put_object_for_mirror( - &self, - path: &RemotePath, - data: Vec, - content_type: Option<&str>, - if_absent: bool, - ) -> rc_core::Result { - self.put_calls - .lock() - .expect("put call mutex should not be poisoned") - .push(PutCall { - key: path.key.clone(), - content_type: content_type.map(ToOwned::to_owned), - data: data.clone(), - if_absent, - }); - - Ok(ObjectInfo::file(path.key.clone(), data.len() as i64)) + let selected = TransferPlan::build(candidates, selection); + let mut skipped_after_selection = 0usize; + let mut items = Vec::new(); + for candidate in selected.items { + let should_copy = match &candidate.payload.target_before { + None => true, + Some(target) if source_matches_target(&candidate.payload.source, target) => false, + Some(_) => overwrite, + }; + if should_copy { + items.push(candidate); + } else { + skipped_after_selection += 1; } } + let mut summary = selected.summary; + summary.planned = items.len(); + summary.skipped = summary + .skipped + .saturating_add(skipped_after_selection) + .saturating_add(source.ignored); + Ok(TransferPlan { items, summary }) +} - #[tokio::test] - async fn test_copy_object_with_metadata_preserves_content_type() { - let source = TestMirrorSource::new(Some("image/jpeg"), b"jpeg-bytes"); - let target = TestMirrorTarget::default(); - let source_path = RemotePath::new("stage", "images", "photo.jpg"); - let target_path = RemotePath::new("prod", "images", "photo.jpg"); +fn build_remove_plan( + source: &MirrorManifest, + target: &MirrorManifest, + selection: &TransferSelection, +) -> rc_core::Result> { + let mut protected = 0usize; + let candidates = target + .entries + .values() + .filter_map(|entry| { + if source.entries.contains_key(&entry.relative_path) { + return None; + } + if path_is_protected(&entry.relative_path, &source.protected_paths) { + protected += 1; + return None; + } + Some(TransferCandidate { + payload: MirrorRemoveOperation { + relative_path: entry.relative_path.clone(), + target: entry.clone(), + }, + source: entry.location.to_string(), + target: entry.location.to_string(), + relative_path: entry.relative_path.clone(), + modified: entry.snapshot.modified, + size_bytes: Some(0), + }) + }) + .collect(); + let mut plan = TransferPlan::build(candidates, selection); + plan.summary.skipped = plan + .summary + .skipped + .saturating_add(protected) + .saturating_add(target.ignored); + Ok(plan) +} - copy_object_with_metadata(&source, &target, &source_path, &target_path, true) +async fn execute_operation_plans( + io: Arc, + copy_plan: TransferPlan, + remove_plan: TransferPlan, + controls: TransferControls, + remove_enabled: bool, +) -> rc_core::Result { + let executor = TransferExecutor::new(controls)?; + let copy = if copy_plan.items.is_empty() { + report_without_execution(copy_plan) + } else { + executor + .execute(copy_plan, { + let io = Arc::clone(&io); + move |candidate| { + let io = Arc::clone(&io); + async move { io.copy(candidate.payload).await } + } + }) .await - .expect("copy should succeed"); - - assert_eq!( - source - .head_calls - .lock() - .expect("head call mutex should not be poisoned") - .as_slice(), - ["photo.jpg"] - ); - assert_eq!( - source - .get_calls - .lock() - .expect("get call mutex should not be poisoned") - .as_slice(), - ["photo.jpg"] - ); - assert_eq!( - target - .put_calls - .lock() - .expect("put call mutex should not be poisoned") - .as_slice(), - [PutCall { - key: "photo.jpg".to_string(), - content_type: Some("image/jpeg".to_string()), - data: b"jpeg-bytes".to_vec(), - if_absent: true, - }] - ); - } + }; + let copies_succeeded = copy.summary.failed == 0 && copy.summary.cancelled == 0; + let blocked_removals = if remove_enabled && !copies_succeeded { + remove_plan.items.len() + } else { + 0 + }; + let remove = if remove_enabled && copies_succeeded { + Some(if remove_plan.items.is_empty() { + report_without_execution(remove_plan) + } else { + executor + .execute(remove_plan, move |candidate| { + let io = Arc::clone(&io); + async move { io.remove(candidate.payload).await } + }) + .await + }) + } else { + None + }; + Ok(MirrorOperationReports { + copy, + remove, + blocked_removals, + }) +} - #[tokio::test] - async fn test_copy_object_with_metadata_passes_none_when_source_has_no_content_type() { - let source = TestMirrorSource::new(None, b"plain-bytes"); - let target = TestMirrorTarget::default(); - let source_path = RemotePath::new("stage", "docs", "readme.txt"); - let target_path = RemotePath::new("prod", "docs", "readme.txt"); +fn report_without_execution(plan: TransferPlan) -> TransferReport { + TransferReport { + summary: plan.summary, + outcomes: Vec::new(), + } +} - copy_object_with_metadata(&source, &target, &source_path, &target_path, false) - .await - .expect("copy should succeed"); - - assert_eq!( - target - .put_calls - .lock() - .expect("put call mutex should not be poisoned") - .as_slice(), - [PutCall { - key: "readme.txt".to_string(), - content_type: None, - data: b"plain-bytes".to_vec(), - if_absent: false, - }] - ); +fn empty_transfer_plan() -> TransferPlan { + TransferPlan { + items: Vec::new(), + summary: TransferSummary::default(), } +} - #[tokio::test] - async fn test_copy_object_with_metadata_falls_back_when_head_lookup_fails() { - let source = TestMirrorSource::with_head_error( - rc_core::Error::Network("head failed".to_string()), - b"plain-bytes", - ); - let target = TestMirrorTarget::default(); - let source_path = RemotePath::new("stage", "docs", "readme.txt"); - let target_path = RemotePath::new("prod", "docs", "readme.txt"); +fn output_dry_run( + formatter: &Formatter, + args: &MirrorArgs, + copy_plan: &TransferPlan, + remove_plan: &TransferPlan, +) { + if formatter.is_json() { + formatter.json(&MirrorOutput { + source: args.source.clone(), + target: args.target.clone(), + copied: copy_plan.items.len(), + removed: remove_plan.items.len(), + skipped: copy_plan + .summary + .skipped + .saturating_add(remove_plan.summary.skipped), + errors: 0, + dry_run: true, + }); + return; + } + for candidate in ©_plan.items { + formatter.println(&format!( + "Would copy: {} -> {}", + formatter.style_file(&candidate.source), + formatter.style_file(&candidate.target) + )); + } + for candidate in &remove_plan.items { + formatter.println(&format!( + "Would remove: {}", + formatter.style_file(&candidate.target) + )); + } + if args.summary { + formatter.println(&format!( + "Summary: {} copies, {} removals, {} skipped", + copy_plan.items.len(), + remove_plan.items.len(), + copy_plan + .summary + .skipped + .saturating_add(remove_plan.summary.skipped) + )); + } +} - copy_object_with_metadata(&source, &target, &source_path, &target_path, false) - .await - .expect("copy should succeed"); - - assert_eq!( - source - .head_calls - .lock() - .expect("head call mutex should not be poisoned") - .as_slice(), - ["readme.txt"] - ); - assert_eq!( - source - .get_calls - .lock() - .expect("get call mutex should not be poisoned") - .as_slice(), - ["readme.txt"] - ); - assert_eq!( - target - .put_calls - .lock() - .expect("put call mutex should not be poisoned") - .as_slice(), - [PutCall { - key: "readme.txt".to_string(), - content_type: None, - data: b"plain-bytes".to_vec(), - if_absent: false, - }] - ); +fn output_reports(formatter: &Formatter, args: &MirrorArgs, reports: &MirrorOperationReports) { + if !formatter.is_json() { + output_outcomes(formatter, "+", &reports.copy); + if let Some(remove) = &reports.remove { + output_outcomes(formatter, "-", remove); + } else if reports.blocked_removals > 0 { + formatter.error("Skipping removals because one or more copy operations failed"); + } } + let removed = reports + .remove + .as_ref() + .map_or(0, |report| report.summary.successful); + let remove_failed = reports + .remove + .as_ref() + .map_or(0, |report| report.summary.failed); + let remove_skipped = reports + .remove + .as_ref() + .map_or(reports.blocked_removals, |report| report.summary.skipped); + let output = MirrorOutput { + source: args.source.clone(), + target: args.target.clone(), + copied: reports.copy.summary.successful, + removed, + skipped: reports.copy.summary.skipped.saturating_add(remove_skipped), + errors: reports.copy.summary.failed.saturating_add(remove_failed), + dry_run: false, + }; + if formatter.is_json() { + formatter.json(&output); + } else if args.summary { + let remove_cancelled = reports + .remove + .as_ref() + .map_or(0, |report| report.summary.cancelled); + let cancelled = reports + .copy + .summary + .cancelled + .saturating_add(remove_cancelled); + formatter.println(&format_human_summary( + &output, + cancelled, + reports.copy.summary.transferred_bytes, + )); + } +} - #[test] - fn test_compare_objects_internal() { - let mut source = HashMap::new(); - source.insert( - "file1.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: Some("abc".to_string()), - }, - ); - source.insert( - "file2.txt".to_string(), - FileInfo { - size: Some(200), - modified: None, - etag: Some("def".to_string()), - }, - ); +fn format_human_summary(output: &MirrorOutput, cancelled: usize, transferred_bytes: u64) -> String { + format!( + "Summary: {} copied, {} removed, {} skipped, {} errors, {} cancelled, {} transferred", + output.copied, + output.removed, + output.skipped, + output.errors, + cancelled, + humansize::format_size(transferred_bytes, humansize::BINARY) + ) +} - let mut target = HashMap::new(); - target.insert( - "file1.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: Some("abc".to_string()), - }, - ); - target.insert( - "file3.txt".to_string(), - FileInfo { - size: Some(300), - modified: None, - etag: Some("ghi".to_string()), - }, - ); +fn output_outcomes(formatter: &Formatter, marker: &str, report: &TransferReport) { + for outcome in &report.outcomes { + match &outcome.state { + TransferOutcomeState::Success { .. } => formatter.println(&format!( + "{marker} {}", + formatter.sanitize_text(&outcome.item.relative_path) + )), + TransferOutcomeState::Failed { error } => formatter.error_with_code( + exit_code_for_error(error), + &format!( + "Mirror operation failed for '{}': {error}", + formatter.sanitize_text(&outcome.item.relative_path) + ), + ), + TransferOutcomeState::Cancelled => formatter.warning(&format!( + "Cancelled before mirror operation: {}", + formatter.sanitize_text(&outcome.item.relative_path) + )), + } + } +} + +fn normalize_relative_path(value: &str) -> rc_core::Result { + if value.starts_with(['/', '\\']) || value.contains('\\') { + return Err(Error::InvalidPath(format!( + "Mirror path must be relative and use '/' separators: {value}" + ))); + } + let mut normalized = Vec::new(); + for component in value.split('/') { + if component.is_empty() || component == "." { + continue; + } + if component == ".." { + return Err(Error::InvalidPath( + "Mirror paths must not contain traversal components".to_string(), + )); + } + validate_portable_component(component)?; + normalized.push(component); + } + if normalized.is_empty() { + return Err(Error::InvalidPath( + "Mirror path does not contain a file name".to_string(), + )); + } + Ok(normalized.join("/")) +} - let entries = compare_objects_internal(&source, &target); - assert_eq!(entries.len(), 3); +fn validate_portable_component(component: &str) -> rc_core::Result<()> { + if component.chars().any(|character| { + character.is_control() || matches!(character, ':' | '<' | '>' | '"' | '|' | '?' | '*') + }) || component.ends_with(['.', ' ']) + { + return Err(Error::InvalidPath(format!( + "Mirror path component is not portable: {component}" + ))); + } + let stem = component.split('.').next().unwrap_or_default(); + let stem = stem.to_ascii_uppercase(); + if matches!(stem.as_str(), "CON" | "PRN" | "AUX" | "NUL") + || (stem.len() == 4 + && (stem.starts_with("COM") || stem.starts_with("LPT")) + && matches!(stem.as_bytes()[3], b'1'..=b'9')) + { + return Err(Error::InvalidPath(format!( + "Mirror path uses a reserved device name: {component}" + ))); + } + Ok(()) +} - // file1.txt should be Same - let f1 = entries.iter().find(|e| e.key == "file1.txt").unwrap(); - assert_eq!(f1.status, DiffStatus::Same); +fn local_relative_path(root: &Path, path: &Path) -> rc_core::Result { + let relative = path + .strip_prefix(root) + .map_err(|error| Error::InvalidPath(error.to_string()))?; + let mut components = Vec::new(); + for component in relative.components() { + let Component::Normal(component) = component else { + return Err(Error::InvalidPath(format!( + "Local mirror entry is not relative to its root: {}", + path.display() + ))); + }; + let component = component.to_str().ok_or_else(|| { + Error::InvalidPath(format!( + "Local mirror entry is not valid UTF-8: {}", + path.display() + )) + })?; + components.push(component); + } + normalize_relative_path(&components.join("/")) +} - // file2.txt should be OnlyFirst - let f2 = entries.iter().find(|e| e.key == "file2.txt").unwrap(); - assert_eq!(f2.status, DiffStatus::OnlyFirst); +fn normalized_remote_root_prefix(key: &str) -> rc_core::Result { + if key.starts_with('/') { + return Err(Error::InvalidPath( + "Remote mirror roots must not start with '/'".to_string(), + )); + } + let key = key.trim_end_matches('/'); + if key.is_empty() { + return Ok(String::new()); + } + Ok(format!("{}/", normalize_relative_path(key)?)) +} - // file3.txt should be OnlySecond - let f3 = entries.iter().find(|e| e.key == "file3.txt").unwrap(); - assert_eq!(f3.status, DiffStatus::OnlySecond); +fn snapshot_from_metadata(metadata: &std::fs::Metadata) -> MirrorSnapshot { + MirrorSnapshot { + size_bytes: Some(metadata.len()), + modified: metadata + .modified() + .ok() + .and_then(|value| value.try_into().ok()), + etag: None, } +} - #[test] - fn test_compare_empty_source() { - let source: HashMap = HashMap::new(); - let mut target = HashMap::new(); - target.insert( - "file.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: Some("abc".to_string()), - }, - ); +fn snapshot_from_object(object: &ObjectInfo) -> rc_core::Result { + let size_bytes = object + .size_bytes + .map(|size| { + u64::try_from(size).map_err(|_| { + Error::Network(format!( + "S3 returned a negative object size for '{}'", + object.key + )) + }) + }) + .transpose()?; + Ok(MirrorSnapshot { + size_bytes, + modified: object.last_modified, + etag: object.etag.clone(), + }) +} + +fn object_size(object: &ObjectInfo) -> Option { + object.size_bytes.and_then(|size| u64::try_from(size).ok()) +} - let entries = compare_objects_internal(&source, &target); - assert_eq!(entries.len(), 1); - assert_eq!(entries[0].status, DiffStatus::OnlySecond); +fn source_matches_target(source: &MirrorEntry, target: &MirrorEntry) -> bool { + let (Some(source_size), Some(target_size)) = + (source.snapshot.size_bytes, target.snapshot.size_bytes) + else { + return false; + }; + if source_size != target_size { + return false; } + if let (Some(source_etag), Some(target_etag)) = (&source.snapshot.etag, &target.snapshot.etag) { + return source_etag == target_etag; + } + match (&source.location, &target.location) { + (MirrorLocation::Remote(_), MirrorLocation::Remote(_)) => false, + (MirrorLocation::Local(_), MirrorLocation::Remote(_)) => { + match (source.snapshot.modified, target.snapshot.modified) { + (Some(source_modified), Some(target_modified)) => { + target_modified >= source_modified + } + _ => false, + } + } + _ => { + source.snapshot.modified.is_some() + && source.snapshot.modified == target.snapshot.modified + } + } +} - #[test] - fn test_compare_empty_target() { - let mut source = HashMap::new(); - source.insert( - "file.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: Some("abc".to_string()), - }, - ); - let target: HashMap = HashMap::new(); +fn path_is_protected(relative_path: &str, protected_paths: &[String]) -> bool { + protected_paths.iter().any(|protected| { + relative_path == protected + || relative_path + .strip_prefix(protected) + .is_some_and(|suffix| suffix.starts_with('/')) + }) +} - let entries = compare_objects_internal(&source, &target); - assert_eq!(entries.len(), 1); - assert_eq!(entries[0].status, DiffStatus::OnlyFirst); +fn validate_local_entry(entry: &MirrorEntry) -> rc_core::Result<()> { + let MirrorLocation::Local(path) = &entry.location else { + return Err(Error::General( + "Expected a local mirror source entry".to_string(), + )); + }; + let metadata = std::fs::symlink_metadata(path)?; + if metadata.file_type().is_symlink() || !metadata.is_file() { + return Err(Error::Conflict(format!( + "Source changed during mirror: {}", + entry.relative_path + ))); } + if snapshot_from_metadata(&metadata) != entry.snapshot { + return Err(Error::Conflict(format!( + "Source changed during mirror: {}", + entry.relative_path + ))); + } + Ok(()) +} - #[test] - fn test_compare_both_empty() { - let source: HashMap = HashMap::new(); - let target: HashMap = HashMap::new(); +fn ensure_snapshot_matches(entry: &MirrorEntry, object: &ObjectInfo) -> rc_core::Result<()> { + if snapshot_from_object(object)? != entry.snapshot { + return Err(Error::Conflict(format!( + "Source changed during mirror: {}", + entry.relative_path + ))); + } + Ok(()) +} - let entries = compare_objects_internal(&source, &target); - assert!(entries.is_empty()); +async fn stage_local_source(source: &Path) -> rc_core::Result { + let mut input = tokio::fs::File::open(source).await?; + let temporary = tempfile::Builder::new() + .prefix("rc-mirror-local-source-") + .suffix(".part") + .tempfile()?; + let (output, staging) = temporary.into_parts(); + let mut output = tokio::fs::File::from_std(output); + if let Err(error) = tokio::io::copy(&mut input, &mut output).await { + return Err(Error::Io(error)); } + if let Err(error) = output.flush().await { + return Err(Error::Io(error)); + } + Ok(staging) +} - #[test] - fn test_compare_different_sizes() { - let mut source = HashMap::new(); - source.insert( - "file.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: Some("abc".to_string()), - }, - ); +fn temporary_mirror_path(label: &str) -> rc_core::Result { + Ok(tempfile::Builder::new() + .prefix(&format!("rc-mirror-{label}-")) + .suffix(".part") + .tempfile()? + .into_temp_path()) +} - let mut target = HashMap::new(); - target.insert( - "file.txt".to_string(), - FileInfo { - size: Some(200), // Different size - modified: None, - etag: Some("def".to_string()), - }, - ); +fn mirror_staging_path(destination: &Path) -> rc_core::Result { + let parent = destination.parent().ok_or_else(|| { + Error::InvalidPath(format!( + "Mirror destination has no parent: {}", + destination.display() + )) + })?; + let file_name = destination.file_name().ok_or_else(|| { + Error::InvalidPath(format!( + "Mirror destination has no file name: {}", + destination.display() + )) + })?; + Ok(tempfile::Builder::new() + .prefix(&format!(".{}.rc-mirror-", file_name.to_string_lossy())) + .tempfile_in(parent)? + .into_temp_path()) +} - let entries = compare_objects_internal(&source, &target); - assert_eq!(entries.len(), 1); - assert_eq!(entries[0].status, DiffStatus::Different); +async fn finish_staged_download( + staging: tempfile::TempPath, + destination: &Path, + validation: rc_core::Result, + modified: Option, + replace_existing: bool, +) -> rc_core::Result { + let bytes = match validation { + Ok(bytes) => bytes, + Err(error) => return Err(error), + }; + if let Some(modified) = modified + && let Err(error) = set_file_modified(&staging, modified) + { + return Err(error); } + let persisted = if replace_existing { + staging.persist(destination) + } else { + staging.persist_noclobber(destination) + }; + persisted.map_err(|error| { + if !replace_existing && error.error.kind() == std::io::ErrorKind::AlreadyExists { + Error::Conflict(format!( + "Destination appeared before mirror completion: {}", + destination.display() + )) + } else { + Error::General(format!( + "Failed to atomically replace mirror destination '{}': {}", + destination.display(), + error.error + )) + } + })?; + Ok(bytes) +} - #[test] - fn test_compare_missing_etag_is_different() { - let source = HashMap::from([( - "file.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: None, - }, - )]); - let target = HashMap::from([( - "file.txt".to_string(), - FileInfo { - size: Some(100), - modified: None, - etag: Some("target-etag".to_string()), - }, - )]); +fn set_file_modified(path: &Path, modified: Timestamp) -> rc_core::Result<()> { + let modified: std::time::SystemTime = modified.into(); + let file = std::fs::OpenOptions::new().write(true).open(path)?; + file.set_times(std::fs::FileTimes::new().set_modified(modified))?; + Ok(()) +} - let entries = compare_objects_internal(&source, &target); +async fn inspect_local_entry( + root: &Path, + relative_path: &str, + expected_path: &Path, +) -> rc_core::Result> { + let path = secure_local_path(root, relative_path, false).await?; + if path != expected_path { + return Err(Error::InvalidPath(format!( + "Local mirror target escaped its root: {}", + expected_path.display() + ))); + } + match tokio::fs::symlink_metadata(&path).await { + Ok(metadata) if metadata.file_type().is_symlink() => Err(Error::InvalidPath(format!( + "Destination is a symbolic link: {}", + path.display() + ))), + Ok(metadata) if metadata.is_file() => Ok(Some(MirrorEntry { + relative_path: relative_path.to_string(), + location: MirrorLocation::Local(path), + snapshot: snapshot_from_metadata(&metadata), + })), + Ok(_) => Err(Error::InvalidPath(format!( + "Destination is not a regular file: {}", + path.display() + ))), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(Error::Io(error)), + } +} - assert_eq!(entries[0].status, DiffStatus::Different); +async fn secure_local_path( + root: &Path, + relative_path: &str, + create_parents: bool, +) -> rc_core::Result { + let relative_path = normalize_relative_path(relative_path)?; + match tokio::fs::symlink_metadata(root).await { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(Error::InvalidPath(format!( + "Local mirror root is a symbolic link: {}", + root.display() + ))); + } + Ok(metadata) if !metadata.is_dir() => { + return Err(Error::InvalidPath(format!( + "Local mirror root is not a directory: {}", + root.display() + ))); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound && create_parents => { + create_local_directory_tree(root).await?; + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(Error::Io(error)), } - #[test] - fn mirror_rejects_overlapping_prefixes_across_aliases_for_same_endpoint() { - let source = RemotePath::new("source", "bucket", "data/source"); - let nested = RemotePath::new("target", "bucket", "data/source/archive"); - let adjacent = RemotePath::new("target", "bucket", "data/source-archive"); + let components = relative_path.split('/').collect::>(); + let mut path = root.to_path_buf(); + for (index, component) in components.iter().enumerate() { + path.push(component); + let is_final = index + 1 == components.len(); + match tokio::fs::symlink_metadata(&path).await { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(Error::InvalidPath(format!( + "Local mirror path component is a symbolic link: {}", + path.display() + ))); + } + Ok(metadata) if !is_final && !metadata.is_dir() => { + return Err(Error::InvalidPath(format!( + "Local mirror parent is not a directory: {}", + path.display() + ))); + } + Ok(_) => {} + Err(error) + if error.kind() == std::io::ErrorKind::NotFound && create_parents && !is_final => + { + match tokio::fs::create_dir(&path).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(Error::Io(error)), + } + let metadata = tokio::fs::symlink_metadata(&path).await?; + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(Error::InvalidPath(format!( + "Unsafe local mirror parent: {}", + path.display() + ))); + } + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(Error::Io(error)), + } + } + Ok(path) +} - assert!(mirror_locations_overlap( - &source, - &nested, - "https://s3.example.com/", - "https://S3.EXAMPLE.COM:443" - )); - assert!(!mirror_locations_overlap( - &source, - &adjacent, - "https://s3.example.com", - "https://s3.example.com" - )); +async fn create_local_directory_tree(root: &Path) -> rc_core::Result<()> { + let mut missing = Vec::new(); + let mut cursor = root.to_path_buf(); + loop { + if cursor.as_os_str().is_empty() { + cursor = PathBuf::from("."); + } + match tokio::fs::symlink_metadata(&cursor).await { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(Error::InvalidPath(format!( + "Local mirror directory component is a symbolic link: {}", + cursor.display() + ))); + } + Ok(metadata) if !metadata.is_dir() => { + return Err(Error::InvalidPath(format!( + "Local mirror directory component is not a directory: {}", + cursor.display() + ))); + } + Ok(_) => break, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + missing.push(cursor.clone()); + cursor = cursor.parent().map(Path::to_path_buf).ok_or_else(|| { + Error::InvalidPath(format!( + "Local mirror root has no existing ancestor: {}", + root.display() + )) + })?; + } + Err(error) => return Err(Error::Io(error)), + } } - #[test] - fn test_mirror_args_defaults() { - let args = MirrorArgs { - source: "src".to_string(), - target: "dst".to_string(), - remove: false, - overwrite: false, - dry_run: false, - parallel: 4, - quiet: false, - }; - assert_eq!(args.parallel, 4); - assert!(!args.remove); - assert!(!args.overwrite); - } - - #[test] - fn test_mirror_output_serialization() { - let output = MirrorOutput { - source: "src/".to_string(), - target: "dst/".to_string(), - copied: 10, - removed: 2, - skipped: 5, - errors: 0, - dry_run: false, - }; - let json = serde_json::to_string(&output).unwrap(); - assert!(json.contains("\"copied\":10")); - assert!(json.contains("\"removed\":2")); - assert!(json.contains("\"dry_run\":false")); + for directory in missing.into_iter().rev() { + match tokio::fs::create_dir(&directory).await { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} + Err(error) => return Err(Error::Io(error)), + } + let metadata = tokio::fs::symlink_metadata(&directory).await?; + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(Error::InvalidPath(format!( + "Unsafe local mirror directory: {}", + directory.display() + ))); + } + } + Ok(()) +} + +fn mirror_locations_overlap( + source: &RemotePath, + target: &RemotePath, + source_endpoint: &str, + target_endpoint: &str, +) -> rc_core::Result { + if source.bucket != target.bucket || !same_endpoint(source_endpoint, target_endpoint) { + return Ok(false); } + let source = normalized_remote_root_prefix(&source.key)?; + let target = normalized_remote_root_prefix(&target.key)?; + Ok(source.is_empty() + || target.is_empty() + || source == target + || source.starts_with(&target) + || target.starts_with(&source)) } + +fn same_endpoint(first: &str, second: &str) -> bool { + match (url::Url::parse(first), url::Url::parse(second)) { + (Ok(first), Ok(second)) => { + first.scheme().eq_ignore_ascii_case(second.scheme()) + && first.host_str().zip(second.host_str()).is_some_and( + |(first_host, second_host)| first_host.eq_ignore_ascii_case(second_host), + ) + && first.port_or_known_default() == second.port_or_known_default() + } + _ => first + .trim_end_matches('/') + .eq_ignore_ascii_case(second.trim_end_matches('/')), + } +} + +fn exit_code_for_error(error: &Error) -> ExitCode { + ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError) +} + +#[cfg(test)] +mod roadmap_tests; diff --git a/crates/cli/src/commands/mirror/roadmap_tests.rs b/crates/cli/src/commands/mirror/roadmap_tests.rs new file mode 100644 index 00000000..c02c9b49 --- /dev/null +++ b/crates/cli/src/commands/mirror/roadmap_tests.rs @@ -0,0 +1,1035 @@ +use std::collections::{BTreeMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; + +use clap::Parser; +use jiff::Timestamp; +use rc_core::{Error, RemotePath, TransferControls, TransferOutcomeState, TransferSelection}; + +use super::*; + +fn snapshot(size: u64, modified: &str, etag: Option<&str>) -> MirrorSnapshot { + MirrorSnapshot { + size_bytes: Some(size), + modified: Some(modified.parse::().expect("valid test timestamp")), + etag: etag.map(ToOwned::to_owned), + } +} + +fn remote_entry(alias: &str, key: &str, relative: &str, etag: &str) -> MirrorEntry { + MirrorEntry { + relative_path: relative.to_string(), + location: MirrorLocation::Remote(RemotePath::new(alias, "bucket", key)), + snapshot: snapshot(4, "2026-07-21T04:00:00Z", Some(etag)), + } +} + +fn local_entry(root: &str, relative: &str) -> MirrorEntry { + MirrorEntry { + relative_path: relative.to_string(), + location: MirrorLocation::Local(PathBuf::from(root).join(relative)), + snapshot: snapshot(4, "2026-07-21T04:00:00Z", None), + } +} + +fn manifest(entries: impl IntoIterator) -> MirrorManifest { + let entries = entries + .into_iter() + .map(|entry| (entry.relative_path.clone(), entry)) + .collect(); + MirrorManifest { + entries, + ..MirrorManifest::default() + } +} + +#[test] +fn relative_paths_are_normalized_and_traversal_is_rejected() { + assert_eq!( + normalize_relative_path("nested//./report.txt").expect("normal relative path"), + "nested/report.txt" + ); + for value in [ + "../secret", + "nested/../../secret", + "/absolute", + "C:/windows", + "nested/bad?.txt", + "nested/control\u{0007}.txt", + ] { + assert!(normalize_relative_path(value).is_err(), "accepted {value}"); + } +} + +#[test] +fn planners_map_all_supported_directions_without_changing_relative_paths() { + let cases = [ + ( + manifest([local_entry("/source", "nested/report.txt")]), + MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + MirrorLocation::Remote(RemotePath::new("dst", "bucket", "backup/nested/report.txt")), + ), + ( + manifest([remote_entry( + "src", + "source/nested/report.txt", + "nested/report.txt", + "etag-1", + )]), + MirrorEndpointSpec::Local(PathBuf::from("/target")), + MirrorLocation::Local(PathBuf::from("/target").join("nested").join("report.txt")), + ), + ( + manifest([remote_entry( + "src", + "source/nested/report.txt", + "nested/report.txt", + "etag-1", + )]), + MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + MirrorLocation::Remote(RemotePath::new("dst", "bucket", "backup/nested/report.txt")), + ), + ]; + + for (source, target, expected_target) in cases { + let plan = build_copy_plan( + &source, + &MirrorManifest::default(), + &target, + &TransferSelection::default(), + false, + ) + .expect("build copy plan"); + assert_eq!(plan.items.len(), 1); + assert_eq!(plan.items[0].relative_path, "nested/report.txt"); + assert_eq!(plan.items[0].payload.target, expected_target); + } +} + +#[test] +fn equivalent_destination_is_restart_safe_and_not_planned_again() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "etag-1", + )]); + let target = manifest([remote_entry( + "dst", + "backup/report.txt", + "report.txt", + "etag-1", + )]); + + let plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + ) + .expect("build restart plan"); + + assert!(plan.items.is_empty()); + assert_eq!(plan.summary.skipped, 1); +} + +#[test] +fn changed_destination_requires_overwrite() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "etag-source", + )]); + let target = manifest([remote_entry( + "dst", + "backup/report.txt", + "report.txt", + "etag-target", + )]); + let destination = MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")); + + let skipped = build_copy_plan( + &source, + &target, + &destination, + &TransferSelection::default(), + false, + ) + .expect("build non-overwrite plan"); + let overwritten = build_copy_plan( + &source, + &target, + &destination, + &TransferSelection::default(), + true, + ) + .expect("build overwrite plan"); + + assert!(skipped.items.is_empty()); + assert_eq!(skipped.summary.skipped, 1); + assert_eq!(overwritten.items.len(), 1); +} + +#[test] +fn removals_never_cross_a_source_symlink_boundary() { + let mut source = MirrorManifest::default(); + source.protected_paths.push("linked".to_string()); + let target = manifest([ + remote_entry("dst", "backup/linked/a.txt", "linked/a.txt", "a"), + remote_entry("dst", "backup/stale.txt", "stale.txt", "b"), + ]); + + let plan = build_remove_plan(&source, &target, &TransferSelection::default()) + .expect("build safe removal plan"); + + assert_eq!(plan.items.len(), 1); + assert_eq!(plan.items[0].relative_path, "stale.txt"); + assert_eq!(plan.summary.skipped, 1); +} + +#[cfg(unix)] +#[test] +fn local_enumeration_skips_symlinks_without_following_them() { + use std::os::unix::fs::symlink; + + let root = tempfile::tempdir().expect("create local mirror root"); + let outside = tempfile::tempdir().expect("create outside directory"); + std::fs::create_dir_all(root.path().join("nested")).expect("create nested directory"); + std::fs::write(root.path().join("nested/file.txt"), b"data").expect("write local file"); + std::fs::write(outside.path().join("secret.txt"), b"secret").expect("write outside file"); + symlink(outside.path(), root.path().join("linked")).expect("create directory symlink"); + + let manifest = enumerate_local_manifest(root.path(), MissingRootPolicy::Error) + .expect("enumerate local root"); + + assert_eq!(manifest.entries.len(), 1); + assert!(manifest.entries.contains_key("nested/file.txt")); + assert_eq!(manifest.protected_paths, ["linked"]); + assert_eq!(manifest.ignored, 1); +} + +#[derive(Default)] +struct MockMirrorIo { + copy_calls: Mutex>, + remove_calls: Mutex>, + copy_failures: HashSet, +} + +#[async_trait::async_trait] +impl MirrorIo for MockMirrorIo { + async fn copy(&self, operation: MirrorCopyOperation) -> rc_core::Result { + self.copy_calls + .lock() + .expect("copy calls lock") + .push(operation.relative_path.clone()); + if self.copy_failures.contains(&operation.relative_path) { + Err(Error::Conflict(format!( + "Source changed during mirror: {}", + operation.relative_path + ))) + } else { + Ok(operation.source.snapshot.size_bytes.unwrap_or_default()) + } + } + + async fn remove(&self, operation: MirrorRemoveOperation) -> rc_core::Result { + self.remove_calls + .lock() + .expect("remove calls lock") + .push(operation.relative_path); + Ok(0) + } +} + +#[tokio::test] +async fn copy_failure_preserves_partial_results_and_blocks_every_removal() { + let source = manifest([ + remote_entry("src", "source/a.txt", "a.txt", "a"), + remote_entry("src", "source/b.txt", "b.txt", "b"), + ]); + let target = manifest([remote_entry( + "dst", + "backup/stale.txt", + "stale.txt", + "stale", + )]); + let copy_plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + ) + .expect("build copy plan"); + let remove_plan = + build_remove_plan(&source, &target, &TransferSelection::default()).expect("remove plan"); + let io = Arc::new(MockMirrorIo { + copy_failures: HashSet::from(["b.txt".to_string()]), + ..MockMirrorIo::default() + }); + let controls = TransferControls { + concurrency: 2, + continue_on_error: true, + ..TransferControls::default() + }; + + let report = execute_operation_plans(Arc::clone(&io), copy_plan, remove_plan, controls, true) + .await + .expect("valid executor controls"); + + assert_eq!(report.copy.summary.successful, 1); + assert_eq!(report.copy.summary.failed, 1); + assert!(report.remove.is_none()); + assert!( + io.remove_calls + .lock() + .expect("remove calls lock") + .is_empty() + ); + assert!(matches!( + report.copy.outcomes[1].state, + TransferOutcomeState::Failed { .. } + )); +} + +#[tokio::test] +async fn a_removal_retry_treats_an_already_absent_target_as_complete() { + let root = tempfile::tempdir().expect("create local target root"); + let target = MirrorEntry { + relative_path: "stale.txt".to_string(), + location: MirrorLocation::Local(root.path().join("stale.txt")), + snapshot: snapshot(4, "2026-07-21T04:00:00Z", None), + }; + let io = LiveMirrorIo { + source: RuntimeEndpoint::Local { + root: root.path().to_path_buf(), + }, + target: RuntimeEndpoint::Local { + root: root.path().to_path_buf(), + }, + }; + + let removed = io + .remove(MirrorRemoveOperation { + relative_path: "stale.txt".to_string(), + target, + }) + .await + .expect("already absent removal is idempotent"); + + assert_eq!(removed, 0); +} + +#[test] +fn changed_local_source_is_rejected_before_mutation() { + let root = tempfile::tempdir().expect("create source root"); + let path = root.path().join("file.txt"); + std::fs::write(&path, b"old").expect("write initial source"); + let entry = enumerate_local_manifest(root.path(), MissingRootPolicy::Error) + .expect("enumerate source") + .entries + .remove("file.txt") + .expect("source entry"); + std::fs::write(&path, b"changed-size").expect("change source"); + + let result = validate_local_entry(&entry); + + assert!(matches!(result, Err(Error::Conflict(_)))); +} + +#[tokio::test] +async fn failed_staged_download_is_cleaned_without_replacing_destination() { + let root = tempfile::tempdir().expect("create target root"); + let destination = root.path().join("file.txt"); + let staging_path = root.path().join(".file.txt.rc-mirror-stage"); + std::fs::write(&destination, b"old").expect("write original destination"); + std::fs::write(&staging_path, b"partial").expect("write staged download"); + + let result = finish_staged_download( + tempfile::TempPath::try_from_path(staging_path.clone()) + .expect("take ownership of staging path"), + &destination, + Err(Error::Conflict("Source changed during mirror".to_string())), + None, + true, + ) + .await; + + assert!(result.is_err()); + assert!(!staging_path.exists()); + assert_eq!( + std::fs::read(&destination).expect("read original destination"), + b"old" + ); +} + +#[test] +fn empty_manifests_produce_empty_deterministic_plans() { + let copy = build_copy_plan( + &MirrorManifest::default(), + &MirrorManifest::default(), + &MirrorEndpointSpec::Local(PathBuf::from("/target")), + &TransferSelection::default(), + true, + ) + .expect("empty copy plan"); + let remove = build_remove_plan( + &MirrorManifest::default(), + &MirrorManifest::default(), + &TransferSelection::default(), + ) + .expect("empty remove plan"); + + assert!(copy.items.is_empty()); + assert!(remove.items.is_empty()); + assert_eq!(copy.summary.planned, 0); + assert_eq!(remove.summary.planned, 0); +} + +#[test] +fn filtered_nested_tree_is_stably_sorted() { + let source = manifest([ + local_entry("/source", "z/last.txt"), + local_entry("/source", "a/first.txt"), + local_entry("/source", "a/private.txt"), + ]); + let selection = TransferSelection::new( + &["**/*.txt".to_string()], + &["**/private.txt".to_string()], + None, + None, + None, + ) + .expect("valid filters"); + + let plan = build_copy_plan( + &source, + &MirrorManifest::default(), + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "root")), + &selection, + false, + ) + .expect("filtered plan"); + + assert_eq!( + plan.items + .iter() + .map(|item| item.relative_path.as_str()) + .collect::>(), + ["a/first.txt", "z/last.txt"] + ); + assert_eq!(plan.summary.skipped, 1); +} + +#[test] +fn target_mapping_rejects_non_normal_relative_paths() { + let target = MirrorEndpointSpec::Local(PathBuf::from("/target")); + assert!(target.location_for("../outside").is_err()); + assert!(normalized_remote_root_prefix("/absolute").is_err()); +} + +#[test] +fn manifest_rejects_normalization_collisions() { + let mut entries = BTreeMap::new(); + insert_manifest_entry(&mut entries, remote_entry("src", "root/a//b", "a/b", "one")) + .expect("insert first entry"); + let result = insert_manifest_entry(&mut entries, remote_entry("src", "root/a/b", "a/b", "two")); + assert!(matches!(result, Err(Error::Conflict(_)))); +} + +#[derive(Debug, PartialEq, Eq)] +struct PathUploadCall { + key: String, + content_type: Option, + condition: RemoteWriteCondition, + size: u64, + staging: PathBuf, +} + +struct TestRemoteSource { + info: ObjectInfo, + after_info: Option, + download_size: u64, + head_error: Option, + head_calls: Mutex>, + download_calls: Arc>>, + block_after_download: Option>, +} + +#[async_trait::async_trait] +impl MirrorRemoteTransfer for TestRemoteSource { + async fn mirror_head(&self, path: &RemotePath) -> rc_core::Result { + let call_count = { + let mut calls = self.head_calls.lock().expect("head calls lock"); + calls.push(path.key.clone()); + calls.len() + }; + if let Some(error) = &self.head_error { + return Err(Error::Network(error.clone())); + } + if call_count > 1 + && let Some(after_info) = &self.after_info + { + return Ok(after_info.clone()); + } + Ok(self.info.clone()) + } + + async fn mirror_download( + &self, + _path: &RemotePath, + destination: &Path, + ) -> rc_core::Result { + self.download_calls + .lock() + .expect("download calls lock") + .push(destination.to_path_buf()); + let file = tokio::fs::OpenOptions::new() + .write(true) + .truncate(true) + .open(destination) + .await?; + file.set_len(self.download_size).await?; + if let Some(started) = &self.block_after_download { + started.notify_one(); + std::future::pending::<()>().await; + } + Ok(self.download_size) + } + + async fn mirror_upload( + &self, + _path: &RemotePath, + _source: &Path, + _content_type: Option<&str>, + _condition: RemoteWriteCondition, + ) -> rc_core::Result { + Err(Error::General( + "test source cannot upload mirror objects".to_string(), + )) + } +} + +#[derive(Default)] +struct TestRemoteTarget { + uploads: Mutex>, + upload_error: Option, +} + +#[async_trait::async_trait] +impl MirrorRemoteTransfer for TestRemoteTarget { + async fn mirror_head(&self, _path: &RemotePath) -> rc_core::Result { + Err(Error::General( + "test target cannot inspect mirror sources".to_string(), + )) + } + + async fn mirror_download( + &self, + _path: &RemotePath, + _destination: &Path, + ) -> rc_core::Result { + Err(Error::General( + "test target cannot download mirror sources".to_string(), + )) + } + + async fn mirror_upload( + &self, + path: &RemotePath, + source: &Path, + content_type: Option<&str>, + condition: RemoteWriteCondition, + ) -> rc_core::Result { + let size = tokio::fs::metadata(source).await?.len(); + self.uploads + .lock() + .expect("upload calls lock") + .push(PathUploadCall { + key: path.key.clone(), + content_type: content_type.map(ToOwned::to_owned), + condition, + size, + staging: source.to_path_buf(), + }); + if let Some(error) = &self.upload_error { + return Err(Error::Network(error.clone())); + } + Ok(ObjectInfo::file(path.key.clone(), size as i64)) + } +} + +fn remote_transfer_fixture( + size: u64, + content_type: Option<&str>, +) -> (TestRemoteSource, MirrorCopyOperation) { + let modified = "2026-07-21T04:00:00Z" + .parse::() + .expect("valid transfer timestamp"); + let mut info = ObjectInfo::file("source/large.bin", size as i64); + info.last_modified = Some(modified); + info.etag = Some("source-etag".to_string()); + info.content_type = content_type.map(ToOwned::to_owned); + let source = MirrorEntry { + relative_path: "large.bin".to_string(), + location: MirrorLocation::Remote(RemotePath::new("source", "bucket", "source/large.bin")), + snapshot: MirrorSnapshot { + size_bytes: Some(size), + modified: Some(modified), + etag: Some("source-etag".to_string()), + }, + }; + let operation = MirrorCopyOperation { + relative_path: source.relative_path.clone(), + source, + target: MirrorLocation::Remote(RemotePath::new("target", "bucket", "backup/large.bin")), + target_before: None, + }; + ( + TestRemoteSource { + info, + after_info: None, + download_size: size, + head_error: None, + head_calls: Mutex::new(Vec::new()), + download_calls: Arc::new(Mutex::new(Vec::new())), + block_after_download: None, + }, + operation, + ) +} + +#[tokio::test] +async fn large_remote_copy_uses_path_streaming_preserves_metadata_and_cleans_staging() { + let size = 64 * 1024 * 1024 + 1; + let (source, operation) = remote_transfer_fixture(size, Some("application/octet-stream")); + let target = TestRemoteTarget::default(); + let source_path = match &operation.source.location { + MirrorLocation::Remote(path) => path.clone(), + MirrorLocation::Local(_) => panic!("expected remote test source"), + }; + let target_path = match &operation.target { + MirrorLocation::Remote(path) => path.clone(), + MirrorLocation::Local(_) => panic!("expected remote test target"), + }; + + let copied = transfer_remote_to_remote( + &source, + &target, + &source_path, + &target_path, + &operation, + RemoteWriteCondition::IfAbsent, + ) + .await + .expect("stream remote object through a file"); + + assert_eq!(copied, size); + assert_eq!( + source + .head_calls + .lock() + .expect("head calls lock") + .as_slice(), + ["source/large.bin", "source/large.bin"] + ); + let uploads = target.uploads.lock().expect("upload calls lock"); + assert_eq!(uploads.len(), 1); + assert_eq!(uploads[0].size, size); + assert_eq!( + uploads[0].content_type.as_deref(), + Some("application/octet-stream") + ); + assert_eq!(uploads[0].condition, RemoteWriteCondition::IfAbsent); + assert!(!uploads[0].staging.exists()); +} + +#[tokio::test] +async fn remote_head_failure_prevents_download_and_upload() { + let (mut source, operation) = remote_transfer_fixture(4, None); + source.head_error = Some("head failed".to_string()); + let target = TestRemoteTarget::default(); + let MirrorLocation::Remote(source_path) = &operation.source.location else { + panic!("expected remote source") + }; + let MirrorLocation::Remote(target_path) = &operation.target else { + panic!("expected remote target") + }; + + let result = transfer_remote_to_remote( + &source, + &target, + source_path, + target_path, + &operation, + RemoteWriteCondition::IfAbsent, + ) + .await; + + assert!(matches!(result, Err(Error::Network(_)))); + assert!( + source + .download_calls + .lock() + .expect("download calls lock") + .is_empty() + ); + assert!(target.uploads.lock().expect("upload calls lock").is_empty()); +} + +#[tokio::test] +async fn failed_remote_upload_removes_the_complete_staging_file() { + let (source, operation) = remote_transfer_fixture(4, None); + let target = TestRemoteTarget { + upload_error: Some("upload failed".to_string()), + ..TestRemoteTarget::default() + }; + let MirrorLocation::Remote(source_path) = &operation.source.location else { + panic!("expected remote source") + }; + let MirrorLocation::Remote(target_path) = &operation.target else { + panic!("expected remote target") + }; + + let result = transfer_remote_to_remote( + &source, + &target, + source_path, + target_path, + &operation, + RemoteWriteCondition::IfAbsent, + ) + .await; + + assert!(result.is_err()); + let uploads = target.uploads.lock().expect("upload calls lock"); + assert_eq!(uploads.len(), 1); + assert!(!uploads[0].staging.exists()); +} + +#[tokio::test] +async fn remote_source_change_after_download_blocks_upload_and_cleans_staging() { + let (mut source, operation) = remote_transfer_fixture(4, None); + let mut changed = source.info.clone(); + changed.etag = Some("changed-etag".to_string()); + source.after_info = Some(changed); + let target = TestRemoteTarget::default(); + let MirrorLocation::Remote(source_path) = &operation.source.location else { + panic!("expected remote source") + }; + let MirrorLocation::Remote(target_path) = &operation.target else { + panic!("expected remote target") + }; + + let result = transfer_remote_to_remote( + &source, + &target, + source_path, + target_path, + &operation, + RemoteWriteCondition::IfAbsent, + ) + .await; + + assert!(matches!(result, Err(Error::Conflict(_)))); + assert!(target.uploads.lock().expect("upload calls lock").is_empty()); + let downloads = source.download_calls.lock().expect("download calls lock"); + assert_eq!(downloads.len(), 1); + assert!(!downloads[0].exists()); +} + +#[tokio::test] +async fn truncated_remote_download_is_rejected_and_cleaned_before_upload() { + let (mut source, operation) = remote_transfer_fixture(4, None); + source.download_size = 3; + let target = TestRemoteTarget::default(); + let MirrorLocation::Remote(source_path) = &operation.source.location else { + panic!("expected remote source") + }; + let MirrorLocation::Remote(target_path) = &operation.target else { + panic!("expected remote target") + }; + + let result = transfer_remote_to_remote( + &source, + &target, + source_path, + target_path, + &operation, + RemoteWriteCondition::IfAbsent, + ) + .await; + + assert!(matches!(result, Err(Error::Conflict(_)))); + assert!(target.uploads.lock().expect("upload calls lock").is_empty()); + let downloads = source.download_calls.lock().expect("download calls lock"); + assert_eq!(downloads.len(), 1); + assert!(!downloads[0].exists()); +} + +#[tokio::test] +async fn cancelling_a_remote_transfer_drops_and_removes_its_staging_file() { + let (mut source, operation) = remote_transfer_fixture(4, None); + let started = Arc::new(tokio::sync::Notify::new()); + source.block_after_download = Some(Arc::clone(&started)); + let download_calls = Arc::clone(&source.download_calls); + let target = TestRemoteTarget::default(); + let source_path = match &operation.source.location { + MirrorLocation::Remote(path) => path, + MirrorLocation::Local(_) => panic!("expected remote source"), + }; + let target_path = match &operation.target { + MirrorLocation::Remote(path) => path, + MirrorLocation::Local(_) => panic!("expected remote target"), + }; + let mut transfer = Box::pin(transfer_remote_to_remote( + &source, + &target, + source_path, + target_path, + &operation, + RemoteWriteCondition::IfAbsent, + )); + tokio::select! { + () = started.notified() => {} + result = &mut transfer => panic!("transfer unexpectedly completed: {result:?}"), + } + drop(transfer); + + let downloads = download_calls.lock().expect("download calls lock"); + assert_eq!(downloads.len(), 1); + assert!(!downloads[0].exists()); +} + +#[test] +fn remote_overwrite_requires_the_planned_etag() { + let mut operation = remote_transfer_fixture(4, None).1; + operation.target_before = Some(remote_entry( + "target", + "backup/large.bin", + "large.bin", + "target-etag", + )); + assert_eq!( + remote_write_condition(&operation).expect("target ETag is available"), + RemoteWriteCondition::IfMatch("target-etag".to_string()) + ); + + operation + .target_before + .as_mut() + .expect("planned target") + .snapshot + .etag = None; + assert!(matches!( + remote_write_condition(&operation), + Err(Error::Conflict(_)) + )); +} + +#[test] +fn remote_entries_without_two_etags_are_never_assumed_equal() { + let source = remote_entry("source", "root/file.txt", "file.txt", "same"); + let mut target = remote_entry("target", "root/file.txt", "file.txt", "same"); + assert!(source_matches_target(&source, &target)); + target.snapshot.etag = None; + assert!(!source_matches_target(&source, &target)); +} + +#[tokio::test] +async fn missing_multilevel_local_target_root_is_created_one_directory_at_a_time() { + let temporary = tempfile::tempdir().expect("create parent directory"); + let existing = temporary.path().join("existing"); + std::fs::create_dir(&existing).expect("create existing ancestor"); + let root = existing.join("a/b/new-root"); + + let destination = secure_local_path(&root, "nested/file.txt", true) + .await + .expect("create safe target directories"); + + assert_eq!(destination, root.join("nested/file.txt")); + for directory in [ + existing.join("a"), + existing.join("a/b"), + root.clone(), + root.join("nested"), + ] { + let metadata = std::fs::symlink_metadata(&directory).expect("created directory metadata"); + assert!(metadata.is_dir()); + assert!(!metadata.file_type().is_symlink()); + } +} + +#[tokio::test] +async fn plain_relative_multilevel_target_uses_the_current_directory_as_its_ancestor() { + let current = std::env::current_dir().expect("read current directory"); + let placeholder = tempfile::Builder::new() + .prefix(".rc-relative-root-") + .tempdir_in(¤t) + .expect("reserve unique relative root"); + let relative_base = PathBuf::from(placeholder.path().file_name().expect("relative root name")); + drop(placeholder); + let root = relative_base.join("foo/bar"); + + let result = secure_local_path(&root, "nested/file.txt", true).await; + + let destination = result.expect("create plain relative target tree"); + assert_eq!(destination, root.join("nested/file.txt")); + assert!(root.join("nested").is_dir()); + std::fs::remove_dir_all(&relative_base).expect("remove relative target tree"); +} + +#[tokio::test] +async fn atomic_replace_failure_keeps_the_previous_destination() { + let temporary = tempfile::tempdir().expect("create target directory"); + let destination = temporary.path().join("file.txt"); + let missing_staging = temporary.path().join("missing-stage"); + std::fs::write(&destination, b"old").expect("write previous destination"); + + let result = finish_staged_download( + tempfile::TempPath::try_from_path(missing_staging) + .expect("take ownership of missing staging path"), + &destination, + Ok(3), + None, + true, + ) + .await; + + assert!(result.is_err()); + assert_eq!( + std::fs::read(&destination).expect("read previous destination"), + b"old" + ); +} + +#[tokio::test] +async fn an_absent_local_target_is_never_clobbered_if_it_appears_before_persist() { + let temporary = tempfile::tempdir().expect("create target directory"); + let destination = temporary.path().join("file.txt"); + let staging_path = temporary.path().join("staging"); + std::fs::write(&staging_path, b"source").expect("write complete staging file"); + std::fs::write(&destination, b"concurrent").expect("write concurrent destination"); + + let result = finish_staged_download( + tempfile::TempPath::try_from_path(staging_path.clone()) + .expect("take ownership of staging path"), + &destination, + Ok(6), + None, + false, + ) + .await; + + assert!(matches!(result, Err(Error::Conflict(_)))); + assert!(!staging_path.exists()); + assert_eq!( + std::fs::read(&destination).expect("read concurrent destination"), + b"concurrent" + ); +} + +#[test] +fn remote_overlap_is_detected_across_aliases_for_the_same_endpoint() { + let source = RemotePath::new("source", "bucket", "data/source"); + let nested = RemotePath::new("target", "bucket", "data/source/archive"); + let adjacent = RemotePath::new("target", "bucket", "data/source-archive"); + + assert!( + mirror_locations_overlap( + &source, + &nested, + "https://s3.example.com/", + "https://S3.EXAMPLE.COM:443" + ) + .expect("valid mirror roots") + ); + assert!( + !mirror_locations_overlap( + &source, + &adjacent, + "https://s3.example.com", + "https://s3.example.com" + ) + .expect("valid adjacent mirror roots") + ); +} + +#[derive(Parser)] +struct MirrorArgumentParser { + #[command(flatten)] + mirror: MirrorArgs, +} + +#[test] +fn mirror_arguments_keep_legacy_parallel_alias_and_safe_defaults() { + let defaults = MirrorArgumentParser::try_parse_from(["test", "source", "target"]) + .expect("parse mirror defaults") + .mirror; + assert_eq!(defaults.concurrency, 4); + assert_eq!(defaults.retry_attempts, 3); + assert!(!defaults.remove); + assert!(!defaults.overwrite); + + let legacy = MirrorArgumentParser::try_parse_from([ + "test", + "source", + "target", + "--parallel", + "7", + "--skip-errors", + ]) + .expect("parse legacy aliases") + .mirror; + assert_eq!(legacy.concurrency, 7); + assert!(legacy.continue_on_error); +} + +#[test] +fn mirror_output_serialization_preserves_the_public_shape() { + let output = MirrorOutput { + source: "src/".to_string(), + target: "dst/".to_string(), + copied: 10, + removed: 2, + skipped: 5, + errors: 0, + dry_run: false, + }; + let json = serde_json::to_value(output).expect("serialize mirror output"); + assert_eq!(json["copied"], 10); + assert_eq!(json["removed"], 2); + assert_eq!(json["dry_run"], false); +} + +#[test] +fn human_summary_reports_cancelled_operations() { + let output = MirrorOutput { + source: "src/".to_string(), + target: "dst/".to_string(), + copied: 1, + removed: 2, + skipped: 3, + errors: 4, + dry_run: false, + }; + + assert_eq!( + format_human_summary(&output, 5, 1024), + "Summary: 1 copied, 2 removed, 3 skipped, 4 errors, 5 cancelled, 1 KiB transferred" + ); +} + +#[test] +fn mirror_errors_map_to_distinct_usage_and_conflict_exit_codes() { + assert_eq!( + exit_code_for_error(&Error::InvalidPath("bad path".to_string())), + ExitCode::UsageError + ); + assert_eq!( + exit_code_for_error(&Error::Conflict("changed".to_string())), + ExitCode::Conflict + ); +} diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index 1240807f..c3751e67 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -368,7 +368,24 @@ fn top_level_command_help_contract() { HelpCase { args: &["mirror"], usage: "Usage: rc mirror [OPTIONS] ", - expected_tokens: &["--remove", "--overwrite", "--dry-run", "--parallel"], + expected_tokens: &[ + "--remove", + "--overwrite", + "--include", + "--exclude", + "--newer-than", + "--older-than", + "--continue-on-error", + "--skip-errors", + "--dry-run", + "--concurrency", + "--parallel", + "--rate-limit", + "--retry-attempts", + "--retry-initial-backoff-ms", + "--retry-max-backoff-ms", + "--summary", + ], }, HelpCase { args: &["tree"], diff --git a/crates/cli/tests/integration.rs b/crates/cli/tests/integration.rs index bbc60e17..ac0c5fc0 100644 --- a/crates/cli/tests/integration.rs +++ b/crates/cli/tests/integration.rs @@ -2136,6 +2136,21 @@ mod diff_operations { mod mirror_operations { use super::*; + fn upload_text_object(config_dir: &std::path::Path, bucket: &str, key: &str, content: &str) { + let temp_file = tempfile::NamedTempFile::new().expect("Failed to create temp file"); + std::fs::write(temp_file.path(), content).expect("Failed to write test object"); + let source = temp_file + .path() + .to_str() + .expect("Temp file path is valid UTF-8"); + let output = run_rc(&["cp", source, &format!("test/{bucket}/{key}")], config_dir); + assert!( + output.status.success(), + "Failed to upload {key}: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + #[test] fn test_mirror_between_buckets() { let (config_dir, bucket_name) = match setup_with_alias("mirror") { @@ -2179,13 +2194,13 @@ mod mirror_operations { "mirror", &format!("test/{}/source/", bucket_name), &format!("test/{}/", bucket_name2), - "--json", ], config_dir.path(), ); assert!( output.status.success(), - "Failed to mirror: {}", + "Failed to mirror; stdout: {}; stderr: {}", + String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); @@ -2256,13 +2271,13 @@ mod mirror_operations { "mirror", &format!("test/{}/source/", bucket_name), &format!("test/{}/", target_bucket), - "--json", ], config_dir.path(), ); assert!( output.status.success(), - "Failed to mirror objects: {}", + "Failed to mirror objects; stdout: {}; stderr: {}", + String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); @@ -2287,6 +2302,105 @@ mod mirror_operations { cleanup_bucket(config_dir.path(), &bucket_name); cleanup_bucket(config_dir.path(), &target_bucket); } + + #[test] + fn test_mirror_local_tree_to_rustfs() { + let (config_dir, bucket_name) = match setup_with_alias("mirrorlocalremote") { + Some(v) => v, + None => panic!("S3 integration test setup failed"), + }; + let source = tempfile::tempdir().expect("Failed to create local source"); + std::fs::create_dir(source.path().join("nested")).expect("Failed to create nested source"); + std::fs::write(source.path().join("root.txt"), b"root-data") + .expect("Failed to write root source"); + std::fs::write(source.path().join("nested/file.txt"), b"nested-data") + .expect("Failed to write nested source"); + + let output = run_rc( + &[ + "mirror", + source.path().to_str().expect("Source path is UTF-8"), + &format!("test/{}/local-mirror/", bucket_name), + "--summary", + "--json", + ], + config_dir.path(), + ); + assert!( + output.status.success(), + "Failed to mirror local tree; stdout: {}; stderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + let payload: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("Invalid mirror JSON output"); + assert_eq!(payload["copied"], 2); + assert_eq!(payload["errors"], 0); + + let output = run_rc( + &[ + "cat", + &format!("test/{}/local-mirror/nested/file.txt", bucket_name), + ], + config_dir.path(), + ); + assert!(output.status.success(), "Failed to read mirrored object"); + assert_eq!(output.stdout, b"nested-data"); + + cleanup_bucket(config_dir.path(), &bucket_name); + } + + #[test] + fn test_mirror_rustfs_tree_to_new_local_root() { + let (config_dir, bucket_name) = match setup_with_alias("mirrorremotelocal") { + Some(v) => v, + None => panic!("S3 integration test setup failed"), + }; + upload_text_object( + config_dir.path(), + &bucket_name, + "remote-source/root.txt", + "root-data", + ); + upload_text_object( + config_dir.path(), + &bucket_name, + "remote-source/nested/file.txt", + "nested-data", + ); + let target_parent = tempfile::tempdir().expect("Failed to create target parent"); + let target = target_parent.path().join("new/deep/restore"); + + let output = run_rc( + &[ + "mirror", + &format!("test/{}/remote-source/", bucket_name), + target.to_str().expect("Target path is UTF-8"), + "--summary", + "--json", + ], + config_dir.path(), + ); + assert!( + output.status.success(), + "Failed to mirror RustFS tree locally: {}", + String::from_utf8_lossy(&output.stderr) + ); + let payload: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("Invalid mirror JSON output"); + assert_eq!(payload["copied"], 2); + assert_eq!(payload["errors"], 0); + assert_eq!( + std::fs::read(target.join("root.txt")).expect("Read local root object"), + b"root-data" + ); + assert_eq!( + std::fs::read(target.join("nested/file.txt")).expect("Read nested local object"), + b"nested-data" + ); + + cleanup_bucket(config_dir.path(), &bucket_name); + } } mod tree_operations { @@ -4160,7 +4274,8 @@ mod option_behavior_operations { ); assert!( output.status.success(), - "mirror --remove --parallel failed: {}", + "mirror --remove --parallel failed; stdout: {}; stderr: {}", + String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); let stdout = String::from_utf8_lossy(&output.stdout); @@ -4202,7 +4317,7 @@ mod option_behavior_operations { } #[test] - fn test_mirror_parallel_zero_returns_usage_error() { + fn test_mirror_concurrency_zero_returns_usage_error() { let (config_dir, bucket_name) = match setup_with_alias("mirrorparallelzero") { Some(v) => v, None => panic!("S3 integration test setup failed"), @@ -4223,7 +4338,7 @@ mod option_behavior_operations { "mirror", &format!("test/{}/", bucket_name), &format!("test/{}/", target_bucket), - "--parallel", + "--concurrency", "0", "--json", ], @@ -4231,11 +4346,11 @@ mod option_behavior_operations { ); assert!( !output.status.success(), - "mirror --parallel 0 should fail with usage error" + "mirror --concurrency 0 should fail with usage error" ); let stderr = String::from_utf8_lossy(&output.stderr); assert!( - stderr.contains("--parallel must be between 1 and 256"), + stderr.contains("Transfer concurrency must be between 1 and 256"), "Unexpected error output: {}", stderr ); diff --git a/crates/cli/tests/mirror_planner.rs b/crates/cli/tests/mirror_planner.rs new file mode 100644 index 00000000..33b44391 --- /dev/null +++ b/crates/cli/tests/mirror_planner.rs @@ -0,0 +1,329 @@ +//! End-to-end mirror planning tests backed by a read-only mock S3 endpoint. + +use std::io::{Read, Write}; +use std::net::TcpListener; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output}; +use std::thread; +use std::time::{Duration, Instant}; + +fn rc_binary() -> PathBuf { + PathBuf::from(env!("CARGO_BIN_EXE_rc")) +} + +fn run_rc(args: &[&str], config_dir: &Path, endpoint: Option<&str>) -> Output { + let mut command = Command::new(rc_binary()); + for (key, _) in std::env::vars_os() { + if key.to_string_lossy().starts_with("RC_HOST_") { + command.env_remove(key); + } + } + command + .args(args) + .env("AWS_EC2_METADATA_DISABLED", "true") + .env("RC_CONFIG_DIR", config_dir); + if let Some(endpoint) = endpoint { + command.env( + "RC_HOST_test", + format!( + "http://accesskey:secretkey@{}", + endpoint.trim_start_matches("http://") + ), + ); + } + command.output().expect("execute rc") +} + +fn start_list_server( + expected_requests: usize, + paginated_source: bool, +) -> (String, thread::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind mock S3 endpoint"); + listener + .set_nonblocking(true) + .expect("configure nonblocking listener"); + let endpoint = format!("http://{}", listener.local_addr().expect("mock endpoint")); + let handle = thread::spawn(move || { + let deadline = Instant::now() + Duration::from_secs(10); + let mut requests = Vec::new(); + let mut source_page = 0usize; + while requests.len() < expected_requests && Instant::now() < deadline { + let (mut stream, _) = match listener.accept() { + Ok(connection) => connection, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(10)); + continue; + } + Err(error) => panic!("accept mock S3 request: {error}"), + }; + stream + .set_nonblocking(false) + .expect("configure blocking mock connection"); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("set request timeout"); + let mut request = Vec::new(); + let mut chunk = [0_u8; 2048]; + loop { + let read = stream.read(&mut chunk).expect("read mock S3 request"); + if read == 0 { + break; + } + request.extend_from_slice(&chunk[..read]); + if request.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + let request = String::from_utf8_lossy(&request).into_owned(); + let request_line = request.lines().next().unwrap_or_default().to_string(); + let body = if request_line.contains("/source-bucket") && paginated_source { + let response = if source_page == 0 { + paginated_source_first_response() + } else { + paginated_source_second_response() + }; + source_page += 1; + response + } else if request_line.contains("/source-bucket") { + source_list_response() + } else { + empty_list_response() + }; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/xml\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + stream + .write_all(response.as_bytes()) + .expect("write mock S3 response"); + requests.push(request_line); + } + requests + }); + (endpoint, handle) +} + +fn source_list_response() -> &'static str { + r#" + + source-bucket + source/ + 1 + 1000 + false + + source/nested/file.txt + 2026-07-21T04:00:00.000Z + "source-etag" + 4 + STANDARD + +"# +} + +fn empty_list_response() -> &'static str { + r#" + + target-bucket + backup/ + 0 + 1000 + false +"# +} + +fn paginated_source_first_response() -> &'static str { + r#" + + source-bucket + source/ + 1 + 1 + true + page-2 + + source/a.txt + 2026-07-21T04:00:00.000Z + "etag-a" + 1 + STANDARD + +"# +} + +fn paginated_source_second_response() -> &'static str { + r#" + + source-bucket + source/ + 1 + 1 + false + + source/z.txt + 2026-07-21T04:00:01.000Z + "etag-z" + 1 + STANDARD + +"# +} + +fn assert_read_only_list_requests( + handle: thread::JoinHandle>, + expected_requests: usize, +) { + let requests = handle.join().expect("join mock S3 endpoint"); + assert_eq!(requests.len(), expected_requests, "{requests:?}"); + assert!( + requests + .iter() + .all(|request| request.starts_with("GET ") && request.contains("list-type=2")), + "dry-run sent a mutating or unexpected request: {requests:?}" + ); +} + +fn parse_success(output: &Output) -> serde_json::Value { + assert_eq!( + output.status.code(), + Some(0), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).expect("mirror output is JSON") +} + +#[test] +fn local_to_remote_dry_run_plans_nested_paths_without_mutation() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let source = tempfile::tempdir().expect("create local source"); + std::fs::create_dir(source.path().join("nested")).expect("create nested source"); + std::fs::write(source.path().join("nested/file.txt"), b"data").expect("write local source"); + let (endpoint, handle) = start_list_server(1, false); + + let output = run_rc( + &[ + "--json", + "mirror", + source.path().to_str().expect("source path is UTF-8"), + "test/target-bucket/backup/", + "--dry-run", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["dry_run"], true); + assert_read_only_list_requests(handle, 1); +} + +#[test] +fn remote_to_local_dry_run_does_not_create_the_target_root() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let target_parent = tempfile::tempdir().expect("create target parent"); + let target = target_parent.path().join("new/restore"); + let (endpoint, handle) = start_list_server(1, false); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + target.to_str().expect("target path is UTF-8"), + "--dry-run", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["dry_run"], true); + assert!(!target.exists()); + assert_read_only_list_requests(handle, 1); +} + +#[test] +fn remote_to_remote_dry_run_reads_both_manifests_without_mutation() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, handle) = start_list_server(2, false); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + "test/target-bucket/backup/", + "--dry-run", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["removed"], 0); + assert_eq!(payload["dry_run"], true); + assert_read_only_list_requests(handle, 2); +} + +#[test] +fn remote_pagination_is_consumed_before_the_deterministic_plan_is_emitted() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let target_parent = tempfile::tempdir().expect("create target parent"); + let target = target_parent.path().join("restore"); + let (endpoint, handle) = start_list_server(2, true); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + target.to_str().expect("target path is UTF-8"), + "--dry-run", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 2); + assert!(!target.exists()); + let requests = handle.join().expect("join paginated mock S3 endpoint"); + assert_eq!(requests.len(), 2, "{requests:?}"); + assert!(requests[1].contains("continuation-token=page-2")); + assert!(requests.iter().all(|request| request.starts_with("GET "))); +} + +#[test] +fn local_to_local_is_rejected_with_the_unsupported_exit_code() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let source = tempfile::tempdir().expect("create source root"); + let target = tempfile::tempdir().expect("create target root"); + + let output = run_rc( + &[ + "--json", + "mirror", + source.path().to_str().expect("source path is UTF-8"), + target.path().to_str().expect("target path is UTF-8"), + ], + config_dir.path(), + None, + ); + + assert_eq!(output.status.code(), Some(7)); + assert!(output.stdout.is_empty()); + let payload: serde_json::Value = + serde_json::from_slice(&output.stderr).expect("mirror error is JSON"); + assert_eq!(payload["code"], 7); + assert!( + payload["error"] + .as_str() + .is_some_and(|message| message.contains("Local-to-local mirror is out of scope")) + ); +} diff --git a/crates/s3/Cargo.toml b/crates/s3/Cargo.toml index 54b64978..4f694bb1 100644 --- a/crates/s3/Cargo.toml +++ b/crates/s3/Cargo.toml @@ -49,6 +49,7 @@ http-body-util.workspace = true sha2.workspace = true hex.workspace = true urlencoding.workspace = true +tempfile.workspace = true # Serialization serde.workspace = true @@ -57,5 +58,4 @@ quick-xml.workspace = true [dev-dependencies] aws-smithy-http-client = { version = "1.1.5", features = ["test-util"] } -tempfile.workspace = true mockall.workspace = true diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 1a92a65d..087a56fa 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -41,7 +41,6 @@ use serde::Deserialize; use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::str::FromStr; -use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use tokio::io::AsyncReadExt; use tokio::io::AsyncWrite; @@ -55,7 +54,20 @@ const S3_REPLICATION_XML_NAMESPACE: &str = "http://s3.amazonaws.com/doc/2006-03- const RUSTFS_FORCE_DELETE_HEADER: &str = "x-rustfs-force-delete"; const REPLICATION_EXTENSION_BODY_LIMIT: u64 = 1024 * 1024; const S3_BYPASS_GOVERNANCE_RETENTION_HEADER: &str = "x-amz-bypass-governance-retention"; -static DOWNLOAD_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0); + +#[derive(Debug, Clone, Copy)] +enum ObjectWritePrecondition<'a> { + None, + IfAbsent, + IfMatch(&'a str), +} + +#[derive(Debug, Clone, Copy)] +struct PathUploadOptions<'a> { + content_type: Option<&'a str>, + encryption: Option<&'a ObjectEncryptionRequest>, + precondition: ObjectWritePrecondition<'a>, +} #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum BucketPolicyErrorKind { @@ -1235,37 +1247,27 @@ impl S3Client { destination.display() )) })?; - let sequence = DOWNLOAD_TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed); - let temporary = parent.join(format!( - ".{}.rc-part-{}-{sequence}", - file_name.to_string_lossy(), - std::process::id() - )); - let mut file = tokio::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temporary) - .await + let temporary_file = tempfile::Builder::new() + .prefix(&format!(".{}.rc-part-", file_name.to_string_lossy())) + .tempfile_in(parent) .map_err(|error| { Error::General(format!( - "create temporary download '{}': {error}", - temporary.display() + "create temporary download in '{}': {error}", + parent.display() )) })?; + let (file, temporary) = temporary_file.into_parts(); + let mut file = tokio::fs::File::from_std(file); let mut body = response.body; let mut bytes_downloaded = 0u64; while let Some(chunk) = match body.try_next().await { Ok(chunk) => chunk, Err(error) => { - drop(file); - let _ = tokio::fs::remove_file(&temporary).await; return Err(Error::Network(error.to_string())); } } { if let Err(error) = file.write_all(&chunk).await { - drop(file); - let _ = tokio::fs::remove_file(&temporary).await; return Err(Error::General(format!( "write download destination '{}': {error}", destination.display() @@ -1276,8 +1278,6 @@ impl S3Client { } if let Err(error) = file.flush().await { - drop(file); - let _ = tokio::fs::remove_file(&temporary).await; return Err(Error::General(format!( "flush download destination '{}': {error}", destination.display() @@ -1285,23 +1285,13 @@ impl S3Client { } drop(file); - let rename_result = tokio::fs::rename(&temporary, destination).await; - #[cfg(windows)] - let rename_result = match rename_result { - Ok(()) => Ok(()), - Err(_) if tokio::fs::try_exists(destination).await.unwrap_or(false) => { - tokio::fs::remove_file(destination).await?; - tokio::fs::rename(&temporary, destination).await - } - Err(error) => Err(error), - }; - if let Err(error) = rename_result { - let _ = tokio::fs::remove_file(&temporary).await; - return Err(Error::General(format!( - "replace download destination '{}': {error}", - destination.display() - ))); - } + temporary.persist(destination).map_err(|error| { + Error::General(format!( + "atomically replace download destination '{}': {}", + destination.display(), + error.error + )) + })?; Ok(bytes_downloaded) } @@ -1537,19 +1527,24 @@ impl S3Client { } /// Format AWS SDK error into a detailed error message - fn format_sdk_error(error: &aws_sdk_s3::error::SdkError) -> String { + fn format_sdk_error(error: &aws_sdk_s3::error::SdkError) -> String + where + E: std::fmt::Display + ProvideErrorMetadata, + { match error { aws_sdk_s3::error::SdkError::ServiceError(service_err) => { let err = service_err.err(); let meta = service_err.raw(); - let mut msg = format!("Service error: {}", err); - // Try to extract additional error information from headers - if let Some(code) = meta.headers().get("x-amz-error-code") - && let Ok(code_str) = std::str::from_utf8(code.as_bytes()) - { - msg.push_str(&format!(" (code: {})", code_str)); + let header_code = meta + .headers() + .get("x-amz-error-code") + .and_then(|value| std::str::from_utf8(value.as_bytes()).ok()); + let code = err.code().or(header_code); + let mut details = vec![format!("status: {}", meta.status().as_u16())]; + if let Some(code) = code { + details.push(format!("code: {code}")); } - msg + format!("Service error: {err} ({})", details.join(", ")) } aws_sdk_s3::error::SdkError::ConstructionFailure(err) => { format!("Request construction failed: {:?}", err) @@ -2162,9 +2157,8 @@ impl S3Client { &self, path: &RemotePath, file_path: &std::path::Path, - content_type: Option<&str>, file_size: u64, - encryption: Option<&ObjectEncryptionRequest>, + options: PathUploadOptions<'_>, ) -> Result { let data = tokio::fs::read(file_path) .await @@ -2177,17 +2171,28 @@ impl S3Client { .bucket(&path.bucket) .key(&path.key) .body(body), - encryption, + options.encryption, ); - if let Some(ct) = content_type { + if let Some(ct) = options.content_type { request = request.content_type(ct); } + request = match options.precondition { + ObjectWritePrecondition::None => request, + ObjectWritePrecondition::IfAbsent => request.if_none_match("*"), + ObjectWritePrecondition::IfMatch(etag) => request.if_match(etag), + }; - let response = request - .send() - .await - .map_err(|e| Error::Network(e.to_string()))?; + let response = request.send().await.map_err(|error| { + if !matches!(options.precondition, ObjectWritePrecondition::None) + && let aws_sdk_s3::error::SdkError::ServiceError(service_error) = &error + && matches!(service_error.raw().status().as_u16(), 409 | 412) + { + Error::Conflict(format!("Object changed before upload: {path}")) + } else { + Error::Network(Self::format_sdk_error(&error)) + } + })?; let mut info = ObjectInfo::file(&path.key, file_size as i64); if let Some(etag) = response.e_tag() { @@ -2213,9 +2218,8 @@ impl S3Client { &self, path: &RemotePath, file_path: &std::path::Path, - content_type: Option<&str>, file_size: u64, - encryption: Option<&ObjectEncryptionRequest>, + options: PathUploadOptions<'_>, on_progress: impl Fn(u64) + Send, ) -> Result { use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; @@ -2237,7 +2241,7 @@ impl S3Client { .bucket(&path.bucket) .key(&path.key); - create_request = match encryption { + create_request = match options.encryption { Some(ObjectEncryptionRequest::SseS3) => create_request .server_side_encryption(aws_sdk_s3::types::ServerSideEncryption::Aes256), Some(ObjectEncryptionRequest::SseKms { key_id }) => create_request @@ -2246,7 +2250,7 @@ impl S3Client { None => create_request, }; - if let Some(ct) = content_type { + if let Some(ct) = options.content_type { create_request = create_request.content_type(ct); } @@ -2275,7 +2279,10 @@ impl S3Client { return Err(error); } }; - if bytes_read == 0 { + // A conditional zero-byte write still needs a multipart completion + // request, because RustFS evaluates destination preconditions there. + // S3 permits the final part to be smaller than the minimum part size. + if bytes_read == 0 && !(file_size == 0 && part_number == 1) { break; } @@ -2337,23 +2344,38 @@ impl S3Client { let completed_upload = CompletedMultipartUpload::builder() .set_parts(Some(completed_parts)) .build(); - let complete_result = self + let mut complete_request = self .inner .complete_multipart_upload() .bucket(&path.bucket) .key(&path.key) .upload_id(&upload_id) - .multipart_upload(completed_upload) - .send() - .await; + .multipart_upload(completed_upload); + complete_request = match options.precondition { + ObjectWritePrecondition::None => complete_request, + ObjectWritePrecondition::IfAbsent => complete_request.if_none_match("*"), + ObjectWritePrecondition::IfMatch(etag) => complete_request.if_match(etag), + }; + let complete_result = complete_request.send().await; let complete_response = match complete_result { Ok(response) => response, - Err(e) => { + Err(error) => { tracing::debug!(upload_id = %upload_id, "Attempting to abort multipart upload after completion failure"); self.abort_multipart_upload_best_effort(path, &upload_id) .await; - return Err(Error::Network(format!("complete multipart upload: {e}"))); + if !matches!(options.precondition, ObjectWritePrecondition::None) + && let aws_sdk_s3::error::SdkError::ServiceError(service_error) = &error + && matches!(service_error.raw().status().as_u16(), 409 | 412) + { + return Err(Error::Conflict(format!( + "Object changed before upload: {path}" + ))); + } + return Err(Error::Network(format!( + "complete multipart upload: {}", + Self::format_sdk_error(&error) + ))); } }; @@ -2379,6 +2401,71 @@ impl S3Client { content_type: Option<&str>, encryption: Option<&ObjectEncryptionRequest>, on_progress: impl Fn(u64) + Send, + ) -> Result { + self.put_object_from_path_with_condition( + path, + file_path, + content_type, + encryption, + ObjectWritePrecondition::None, + on_progress, + ) + .await + } + + /// Upload a local file path only when the destination object does not exist. + /// + /// The precondition is applied to `PutObject` for single-part uploads and to + /// `CompleteMultipartUpload` for multipart uploads, so a concurrent writer + /// cannot be overwritten between mirror planning and completion. + pub async fn put_object_from_path_if_absent( + &self, + path: &RemotePath, + file_path: &std::path::Path, + content_type: Option<&str>, + encryption: Option<&ObjectEncryptionRequest>, + on_progress: impl Fn(u64) + Send, + ) -> Result { + self.put_object_from_path_with_condition( + path, + file_path, + content_type, + encryption, + ObjectWritePrecondition::IfAbsent, + on_progress, + ) + .await + } + + /// Upload a local file path only when the destination still has `etag`. + pub async fn put_object_from_path_if_match( + &self, + path: &RemotePath, + file_path: &std::path::Path, + content_type: Option<&str>, + encryption: Option<&ObjectEncryptionRequest>, + etag: &str, + on_progress: impl Fn(u64) + Send, + ) -> Result { + self.put_object_from_path_with_condition( + path, + file_path, + content_type, + encryption, + ObjectWritePrecondition::IfMatch(etag), + on_progress, + ) + .await + } + + async fn put_object_from_path_with_condition( + &self, + path: &RemotePath, + file_path: &std::path::Path, + content_type: Option<&str>, + encryption: Option<&ObjectEncryptionRequest>, + precondition: ObjectWritePrecondition<'_>, + on_progress: impl Fn(u64) + Send, ) -> Result { let metadata = tokio::fs::metadata(file_path).await.map_err(|e| { Error::General(format!("read metadata for '{}': {e}", file_path.display())) @@ -2391,25 +2478,23 @@ impl S3Client { } let file_size = metadata.len(); - if Self::should_use_multipart(file_size) { - self.put_object_multipart_from_path( - path, - file_path, - content_type, - file_size, - encryption, - on_progress, - ) - .await + let options = PathUploadOptions { + content_type, + encryption, + precondition, + }; + // RustFS evaluates write preconditions for multipart completion. Keep + // ordinary small uploads on PutObject, but route conditional path writes + // through multipart so mirror retains compare-and-swap semantics on the + // currently deployed service. + if Self::should_use_multipart(file_size) + || !matches!(precondition, ObjectWritePrecondition::None) + { + self.put_object_multipart_from_path(path, file_path, file_size, options, on_progress) + .await } else { - self.put_object_single_part_from_path( - path, - file_path, - content_type, - file_size, - encryption, - ) - .await + self.put_object_single_part_from_path(path, file_path, file_size, options) + .await } } } @@ -2576,8 +2661,14 @@ impl ObjectStore for S3Client { .send() .await .map_err(|e| { - let err_str = e.to_string(); - if err_str.contains("NotFound") || err_str.contains("NoSuchKey") { + let missing = e + .raw_response() + .is_some_and(|response| response.status().as_u16() == 404) + || e.as_service_error().is_some_and(|error| { + matches!(error.code(), Some("NotFound" | "NoSuchKey")) + }); + let err_str = Self::format_sdk_error(&e); + if missing || err_str.contains("NotFound") || err_str.contains("NoSuchKey") { Error::NotFound(path.to_string()) } else { Error::Network(err_str) @@ -3790,7 +3881,9 @@ impl ObjectStore for S3Client { #[cfg(test)] mod tests { use super::*; - use aws_smithy_http_client::test_util::{CaptureRequestReceiver, capture_request}; + use aws_smithy_http_client::test_util::{ + CaptureRequestReceiver, ReplayEvent, StaticReplayClient, capture_request, + }; use std::collections::HashMap; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; @@ -3866,6 +3959,65 @@ mod tests { (client, request_receiver) } + fn test_s3_client_with_response_sequence( + responses: Vec>, + ) -> (S3Client, StaticReplayClient) { + let events = responses + .into_iter() + .enumerate() + .map(|(index, response)| { + let request = http::Request::builder() + .uri(format!("https://example.com/expected-{index}")) + .body(SdkBody::empty()) + .expect("build replay request"); + ReplayEvent::new(request, response) + }) + .collect(); + let replay = StaticReplayClient::new(events); + let credentials = Credentials::new( + "access-key", + "secret-key", + None, + None, + "rc-test-credentials", + ); + let config = aws_sdk_s3::config::Builder::new() + .credentials_provider(credentials) + .endpoint_url("https://example.com") + .region(aws_sdk_s3::config::Region::new("us-east-1")) + .force_path_style(true) + .retry_config(aws_smithy_types::retry::RetryConfig::disabled()) + .behavior_version_latest() + .http_client(replay.clone()) + .build(); + let alias = Alias::new("test", "https://example.com", "access-key", "secret-key"); + let client = S3Client { + inner: aws_sdk_s3::Client::from_conf(config.clone()), + presign_inner: aws_sdk_s3::Client::from_conf(config), + xml_http_client: reqwest::Client::new(), + alias, + request_headers: Vec::new(), + }; + (client, replay) + } + + #[tokio::test] + async fn head_object_maps_bare_http_404_to_not_found() { + let response = http::Response::builder() + .status(404) + .body(SdkBody::empty()) + .expect("build head object response"); + let (client, _) = test_s3_client(Some(response)); + let path = RemotePath::new("test", "bucket", "missing.txt"); + + let result = client.head_object(&path).await; + + assert!( + matches!(result, Err(Error::NotFound(_))), + "unexpected result: {result:?}" + ); + } + fn read_xml_request(stream: &mut TcpStream) -> CapturedXmlRequest { let mut buffer = Vec::new(); let mut chunk = [0_u8; 1024]; @@ -5020,7 +5172,7 @@ mod tests { } #[tokio::test] - async fn conditional_mirror_writes_and_deletes_set_precondition_headers() { + async fn conditional_buffer_writes_and_deletes_set_precondition_headers() { let put_response = http::Response::builder() .status(200) .body(SdkBody::from("")) @@ -5048,6 +5200,249 @@ mod tests { assert_eq!(delete_request.headers().get("if-match"), Some("etag-value")); } + #[tokio::test] + async fn conditional_path_writes_complete_with_precondition_headers() { + let complete_response = || { + http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .body(SdkBody::from( + r#"bucketkey.txt"final-etag""#, + )) + .expect("build multipart complete response") + }; + let path = RemotePath::new("test", "bucket", "key.txt"); + let mut source = tempfile::NamedTempFile::new().expect("create upload source"); + source.write_all(b"path-data").expect("write upload source"); + + let (path_upload_client, path_upload_replay) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + multipart_part_response(), + complete_response(), + ]); + path_upload_client + .put_object_from_path_if_absent(&path, source.path(), None, None, |_| {}) + .await + .expect("conditional path upload"); + let path_upload_requests = path_upload_replay.actual_requests().collect::>(); + assert_eq!(path_upload_requests.len(), 3); + assert_eq!( + path_upload_requests[2].headers().get("if-none-match"), + Some("*") + ); + + let (matched_upload_client, matched_upload_replay) = + test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + multipart_part_response(), + complete_response(), + ]); + matched_upload_client + .put_object_from_path_if_match( + &path, + source.path(), + None, + None, + "expected-etag", + |_| {}, + ) + .await + .expect("matched path upload"); + let matched_upload_requests = matched_upload_replay.actual_requests().collect::>(); + assert_eq!(matched_upload_requests.len(), 3); + assert_eq!( + matched_upload_requests[2].headers().get("if-match"), + Some("expected-etag") + ); + } + + #[tokio::test] + async fn conditional_empty_path_write_uploads_one_empty_part() { + let complete_response = http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .body(SdkBody::from( + r#"bucketempty.txt"final-etag""#, + )) + .expect("build multipart complete response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + multipart_part_response(), + complete_response, + ]); + let path = RemotePath::new("test", "bucket", "empty.txt"); + let source = tempfile::NamedTempFile::new().expect("create empty upload source"); + + client + .put_object_from_path_if_absent(&path, source.path(), None, None, |_| {}) + .await + .expect("conditional empty path upload"); + + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 3); + assert!(requests[1].uri().contains("partNumber=1")); + assert_eq!(requests[2].headers().get("if-none-match"), Some("*")); + } + + fn multipart_create_response() -> http::Response { + http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .body(SdkBody::from( + r#"bucketkey.txtupload-id"#, + )) + .expect("build multipart create response") + } + + fn multipart_part_response() -> http::Response { + http::Response::builder() + .status(200) + .header("etag", "\"part-etag\"") + .body(SdkBody::empty()) + .expect("build multipart part response") + } + + #[tokio::test] + async fn conditional_multipart_completion_sets_if_none_match() { + let complete_response = http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .body(SdkBody::from( + r#"bucketkey.txt"final-etag""#, + )) + .expect("build multipart complete response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + multipart_part_response(), + complete_response, + ]); + let path = RemotePath::new("test", "bucket", "key.txt"); + let mut source = tempfile::NamedTempFile::new().expect("create multipart source"); + source.write_all(b"data").expect("write multipart source"); + + client + .put_object_multipart_from_path( + &path, + source.path(), + 4, + PathUploadOptions { + content_type: Some("text/plain"), + encryption: None, + precondition: ObjectWritePrecondition::IfAbsent, + }, + |_| {}, + ) + .await + .expect("complete conditional multipart upload"); + + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 3); + assert_eq!(requests[2].headers().get("if-none-match"), Some("*")); + assert!(requests[2].uri().contains("uploadId=upload-id")); + } + + #[tokio::test] + async fn conditional_multipart_conflicts_are_mapped_and_aborted() { + for (status, code) in [ + (409_u16, "ConditionalRequestConflict"), + (412_u16, "PreconditionFailed"), + ] { + let complete_response = http::Response::builder() + .status(status) + .header("content-type", "application/xml") + .header("x-amz-error-code", code) + .body(SdkBody::from(format!( + "{code}conditional write failed" + ))) + .expect("build multipart conflict response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build multipart abort response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + multipart_part_response(), + complete_response, + abort_response, + ]); + let path = RemotePath::new("test", "bucket", "key.txt"); + let mut source = tempfile::NamedTempFile::new().expect("create multipart source"); + source.write_all(b"data").expect("write multipart source"); + + let result = client + .put_object_multipart_from_path( + &path, + source.path(), + 4, + PathUploadOptions { + content_type: None, + encryption: None, + precondition: ObjectWritePrecondition::IfMatch("expected-etag"), + }, + |_| {}, + ) + .await; + + assert!(matches!(result, Err(Error::Conflict(_))), "status {status}"); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 4, "status {status}"); + assert_eq!( + requests[2].headers().get("if-match"), + Some("expected-etag"), + "status {status}" + ); + assert_eq!(requests[3].method(), "DELETE", "status {status}"); + assert!( + requests[3].uri().contains("uploadId=upload-id"), + "status {status}" + ); + } + } + + #[tokio::test] + async fn conditional_multipart_service_errors_preserve_response_metadata() { + let complete_response = http::Response::builder() + .status(500) + .header("content-type", "application/xml") + .body(SdkBody::from( + "InternalErrorconditional completion failed", + )) + .expect("build multipart service error response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build multipart abort response"); + let (client, _) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + multipart_part_response(), + complete_response, + abort_response, + ]); + let path = RemotePath::new("test", "bucket", "key.txt"); + let mut source = tempfile::NamedTempFile::new().expect("create multipart source"); + source.write_all(b"data").expect("write multipart source"); + + let result = client + .put_object_multipart_from_path( + &path, + source.path(), + 4, + PathUploadOptions { + content_type: None, + encryption: None, + precondition: ObjectWritePrecondition::IfAbsent, + }, + |_| {}, + ) + .await; + + let Err(Error::Network(message)) = result else { + panic!("expected a network error"); + }; + assert!(message.contains("status: 500"), "{message}"); + assert!(message.contains("code: InternalError"), "{message}"); + } + #[tokio::test] async fn custom_headers_are_added_before_sending_sdk_requests() { let (client, request_receiver) = test_s3_client_with_endpoint_and_headers( diff --git a/docs/reference/rc/mirror.md b/docs/reference/rc/mirror.md index edd27a1e..bec2801b 100644 --- a/docs/reference/rc/mirror.md +++ b/docs/reference/rc/mirror.md @@ -2,7 +2,7 @@ ## Purpose -`rc mirror` synchronizes objects from one remote S3-compatible location to another remote S3-compatible location. +`rc mirror` synchronizes a directory-like tree between the local filesystem and S3-compatible storage. It supports local-to-remote, remote-to-local, and remote-to-remote synchronization. Local-to-local synchronization is rejected; use a filesystem synchronization tool for that case. ## Syntax @@ -14,24 +14,62 @@ rc [GLOBAL OPTIONS] mirror [OPTIONS] | Parameter | Description | | --- | --- | -| `SOURCE` | Remote source bucket or prefix path, in `ALIAS/BUCKET[/PREFIX]` form. | -| `TARGET` | Remote destination bucket or prefix path, in `ALIAS/BUCKET[/PREFIX]` form. | -| `--overwrite` | Overwrite changed destination objects. | -| `--remove` | Delete target objects that no longer exist in the source. | -| `-n, --dry-run` | Show planned changes without applying them. | -| `-P, --parallel` | Number of parallel transfer workers. Defaults to `4`. | +| `SOURCE` | A local directory or remote bucket/prefix in `ALIAS/BUCKET[/PREFIX]` form. Prefix ambiguous relative local paths with `./` or `../`. | +| `TARGET` | A local directory or remote bucket/prefix in `ALIAS/BUCKET[/PREFIX]` form. Prefix a new relative local directory with `./` or `../`. | +| `--overwrite` | Replace changed destination files or objects. Existing remote objects are replaced only if their planned ETag still matches. | +| `--remove` | Remove selected destination files or objects that are absent from the source. Removal starts only after every copy succeeds. | +| `--include ` | Include source-relative paths matching the glob. Repeatable. When present, unmatched paths are skipped. | +| `--exclude ` | Exclude source-relative paths matching the glob. Repeatable; exclusion always wins. | +| `--newer-than ` | Select entries modified more recently than the age, such as `1h` or `7d`. | +| `--older-than ` | Select entries modified less recently than the age, such as `1h` or `7d`. | +| `--continue-on-error`, `--skip-errors` | Continue independent copy operations after a per-entry failure. Failed copies still block all removals. | +| `-n, --dry-run` | Emit the deterministic plan without creating, replacing, or removing files or objects. | +| `-P, --concurrency ` | Maximum operations in flight across the command. Defaults to `4`; valid range is `1..=256`. `--parallel` remains a visible compatibility alias. | +| `--rate-limit ` | Apply aggregate transfer pacing, such as `10MiB/s`. | +| `--retry-attempts ` | Maximum attempts for transient failures. Defaults to `3`. | +| `--retry-initial-backoff-ms ` | Initial transient retry backoff. Defaults to `100`. | +| `--retry-max-backoff-ms ` | Maximum transient retry backoff. Defaults to `10000`. | +| `--summary` | Print deterministic aggregate counts and transferred bytes in human output. | +| `--quiet` | Suppress non-error command output. The global `--quiet` option has the same effect. | ## Examples ```bash -rc mirror local/source backup/source --parallel 8 -rc mirror local/web/site backup/web/site --dry-run -rc mirror local/web/site backup/web/site --overwrite --remove +rc mirror ./site/ rustfs/web/site/ --overwrite --summary +rc mirror rustfs/archive/ ./restore/ --remove --dry-run +rc mirror stage/data/ prod/data/ --include '**/*.json' --exclude '**/private/**' +rc mirror stage/data/ prod/data/ --overwrite --remove --concurrency 8 --rate-limit 20MiB/s ``` ## Behavior -Mirror is intended for remote prefix-level synchronization. Local filesystem paths are not supported by the current implementation. Use `--dry-run` first when combining `--overwrite` and `--remove`. +### Root and path mapping + +Both operands are directory-like roots. Every selected source-relative path is appended to the destination root without flattening. Remote prefixes are normalized to one trailing `/`, so `alias/bucket/prefix` and `alias/bucket/prefix/` map the same tree. Remote listing is paginated and the final plan is sorted by normalized relative path. + +Remote keys that are absolute, contain traversal, use backslashes, collide after normalization, or cannot be represented portably on supported local platforms are rejected. A new relative local target should be written explicitly, for example `./restore/`, so it cannot be confused with `ALIAS/BUCKET` syntax. + +### Comparison and restart behavior + +Entries are compared by size and the strongest stable metadata available. Remote-to-remote equality requires matching ETags. Downloads preserve the source modification time, and local-to-remote restart checks accept a same-size destination written no earlier than the source. A completed entry is skipped on a restarted command. + +`--overwrite` authorizes replacing a changed destination; it does not disable concurrency checks. Mirror revalidates sources and compares local destination metadata again before persistence, so changes observed by those checks fail with the conflict exit code. New remote objects use `If-None-Match: *`, while existing remote objects and remote removals use the planned ETag as a condition; these service-side conditions also reject remote races after the final client-side check. Local replacement is atomic but is not a filesystem compare-and-swap, so a local writer racing after the final metadata check may be replaced. + +### Local filesystem safety + +Mirror does not follow symbolic links. Source symlinks and special files are skipped, and their destination-relative paths are protected from `--remove`. A destination symlink or special file is a conflict. Missing destination directories are created one component at a time with no-follow validation. + +Remote downloads are written to a unique sibling staging file and atomically persisted only after the source is revalidated. Failed or interrupted transfers remove staging files and preserve the previous destination. Remote-to-remote transfers use a bounded-memory temporary file instead of buffering the complete object in memory. + +### Failure, removal, and dry-run semantics + +Copy and removal phases use the shared transfer controls for filtering, concurrency, rate pacing, retry, cancellation, and summaries. Removals are a separate second phase and are withheld if any copy fails or is cancelled, including when `--continue-on-error` is set. A failed run keeps successful copies so the same command can resume deterministically. + +`--dry-run` may list remote prefixes and read local metadata to build the plan, but it performs no uploads, downloads, directory creation, replacements, or removals. Use it before combining `--overwrite` and `--remove`. + +### Compatibility and migration + +Existing remote-to-remote commands continue to work. `--parallel` is retained as an alias of the shared `--concurrency` option. Mirror no longer falls back to an unconditional byte copy when source metadata lookup fails, and missing remote ETags are no longer treated as proof of equality. Automation that depended on either unsafe fallback must handle explicit network or conflict exits and retry after re-planning. Global options shown in command syntax use the same meaning everywhere: