diff --git a/crates/cli/src/commands/mirror.rs b/crates/cli/src/commands/mirror.rs index 32a40777..60fd2ab5 100644 --- a/crates/cli/src/commands/mirror.rs +++ b/crates/cli/src/commands/mirror.rs @@ -1,16 +1,18 @@ //! mirror command - Synchronize trees between local filesystems and S3-compatible storage. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; +use std::future::Future; use std::path::{Component, Path, PathBuf}; use std::sync::Arc; -use clap::Args; +use clap::{Args, ValueEnum}; 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, + AliasManager, Error, ListOptions, ObjectAttributes, ObjectInfo, ObjectStore as _, + ObjectWriteOptions, ParsedPath, RemotePath, TransferCandidate, TransferControls, + TransferExecutor, TransferOutcomeState, TransferPlan, TransferReport, TransferSelection, + TransferSummary, parse_path, }; use rc_s3::S3Client; use serde::Serialize; @@ -20,6 +22,21 @@ use super::cp::{parse_age_cutoff, parse_byte_rate}; use crate::exit_code::ExitCode; use crate::output::{Formatter, OutputConfig}; +const SOURCE_IDENTITY_METADATA_KEY: &str = "rc-source-etag"; + +/// How `rc mirror` decides that a destination object already matches the source. +#[derive(Copy, Clone, Debug, Default, Eq, PartialEq, ValueEnum)] +pub enum CompareMode { + /// Skip when ETags match, or when size matches and destination metadata + /// records the source ETag from a previous `rc mirror` copy. + #[default] + Auto, + /// Skip only when destination and source ETags are identical. + Etag, + /// Skip when object sizes match, ignoring ETag differences. + Size, +} + const MIRROR_AFTER_HELP: &str = "\ Examples: rc mirror ./site/ local/web/site/ --overwrite @@ -92,6 +109,10 @@ pub struct MirrorArgs { #[arg(long)] pub summary: bool, + /// How existing destination objects are compared before skipping a copy + #[arg(long, value_enum, default_value_t = CompareMode::Auto)] + pub compare: CompareMode, + /// Suppress non-error mirror output (legacy command-local alias) #[arg(long)] pub quiet: bool, @@ -113,6 +134,18 @@ struct MirrorSnapshot { size_bytes: Option, modified: Option, etag: Option, + // Preserved source ETag from `x-amz-meta-rc-source-etag`. ListObjects does + // not return user metadata, so this is filled from HeadObject during Auto + // compare and must not participate in destination race detection. + identity_etag: Option, +} + +impl MirrorSnapshot { + fn content_matches(&self, other: &Self) -> bool { + self.size_bytes == other.size_bytes + && self.modified == other.modified + && self.etag == other.etag + } } #[derive(Debug, Clone, PartialEq, Eq)] @@ -210,6 +243,7 @@ trait MirrorRemoteTransfer: Send + Sync { source: &Path, content_type: Option<&str>, condition: RemoteWriteCondition, + identity_etag: Option<&str>, ) -> rc_core::Result; } @@ -230,15 +264,37 @@ impl MirrorRemoteTransfer for S3Client { source: &Path, content_type: Option<&str>, condition: RemoteWriteCondition, + identity_etag: Option<&str>, ) -> rc_core::Result { + let mut user_metadata = HashMap::new(); + if let Some(identity_etag) = identity_etag { + user_metadata.insert( + SOURCE_IDENTITY_METADATA_KEY.to_string(), + identity_etag.to_string(), + ); + } + let options = ObjectWriteOptions { + attributes: Some(ObjectAttributes { + content_type: content_type.map(ToString::to_string), + user_metadata, + ..ObjectAttributes::default() + }), + ..ObjectWriteOptions::default() + }; match condition { RemoteWriteCondition::IfAbsent => { - self.put_object_from_path_if_absent(path, source, content_type, None, |_| {}) + self.put_object_from_path_if_absent_with_options(path, source, &options, |_| {}) .await } RemoteWriteCondition::IfMatch(etag) => { - self.put_object_from_path_if_match(path, source, content_type, None, &etag, |_| {}) - .await + self.put_object_from_path_if_match_with_options( + path, + source, + &options, + &etag, + |_| {}, + ) + .await } } } @@ -297,6 +353,7 @@ impl RuntimeEndpoint { struct LiveMirrorIo { source: RuntimeEndpoint, target: RuntimeEndpoint, + compare: CompareMode, } #[async_trait::async_trait] @@ -324,7 +381,7 @@ impl MirrorIo for LiveMirrorIo { // A previous attempt may have removed the entry before its response was lost. return Ok(0); }; - if current.snapshot != operation.target.snapshot { + if !current.snapshot.content_matches(&operation.target.snapshot) { return Err(Error::Conflict(format!( "Destination changed before removal: {}", operation.relative_path @@ -401,7 +458,13 @@ impl LiveMirrorIo { 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) + .mirror_upload( + target_path, + &staged, + content_type.as_deref(), + condition, + None, + ) .await?; Ok(object_size(&info).unwrap_or(size)) } @@ -433,7 +496,12 @@ impl LiveMirrorIo { )); }; - match self.target_disposition(operation).await? { + let disposition = + source_preflight_then_target(source_client.as_ref(), source_path, operation, || { + self.target_disposition(operation) + }) + .await?; + match disposition { TargetDisposition::AlreadyComplete => { return Ok(operation.source.snapshot.size_bytes.unwrap_or_default()); } @@ -541,22 +609,50 @@ impl LiveMirrorIo { 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); - } + target_disposition_for_current(operation, current.as_ref(), self.compare) + } +} - 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 - ))), +async fn source_preflight_then_target( + source_client: &S, + source_path: &RemotePath, + operation: &MirrorCopyOperation, + target_check: F, +) -> rc_core::Result +where + S: MirrorRemoteTransfer + ?Sized, + F: FnOnce() -> Fut, + Fut: Future>, +{ + let before = source_client.mirror_head(source_path).await?; + ensure_snapshot_matches(&operation.source, &before)?; + target_check().await +} + +fn target_disposition_for_current( + operation: &MirrorCopyOperation, + current: Option<&MirrorEntry>, + compare: CompareMode, +) -> rc_core::Result { + if let Some(current) = current + && operation + .target_before + .as_ref() + .is_some_and(|expected| expected.snapshot.content_matches(¤t.snapshot)) + && source_matches_target(&operation.source, current, compare) + { + return Ok(TargetDisposition::AlreadyComplete); + } + + match (&operation.target_before, current) { + (None, None) => Ok(TargetDisposition::Ready), + (Some(expected), Some(current)) if expected.snapshot.content_matches(¤t.snapshot) => { + Ok(TargetDisposition::Ready) } + _ => Err(Error::Conflict(format!( + "Destination changed after mirror planning: {}", + operation.relative_path + ))), } } @@ -619,7 +715,13 @@ where .as_deref() .or(before.content_type.as_deref()); let info = target_client - .mirror_upload(target_path, &staging, content_type, condition) + .mirror_upload( + target_path, + &staging, + content_type, + condition, + operation.source.snapshot.etag.as_deref(), + ) .await?; Ok(object_size(&info) .or(operation.source.snapshot.size_bytes) @@ -691,7 +793,7 @@ pub async fn execute(args: MirrorArgs, mut output_config: OutputConfig) -> ExitC ); } }; - let (target_runtime, target_manifest) = + let (target_runtime, mut target_manifest) = match prepare_endpoint(&target, MissingRootPolicy::Empty, &alias_manager).await { Ok(prepared) => prepared, Err(error) => { @@ -701,12 +803,26 @@ pub async fn execute(args: MirrorArgs, mut output_config: OutputConfig) -> ExitC ); } }; + if let Err(error) = enrich_destination_identity( + &source_manifest, + &mut target_manifest, + &target_runtime, + args.compare, + ) + .await + { + return formatter.fail( + exit_code_for_error(&error), + &format!("Failed to inspect destination identity: {error}"), + ); + } let copy_plan = match build_copy_plan( &source_manifest, &target_manifest, &target_runtime.spec(), &selection, args.overwrite, + args.compare, ) { Ok(plan) => plan, Err(error) => return formatter.fail(exit_code_for_error(&error), &error.to_string()), @@ -728,6 +844,7 @@ pub async fn execute(args: MirrorArgs, mut output_config: OutputConfig) -> ExitC let io = Arc::new(LiveMirrorIo { source: source_runtime, target: target_runtime, + compare: args.compare, }); let reports = match execute_operation_plans(io, copy_plan, remove_plan, controls, args.remove).await { @@ -1025,6 +1142,7 @@ fn build_copy_plan( target_endpoint: &MirrorEndpointSpec, selection: &TransferSelection, overwrite: bool, + compare: CompareMode, ) -> rc_core::Result> { let mut candidates = Vec::with_capacity(source.entries.len()); for entry in source.entries.values() { @@ -1060,7 +1178,9 @@ fn build_copy_plan( 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(target) if source_matches_target(&candidate.payload.source, target, compare) => { + false + } Some(_) => overwrite, }; if should_copy { @@ -1414,6 +1534,7 @@ fn snapshot_from_metadata(metadata: &std::fs::Metadata) -> MirrorSnapshot { .ok() .and_then(|value| value.try_into().ok()), etag: None, + identity_etag: None, } } @@ -1433,14 +1554,55 @@ fn snapshot_from_object(object: &ObjectInfo) -> rc_core::Result size_bytes, modified: object.last_modified, etag: object.etag.clone(), + identity_etag: identity_etag_from_metadata(object.metadata.as_ref()), }) } +fn identity_etag_from_metadata(metadata: Option<&HashMap>) -> Option { + metadata.and_then(|metadata| { + metadata.iter().find_map(|(key, value)| { + let normalized = key.to_ascii_lowercase(); + let key = normalized + .strip_prefix("x-amz-meta-") + .unwrap_or(normalized.as_str()); + (key == SOURCE_IDENTITY_METADATA_KEY && !value.is_empty()).then(|| value.clone()) + }) + }) +} + +fn destination_needs_identity_lookup( + source: &MirrorEntry, + target: &MirrorEntry, + compare: CompareMode, +) -> bool { + if !matches!(compare, CompareMode::Auto) { + return false; + } + if source.snapshot.size_bytes.is_none() + || source.snapshot.size_bytes != target.snapshot.size_bytes + { + return false; + } + let Some(source_etag) = source.snapshot.etag.as_ref() else { + return false; + }; + if target.snapshot.etag.as_ref() == Some(source_etag) { + return false; + } + if target.snapshot.identity_etag.is_some() { + return false; + } + matches!( + (&source.location, &target.location), + (MirrorLocation::Remote(_), MirrorLocation::Remote(_)) + ) +} + fn object_size(object: &ObjectInfo) -> Option { object.size_bytes.and_then(|size| u64::try_from(size).ok()) } -fn source_matches_target(source: &MirrorEntry, target: &MirrorEntry) -> bool { +fn source_matches_target(source: &MirrorEntry, target: &MirrorEntry, compare: CompareMode) -> bool { let (Some(source_size), Some(target_size)) = (source.snapshot.size_bytes, target.snapshot.size_bytes) else { @@ -1449,24 +1611,78 @@ fn source_matches_target(source: &MirrorEntry, target: &MirrorEntry) -> bool { 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 + match compare { + CompareMode::Size => true, + CompareMode::Etag => { + source.snapshot.etag.is_some() && source.snapshot.etag == target.snapshot.etag + } + CompareMode::Auto => { + if source.snapshot.etag.is_some() && source.snapshot.etag == target.snapshot.etag { + return true; + } + if source + .snapshot + .etag + .as_ref() + .zip(target.snapshot.identity_etag.as_ref()) + .is_some_and(|(source_etag, identity_etag)| source_etag == identity_etag) + { + return true; + } + 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 } - _ => false, } } - _ => { - source.snapshot.modified.is_some() - && source.snapshot.modified == target.snapshot.modified + } +} + +async fn enrich_destination_identity( + source: &MirrorManifest, + target: &mut MirrorManifest, + target_runtime: &RuntimeEndpoint, + compare: CompareMode, +) -> rc_core::Result<()> { + let RuntimeEndpoint::Remote { client, .. } = target_runtime else { + return Ok(()); + }; + if !matches!(compare, CompareMode::Auto) { + return Ok(()); + } + for (relative_path, source_entry) in &source.entries { + let Some(target_entry) = target.entries.get(relative_path) else { + continue; + }; + if !destination_needs_identity_lookup(source_entry, target_entry, compare) { + continue; + } + let MirrorLocation::Remote(path) = &target_entry.location else { + continue; + }; + let path = path.clone(); + match client.head_object(&path).await { + Ok(info) => { + if let Some(target_entry) = target.entries.get_mut(relative_path) { + target_entry.snapshot.identity_etag = + identity_etag_from_metadata(info.metadata.as_ref()); + } + } + Err(Error::NotFound(_)) => {} + Err(error) => return Err(error), } } + Ok(()) } fn path_is_protected(relative_path: &str, protected_paths: &[String]) -> bool { @@ -1501,7 +1717,11 @@ fn validate_local_entry(entry: &MirrorEntry) -> rc_core::Result<()> { } fn ensure_snapshot_matches(entry: &MirrorEntry, object: &ObjectInfo) -> rc_core::Result<()> { - if snapshot_from_object(object)? != entry.snapshot { + let current = snapshot_from_object(object)?; + if current.size_bytes != entry.snapshot.size_bytes + || current.modified != entry.snapshot.modified + || current.etag != entry.snapshot.etag + { return Err(Error::Conflict(format!( "Source changed during mirror: {}", entry.relative_path diff --git a/crates/cli/src/commands/mirror/roadmap_tests.rs b/crates/cli/src/commands/mirror/roadmap_tests.rs index c02c9b49..948f7452 100644 --- a/crates/cli/src/commands/mirror/roadmap_tests.rs +++ b/crates/cli/src/commands/mirror/roadmap_tests.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; @@ -9,10 +9,20 @@ use rc_core::{Error, RemotePath, TransferControls, TransferOutcomeState, Transfe use super::*; fn snapshot(size: u64, modified: &str, etag: Option<&str>) -> MirrorSnapshot { + snapshot_with_identity(size, modified, etag, None) +} + +fn snapshot_with_identity( + size: u64, + modified: &str, + etag: Option<&str>, + identity_etag: Option<&str>, +) -> MirrorSnapshot { MirrorSnapshot { size_bytes: Some(size), modified: Some(modified.parse::().expect("valid test timestamp")), etag: etag.map(ToOwned::to_owned), + identity_etag: identity_etag.map(ToOwned::to_owned), } } @@ -98,6 +108,7 @@ fn planners_map_all_supported_directions_without_changing_relative_paths() { &target, &TransferSelection::default(), false, + CompareMode::Auto, ) .expect("build copy plan"); assert_eq!(plan.items.len(), 1); @@ -127,6 +138,7 @@ fn equivalent_destination_is_restart_safe_and_not_planned_again() { &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), &TransferSelection::default(), true, + CompareMode::Auto, ) .expect("build restart plan"); @@ -156,6 +168,7 @@ fn changed_destination_requires_overwrite() { &destination, &TransferSelection::default(), false, + CompareMode::Auto, ) .expect("build non-overwrite plan"); let overwritten = build_copy_plan( @@ -164,6 +177,7 @@ fn changed_destination_requires_overwrite() { &destination, &TransferSelection::default(), true, + CompareMode::Auto, ) .expect("build overwrite plan"); @@ -261,6 +275,7 @@ async fn copy_failure_preserves_partial_results_and_blocks_every_removal() { &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), &TransferSelection::default(), true, + CompareMode::Auto, ) .expect("build copy plan"); let remove_plan = @@ -309,6 +324,7 @@ async fn a_removal_retry_treats_an_already_absent_target_as_complete() { target: RuntimeEndpoint::Local { root: root.path().to_path_buf(), }, + compare: CompareMode::Auto, }; let removed = io @@ -373,6 +389,7 @@ fn empty_manifests_produce_empty_deterministic_plans() { &MirrorEndpointSpec::Local(PathBuf::from("/target")), &TransferSelection::default(), true, + CompareMode::Auto, ) .expect("empty copy plan"); let remove = build_remove_plan( @@ -410,6 +427,7 @@ fn filtered_nested_tree_is_stably_sorted() { &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "root")), &selection, false, + CompareMode::Auto, ) .expect("filtered plan"); @@ -446,6 +464,7 @@ struct PathUploadCall { condition: RemoteWriteCondition, size: u64, staging: PathBuf, + identity_etag: Option, } struct TestRemoteSource { @@ -505,6 +524,7 @@ impl MirrorRemoteTransfer for TestRemoteSource { _source: &Path, _content_type: Option<&str>, _condition: RemoteWriteCondition, + _identity_etag: Option<&str>, ) -> rc_core::Result { Err(Error::General( "test source cannot upload mirror objects".to_string(), @@ -542,6 +562,7 @@ impl MirrorRemoteTransfer for TestRemoteTarget { source: &Path, content_type: Option<&str>, condition: RemoteWriteCondition, + identity_etag: Option<&str>, ) -> rc_core::Result { let size = tokio::fs::metadata(source).await?.len(); self.uploads @@ -553,6 +574,7 @@ impl MirrorRemoteTransfer for TestRemoteTarget { condition, size, staging: source.to_path_buf(), + identity_etag: identity_etag.map(ToOwned::to_owned), }); if let Some(error) = &self.upload_error { return Err(Error::Network(error.clone())); @@ -579,6 +601,7 @@ fn remote_transfer_fixture( size_bytes: Some(size), modified: Some(modified), etag: Some("source-etag".to_string()), + identity_etag: None, }, }; let operation = MirrorCopyOperation { @@ -643,6 +666,7 @@ async fn large_remote_copy_uses_path_streaming_preserves_metadata_and_cleans_sta Some("application/octet-stream") ); assert_eq!(uploads[0].condition, RemoteWriteCondition::IfAbsent); + assert_eq!(uploads[0].identity_etag.as_deref(), Some("source-etag")); assert!(!uploads[0].staging.exists()); } @@ -769,6 +793,26 @@ async fn truncated_remote_download_is_rejected_and_cleaned_before_upload() { assert!(!downloads[0].exists()); } +#[tokio::test] +async fn remote_source_preflight_runs_before_target_identity_check() { + let (mut source, operation) = remote_transfer_fixture(4, None); + source.info.etag = Some("replacement-etag".to_string()); + let MirrorLocation::Remote(source_path) = &operation.source.location else { + panic!("expected remote source") + }; + let target_checked = Arc::new(Mutex::new(false)); + let target_checked_for_check = Arc::clone(&target_checked); + + let result = source_preflight_then_target(&source, source_path, &operation, || async move { + *target_checked_for_check.lock().expect("target check lock") = true; + Ok::<(), Error>(()) + }) + .await; + + assert!(matches!(result, Err(Error::Conflict(_)))); + assert!(!*target_checked.lock().expect("target check lock")); +} + #[tokio::test] async fn cancelling_a_remote_transfer_drops_and_removes_its_staging_file() { let (mut source, operation) = remote_transfer_fixture(4, None); @@ -833,9 +877,340 @@ fn remote_overwrite_requires_the_planned_etag() { 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)); + assert!(source_matches_target(&source, &target, CompareMode::Auto)); target.snapshot.etag = None; - assert!(!source_matches_target(&source, &target)); + assert!(!source_matches_target(&source, &target, CompareMode::Auto)); +} + +fn remote_entry_with_identity( + alias: &str, + key: &str, + relative: &str, + etag: &str, + identity_etag: &str, +) -> MirrorEntry { + let mut entry = remote_entry(alias, key, relative, etag); + entry.snapshot.identity_etag = Some(identity_etag.to_string()); + entry +} + +#[test] +fn auto_compare_skips_when_source_etag_is_preserved_in_destination_metadata() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "source-etag", + )]); + let target = manifest([remote_entry_with_identity( + "dst", + "backup/report.txt", + "report.txt", + "multipart-etag-1", + "source-etag", + )]); + + let plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + CompareMode::Auto, + ) + .expect("build identity-aware plan"); + + assert!(plan.items.is_empty()); + assert_eq!(plan.summary.skipped, 1); +} + +#[test] +fn runtime_identity_shortcut_requires_the_planned_destination_snapshot() { + let source = remote_entry("src", "source/report.txt", "report.txt", "source-etag"); + let current = remote_entry_with_identity( + "dst", + "backup/report.txt", + "report.txt", + "multipart-etag-1", + "source-etag", + ); + let mut operation = MirrorCopyOperation { + relative_path: "report.txt".to_string(), + source: source.clone(), + target: current.location.clone(), + target_before: Some(current.clone()), + }; + + assert!(matches!( + target_disposition_for_current(&operation, Some(¤t), CompareMode::Auto), + Ok(TargetDisposition::AlreadyComplete) + )); + + let mut replaced = current.clone(); + replaced.snapshot.etag = Some("multipart-etag-2".to_string()); + assert!(matches!( + target_disposition_for_current(&operation, Some(&replaced), CompareMode::Auto), + Err(Error::Conflict(_)) + )); + + operation.target_before = None; + assert!(matches!( + target_disposition_for_current(&operation, Some(¤t), CompareMode::Auto), + Err(Error::Conflict(_)) + )); +} + +#[test] +fn runtime_identity_shortcut_requires_the_current_source_snapshot() { + let source = remote_entry("src", "source/report.txt", "report.txt", "source-etag"); + let mut current = ObjectInfo::file("source/report.txt", 4); + current.last_modified = source.snapshot.modified; + current.etag = source.snapshot.etag.clone(); + assert!(ensure_snapshot_matches(&source, ¤t).is_ok()); + + current.etag = Some("replacement-etag".to_string()); + assert!(matches!( + ensure_snapshot_matches(&source, ¤t), + Err(Error::Conflict(_)) + )); +} + +#[test] +fn auto_compare_copies_when_identity_metadata_is_missing() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "source-etag", + )]); + let target = manifest([remote_entry( + "dst", + "backup/report.txt", + "report.txt", + "multipart-etag-1", + )]); + + let plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + CompareMode::Auto, + ) + .expect("build plan without identity metadata"); + + assert_eq!(plan.items.len(), 1); + assert_eq!(plan.summary.skipped, 0); +} + +#[test] +fn auto_compare_copies_when_identity_metadata_does_not_match() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "source-etag", + )]); + let target = manifest([remote_entry_with_identity( + "dst", + "backup/report.txt", + "report.txt", + "multipart-etag-1", + "other-etag", + )]); + + let plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + CompareMode::Auto, + ) + .expect("build plan with mismatched identity"); + + assert_eq!(plan.items.len(), 1); + assert_eq!(plan.summary.skipped, 0); +} + +#[test] +fn size_mismatch_never_skips_regardless_of_compare_mode() { + let source = remote_entry("src", "source/report.txt", "report.txt", "source-etag"); + let mut target = remote_entry_with_identity( + "dst", + "backup/report.txt", + "report.txt", + "source-etag", + "source-etag", + ); + target.snapshot.size_bytes = Some(8); + + for compare in [CompareMode::Auto, CompareMode::Etag, CompareMode::Size] { + assert!( + !source_matches_target(&source, &target, compare), + "{compare:?} must copy when sizes differ" + ); + assert!( + !destination_needs_identity_lookup(&source, &target, compare), + "{compare:?} must not HeadObject when sizes differ" + ); + } + + let plan = build_copy_plan( + &manifest([source]), + &manifest([target]), + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + CompareMode::Auto, + ) + .expect("build plan for size mismatch"); + + assert_eq!(plan.items.len(), 1); + assert_eq!(plan.summary.skipped, 0); +} + +#[test] +fn etag_compare_still_copies_when_only_identity_metadata_matches() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "source-etag", + )]); + let target = manifest([remote_entry_with_identity( + "dst", + "backup/report.txt", + "report.txt", + "multipart-etag-1", + "source-etag", + )]); + + let plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + CompareMode::Etag, + ) + .expect("build etag-only plan"); + + assert_eq!(plan.items.len(), 1); +} + +#[test] +fn size_compare_skips_when_sizes_match_even_if_etags_differ() { + let source = manifest([remote_entry( + "src", + "source/report.txt", + "report.txt", + "source-etag", + )]); + let target = manifest([remote_entry( + "dst", + "backup/report.txt", + "report.txt", + "multipart-etag-1", + )]); + + let plan = build_copy_plan( + &source, + &target, + &MirrorEndpointSpec::Remote(RemotePath::new("dst", "bucket", "backup")), + &TransferSelection::default(), + true, + CompareMode::Size, + ) + .expect("build size-only plan"); + + assert!(plan.items.is_empty()); + assert_eq!(plan.summary.skipped, 1); +} + +#[test] +fn identity_metadata_is_read_case_insensitively() { + let metadata = HashMap::from([("Rc-Source-Etag".to_string(), "abc".to_string())]); + assert_eq!( + identity_etag_from_metadata(Some(&metadata)).as_deref(), + Some("abc") + ); + assert_eq!( + identity_etag_from_metadata(Some(&HashMap::from([( + "x-amz-meta-rc-source-etag".to_string(), + "abc".to_string() + )]))) + .as_deref(), + Some("abc") + ); + assert_eq!( + identity_etag_from_metadata(Some(&HashMap::from([( + "rc-source-etag".to_string(), + String::new() + )]))), + None + ); + assert_eq!(identity_etag_from_metadata(None), None); +} + +#[test] +fn destination_identity_lookup_is_limited_to_auto_remote_etag_mismatches() { + let source = remote_entry("src", "source/file.txt", "file.txt", "source-etag"); + let mismatched = remote_entry("dst", "backup/file.txt", "file.txt", "other-etag"); + let matching = remote_entry("dst", "backup/file.txt", "file.txt", "source-etag"); + let already_enriched = remote_entry_with_identity( + "dst", + "backup/file.txt", + "file.txt", + "other-etag", + "source-etag", + ); + let local_source = local_entry("/src", "file.txt"); + + assert!(destination_needs_identity_lookup( + &source, + &mismatched, + CompareMode::Auto + )); + assert!(!destination_needs_identity_lookup( + &source, + &matching, + CompareMode::Auto + )); + assert!(!destination_needs_identity_lookup( + &source, + &already_enriched, + CompareMode::Auto + )); + assert!(!destination_needs_identity_lookup( + &source, + &mismatched, + CompareMode::Etag + )); + assert!(!destination_needs_identity_lookup( + &source, + &mismatched, + CompareMode::Size + )); + assert!(!destination_needs_identity_lookup( + &local_source, + &mismatched, + CompareMode::Auto + )); +} + +#[test] +fn destination_race_check_ignores_identity_metadata() { + let planned = snapshot(4, "2026-07-21T04:00:00Z", Some("dest-etag")); + let current = snapshot_with_identity( + 4, + "2026-07-21T04:00:00Z", + Some("dest-etag"), + Some("source-etag"), + ); + assert!(planned.content_matches(¤t)); + assert_ne!(planned, current); } #[tokio::test] diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index 22a24fde..2b22ead0 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -439,6 +439,7 @@ fn top_level_command_help_contract() { "--retry-initial-backoff-ms", "--retry-max-backoff-ms", "--summary", + "--compare", ], }, HelpCase { diff --git a/crates/cli/tests/mirror_planner.rs b/crates/cli/tests/mirror_planner.rs index 33b44391..af0f32d7 100644 --- a/crates/cli/tests/mirror_planner.rs +++ b/crates/cli/tests/mirror_planner.rs @@ -299,6 +299,275 @@ fn remote_pagination_is_consumed_before_the_deterministic_plan_is_emitted() { assert!(requests.iter().all(|request| request.starts_with("GET "))); } +fn destination_list_response_with(etag: &str, size: u64) -> String { + format!( + r#" + + target-bucket + backup/ + 1 + 1000 + false + + backup/nested/file.txt + 2026-07-21T04:00:00.000Z + "{etag}" + {size} + STANDARD + +"# + ) +} + +fn start_identity_server( + expected_requests: usize, + destination_etag: &'static str, + identity_etag: Option<&'static str>, +) -> (String, thread::JoinHandle>) { + start_identity_server_with(expected_requests, destination_etag, identity_etag, 4) +} + +fn start_identity_server_with( + expected_requests: usize, + destination_etag: &'static str, + identity_etag: Option<&'static str>, + destination_size: u64, +) -> (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(); + 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 response = if request_line.starts_with("HEAD ") { + let mut headers = vec![ + "HTTP/1.1 200 OK".to_string(), + "content-length: 4".to_string(), + format!("etag: \"{destination_etag}\""), + "last-modified: Tue, 21 Jul 2026 04:00:00 GMT".to_string(), + "connection: close".to_string(), + ]; + if let Some(identity_etag) = identity_etag { + headers.push(format!("x-amz-meta-rc-source-etag: {identity_etag}")); + } + format!("{}\r\n\r\n", headers.join("\r\n")) + } else { + let body = if request_line.contains("/source-bucket") { + source_list_response().to_string() + } else { + destination_list_response_with(destination_etag, destination_size) + }; + 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 assert_read_only_identity_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.starts_with("HEAD ") }), + "identity planning sent a mutating request: {requests:?}" + ); + assert!( + requests + .iter() + .any(|request| request.starts_with("HEAD ") && request.contains("target-bucket")), + "auto compare should HeadObject the destination: {requests:?}" + ); +} + +#[test] +fn remote_to_remote_auto_compare_skips_when_destination_preserves_source_etag() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, handle) = start_identity_server(3, "multipart-etag-1", Some("source-etag")); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + "test/target-bucket/backup/", + "--overwrite", + "--dry-run", + "--compare", + "auto", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 0); + assert_eq!(payload["skipped"], 1); + assert_eq!(payload["dry_run"], true); + assert_read_only_identity_requests(handle, 3); +} + +#[test] +fn remote_to_remote_etag_compare_recopies_when_stored_etags_differ() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, handle) = start_identity_server(2, "multipart-etag-1", Some("source-etag")); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + "test/target-bucket/backup/", + "--overwrite", + "--dry-run", + "--compare", + "etag", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["dry_run"], true); + let requests = handle.join().expect("join mock S3 endpoint"); + assert_eq!(requests.len(), 2, "{requests:?}"); + assert!( + requests.iter().all(|request| request.starts_with("GET ")), + "etag compare should not HeadObject destinations: {requests:?}" + ); +} + +#[test] +fn remote_to_remote_auto_compare_recopies_when_destination_identity_is_missing() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, handle) = start_identity_server(3, "multipart-etag-1", None); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + "test/target-bucket/backup/", + "--overwrite", + "--dry-run", + "--compare", + "auto", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["skipped"], 0); + assert_eq!(payload["dry_run"], true); + assert_read_only_identity_requests(handle, 3); +} + +#[test] +fn remote_to_remote_auto_compare_recopies_when_destination_identity_mismatches() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, handle) = start_identity_server(3, "multipart-etag-1", Some("other-etag")); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + "test/target-bucket/backup/", + "--overwrite", + "--dry-run", + "--compare", + "auto", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["skipped"], 0); + assert_eq!(payload["dry_run"], true); + assert_read_only_identity_requests(handle, 3); +} + +#[test] +fn remote_to_remote_auto_compare_recopies_when_sizes_differ_without_head() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, handle) = start_identity_server_with(2, "multipart-etag-1", None, 8); + + let output = run_rc( + &[ + "--json", + "mirror", + "test/source-bucket/source/", + "test/target-bucket/backup/", + "--overwrite", + "--dry-run", + "--compare", + "auto", + ], + config_dir.path(), + Some(&endpoint), + ); + + let payload = parse_success(&output); + assert_eq!(payload["copied"], 1); + assert_eq!(payload["skipped"], 0); + assert_eq!(payload["dry_run"], true); + let requests = handle.join().expect("join mock S3 endpoint"); + assert_eq!(requests.len(), 2, "{requests:?}"); + assert!( + requests.iter().all(|request| request.starts_with("GET ")), + "size mismatch should not HeadObject destinations: {requests:?}" + ); +} + #[test] fn local_to_local_is_rejected_with_the_unsupported_exit_code() { let config_dir = tempfile::tempdir().expect("create config dir"); diff --git a/crates/cli/tests/rm_purge.rs b/crates/cli/tests/rm_purge.rs index eeb4d438..bb52b2f9 100644 --- a/crates/cli/tests/rm_purge.rs +++ b/crates/cli/tests/rm_purge.rs @@ -49,6 +49,10 @@ impl TestServer { while !thread_stop.load(Ordering::SeqCst) { match listener.accept() { Ok((mut stream, _)) => { + // Accepted sockets can inherit nonblocking mode on some platforms. + stream + .set_nonblocking(false) + .expect("set accepted stream blocking"); if let Some(request) = read_request(&mut stream) { let response = response_for(&request); thread_requests diff --git a/crates/cli/tests/versioned_objects.rs b/crates/cli/tests/versioned_objects.rs index 6818ea02..18f82c9a 100644 --- a/crates/cli/tests/versioned_objects.rs +++ b/crates/cli/tests/versioned_objects.rs @@ -63,6 +63,10 @@ impl TestServer { while !thread_stop.load(Ordering::SeqCst) { match listener.accept() { Ok((mut stream, _)) => { + // Accepted sockets can inherit nonblocking mode on some platforms. + stream + .set_nonblocking(false) + .expect("set accepted stream blocking"); if let Some(request) = read_request(&mut stream) { let response = response_for(mode, &request); thread_requests diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index bf7844d8..c3540413 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -3979,10 +3979,47 @@ impl S3Client { encryption: encryption.cloned().map(ObjectWriteEncryption::Managed), ..ObjectWriteOptions::default() }; - self.put_object_from_path_with_condition( + self.put_object_from_path_if_match_with_options( path, file_path, &options, + etag, + on_progress, + ) + .await + } + + /// Upload a local file with explicit write options only when the object is absent. + pub async fn put_object_from_path_if_absent_with_options( + &self, + path: &RemotePath, + file_path: &std::path::Path, + options: &ObjectWriteOptions, + on_progress: impl Fn(u64) + Send, + ) -> Result { + self.put_object_from_path_with_condition( + path, + file_path, + options, + ObjectWritePrecondition::IfAbsent, + on_progress, + ) + .await + } + + /// Upload a local file with explicit write options only when `etag` still matches. + pub async fn put_object_from_path_if_match_with_options( + &self, + path: &RemotePath, + file_path: &std::path::Path, + options: &ObjectWriteOptions, + etag: &str, + on_progress: impl Fn(u64) + Send, + ) -> Result { + self.put_object_from_path_with_condition( + path, + file_path, + options, ObjectWritePrecondition::IfMatch(etag), on_progress, ) diff --git a/docs/reference/rc/mirror.md b/docs/reference/rc/mirror.md index bec2801b..5029f94e 100644 --- a/docs/reference/rc/mirror.md +++ b/docs/reference/rc/mirror.md @@ -30,6 +30,7 @@ rc [GLOBAL OPTIONS] mirror [OPTIONS] | `--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. | +| `--compare ` | Choose how existing destination objects are compared before a copy is skipped. Defaults to `auto`. | | `--quiet` | Suppress non-error command output. The global `--quiet` option has the same effect. | ## Examples @@ -39,6 +40,7 @@ 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 +rc mirror stage/data/ prod/data/ --overwrite --compare auto ``` ## Behavior @@ -51,7 +53,17 @@ Remote keys that are absolute, contain traversal, use backslashes, collide after ### 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. +Entries are compared by size and the strongest stable metadata available. `--compare` selects the skip rule: + +| Mode | Skip a copy when | +| --- | --- | +| `auto` (default) | Destination and source ETags match, or sizes match and destination user metadata `x-amz-meta-rc-source-etag` records the source ETag from a previous `rc mirror` copy. | +| `etag` | Destination and source ETags are identical. Multipart re-uploads that change the stored ETag are treated as different. | +| `size` | Object sizes match, ignoring ETag differences. | + +Remote-to-remote copies download through a temporary file and upload to the destination. Multipart completion often stores a different ETag than the source, so a second `auto` run would recopy every object if it compared ListObjects ETags alone. `rc mirror` therefore writes `x-amz-meta-rc-source-etag` on remote uploads that have a source ETag. ListObjects does not return user metadata, so `auto` issues HeadObject for same-size destinations whose listed ETags differ. + +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. @@ -71,6 +83,10 @@ Copy and removal phases use the shared transfer controls for filtering, concurre 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. +### BREAKING incremental identity contract migration + +`--compare auto|etag|size` and destination metadata `x-amz-meta-rc-source-etag` are additive. Default skip-on-matching-ETag behavior is unchanged. Objects copied before this change still recopy once under `auto`, then skip. This PR must be marked `BREAKING` because `docs/reference/rc/mirror.md` is a protected CLI behavior contract. No JSON schema or config `schema_version` bump applies. + Global options shown in command syntax use the same meaning everywhere: | Option | Description | diff --git a/scripts/regression/mirror-identity.sh b/scripts/regression/mirror-identity.sh new file mode 100755 index 00000000..6ada4038 --- /dev/null +++ b/scripts/regression/mirror-identity.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +# +# Regression tests for incremental mirror identity (#342). +# +# Usage: +# ./scripts/regression/mirror-identity.sh +# +# These tests do not require a running S3 backend. They cover: +# - Auto compare skipping recopies when destination metadata preserves the +# source ETag after a multipart re-upload +# - ETag compare still treating a changed stored ETag as different +# - Size compare ignoring ETag differences +# - CLI help/parse contract for --compare +# + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +cd "$PROJECT_ROOT" + +RED='\033[0;31m' +GREEN='\033[0;32m' +BLUE='\033[0;34m' +NC='\033[0m' + +log_info() { echo -e "${BLUE}[INFO]${NC} $*"; } +log_success() { echo -e "${GREEN}[PASS]${NC} $*"; } +log_error() { echo -e "${RED}[FAIL]${NC} $*"; } + +run_tests() { + local description="$1" + shift + log_info "$description" + if "$@"; then + log_success "$description" + else + log_error "$description" + return 1 + fi +} + +log_info "Mirror identity regression suite" + +run_tests "mirror identity unit tests" \ + cargo test -p rustfs-cli --lib -- \ + auto_compare_skips_when_source_etag_is_preserved_in_destination_metadata \ + auto_compare_copies_when_identity_metadata_is_missing \ + auto_compare_copies_when_identity_metadata_does_not_match \ + size_mismatch_never_skips_regardless_of_compare_mode \ + etag_compare_still_copies_when_only_identity_metadata_matches \ + size_compare_skips_when_sizes_match_even_if_etags_differ \ + identity_metadata_is_read_case_insensitively \ + destination_identity_lookup_is_limited_to_auto_remote_etag_mismatches \ + destination_race_check_ignores_identity_metadata \ + large_remote_copy_uses_path_streaming_preserves_metadata_and_cleans_staging \ + remote_entries_without_two_etags_are_never_assumed_equal \ + equivalent_destination_is_restart_safe_and_not_planned_again \ + changed_destination_requires_overwrite + +run_tests "mirror identity planner integration tests" \ + cargo test -p rustfs-cli --test mirror_planner -- \ + remote_to_remote_auto_compare_skips_when_destination_preserves_source_etag \ + remote_to_remote_etag_compare_recopies_when_stored_etags_differ \ + remote_to_remote_auto_compare_recopies_when_destination_identity_is_missing \ + remote_to_remote_auto_compare_recopies_when_destination_identity_mismatches \ + remote_to_remote_auto_compare_recopies_when_sizes_differ_without_head \ + remote_to_remote_dry_run_reads_both_manifests_without_mutation + +run_tests "CLI help contract includes --compare" \ + cargo test -p rustfs-cli --test help_contract -- top_level_command_help_contract + +log_success "Mirror identity regression suite passed"