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