1use clap::Args;
6use jiff::Timestamp;
7use rc_core::alias::RetryConfig;
8use rc_core::{
9 AliasManager, Error, MetadataDirective, MultipartCopyCancellation, MultipartCopyOptions,
10 ObjectAttributes, ObjectEncryptionRequest, ObjectInfo, ObjectKeyPolicy, ObjectStore as _,
11 ObjectWriteOptions, ParsedPath, RemotePath, SseCustomerKey, TransferCancellation,
12 TransferCandidate, TransferControls, TransferCopyOptions, TransferExecutor,
13 TransferOutcomeState, TransferPlan, TransferReadOptions, TransferSelection,
14 normalize_relative_key, parse_path, relative_local_path_from_key,
15};
16use rc_s3::S3Client;
17use serde::Serialize;
18use std::collections::{BTreeSet, HashMap, HashSet};
19use std::fmt;
20use std::path::{Path, PathBuf};
21use std::sync::{Arc, Mutex as StdMutex};
22use tokio::sync::Mutex as AsyncMutex;
23
24use crate::exit_code::ExitCode;
25use crate::output::{Formatter, OutputConfig, ProgressBar, V3SuccessEnvelope};
26use crate::secret_input::{SecretLocator, resolve_secret_locator};
27
28use super::object_identity::set_source_identity;
29use super::transfer_fidelity::{MetadataDirectiveArg, TaggingDirectiveArg, TransferFidelityArgs};
30
31const CP_AFTER_HELP: &str = "\
32Examples:
33 rc object copy ./report.json local/my-bucket/reports/
34 rc cp ./report.json local/my-bucket/reports/
35 rc object copy local/source-bucket/archive.tar.gz ./downloads/archive.tar.gz";
36
37pub(crate) const GET_AFTER_HELP: &str = "\
38Examples:
39 rc get local/my-bucket/report.json ./report.json
40 rc get local/my-bucket/archive.tar.gz ./downloads/archive.tar.gz";
41
42pub(crate) const PUT_AFTER_HELP: &str = "\
43Examples:
44 rc put ./report.json local/my-bucket/reports/
45 rc put ./january.csv ./february.csv local/my-bucket/reports/";
46
47const REMOTE_PATH_SUGGESTION: &str =
48 "Use a local filesystem path or a remote path in the form alias/bucket[/key].";
49const DEFAULT_TRANSFER_CONCURRENCY: usize = 4;
50const DEFAULT_RETRY_ATTEMPTS: u32 = 3;
51const DEFAULT_RETRY_INITIAL_BACKOFF_MS: u64 = 100;
52const DEFAULT_RETRY_MAX_BACKOFF_MS: u64 = 10_000;
53#[cfg(test)]
54const MAX_SINGLE_COPY_SIZE: u64 = rc_core::S3_SINGLE_COPY_MAX_SIZE;
55
56#[derive(Args, Clone)]
58#[command(after_help = CP_AFTER_HELP)]
59pub struct CpArgs {
60 #[arg(required = true, num_args = 1.., value_name = "SOURCE")]
62 pub sources: Vec<String>,
63
64 pub target: String,
66
67 #[arg(short, long)]
69 pub recursive: bool,
70
71 #[arg(short, long, conflicts_with = "metadata_directive")]
73 pub preserve: bool,
74
75 #[arg(long, value_enum)]
77 pub(crate) metadata_directive: Option<MetadataDirectiveArg>,
78
79 #[arg(long, value_enum)]
81 pub(crate) tagging_directive: Option<TaggingDirectiveArg>,
82
83 #[arg(long)]
85 pub continue_on_error: bool,
86
87 #[arg(
89 long,
90 default_value_t = true,
91 action = clap::ArgAction::Set,
92 num_args = 0..=1,
93 default_missing_value = "true"
94 )]
95 pub overwrite: bool,
96
97 #[arg(long)]
99 pub skip_existing: bool,
100
101 #[arg(long)]
103 pub dry_run: bool,
104
105 #[arg(long)]
107 pub storage_class: Option<String>,
108
109 #[arg(long)]
111 pub content_type: Option<String>,
112
113 #[command(flatten)]
114 pub(crate) fidelity: TransferFidelityArgs,
115
116 #[arg(long = "enc-s3")]
118 pub enc_s3: Vec<String>,
119
120 #[arg(long = "enc-kms")]
122 pub enc_kms: Vec<String>,
123
124 #[arg(long = "enc-c-source-key-file")]
126 pub enc_c_source_key_file: Option<PathBuf>,
127
128 #[arg(long = "enc-c-source-key-env")]
130 pub enc_c_source_key_env: Option<String>,
131
132 #[arg(long = "enc-c-destination-key-file")]
134 pub enc_c_destination_key_file: Option<PathBuf>,
135
136 #[arg(long = "enc-c-destination-key-env")]
138 pub enc_c_destination_key_env: Option<String>,
139
140 #[arg(skip)]
141 pub(crate) source_customer_key: Option<SseCustomerKey>,
142
143 #[arg(skip)]
144 pub(crate) destination_customer_key: Option<SseCustomerKey>,
145
146 #[arg(long)]
148 pub include: Vec<String>,
149
150 #[arg(long)]
152 pub exclude: Vec<String>,
153
154 #[arg(long)]
156 pub newer_than: Option<String>,
157
158 #[arg(long)]
160 pub older_than: Option<String>,
161
162 #[arg(long)]
164 pub rewind: Option<String>,
165
166 #[arg(long)]
168 pub concurrency: Option<usize>,
169
170 #[arg(long)]
172 pub rate_limit: Option<String>,
173
174 #[arg(long)]
176 pub retry_attempts: Option<u32>,
177
178 #[arg(long)]
180 pub retry_initial_backoff_ms: Option<u64>,
181
182 #[arg(long)]
184 pub retry_max_backoff_ms: Option<u64>,
185
186 #[arg(long)]
188 pub fail_empty: bool,
189
190 #[arg(long)]
192 pub summary: bool,
193
194 #[arg(long)]
196 pub portable_names: bool,
197}
198
199impl fmt::Debug for CpArgs {
200 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
201 formatter.write_str("CpArgs { .. }")
202 }
203}
204
205impl CpArgs {
206 pub(crate) fn single(source: impl Into<String>, target: impl Into<String>) -> Self {
207 Self {
208 sources: vec![source.into()],
209 target: target.into(),
210 recursive: false,
211 preserve: false,
212 metadata_directive: None,
213 tagging_directive: None,
214 continue_on_error: false,
215 overwrite: true,
216 skip_existing: false,
217 dry_run: false,
218 storage_class: None,
219 content_type: None,
220 fidelity: TransferFidelityArgs::default(),
221 enc_s3: Vec::new(),
222 enc_kms: Vec::new(),
223 enc_c_source_key_file: None,
224 enc_c_source_key_env: None,
225 enc_c_destination_key_file: None,
226 enc_c_destination_key_env: None,
227 source_customer_key: None,
228 destination_customer_key: None,
229 include: Vec::new(),
230 exclude: Vec::new(),
231 newer_than: None,
232 older_than: None,
233 rewind: None,
234 concurrency: None,
235 rate_limit: None,
236 retry_attempts: None,
237 retry_initial_backoff_ms: None,
238 retry_max_backoff_ms: None,
239 fail_empty: false,
240 summary: false,
241 portable_names: false,
242 }
243 }
244
245 fn local_key_policy(&self) -> ObjectKeyPolicy {
246 ObjectKeyPolicy::for_local_destination(self.portable_names)
247 }
248}
249
250#[derive(Args, Debug)]
252#[command(
253 override_usage = "rc get [OPTIONS] <SOURCE> <TARGET>",
254 after_help = GET_AFTER_HELP
255)]
256pub struct GetArgs {
257 #[command(flatten)]
258 pub transfer: CpArgs,
259}
260
261#[derive(Args, Debug)]
263#[command(after_help = PUT_AFTER_HELP)]
264pub struct PutArgs {
265 #[command(flatten)]
266 pub transfer: CpArgs,
267}
268
269#[derive(Clone, Copy, Debug, Eq, PartialEq)]
270enum TransferAlias {
271 Copy,
272 Get,
273 Put,
274}
275
276#[derive(Debug, Serialize)]
277struct CpOutput {
278 status: &'static str,
279 source: String,
280 target: String,
281 #[serde(skip_serializing_if = "Option::is_none")]
282 size_bytes: Option<i64>,
283 #[serde(skip_serializing_if = "Option::is_none")]
284 size_human: Option<String>,
285 #[serde(skip_serializing_if = "Option::is_none")]
286 version_id: Option<String>,
287 #[serde(skip_serializing_if = "Option::is_none")]
288 source_version_id: Option<String>,
289}
290
291#[derive(Debug, Serialize)]
292struct VersionCopyData {
293 operation: &'static str,
294 source: String,
295 target: String,
296 source_version_id: Option<String>,
297 version_id: Option<String>,
298 size_bytes: Option<i64>,
299 size_human: Option<String>,
300}
301
302pub async fn execute(args: CpArgs, output_config: OutputConfig) -> ExitCode {
304 execute_with_alias(args, output_config, TransferAlias::Copy).await
305}
306
307pub async fn execute_get(args: GetArgs, output_config: OutputConfig) -> ExitCode {
309 execute_with_alias(args.transfer, output_config, TransferAlias::Get).await
310}
311
312pub async fn execute_put(args: PutArgs, output_config: OutputConfig) -> ExitCode {
314 execute_with_alias(args.transfer, output_config, TransferAlias::Put).await
315}
316
317async fn execute_with_alias(
318 mut args: CpArgs,
319 output_config: OutputConfig,
320 alias: TransferAlias,
321) -> ExitCode {
322 let formatter = Formatter::new(output_config);
323
324 if alias == TransferAlias::Get && args.sources.len() != 1 {
325 return formatter.fail(
326 ExitCode::UsageError,
327 "get requires exactly one remote source and one local target",
328 );
329 }
330
331 if let Err(error) = validate_destination_storage_class(args.storage_class.as_deref()) {
332 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
333 }
334 if let Err(error) = object_write_options(
335 &args.fidelity,
336 args.content_type.as_deref(),
337 None,
338 None,
339 args.storage_class.clone(),
340 ) {
341 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
342 }
343
344 if formatter.is_json() && uses_transfer_planner(&args) {
345 return formatter.fail(
346 ExitCode::UnsupportedFeature,
347 "Bulk transfer planning currently supports human output only; JSON batch output requires the versioned output contract",
348 );
349 }
350
351 let selection = match build_transfer_selection(&args, Timestamp::now()) {
352 Ok(selection) => selection,
353 Err(error) => return formatter.fail(ExitCode::UsageError, &error),
354 };
355 let controls = match build_transfer_controls(&args) {
356 Ok(controls) => controls,
357 Err(error) => return formatter.fail(ExitCode::UsageError, &error),
358 };
359
360 let alias_manager = AliasManager::new();
361
362 let mut sources = Vec::with_capacity(args.sources.len());
364 for source in &args.sources {
365 let parsed = match alias {
366 TransferAlias::Put => Ok(ParsedPath::Local(PathBuf::from(source))),
369 TransferAlias::Copy | TransferAlias::Get => {
370 parse_cp_path(source, alias_manager.as_ref().ok())
371 }
372 };
373 match parsed {
374 Ok(path) => sources.push(path),
375 Err(error) => {
376 return formatter.fail_with_suggestion(
377 ExitCode::UsageError,
378 &format!("Invalid source path '{source}': {error}"),
379 REMOTE_PATH_SUGGESTION,
380 );
381 }
382 }
383 }
384
385 let parsed_target = match alias {
386 TransferAlias::Get => Ok(ParsedPath::Local(PathBuf::from(&args.target))),
389 TransferAlias::Copy | TransferAlias::Put => {
390 parse_cp_path(&args.target, alias_manager.as_ref().ok())
391 }
392 };
393 let target = match parsed_target {
394 Ok(p) => p,
395 Err(e) => {
396 return formatter.fail_with_suggestion(
397 ExitCode::UsageError,
398 &format!("Invalid target path: {e}"),
399 REMOTE_PATH_SUGGESTION,
400 );
401 }
402 };
403 if let Err(error) = validate_alias_direction(alias, &sources, &target) {
404 return formatter.fail(ExitCode::UsageError, error);
405 }
406 if let Err(error) = validate_fidelity_directions(&args, &sources, &target) {
407 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
408 }
409 let source_key_locator = match resolve_secret_locator(
410 args.enc_c_source_key_file.clone(),
411 args.enc_c_source_key_env.clone(),
412 ) {
413 Ok(locator) => locator,
414 Err(error) => {
415 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
416 }
417 };
418 let destination_key_locator = match resolve_secret_locator(
419 args.enc_c_destination_key_file.clone(),
420 args.enc_c_destination_key_env.clone(),
421 ) {
422 Ok(locator) => locator,
423 Err(error) => {
424 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
425 }
426 };
427 if source_key_locator.is_some()
428 && sources
429 .iter()
430 .any(|path| matches!(path, ParsedPath::Local(_)))
431 {
432 return formatter.fail(
433 ExitCode::UsageError,
434 "SSE-C source keys require every source to be remote",
435 );
436 }
437 if destination_key_locator.is_some() && matches!(target, ParsedPath::Local(_)) {
438 return formatter.fail(
439 ExitCode::UsageError,
440 "SSE-C destination keys require a remote destination",
441 );
442 }
443 if args.storage_class.is_some() && matches!(target, ParsedPath::Local(_)) {
444 return formatter.fail(
445 ExitCode::UsageError,
446 "--storage-class requires a remote destination",
447 );
448 }
449
450 let target_is_container = is_container_target(&args.target, &target);
451 if args.sources.len() > 1 && !target_is_container {
452 return formatter.fail(
453 ExitCode::UsageError,
454 "Multiple copy sources require a directory or remote prefix destination ending in '/'",
455 );
456 }
457
458 let target_encryption = match parse_destination_encryption(&args.enc_s3, &args.enc_kms, &target)
459 {
460 Ok(encryption) => encryption,
461 Err(error) => return formatter.fail(ExitCode::UsageError, &error),
462 };
463 if destination_key_locator.is_some() && target_encryption.is_some() {
464 return formatter.fail(
465 ExitCode::UsageError,
466 "SSE-C destination keys cannot be combined with --enc-s3 or --enc-kms",
467 );
468 }
469 if !args.dry_run
470 && (source_key_locator.is_some() || destination_key_locator.is_some())
471 && sources
472 .iter()
473 .all(|path| matches!(path, ParsedPath::Remote(_)))
474 && matches!(target, ParsedPath::Remote(_))
475 {
476 return formatter.fail(
477 ExitCode::UnsupportedFeature,
478 "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467",
479 );
480 }
481 if !args.dry_run {
482 args.source_customer_key = match load_customer_key(source_key_locator.as_ref()) {
483 Ok(key) => key,
484 Err(error) => {
485 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
486 }
487 };
488 args.destination_customer_key = match load_customer_key(destination_key_locator.as_ref()) {
489 Ok(key) => key,
490 Err(error) => {
491 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
492 }
493 };
494 }
495
496 let single_local_to_local = matches!(
497 (sources.as_slice(), &target),
498 ([ParsedPath::Local(_)], ParsedPath::Local(_))
499 );
500 if uses_transfer_planner(&args) && !single_local_to_local {
501 let alias_manager = match alias_manager.as_ref() {
502 Ok(manager) => manager,
503 Err(error) => {
504 return formatter.fail(
505 ExitCode::UsageError,
506 &format!("Failed to load aliases: {error}"),
507 );
508 }
509 };
510 return execute_transfer_plan(
511 &args,
512 sources,
513 target,
514 target_is_container,
515 target_encryption,
516 selection,
517 controls,
518 &formatter,
519 alias_manager,
520 )
521 .await;
522 }
523
524 let Some(source) = sources.first() else {
525 return formatter.fail(ExitCode::UsageError, "At least one copy source is required");
526 };
527
528 execute_single_copy(
529 source,
530 &target,
531 &args,
532 &formatter,
533 target_encryption.as_ref(),
534 )
535 .await
536}
537
538fn validate_alias_direction(
539 alias: TransferAlias,
540 sources: &[ParsedPath],
541 target: &ParsedPath,
542) -> Result<(), &'static str> {
543 match alias {
544 TransferAlias::Copy => Ok(()),
545 TransferAlias::Get if sources.len() == 1 && sources[0].is_remote() && target.is_local() => {
546 Ok(())
547 }
548 TransferAlias::Get => Err("get requires exactly one remote source and one local target"),
549 TransferAlias::Put if sources.iter().all(ParsedPath::is_local) && target.is_remote() => {
550 Ok(())
551 }
552 TransferAlias::Put => Err("put requires one or more local sources and one remote target"),
553 }
554}
555
556fn uses_transfer_planner(args: &CpArgs) -> bool {
557 args.sources.len() > 1
558 || args.recursive
559 || args.skip_existing
560 || !args.overwrite
561 || !args.include.is_empty()
562 || !args.exclude.is_empty()
563 || args.newer_than.is_some()
564 || args.older_than.is_some()
565 || args.rewind.is_some()
566 || args.concurrency.is_some()
567 || args.rate_limit.is_some()
568 || args.retry_attempts.is_some()
569 || args.retry_initial_backoff_ms.is_some()
570 || args.retry_max_backoff_ms.is_some()
571 || args.fail_empty
572 || args.summary
573}
574
575pub(super) fn validate_destination_storage_class(value: Option<&str>) -> rc_core::Result<()> {
576 let Some(value) = value else {
577 return Ok(());
578 };
579 match value {
580 "STANDARD" | "REDUCED_REDUNDANCY" => Ok(()),
581 "DEEP_ARCHIVE"
582 | "EXPRESS_ONEZONE"
583 | "FSX_ONTAP"
584 | "FSX_OPENZFS"
585 | "GLACIER"
586 | "GLACIER_IR"
587 | "INTELLIGENT_TIERING"
588 | "ONEZONE_IA"
589 | "OUTPOSTS"
590 | "SNOW"
591 | "STANDARD_IA" => Err(Error::UnsupportedFeature(format!(
592 "RustFS beta.10 does not provide meaningful storage policy '{value}'"
593 ))),
594 value => Err(Error::InvalidPath(format!(
595 "Unknown destination storage class '{value}'"
596 ))),
597 }
598}
599
600fn object_write_options(
601 fidelity: &TransferFidelityArgs,
602 content_type: Option<&str>,
603 encryption: Option<&ObjectEncryptionRequest>,
604 customer_key: Option<&SseCustomerKey>,
605 storage_class: Option<String>,
606) -> rc_core::Result<ObjectWriteOptions> {
607 fidelity.build_write_options(content_type, encryption, customer_key, storage_class)
608}
609
610fn requested_metadata_directive(args: &CpArgs) -> Option<MetadataDirective> {
611 if args.preserve {
612 Some(MetadataDirective::Copy)
613 } else {
614 args.metadata_directive.map(Into::into)
615 }
616}
617
618fn transfer_copy_options(
619 args: &CpArgs,
620 source_version_id: Option<String>,
621 encryption: Option<&ObjectEncryptionRequest>,
622) -> rc_core::Result<TransferCopyOptions> {
623 let metadata_directive = requested_metadata_directive(args);
624 let tagging_directive = args.tagging_directive.map(Into::into);
625 let mut destination = object_write_options(
626 &args.fidelity,
627 args.content_type.as_deref(),
628 encryption,
629 args.destination_customer_key.as_ref(),
630 args.storage_class.clone(),
631 )?;
632 if matches!(metadata_directive, Some(MetadataDirective::Replace))
633 && destination.attributes.is_none()
634 {
635 destination.attributes = Some(Default::default());
636 }
637 if matches!(tagging_directive, Some(rc_core::TaggingDirective::Replace))
638 && destination.tags.is_none()
639 {
640 destination.tags = Some(Default::default());
641 }
642 let options = TransferCopyOptions {
643 source: TransferReadOptions {
644 version_id: source_version_id,
645 customer_key: args.source_customer_key.clone(),
646 ..TransferReadOptions::default()
647 },
648 metadata_directive,
649 tagging_directive,
650 destination,
651 };
652 options.validate()?;
653 Ok(options)
654}
655
656fn validate_fidelity_directions(
657 args: &CpArgs,
658 sources: &[ParsedPath],
659 target: &ParsedPath,
660) -> rc_core::Result<()> {
661 let all_remote = sources
662 .iter()
663 .all(|source| matches!(source, ParsedPath::Remote(_)));
664 let any_remote = sources
665 .iter()
666 .any(|source| matches!(source, ParsedPath::Remote(_)));
667 let target_remote = matches!(target, ParsedPath::Remote(_));
668 let has_copy_directive =
669 args.preserve || args.metadata_directive.is_some() || args.tagging_directive.is_some();
670 if has_copy_directive && !(all_remote && target_remote) {
671 return Err(Error::InvalidPath(
672 "Metadata and tagging copy directives require remote sources and a remote destination"
673 .to_string(),
674 ));
675 }
676 if !target_remote
677 && (args.storage_class.is_some()
678 || args.content_type.is_some()
679 || args.fidelity.has_write_policy()
680 || !args.enc_s3.is_empty()
681 || !args.enc_kms.is_empty()
682 || args.enc_c_destination_key_file.is_some()
683 || args.enc_c_destination_key_env.is_some())
684 {
685 return Err(Error::InvalidPath(
686 "Destination transfer policies require a remote destination".to_string(),
687 ));
688 }
689 let same_alias_remote_copy = target_remote
690 && sources.iter().any(|source| {
691 matches!(
692 source,
693 ParsedPath::Remote(source)
694 if target
695 .as_remote()
696 .is_some_and(|target| source.alias == target.alias)
697 )
698 });
699 if any_remote && target_remote {
700 let copy_options = transfer_copy_options(args, None, None)?;
701 if same_alias_remote_copy
702 && matches!(
703 copy_options.metadata_directive,
704 Some(MetadataDirective::Replace)
705 )
706 {
707 return Err(Error::UnsupportedFeature(
708 "RustFS beta.10 does not preserve complete metadata REPLACE semantics; tracked by rustfs/backlog#1463"
709 .to_string(),
710 ));
711 }
712 if copy_options.tagging_directive.is_some() || copy_options.destination.tags.is_some() {
713 return Err(Error::UnsupportedFeature(
714 "RustFS beta.10 does not preserve CopyObject tagging directives; tracked by rustfs/backlog#1462"
715 .to_string(),
716 ));
717 }
718 if copy_options.destination.checksum.is_some() {
719 return Err(Error::UnsupportedFeature(
720 "RustFS beta.10 does not preserve CopyObject checksum selection; tracked by rustfs/backlog#1466"
721 .to_string(),
722 ));
723 }
724 }
725 Ok(())
726}
727
728fn load_customer_key(locator: Option<&SecretLocator>) -> rc_core::Result<Option<SseCustomerKey>> {
729 locator.map(SecretLocator::load_customer_key).transpose()
730}
731
732async fn execute_single_copy(
733 source: &ParsedPath,
734 target: &ParsedPath,
735 args: &CpArgs,
736 formatter: &Formatter,
737 encryption: Option<&ObjectEncryptionRequest>,
738) -> ExitCode {
739 match (source, target) {
741 (ParsedPath::Local(src), ParsedPath::Remote(dst)) => {
742 copy_local_to_s3_prepared(src, dst, args, formatter, encryption).await
743 }
744 (ParsedPath::Remote(src), ParsedPath::Local(dst)) => {
745 copy_s3_to_local(src, dst, args, formatter).await
746 }
747 (ParsedPath::Remote(src), ParsedPath::Remote(dst)) => {
748 if args.recursive {
749 return formatter.fail(
750 ExitCode::UnsupportedFeature,
751 "Recursive S3-to-S3 copy is not implemented",
752 );
753 }
754 copy_s3_to_s3_prepared(src, dst, args, formatter, encryption).await
755 }
756 (ParsedPath::Local(_), ParsedPath::Local(_)) => formatter.fail_with_suggestion(
757 ExitCode::UsageError,
758 "Cannot copy between two local paths. Use system cp command.",
759 "Use your local shell cp command when both paths are on the filesystem.",
760 ),
761 }
762}
763
764#[derive(Debug, Clone)]
765enum CpOperation {
766 LocalToRemote {
767 source: PathBuf,
768 target: RemotePath,
769 encryption: Option<ObjectEncryptionRequest>,
770 },
771 RemoteToLocal {
772 source: RemotePath,
773 target: PathBuf,
774 },
775 RemoteToRemote {
776 source: RemotePath,
777 target: RemotePath,
778 source_info: Box<ObjectInfo>,
779 encryption: Option<ObjectEncryptionRequest>,
780 },
781}
782
783#[derive(Debug, Clone)]
784struct PlannedCopyDetail {
785 source_version_id: Option<String>,
786 destination_version_id: Option<String>,
787 upload_id: Option<String>,
788}
789
790type PlannedCopyDetails = Arc<AsyncMutex<HashMap<(String, String), PlannedCopyDetail>>>;
791
792#[derive(Debug)]
793struct PlannedCopyProgress {
794 positions: StdMutex<HashMap<(String, String), u64>>,
795 bar: Option<ProgressBar>,
796}
797
798impl PlannedCopyProgress {
799 fn new(output_config: OutputConfig, total: u64) -> Self {
800 Self {
801 positions: StdMutex::new(HashMap::new()),
802 bar: (total > 0).then(|| ProgressBar::new(output_config, total)),
803 }
804 }
805
806 fn reset(&self, key: &(String, String)) {
807 self.set(key, 0);
808 }
809
810 fn set(&self, key: &(String, String), bytes: u64) {
811 let mut positions = self
812 .positions
813 .lock()
814 .expect("planned copy progress lock should not be poisoned");
815 positions.insert(key.clone(), bytes);
816 let aggregate = positions.values().copied().fold(0_u64, u64::saturating_add);
817 if let Some(bar) = &self.bar {
818 bar.set_position(aggregate);
819 }
820 }
821
822 fn finish(&self) {
823 if let Some(bar) = &self.bar {
824 bar.finish_and_clear();
825 }
826 }
827}
828
829type SharedPlannedCopyProgress = Arc<PlannedCopyProgress>;
830
831#[allow(clippy::too_many_arguments)]
832async fn execute_transfer_plan(
833 args: &CpArgs,
834 sources: Vec<ParsedPath>,
835 target: ParsedPath,
836 target_is_container: bool,
837 encryption: Option<ObjectEncryptionRequest>,
838 selection: TransferSelection,
839 controls: TransferControls,
840 formatter: &Formatter,
841 alias_manager: &AliasManager,
842) -> ExitCode {
843 let candidates = match build_transfer_candidates(
844 &sources,
845 &target,
846 target_is_container,
847 args.recursive,
848 encryption,
849 args.source_customer_key.as_ref(),
850 alias_manager,
851 args.local_key_policy(),
852 )
853 .await
854 {
855 Ok(candidates) => candidates,
856 Err(error) => {
857 return formatter.fail(
858 exit_code_for_core_error(&error),
859 &format!("Failed to plan copy: {error}"),
860 );
861 }
862 };
863
864 let mut plan = TransferPlan::build(candidates, &selection);
865 if let Err(error) = validate_storage_class_plan(&plan, args.storage_class.as_deref()) {
866 return formatter.fail(exit_code_for_core_error(&error), &error.to_string());
867 }
868 if let Err(error) = validate_plan_targets(&plan) {
869 return formatter.fail(ExitCode::UsageError, &error.to_string());
870 }
871 if plan.items.is_empty() {
872 if args.fail_empty {
873 return formatter.fail(
874 ExitCode::NotFound,
875 "No copy sources matched the requested selection",
876 );
877 }
878 if args.summary || args.recursive || args.sources.len() > 1 {
879 print_transfer_summary(formatter, &plan.summary);
880 }
881 return ExitCode::Success;
882 }
883
884 let clients = match create_planned_client_cache(&plan.items, alias_manager).await {
885 Ok(clients) => Arc::new(clients),
886 Err(error) => {
887 return formatter.fail(
888 exit_code_for_core_error(&error),
889 &format!("Failed to prepare copy clients: {error}"),
890 );
891 }
892 };
893 let skipped_existing = if args.skip_existing || !args.overwrite {
894 match skip_existing_remote_targets(
895 &mut plan,
896 &clients,
897 args.destination_customer_key.as_ref(),
898 )
899 .await
900 {
901 Ok(skipped) => skipped,
902 Err(error) => {
903 return formatter.fail(
904 exit_code_for_core_error(&error),
905 &format!("Failed to inspect copy destinations: {error}"),
906 );
907 }
908 }
909 } else {
910 Vec::new()
911 };
912
913 if args.dry_run {
914 for item in &plan.items {
915 formatter.println(&format!(
916 "Would copy: {} -> {}{}",
917 formatter.style_file(&item.source),
918 formatter.style_file(&item.target),
919 transfer_policy_suffix(args)
920 ));
921 }
922 for item in &skipped_existing {
923 print_skipped_existing(formatter, item, true);
924 }
925 if args.summary || args.recursive || args.sources.len() > 1 {
926 print_transfer_summary(formatter, &plan.summary);
927 }
928 return ExitCode::Success;
929 }
930
931 for item in &skipped_existing {
932 print_skipped_existing(formatter, item, false);
933 }
934 if plan.items.is_empty() {
935 if args.summary || args.recursive || args.sources.len() > 1 {
936 print_transfer_summary(formatter, &plan.summary);
937 }
938 return ExitCode::Success;
939 }
940
941 let total_bytes = plan
942 .items
943 .iter()
944 .filter_map(|item| {
945 if matches!(item.payload, CpOperation::RemoteToRemote { .. }) {
946 item.size_bytes
947 } else {
948 None
949 }
950 })
951 .sum::<u64>();
952 let executor = match TransferExecutor::new(controls) {
953 Ok(executor) => executor,
954 Err(error) => {
955 return formatter.fail(ExitCode::UsageError, &error.to_string());
956 }
957 };
958 let operation_args = Arc::new(args.clone());
959 let transfer_cancellation = TransferCancellation::new();
960 let multipart_cancellation = MultipartCopyCancellation::new();
961 let signal_task = tokio::spawn({
962 let transfer_cancellation = transfer_cancellation.clone();
963 let multipart_cancellation = multipart_cancellation.clone();
964 async move {
965 if tokio::signal::ctrl_c().await.is_ok() {
966 multipart_cancellation.cancel();
967 transfer_cancellation.cancel();
968 }
969 }
970 });
971 let copy_details = PlannedCopyDetails::default();
972 let copy_progress = Arc::new(PlannedCopyProgress::new(
973 formatter.output_config(),
974 total_bytes,
975 ));
976 let report = executor
977 .execute_with_cancellation(plan, transfer_cancellation, {
978 let operation_args = Arc::clone(&operation_args);
979 let clients = Arc::clone(&clients);
980 let multipart_cancellation = multipart_cancellation.clone();
981 let copy_details = Arc::clone(©_details);
982 let copy_progress = Arc::clone(©_progress);
983 move |item| {
984 let operation_args = Arc::clone(&operation_args);
985 let clients = Arc::clone(&clients);
986 let multipart_cancellation = multipart_cancellation.clone();
987 let copy_details = Arc::clone(©_details);
988 let copy_progress = Arc::clone(©_progress);
989 async move {
990 execute_planned_operation(
991 item,
992 &operation_args,
993 &clients,
994 &multipart_cancellation,
995 ©_details,
996 ©_progress,
997 )
998 .await
999 }
1000 }
1001 })
1002 .await;
1003 signal_task.abort();
1004 let _ = signal_task.await;
1005 copy_progress.finish();
1006
1007 let copy_details = copy_details.lock().await;
1008 for outcome in &report.outcomes {
1009 match &outcome.state {
1010 TransferOutcomeState::Success { bytes_transferred } => {
1011 let key = (outcome.item.source.clone(), outcome.item.target.clone());
1012 print_planned_success(
1013 formatter,
1014 &outcome.item,
1015 *bytes_transferred,
1016 copy_details.get(&key),
1017 );
1018 }
1019 TransferOutcomeState::Failed { error } => {
1020 formatter.error_with_code(exit_code_for_core_error(error), &error.to_string());
1021 }
1022 TransferOutcomeState::Cancelled { error } => {
1023 if let Some(error) = error {
1024 formatter.warning(&format!(
1025 "Cancelled transfer: {} ({error})",
1026 outcome.item.source
1027 ));
1028 } else {
1029 formatter.warning(&format!(
1030 "Cancelled before transfer: {}",
1031 outcome.item.source
1032 ));
1033 }
1034 }
1035 }
1036 }
1037
1038 if args.summary || args.recursive || args.sources.len() > 1 {
1039 print_transfer_summary(formatter, &report.summary);
1040 }
1041
1042 if report.was_cancelled {
1043 ExitCode::Interrupted
1044 } else {
1045 report
1046 .first_failure()
1047 .map_or(ExitCode::Success, exit_code_for_core_error)
1048 }
1049}
1050
1051fn transfer_policy_suffix(args: &CpArgs) -> String {
1052 let mut policies = Vec::new();
1053 if let Some(value) = args.storage_class.as_deref() {
1054 policies.push(format!("storage-class={value}"));
1055 }
1056 if let Some(directive) = requested_metadata_directive(args) {
1057 policies.push(format!(
1058 "metadata={}",
1059 match directive {
1060 MetadataDirective::Copy => "copy",
1061 MetadataDirective::Replace => "replace",
1062 }
1063 ));
1064 } else if args.fidelity.has_attribute_policy() || args.content_type.is_some() {
1065 policies.push(format!("metadata=write({})", args.fidelity.metadata.len()));
1066 }
1067 if let Some(directive) = args.tagging_directive {
1068 policies.push(format!(
1069 "tags={}({})",
1070 match directive {
1071 TaggingDirectiveArg::Copy => "copy",
1072 TaggingDirectiveArg::Replace => "replace",
1073 },
1074 args.fidelity.tags.len()
1075 ));
1076 } else if args.fidelity.has_tag_policy() {
1077 policies.push(format!("tags=write({})", args.fidelity.tags.len()));
1078 }
1079 if args.fidelity.checksum.is_some() {
1080 policies.push("checksum=sha256".to_string());
1081 }
1082 if let Some(mode) = args.fidelity.retention_mode.as_deref() {
1083 policies.push(format!("retention={}", mode.to_ascii_lowercase()));
1084 }
1085 if let Some(state) = args.fidelity.legal_hold.as_deref() {
1086 policies.push(format!("legal-hold={}", state.to_ascii_lowercase()));
1087 }
1088 if args.enc_c_destination_key_file.is_some() || args.enc_c_destination_key_env.is_some() {
1089 policies.push("encryption=sse-c".to_string());
1090 } else if !args.enc_kms.is_empty() {
1091 policies.push("encryption=sse-kms".to_string());
1092 } else if !args.enc_s3.is_empty() {
1093 policies.push("encryption=sse-s3".to_string());
1094 }
1095 if policies.is_empty() {
1096 String::new()
1097 } else {
1098 format!(" [policy:{}]", policies.join(","))
1099 }
1100}
1101
1102fn validate_storage_class_plan(
1103 plan: &TransferPlan<CpOperation>,
1104 storage_class: Option<&str>,
1105) -> rc_core::Result<()> {
1106 if storage_class.is_none() {
1107 return Ok(());
1108 }
1109 for item in &plan.items {
1110 let size = item.size_bytes.ok_or_else(|| {
1111 Error::UnsupportedFeature(format!(
1112 "Cannot guarantee storage class for a transfer with unknown size: {}",
1113 item.source
1114 ))
1115 })?;
1116 let multipart = match &item.payload {
1117 CpOperation::LocalToRemote { .. } => size > MULTIPART_THRESHOLD,
1118 CpOperation::RemoteToRemote { source, target, .. } if source.alias != target.alias => {
1119 size > MULTIPART_THRESHOLD
1120 }
1121 CpOperation::RemoteToRemote { .. } => rc_core::requires_multipart_copy(size),
1122 CpOperation::RemoteToLocal { .. } => false,
1123 };
1124 if multipart {
1125 return Err(Error::UnsupportedFeature(format!(
1126 "RustFS beta.10 does not persist storage class for multipart transfer: {}",
1127 item.source
1128 )));
1129 }
1130 }
1131 Ok(())
1132}
1133
1134async fn execute_planned_operation(
1135 item: TransferCandidate<CpOperation>,
1136 args: &CpArgs,
1137 clients: &HashMap<String, Arc<S3Client>>,
1138 multipart_cancellation: &MultipartCopyCancellation,
1139 copy_details: &PlannedCopyDetails,
1140 copy_progress: &SharedPlannedCopyProgress,
1141) -> rc_core::Result<u64> {
1142 let client = planned_client(clients, operation_alias(&item.payload))?;
1143 match &item.payload {
1144 CpOperation::LocalToRemote {
1145 source,
1146 target,
1147 encryption,
1148 } => perform_planned_upload(client, source, target, encryption.as_ref(), args).await,
1149 CpOperation::RemoteToLocal { source, target } => {
1150 perform_planned_download(client, source, target, args).await
1151 }
1152 CpOperation::RemoteToRemote {
1153 source,
1154 target,
1155 source_info,
1156 encryption,
1157 } => {
1158 let source_client = planned_client(clients, &source.alias)?;
1159 let target_client = planned_client(clients, &target.alias)?;
1160 let progress_key = (item.source.clone(), item.target.clone());
1161 copy_progress.reset(&progress_key);
1162 let result = perform_planned_remote_copy(
1163 source_client,
1164 target_client,
1165 source,
1166 target,
1167 source_info,
1168 encryption.as_ref(),
1169 multipart_cancellation,
1170 &|bytes| copy_progress.set(&progress_key, bytes),
1171 args,
1172 )
1173 .await?;
1174 copy_progress.set(&progress_key, result.bytes_copied);
1175 copy_details.lock().await.insert(
1176 (item.source, item.target),
1177 PlannedCopyDetail {
1178 source_version_id: result.source_version_id,
1179 destination_version_id: result.destination_version_id,
1180 upload_id: result.upload_id,
1181 },
1182 );
1183 Ok(result.bytes_copied)
1184 }
1185 }
1186}
1187
1188async fn perform_planned_upload(
1189 client: &S3Client,
1190 source: &Path,
1191 target: &RemotePath,
1192 encryption: Option<&ObjectEncryptionRequest>,
1193 args: &CpArgs,
1194) -> rc_core::Result<u64> {
1195 let metadata = tokio::fs::metadata(source).await?;
1196 let file_size = metadata.len();
1197 let guessed_type = mime_guess::from_path(source)
1198 .first()
1199 .map(|mime| mime.essence_str().to_string());
1200 let content_type = select_upload_content_type(
1201 args.content_type.as_deref(),
1202 guessed_type.as_deref(),
1203 file_size,
1204 );
1205 let options = object_write_options(
1206 &args.fidelity,
1207 content_type,
1208 encryption,
1209 args.destination_customer_key.as_ref(),
1210 args.storage_class.clone(),
1211 )?;
1212 let info = client
1213 .put_object_from_path_with_options(target, source, &options, |_| {})
1214 .await?;
1215 Ok(info
1216 .size_bytes
1217 .and_then(|size| u64::try_from(size).ok())
1218 .unwrap_or(file_size))
1219}
1220
1221async fn perform_planned_download(
1222 client: &S3Client,
1223 source: &RemotePath,
1224 target: &Path,
1225 args: &CpArgs,
1226) -> rc_core::Result<u64> {
1227 if target.exists() && !args.overwrite {
1228 return Err(Error::Conflict(format!(
1229 "Destination exists: {}. Use --overwrite to replace.",
1230 target.display()
1231 )));
1232 }
1233 if let Some(parent) = target.parent() {
1234 tokio::fs::create_dir_all(parent).await?;
1235 }
1236 client
1237 .download_object_to_path_with_transfer_options(
1238 source,
1239 target,
1240 &TransferReadOptions {
1241 customer_key: args.source_customer_key.clone(),
1242 ..TransferReadOptions::default()
1243 },
1244 |_, _| {},
1245 )
1246 .await
1247}
1248
1249struct PlannedRemoteCopyResult {
1250 bytes_copied: u64,
1251 source_version_id: Option<String>,
1252 source_etag: Option<String>,
1253 destination_version_id: Option<String>,
1254 upload_id: Option<String>,
1255 object: ObjectInfo,
1256}
1257
1258#[allow(clippy::too_many_arguments)]
1259async fn perform_planned_remote_copy(
1260 source_client: &S3Client,
1261 target_client: &S3Client,
1262 source: &RemotePath,
1263 target: &RemotePath,
1264 source_info: &ObjectInfo,
1265 encryption: Option<&ObjectEncryptionRequest>,
1266 cancellation: &MultipartCopyCancellation,
1267 on_progress: &(dyn Fn(u64) + Send + Sync),
1268 args: &CpArgs,
1269) -> rc_core::Result<PlannedRemoteCopyResult> {
1270 if source.alias != target.alias {
1271 return perform_cross_alias_remote_copy(
1272 source_client,
1273 target_client,
1274 source,
1275 target,
1276 source_info,
1277 encryption,
1278 on_progress,
1279 args,
1280 )
1281 .await;
1282 }
1283 if args.source_customer_key.is_some() || args.destination_customer_key.is_some() {
1284 return Err(Error::UnsupportedFeature(
1285 "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467"
1286 .to_string(),
1287 ));
1288 }
1289 let planned_size = source_info
1290 .size_bytes
1291 .and_then(|size| u64::try_from(size).ok())
1292 .ok_or_else(|| Error::InvalidPath(format!("Source size is unavailable: {source}")))?;
1293 if rc_core::requires_multipart_copy(planned_size) {
1294 if args.storage_class.is_some() {
1295 return Err(Error::UnsupportedFeature(
1296 "RustFS beta.10 does not persist storage class for multipart copies".to_string(),
1297 ));
1298 }
1299 let current = source_client.head_object(source).await?;
1300 if !source_identity_matches(source_info, ¤t) {
1301 return Err(Error::Conflict(format!(
1302 "Source changed after copy planning: {source}"
1303 )));
1304 }
1305 let options = multipart_options_from_source(¤t)?;
1306 let transfer = transfer_copy_options(args, current.version_id.clone(), encryption)?;
1307 let copied = source_client
1308 .multipart_copy_with_transfer_options(
1309 source,
1310 target,
1311 &options,
1312 &transfer,
1313 cancellation,
1314 on_progress,
1315 )
1316 .await?;
1317 return Ok(PlannedRemoteCopyResult {
1318 bytes_copied: copied.bytes_copied,
1319 source_version_id: options.source_version_id,
1320 source_etag: current.etag.clone(),
1321 destination_version_id: copied.object.version_id.clone(),
1322 upload_id: Some(copied.upload_id),
1323 object: copied.object,
1324 });
1325 }
1326 let options = transfer_copy_options(args, source_info.version_id.clone(), encryption)?;
1327 let copied = source_client
1328 .copy_object_with_transfer_options(source, target, &options)
1329 .await?;
1330 let bytes_copied = copied
1331 .size_bytes
1332 .and_then(|size| u64::try_from(size).ok())
1333 .unwrap_or(planned_size);
1334 on_progress(bytes_copied);
1335 Ok(PlannedRemoteCopyResult {
1336 bytes_copied,
1337 source_version_id: copied
1338 .source_version_id
1339 .clone()
1340 .or_else(|| source_info.version_id.clone()),
1341 source_etag: source_info.etag.clone(),
1342 destination_version_id: copied.version_id.clone(),
1343 upload_id: None,
1344 object: copied,
1345 })
1346}
1347
1348#[allow(clippy::too_many_arguments)]
1349async fn perform_cross_alias_remote_copy(
1350 source_client: &S3Client,
1351 target_client: &S3Client,
1352 source: &RemotePath,
1353 target: &RemotePath,
1354 source_info: &ObjectInfo,
1355 encryption: Option<&ObjectEncryptionRequest>,
1356 on_progress: &(dyn Fn(u64) + Send + Sync),
1357 args: &CpArgs,
1358) -> rc_core::Result<PlannedRemoteCopyResult> {
1359 if args.source_customer_key.is_some() || args.destination_customer_key.is_some() {
1363 return Err(Error::UnsupportedFeature(
1364 "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467"
1365 .to_string(),
1366 ));
1367 }
1368 let planned_size = source_info
1369 .size_bytes
1370 .and_then(|size| u64::try_from(size).ok())
1371 .ok_or_else(|| Error::InvalidPath(format!("Source size is unavailable: {source}")))?;
1372 if args.storage_class.is_some() && planned_size > MULTIPART_THRESHOLD {
1373 return Err(Error::UnsupportedFeature(
1374 "RustFS beta.10 does not persist storage class for multipart uploads".to_string(),
1375 ));
1376 }
1377
1378 let current = source_client.head_object(source).await?;
1379 if !source_identity_matches(source_info, ¤t) {
1380 return Err(Error::Conflict(format!(
1381 "Source changed after copy planning: {source}"
1382 )));
1383 }
1384
1385 let staging = tempfile::Builder::new()
1386 .prefix("rc-cp-cross-alias-")
1387 .suffix(".part")
1388 .tempfile()?
1389 .into_temp_path();
1390 async {
1391 let downloaded = source_client
1392 .download_object_to_path_with_transfer_options(
1393 source,
1394 &staging,
1395 &TransferReadOptions {
1396 version_id: current.version_id.clone(),
1397 customer_key: args.source_customer_key.clone(),
1398 ..TransferReadOptions::default()
1399 },
1400 |copied, _total| on_progress(copied),
1401 )
1402 .await?;
1403 if downloaded != planned_size {
1404 return Err(Error::Conflict(format!(
1405 "Source changed after copy planning: {source}"
1406 )));
1407 }
1408 let after = source_client
1409 .head_object_with_transfer_options(
1410 source,
1411 &TransferReadOptions {
1412 version_id: current.version_id.clone(),
1413 customer_key: args.source_customer_key.clone(),
1414 ..TransferReadOptions::default()
1415 },
1416 )
1417 .await?;
1418 if !source_identity_matches(¤t, &after) {
1419 return Err(Error::Conflict(format!(
1420 "Source changed after copy planning: {source}"
1421 )));
1422 }
1423 let options = piped_copy_write_options(args, ¤t, encryption)?;
1424 let object = target_client
1425 .put_object_from_path_with_options(target, &staging, &options, |copied| {
1426 on_progress(copied)
1427 })
1428 .await?;
1429 let bytes_copied = object
1430 .size_bytes
1431 .and_then(|size| u64::try_from(size).ok())
1432 .unwrap_or(downloaded);
1433 on_progress(bytes_copied);
1434 Ok(PlannedRemoteCopyResult {
1435 bytes_copied,
1436 source_version_id: current.version_id.clone(),
1437 source_etag: current.etag.clone(),
1438 destination_version_id: object.version_id.clone(),
1439 upload_id: None,
1440 object,
1441 })
1442 }
1443 .await
1444}
1445
1446fn source_identity_matches(planned: &ObjectInfo, current: &ObjectInfo) -> bool {
1447 if planned.size_bytes != current.size_bytes {
1448 return false;
1449 }
1450 match (&planned.etag, ¤t.etag) {
1451 (Some(planned), Some(current)) => planned == current,
1452 (None, None) => {
1455 planned.version_id.is_some()
1456 && planned.version_id.as_ref() == current.version_id.as_ref()
1457 }
1458 _ => false,
1459 }
1460}
1461
1462#[derive(Debug, Clone)]
1468pub(super) struct CrossAliasCopyResult {
1469 pub(super) object: ObjectInfo,
1470 pub(super) source_version_id: Option<String>,
1471 pub(super) source_etag: Option<String>,
1472}
1473
1474pub(super) async fn copy_object_across_aliases(
1475 source_client: &S3Client,
1476 target_client: &S3Client,
1477 source: &RemotePath,
1478 target: &RemotePath,
1479 encryption: Option<&ObjectEncryptionRequest>,
1480) -> rc_core::Result<CrossAliasCopyResult> {
1481 let source_info = source_client.head_object(source).await?;
1482 let args = CpArgs::single(source.to_string(), target.to_string());
1483 let ignore_progress = |_: u64| {};
1484 let result = perform_cross_alias_remote_copy(
1485 source_client,
1486 target_client,
1487 source,
1488 target,
1489 &source_info,
1490 encryption,
1491 &ignore_progress,
1492 &args,
1493 )
1494 .await?;
1495 Ok(CrossAliasCopyResult {
1496 object: result.object,
1497 source_version_id: result.source_version_id,
1498 source_etag: result.source_etag,
1499 })
1500}
1501
1502fn piped_copy_write_options(
1503 args: &CpArgs,
1504 source: &ObjectInfo,
1505 encryption: Option<&ObjectEncryptionRequest>,
1506) -> rc_core::Result<ObjectWriteOptions> {
1507 let mut options = object_write_options(
1508 &args.fidelity,
1509 args.content_type.as_deref(),
1510 encryption,
1511 args.destination_customer_key.as_ref(),
1512 args.storage_class.clone(),
1513 )?;
1514 let replace_metadata = matches!(
1515 requested_metadata_directive(args),
1516 Some(MetadataDirective::Replace)
1517 );
1518 let mut attributes = options.attributes.take().unwrap_or_default();
1519 if !replace_metadata {
1520 if attributes.content_type.is_none() {
1521 attributes.content_type = source.content_type.clone();
1522 }
1523 if attributes.user_metadata.is_empty()
1524 && let Some(metadata) = &source.metadata
1525 {
1526 attributes.user_metadata.clone_from(metadata);
1527 }
1528 }
1529 if let Some(source_etag) = source.etag.as_deref() {
1534 set_source_identity(&mut attributes, source_etag);
1535 }
1536 if attributes != ObjectAttributes::default() {
1537 options.attributes = Some(attributes);
1538 }
1539 Ok(options)
1540}
1541
1542#[cfg(test)]
1543fn requires_multipart_copy(planned_size: Option<u64>) -> bool {
1544 planned_size.is_some_and(rc_core::requires_multipart_copy)
1545}
1546
1547fn multipart_options_from_source(source: &ObjectInfo) -> rc_core::Result<MultipartCopyOptions> {
1548 let source_size = source
1549 .size_bytes
1550 .and_then(|size| u64::try_from(size).ok())
1551 .ok_or_else(|| {
1552 Error::InvalidPath("Multipart copy source size is unavailable".to_string())
1553 })?;
1554 let source_etag = source.etag.clone().ok_or_else(|| {
1555 Error::InvalidPath("Multipart copy source ETag is unavailable".to_string())
1556 })?;
1557 let mut options = MultipartCopyOptions::new(source_size, source_etag)?;
1558 options.source_version_id = source.version_id.clone();
1559 options.content_type = source.content_type.clone();
1560 let mut attributes = ObjectAttributes {
1561 user_metadata: source.metadata.clone().unwrap_or_default(),
1562 ..ObjectAttributes::default()
1563 };
1564 if let Some(source_etag) = source.etag.as_deref() {
1565 set_source_identity(&mut attributes, source_etag);
1566 }
1567 options.metadata = attributes.user_metadata;
1568 Ok(options)
1569}
1570
1571fn operation_alias(operation: &CpOperation) -> &str {
1572 match operation {
1573 CpOperation::LocalToRemote { target, .. } => &target.alias,
1574 CpOperation::RemoteToLocal { source, .. } | CpOperation::RemoteToRemote { source, .. } => {
1575 &source.alias
1576 }
1577 }
1578}
1579
1580fn operation_client_aliases(operation: &CpOperation) -> Vec<&str> {
1581 match operation {
1582 CpOperation::LocalToRemote { target, .. } => vec![target.alias.as_str()],
1583 CpOperation::RemoteToLocal { source, .. } => vec![source.alias.as_str()],
1584 CpOperation::RemoteToRemote { source, target, .. } => {
1585 if source.alias == target.alias {
1586 vec![source.alias.as_str()]
1587 } else {
1588 vec![source.alias.as_str(), target.alias.as_str()]
1589 }
1590 }
1591 }
1592}
1593
1594fn planned_client_aliases(items: &[TransferCandidate<CpOperation>]) -> BTreeSet<String> {
1595 items
1596 .iter()
1597 .flat_map(|item| {
1598 operation_client_aliases(&item.payload)
1599 .into_iter()
1600 .map(ToOwned::to_owned)
1601 })
1602 .collect()
1603}
1604
1605async fn create_planned_client_cache(
1606 items: &[TransferCandidate<CpOperation>],
1607 alias_manager: &AliasManager,
1608) -> rc_core::Result<HashMap<String, Arc<S3Client>>> {
1609 let mut clients = HashMap::new();
1610 for alias_name in planned_client_aliases(items) {
1611 match create_leaf_s3_client(alias_manager, &alias_name).await {
1612 Ok(client) => {
1613 clients.insert(alias_name, Arc::new(client));
1614 }
1615 Err(Error::AliasNotFound(_)) => {}
1618 Err(error) => return Err(error),
1619 }
1620 }
1621 Ok(clients)
1622}
1623
1624fn planned_client<'a>(
1625 clients: &'a HashMap<String, Arc<S3Client>>,
1626 alias_name: &str,
1627) -> rc_core::Result<&'a S3Client> {
1628 clients
1629 .get(alias_name)
1630 .map(Arc::as_ref)
1631 .ok_or_else(|| Error::AliasNotFound(alias_name.to_string()))
1632}
1633
1634async fn create_leaf_s3_client(
1635 alias_manager: &AliasManager,
1636 alias_name: &str,
1637) -> rc_core::Result<S3Client> {
1638 let mut alias = alias_manager
1639 .get(alias_name)
1640 .map_err(|_| Error::AliasNotFound(alias_name.to_string()))?;
1641 alias.retry = Some(RetryConfig {
1644 max_attempts: 1,
1645 initial_backoff_ms: 1,
1646 max_backoff_ms: 1,
1647 });
1648 S3Client::new(alias).await
1649}
1650
1651fn print_planned_success(
1652 formatter: &Formatter,
1653 item: &TransferCandidate<CpOperation>,
1654 bytes_transferred: u64,
1655 detail: Option<&PlannedCopyDetail>,
1656) {
1657 let size_bytes = i64::try_from(bytes_transferred).ok();
1658 let size_human = size_bytes.map(|size| humansize::format_size(size as u64, humansize::BINARY));
1659 if formatter.is_json() {
1660 formatter.json(&CpOutput {
1661 status: "success",
1662 source: item.source.clone(),
1663 target: item.target.clone(),
1664 size_bytes,
1665 size_human,
1666 version_id: None,
1667 source_version_id: None,
1668 });
1669 } else {
1670 let mut line = format!(
1671 "{} -> {} ({})",
1672 formatter.style_file(&item.source),
1673 formatter.style_file(&item.target),
1674 formatter.style_size(size_human.as_deref().unwrap_or_default())
1675 );
1676 if let Some(detail) = detail {
1677 if let Some(version_id) = &detail.source_version_id {
1678 line.push_str(&format!(" source-version={version_id}"));
1679 }
1680 if let Some(version_id) = &detail.destination_version_id {
1681 line.push_str(&format!(" destination-version={version_id}"));
1682 }
1683 if let Some(upload_id) = &detail.upload_id {
1684 line.push_str(&format!(" upload-id={upload_id}"));
1685 }
1686 }
1687 formatter.println(&line);
1688 }
1689}
1690
1691fn print_skipped_existing(
1692 formatter: &Formatter,
1693 item: &TransferCandidate<CpOperation>,
1694 dry_run: bool,
1695) {
1696 let action = if dry_run {
1697 "Would skip existing"
1698 } else {
1699 "Skipped existing"
1700 };
1701 formatter.println(&format!(
1702 "{action}: {} -> {}",
1703 formatter.style_file(&item.source),
1704 formatter.style_file(&item.target)
1705 ));
1706}
1707
1708fn print_transfer_summary(formatter: &Formatter, summary: &rc_core::TransferSummary) {
1709 formatter.println(&format!(
1710 "Summary: {} planned, {} skipped, {} succeeded, {} failed, {} cancelled, {} transferred",
1711 summary.planned,
1712 summary.skipped,
1713 summary.successful,
1714 summary.failed,
1715 summary.cancelled,
1716 humansize::format_size(summary.transferred_bytes, humansize::BINARY)
1717 ));
1718}
1719
1720pub(super) fn exit_code_for_core_error(error: &Error) -> ExitCode {
1721 ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError)
1722}
1723
1724fn build_transfer_selection(args: &CpArgs, now: Timestamp) -> Result<TransferSelection, String> {
1725 let newer_than = args
1726 .newer_than
1727 .as_deref()
1728 .map(|value| parse_age_cutoff(value, now))
1729 .transpose()?;
1730 let older_than = args
1731 .older_than
1732 .as_deref()
1733 .map(|value| parse_age_cutoff(value, now))
1734 .transpose()?;
1735 let rewind = args
1736 .rewind
1737 .as_deref()
1738 .map(|value| parse_rewind_cutoff(value, now))
1739 .transpose()?;
1740
1741 TransferSelection::new(&args.include, &args.exclude, newer_than, older_than, rewind)
1742 .map_err(|error| error.to_string())
1743}
1744
1745fn build_transfer_controls(args: &CpArgs) -> Result<TransferControls, String> {
1746 let bytes_per_second = args
1747 .rate_limit
1748 .as_deref()
1749 .map(parse_byte_rate)
1750 .transpose()?;
1751 let controls = TransferControls {
1752 concurrency: args.concurrency.unwrap_or(DEFAULT_TRANSFER_CONCURRENCY),
1753 bytes_per_second,
1754 retry: RetryConfig {
1755 max_attempts: args.retry_attempts.unwrap_or(DEFAULT_RETRY_ATTEMPTS),
1756 initial_backoff_ms: args
1757 .retry_initial_backoff_ms
1758 .unwrap_or(DEFAULT_RETRY_INITIAL_BACKOFF_MS),
1759 max_backoff_ms: args
1760 .retry_max_backoff_ms
1761 .unwrap_or(DEFAULT_RETRY_MAX_BACKOFF_MS),
1762 },
1763 continue_on_error: args.continue_on_error,
1764 };
1765 controls.validate().map_err(|error| error.to_string())?;
1766 Ok(controls)
1767}
1768
1769pub(super) fn parse_age_cutoff(value: &str, now: Timestamp) -> Result<Timestamp, String> {
1770 let value = value.trim();
1771 if value.is_empty() {
1772 return Err("Transfer age must not be empty".to_string());
1773 }
1774 let suffix_index = value
1775 .find(|character: char| character.is_ascii_alphabetic())
1776 .unwrap_or(value.len());
1777 let number = &value[..suffix_index];
1778 let suffix = &value[suffix_index..];
1779 let amount: i64 = number
1780 .parse()
1781 .map_err(|_| format!("Invalid transfer age number: {number}"))?;
1782 if amount < 0 {
1783 return Err("Transfer age must not be negative".to_string());
1784 }
1785 let multiplier = match suffix.to_ascii_lowercase().as_str() {
1786 "" | "s" => 1,
1787 "m" => 60,
1788 "h" => 3_600,
1789 "d" => 86_400,
1790 "w" => 604_800,
1791 _ => return Err(format!("Unknown transfer age suffix: {suffix}")),
1792 };
1793 let seconds = amount
1794 .checked_mul(multiplier)
1795 .ok_or_else(|| "Transfer age is too large".to_string())?;
1796 now.checked_sub(jiff::Span::new().seconds(seconds))
1797 .map_err(|error| format!("Transfer age overflow: {error}"))
1798}
1799
1800fn parse_rewind_cutoff(value: &str, now: Timestamp) -> Result<Timestamp, String> {
1801 value
1802 .parse::<Timestamp>()
1803 .or_else(|_| parse_age_cutoff(value, now))
1804 .map_err(|error| format!("Invalid rewind value '{value}': {error}"))
1805}
1806
1807pub(super) fn parse_byte_rate(value: &str) -> Result<u64, String> {
1808 let normalized = value.trim().to_ascii_lowercase();
1809 let normalized = normalized
1810 .strip_suffix("/s")
1811 .or_else(|| normalized.strip_suffix("ps"))
1812 .unwrap_or(&normalized);
1813 let suffix_index = normalized
1814 .find(|character: char| character.is_ascii_alphabetic())
1815 .unwrap_or(normalized.len());
1816 let number = &normalized[..suffix_index];
1817 let suffix = &normalized[suffix_index..];
1818 let amount: u64 = number
1819 .parse()
1820 .map_err(|_| format!("Invalid transfer rate number: {number}"))?;
1821 let multiplier = match suffix {
1822 "" | "b" => 1u64,
1823 "k" | "kb" => 1_000,
1824 "ki" | "kib" => 1_024,
1825 "m" | "mb" => 1_000_000,
1826 "mi" | "mib" => 1_048_576,
1827 "g" | "gb" => 1_000_000_000,
1828 "gi" | "gib" => 1_073_741_824,
1829 _ => return Err(format!("Unknown transfer rate suffix: {suffix}")),
1830 };
1831 let rate = amount
1832 .checked_mul(multiplier)
1833 .ok_or_else(|| "Transfer rate is too large".to_string())?;
1834 if rate == 0 {
1835 return Err("Transfer rate must be greater than zero".to_string());
1836 }
1837 Ok(rate)
1838}
1839
1840fn is_container_target(raw: &str, target: &ParsedPath) -> bool {
1841 match target {
1842 ParsedPath::Local(path) => path.is_dir() || raw.ends_with(['/', '\\']),
1843 ParsedPath::Remote(path) => path.key.is_empty() || raw.ends_with('/'),
1844 }
1845}
1846
1847#[allow(clippy::too_many_arguments)]
1848async fn build_transfer_candidates(
1849 sources: &[ParsedPath],
1850 target: &ParsedPath,
1851 target_is_container: bool,
1852 recursive: bool,
1853 encryption: Option<ObjectEncryptionRequest>,
1854 source_customer_key: Option<&SseCustomerKey>,
1855 alias_manager: &AliasManager,
1856 key_policy: ObjectKeyPolicy,
1857) -> rc_core::Result<Vec<TransferCandidate<CpOperation>>> {
1858 let mut candidates = Vec::new();
1859 let mut planning_clients = HashMap::new();
1860 let multiple_sources = sources.len() > 1;
1861
1862 for source in sources {
1863 match source {
1864 ParsedPath::Local(source) => build_local_candidates(
1865 source,
1866 target,
1867 target_is_container,
1868 recursive,
1869 multiple_sources,
1870 encryption.clone(),
1871 &mut candidates,
1872 )?,
1873 ParsedPath::Remote(source) => {
1874 build_remote_candidates(
1875 source,
1876 target,
1877 target_is_container,
1878 recursive,
1879 multiple_sources,
1880 encryption.clone(),
1881 source_customer_key,
1882 alias_manager,
1883 &mut planning_clients,
1884 &mut candidates,
1885 key_policy,
1886 )
1887 .await?;
1888 }
1889 }
1890 }
1891
1892 Ok(candidates)
1893}
1894
1895fn validate_plan_targets(plan: &TransferPlan<CpOperation>) -> rc_core::Result<()> {
1896 let mut targets = HashSet::with_capacity(plan.items.len());
1897 for candidate in &plan.items {
1898 if let CpOperation::RemoteToRemote { source, target, .. } = &candidate.payload
1899 && source == target
1900 {
1901 return Err(Error::InvalidPath(format!(
1902 "Source and destination resolve to the same object '{}'",
1903 candidate.source
1904 )));
1905 }
1906 if !targets.insert(candidate.target.clone()) {
1907 return Err(Error::InvalidPath(format!(
1908 "Multiple sources resolve to the same destination '{}'",
1909 candidate.target
1910 )));
1911 }
1912 }
1913 Ok(())
1914}
1915
1916async fn skip_existing_remote_targets(
1917 plan: &mut TransferPlan<CpOperation>,
1918 clients: &HashMap<String, Arc<S3Client>>,
1919 destination_customer_key: Option<&SseCustomerKey>,
1920) -> rc_core::Result<Vec<TransferCandidate<CpOperation>>> {
1921 let mut retained = Vec::with_capacity(plan.items.len());
1922 let mut skipped = Vec::new();
1923 for candidate in plan.items.drain(..) {
1924 let destination = match &candidate.payload {
1925 CpOperation::LocalToRemote { target, .. }
1926 | CpOperation::RemoteToRemote { target, .. } => Some(target),
1927 CpOperation::RemoteToLocal { .. } => None,
1928 };
1929 let Some(destination) = destination else {
1930 retained.push(candidate);
1931 continue;
1932 };
1933 let client = planned_client(clients, &destination.alias)?;
1934 match client
1935 .head_object_with_transfer_options(
1936 destination,
1937 &TransferReadOptions {
1938 customer_key: destination_customer_key.cloned(),
1939 ..TransferReadOptions::default()
1940 },
1941 )
1942 .await
1943 {
1944 Ok(_) => skipped.push(candidate),
1945 Err(Error::NotFound(_)) | Err(Error::VersionNotFound { .. }) => {
1946 retained.push(candidate);
1947 }
1948 Err(error) => return Err(error),
1949 }
1950 }
1951 plan.items = retained;
1952 plan.summary.planned = plan.items.len();
1953 plan.summary.skipped = plan.summary.skipped.saturating_add(skipped.len());
1954 Ok(skipped)
1955}
1956
1957#[allow(clippy::too_many_arguments)]
1958fn build_local_candidates(
1959 source: &Path,
1960 target: &ParsedPath,
1961 target_is_container: bool,
1962 recursive: bool,
1963 multiple_sources: bool,
1964 encryption: Option<ObjectEncryptionRequest>,
1965 candidates: &mut Vec<TransferCandidate<CpOperation>>,
1966) -> rc_core::Result<()> {
1967 let ParsedPath::Remote(target) = target else {
1968 return Err(Error::InvalidPath(
1969 "Cannot copy between two local paths. Use the system cp command.".to_string(),
1970 ));
1971 };
1972 let metadata = std::fs::metadata(source).map_err(|error| {
1973 if error.kind() == std::io::ErrorKind::NotFound {
1974 Error::NotFound(format!("Source not found: {}", source.display()))
1975 } else {
1976 Error::Io(error)
1977 }
1978 })?;
1979
1980 if metadata.is_file() {
1981 let name = local_file_name(source)?;
1982 let destination = if target_is_container {
1983 remote_child(target, &name, ObjectKeyPolicy::for_remote_destination())?
1984 } else {
1985 normalize_remote_target(target, ObjectKeyPolicy::for_remote_destination())?
1986 };
1987 candidates.push(local_transfer_candidate(
1988 source.to_path_buf(),
1989 destination,
1990 name,
1991 metadata,
1992 encryption,
1993 ));
1994 return Ok(());
1995 }
1996
1997 if !recursive {
1998 return Err(Error::InvalidPath(
1999 "Source is a directory. Use -r/--recursive to copy directories.".to_string(),
2000 ));
2001 }
2002 if !target_is_container {
2003 return Err(Error::InvalidPath(
2004 "Recursive copy requires a directory or remote prefix destination ending in '/'"
2005 .to_string(),
2006 ));
2007 }
2008
2009 let mut files = Vec::new();
2010 collect_local_files(source, source, &mut files)?;
2011 files.sort_by(|left, right| left.1.cmp(&right.1));
2012 let source_root = multiple_sources
2013 .then(|| local_file_name(source))
2014 .transpose()?;
2015 for (path, relative, metadata) in files {
2016 let relative = relative.replace('\\', "/");
2017 let target_relative = source_root
2018 .as_deref()
2019 .map_or_else(|| relative.clone(), |root| format!("{root}/{relative}"));
2020 candidates.push(local_transfer_candidate(
2021 path,
2022 remote_child(
2023 target,
2024 &target_relative,
2025 ObjectKeyPolicy::for_remote_destination(),
2026 )?,
2027 relative,
2028 metadata,
2029 encryption.clone(),
2030 ));
2031 }
2032 Ok(())
2033}
2034
2035fn local_file_name(path: &Path) -> rc_core::Result<String> {
2036 path.file_name()
2037 .map(|name| name.to_string_lossy().to_string())
2038 .filter(|name| !name.is_empty())
2039 .ok_or_else(|| Error::InvalidPath(format!("Path has no file name: {}", path.display())))
2040}
2041
2042fn local_transfer_candidate(
2043 source: PathBuf,
2044 target: RemotePath,
2045 relative_path: String,
2046 metadata: std::fs::Metadata,
2047 encryption: Option<ObjectEncryptionRequest>,
2048) -> TransferCandidate<CpOperation> {
2049 TransferCandidate {
2050 payload: CpOperation::LocalToRemote {
2051 source: source.clone(),
2052 target: target.clone(),
2053 encryption,
2054 },
2055 source: source.display().to_string(),
2056 target: target.to_string(),
2057 relative_path,
2058 modified: metadata
2059 .modified()
2060 .ok()
2061 .and_then(|time| time.try_into().ok()),
2062 size_bytes: Some(metadata.len()),
2063 }
2064}
2065
2066fn collect_local_files(
2067 directory: &Path,
2068 base: &Path,
2069 files: &mut Vec<(PathBuf, String, std::fs::Metadata)>,
2070) -> rc_core::Result<()> {
2071 let mut entries = std::fs::read_dir(directory)?.collect::<std::io::Result<Vec<_>>>()?;
2072 entries.sort_by_key(std::fs::DirEntry::file_name);
2073 for entry in entries {
2074 let path = entry.path();
2075 let file_type = entry.file_type()?;
2076 if file_type.is_symlink() {
2077 return Err(Error::InvalidPath(format!(
2078 "Symbolic links are not supported in recursive copy: {}",
2079 path.display()
2080 )));
2081 }
2082 let metadata = entry.metadata()?;
2083 if file_type.is_file() {
2084 let relative = path
2085 .strip_prefix(base)
2086 .map_err(|error| Error::InvalidPath(error.to_string()))?
2087 .to_string_lossy()
2088 .to_string();
2089 files.push((path, relative, metadata));
2090 } else if file_type.is_dir() {
2091 collect_local_files(&path, base, files)?;
2092 }
2093 }
2094 Ok(())
2095}
2096
2097#[allow(clippy::too_many_arguments)]
2098async fn build_remote_candidates(
2099 source: &RemotePath,
2100 target: &ParsedPath,
2101 target_is_container: bool,
2102 recursive: bool,
2103 multiple_sources: bool,
2104 encryption: Option<ObjectEncryptionRequest>,
2105 source_customer_key: Option<&SseCustomerKey>,
2106 alias_manager: &AliasManager,
2107 planning_clients: &mut HashMap<String, Arc<S3Client>>,
2108 candidates: &mut Vec<TransferCandidate<CpOperation>>,
2109 key_policy: ObjectKeyPolicy,
2110) -> rc_core::Result<()> {
2111 let is_prefix = source.key.is_empty() || source.key.ends_with('/') || recursive;
2112 let client = planning_client(planning_clients, alias_manager, &source.alias).await?;
2113
2114 if is_prefix {
2115 if !recursive {
2116 return Err(Error::InvalidPath(
2117 "Remote prefix copy requires -r/--recursive".to_string(),
2118 ));
2119 }
2120 if !target_is_container {
2121 return Err(Error::InvalidPath(
2122 "Recursive copy requires a directory or remote prefix destination ending in '/'"
2123 .to_string(),
2124 ));
2125 }
2126
2127 let listing_source = recursive_listing_source(source);
2128 if let ParsedPath::Remote(target) = target
2129 && remote_copy_scopes_overlap(&listing_source, target)
2130 {
2131 return Err(Error::Conflict(format!(
2132 "Recursive source '{}' overlaps destination '{}'",
2133 listing_source, target
2134 )));
2135 }
2136
2137 let source_root = recursive_source_root(&listing_source, multiple_sources);
2138 let mut continuation_token = None;
2139 let mut seen_tokens = HashSet::new();
2140 loop {
2141 let result = client
2142 .list_objects(
2143 &listing_source,
2144 rc_core::ListOptions {
2145 recursive: true,
2146 max_keys: Some(1_000),
2147 continuation_token: continuation_token.clone(),
2148 ..Default::default()
2149 },
2150 )
2151 .await?;
2152 for object in result.items.into_iter().filter(|object| !object.is_dir) {
2153 let object_source =
2154 RemotePath::new(&listing_source.alias, &listing_source.bucket, &object.key);
2155 match target {
2156 ParsedPath::Local(target_root) => {
2157 let relative = safe_download_relative_path(
2158 &object.key,
2159 &listing_source.key,
2160 key_policy,
2161 )
2162 .map_err(Error::InvalidPath)?;
2163 let relative_string = relative.to_string_lossy().replace('\\', "/");
2164 let target_relative = if source_root.is_empty() {
2165 relative.clone()
2166 } else {
2167 PathBuf::from(&source_root).join(&relative)
2168 };
2169 let destination = safe_download_destination(target_root, &target_relative)
2170 .await
2171 .map_err(Error::InvalidPath)?;
2172 candidates.push(TransferCandidate {
2173 payload: CpOperation::RemoteToLocal {
2174 source: object_source.clone(),
2175 target: destination.clone(),
2176 },
2177 source: object_source.to_string(),
2178 target: destination.display().to_string(),
2179 relative_path: relative_string,
2180 modified: object.last_modified,
2181 size_bytes: object.size_bytes.and_then(|size| u64::try_from(size).ok()),
2182 });
2183 }
2184 ParsedPath::Remote(target) => {
2185 let (destination, relative) = recursive_remote_target(
2186 &listing_source,
2187 target,
2188 &object.key,
2189 multiple_sources,
2190 ObjectKeyPolicy::for_remote_destination(),
2191 )?;
2192 let size_bytes =
2193 object.size_bytes.and_then(|size| u64::try_from(size).ok());
2194 candidates.push(TransferCandidate {
2195 payload: CpOperation::RemoteToRemote {
2196 source: object_source.clone(),
2197 target: destination.clone(),
2198 source_info: Box::new(object.clone()),
2199 encryption: encryption.clone(),
2200 },
2201 source: object_source.to_string(),
2202 target: destination.to_string(),
2203 relative_path: relative,
2204 modified: object.last_modified,
2205 size_bytes,
2206 });
2207 }
2208 }
2209 }
2210 if !result.truncated {
2211 break;
2212 }
2213 let next_token = result.continuation_token.ok_or_else(|| {
2214 Error::InvalidPath(
2215 "Truncated object listing did not include a continuation token".to_string(),
2216 )
2217 })?;
2218 if !seen_tokens.insert(next_token.clone()) {
2219 return Err(Error::Conflict(
2220 "Object listing repeated a continuation token".to_string(),
2221 ));
2222 }
2223 continuation_token = Some(next_token);
2224 }
2225 return Ok(());
2226 }
2227
2228 let object = client
2229 .head_object_with_transfer_options(
2230 source,
2231 &TransferReadOptions {
2232 customer_key: source_customer_key.cloned(),
2233 ..TransferReadOptions::default()
2234 },
2235 )
2236 .await?;
2237 let name = source
2238 .key
2239 .rsplit('/')
2240 .next()
2241 .filter(|name| !name.is_empty())
2242 .ok_or_else(|| Error::InvalidPath(format!("Object key is empty: {source}")))?;
2243 let size_bytes = object.size_bytes.and_then(|size| u64::try_from(size).ok());
2244 let modified = object.last_modified;
2245
2246 match target {
2247 ParsedPath::Local(target) => {
2248 let destination = if target_is_container {
2249 let relative = safe_download_relative_path(name, "", key_policy)
2250 .map_err(Error::InvalidPath)?;
2251 safe_download_destination(target, &relative)
2252 .await
2253 .map_err(Error::InvalidPath)?
2254 } else {
2255 target.clone()
2256 };
2257 candidates.push(TransferCandidate {
2258 payload: CpOperation::RemoteToLocal {
2259 source: source.clone(),
2260 target: destination.clone(),
2261 },
2262 source: source.to_string(),
2263 target: destination.display().to_string(),
2264 relative_path: name.to_string(),
2265 modified,
2266 size_bytes,
2267 });
2268 }
2269 ParsedPath::Remote(target) => {
2270 let destination = if target_is_container {
2271 remote_child(target, name, ObjectKeyPolicy::for_remote_destination())?
2272 } else {
2273 normalize_remote_target(target, ObjectKeyPolicy::for_remote_destination())?
2274 };
2275 candidates.push(TransferCandidate {
2276 payload: CpOperation::RemoteToRemote {
2277 source: source.clone(),
2278 target: destination.clone(),
2279 source_info: Box::new(object),
2280 encryption,
2281 },
2282 source: source.to_string(),
2283 target: destination.to_string(),
2284 relative_path: name.to_string(),
2285 modified,
2286 size_bytes,
2287 });
2288 }
2289 }
2290 Ok(())
2291}
2292
2293async fn planning_client(
2294 clients: &mut HashMap<String, Arc<S3Client>>,
2295 alias_manager: &AliasManager,
2296 alias_name: &str,
2297) -> rc_core::Result<Arc<S3Client>> {
2298 if let Some(client) = clients.get(alias_name) {
2299 return Ok(Arc::clone(client));
2300 }
2301 let alias = alias_manager
2302 .get(alias_name)
2303 .map_err(|_| Error::AliasNotFound(alias_name.to_string()))?;
2304 let client = Arc::new(S3Client::new(alias).await?);
2305 clients.insert(alias_name.to_string(), Arc::clone(&client));
2306 Ok(client)
2307}
2308
2309fn remote_child(
2310 parent: &RemotePath,
2311 relative: &str,
2312 policy: ObjectKeyPolicy,
2313) -> rc_core::Result<RemotePath> {
2314 let parent_key = normalize_remote_prefix(&parent.key, policy)?;
2315 let relative = normalize_relative_key(relative, policy)?;
2316 let key = if parent_key.is_empty() {
2317 relative
2318 } else {
2319 format!("{parent_key}/{relative}")
2320 };
2321 Ok(RemotePath::new(&parent.alias, &parent.bucket, key))
2322}
2323
2324fn normalize_remote_target(
2325 target: &RemotePath,
2326 policy: ObjectKeyPolicy,
2327) -> rc_core::Result<RemotePath> {
2328 let key = normalize_remote_prefix(&target.key, policy)?;
2329 if target.key.ends_with('/') && !key.is_empty() {
2330 Ok(RemotePath::new(
2331 &target.alias,
2332 &target.bucket,
2333 format!("{key}/"),
2334 ))
2335 } else {
2336 Ok(RemotePath::new(&target.alias, &target.bucket, key))
2337 }
2338}
2339
2340fn normalize_remote_prefix(key: &str, policy: ObjectKeyPolicy) -> rc_core::Result<String> {
2341 if key.is_empty() {
2342 return Ok(String::new());
2343 }
2344 normalize_relative_key(key, policy)
2345}
2346
2347fn recursive_listing_source(source: &RemotePath) -> RemotePath {
2348 let key = if source.key.is_empty() || source.key.ends_with('/') {
2349 source.key.clone()
2350 } else {
2351 format!("{}/", source.key)
2352 };
2353 RemotePath::new(&source.alias, &source.bucket, key)
2354}
2355
2356fn recursive_source_root(source: &RemotePath, multiple_sources: bool) -> String {
2357 if !multiple_sources {
2358 return String::new();
2359 }
2360 source
2361 .key
2362 .trim_end_matches('/')
2363 .rsplit('/')
2364 .next()
2365 .filter(|value| !value.is_empty())
2366 .unwrap_or(&source.bucket)
2367 .to_string()
2368}
2369
2370fn recursive_remote_target(
2371 source: &RemotePath,
2372 target: &RemotePath,
2373 object_key: &str,
2374 multiple_sources: bool,
2375 policy: ObjectKeyPolicy,
2376) -> rc_core::Result<(RemotePath, String)> {
2377 let relative = object_key.strip_prefix(&source.key).ok_or_else(|| {
2378 Error::InvalidPath(format!(
2379 "Listed object key '{object_key}' is outside source prefix '{}'",
2380 source.key
2381 ))
2382 })?;
2383 if relative.is_empty() {
2384 return Err(Error::InvalidPath(format!(
2385 "Listed object key '{object_key}' does not identify a child of source prefix '{}'",
2386 source.key
2387 )));
2388 }
2389 let destination_relative = match recursive_source_root(source, multiple_sources) {
2390 root if root.is_empty() => relative.to_string(),
2391 root => format!("{root}/{relative}"),
2392 };
2393 let normalized_relative = normalize_relative_key(relative, policy)?;
2394 let normalized_destination_relative = normalize_relative_key(&destination_relative, policy)?;
2395 Ok((
2396 remote_child(target, &normalized_destination_relative, policy)?,
2397 normalized_relative,
2398 ))
2399}
2400
2401fn remote_copy_scopes_overlap(source: &RemotePath, target: &RemotePath) -> bool {
2402 if source.alias != target.alias || source.bucket != target.bucket {
2403 return false;
2404 }
2405 let source_prefix = recursive_listing_source(source).key;
2406 let target_prefix = recursive_listing_source(target).key;
2407 source_prefix.is_empty()
2408 || target_prefix.is_empty()
2409 || source_prefix.starts_with(&target_prefix)
2410 || target_prefix.starts_with(&source_prefix)
2411}
2412
2413fn parse_cp_path(path: &str, alias_manager: Option<&AliasManager>) -> rc_core::Result<ParsedPath> {
2414 let parsed = parse_path(path)?;
2415
2416 let ParsedPath::Remote(remote) = &parsed else {
2417 return Ok(parsed);
2418 };
2419
2420 if let Some(manager) = alias_manager
2421 && matches!(manager.exists(&remote.alias), Ok(true))
2422 {
2423 return Ok(parsed);
2424 }
2425
2426 if Path::new(path).exists() {
2427 return Ok(ParsedPath::Local(PathBuf::from(path)));
2428 }
2429
2430 Ok(parsed)
2431}
2432
2433async fn copy_local_to_s3_prepared(
2434 src: &Path,
2435 dst: &RemotePath,
2436 args: &CpArgs,
2437 formatter: &Formatter,
2438 encryption: Option<&ObjectEncryptionRequest>,
2439) -> ExitCode {
2440 if !src.exists() {
2442 return formatter.fail_with_suggestion(
2443 ExitCode::NotFound,
2444 &format!("Source not found: {}", src.display()),
2445 "Check the local source path and retry the copy command.",
2446 );
2447 }
2448
2449 if src.is_dir() && !args.recursive {
2451 return formatter.fail_with_suggestion(
2452 ExitCode::UsageError,
2453 "Source is a directory. Use -r/--recursive to copy directories.",
2454 "Retry with -r or --recursive to copy a directory tree.",
2455 );
2456 }
2457
2458 let alias_manager = match AliasManager::new() {
2460 Ok(am) => am,
2461 Err(e) => {
2462 formatter.error(&format!("Failed to load aliases: {e}"));
2463 return ExitCode::GeneralError;
2464 }
2465 };
2466
2467 let alias = match alias_manager.get(&dst.alias) {
2468 Ok(a) => a,
2469 Err(_) => {
2470 return formatter.fail_with_suggestion(
2471 ExitCode::NotFound,
2472 &format!("Alias '{}' not found", dst.alias),
2473 "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
2474 );
2475 }
2476 };
2477 let client = match S3Client::new(alias).await {
2478 Ok(c) => c,
2479 Err(e) => {
2480 return formatter.fail(
2481 ExitCode::NetworkError,
2482 &format!("Failed to create S3 client: {e}"),
2483 );
2484 }
2485 };
2486
2487 if src.is_file() {
2488 upload_file(&client, src, dst, args, formatter, encryption).await
2490 } else {
2491 upload_directory(&client, src, dst, args, formatter, encryption).await
2493 }
2494}
2495
2496const MULTIPART_THRESHOLD: u64 = rc_s3::multipart::DEFAULT_PART_SIZE;
2498const DOWNLOAD_PROGRESS_THRESHOLD: u64 = 4 * 1024 * 1024;
2500
2501fn update_download_progress(
2502 progress: &mut Option<ProgressBar>,
2503 output_config: &OutputConfig,
2504 bytes_downloaded: u64,
2505 total_size: Option<u64>,
2506) {
2507 let Some(total_size) = total_size else {
2508 return;
2509 };
2510
2511 if total_size < DOWNLOAD_PROGRESS_THRESHOLD {
2512 return;
2513 }
2514
2515 let progress_bar =
2516 progress.get_or_insert_with(|| ProgressBar::new(output_config.clone(), total_size));
2517 progress_bar.set_position(bytes_downloaded);
2518}
2519
2520fn print_upload_success(
2521 formatter: &Formatter,
2522 info: &rc_core::ObjectInfo,
2523 src_display: &str,
2524 dst_display: &str,
2525) {
2526 if formatter.is_json() {
2527 print_copy_json(formatter, info, src_display, dst_display);
2528 } else {
2529 let styled_src = formatter.style_file(src_display);
2530 let styled_dst = formatter.style_file(dst_display);
2531 let styled_size = formatter.style_size(&info.size_human.clone().unwrap_or_default());
2532 formatter.println(&format!("{styled_src} -> {styled_dst} ({styled_size})"));
2533 }
2534}
2535
2536async fn upload_file(
2537 client: &S3Client,
2538 src: &Path,
2539 dst: &RemotePath,
2540 args: &CpArgs,
2541 formatter: &Formatter,
2542 encryption: Option<&ObjectEncryptionRequest>,
2543) -> ExitCode {
2544 let dst_key = if dst.key.is_empty() || dst.key.ends_with('/') {
2546 let filename = src.file_name().unwrap_or_default().to_string_lossy();
2548 format!("{}{}", dst.key, filename)
2549 } else {
2550 dst.key.clone()
2551 };
2552
2553 let target = RemotePath::new(&dst.alias, &dst.bucket, &dst_key);
2554 let src_display = src.display().to_string();
2555 let dst_display = format!("{}/{}/{}", dst.alias, dst.bucket, dst_key);
2556
2557 let file_size = match std::fs::metadata(src) {
2559 Ok(m) => m.len(),
2560 Err(e) => {
2561 return formatter.fail(
2562 ExitCode::GeneralError,
2563 &format!("Failed to read {src_display}: {e}"),
2564 );
2565 }
2566 };
2567 if args.storage_class.is_some() && file_size > MULTIPART_THRESHOLD {
2568 return formatter.fail(
2569 ExitCode::UnsupportedFeature,
2570 "RustFS beta.10 does not persist storage class for multipart uploads",
2571 );
2572 }
2573 if args.dry_run {
2574 let styled_src = formatter.style_file(&src_display);
2575 let styled_dst = formatter.style_file(&dst_display);
2576 formatter.println(&format!(
2577 "Would copy: {styled_src} -> {styled_dst}{}",
2578 transfer_policy_suffix(args)
2579 ));
2580 return ExitCode::Success;
2581 }
2582
2583 let guessed_type: Option<String> = mime_guess::from_path(src)
2585 .first()
2586 .map(|m| m.essence_str().to_string());
2587 let content_type = select_upload_content_type(
2588 args.content_type.as_deref(),
2589 guessed_type.as_deref(),
2590 file_size,
2591 );
2592
2593 let progress = if file_size > MULTIPART_THRESHOLD {
2595 tracing::debug!(
2596 file_size,
2597 threshold = MULTIPART_THRESHOLD,
2598 "Using multipart upload for large file"
2599 );
2600 Some(ProgressBar::new(formatter.output_config(), file_size))
2601 } else {
2602 tracing::debug!(file_size, "Using single put_object for small file");
2603 None
2604 };
2605
2606 let upload_result = match object_write_options(
2608 &args.fidelity,
2609 content_type,
2610 encryption,
2611 args.destination_customer_key.as_ref(),
2612 args.storage_class.clone(),
2613 ) {
2614 Ok(options) => {
2615 client
2616 .put_object_from_path_with_options(&target, src, &options, |bytes_sent| {
2617 if let Some(ref pb) = progress {
2618 pb.set_position(bytes_sent);
2619 }
2620 })
2621 .await
2622 }
2623 Err(error) => Err(error),
2624 };
2625 match upload_result {
2626 Ok(info) => {
2627 if let Some(ref pb) = progress {
2628 pb.finish_and_clear();
2629 }
2630 print_upload_success(formatter, &info, &src_display, &dst_display);
2631 ExitCode::Success
2632 }
2633 Err(e) => {
2634 if let Some(ref pb) = progress {
2635 pb.finish_and_clear();
2636 }
2637 formatter.fail(
2638 exit_code_for_core_error(&e),
2639 &format!("Failed to upload {src_display}: {e}"),
2640 )
2641 }
2642 }
2643}
2644
2645fn select_upload_content_type<'a>(
2646 explicit_type: Option<&'a str>,
2647 guessed_type: Option<&'a str>,
2648 file_size: u64,
2649) -> Option<&'a str> {
2650 if file_size > MULTIPART_THRESHOLD {
2651 explicit_type
2652 } else {
2653 explicit_type.or(guessed_type)
2654 }
2655}
2656
2657async fn upload_directory(
2658 client: &S3Client,
2659 src: &Path,
2660 dst: &RemotePath,
2661 args: &CpArgs,
2662 formatter: &Formatter,
2663 encryption: Option<&ObjectEncryptionRequest>,
2664) -> ExitCode {
2665 use std::fs;
2666
2667 let mut success_count = 0;
2668 let mut error_count = 0;
2669
2670 fn walk_dir(dir: &Path, base: &Path) -> std::io::Result<Vec<(std::path::PathBuf, String)>> {
2672 let mut files = Vec::new();
2673 for entry in fs::read_dir(dir)? {
2674 let entry = entry?;
2675 let path = entry.path();
2676 if path.is_file() {
2677 let relative = path.strip_prefix(base).unwrap_or(&path);
2678 let relative_str = relative.to_string_lossy().to_string();
2679 files.push((path, relative_str));
2680 } else if path.is_dir() {
2681 files.extend(walk_dir(&path, base)?);
2682 }
2683 }
2684 Ok(files)
2685 }
2686
2687 let files = match walk_dir(src, src) {
2688 Ok(f) => f,
2689 Err(e) => {
2690 return formatter.fail(
2691 ExitCode::GeneralError,
2692 &format!("Failed to read directory: {e}"),
2693 );
2694 }
2695 };
2696
2697 for (file_path, relative_path) in files {
2698 let dst_key = if dst.key.is_empty() {
2700 relative_path.replace('\\', "/")
2701 } else if dst.key.ends_with('/') {
2702 format!("{}{}", dst.key, relative_path.replace('\\', "/"))
2703 } else {
2704 format!("{}/{}", dst.key, relative_path.replace('\\', "/"))
2705 };
2706
2707 let target = RemotePath::new(&dst.alias, &dst.bucket, &dst_key);
2708
2709 let result = upload_file(client, &file_path, &target, args, formatter, encryption).await;
2710
2711 if result == ExitCode::Success {
2712 success_count += 1;
2713 } else {
2714 error_count += 1;
2715 if !args.continue_on_error {
2716 return result;
2717 }
2718 }
2719 }
2720
2721 if error_count > 0 {
2722 formatter.warning(&format!(
2723 "Completed with errors: {success_count} succeeded, {error_count} failed"
2724 ));
2725 ExitCode::GeneralError
2726 } else {
2727 if !formatter.is_json() {
2728 formatter.success(&format!("Uploaded {success_count} file(s)."));
2729 }
2730 ExitCode::Success
2731 }
2732}
2733
2734async fn copy_s3_to_local(
2735 src: &RemotePath,
2736 dst: &Path,
2737 args: &CpArgs,
2738 formatter: &Formatter,
2739) -> ExitCode {
2740 let alias_manager = match AliasManager::new() {
2742 Ok(am) => am,
2743 Err(e) => {
2744 formatter.error(&format!("Failed to load aliases: {e}"));
2745 return ExitCode::GeneralError;
2746 }
2747 };
2748
2749 let alias = match alias_manager.get(&src.alias) {
2750 Ok(a) => a,
2751 Err(_) => {
2752 return formatter.fail_with_suggestion(
2753 ExitCode::NotFound,
2754 &format!("Alias '{}' not found", src.alias),
2755 "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
2756 );
2757 }
2758 };
2759 let client = match S3Client::new(alias).await {
2760 Ok(c) => c,
2761 Err(e) => {
2762 return formatter.fail(
2763 ExitCode::NetworkError,
2764 &format!("Failed to create S3 client: {e}"),
2765 );
2766 }
2767 };
2768
2769 let is_prefix = src.key.is_empty() || src.key.ends_with('/');
2771
2772 if is_prefix || args.recursive {
2773 download_prefix(&client, src, dst, args, formatter).await
2775 } else {
2776 download_file(&client, src, dst, args, formatter).await
2778 }
2779}
2780
2781pub(super) async fn download_file(
2782 client: &S3Client,
2783 src: &RemotePath,
2784 dst: &Path,
2785 args: &CpArgs,
2786 formatter: &Formatter,
2787) -> ExitCode {
2788 let src_display = format!("{}/{}/{}", src.alias, src.bucket, src.key);
2789
2790 let dst_path = if dst.is_dir() || dst.to_string_lossy().ends_with('/') {
2792 let filename = src.key.rsplit('/').next().unwrap_or(&src.key);
2793 let filename = match safe_download_relative_path(filename, "", args.local_key_policy()) {
2794 Ok(filename) => filename,
2795 Err(error) => {
2796 return formatter.fail(
2797 ExitCode::UsageError,
2798 &format!("Unsafe object key '{}': {error}", src.key),
2799 );
2800 }
2801 };
2802 dst.join(filename)
2803 } else {
2804 dst.to_path_buf()
2805 };
2806
2807 let dst_display = dst_path.display().to_string();
2808
2809 if args.dry_run {
2810 let styled_src = formatter.style_file(&src_display);
2811 let styled_dst = formatter.style_file(&dst_display);
2812 formatter.println(&format!("Would copy: {styled_src} -> {styled_dst}"));
2813 return ExitCode::Success;
2814 }
2815
2816 if dst_path.exists() && !args.overwrite {
2818 return formatter.fail_with_suggestion(
2819 ExitCode::Conflict,
2820 &format!("Destination exists: {dst_display}. Use --overwrite to replace."),
2821 "Retry with --overwrite if replacing the destination file is intended.",
2822 );
2823 }
2824
2825 if let Some(parent) = dst_path.parent()
2827 && !parent.exists()
2828 && let Err(e) = std::fs::create_dir_all(parent)
2829 {
2830 return formatter.fail(
2831 ExitCode::GeneralError,
2832 &format!("Failed to create directory: {e}"),
2833 );
2834 }
2835
2836 let output_config = formatter.output_config();
2837 let mut progress = None;
2838
2839 let result = client
2841 .download_object_to_path_with_transfer_options(
2842 src,
2843 &dst_path,
2844 &TransferReadOptions {
2845 customer_key: args.source_customer_key.clone(),
2846 ..TransferReadOptions::default()
2847 },
2848 |bytes_downloaded, total_size| {
2849 update_download_progress(
2850 &mut progress,
2851 &output_config,
2852 bytes_downloaded,
2853 total_size,
2854 );
2855 },
2856 )
2857 .await;
2858
2859 if let Some(ref pb) = progress {
2860 pb.finish_and_clear();
2861 }
2862
2863 match result {
2864 Ok(size) => {
2865 let size = size as i64;
2866
2867 if formatter.is_json() {
2868 let output = CpOutput {
2869 status: "success",
2870 source: src_display,
2871 target: dst_display,
2872 size_bytes: Some(size),
2873 size_human: Some(humansize::format_size(size as u64, humansize::BINARY)),
2874 version_id: None,
2875 source_version_id: None,
2876 };
2877 formatter.json(&output);
2878 } else {
2879 let styled_src = formatter.style_file(&src_display);
2880 let styled_dst = formatter.style_file(&dst_display);
2881 let styled_size =
2882 formatter.style_size(&humansize::format_size(size as u64, humansize::BINARY));
2883 formatter.println(&format!("{styled_src} -> {styled_dst} ({styled_size})"));
2884 }
2885 ExitCode::Success
2886 }
2887 Err(e) => {
2888 let err_str = e.to_string();
2889 if err_str.contains("NotFound") || err_str.contains("NoSuchKey") {
2890 formatter.fail_with_suggestion(
2891 ExitCode::NotFound,
2892 &format!("Object not found: {src_display}"),
2893 "Check the object key and bucket path, then retry the copy command.",
2894 )
2895 } else {
2896 formatter.fail(
2897 ExitCode::NetworkError,
2898 &format!("Failed to download {src_display}: {e}"),
2899 )
2900 }
2901 }
2902 }
2903}
2904
2905async fn download_prefix(
2906 client: &S3Client,
2907 src: &RemotePath,
2908 dst: &Path,
2909 args: &CpArgs,
2910 formatter: &Formatter,
2911) -> ExitCode {
2912 use rc_core::ListOptions;
2913
2914 let mut success_count = 0;
2915 let mut error_count = 0;
2916 let mut continuation_token: Option<String> = None;
2917
2918 loop {
2919 let options = ListOptions {
2920 recursive: true,
2921 max_keys: Some(1000),
2922 continuation_token: continuation_token.clone(),
2923 ..Default::default()
2924 };
2925
2926 match client.list_objects(src, options).await {
2927 Ok(result) => {
2928 for item in result.items {
2929 if item.is_dir {
2930 continue;
2931 }
2932
2933 let relative_path = match safe_download_relative_path(
2935 &item.key,
2936 &src.key,
2937 args.local_key_policy(),
2938 ) {
2939 Ok(path) => path,
2940 Err(error) => {
2941 error_count += 1;
2942 formatter.error(&format!(
2943 "Refusing unsafe object key '{}': {error}",
2944 item.key
2945 ));
2946 if !args.continue_on_error {
2947 return ExitCode::UsageError;
2948 }
2949 continue;
2950 }
2951 };
2952 let dst_path = match safe_download_destination(dst, &relative_path).await {
2953 Ok(path) => path,
2954 Err(error) => {
2955 error_count += 1;
2956 formatter.error(&format!(
2957 "Refusing unsafe destination for '{}': {error}",
2958 item.key
2959 ));
2960 if !args.continue_on_error {
2961 return ExitCode::UsageError;
2962 }
2963 continue;
2964 }
2965 };
2966
2967 let obj_src = RemotePath::new(&src.alias, &src.bucket, &item.key);
2968 let result = download_file(client, &obj_src, &dst_path, args, formatter).await;
2969
2970 if result == ExitCode::Success {
2971 success_count += 1;
2972 } else {
2973 error_count += 1;
2974 if !args.continue_on_error {
2975 return result;
2976 }
2977 }
2978 }
2979
2980 if result.truncated {
2981 continuation_token = result.continuation_token;
2982 } else {
2983 break;
2984 }
2985 }
2986 Err(e) => {
2987 return formatter.fail(
2988 ExitCode::NetworkError,
2989 &format!("Failed to list objects: {e}"),
2990 );
2991 }
2992 }
2993 }
2994
2995 if error_count > 0 {
2996 formatter.warning(&format!(
2997 "Completed with errors: {success_count} succeeded, {error_count} failed"
2998 ));
2999 ExitCode::GeneralError
3000 } else if success_count == 0 {
3001 formatter.warning("No objects found to download.");
3002 ExitCode::Success
3003 } else {
3004 if !formatter.is_json() {
3005 formatter.success(&format!("Downloaded {success_count} file(s)."));
3006 }
3007 ExitCode::Success
3008 }
3009}
3010
3011pub(super) fn safe_download_relative_path(
3012 key: &str,
3013 prefix: &str,
3014 policy: ObjectKeyPolicy,
3015) -> Result<PathBuf, String> {
3016 relative_local_path_from_key(key, prefix, policy).map_err(|error| error.to_string())
3017}
3018
3019pub(super) async fn safe_download_destination(
3020 root: &Path,
3021 relative: &Path,
3022) -> Result<PathBuf, String> {
3023 let mut destination = root.to_path_buf();
3024 for component in relative.components() {
3025 let std::path::Component::Normal(component) = component else {
3026 return Err("destination path contains a non-normal component".to_string());
3027 };
3028 destination.push(component);
3029 match tokio::fs::symlink_metadata(&destination).await {
3030 Ok(metadata) if metadata.file_type().is_symlink() => {
3031 return Err(format!(
3032 "destination component '{}' is a symbolic link",
3033 destination.display()
3034 ));
3035 }
3036 Ok(_) => {}
3037 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
3038 Err(error) => {
3039 return Err(format!(
3040 "failed to inspect destination '{}': {error}",
3041 destination.display()
3042 ));
3043 }
3044 }
3045 }
3046
3047 Ok(destination)
3048}
3049
3050async fn copy_s3_to_s3_prepared(
3051 src: &RemotePath,
3052 dst: &RemotePath,
3053 args: &CpArgs,
3054 formatter: &Formatter,
3055 encryption: Option<&ObjectEncryptionRequest>,
3056) -> ExitCode {
3057 if args.source_customer_key.is_some() || args.destination_customer_key.is_some() {
3058 return formatter.fail(
3059 ExitCode::UnsupportedFeature,
3060 "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467",
3061 );
3062 }
3063
3064 let alias_manager = match AliasManager::new() {
3065 Ok(am) => am,
3066 Err(e) => {
3067 formatter.error(&format!("Failed to load aliases: {e}"));
3068 return ExitCode::GeneralError;
3069 }
3070 };
3071
3072 let source_alias = match alias_manager.get(&src.alias) {
3075 Ok(a) => a,
3076 Err(_) => {
3077 return formatter.fail_with_suggestion(
3078 ExitCode::NotFound,
3079 &format!("Alias '{}' not found", src.alias),
3080 "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
3081 );
3082 }
3083 };
3084 let source_client = match S3Client::new(source_alias).await {
3085 Ok(c) => c,
3086 Err(e) => {
3087 return formatter.fail(
3088 ExitCode::NetworkError,
3089 &format!("Failed to create S3 client: {e}"),
3090 );
3091 }
3092 };
3093 let target_client;
3094 let target_client_ref = if src.alias == dst.alias {
3095 &source_client
3096 } else {
3097 let destination_alias = match alias_manager.get(&dst.alias) {
3098 Ok(a) => a,
3099 Err(_) => {
3100 return formatter.fail_with_suggestion(
3101 ExitCode::NotFound,
3102 &format!("Alias '{}' not found", dst.alias),
3103 "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.",
3104 );
3105 }
3106 };
3107 target_client = match S3Client::new(destination_alias).await {
3108 Ok(c) => c,
3109 Err(e) => {
3110 return formatter.fail(
3111 ExitCode::NetworkError,
3112 &format!("Failed to create destination S3 client: {e}"),
3113 );
3114 }
3115 };
3116 &target_client
3117 };
3118
3119 let src_display = format!("{}/{}/{}", src.alias, src.bucket, src.key);
3120 let dst_display = format!("{}/{}/{}", dst.alias, dst.bucket, dst.key);
3121
3122 if args.dry_run && args.storage_class.is_none() {
3123 let styled_src = formatter.style_file(&src_display);
3124 let styled_dst = formatter.style_file(&dst_display);
3125 formatter.println(&format!(
3126 "Would copy: {styled_src} -> {styled_dst}{}",
3127 transfer_policy_suffix(args)
3128 ));
3129 return ExitCode::Success;
3130 }
3131
3132 let source_info = match source_client.head_object(src).await {
3133 Ok(info) => info,
3134 Err(Error::NotFound(_)) => {
3135 return formatter.fail_with_suggestion(
3136 ExitCode::NotFound,
3137 &format!("Source not found: {src_display}"),
3138 "Check the source bucket and object key, then retry the copy command.",
3139 );
3140 }
3141 Err(error) => {
3142 return formatter.fail(
3143 exit_code_for_core_error(&error),
3144 &format!("Failed to inspect source object: {error}"),
3145 );
3146 }
3147 };
3148 let source_size = source_info
3149 .size_bytes
3150 .and_then(|size| u64::try_from(size).ok());
3151 if args.storage_class.is_some()
3152 && source_size.is_none_or(|size| {
3153 if src.alias == dst.alias {
3154 rc_core::requires_multipart_copy(size)
3155 } else {
3156 size > MULTIPART_THRESHOLD
3157 }
3158 })
3159 {
3160 return formatter.fail(
3161 ExitCode::UnsupportedFeature,
3162 "RustFS beta.10 does not persist storage class for multipart or unknown-size copies",
3163 );
3164 }
3165 if args.dry_run {
3166 let styled_src = formatter.style_file(&src_display);
3167 let styled_dst = formatter.style_file(&dst_display);
3168 formatter.println(&format!(
3169 "Would copy: {styled_src} -> {styled_dst}{}",
3170 transfer_policy_suffix(args)
3171 ));
3172 return ExitCode::Success;
3173 }
3174
3175 let cancellation = MultipartCopyCancellation::new();
3176 let transfer_cancellation = TransferCancellation::new();
3177 let signal_task = tokio::spawn({
3178 let cancellation = cancellation.clone();
3179 let transfer_cancellation = transfer_cancellation.clone();
3180 async move {
3181 if tokio::signal::ctrl_c().await.is_ok() {
3182 cancellation.cancel();
3183 transfer_cancellation.cancel();
3184 }
3185 }
3186 });
3187 let ignore_progress = |_: u64| {};
3188 let copy = perform_planned_remote_copy(
3189 &source_client,
3190 target_client_ref,
3191 src,
3192 dst,
3193 &source_info,
3194 encryption,
3195 &cancellation,
3196 &ignore_progress,
3197 args,
3198 );
3199 tokio::pin!(copy);
3200 let result = tokio::select! {
3201 biased;
3202 _ = transfer_cancellation.cancelled() => {
3203 let _ = copy.await;
3206 Err(Error::Interrupted("Copy interrupted".to_string()))
3207 }
3208 result = &mut copy => {
3209 if transfer_cancellation.is_cancelled() {
3210 Err(Error::Interrupted("Copy interrupted".to_string()))
3211 } else {
3212 result
3213 }
3214 }
3215 };
3216 signal_task.abort();
3217 let _ = signal_task.await;
3218
3219 match result {
3220 Ok(result) => {
3221 if formatter.is_json() {
3222 print_copy_json(formatter, &result.object, &src_display, &dst_display);
3223 } else {
3224 let styled_src = formatter.style_file(&src_display);
3225 let styled_dst = formatter.style_file(&dst_display);
3226 let styled_size =
3227 formatter.style_size(&result.object.size_human.unwrap_or_else(|| {
3228 humansize::format_size(result.bytes_copied, humansize::BINARY)
3229 }));
3230 let upload = result
3231 .upload_id
3232 .map(|upload_id| format!(" upload-id={upload_id}"))
3233 .unwrap_or_default();
3234 formatter.println(&format!(
3235 "{styled_src} -> {styled_dst} ({styled_size}){upload}"
3236 ));
3237 }
3238 ExitCode::Success
3239 }
3240 Err(error) => {
3241 if matches!(error, Error::NotFound(_) | Error::VersionNotFound { .. }) {
3242 formatter.fail_with_suggestion(
3243 ExitCode::NotFound,
3244 &format!("Source not found: {src_display}"),
3245 "Check the source bucket and object key, then retry the copy command.",
3246 )
3247 } else {
3248 formatter.fail(
3249 exit_code_for_core_error(&error),
3250 &format!("Failed to copy: {error}"),
3251 )
3252 }
3253 }
3254 }
3255}
3256
3257fn print_copy_json(formatter: &Formatter, info: &rc_core::ObjectInfo, source: &str, target: &str) {
3258 if info.version_id.is_some() || info.source_version_id.is_some() {
3259 formatter.json(&V3SuccessEnvelope::versioned_objects(VersionCopyData {
3260 operation: "copy",
3261 source: source.to_string(),
3262 target: target.to_string(),
3263 source_version_id: info.source_version_id.clone(),
3264 version_id: info.version_id.clone(),
3265 size_bytes: info.size_bytes,
3266 size_human: info.size_human.clone(),
3267 }));
3268 } else {
3269 formatter.json(&CpOutput {
3270 status: "success",
3271 source: source.to_string(),
3272 target: target.to_string(),
3273 size_bytes: info.size_bytes,
3274 size_human: info.size_human.clone(),
3275 version_id: None,
3276 source_version_id: None,
3277 });
3278 }
3279}
3280
3281fn parse_kms_target(value: &str) -> Result<(String, String), String> {
3282 let (target, key_id) = value
3283 .split_once('=')
3284 .ok_or_else(|| "Expected TARGET=KMS_KEY_ID for --enc-kms".to_string())?;
3285
3286 if target.is_empty() || key_id.is_empty() {
3287 return Err("Expected TARGET=KMS_KEY_ID for --enc-kms".to_string());
3288 }
3289
3290 Ok((target.to_string(), key_id.to_string()))
3291}
3292
3293pub(crate) fn parse_destination_encryption(
3294 enc_s3: &[String],
3295 enc_kms: &[String],
3296 target: &ParsedPath,
3297) -> Result<Option<ObjectEncryptionRequest>, String> {
3298 if enc_s3.is_empty() && enc_kms.is_empty() {
3299 return Ok(None);
3300 }
3301
3302 let remote = match target {
3303 ParsedPath::Remote(remote) => remote,
3304 ParsedPath::Local(_) => {
3305 return Err("Destination encryption flags must reference a remote destination".into());
3306 }
3307 };
3308
3309 let target_display = remote.to_string();
3310 let s3_matches = enc_s3.iter().any(|value| value == &target_display);
3311 let kms_targets = enc_kms
3312 .iter()
3313 .map(|value| parse_kms_target(value))
3314 .collect::<Result<Vec<_>, _>>()?;
3315 let kms_match = kms_targets
3316 .iter()
3317 .find(|(candidate, _)| candidate == &target_display);
3318
3319 if !enc_s3.is_empty() && !s3_matches {
3320 return Err(format!(
3321 "--enc-s3 target must exactly match the remote destination: {target_display}"
3322 ));
3323 }
3324
3325 if !enc_kms.is_empty() && kms_match.is_none() {
3326 return Err(format!(
3327 "--enc-kms target must exactly match the remote destination: {target_display}"
3328 ));
3329 }
3330
3331 match (s3_matches, kms_match) {
3332 (true, Some(_)) => Err(format!(
3333 "--enc-s3 and --enc-kms cannot target the same destination: {target_display}"
3334 )),
3335 (true, None) => Ok(Some(ObjectEncryptionRequest::SseS3)),
3336 (false, Some((_, key_id))) => Ok(Some(ObjectEncryptionRequest::SseKms {
3337 key_id: key_id.clone(),
3338 })),
3339 (false, None) => Ok(None),
3340 }
3341}
3342
3343#[cfg(test)]
3344mod tests {
3345 use super::*;
3346 use rc_core::{Alias, ConfigManager};
3347 use tempfile::TempDir;
3348
3349 fn temp_alias_manager() -> (AliasManager, TempDir) {
3350 let temp_dir = TempDir::new().expect("create temp dir");
3351 let config_path = temp_dir.path().join("config.toml");
3352 let config_manager = ConfigManager::with_path(config_path);
3353 let alias_manager = AliasManager::with_config_manager(config_manager);
3354 (alias_manager, temp_dir)
3355 }
3356
3357 #[test]
3358 fn test_parse_local_path() {
3359 let result = parse_path("./file.txt").unwrap();
3360 assert!(matches!(result, ParsedPath::Local(_)));
3361 }
3362
3363 #[test]
3364 fn test_parse_remote_path() {
3365 let result = parse_path("myalias/bucket/file.txt").unwrap();
3366 assert!(matches!(result, ParsedPath::Remote(_)));
3367 }
3368
3369 #[test]
3370 fn test_parse_local_absolute_path() {
3371 #[cfg(unix)]
3373 let path = "/home/user/file.txt";
3374 #[cfg(windows)]
3375 let path = "C:\\Users\\user\\file.txt";
3376
3377 let result = parse_path(path).unwrap();
3378 assert!(matches!(result, ParsedPath::Local(_)));
3379 if let ParsedPath::Local(p) = result {
3380 assert!(p.is_absolute());
3381 }
3382 }
3383
3384 #[test]
3385 fn test_parse_local_relative_path() {
3386 let result = parse_path("../file.txt").unwrap();
3387 assert!(matches!(result, ParsedPath::Local(_)));
3388 }
3389
3390 #[test]
3391 fn test_parse_remote_path_bucket_only() {
3392 let result = parse_path("myalias/bucket/").unwrap();
3393 assert!(matches!(result, ParsedPath::Remote(_)));
3394 if let ParsedPath::Remote(r) = result {
3395 assert_eq!(r.alias, "myalias");
3396 assert_eq!(r.bucket, "bucket");
3397 assert!(r.key.is_empty());
3398 }
3399 }
3400
3401 #[test]
3402 fn test_parse_remote_path_with_deep_key() {
3403 let result = parse_path("myalias/bucket/dir1/dir2/file.txt").unwrap();
3404 assert!(matches!(result, ParsedPath::Remote(_)));
3405 if let ParsedPath::Remote(r) = result {
3406 assert_eq!(r.alias, "myalias");
3407 assert_eq!(r.bucket, "bucket");
3408 assert_eq!(r.key, "dir1/dir2/file.txt");
3409 }
3410 }
3411
3412 #[test]
3413 fn test_download_progress_created_for_large_transfer() {
3414 let output_config = OutputConfig::default();
3415 let mut progress = None;
3416
3417 update_download_progress(
3418 &mut progress,
3419 &output_config,
3420 1024,
3421 Some(DOWNLOAD_PROGRESS_THRESHOLD),
3422 );
3423
3424 let progress = progress.expect("large download should create progress bar");
3425 assert!(progress.is_visible());
3426 progress.finish_and_clear();
3427 }
3428
3429 #[test]
3430 fn test_download_progress_skips_small_transfer() {
3431 let output_config = OutputConfig::default();
3432 let mut progress = None;
3433
3434 update_download_progress(
3435 &mut progress,
3436 &output_config,
3437 1024,
3438 Some(DOWNLOAD_PROGRESS_THRESHOLD - 1),
3439 );
3440
3441 assert!(progress.is_none());
3442 }
3443
3444 #[test]
3445 fn test_download_progress_skips_unknown_total_size() {
3446 let output_config = OutputConfig::default();
3447 let mut progress = None;
3448
3449 update_download_progress(&mut progress, &output_config, 1024, None);
3450
3451 assert!(progress.is_none());
3452 }
3453
3454 #[test]
3455 fn test_download_progress_respects_no_progress_config() {
3456 let output_config = OutputConfig {
3457 no_progress: true,
3458 ..Default::default()
3459 };
3460 let mut progress = None;
3461
3462 update_download_progress(
3463 &mut progress,
3464 &output_config,
3465 1024,
3466 Some(DOWNLOAD_PROGRESS_THRESHOLD),
3467 );
3468
3469 let progress = progress.expect("large download should create progress state");
3470 assert!(!progress.is_visible());
3471 }
3472
3473 #[test]
3474 fn download_relative_path_preserves_safe_nested_keys() {
3475 let relative = safe_download_relative_path(
3476 "reports/2026/july/data.csv",
3477 "reports/",
3478 ObjectKeyPolicy::Logical,
3479 )
3480 .expect("safe key should resolve");
3481
3482 assert_eq!(
3483 relative,
3484 PathBuf::from("2026").join("july").join("data.csv")
3485 );
3486 }
3487
3488 #[test]
3489 fn download_relative_path_rejects_traversal_and_absolute_keys() {
3490 for key in [
3491 "reports/../../escaped",
3492 "reports/..\\..\\escaped",
3493 "/absolute/path",
3494 ] {
3495 assert!(
3496 safe_download_relative_path(key, "reports/", ObjectKeyPolicy::Logical).is_err(),
3497 "unsafe key should be rejected: {key}"
3498 );
3499 }
3500 }
3501
3502 #[test]
3503 fn download_relative_path_accepts_colon_keys_on_logical_destinations() {
3504 let relative = safe_download_relative_path(
3505 "loki/fake/deadbeef/19f6abd9af4:19f6abe0e77:499628ff",
3506 "loki/",
3507 ObjectKeyPolicy::Logical,
3508 )
3509 .expect("colon keys are valid Unix file names");
3510
3511 assert_eq!(
3512 relative,
3513 PathBuf::from("fake")
3514 .join("deadbeef")
3515 .join("19f6abd9af4:19f6abe0e77:499628ff")
3516 );
3517 }
3518
3519 #[test]
3520 fn download_relative_path_rejects_colon_keys_when_portable_names_requested() {
3521 for key in [
3522 "reports/C:/escaped",
3523 "reports/safe:stream",
3524 "reports/CON.txt",
3525 "reports/trailing.",
3526 ] {
3527 assert!(
3528 safe_download_relative_path(key, "reports/", ObjectKeyPolicy::WindowsPortable)
3529 .is_err(),
3530 "portable-unsafe key should be rejected: {key}"
3531 );
3532 }
3533 }
3534
3535 #[cfg(unix)]
3536 #[tokio::test]
3537 async fn download_destination_rejects_existing_symlink_components() {
3538 use std::os::unix::fs::symlink;
3539
3540 let root = tempfile::tempdir().expect("create destination root");
3541 let outside = tempfile::tempdir().expect("create outside directory");
3542 symlink(outside.path(), root.path().join("linked")).expect("create test symlink");
3543
3544 let result =
3545 safe_download_destination(root.path(), &PathBuf::from("linked/file.txt")).await;
3546
3547 assert!(result.is_err());
3548 }
3549
3550 #[cfg(unix)]
3551 #[tokio::test]
3552 async fn download_destination_rejects_existing_symlink_file() {
3553 use std::os::unix::fs::symlink;
3554
3555 let root = tempfile::tempdir().expect("create destination root");
3556 let outside = tempfile::tempdir().expect("create outside directory");
3557 let outside_file = outside.path().join("file.txt");
3558 std::fs::write(&outside_file, b"outside").expect("write outside file");
3559 symlink(&outside_file, root.path().join("file.txt")).expect("create test symlink");
3560
3561 let result = safe_download_destination(root.path(), &PathBuf::from("file.txt")).await;
3562
3563 assert!(result.is_err());
3564 }
3565
3566 #[cfg(unix)]
3567 #[test]
3568 fn recursive_source_collection_rejects_symbolic_links() {
3569 use std::os::unix::fs::symlink;
3570
3571 let root = tempfile::tempdir().expect("create source root");
3572 let outside = tempfile::NamedTempFile::new().expect("create outside file");
3573 symlink(outside.path(), root.path().join("linked.txt")).expect("create source symlink");
3574 let mut files = Vec::new();
3575
3576 let result = collect_local_files(root.path(), root.path(), &mut files);
3577
3578 assert!(result.is_err());
3579 assert!(files.is_empty());
3580 }
3581
3582 #[test]
3583 fn test_select_upload_content_type_uses_guess_for_small_files() {
3584 let selected =
3585 select_upload_content_type(None, Some("text/plain"), MULTIPART_THRESHOLD - 1);
3586
3587 assert_eq!(selected, Some("text/plain"));
3588 }
3589
3590 #[test]
3591 fn test_select_upload_content_type_skips_guess_for_multipart_files() {
3592 let selected =
3593 select_upload_content_type(None, Some("text/plain"), MULTIPART_THRESHOLD + 1);
3594
3595 assert_eq!(selected, None);
3596 }
3597
3598 #[test]
3599 fn test_select_upload_content_type_uses_guess_at_multipart_boundary() {
3600 let selected = select_upload_content_type(None, Some("text/plain"), MULTIPART_THRESHOLD);
3601
3602 assert_eq!(selected, Some("text/plain"));
3603 }
3604
3605 #[test]
3606 fn test_select_upload_content_type_keeps_explicit_type_for_multipart_files() {
3607 let selected = select_upload_content_type(
3608 Some("application/octet-stream"),
3609 Some("text/plain"),
3610 MULTIPART_THRESHOLD + 1,
3611 );
3612
3613 assert_eq!(selected, Some("application/octet-stream"));
3614 }
3615
3616 #[test]
3617 fn test_parse_cp_path_prefers_existing_local_path_when_alias_missing() {
3618 let (alias_manager, temp_dir) = temp_alias_manager();
3619 let full = temp_dir.path().join("issue-2094-local").join("file.txt");
3620 let full_str = full.to_string_lossy().to_string();
3621
3622 if let Some(parent) = full.parent() {
3623 std::fs::create_dir_all(parent).expect("create parent dirs");
3624 }
3625 std::fs::write(&full, b"test").expect("write local file");
3626
3627 let parsed = parse_cp_path(&full_str, Some(&alias_manager)).expect("parse path");
3628 assert!(matches!(parsed, ParsedPath::Local(_)));
3629 }
3630
3631 #[test]
3632 fn test_parse_cp_path_keeps_remote_when_alias_exists() {
3633 let (alias_manager, _temp_dir) = temp_alias_manager();
3634 alias_manager
3635 .set(Alias::new("target", "http://localhost:9000", "a", "b"))
3636 .expect("set alias");
3637
3638 let parsed = parse_cp_path("target/bucket/file.txt", Some(&alias_manager))
3639 .expect("parse remote path");
3640 assert!(matches!(parsed, ParsedPath::Remote(_)));
3641 }
3642
3643 #[test]
3644 fn test_parse_cp_path_keeps_remote_when_local_missing() {
3645 let (alias_manager, _temp_dir) = temp_alias_manager();
3646 let parsed = parse_cp_path("missing/bucket/file.txt", Some(&alias_manager))
3647 .expect("parse remote path");
3648 assert!(matches!(parsed, ParsedPath::Remote(_)));
3649 }
3650
3651 #[test]
3652 fn test_cp_args_defaults() {
3653 let args = CpArgs::single("src", "dst");
3654 assert!(args.overwrite);
3655 assert!(!args.recursive);
3656 assert!(!args.dry_run);
3657 assert_eq!(args.concurrency, None);
3658 assert_eq!(args.retry_attempts, None);
3659 let controls = build_transfer_controls(&args).expect("default controls");
3660 assert_eq!(controls.concurrency, DEFAULT_TRANSFER_CONCURRENCY);
3661 assert_eq!(controls.retry.max_attempts, DEFAULT_RETRY_ATTEMPTS);
3662 assert_eq!(
3663 controls.retry.initial_backoff_ms,
3664 DEFAULT_RETRY_INITIAL_BACKOFF_MS
3665 );
3666 assert_eq!(controls.retry.max_backoff_ms, DEFAULT_RETRY_MAX_BACKOFF_MS);
3667 }
3668
3669 #[test]
3670 fn explicit_default_transfer_controls_enable_planner_routing() {
3671 let defaults = CpArgs::single("src", "dst");
3672 assert!(!uses_transfer_planner(&defaults));
3673
3674 let mut explicit_concurrency = defaults.clone();
3675 explicit_concurrency.concurrency = Some(DEFAULT_TRANSFER_CONCURRENCY);
3676 let mut explicit_attempts = defaults.clone();
3677 explicit_attempts.retry_attempts = Some(DEFAULT_RETRY_ATTEMPTS);
3678 let mut explicit_initial_backoff = defaults.clone();
3679 explicit_initial_backoff.retry_initial_backoff_ms = Some(DEFAULT_RETRY_INITIAL_BACKOFF_MS);
3680 let mut explicit_max_backoff = defaults;
3681 explicit_max_backoff.retry_max_backoff_ms = Some(DEFAULT_RETRY_MAX_BACKOFF_MS);
3682
3683 for args in [
3684 explicit_concurrency,
3685 explicit_attempts,
3686 explicit_initial_backoff,
3687 explicit_max_backoff,
3688 ] {
3689 assert!(uses_transfer_planner(&args));
3690 }
3691 }
3692
3693 #[test]
3694 fn planned_client_aliases_are_deduplicated() {
3695 let first = TransferCandidate {
3696 payload: CpOperation::LocalToRemote {
3697 source: PathBuf::from("first.txt"),
3698 target: RemotePath::new("shared", "bucket", "first.txt"),
3699 encryption: None,
3700 },
3701 source: "first.txt".to_string(),
3702 target: "shared/bucket/first.txt".to_string(),
3703 relative_path: "first.txt".to_string(),
3704 modified: None,
3705 size_bytes: Some(1),
3706 };
3707 let second = TransferCandidate {
3708 payload: CpOperation::RemoteToLocal {
3709 source: RemotePath::new("shared", "bucket", "second.txt"),
3710 target: PathBuf::from("second.txt"),
3711 },
3712 source: "shared/bucket/second.txt".to_string(),
3713 target: "second.txt".to_string(),
3714 relative_path: "second.txt".to_string(),
3715 modified: None,
3716 size_bytes: Some(2),
3717 };
3718
3719 assert_eq!(
3720 planned_client_aliases(&[first, second])
3721 .into_iter()
3722 .collect::<Vec<_>>(),
3723 ["shared"]
3724 );
3725 }
3726
3727 #[test]
3728 fn planned_client_aliases_include_both_sides_of_a_cross_alias_copy() {
3729 let candidate = TransferCandidate {
3730 payload: CpOperation::RemoteToRemote {
3731 source: RemotePath::new("alpha", "source", "file.txt"),
3732 target: RemotePath::new("beta", "target", "file.txt"),
3733 source_info: Box::new(ObjectInfo::file("file.txt", 4)),
3734 encryption: None,
3735 },
3736 source: "alpha/source/file.txt".to_string(),
3737 target: "beta/target/file.txt".to_string(),
3738 relative_path: "file.txt".to_string(),
3739 modified: None,
3740 size_bytes: Some(4),
3741 };
3742
3743 assert_eq!(
3744 planned_client_aliases(&[candidate])
3745 .into_iter()
3746 .collect::<Vec<_>>(),
3747 ["alpha", "beta"]
3748 );
3749 }
3750
3751 #[test]
3752 fn planned_client_aliases_keep_same_alias_remote_copy_on_one_alias() {
3753 let candidate = TransferCandidate {
3754 payload: CpOperation::RemoteToRemote {
3755 source: RemotePath::new("shared", "source", "file.txt"),
3756 target: RemotePath::new("shared", "target", "file.txt"),
3757 source_info: Box::new(ObjectInfo::file("file.txt", 4)),
3758 encryption: None,
3759 },
3760 source: "shared/source/file.txt".to_string(),
3761 target: "shared/target/file.txt".to_string(),
3762 relative_path: "file.txt".to_string(),
3763 modified: None,
3764 size_bytes: Some(4),
3765 };
3766
3767 assert_eq!(
3768 planned_client_aliases(&[candidate])
3769 .into_iter()
3770 .collect::<Vec<_>>(),
3771 ["shared"]
3772 );
3773 }
3774
3775 #[tokio::test]
3776 async fn planning_client_reuses_one_connection_pool_per_alias() {
3777 let (alias_manager, _temp_dir) = temp_alias_manager();
3778 alias_manager
3779 .set(Alias::new(
3780 "shared",
3781 "http://localhost:9000",
3782 "access",
3783 "secret",
3784 ))
3785 .expect("set alias");
3786 let mut clients = HashMap::new();
3787
3788 let first = planning_client(&mut clients, &alias_manager, "shared")
3789 .await
3790 .expect("create first client");
3791 let second = planning_client(&mut clients, &alias_manager, "shared")
3792 .await
3793 .expect("reuse first client");
3794
3795 assert_eq!(clients.len(), 1);
3796 assert!(Arc::ptr_eq(&first, &second));
3797 }
3798
3799 #[test]
3800 fn planned_remote_copy_uses_candidate_size_for_multipart_guard() {
3801 let candidate = TransferCandidate {
3802 payload: CpOperation::RemoteToRemote {
3803 source: RemotePath::new("shared", "source", "large.bin"),
3804 target: RemotePath::new("shared", "target", "large.bin"),
3805 source_info: Box::new(ObjectInfo::file(
3806 "large.bin",
3807 (MAX_SINGLE_COPY_SIZE + 1) as i64,
3808 )),
3809 encryption: None,
3810 },
3811 source: "shared/source/large.bin".to_string(),
3812 target: "shared/target/large.bin".to_string(),
3813 relative_path: "large.bin".to_string(),
3814 modified: None,
3815 size_bytes: Some(MAX_SINGLE_COPY_SIZE + 1),
3816 };
3817
3818 assert!(requires_multipart_copy(candidate.size_bytes));
3819 }
3820
3821 #[test]
3822 fn planned_remote_copy_uses_exact_five_gib_boundary() {
3823 assert!(!requires_multipart_copy(Some(MAX_SINGLE_COPY_SIZE)));
3824 assert!(requires_multipart_copy(Some(MAX_SINGLE_COPY_SIZE + 1)));
3825 }
3826
3827 #[test]
3828 fn planned_progress_replaces_attempt_bytes_instead_of_double_counting() {
3829 let progress = PlannedCopyProgress::new(
3830 OutputConfig {
3831 no_progress: true,
3832 ..OutputConfig::default()
3833 },
3834 20,
3835 );
3836 let first = ("source/a".to_string(), "target/a".to_string());
3837 let second = ("source/b".to_string(), "target/b".to_string());
3838
3839 progress.set(&first, 7);
3840 progress.set(&second, 5);
3841 progress.reset(&first);
3842 progress.set(&first, 15);
3843
3844 let positions = progress
3845 .positions
3846 .lock()
3847 .expect("planned copy progress lock");
3848 assert_eq!(positions.get(&first), Some(&15));
3849 assert_eq!(positions.get(&second), Some(&5));
3850 assert_eq!(
3851 positions.values().copied().fold(0_u64, u64::saturating_add),
3852 20
3853 );
3854 }
3855
3856 #[test]
3857 fn recursive_remote_mapping_preserves_source_relative_keys() {
3858 let source = RemotePath::new("shared", "source", "src/");
3859 let target = RemotePath::new("shared", "destination", "archive/");
3860
3861 let (destination, relative) = recursive_remote_target(
3862 &source,
3863 &target,
3864 "src/nested/report.csv",
3865 false,
3866 ObjectKeyPolicy::for_remote_destination(),
3867 )
3868 .expect("map recursive object");
3869
3870 assert_eq!(relative, "nested/report.csv");
3871 assert_eq!(destination.key, "archive/nested/report.csv");
3872 }
3873
3874 #[test]
3875 fn recursive_bucket_root_mapping_keeps_the_full_object_key() {
3876 let source = RemotePath::new("shared", "source", "");
3877 let target = RemotePath::new("shared", "destination", "archive/");
3878
3879 let (destination, relative) = recursive_remote_target(
3880 &source,
3881 &target,
3882 "nested/report.csv",
3883 false,
3884 ObjectKeyPolicy::for_remote_destination(),
3885 )
3886 .expect("map bucket object");
3887
3888 assert_eq!(relative, "nested/report.csv");
3889 assert_eq!(destination.key, "archive/nested/report.csv");
3890 }
3891
3892 #[test]
3893 fn recursive_remote_mapping_rejects_unsafe_listed_keys() {
3894 let source = RemotePath::new("shared", "source", "src/");
3895 let target = RemotePath::new("shared", "destination", "archive/");
3896
3897 for object_key in [
3898 "/absolute.txt",
3899 "src/../escape.txt",
3900 "src\\escape.txt",
3901 "src/control\u{0007}.txt",
3902 ] {
3903 assert!(
3904 recursive_remote_target(
3905 &source,
3906 &target,
3907 object_key,
3908 false,
3909 ObjectKeyPolicy::for_remote_destination(),
3910 )
3911 .is_err(),
3912 "unsafe listed key should be rejected: {object_key:?}"
3913 );
3914 }
3915 }
3916
3917 #[test]
3918 fn remote_child_rejects_unsafe_relative_keys() {
3919 let target = RemotePath::new("shared", "destination", "archive/");
3920
3921 for relative in [
3922 "../escape.txt",
3923 "nested\\escape.txt",
3924 "nested/control\u{0007}.txt",
3925 ] {
3926 assert!(
3927 remote_child(&target, relative, ObjectKeyPolicy::for_remote_destination(),)
3928 .is_err(),
3929 "unsafe relative key should be rejected: {relative:?}"
3930 );
3931 }
3932 }
3933
3934 #[test]
3935 fn remote_target_rejects_absolute_prefixes() {
3936 let target = RemotePath::new("shared", "destination", "/archive/");
3937
3938 assert!(
3939 normalize_remote_target(&target, ObjectKeyPolicy::for_remote_destination()).is_err()
3940 );
3941 }
3942
3943 #[test]
3944 fn recursive_remote_overlap_is_boundary_aware_and_symmetric() {
3945 let source = RemotePath::new("shared", "bucket", "src/");
3946 let child = RemotePath::new("shared", "bucket", "src/archive/");
3947 let sibling = RemotePath::new("shared", "bucket", "src-old/");
3948 let other_bucket = RemotePath::new("shared", "other", "src/archive/");
3949
3950 assert!(remote_copy_scopes_overlap(&source, &child));
3951 assert!(remote_copy_scopes_overlap(&child, &source));
3952 assert!(!remote_copy_scopes_overlap(&source, &sibling));
3953 assert!(!remote_copy_scopes_overlap(&source, &other_bucket));
3954 }
3955
3956 #[test]
3957 fn multipart_options_require_and_preserve_planned_source_identity() {
3958 let mut source = rc_core::ObjectInfo::file("large.bin", (MAX_SINGLE_COPY_SIZE + 1) as i64);
3959 source.etag = Some("planned-etag".to_string());
3960 source.version_id = Some("source-version".to_string());
3961 source.content_type = Some("application/octet-stream".to_string());
3962 source.metadata = Some(HashMap::from([(
3963 "project".to_string(),
3964 "archive".to_string(),
3965 )]));
3966
3967 let options = multipart_options_from_source(&source).expect("multipart options");
3968
3969 assert_eq!(options.source_size, MAX_SINGLE_COPY_SIZE + 1);
3970 assert_eq!(options.source_etag, "planned-etag");
3971 assert_eq!(options.source_version_id.as_deref(), Some("source-version"));
3972 assert_eq!(
3973 options.content_type.as_deref(),
3974 Some("application/octet-stream")
3975 );
3976 assert_eq!(
3977 options.metadata.get("project").map(String::as_str),
3978 Some("archive")
3979 );
3980 assert_eq!(
3981 options.metadata.get("rc-source-etag").map(String::as_str),
3982 Some("planned-etag")
3983 );
3984 }
3985
3986 #[test]
3987 fn transfer_age_and_rewind_use_utc_cutoffs() {
3988 let now: Timestamp = "2026-07-21T12:00:00Z".parse().expect("valid UTC timestamp");
3989
3990 assert_eq!(
3991 parse_age_cutoff("1h", now).expect("valid age").to_string(),
3992 "2026-07-21T11:00:00Z"
3993 );
3994 assert_eq!(
3995 parse_rewind_cutoff("2026-07-20T08:30:00+08:00", now)
3996 .expect("valid offset timestamp")
3997 .to_string(),
3998 "2026-07-20T00:30:00Z"
3999 );
4000 }
4001
4002 #[test]
4003 fn transfer_rate_parser_supports_decimal_and_binary_units() {
4004 assert_eq!(parse_byte_rate("10MB/s").expect("decimal rate"), 10_000_000);
4005 assert_eq!(parse_byte_rate("10MiB/s").expect("binary rate"), 10_485_760);
4006 assert!(parse_byte_rate("0").is_err());
4007 assert!(parse_byte_rate("10widgets/s").is_err());
4008 }
4009
4010 #[tokio::test]
4011 async fn planner_rejects_two_sources_that_resolve_to_one_target() {
4012 let (alias_manager, temp_dir) = temp_alias_manager();
4013 let first_dir = temp_dir.path().join("first");
4014 let second_dir = temp_dir.path().join("second");
4015 std::fs::create_dir_all(&first_dir).expect("create first source dir");
4016 std::fs::create_dir_all(&second_dir).expect("create second source dir");
4017 let first = first_dir.join("same.txt");
4018 let second = second_dir.join("same.txt");
4019 std::fs::write(&first, b"first").expect("write first source");
4020 std::fs::write(&second, b"second").expect("write second source");
4021
4022 let candidates = build_transfer_candidates(
4023 &[ParsedPath::Local(first), ParsedPath::Local(second)],
4024 &ParsedPath::Remote(RemotePath::new("target", "bucket", "prefix/")),
4025 true,
4026 false,
4027 None,
4028 None,
4029 &alias_manager,
4030 ObjectKeyPolicy::Logical,
4031 )
4032 .await
4033 .expect("sources can be expanded before selection");
4034 let plan = TransferPlan::build(candidates, &TransferSelection::default());
4035 let error = validate_plan_targets(&plan).expect_err("colliding destinations must fail");
4036
4037 assert!(error.to_string().contains("same destination"));
4038 }
4039
4040 #[test]
4041 fn parse_enc_kms_target_requires_equals_separator() {
4042 let error = parse_kms_target("local/bucket/file.txt").expect_err("missing key separator");
4043 assert!(error.contains("Expected TARGET=KMS_KEY_ID"));
4044 }
4045
4046 #[test]
4047 fn destination_encryption_rejects_local_targets() {
4048 let error = parse_destination_encryption(
4049 &[String::from("./local.txt")],
4050 &[],
4051 &ParsedPath::Local(std::path::PathBuf::from("./local.txt")),
4052 )
4053 .expect_err("local target should be rejected");
4054
4055 assert!(error.contains("must reference a remote destination"));
4056 }
4057
4058 #[test]
4059 fn destination_encryption_detects_conflicting_flags_for_same_target() {
4060 let target = ParsedPath::Remote(RemotePath::new("local", "bucket", "file.txt"));
4061 let error = parse_destination_encryption(
4062 &[String::from("local/bucket/file.txt")],
4063 &[String::from("local/bucket/file.txt=kms-key")],
4064 &target,
4065 )
4066 .expect_err("same target conflict should fail");
4067
4068 assert!(error.contains("cannot target the same destination"));
4069 }
4070
4071 #[test]
4072 fn destination_encryption_rejects_unmatched_s3_target() {
4073 let target = ParsedPath::Remote(RemotePath::new("local", "bucket", "file.txt"));
4074 let error =
4075 parse_destination_encryption(&[String::from("local/bucket/typo.txt")], &[], &target)
4076 .expect_err("unmatched s3 target should fail");
4077
4078 assert!(error.contains("must exactly match the remote destination"));
4079 }
4080
4081 #[test]
4082 fn destination_encryption_rejects_unmatched_kms_target() {
4083 let target = ParsedPath::Remote(RemotePath::new("local", "bucket", "file.txt"));
4084 let error = parse_destination_encryption(
4085 &[],
4086 &[String::from("local/bucket/typo.txt=kms-key")],
4087 &target,
4088 )
4089 .expect_err("unmatched kms target should fail");
4090
4091 assert!(error.contains("must exactly match the remote destination"));
4092 }
4093
4094 #[test]
4095 fn test_cp_output_serialization() {
4096 let output = CpOutput {
4097 status: "success",
4098 source: "src/file.txt".to_string(),
4099 target: "dst/file.txt".to_string(),
4100 size_bytes: Some(1024),
4101 size_human: Some("1 KiB".to_string()),
4102 version_id: Some("destination-v2".to_string()),
4103 source_version_id: Some("source-v1".to_string()),
4104 };
4105 let json = serde_json::to_string(&output).unwrap();
4106 assert!(json.contains("\"status\":\"success\""));
4107 assert!(json.contains("\"size_bytes\":1024"));
4108 assert!(json.contains("\"version_id\":\"destination-v2\""));
4109 assert!(json.contains("\"source_version_id\":\"source-v1\""));
4110 }
4111
4112 #[test]
4113 fn test_cp_output_skips_none_fields() {
4114 let output = CpOutput {
4115 status: "success",
4116 source: "src".to_string(),
4117 target: "dst".to_string(),
4118 size_bytes: None,
4119 size_human: None,
4120 version_id: None,
4121 source_version_id: None,
4122 };
4123 let json = serde_json::to_string(&output).unwrap();
4124 assert!(!json.contains("size_bytes"));
4125 assert!(!json.contains("size_human"));
4126 assert!(!json.contains("version_id"));
4127 }
4128
4129 #[test]
4130 fn versioned_copy_output_uses_v3_and_preserves_both_version_ids() {
4131 let envelope = V3SuccessEnvelope::versioned_objects(VersionCopyData {
4132 operation: "copy",
4133 source: "src/object.txt".to_string(),
4134 target: "dst/object.txt".to_string(),
4135 source_version_id: Some("source-v1".to_string()),
4136 version_id: Some("destination-v2".to_string()),
4137 size_bytes: Some(1024),
4138 size_human: Some("1 KiB".to_string()),
4139 });
4140
4141 let json = serde_json::to_value(envelope).expect("serialize versioned copy output");
4142 assert_eq!(json["schema_version"], 3);
4143 assert_eq!(json["type"], "versioned_objects");
4144 assert_eq!(json["data"]["operation"], "copy");
4145 assert_eq!(json["data"]["source_version_id"], "source-v1");
4146 assert_eq!(json["data"]["version_id"], "destination-v2");
4147 }
4148
4149 #[tokio::test]
4150 async fn storage_class_validation_has_usage_and_unsupported_exit_codes() {
4151 for (storage_class, expected) in [
4152 ("not-a-class", ExitCode::UsageError),
4153 ("STANDARD_IA", ExitCode::UnsupportedFeature),
4154 ] {
4155 let mut args = CpArgs::single("source.txt", "local/bucket/target.txt");
4156 args.storage_class = Some(storage_class.to_string());
4157
4158 assert_eq!(execute(args, OutputConfig::default()).await, expected);
4159 }
4160 }
4161
4162 #[test]
4163 fn fidelity_direction_preflight_rejects_silent_or_beta10_unsupported_paths() {
4164 let local = ParsedPath::Local(PathBuf::from("report.json"));
4165 let source = ParsedPath::Remote(RemotePath::new("test", "source", "report.json"));
4166 let target = ParsedPath::Remote(RemotePath::new("test", "target", "report.json"));
4167
4168 let mut preserve_upload = CpArgs::single("report.json", "test/target/report.json");
4169 preserve_upload.preserve = true;
4170 assert!(matches!(
4171 validate_fidelity_directions(&preserve_upload, std::slice::from_ref(&local), &target),
4172 Err(Error::InvalidPath(_))
4173 ));
4174
4175 let mut replace = CpArgs::single("test/source/report.json", "test/target/report.json");
4176 replace.metadata_directive = Some(MetadataDirectiveArg::Replace);
4177 assert!(matches!(
4178 validate_fidelity_directions(&replace, std::slice::from_ref(&source), &target),
4179 Err(Error::UnsupportedFeature(_))
4180 ));
4181
4182 let cross_alias_source =
4183 ParsedPath::Remote(RemotePath::new("source", "source", "report.json"));
4184 let cross_alias_target =
4185 ParsedPath::Remote(RemotePath::new("destination", "target", "report.json"));
4186 assert!(
4187 validate_fidelity_directions(
4188 &replace,
4189 std::slice::from_ref(&cross_alias_source),
4190 &cross_alias_target,
4191 )
4192 .is_ok(),
4193 "cross-alias metadata REPLACE uses the upload path"
4194 );
4195
4196 let mut tags = CpArgs::single("test/source/report.json", "test/target/report.json");
4197 tags.tagging_directive = Some(TaggingDirectiveArg::Replace);
4198 tags.fidelity.tags = vec!["env=prod".to_string()];
4199 assert!(matches!(
4200 validate_fidelity_directions(&tags, std::slice::from_ref(&source), &target),
4201 Err(Error::UnsupportedFeature(_))
4202 ));
4203
4204 let mut checksum = CpArgs::single("test/source/report.json", "test/target/report.json");
4205 checksum.fidelity.checksum = Some("sha256".to_string());
4206 assert!(matches!(
4207 validate_fidelity_directions(&checksum, std::slice::from_ref(&source), &target),
4208 Err(Error::UnsupportedFeature(_))
4209 ));
4210 }
4211
4212 #[test]
4213 fn source_identity_validation_rejects_same_size_etag_changes() {
4214 let mut planned = ObjectInfo::file("report.json", 4);
4215 planned.etag = Some("planned".to_string());
4216 let mut current = planned.clone();
4217 assert!(source_identity_matches(&planned, ¤t));
4218
4219 current.etag = Some("changed".to_string());
4220 assert!(!source_identity_matches(&planned, ¤t));
4221 current.etag = None;
4222 assert!(!source_identity_matches(&planned, ¤t));
4223
4224 planned.etag = None;
4225 assert!(!source_identity_matches(&planned, ¤t));
4226 }
4227
4228 #[test]
4229 fn preserve_builds_explicit_metadata_copy_without_replacement_payload() {
4230 let mut args = CpArgs::single("test/source/report.json", "test/target/report.json");
4231 args.preserve = true;
4232 args.fidelity.retention_mode = Some("GOVERNANCE".to_string());
4233 args.fidelity.retain_until = Some("2099-01-02T03:04:05Z".to_string());
4234 args.fidelity.legal_hold = Some("ON".to_string());
4235
4236 let options =
4237 transfer_copy_options(&args, Some("source-v1".to_string()), None).expect("copy policy");
4238 assert_eq!(options.metadata_directive, Some(MetadataDirective::Copy));
4239 assert!(options.destination.attributes.is_none());
4240 assert_eq!(options.source.version_id.as_deref(), Some("source-v1"));
4241 assert!(options.destination.retention.is_some());
4242 assert_eq!(
4243 options.destination.legal_hold,
4244 Some(rc_core::LegalHoldStatus::On)
4245 );
4246 }
4247
4248 #[test]
4249 fn storage_class_plan_rejects_multipart_without_mutation() {
4250 let plan = TransferPlan::build(
4251 vec![TransferCandidate {
4252 payload: CpOperation::RemoteToRemote {
4253 source: RemotePath::new("local", "source", "large.bin"),
4254 target: RemotePath::new("local", "target", "large.bin"),
4255 source_info: Box::new(ObjectInfo::file(
4256 "large.bin",
4257 (MAX_SINGLE_COPY_SIZE + 1) as i64,
4258 )),
4259 encryption: None,
4260 },
4261 source: "local/source/large.bin".to_string(),
4262 target: "local/target/large.bin".to_string(),
4263 relative_path: "large.bin".to_string(),
4264 modified: None,
4265 size_bytes: Some(MAX_SINGLE_COPY_SIZE + 1),
4266 }],
4267 &TransferSelection::default(),
4268 );
4269
4270 assert!(matches!(
4271 validate_storage_class_plan(&plan, Some("STANDARD")),
4272 Err(Error::UnsupportedFeature(_))
4273 ));
4274 }
4275
4276 #[test]
4277 fn storage_class_plan_rejects_cross_alias_multipart_uploads() {
4278 let plan = TransferPlan::build(
4279 vec![TransferCandidate {
4280 payload: CpOperation::RemoteToRemote {
4281 source: RemotePath::new("alpha", "source", "medium.bin"),
4282 target: RemotePath::new("beta", "target", "medium.bin"),
4283 source_info: Box::new(ObjectInfo::file(
4284 "medium.bin",
4285 (MULTIPART_THRESHOLD + 1) as i64,
4286 )),
4287 encryption: None,
4288 },
4289 source: "alpha/source/medium.bin".to_string(),
4290 target: "beta/target/medium.bin".to_string(),
4291 relative_path: "medium.bin".to_string(),
4292 modified: None,
4293 size_bytes: Some(MULTIPART_THRESHOLD + 1),
4294 }],
4295 &TransferSelection::default(),
4296 );
4297
4298 assert!(matches!(
4299 validate_storage_class_plan(&plan, Some("STANDARD")),
4300 Err(Error::UnsupportedFeature(_))
4301 ));
4302 }
4303
4304 #[test]
4305 fn piped_copy_preserves_source_content_type_unless_replaced() {
4306 let mut source = ObjectInfo::file("file.txt", 4);
4307 source.content_type = Some("text/plain".to_string());
4308 let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4309
4310 let copied = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4311 assert_eq!(
4312 copied
4313 .attributes
4314 .as_ref()
4315 .and_then(|value| value.content_type.as_deref()),
4316 Some("text/plain")
4317 );
4318
4319 let mut replace = args;
4320 replace.metadata_directive = Some(MetadataDirectiveArg::Replace);
4321 let replaced = piped_copy_write_options(&replace, &source, None).expect("replace metadata");
4322 assert!(
4323 replaced
4324 .attributes
4325 .as_ref()
4326 .is_none_or(|value| value.content_type.is_none())
4327 );
4328 }
4329
4330 #[test]
4331 fn piped_copy_records_the_source_etag_as_identity_metadata() {
4332 let mut source = ObjectInfo::file("file.txt", 4);
4333 source.etag = Some("source-etag".to_string());
4334 let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4335
4336 let options = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4337
4338 let attributes = options.attributes.as_ref().expect("identity attributes");
4339 assert_eq!(
4340 super::super::object_identity::identity_etag_from_metadata(Some(
4341 &attributes.user_metadata
4342 ))
4343 .as_deref(),
4344 Some("source-etag"),
4345 "a later mirror --compare auto must be able to skip this object"
4346 );
4347 }
4348
4349 #[test]
4350 fn piped_copy_records_identity_even_when_metadata_is_replaced() {
4351 let mut source = ObjectInfo::file("file.txt", 4);
4352 source.etag = Some("source-etag".to_string());
4353 source.metadata = Some(HashMap::from([(
4354 "owner".to_string(),
4355 "storage".to_string(),
4356 )]));
4357 let mut args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4358 args.metadata_directive = Some(MetadataDirectiveArg::Replace);
4359
4360 let options = piped_copy_write_options(&args, &source, None).expect("replace metadata");
4361
4362 let attributes = options.attributes.as_ref().expect("identity attributes");
4363 assert_eq!(
4364 super::super::object_identity::identity_etag_from_metadata(Some(
4365 &attributes.user_metadata
4366 ))
4367 .as_deref(),
4368 Some("source-etag"),
4369 "identity is rc bookkeeping, not user metadata"
4370 );
4371 assert!(
4372 !attributes.user_metadata.contains_key("owner"),
4373 "replace must still drop source user metadata"
4374 );
4375 }
4376
4377 #[test]
4378 fn piped_copy_omits_identity_when_the_source_has_no_etag() {
4379 let source = ObjectInfo::file("file.txt", 4);
4380 let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4381
4382 let options = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4383
4384 assert!(
4385 options
4386 .attributes
4387 .as_ref()
4388 .is_none_or(|attributes| attributes.user_metadata.is_empty()),
4389 "without a source ETag there is no identity to record"
4390 );
4391 }
4392
4393 #[test]
4394 fn piped_copy_preserves_source_user_metadata_unless_replaced() {
4395 let mut source = ObjectInfo::file("file.txt", 4);
4396 source.metadata = Some(HashMap::from([(
4397 "owner".to_string(),
4398 "storage".to_string(),
4399 )]));
4400 let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
4401
4402 let copied = piped_copy_write_options(&args, &source, None).expect("copy metadata");
4403 assert_eq!(
4404 copied
4405 .attributes
4406 .as_ref()
4407 .and_then(|value| value.user_metadata.get("owner"))
4408 .map(String::as_str),
4409 Some("storage")
4410 );
4411
4412 let mut replace = args;
4413 replace.metadata_directive = Some(MetadataDirectiveArg::Replace);
4414 let replaced = piped_copy_write_options(&replace, &source, None).expect("replace metadata");
4415 assert!(
4416 replaced
4417 .attributes
4418 .as_ref()
4419 .is_none_or(|value| value.user_metadata.is_empty())
4420 );
4421 }
4422
4423 #[test]
4424 fn get_alias_accepts_only_one_remote_source_and_local_target() {
4425 let remote = ParsedPath::Remote(RemotePath::new("local", "reports", "report.json"));
4426 let other_remote = ParsedPath::Remote(RemotePath::new("local", "reports", "other.json"));
4427 let local = ParsedPath::Local(PathBuf::from("./report.json"));
4428
4429 assert_eq!(
4430 validate_alias_direction(TransferAlias::Get, std::slice::from_ref(&remote), &local),
4431 Ok(())
4432 );
4433 assert_eq!(
4434 validate_alias_direction(TransferAlias::Get, &[remote.clone(), other_remote], &local),
4435 Err("get requires exactly one remote source and one local target")
4436 );
4437 assert_eq!(
4438 validate_alias_direction(TransferAlias::Get, std::slice::from_ref(&local), &remote),
4439 Err("get requires exactly one remote source and one local target")
4440 );
4441 }
4442
4443 #[test]
4444 fn put_alias_accepts_only_local_sources_and_remote_target() {
4445 let first_local = ParsedPath::Local(PathBuf::from("./january.csv"));
4446 let second_local = ParsedPath::Local(PathBuf::from("./february.csv"));
4447 let remote = ParsedPath::Remote(RemotePath::new("local", "reports", ""));
4448
4449 assert_eq!(
4450 validate_alias_direction(
4451 TransferAlias::Put,
4452 &[first_local.clone(), second_local],
4453 &remote
4454 ),
4455 Ok(())
4456 );
4457 assert_eq!(
4458 validate_alias_direction(
4459 TransferAlias::Put,
4460 std::slice::from_ref(&remote),
4461 &first_local
4462 ),
4463 Err("put requires one or more local sources and one remote target")
4464 );
4465 }
4466}