diff --git a/Cargo.lock b/Cargo.lock index f8616017..09e52c4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1662,6 +1662,12 @@ dependencies = [ "libm", ] +[[package]] +name = "humantime" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15" + [[package]] name = "hybrid-array" version = "0.4.13" @@ -2614,6 +2620,7 @@ dependencies = [ "http 1.4.2", "http-body 1.0.1", "http-body-util", + "humantime", "jiff", "mockall", "quick-xml", @@ -2824,6 +2831,7 @@ dependencies = [ "console", "glob", "humansize", + "humantime", "indicatif", "insta", "jiff", diff --git a/Cargo.toml b/Cargo.toml index d424d3df..c37d8eb7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -56,6 +56,7 @@ console = "0.16" dirs = "6.0" jiff = { version = "0.2", features = ["serde"] } humansize = "2.1" +humantime = "2.3" bytes = "1.11" url = "2.5" futures = "0.3" diff --git a/crates/cli/Cargo.toml b/crates/cli/Cargo.toml index f65367be..23ab24ea 100644 --- a/crates/cli/Cargo.toml +++ b/crates/cli/Cargo.toml @@ -48,6 +48,7 @@ console.workspace = true # Utilities jiff.workspace = true humansize.workspace = true +humantime.workspace = true mime_guess.workspace = true glob.workspace = true shlex.workspace = true diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index 34606ca8..d39c1bdd 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -928,6 +928,33 @@ mod tests { } } + #[test] + fn cli_accepts_bucket_replication_diff_prefix() { + let cli = Cli::try_parse_from([ + "rc", + "bucket", + "replication", + "diff", + "local/my-bucket", + "--prefix", + "reports/2026/", + ]) + .expect("parse bucket replication diff"); + + match cli.command { + Commands::Bucket(args) => match args.command { + bucket::BucketCommands::Replication(replicate::ReplicateArgs { + command: replicate::ReplicateCommands::Diff(arg), + }) => { + assert_eq!(arg.path, "local/my-bucket"); + assert_eq!(arg.prefix.as_deref(), Some("reports/2026/")); + } + other => panic!("expected bucket replication diff command, got {:?}", other), + }, + other => panic!("expected bucket command, got {:?}", other), + } + } + #[test] fn cli_accepts_bucket_replication_add_tls_flags() { let cli = Cli::try_parse_from([ @@ -957,6 +984,103 @@ mod tests { } } + #[test] + fn cli_accepts_bucket_replication_check_confirmation() { + let cli = Cli::try_parse_from([ + "rc", + "bucket", + "replication", + "check", + "local/my-bucket", + "--yes", + "--force", + ]) + .expect("parse bucket replication check"); + + match cli.command { + Commands::Bucket(args) => match args.command { + bucket::BucketCommands::Replication(replicate::ReplicateArgs { + command: replicate::ReplicateCommands::Check(arg), + }) => { + assert_eq!(arg.path, "local/my-bucket"); + assert!(arg.yes); + assert!(arg.force); + } + other => panic!("expected bucket replication check command, got {other:?}"), + }, + other => panic!("expected bucket command, got {other:?}"), + } + } + + #[test] + fn cli_accepts_bucket_replication_resync_lifecycle() { + let start = Cli::try_parse_from([ + "rc", + "bucket", + "replication", + "resync", + "start", + "local/my-bucket", + "--target-arn", + "arn:rustfs:replication::id:backup", + "--older-than", + "7d", + "--reset-id", + "caller-id", + "--yes", + ]) + .expect("parse bucket replication resync start"); + + match start.command { + Commands::Bucket(args) => match args.command { + bucket::BucketCommands::Replication(replicate::ReplicateArgs { + command: + replicate::ReplicateCommands::Resync(replicate::ResyncCommands::Start(arg)), + }) => { + assert_eq!(arg.path, "local/my-bucket"); + assert_eq!( + arg.target_arn.as_deref(), + Some("arn:rustfs:replication::id:backup") + ); + assert_eq!(arg.older_than.as_deref(), Some("7d")); + assert_eq!(arg.reset_id.as_deref(), Some("caller-id")); + assert!(arg.yes); + } + other => panic!("expected bucket replication resync start, got {other:?}"), + }, + other => panic!("expected bucket command, got {other:?}"), + } + + let status = Cli::try_parse_from([ + "rc", + "bucket", + "replication", + "resync", + "status", + "local/my-bucket", + "--target-arn", + "arn:rustfs:replication::id:backup", + ]) + .expect("parse bucket replication resync status"); + + match status.command { + Commands::Bucket(args) => match args.command { + bucket::BucketCommands::Replication(replicate::ReplicateArgs { + command: + replicate::ReplicateCommands::Resync(replicate::ResyncCommands::Status(arg)), + }) => { + assert_eq!(arg.path, "local/my-bucket"); + assert_eq!( + arg.target_arn.as_deref(), + Some("arn:rustfs:replication::id:backup") + ); + } + other => panic!("expected bucket replication resync status, got {other:?}"), + }, + other => panic!("expected bucket command, got {other:?}"), + } + } + #[test] fn cli_accepts_bucket_remove_subcommand() { let cli = Cli::try_parse_from(["rc", "bucket", "remove", "local/my-bucket"]) diff --git a/crates/cli/src/commands/replicate.rs b/crates/cli/src/commands/replicate.rs index 98ac215e..7f0a13fc 100644 --- a/crates/cli/src/commands/replicate.rs +++ b/crates/cli/src/commands/replicate.rs @@ -5,15 +5,16 @@ use clap::{Args, Subcommand}; use comfy_table::{Cell, Table}; -use rc_core::admin::AdminApi; +use rc_core::admin::{AdminApi, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry}; use rc_core::replication::{ BucketTarget, BucketTargetCredentials, ReplicationConfiguration, ReplicationDestination, - ReplicationRule, ReplicationRuleStatus, + ReplicationResyncStartOptions, ReplicationResyncStartResult, ReplicationResyncStatus, + ReplicationResyncTargetStatus, ReplicationRule, ReplicationRuleStatus, }; -use rc_core::{AliasManager, ObjectStore as _}; +use rc_core::{AliasManager, Error, ObjectStore as _}; use rc_s3::{AdminClient, S3Client}; use serde::{Deserialize, Serialize}; -use std::collections::{BTreeSet, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::path::{Path, PathBuf}; use crate::exit_code::ExitCode; @@ -60,6 +61,9 @@ pub enum ReplicateCommands { /// Show replication status/metrics for a bucket Status(BucketArg), + /// Scan for object versions that have not replicated + Diff(DiffArgs), + /// Remove replication rules from a bucket Remove(RemoveArgs), @@ -68,6 +72,76 @@ pub enum ReplicateCommands { /// Import replication configuration from a JSON file Import(ImportArgs), + + /// Actively validate configured replication targets + Check(CheckArgs), + + /// Manage bucket replication resync operations + #[command(subcommand)] + Resync(ResyncCommands), +} + +#[derive(Args, Debug)] +pub struct CheckArgs { + /// Source bucket path (ALIAS/BUCKET) + pub path: String, + + /// Confirm the active remote write/delete validation probe + #[arg(long)] + pub yes: bool, + + /// Force operation even if generic replication capability detection fails + #[arg(long)] + pub force: bool, +} + +#[derive(Subcommand, Debug)] +pub enum ResyncCommands { + /// Start resyncing previously replicated objects + Start(ResyncStartArgs), + + /// Show persisted server-side resync status + Status(ResyncStatusArgs), +} + +#[derive(Args, Debug)] +pub struct ResyncStartArgs { + /// Source bucket path (ALIAS/BUCKET) + pub path: String, + + /// Select a configured replication target ARN + #[arg(long)] + pub target_arn: Option, + + /// Only resync objects older than this duration (for example, 7d10h31s) + #[arg(long)] + pub older_than: Option, + + /// Supply a stable resync identifier instead of using a server-generated ID + #[arg(long)] + pub reset_id: Option, + + /// Confirm the active server-side resync + #[arg(long)] + pub yes: bool, + + /// Force operation even if generic replication capability detection fails + #[arg(long)] + pub force: bool, +} + +#[derive(Args, Debug)] +pub struct ResyncStatusArgs { + /// Source bucket path (ALIAS/BUCKET) + pub path: String, + + /// Filter status to one configured replication target ARN + #[arg(long)] + pub target_arn: Option, + + /// Force operation even if generic replication capability detection fails + #[arg(long)] + pub force: bool, } #[derive(Args, Debug)] @@ -80,6 +154,16 @@ pub struct BucketArg { pub force: bool, } +#[derive(Args, Debug)] +pub struct DiffArgs { + /// Source bucket path (ALIAS/BUCKET) + pub path: String, + + /// Limit the scan to object keys below this prefix + #[arg(long)] + pub prefix: Option, +} + #[derive(Args, Debug)] #[command(after_help = REPLICATE_ADD_AFTER_HELP)] pub struct AddArgs { @@ -251,6 +335,41 @@ struct ReplicateOperationOutput { action: String, } +#[derive(Debug, Serialize)] +struct ReplicationV3Output { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + data: ReplicationV3Data, +} + +#[derive(Debug, Serialize)] +struct ReplicationV3Data { + operation: &'static str, + bucket: String, + valid: Option, + targets: Vec, +} + +#[derive(Debug, Serialize)] +struct ReplicationV3Target { + target_arn: String, + reset_id: String, + reset_before: Option, + started_at: Option, + last_updated_at: Option, + state: Option, + server_state: Option, + replicated_count: Option, + replicated_size: Option, + failed_count: Option, + failed_size: Option, + current_bucket: Option, + current_object: Option, + error: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] struct ReplicationExport { @@ -266,6 +385,79 @@ struct ReplicationTargetTlsSettings { ca_cert_pem: Option, } +#[derive(Debug, Serialize)] +struct ReplicationDiffSuccessOutput { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + data: ReplicationDiffData, +} + +#[derive(Debug, Serialize)] +struct ReplicationDiffData { + operation: &'static str, + bucket: String, + prefix: Option, + entries: Vec, + scan: ReplicationDiffScanOutput, + extensions: BTreeMap, +} + +#[derive(Debug, Serialize)] +struct ReplicationDiffEntryOutput { + object: String, + version_id: Option, + delete_marker: bool, + size_bytes: u64, + replication_status: String, + last_modified: Option, + extensions: BTreeMap, +} + +#[derive(Debug, Serialize)] +struct ReplicationDiffScanOutput { + scanned_versions: usize, + truncated: bool, + resumable: bool, +} + +#[derive(Debug, Serialize)] +struct ReplicationDiffErrorOutput { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + error: ReplicationDiffError, +} + +#[derive(Debug, Serialize)] +#[serde(untagged)] +enum ReplicationDiffError { + Unsupported(ReplicationDiffUnsupportedError), + Standard(ReplicationDiffStandardError), +} + +#[derive(Debug, Serialize)] +struct ReplicationDiffUnsupportedError { + #[serde(rename = "type")] + error_type: &'static str, + message: String, + retryable: bool, + capability: &'static str, + server: Option, + suggestion: Option<&'static str>, +} + +#[derive(Debug, Serialize)] +struct ReplicationDiffStandardError { + #[serde(rename = "type")] + error_type: &'static str, + message: String, + retryable: bool, + suggestion: Option<&'static str>, +} + // ==================== execute ==================== /// Execute the replicate command @@ -275,9 +467,263 @@ pub async fn execute(args: ReplicateArgs, output_config: OutputConfig) -> ExitCo ReplicateCommands::Update(args) => execute_update(args, output_config).await, ReplicateCommands::List(args) => execute_list(args, output_config).await, ReplicateCommands::Status(args) => execute_status(args, output_config).await, + ReplicateCommands::Diff(args) => execute_diff(args, output_config).await, ReplicateCommands::Remove(args) => execute_remove(args, output_config).await, ReplicateCommands::Export(args) => execute_export(args, output_config).await, ReplicateCommands::Import(args) => execute_import(args, output_config).await, + ReplicateCommands::Check(args) => execute_check(args, output_config).await, + ReplicateCommands::Resync(ResyncCommands::Start(args)) => { + execute_resync_start(args, output_config).await + } + ReplicateCommands::Resync(ResyncCommands::Status(args)) => { + execute_resync_status(args, output_config).await + } + } +} + +// ==================== Diff ==================== + +async fn execute_diff(args: DiffArgs, output_config: OutputConfig) -> ExitCode { + let formatter = Formatter::new(output_config); + let (alias_name, bucket) = match parse_bucket_path(&args.path) { + Ok(parts) => parts, + Err(error) => { + return formatter.fail_with_suggestion( + ExitCode::UsageError, + &error, + "Use a bucket path in the form alias/bucket.", + ); + } + }; + let client = match setup_admin_client(&alias_name, &formatter) { + Ok(client) => client, + Err(code) => return code, + }; + + execute_diff_with_api(&bucket, args.prefix, &client, &formatter).await +} + +async fn execute_diff_with_api( + bucket: &str, + prefix: Option, + api: &dyn ReplicationDiffApi, + formatter: &Formatter, +) -> ExitCode { + match api.replication_diff(bucket, prefix.as_deref()).await { + Ok(diff) => { + if formatter.is_json() { + formatter.json(&replication_diff_output(bucket, prefix, diff)); + } else { + for line in replication_diff_lines(bucket, prefix.as_deref(), &diff, formatter) { + formatter.println(&line); + } + } + ExitCode::Success + } + Err(error) => emit_replication_diff_error(&error, formatter), + } +} + +fn replication_diff_output( + bucket: &str, + prefix: Option, + mut diff: ReplicationDiff, +) -> ReplicationDiffSuccessOutput { + sort_replication_diff_entries(&mut diff.entries); + ReplicationDiffSuccessOutput { + schema_version: 3, + output_type: "replication", + status: "success", + data: ReplicationDiffData { + operation: "diff", + bucket: bucket.to_string(), + prefix, + entries: diff + .entries + .into_iter() + .map(|entry| ReplicationDiffEntryOutput { + object: entry.object, + version_id: entry.version_id, + delete_marker: entry.delete_marker, + size_bytes: entry.size_bytes, + replication_status: entry.replication_status, + last_modified: entry.last_modified, + extensions: entry.extra, + }) + .collect(), + scan: ReplicationDiffScanOutput { + scanned_versions: diff.scanned_versions, + truncated: diff.is_truncated, + resumable: false, + }, + extensions: diff.extra, + }, + } +} + +fn replication_diff_lines( + bucket: &str, + prefix: Option<&str>, + diff: &ReplicationDiff, + formatter: &Formatter, +) -> Vec { + let bucket = formatter.sanitize_text(bucket); + let scope = match prefix { + Some(prefix) => format!("{bucket} (prefix: {})", formatter.sanitize_text(prefix)), + None => bucket, + }; + let mut entries = diff.entries.clone(); + sort_replication_diff_entries(&mut entries); + let mut lines = vec![format!("Replication diff for '{scope}':")]; + + if diff.is_truncated { + lines.push(format!( + "Partial scan: inspected {} versions; results are non-resumable. Narrow --prefix and run again for a more complete view.", + diff.scanned_versions + )); + } else { + lines.push(format!( + "Complete scan: inspected {} versions.", + diff.scanned_versions + )); + } + + if entries.is_empty() { + lines.push(if diff.is_truncated { + "No pending or failed versions were found in this partial scan; this does not prove the bucket has no replication backlog.".to_string() + } else { + "No pending or failed versions found.".to_string() + }); + return lines; + } + + lines.push(String::new()); + lines.push("OBJECT VERSION TYPE SIZE STATUS LAST MODIFIED".to_string()); + for entry in entries { + lines.push(format!( + "{} {} {} {} {} {}", + formatter.sanitize_text(&entry.object), + formatter.sanitize_text(entry.version_id.as_deref().unwrap_or("-")), + if entry.delete_marker { + "delete-marker" + } else { + "object" + }, + entry.size_bytes, + formatter.sanitize_text(&entry.replication_status), + formatter.sanitize_text( + entry + .last_modified + .as_ref() + .map(ToString::to_string) + .as_deref() + .unwrap_or("-") + ) + )); + } + lines +} + +fn sort_replication_diff_entries(entries: &mut [ReplicationDiffEntry]) { + entries.sort_by(|left, right| { + ( + &left.object, + &left.version_id, + left.delete_marker, + &left.replication_status, + &left.last_modified, + ) + .cmp(&( + &right.object, + &right.version_id, + right.delete_marker, + &right.replication_status, + &right.last_modified, + )) + }); +} + +fn emit_replication_diff_error(error: &Error, formatter: &Formatter) -> ExitCode { + let code = ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError); + let message = format!("Failed to scan replication diff: {error}"); + if formatter.is_json() { + formatter.json_error(&replication_diff_error_output(error, code, message)); + } else { + formatter.error_with_code(code, &message); + } + code +} + +fn replication_diff_error_output( + error: &Error, + code: ExitCode, + message: String, +) -> ReplicationDiffErrorOutput { + let error = if matches!(error, Error::UnsupportedFeature(_)) { + ReplicationDiffError::Unsupported(ReplicationDiffUnsupportedError { + error_type: "unsupported_feature", + message, + retryable: false, + capability: "replication_diff", + server: None, + suggestion: Some( + "Upgrade RustFS or verify that the replication diff route is enabled.", + ), + }) + } else { + let (error_type, retryable, suggestion) = replication_diff_error_metadata(code); + ReplicationDiffError::Standard(ReplicationDiffStandardError { + error_type, + message, + retryable, + suggestion, + }) + }; + ReplicationDiffErrorOutput { + schema_version: 3, + output_type: "replication", + status: "error", + error, + } +} + +const fn replication_diff_error_metadata( + code: ExitCode, +) -> (&'static str, bool, Option<&'static str>) { + match code { + ExitCode::UsageError => ( + "usage_error", + false, + Some("Verify the alias configuration and command arguments."), + ), + ExitCode::NetworkError => ( + "network_error", + true, + Some("Verify the endpoint and network connectivity, then retry."), + ), + ExitCode::AuthError => ( + "auth_error", + false, + Some("Verify the alias credentials and replication admin permission."), + ), + ExitCode::NotFound => ( + "not_found", + false, + Some("Verify the bucket name and that replication is configured."), + ), + ExitCode::Conflict => ( + "conflict", + false, + Some("Review the replication configuration and retry."), + ), + ExitCode::Interrupted => ( + "interrupted", + true, + Some("Run the diff again if the scan is still needed."), + ), + ExitCode::Success | ExitCode::GeneralError | ExitCode::UnsupportedFeature => { + ("general_error", false, None) + } } } @@ -1046,6 +1492,474 @@ async fn execute_import(args: ImportArgs, output_config: OutputConfig) -> ExitCo } } +// ==================== Target check and resync ==================== + +async fn execute_check(args: CheckArgs, output_config: OutputConfig) -> ExitCode { + let formatter = Formatter::new(output_config); + let (alias, bucket) = match validate_resync_inputs(&args.path, None, None, None) { + Ok(validated) => validated, + Err(error) => return fail_replication(&formatter, &error, "check"), + }; + if !args.yes { + return fail_replication( + &formatter, + &Error::InvalidPath( + "Replication check performs a remote write/delete probe; pass --yes to confirm" + .to_string(), + ), + "check", + ); + } + + let client = match setup_replication_operation_client(&alias, args.force).await { + Ok(client) => client, + Err(error) => return fail_replication(&formatter, &error, "check"), + }; + match client.check_bucket_replication(&bucket).await { + Ok(()) => { + output_replication_success(&formatter, "check", bucket.clone(), Some(true), Vec::new()); + if !formatter.is_json() { + let display_path = formatter.sanitize_text(&format!("{alias}/{bucket}")); + formatter.success(&format!( + "Replication targets for '{display_path}' passed the active validation probe." + )); + } + ExitCode::Success + } + Err(error) => fail_replication(&formatter, &error, "check"), + } +} + +async fn execute_resync_start(args: ResyncStartArgs, output_config: OutputConfig) -> ExitCode { + let formatter = Formatter::new(output_config); + let (alias, bucket) = match validate_resync_inputs( + &args.path, + args.target_arn.as_deref(), + args.reset_id.as_deref(), + args.older_than.as_deref(), + ) { + Ok(validated) => validated, + Err(error) => return fail_replication(&formatter, &error, "resync_start"), + }; + if !args.yes { + return fail_replication( + &formatter, + &Error::InvalidPath("Starting a replication resync requires --yes".to_string()), + "resync_start", + ); + } + let older_than = match args + .older_than + .as_deref() + .map(parse_resync_duration) + .transpose() + { + Ok(value) => value, + Err(error) => return fail_replication(&formatter, &error, "resync_start"), + }; + let client = match setup_replication_operation_client(&alias, args.force).await { + Ok(client) => client, + Err(error) => return fail_replication(&formatter, &error, "resync_start"), + }; + let options = ReplicationResyncStartOptions { + target_arn: args.target_arn, + older_than, + reset_id: args.reset_id, + }; + match client + .start_bucket_replication_resync(&bucket, options) + .await + { + Ok(result) => { + let target = start_target_output(&result); + output_replication_success( + &formatter, + "resync_start", + bucket.clone(), + None, + vec![target], + ); + if !formatter.is_json() { + let display_path = formatter.sanitize_text(&format!("{alias}/{bucket}")); + formatter.success(&format!( + "Replication resync started for '{display_path}' target '{}' with ID '{}'.", + formatter.sanitize_text(&result.target_arn), + formatter.sanitize_text(&result.reset_id) + )); + } + ExitCode::Success + } + Err(error) => fail_replication(&formatter, &error, "resync_start"), + } +} + +async fn execute_resync_status(args: ResyncStatusArgs, output_config: OutputConfig) -> ExitCode { + let formatter = Formatter::new(output_config); + let (alias, bucket) = + match validate_resync_inputs(&args.path, args.target_arn.as_deref(), None, None) { + Ok(validated) => validated, + Err(error) => return fail_replication(&formatter, &error, "resync_status"), + }; + let client = match setup_replication_operation_client(&alias, args.force).await { + Ok(client) => client, + Err(error) => return fail_replication(&formatter, &error, "resync_status"), + }; + match client + .bucket_replication_resync_status(&bucket, args.target_arn.as_deref()) + .await + { + Ok(status) => { + output_resync_status(&formatter, &alias, &bucket, status); + ExitCode::Success + } + Err(error) => fail_replication(&formatter, &error, "resync_status"), + } +} + +fn output_resync_status( + formatter: &Formatter, + alias: &str, + bucket: &str, + status: ReplicationResyncStatus, +) { + let targets = status + .targets + .iter() + .cloned() + .map(status_target_output) + .collect::>(); + output_replication_success( + formatter, + "resync_status", + bucket.to_string(), + None, + targets, + ); + if formatter.is_json() { + return; + } + if status.targets.is_empty() { + formatter.println("No replication resync status is available."); + return; + } + let display_path = formatter.sanitize_text(&format!("{alias}/{bucket}")); + formatter.println(&format!("Replication resync status for '{display_path}':")); + for target in status.targets { + let has_error = target.error.is_some(); + let lines = resync_status_human_lines(formatter, &target); + let last_index = lines.len().saturating_sub(1); + for (index, line) in lines.into_iter().enumerate() { + if has_error && index == last_index { + formatter.warning(&line); + } else { + formatter.println(&line); + } + } + } +} + +fn resync_status_human_lines( + formatter: &Formatter, + target: &ReplicationResyncTargetStatus, +) -> Vec { + let state = serde_json::to_value(&target.state) + .ok() + .and_then(|value| value.as_str().map(str::to_string)) + .unwrap_or_else(|| "unknown".to_string()); + let server_value = |value: &str| { + if value.is_empty() { + "".to_string() + } else { + formatter.sanitize_text(value) + } + }; + let timestamp = |value: Option<&jiff::Timestamp>| { + value + .map(|value| formatter.sanitize_text(&value.to_string())) + .unwrap_or_else(|| "".to_string()) + }; + let optional_value = |value: Option<&str>| { + value + .map(server_value) + .unwrap_or_else(|| "".to_string()) + }; + + let mut lines = vec![ + format!(" target ARN: {}", server_value(&target.target_arn)), + format!(" reset ID: {}", server_value(&target.reset_id)), + format!( + " reset before: {}", + timestamp(target.reset_before.as_ref()) + ), + format!(" started at: {}", timestamp(target.started_at.as_ref())), + format!( + " last update (server EndTime): {}", + timestamp(target.last_updated_at.as_ref()) + ), + format!(" state: {state}"), + format!(" server state: {}", server_value(&target.server_state)), + format!( + " replicated: {} objects ({} bytes)", + target.replicated_count, target.replicated_size + ), + format!( + " failed: {} objects ({} bytes)", + target.failed_count, target.failed_size + ), + format!( + " current bucket: {}", + optional_value(target.current_bucket.as_deref()) + ), + format!( + " current object: {}", + optional_value(target.current_object.as_deref()) + ), + ]; + if let Some(error) = target.error.as_deref() { + lines.push(format!(" server error detail: {}", server_value(error))); + } + lines +} + +fn output_replication_success( + formatter: &Formatter, + operation: &'static str, + bucket: String, + valid: Option, + targets: Vec, +) { + if formatter.is_json() { + formatter.json(&ReplicationV3Output { + schema_version: 3, + output_type: "replication_operations", + status: "success", + data: ReplicationV3Data { + operation, + bucket, + valid, + targets, + }, + }); + } +} + +fn start_target_output(result: &ReplicationResyncStartResult) -> ReplicationV3Target { + ReplicationV3Target { + target_arn: result.target_arn.clone(), + reset_id: result.reset_id.clone(), + reset_before: None, + started_at: None, + last_updated_at: None, + state: None, + server_state: None, + replicated_count: None, + replicated_size: None, + failed_count: None, + failed_size: None, + current_bucket: None, + current_object: None, + error: None, + } +} + +fn status_target_output(target: ReplicationResyncTargetStatus) -> ReplicationV3Target { + ReplicationV3Target { + target_arn: target.target_arn, + reset_id: target.reset_id, + reset_before: target.reset_before, + started_at: target.started_at, + last_updated_at: target.last_updated_at, + state: Some(target.state), + server_state: Some(target.server_state), + replicated_count: Some(target.replicated_count), + replicated_size: Some(target.replicated_size), + failed_count: Some(target.failed_count), + failed_size: Some(target.failed_size), + current_bucket: target.current_bucket, + current_object: target.current_object, + error: target.error, + } +} + +fn fail_replication(formatter: &Formatter, error: &Error, operation: &str) -> ExitCode { + let code = exit_code_from_replication_error(error); + let message = format!("Replication {operation} failed: {error}"); + if formatter.is_json() { + formatter.json_error(&replication_error_json(error, operation)); + code + } else { + formatter.fail(code, &message) + } +} + +fn replication_error_json(error: &Error, operation: &str) -> serde_json::Value { + let code = exit_code_from_replication_error(error); + let message = format!("Replication {operation} failed: {error}"); + let (error_type, retryable, suggestion) = replication_error_metadata(code); + let error_value = if code == ExitCode::UnsupportedFeature { + let capability = if operation == "check" { + "bucket_replication_check" + } else { + "bucket_replication_resync" + }; + serde_json::json!({ + "type": "unsupported_feature", + "message": message, + "retryable": false, + "capability": capability, + "server": null, + "suggestion": suggestion, + }) + } else { + serde_json::json!({ + "type": error_type, + "message": message, + "retryable": retryable, + "suggestion": suggestion, + }) + }; + serde_json::json!({ + "schema_version": 3, + "type": "replication_operations", + "status": "error", + "error": error_value, + }) +} + +fn exit_code_from_replication_error(error: &Error) -> ExitCode { + match error { + Error::InvalidPath(_) | Error::Config(_) => ExitCode::UsageError, + Error::Network(_) => ExitCode::NetworkError, + Error::Auth(_) => ExitCode::AuthError, + Error::NotFound(_) | Error::AliasNotFound(_) => ExitCode::NotFound, + Error::Conflict(_) | Error::AliasExists(_) => ExitCode::Conflict, + Error::UnsupportedFeature(_) => ExitCode::UnsupportedFeature, + _ => ExitCode::GeneralError, + } +} + +const fn replication_error_metadata(code: ExitCode) -> (&'static str, bool, Option<&'static str>) { + match code { + ExitCode::UsageError => ( + "usage_error", + false, + Some("Review the command arguments and required --yes confirmation."), + ), + ExitCode::NetworkError => ( + "network_error", + true, + Some("Verify endpoint connectivity and retry."), + ), + ExitCode::AuthError => ( + "auth_error", + false, + Some("Verify S3 replication permissions and alias credentials."), + ), + ExitCode::NotFound => ( + "not_found", + false, + Some("Verify the bucket and replication configuration."), + ), + ExitCode::Conflict => ( + "conflict", + false, + Some("Review the replication target configuration and server error."), + ), + ExitCode::Interrupted => ("interrupted", true, Some("Retry the operation if needed.")), + ExitCode::Success | ExitCode::GeneralError | ExitCode::UnsupportedFeature => { + ("general_error", false, None) + } + } +} + +fn validate_resync_inputs( + path: &str, + target_arn: Option<&str>, + reset_id: Option<&str>, + older_than: Option<&str>, +) -> Result<(String, String), Error> { + let (alias, bucket) = parse_bucket_path(path).map_err(Error::InvalidPath)?; + validate_s3_bucket_name(&bucket)?; + if let Some(target_arn) = target_arn { + validate_target_arn(target_arn)?; + } + if let Some(reset_id) = reset_id { + validate_reset_id(reset_id)?; + } + if let Some(older_than) = older_than { + parse_resync_duration(older_than)?; + } + Ok((alias, bucket)) +} + +fn validate_s3_bucket_name(bucket: &str) -> Result<(), Error> { + let valid_length = (3..=63).contains(&bucket.len()); + let valid_characters = bucket.bytes().all(|byte| { + byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'-' | b'.') + }); + let valid_labels = bucket + .split('.') + .all(|label| !label.is_empty() && !label.starts_with('-') && !label.ends_with('-')); + if !valid_length + || !valid_characters + || !valid_labels + || bucket.parse::().is_ok() + { + return Err(Error::InvalidPath(format!( + "Invalid S3 bucket name: {bucket}" + ))); + } + Ok(()) +} + +fn validate_target_arn(target_arn: &str) -> Result<(), Error> { + if target_arn.is_empty() + || target_arn.trim() != target_arn + || target_arn.len() > 2048 + || target_arn.chars().any(char::is_control) + || target_arn.chars().any(char::is_whitespace) + || !target_arn.starts_with("arn:") + || !target_arn.contains(":replication:") + { + return Err(Error::InvalidPath( + "--target-arn must be a bounded replication ARN without whitespace or control characters" + .to_string(), + )); + } + Ok(()) +} + +fn validate_reset_id(reset_id: &str) -> Result<(), Error> { + if reset_id.is_empty() + || reset_id.trim() != reset_id + || reset_id.len() > 256 + || reset_id.chars().any(char::is_control) + { + return Err(Error::InvalidPath( + "--reset-id must be 1 to 256 bytes without surrounding whitespace or control characters" + .to_string(), + )); + } + Ok(()) +} + +fn parse_resync_duration(value: &str) -> Result { + let duration = humantime::parse_duration(value).map_err(|error| { + Error::InvalidPath(format!("Invalid --older-than duration '{value}': {error}")) + })?; + if duration.is_zero() { + return Err(Error::InvalidPath( + "--older-than must be greater than zero".to_string(), + )); + } + if duration.as_secs() > i64::MAX as u64 { + return Err(Error::InvalidPath( + "--older-than exceeds the RustFS duration range".to_string(), + )); + } + Ok(duration) +} + // ==================== Helpers ==================== fn parse_bucket_path(path: &str) -> Result<(String, String), String> { @@ -1092,6 +2006,28 @@ fn resolve_alias(alias_name: &str, formatter: &Formatter) -> Result Result { + let alias_manager = AliasManager::new()?; + let alias = alias_manager.get(alias_name)?; + let client = S3Client::new(alias).await?; + + match client.capabilities().await { + Ok(capabilities) if !force && !capabilities.replication => { + return Err(Error::UnsupportedFeature( + "Backend does not advertise generic bucket replication support".to_string(), + )); + } + Ok(_) => {} + Err(_) if force => {} + Err(error) => return Err(error), + } + + Ok(client) +} + async fn setup_s3_client( alias_name: &str, bucket: &str, @@ -1381,9 +2317,66 @@ fn strip_endpoint_path(endpoint: &str) -> String { #[cfg(test)] mod tests { use super::*; + use async_trait::async_trait; use std::collections::HashMap; use tempfile::NamedTempFile; + enum StubDiffOutcome { + Success(ReplicationDiff), + Auth, + NotFound, + Unsupported, + Network, + General, + } + + struct StubReplicationDiffApi { + outcome: StubDiffOutcome, + } + + #[async_trait] + impl ReplicationDiffApi for StubReplicationDiffApi { + async fn replication_diff( + &self, + _bucket: &str, + _prefix: Option<&str>, + ) -> rc_core::Result { + match &self.outcome { + StubDiffOutcome::Success(diff) => Ok(diff.clone()), + StubDiffOutcome::Auth => Err(Error::Auth("Access denied".to_string())), + StubDiffOutcome::NotFound => { + Err(Error::NotFound("replication is not configured".to_string())) + } + StubDiffOutcome::Unsupported => { + Err(Error::UnsupportedFeature("route unavailable".to_string())) + } + StubDiffOutcome::Network => Err(Error::Network("connection reset".to_string())), + StubDiffOutcome::General => Err(Error::General("malformed response".to_string())), + } + } + } + + fn diff_with_entries(entries: Vec) -> ReplicationDiff { + ReplicationDiff { + entries, + is_truncated: false, + scanned_versions: 24, + extra: BTreeMap::new(), + } + } + + fn diff_entry(object: &str, version_id: Option<&str>) -> ReplicationDiffEntry { + ReplicationDiffEntry { + object: object.to_string(), + version_id: version_id.map(str::to_string), + size_bytes: 42, + delete_marker: false, + replication_status: "FAILED".to_string(), + last_modified: Some("2026-07-21T04:00:00Z".parse().expect("valid timestamp")), + extra: BTreeMap::new(), + } + } + #[test] fn test_parse_bucket_path_success() { let (alias, bucket) = parse_bucket_path("local/my-bucket").expect("should parse"); @@ -1404,6 +2397,107 @@ mod tests { assert!(parse_bucket_path("local/my-bucket/object.txt").is_err()); } + #[test] + fn replication_diff_json_is_deterministic_and_preserves_extensions() { + let mut second = diff_entry("reports/b.json", Some("v2")); + second.extra.insert( + "TargetDetail".to_string(), + serde_json::json!({ "attempts": 2 }), + ); + let mut diff = diff_with_entries(vec![second, diff_entry("reports/a.json", Some("v1"))]); + diff.extra + .insert("ServerRevision".to_string(), serde_json::json!(7)); + + let value = serde_json::to_value(replication_diff_output( + "source", + Some("reports/".to_string()), + diff, + )) + .expect("replication diff output should serialize"); + + assert_eq!(value["schema_version"], 3); + assert_eq!(value["type"], "replication"); + assert_eq!(value["data"]["operation"], "diff"); + assert_eq!(value["data"]["entries"][0]["object"], "reports/a.json"); + assert_eq!( + value["data"]["entries"][1]["extensions"]["TargetDetail"]["attempts"], + 2 + ); + assert_eq!(value["data"]["extensions"]["ServerRevision"], 7); + assert_eq!(value["data"]["scan"]["resumable"], false); + } + + #[test] + fn truncated_empty_human_output_is_partial_and_sanitized() { + let formatter = Formatter::new(OutputConfig { + no_color: true, + ..Default::default() + }); + let diff = ReplicationDiff { + entries: Vec::new(), + is_truncated: true, + scanned_versions: 10000, + extra: BTreeMap::new(), + }; + + let lines = replication_diff_lines( + "bucket\n\u{1b}[31m", + Some("archive\r\n2026/"), + &diff, + &formatter, + ); + + assert!(lines.iter().any(|line| line.contains("Partial scan"))); + assert!(lines.iter().any(|line| line.contains("non-resumable"))); + assert!(lines.iter().any(|line| line.contains("does not prove"))); + assert!(lines.iter().all(|line| !line.contains('\u{1b}'))); + assert!(lines.iter().all(|line| !line.contains('\r'))); + assert!(lines.iter().all(|line| !line.contains('\n'))); + } + + #[tokio::test] + async fn replication_diff_command_preserves_success_and_error_exit_codes() { + let formatter = Formatter::new(OutputConfig { + quiet: true, + ..Default::default() + }); + let cases = [ + ( + StubDiffOutcome::Success(diff_with_entries(Vec::new())), + ExitCode::Success, + ), + (StubDiffOutcome::Auth, ExitCode::AuthError), + (StubDiffOutcome::NotFound, ExitCode::NotFound), + (StubDiffOutcome::Unsupported, ExitCode::UnsupportedFeature), + (StubDiffOutcome::Network, ExitCode::NetworkError), + (StubDiffOutcome::General, ExitCode::GeneralError), + ]; + + for (outcome, expected) in cases { + let api = StubReplicationDiffApi { outcome }; + assert_eq!( + execute_diff_with_api("source", None, &api, &formatter).await, + expected + ); + } + } + + #[test] + fn replication_diff_unsupported_error_uses_specialized_v3_shape() { + let error = Error::UnsupportedFeature("route unavailable".to_string()); + let value = serde_json::to_value(replication_diff_error_output( + &error, + ExitCode::UnsupportedFeature, + format!("Failed to scan replication diff: {error}"), + )) + .expect("replication diff error should serialize"); + + assert_eq!(value["type"], "replication"); + assert_eq!(value["error"]["type"], "unsupported_feature"); + assert_eq!(value["error"]["capability"], "replication_diff"); + assert_eq!(value["error"]["server"], serde_json::Value::Null); + } + #[test] fn test_parse_replicate_flags_none() { let (d, dm, eo) = parse_replicate_flags(None); @@ -1760,4 +2854,193 @@ mod tests { let code = execute(args, OutputConfig::default()).await; assert_eq!(code, ExitCode::UsageError); } + + #[test] + fn resync_input_validation_is_local_and_conservative() { + assert!( + validate_resync_inputs( + "local/source-bucket", + Some("arn:rustfs:replication::id:target"), + Some("caller-reset-1"), + Some("7d10h31s") + ) + .is_ok() + ); + assert!(validate_resync_inputs("local/Bad_Bucket", None, None, None).is_err()); + assert!( + validate_resync_inputs( + "local/source-bucket", + Some("arn:rustfs:replication::id:target\n"), + None, + None + ) + .is_err() + ); + assert!( + validate_resync_inputs("local/source-bucket", None, Some(" reset-1"), None).is_err() + ); + } + + #[test] + fn resync_duration_accepts_large_values_and_rejects_server_overflow() { + assert_eq!( + parse_resync_duration("1h").expect("one hour"), + std::time::Duration::from_secs(3600) + ); + assert!(parse_resync_duration("1000y").is_ok()); + assert!(parse_resync_duration("0s").is_err()); + assert!(parse_resync_duration("-1h").is_err()); + assert!(parse_resync_duration("9223372036854775808s").is_err()); + } + + #[test] + fn replication_v3_status_output_preserves_server_state() { + let target = status_target_output(ReplicationResyncTargetStatus { + target_arn: "arn:rustfs:replication::id:backup".to_string(), + reset_id: "reset-1".to_string(), + reset_before: None, + started_at: None, + last_updated_at: None, + state: rc_core::ReplicationResyncState::Unknown, + server_state: "FutureState".to_string(), + replicated_count: 3, + replicated_size: 30, + failed_count: 2, + failed_size: 20, + current_bucket: Some("source-bucket".to_string()), + current_object: Some("last.txt".to_string()), + error: Some("target unavailable".to_string()), + }); + let value = serde_json::to_value(ReplicationV3Output { + schema_version: 3, + output_type: "replication_operations", + status: "success", + data: ReplicationV3Data { + operation: "resync_status", + bucket: "source-bucket".to_string(), + valid: None, + targets: vec![target], + }, + }) + .expect("serialize replication v3 output"); + + assert_eq!(value["schema_version"], 3); + assert_eq!(value["type"], "replication_operations"); + assert_eq!(value["data"]["targets"][0]["state"], "unknown"); + assert_eq!(value["data"]["targets"][0]["server_state"], "FutureState"); + } + + #[test] + fn replication_operation_errors_classify_auth_not_found_and_missing_routes() { + let cases = [ + ( + Error::Auth("access denied".to_string()), + ExitCode::AuthError, + "auth_error", + ), + ( + Error::NotFound("replication configuration missing".to_string()), + ExitCode::NotFound, + "not_found", + ), + ( + Error::UnsupportedFeature("extension route missing".to_string()), + ExitCode::UnsupportedFeature, + "unsupported_feature", + ), + ]; + + for (error, expected_code, expected_type) in cases { + assert_eq!(exit_code_from_replication_error(&error), expected_code); + let value = replication_error_json(&error, "resync_status"); + assert_eq!(value["schema_version"], 3); + assert_eq!(value["type"], "replication_operations"); + assert_eq!(value["status"], "error"); + assert_eq!(value["error"]["type"], expected_type); + } + + let unsupported = replication_error_json( + &Error::UnsupportedFeature("extension route missing".to_string()), + "check", + ); + assert_eq!( + unsupported["error"]["capability"], + "bucket_replication_check" + ); + } + + #[test] + fn resync_status_human_lines_retain_and_sanitize_server_fields() { + let formatter = Formatter::new(OutputConfig { + no_color: true, + ..OutputConfig::default() + }); + let target = ReplicationResyncTargetStatus { + target_arn: "arn:target\nspoof".to_string(), + reset_id: "reset\rid".to_string(), + reset_before: Some("2026-07-01T00:00:00Z".parse().expect("timestamp")), + started_at: Some("2026-07-02T00:00:00Z".parse().expect("timestamp")), + last_updated_at: Some("2026-07-02T00:01:00Z".parse().expect("timestamp")), + state: rc_core::ReplicationResyncState::Unknown, + server_state: "Future\tState".to_string(), + replicated_count: 3, + replicated_size: 30, + failed_count: 2, + failed_size: 20, + current_bucket: Some("source\nbucket".to_string()), + current_object: Some("last\r.txt".to_string()), + error: Some("target\tunavailable".to_string()), + }; + + let lines = resync_status_human_lines(&formatter, &target); + let output = lines.join("\n"); + for expected in [ + "arn:target\\nspoof", + "reset\\rid", + "reset before: 2026-07-01T00:00:00Z", + "started at: 2026-07-02T00:00:00Z", + "last update (server EndTime): 2026-07-02T00:01:00Z", + "server state: Future\\tState", + "current bucket: source\\nbucket", + "current object: last\\r.txt", + "server error detail: target\\tunavailable", + ] { + assert!(output.contains(expected), "missing human field: {expected}"); + } + assert!(!output.contains('\r')); + assert!(!output.contains('\t')); + } + + #[tokio::test] + async fn check_requires_confirmation_before_alias_or_network_setup() { + let args = ReplicateArgs { + command: ReplicateCommands::Check(CheckArgs { + path: "missing-alias/source-bucket".to_string(), + yes: false, + force: false, + }), + }; + + let code = execute(args, OutputConfig::default()).await; + + assert_eq!(code, ExitCode::UsageError); + } + + #[tokio::test] + async fn resync_start_requires_confirmation_before_alias_or_network_setup() { + let args = ReplicateArgs { + command: ReplicateCommands::Resync(ResyncCommands::Start(ResyncStartArgs { + path: "missing-alias/source-bucket".to_string(), + target_arn: None, + older_than: None, + reset_id: None, + yes: false, + force: false, + })), + }; + + let code = execute(args, OutputConfig::default()).await; + + assert_eq!(code, ExitCode::UsageError); + } } diff --git a/crates/cli/tests/fixtures/output_v3/replication/empty.json b/crates/cli/tests/fixtures/output_v3/replication/empty.json new file mode 100644 index 00000000..fae2261f --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication/empty.json @@ -0,0 +1,17 @@ +{ + "schema_version": 3, + "type": "replication", + "status": "success", + "data": { + "operation": "diff", + "bucket": "source", + "prefix": null, + "entries": [], + "scan": { + "scanned_versions": 0, + "truncated": false, + "resumable": false + }, + "extensions": {} + } +} diff --git a/crates/cli/tests/fixtures/output_v3/replication/error.json b/crates/cli/tests/fixtures/output_v3/replication/error.json new file mode 100644 index 00000000..db09923a --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication/error.json @@ -0,0 +1,11 @@ +{ + "schema_version": 3, + "type": "replication", + "status": "error", + "error": { + "type": "not_found", + "message": "Failed to scan replication diff: Not found: replication is not configured", + "retryable": false, + "suggestion": "Verify the bucket name and that replication is configured." + } +} diff --git a/crates/cli/tests/fixtures/output_v3/replication/success.json b/crates/cli/tests/fixtures/output_v3/replication/success.json new file mode 100644 index 00000000..9a865aa1 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication/success.json @@ -0,0 +1,31 @@ +{ + "schema_version": 3, + "type": "replication", + "status": "success", + "data": { + "operation": "diff", + "bucket": "source", + "prefix": "reports/", + "entries": [ + { + "object": "reports/a.json", + "version_id": "v1", + "delete_marker": false, + "size_bytes": 42, + "replication_status": "FAILED", + "last_modified": "2026-07-21T04:00:00Z", + "extensions": { + "TargetDetail": { "attempts": 2 } + } + } + ], + "scan": { + "scanned_versions": 24, + "truncated": false, + "resumable": false + }, + "extensions": { + "ServerRevision": 7 + } + } +} diff --git a/crates/cli/tests/fixtures/output_v3/replication/truncated.json b/crates/cli/tests/fixtures/output_v3/replication/truncated.json new file mode 100644 index 00000000..a4cea0db --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication/truncated.json @@ -0,0 +1,17 @@ +{ + "schema_version": 3, + "type": "replication", + "status": "success", + "data": { + "operation": "diff", + "bucket": "source", + "prefix": "archive/", + "entries": [], + "scan": { + "scanned_versions": 10000, + "truncated": true, + "resumable": false + }, + "extensions": {} + } +} diff --git a/crates/cli/tests/fixtures/output_v3/replication_operations/empty.json b/crates/cli/tests/fixtures/output_v3/replication_operations/empty.json new file mode 100644 index 00000000..2d65fce4 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication_operations/empty.json @@ -0,0 +1 @@ +{"schema_version":3,"type":"replication_operations","status":"success","data":{"operation":"resync_status","bucket":"archive","valid":null,"targets":[]}} diff --git a/crates/cli/tests/fixtures/output_v3/replication_operations/error.json b/crates/cli/tests/fixtures/output_v3/replication_operations/error.json new file mode 100644 index 00000000..c51bd0e7 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication_operations/error.json @@ -0,0 +1 @@ +{"schema_version":3,"type":"replication_operations","status":"error","error":{"type":"conflict","message":"Replication check failed: target versioning is disabled","retryable":false,"suggestion":"Review the replication target configuration and server error."}} diff --git a/crates/cli/tests/fixtures/output_v3/replication_operations/success.json b/crates/cli/tests/fixtures/output_v3/replication_operations/success.json new file mode 100644 index 00000000..5c4507bb --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication_operations/success.json @@ -0,0 +1 @@ +{"schema_version":3,"type":"replication_operations","status":"success","data":{"operation":"resync_status","bucket":"archive","valid":null,"targets":[{"target_arn":"arn:rustfs:replication::id:backup","reset_id":"","reset_before":"2026-07-01T00:00:00Z","started_at":"2026-07-02T00:00:00Z","last_updated_at":"2026-07-02T00:01:00Z","state":"failed","server_state":"Failed","replicated_count":3,"replicated_size":30,"failed_count":2,"failed_size":20,"current_bucket":"archive","current_object":"last.txt","error":"target unavailable"}]}} diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index 1a0db130..f0c6fac7 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -605,6 +605,40 @@ fn nested_subcommand_help_contract() { "The path is resolved on the CLI machine,", ], }, + HelpCase { + args: &["bucket", "replication", "check"], + usage: "Usage: rc bucket replication check [OPTIONS] ", + expected_tokens: &[ + "--yes", + "--force", + "active remote write/delete validation probe", + ], + }, + HelpCase { + args: &["bucket", "replication", "resync"], + usage: "Usage: rc bucket replication resync [OPTIONS] ", + expected_tokens: &["start", "status"], + }, + HelpCase { + args: &["bucket", "replication", "resync", "start"], + usage: "Usage: rc bucket replication resync start [OPTIONS] ", + expected_tokens: &[ + "--target-arn", + "--older-than", + "--reset-id", + "--yes", + "--force", + ], + }, + HelpCase { + args: &["bucket", "replication", "resync", "status"], + usage: "Usage: rc bucket replication resync status [OPTIONS] ", + expected_tokens: &[ + "--target-arn", + "--force", + "persisted server-side resync status", + ], + }, HelpCase { args: &["bucket", "event", "add"], usage: "Usage: rc bucket event add [OPTIONS] ", diff --git a/crates/cli/tests/output_schema_v3.rs b/crates/cli/tests/output_schema_v3.rs index 8572775c..f865ebef 100644 --- a/crates/cli/tests/output_schema_v3.rs +++ b/crates/cli/tests/output_schema_v3.rs @@ -15,6 +15,8 @@ const V3_FAMILIES: &[&str] = &[ "usage", "metrics", "admin_operations", + "replication", + "replication_operations", ]; fn repository_root() -> PathBuf { @@ -163,6 +165,48 @@ fn v3_allows_unknown_server_fields() { assert_valid(&validator, &value, "extended capabilities output"); } +#[test] +fn replication_truncated_fixture_is_explicitly_non_resumable() { + let validator = load_validator(3); + let value = load_json(&fixture_path("replication", "truncated")); + + assert_eq!(value["data"]["scan"]["truncated"], true); + assert_eq!(value["data"]["scan"]["resumable"], false); + assert_valid(&validator, &value, "truncated replication diff"); +} + +#[test] +fn replication_status_allows_empty_reset_id_but_start_requires_exactly_one_target() { + let validator = load_validator(3); + let status = load_json(&fixture_path("replication_operations", "success")); + assert_eq!(status["data"]["targets"][0]["reset_id"], ""); + assert_valid(&validator, &status, "status with persisted empty reset ID"); + + let mut start = status; + start["data"]["operation"] = Value::String("resync_start".to_string()); + assert!( + !validator.is_valid(&start), + "start output must retain a nonempty server reset ID" + ); + start["data"]["targets"][0]["reset_id"] = Value::String("server-id".to_string()); + assert_valid(&validator, &start, "start with nonempty server reset ID"); + + let mut empty_targets = start.clone(); + empty_targets["data"]["targets"] = serde_json::json!([]); + assert!( + !validator.is_valid(&empty_targets), + "start output must contain exactly one target" + ); + + let mut multiple_targets = start; + let target = multiple_targets["data"]["targets"][0].clone(); + multiple_targets["data"]["targets"] = serde_json::json!([target.clone(), target]); + assert!( + !validator.is_valid(&multiple_targets), + "start output must not contain multiple targets" + ); +} + #[test] fn v3_rejects_field_renames_and_type_changes() { let validator = load_validator(3); diff --git a/crates/cli/tests/replication_diff.rs b/crates/cli/tests/replication_diff.rs new file mode 100644 index 00000000..ce4f7ddc --- /dev/null +++ b/crates/cli/tests/replication_diff.rs @@ -0,0 +1,120 @@ +#![cfg(not(windows))] + +mod admin_support; + +use std::fs; +use std::path::Path; +use std::process::Command; +use std::time::Duration; + +use admin_support::{ + rc_binary, rc_host_alias, start_admin_sequence_test_server, start_admin_test_server, +}; +use jsonschema::Validator; +use serde_json::Value; + +fn output_v3_validator() -> Validator { + let path = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../schemas/output_v3.json"); + let contents = fs::read_to_string(&path) + .unwrap_or_else(|error| panic!("failed to read {}: {error}", path.display())); + let schema: Value = serde_json::from_str(&contents).expect("output v3 schema should parse"); + jsonschema::validator_for(&schema).expect("output v3 schema should compile") +} + +fn assert_valid_v3(value: &Value) { + let errors = output_v3_validator() + .iter_errors(value) + .map(|error| error.to_string()) + .collect::>(); + assert!( + errors.is_empty(), + "replication diff output must satisfy output v3:\n{}", + errors.join("\n") + ); +} + +#[test] +fn replication_diff_emits_v3_json_and_forwards_prefix() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_test_server( + r#"{ + "Entries":[{ + "Object":"reports/a.json", + "VersionID":"v1", + "Size":42, + "IsDeleteMarker":false, + "ReplicationStatus":"FAILED", + "LastModified":"2026-07-21T04:00:00Z", + "TargetDetail":{"attempts":2} + }], + "IsTruncated":false, + "ScannedVersions":24, + "ServerRevision":7 + }"#, + ); + + let output = Command::new(rc_binary()) + .args([ + "--json", + "bucket", + "replication", + "diff", + "myalias/source", + "--prefix", + "reports/2026 Q3/", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let payload: Value = serde_json::from_slice(&output.stdout).expect("JSON output"); + assert_eq!(payload["schema_version"], 3); + assert_eq!(payload["type"], "replication"); + assert_eq!(payload["data"]["operation"], "diff"); + assert_eq!(payload["data"]["entries"][0]["version_id"], "v1"); + assert_eq!(payload["data"]["extensions"]["ServerRevision"], 7); + assert_valid_v3(&payload); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + assert_eq!(request.method, "POST"); + assert_eq!( + request.target, + "/rustfs/admin/v3/replication/diff?bucket=source&prefix=reports%2F2026%20Q3%2F" + ); + handle.join().expect("admin test server finished"); +} + +#[test] +fn replication_diff_generic_404_is_v3_unsupported_error() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = + start_admin_sequence_test_server(vec![("404 Not Found", r#"{"message":"route missing"}"#)]); + + let output = Command::new(rc_binary()) + .args(["--json", "bucket", "replication", "diff", "myalias/source"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run rc command"); + + assert_eq!(output.status.code(), Some(7)); + assert!(output.stdout.is_empty()); + let payload: Value = serde_json::from_slice(&output.stderr).expect("JSON error output"); + assert_eq!(payload["type"], "replication"); + assert_eq!(payload["error"]["type"], "unsupported_feature"); + assert_eq!(payload["error"]["capability"], "replication_diff"); + assert_valid_v3(&payload); + + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured admin request"); + handle.join().expect("admin test server finished"); +} diff --git a/crates/cli/tests/replication_operations.rs b/crates/cli/tests/replication_operations.rs new file mode 100644 index 00000000..21ab1abd --- /dev/null +++ b/crates/cli/tests/replication_operations.rs @@ -0,0 +1,124 @@ +//! Process-level contracts for bucket replication check and resync commands. + +use std::process::{Command, Output}; + +use serde_json::Value; +use tempfile::TempDir; + +fn run_rc(args: &[&str]) -> (Output, TempDir) { + let config_dir = tempfile::tempdir().expect("create isolated config dir"); + let output = Command::new(env!("CARGO_BIN_EXE_rc")) + .args(args) + .env("RC_CONFIG_DIR", config_dir.path()) + .output() + .expect("execute rc"); + (output, config_dir) +} + +fn assert_v3_usage_error(output: &Output, operation: &str) { + assert_v3_error(output, operation, 2, "usage_error"); +} + +fn assert_v3_error(output: &Output, operation: &str, code: i32, error_type: &str) { + assert_eq!(output.status.code(), Some(code)); + assert!(output.stdout.is_empty(), "JSON errors belong on stderr"); + let value: Value = serde_json::from_slice(&output.stderr).expect("v3 error JSON"); + assert_eq!(value["schema_version"], 3); + assert_eq!(value["type"], "replication_operations"); + assert_eq!(value["status"], "error"); + assert_eq!(value["error"]["type"], error_type); + assert!( + value["error"]["message"] + .as_str() + .is_some_and(|message| message.contains(operation)) + ); +} + +#[test] +fn replication_check_requires_confirmation_before_alias_setup() { + let (output, _config_dir) = run_rc(&[ + "--json", + "bucket", + "replication", + "check", + "missing-alias/source-bucket", + ]); + + assert_v3_usage_error(&output, "check"); + assert!(String::from_utf8_lossy(&output.stderr).contains("--yes")); +} + +#[test] +fn replication_check_missing_alias_is_not_found_before_network() { + let (output, _config_dir) = run_rc(&[ + "--json", + "bucket", + "replication", + "check", + "missing-alias/source-bucket", + "--yes", + ]); + + assert_v3_error(&output, "check", 5, "not_found"); +} + +#[test] +fn replication_resync_start_requires_confirmation_before_alias_setup() { + let (output, _config_dir) = run_rc(&[ + "--json", + "bucket", + "replication", + "resync", + "start", + "missing-alias/source-bucket", + ]); + + assert_v3_usage_error(&output, "resync_start"); + assert!(String::from_utf8_lossy(&output.stderr).contains("--yes")); +} + +#[test] +fn replication_resync_start_missing_alias_is_not_found_before_network() { + let (output, _config_dir) = run_rc(&[ + "--json", + "bucket", + "replication", + "resync", + "start", + "missing-alias/source-bucket", + "--yes", + ]); + + assert_v3_error(&output, "resync_start", 5, "not_found"); +} + +#[test] +fn replication_resync_status_validates_target_before_alias_setup() { + let (output, _config_dir) = run_rc(&[ + "--json", + "bucket", + "replication", + "resync", + "status", + "missing-alias/source-bucket", + "--target-arn", + "not-an-arn", + ]); + + assert_v3_usage_error(&output, "resync_status"); + assert!(String::from_utf8_lossy(&output.stderr).contains("--target-arn")); +} + +#[test] +fn replication_resync_status_missing_alias_is_not_found_before_network() { + let (output, _config_dir) = run_rc(&[ + "--json", + "bucket", + "replication", + "resync", + "status", + "missing-alias/source-bucket", + ]); + + assert_v3_error(&output, "resync_status", 5, "not_found"); +} diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index 9b9f855f..7a1c4cdc 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -5,6 +5,7 @@ mod capabilities; mod cluster; +mod replication; mod site; pub mod tier; mod types; @@ -22,6 +23,9 @@ pub use cluster::{ RebalancePoolProgress, RebalancePoolStatus, RebalanceStartResult, RebalanceStatus, ServerInfo, UsageInfo, }; +pub use replication::{ + MAX_REPLICATION_DIFF_RESPONSE_BYTES, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry, +}; pub use site::{PeerSiteSpec, ServiceActionResult, SiteRemoveSpec, SiteStatusOptions}; pub use tier::{ TierAliyun, TierAzure, TierConfig, TierCreds, TierGCS, TierHuaweicloud, TierMinIO, TierR2, diff --git a/crates/core/src/admin/replication.rs b/crates/core/src/admin/replication.rs new file mode 100644 index 00000000..44f7912d --- /dev/null +++ b/crates/core/src/admin/replication.rs @@ -0,0 +1,123 @@ +//! Typed contracts for read-only replication diff inspection. + +use std::collections::BTreeMap; + +use async_trait::async_trait; +use jiff::Timestamp; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use crate::Result; + +/// Maximum encoded size accepted for one replication diff response. +pub const MAX_REPLICATION_DIFF_RESPONSE_BYTES: usize = 8 * 1024 * 1024; + +/// A bounded, on-demand scan of object versions that have not replicated. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ReplicationDiff { + #[serde(rename = "Entries")] + pub entries: Vec, + #[serde(rename = "IsTruncated")] + pub is_truncated: bool, + #[serde(rename = "ScannedVersions")] + pub scanned_versions: usize, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +/// One pending or failed object version returned by a replication diff. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ReplicationDiffEntry { + #[serde(rename = "Object")] + pub object: String, + #[serde(rename = "VersionID")] + pub version_id: Option, + #[serde(rename = "Size")] + pub size_bytes: u64, + #[serde(rename = "IsDeleteMarker")] + pub delete_marker: bool, + #[serde(rename = "ReplicationStatus")] + pub replication_status: String, + #[serde(rename = "LastModified")] + pub last_modified: Option, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +/// Read-only RustFS replication diff operations. +#[async_trait] +pub trait ReplicationDiffApi: Send + Sync { + /// Scan a bucket, optionally below a prefix, for pending or failed versions. + async fn replication_diff(&self, bucket: &str, prefix: Option<&str>) + -> Result; +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn response_preserves_unknown_fields_and_typed_entries() { + let response: ReplicationDiff = serde_json::from_str( + r#"{ + "Entries": [{ + "Object": "reports/a.json", + "VersionID": "v1", + "Size": 42, + "IsDeleteMarker": false, + "ReplicationStatus": "FAILED", + "LastModified": "2026-07-21T04:00:00Z", + "TargetDetail": {"attempts": 2} + }], + "IsTruncated": true, + "ScannedVersions": 10000, + "ServerRevision": 7 + }"#, + ) + .expect("typed replication diff"); + + assert_eq!(response.entries[0].size_bytes, 42); + assert_eq!(response.entries[0].version_id.as_deref(), Some("v1")); + assert_eq!(response.entries[0].extra["TargetDetail"]["attempts"], 2); + assert_eq!(response.extra["ServerRevision"], 7); + } + + #[test] + fn response_accepts_delete_marker_without_version_or_timestamp() { + let response: ReplicationDiff = serde_json::from_str( + r#"{ + "Entries": [{ + "Object": "removed.txt", + "VersionID": null, + "Size": 0, + "IsDeleteMarker": true, + "ReplicationStatus": "PENDING", + "LastModified": null + }], + "IsTruncated": false, + "ScannedVersions": 1 + }"#, + ) + .expect("delete marker diff"); + + assert!(response.entries[0].delete_marker); + assert!(response.entries[0].version_id.is_none()); + assert!(response.entries[0].last_modified.is_none()); + } + + #[test] + fn response_rejects_negative_sizes_and_malformed_timestamps() { + for payload in [ + r#"{"Entries":[{"Object":"a","VersionID":null,"Size":-1,"IsDeleteMarker":false,"ReplicationStatus":"FAILED","LastModified":null}],"IsTruncated":false,"ScannedVersions":1}"#, + r#"{"Entries":[{"Object":"a","VersionID":null,"Size":1,"IsDeleteMarker":false,"ReplicationStatus":"FAILED","LastModified":"yesterday"}],"IsTruncated":false,"ScannedVersions":1}"#, + ] { + assert!(serde_json::from_str::(payload).is_err()); + } + } + + #[test] + fn response_requires_scan_completeness_fields() { + let payload = r#"{"Entries":[]}"#; + assert!(serde_json::from_str::(payload).is_err()); + } +} diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index 906f091d..7c53f910 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -37,7 +37,8 @@ pub use lifecycle::{ pub use path::{ParsedPath, RemotePath, parse_object_path, parse_path}; pub use replication::{ BucketTarget, BucketTargetCredentials, ReplicationConfiguration, ReplicationDestination, - ReplicationRule, ReplicationRuleStatus, + ReplicationResyncStartOptions, ReplicationResyncStartResult, ReplicationResyncState, + ReplicationResyncStatus, ReplicationResyncTargetStatus, ReplicationRule, ReplicationRuleStatus, }; pub use retry::{RetryBuilder, is_retryable_error, retry_with_backoff}; pub use select::{ diff --git a/crates/core/src/replication.rs b/crates/core/src/replication.rs index 39c000ee..b3efbdc6 100644 --- a/crates/core/src/replication.rs +++ b/crates/core/src/replication.rs @@ -5,6 +5,80 @@ use serde::{Deserialize, Serialize}; use std::fmt; +use std::time::Duration; + +/// Options for starting a RustFS bucket replication resync. +#[derive(Debug, Clone, PartialEq, Eq, Default)] +pub struct ReplicationResyncStartOptions { + /// Select a configured replication target. The server may infer the only target. + pub target_arn: Option, + /// Only resync objects older than this duration. + pub older_than: Option, + /// Caller-supplied operation identifier. The server generates one when omitted. + pub reset_id: Option, +} + +/// A target accepted by a bucket replication resync start request. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReplicationResyncStartResult { + pub target_arn: String, + pub reset_id: String, +} + +/// Server state for a bucket replication resync target. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ReplicationResyncState { + NotStarted, + Pending, + Ongoing, + Completed, + Failed, + Canceled, + Unknown, +} + +impl ReplicationResyncState { + pub fn from_server(value: &str) -> Self { + match value { + "" => Self::NotStarted, + "Pending" => Self::Pending, + "Ongoing" => Self::Ongoing, + "Completed" => Self::Completed, + "Failed" => Self::Failed, + "Canceled" => Self::Canceled, + _ => Self::Unknown, + } + } +} + +/// Status for one configured replication target. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ReplicationResyncTargetStatus { + pub target_arn: String, + pub reset_id: String, + pub reset_before: Option, + pub started_at: Option, + /// RustFS beta.10 calls this `EndTime`, but populates it from last-update time. + pub last_updated_at: Option, + pub state: ReplicationResyncState, + /// Exact state string returned by the server, including future values. + pub server_state: String, + pub replicated_count: u64, + pub replicated_size: u64, + pub failed_count: u64, + pub failed_size: u64, + pub current_bucket: Option, + pub current_object: Option, + /// Optional server detail. RustFS beta.10 does not reliably persist this field. + pub error: Option, +} + +/// Current server-side bucket replication resync state. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +pub struct ReplicationResyncStatus { + pub targets: Vec, +} // ==================== S3 Replication Config Types ==================== @@ -198,6 +272,38 @@ mod tests { assert!("invalid".parse::().is_err()); } + #[test] + fn resync_state_preserves_empty_and_future_wire_values() { + assert_eq!( + ReplicationResyncState::from_server("Ongoing"), + ReplicationResyncState::Ongoing + ); + assert_eq!( + ReplicationResyncState::from_server(""), + ReplicationResyncState::NotStarted + ); + assert_eq!( + ReplicationResyncState::from_server("FutureState"), + ReplicationResyncState::Unknown + ); + } + + #[test] + fn resync_start_options_distinguish_server_and_caller_ids() { + let generated = ReplicationResyncStartOptions { + target_arn: Some("arn:rustfs:replication::id:dest".to_string()), + older_than: Some(Duration::from_secs(3600)), + reset_id: None, + }; + let explicit = ReplicationResyncStartOptions { + reset_id: Some("reset-1".to_string()), + ..generated.clone() + }; + + assert_eq!(generated.reset_id, None); + assert_eq!(explicit.reset_id.as_deref(), Some("reset-1")); + } + #[test] fn test_replication_configuration_serialization() { let config = ReplicationConfiguration { diff --git a/crates/core/src/traits.rs b/crates/core/src/traits.rs index 34fec473..4124bc66 100644 --- a/crates/core/src/traits.rs +++ b/crates/core/src/traits.rs @@ -15,7 +15,10 @@ use crate::encryption::{BucketEncryption, ObjectEncryptionRequest}; use crate::error::Result; use crate::lifecycle::LifecycleRule; use crate::path::RemotePath; -use crate::replication::ReplicationConfiguration; +use crate::replication::{ + ReplicationConfiguration, ReplicationResyncStartOptions, ReplicationResyncStartResult, + ReplicationResyncStatus, +}; use crate::select::SelectOptions; /// Metadata for an object version @@ -419,6 +422,23 @@ pub trait ObjectStore: Send + Sync { /// Delete bucket replication configuration. async fn delete_bucket_replication(&self, bucket: &str) -> Result<()>; + /// Actively validate configured replication targets. + async fn check_bucket_replication(&self, bucket: &str) -> Result<()>; + + /// Start a server-side bucket replication resync. + async fn start_bucket_replication_resync( + &self, + bucket: &str, + options: ReplicationResyncStartOptions, + ) -> Result; + + /// Read persisted server-side bucket replication resync status. + async fn bucket_replication_resync_status( + &self, + bucket: &str, + target_arn: Option<&str>, + ) -> Result; + /// Get bucket CORS rules. Returns empty vec if no CORS config exists. async fn get_bucket_cors(&self, bucket: &str) -> Result>; diff --git a/crates/s3/Cargo.toml b/crates/s3/Cargo.toml index acd05a62..54b64978 100644 --- a/crates/s3/Cargo.toml +++ b/crates/s3/Cargo.toml @@ -36,6 +36,7 @@ tracing.workspace = true # Utilities jiff.workspace = true +humantime.workspace = true bytes.workspace = true url.workspace = true diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index d253e19f..d287cce6 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -9,15 +9,17 @@ use aws_sigv4::http_request::{ SignableBody, SignableRequest, SignatureLocation, SigningSettings, sign, }; use aws_sigv4::sign::v4; +use futures::StreamExt; use rc_core::admin::{ AccessKeyInfo, AdminApi, BucketQuota, CapabilityApi, CapabilityAvailability, CapabilityEntry, CapabilityReport, ClusterInfo, ClusterSnapshotMetadata, ClusterSnapshotSummary, CreateServiceAccountRequest, DecommissionPoolStatus, DecommissionStatus, ExtensionsCatalog, Group, GroupStatus, HealRuntimeState, HealScanMode, HealStartRequest, HealStatus, - HealTaskRequest, PeerSiteSpec, Policy, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, - RebalanceStartResult, RebalanceStatus, RuntimeCapabilitiesSnapshot, RuntimeCapabilityStatus, - ServiceAccount, ServiceAccountCreateResponse, ServiceActionResult, SiteRemoveSpec, - SiteStatusOptions, UpdateGroupMembersRequest, UpdateServiceAccountRequest, User, UserStatus, + HealTaskRequest, MAX_REPLICATION_DIFF_RESPONSE_BYTES, PeerSiteSpec, Policy, PolicyEntity, + PolicyInfo, PoolStatus, PoolTarget, RebalanceStartResult, RebalanceStatus, ReplicationDiff, + ReplicationDiffApi, RuntimeCapabilitiesSnapshot, RuntimeCapabilityStatus, ServiceAccount, + ServiceAccountCreateResponse, ServiceActionResult, SiteRemoveSpec, SiteStatusOptions, + UpdateGroupMembersRequest, UpdateServiceAccountRequest, User, UserStatus, }; use rc_core::{Alias, Error, Result}; use reqwest::header::{CONTENT_TYPE, HOST, HeaderMap, HeaderName, HeaderValue}; @@ -358,6 +360,63 @@ impl AdminClient { } } + async fn request_bounded_json Deserialize<'de>>( + &self, + method: Method, + path: &str, + query: Option<&[(&str, &str)]>, + body: Option<&[u8]>, + max_response_bytes: usize, + response_name: &str, + ) -> Result { + let mut url = self.admin_url(path); + if let Some(query) = query { + let query_string = query + .iter() + .map(|(key, value)| { + format!( + "{}={}", + urlencoding::encode(key), + urlencoding::encode(value) + ) + }) + .collect::>() + .join("&"); + if !query_string.is_empty() { + url.push('?'); + url.push_str(&query_string); + } + } + + let body_bytes = body.unwrap_or(&[]); + let headers = self.request_headers(body_bytes)?; + let signed_headers = self + .sign_request(&method, &url, &headers, body_bytes) + .await?; + let mut request_builder = self.http_client.request(method, &url); + for (name, value) in signed_headers.iter() { + request_builder = request_builder.header(name, value); + } + if !body_bytes.is_empty() { + request_builder = request_builder.body(body_bytes.to_vec()); + } + + let response = request_builder + .send() + .await + .map_err(|error| Error::Network(format!("Request failed: {error}")))?; + let status = response.status(); + let response_body = + read_bounded_response_body(response, max_response_bytes, response_name).await?; + if !status.is_success() { + return Err( + self.map_replication_diff_error(status, &String::from_utf8_lossy(&response_body)) + ); + } + + serde_json::from_slice(&response_body).map_err(Error::Json) + } + /// Make a signed request that returns no body async fn request_no_response( &self, @@ -462,6 +521,32 @@ impl AdminClient { _ => Error::Network(format!("HTTP {}: {}", status.as_u16(), body)), } } + + fn map_replication_diff_error(&self, status: StatusCode, body: &str) -> Error { + if status != StatusCode::NOT_FOUND { + return self.map_error(status, body); + } + + let structured_error = parse_admin_error(body); + if structured_error.as_ref().is_some_and(|error| { + matches!( + error.code.as_deref(), + Some( + "NoSuchBucket" + | "ReplicationConfigurationNotFoundError" + | "ReplicationConfigurationNotFound" + ) + ) + }) { + return Error::NotFound(body.to_string()); + } + + let reason = structured_error + .and_then(|error| error.message) + .filter(|message| !message.trim().is_empty()) + .unwrap_or_else(|| "the replication diff route was not found".to_string()); + Error::UnsupportedFeature(reason) + } } #[derive(Debug, Deserialize)] @@ -488,6 +573,36 @@ fn parse_admin_error(body: &str) -> Option { .or_else(|| quick_xml::de::from_str(body).ok()) } +async fn read_bounded_response_body( + response: reqwest::Response, + max_response_bytes: usize, + response_name: &str, +) -> Result> { + if response + .content_length() + .is_some_and(|length| length > max_response_bytes as u64) + { + return Err(Error::General(format!( + "{response_name} exceeded the {max_response_bytes}-byte response limit" + ))); + } + + let mut body = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = + chunk.map_err(|error| Error::Network(format!("Failed to read response: {error}")))?; + if body.len().saturating_add(chunk.len()) > max_response_bytes { + return Err(Error::General(format!( + "{response_name} exceeded the {max_response_bytes}-byte response limit" + ))); + } + body.extend_from_slice(&chunk); + } + + Ok(body) +} + /// Response wrapper for user list #[derive(Debug, Deserialize)] struct UserListResponse(HashMap); @@ -1729,6 +1844,30 @@ impl AdminApi for AdminClient { } } +#[async_trait] +impl ReplicationDiffApi for AdminClient { + async fn replication_diff( + &self, + bucket: &str, + prefix: Option<&str>, + ) -> Result { + let mut query = vec![("bucket", bucket)]; + if let Some(prefix) = prefix { + query.push(("prefix", prefix)); + } + + self.request_bounded_json( + Method::POST, + "/replication/diff", + Some(&query), + None, + MAX_REPLICATION_DIFF_RESPONSE_BYTES, + "Replication diff response", + ) + .await + } +} + #[cfg(test)] mod tests { use super::*; @@ -1849,6 +1988,75 @@ mod tests { (endpoint, receiver, handle) } + fn start_admin_declared_length_server( + response_status: &'static str, + content_length: usize, + ) -> ( + String, + mpsc::Receiver, + thread::JoinHandle<()>, + ) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + let (sender, receiver) = mpsc::channel(); + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept request"); + sender + .send(read_admin_request(&mut stream)) + .expect("send captured request"); + let response = format!( + "HTTP/1.1 {response_status}\r\ncontent-length: {content_length}\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n" + ); + stream + .write_all(response.as_bytes()) + .expect("write HTTP headers"); + }); + + (endpoint, receiver, handle) + } + + fn start_admin_chunked_overflow_server() -> ( + String, + mpsc::Receiver, + mpsc::Receiver<()>, + ) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + let (sender, receiver) = mpsc::channel(); + let (completion_sender, completion_receiver) = mpsc::channel(); + thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept request"); + sender + .send(read_admin_request(&mut stream)) + .expect("send captured request"); + stream + .set_write_timeout(Some(Duration::from_secs(2))) + .expect("set response write timeout"); + + let header = b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n"; + let chunk = vec![b'x'; 64 * 1024]; + let mut remaining = MAX_REPLICATION_DIFF_RESPONSE_BYTES; + let mut write_failed = stream.write_all(header).is_err(); + while remaining > 0 && !write_failed { + let chunk_len = remaining.min(chunk.len()); + let chunk_header = format!("{chunk_len:x}\r\n"); + write_failed = stream.write_all(chunk_header.as_bytes()).is_err() + || stream.write_all(&chunk[..chunk_len]).is_err() + || stream.write_all(b"\r\n").is_err(); + remaining -= chunk_len; + } + if !write_failed { + // One additional byte is sufficient to exercise the streaming limit. Sending + // megabytes beyond the limit can block when the client intentionally stops + // reading, especially with Windows socket buffering behavior. + let _ = stream.write_all(b"1\r\nx\r\n"); + } + let _ = completion_sender.send(()); + }); + + (endpoint, receiver, completion_receiver) + } + fn admin_client_for_endpoint(endpoint: &str) -> AdminClient { let alias = Alias::new("test", endpoint, "access", "secret"); AdminClient::new(&alias).expect("admin client should build") @@ -3393,4 +3601,157 @@ mod tests { Err(e) => panic!("Expected Error::Network for invalid PEM, got Err({e})"), } } + + #[tokio::test] + async fn replication_diff_posts_empty_signed_body_and_preserves_extensions() { + let (endpoint, receiver, handle) = start_admin_test_server( + "200 OK", + r#"{ + "Entries":[{ + "Object":"reports/a.json", + "VersionID":"v1", + "Size":42, + "IsDeleteMarker":false, + "ReplicationStatus":"FAILED", + "LastModified":"2026-07-21T04:00:00Z", + "TargetDetail":{"attempts":2} + }], + "IsTruncated":false, + "ScannedVersions":24, + "ServerRevision":7 + }"#, + ); + let client = admin_client_for_endpoint(&endpoint); + + let diff = client + .replication_diff("source bucket", Some("reports/2026 Q3/")) + .await + .expect("replication diff"); + + assert_eq!(diff.entries[0].object, "reports/a.json"); + assert_eq!(diff.entries[0].extra["TargetDetail"]["attempts"], 2); + assert_eq!(diff.extra["ServerRevision"], 7); + let request = receiver.recv().expect("captured request"); + assert_eq!(request.method, "POST"); + assert_eq!( + request.target, + "/rustfs/admin/v3/replication/diff?bucket=source%20bucket&prefix=reports%2F2026%20Q3%2F" + ); + assert!(request.body.is_empty()); + assert!(request.headers.to_ascii_lowercase().contains( + "x-amz-content-sha256: e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" + )); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn replication_diff_omits_prefix_query_when_not_requested() { + let (endpoint, receiver, handle) = start_admin_test_server( + "200 OK", + r#"{"Entries":[],"IsTruncated":false,"ScannedVersions":0}"#, + ); + let client = anonymous_admin_client_for_endpoint(&endpoint); + + let diff = client + .replication_diff("source", None) + .await + .expect("empty replication diff"); + + assert!(diff.entries.is_empty()); + assert_eq!( + receiver.recv().expect("captured request").target, + "/rustfs/admin/v3/replication/diff?bucket=source" + ); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn replication_diff_maps_auth_missing_and_unsupported_responses() { + let cases = [ + ( + "403 Forbidden", + r#"{"Code":"AccessDenied","Message":"denied"}"#, + "auth", + ), + ( + "404 Not Found", + r#"{"Code":"NoSuchBucket","Message":"missing bucket"}"#, + "not_found", + ), + ( + "404 Not Found", + r#"{"Code":"ReplicationConfigurationNotFoundError","Message":"replication is not configured"}"#, + "not_found", + ), + ( + "404 Not Found", + r#"{"message":"route missing"}"#, + "unsupported", + ), + ( + "501 Not Implemented", + r#"{"Code":"NotImplemented","Message":"not implemented"}"#, + "unsupported", + ), + ]; + + for (status, body, expected) in cases { + let (endpoint, _receiver, handle) = start_admin_test_server(status, body); + let error = anonymous_admin_client_for_endpoint(&endpoint) + .replication_diff("source", None) + .await + .expect_err("HTTP error response"); + match expected { + "auth" => assert!(matches!(error, Error::Auth(_))), + "not_found" => assert!(matches!(error, Error::NotFound(_))), + "unsupported" => assert!(matches!(error, Error::UnsupportedFeature(_))), + _ => panic!("unexpected test expectation"), + } + handle.join().expect("server thread"); + } + } + + #[tokio::test] + async fn replication_diff_rejects_malformed_json_and_network_failure() { + let (endpoint, _receiver, handle) = start_admin_test_server("200 OK", "not-json"); + let malformed = anonymous_admin_client_for_endpoint(&endpoint) + .replication_diff("source", None) + .await + .expect_err("malformed JSON response"); + assert!(matches!(malformed, Error::Json(_))); + handle.join().expect("server thread"); + + let listener = TcpListener::bind("127.0.0.1:0").expect("reserve unused port"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + drop(listener); + let network = anonymous_admin_client_for_endpoint(&endpoint) + .replication_diff("source", None) + .await + .expect_err("connection failure"); + assert!(matches!(network, Error::Network(_))); + } + + #[tokio::test] + async fn replication_diff_rejects_declared_and_chunked_overflow() { + let (endpoint, _receiver, handle) = start_admin_declared_length_server( + "403 Forbidden", + MAX_REPLICATION_DIFF_RESPONSE_BYTES + 1, + ); + let declared = anonymous_admin_client_for_endpoint(&endpoint) + .replication_diff("source", None) + .await + .expect_err("declared overflow"); + assert!(matches!(declared, Error::General(message) if message.contains("response limit"))); + handle.join().expect("server thread"); + + let (endpoint, _receiver, completion) = start_admin_chunked_overflow_server(); + let chunked = anonymous_admin_client_for_endpoint(&endpoint) + .replication_diff("source", None) + .await + .expect_err("chunked overflow"); + assert!(matches!(chunked, Error::General(message) if message.contains("response limit"))); + completion + .recv_timeout(Duration::from_secs(5)) + .expect("chunked overflow server should complete within its socket timeout"); + } } diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 4bd14fcb..16c9023b 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -30,14 +30,17 @@ use quick_xml::de::from_str as from_xml_str; use rc_core::{ Alias, BucketEncryption, BucketNotification, Capabilities, CorsRule, Error, LifecycleRule, ListOptions, ListResult, NotificationTarget, ObjectEncryptionRequest, ObjectInfo, ObjectStore, - ObjectVersion, ObjectVersionListResult, RemotePath, ReplicationConfiguration, RequestHeader, - Result, SelectOptions, global_request_headers, + ObjectVersion, ObjectVersionListResult, RemotePath, ReplicationConfiguration, + ReplicationResyncStartOptions, ReplicationResyncStartResult, ReplicationResyncState, + ReplicationResyncStatus, ReplicationResyncTargetStatus, RequestHeader, Result, SelectOptions, + global_request_headers, }; use reqwest::Method; use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue}; use serde::Deserialize; use sha2::{Digest, Sha256}; use std::collections::HashMap; +use std::str::FromStr; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use tokio::io::AsyncReadExt; @@ -50,6 +53,7 @@ const SINGLE_PUT_OBJECT_MAX_SIZE: u64 = crate::multipart::DEFAULT_PART_SIZE; const S3_SERVICE_NAME: &str = "s3"; const S3_REPLICATION_XML_NAMESPACE: &str = "http://s3.amazonaws.com/doc/2006-03-01/"; const RUSTFS_FORCE_DELETE_HEADER: &str = "x-rustfs-force-delete"; +const REPLICATION_EXTENSION_BODY_LIMIT: u64 = 1024 * 1024; static DOWNLOAD_TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(0); #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -255,6 +259,50 @@ struct ReplicationStatusXml { status: Option, } +#[derive(Debug, Deserialize)] +#[serde(rename_all = "PascalCase")] +struct ReplicationResyncResponseDto { + #[serde(default)] + targets: Vec, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "PascalCase")] +struct ReplicationResyncTargetDto { + arn: String, + #[serde(rename = "ResetID")] + reset_id: String, + #[serde(default)] + reset_before_date: Option, + #[serde(default)] + start_time: Option, + #[serde(default)] + end_time: Option, + #[serde(default)] + status: Option, + #[serde(default)] + replicated_count: Option, + #[serde(default)] + replicated_size: Option, + #[serde(default)] + failed_count: Option, + #[serde(default)] + failed_size: Option, + #[serde(default)] + bucket: Option, + #[serde(default)] + object: Option, + #[serde(default)] + error: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "PascalCase")] +struct S3ExtensionErrorDto { + code: String, + message: String, +} + #[derive(Debug, Deserialize)] #[serde(rename_all = "PascalCase")] struct CorsConfigurationXml { @@ -1512,6 +1560,217 @@ impl S3Client { Ok(url) } + fn replication_extension_url( + &self, + bucket: &str, + marker: &str, + query: &[(&str, String)], + ) -> Result { + let mut url = + reqwest::Url::parse(self.alias.endpoint.trim_end_matches('/')).map_err(|error| { + Error::Network(format!( + "Invalid endpoint '{}': {error}", + self.alias.endpoint + )) + })?; + { + let mut segments = url.path_segments_mut().map_err(|_| { + Error::Network(format!( + "Endpoint '{}' does not support path-style bucket operations", + self.alias.endpoint + )) + })?; + segments.pop_if_empty(); + segments.push(bucket); + } + + url.set_query(Some(marker)); + if !query.is_empty() { + let mut serializer = url.query_pairs_mut(); + for (name, value) in query { + serializer.append_pair(name, value); + } + } + Ok(url) + } + + async fn signed_replication_extension_request( + &self, + method: Method, + url: reqwest::Url, + ) -> Result> { + let body = []; + let mut headers = HeaderMap::new(); + headers.insert( + "x-amz-content-sha256", + HeaderValue::from_str(&Self::sha256_hash(&body)) + .map_err(|error| Error::Auth(format!("Invalid content hash header: {error}")))?, + ); + headers.insert( + "host", + HeaderValue::from_str(&self.request_host(&url)?) + .map_err(|error| Error::Auth(format!("Invalid host header: {error}")))?, + ); + for header in &self.request_headers { + let name = HeaderName::from_bytes(header.name.as_bytes()) + .map_err(|error| Error::Auth(format!("Invalid custom header name: {error}")))?; + let value = HeaderValue::from_str(&header.value) + .map_err(|error| Error::Auth(format!("Invalid custom header value: {error}")))?; + headers.insert(name, value); + } + + let signed_headers = self + .sign_xml_request(&method, url.as_str(), &headers, &body) + .await?; + let mut request = self.xml_http_client.request(method, url); + for (name, value) in &signed_headers { + request = request.header(name, value); + } + + let response = request + .send() + .await + .map_err(|error| Error::Network(format!("Replication request failed: {error}")))?; + let status = response.status(); + if response + .content_length() + .is_some_and(|length| length > REPLICATION_EXTENSION_BODY_LIMIT) + { + return Err(Error::General(format!( + "Replication response exceeds the {} byte limit", + REPLICATION_EXTENSION_BODY_LIMIT + ))); + } + + let mut bytes = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.try_next().await.map_err(|error| { + Error::Network(format!("Failed to read replication response: {error}")) + })? { + let next_len = bytes.len().saturating_add(chunk.len()); + if next_len as u64 > REPLICATION_EXTENSION_BODY_LIMIT { + return Err(Error::General(format!( + "Replication response exceeds the {} byte limit", + REPLICATION_EXTENSION_BODY_LIMIT + ))); + } + bytes.extend_from_slice(&chunk); + } + + if !status.is_success() { + return Err(self.map_replication_extension_error(status, &bytes)); + } + Ok(bytes) + } + + fn map_replication_extension_error(&self, status: reqwest::StatusCode, body: &[u8]) -> Error { + let text = String::from_utf8_lossy(body); + let parsed = from_xml_str::(&text).ok(); + let code = parsed.as_ref().map(|error| error.code.as_str()); + let detail = parsed + .as_ref() + .map(|error| error.message.trim()) + .filter(|message| !message.is_empty()) + .unwrap_or_else(|| text.trim()); + let mut message = if detail.is_empty() { + format!("HTTP {}", status.as_u16()) + } else if let Some(code) = code { + format!("{code}: {detail}") + } else { + format!("HTTP {}: {detail}", status.as_u16()) + }; + for sensitive in [&self.alias.access_key, &self.alias.secret_key] { + if !sensitive.is_empty() { + message = message.replace(sensitive, "[REDACTED]"); + } + } + + if status == reqwest::StatusCode::UNAUTHORIZED + || status == reqwest::StatusCode::FORBIDDEN + || matches!(code, Some("AccessDenied" | "Unauthorized" | "Forbidden")) + { + Error::Auth(message) + } else if status == reqwest::StatusCode::NOT_IMPLEMENTED + || status == reqwest::StatusCode::METHOD_NOT_ALLOWED + || matches!(code, Some("NotImplemented" | "MethodNotAllowed")) + { + Error::UnsupportedFeature(message) + } else if matches!( + code, + Some( + "NoSuchBucket" + | "ReplicationConfigurationNotFoundError" + | "ReplicationConfigurationNotFound" + ) + ) { + Error::NotFound(message) + } else if status == reqwest::StatusCode::NOT_FOUND { + Error::UnsupportedFeature(message) + } else if status == reqwest::StatusCode::CONFLICT + || status == reqwest::StatusCode::BAD_REQUEST + || matches!(code, Some("InvalidRequest" | "InvalidBucketState")) + { + Error::Conflict(message) + } else if status.is_server_error() { + Error::Network(message) + } else { + Error::General(message) + } + } + + fn parse_replication_timestamp( + value: Option, + field: &str, + ) -> Result> { + value + .map(|value| { + jiff::Timestamp::from_str(&value).map_err(|error| { + Error::General(format!("Malformed replication {field}: {error}")) + }) + }) + .transpose() + } + + fn convert_resync_status_target( + target: ReplicationResyncTargetDto, + ) -> Result { + if target.arn.is_empty() { + return Err(Error::General( + "Malformed replication status target: missing ARN".to_string(), + )); + } + let server_state = target.status.unwrap_or_default(); + let state = ReplicationResyncState::from_server(&server_state); + let nonnegative = |value: Option, field: &str| { + u64::try_from(value.unwrap_or_default()).map_err(|_| { + Error::General(format!("Malformed replication status: negative {field}")) + }) + }; + + Ok(ReplicationResyncTargetStatus { + target_arn: target.arn, + reset_id: target.reset_id, + reset_before: Self::parse_replication_timestamp( + target.reset_before_date, + "reset-before timestamp", + )?, + started_at: Self::parse_replication_timestamp(target.start_time, "start timestamp")?, + last_updated_at: Self::parse_replication_timestamp( + target.end_time, + "last-update timestamp", + )?, + state, + server_state, + replicated_count: nonnegative(target.replicated_count, "replicated count")?, + replicated_size: nonnegative(target.replicated_size, "replicated size")?, + failed_count: nonnegative(target.failed_count, "failed count")?, + failed_size: nonnegative(target.failed_size, "failed size")?, + current_bucket: target.bucket.filter(|value| !value.is_empty()), + current_object: target.object.filter(|value| !value.is_empty()), + error: target.error.filter(|value| !value.is_empty()), + }) + } + fn cors_url(&self, bucket: &str) -> Result { let mut url = reqwest::Url::parse(self.alias.endpoint.trim_end_matches('/')).map_err(|e| { @@ -3384,6 +3643,90 @@ impl ObjectStore for S3Client { Ok(()) } + async fn check_bucket_replication(&self, bucket: &str) -> Result<()> { + let url = self.replication_extension_url(bucket, "replication-check", &[])?; + let body = self + .signed_replication_extension_request(Method::GET, url) + .await?; + if !body.is_empty() { + return Err(Error::General( + "Malformed replication check response: expected an empty body".to_string(), + )); + } + Ok(()) + } + + async fn start_bucket_replication_resync( + &self, + bucket: &str, + options: ReplicationResyncStartOptions, + ) -> Result { + let mut query = Vec::new(); + if let Some(target_arn) = options.target_arn { + query.push(("arn", target_arn)); + } + if let Some(older_than) = options.older_than { + query.push(( + "older-than", + humantime::format_duration(older_than).to_string(), + )); + } + if let Some(reset_id) = options.reset_id { + query.push(("reset-id", reset_id)); + } + let url = self.replication_extension_url(bucket, "replication-reset", &query)?; + let body = self + .signed_replication_extension_request(Method::PUT, url) + .await?; + let mut response: ReplicationResyncResponseDto = + serde_json::from_slice(&body).map_err(|error| { + Error::General(format!("Malformed replication start response: {error}")) + })?; + if response.targets.len() != 1 { + return Err(Error::General(format!( + "Malformed replication start response: expected one target, got {}", + response.targets.len() + ))); + } + let target = response + .targets + .pop() + .expect("one target was verified before removing it"); + if target.arn.is_empty() || target.reset_id.is_empty() { + return Err(Error::General( + "Malformed replication start response: missing ARN or reset ID".to_string(), + )); + } + Ok(ReplicationResyncStartResult { + target_arn: target.arn, + reset_id: target.reset_id, + }) + } + + async fn bucket_replication_resync_status( + &self, + bucket: &str, + target_arn: Option<&str>, + ) -> Result { + let query = target_arn + .map(|target_arn| vec![("arn", target_arn.to_string())]) + .unwrap_or_default(); + let url = self.replication_extension_url(bucket, "replication-reset-status", &query)?; + let body = self + .signed_replication_extension_request(Method::GET, url) + .await?; + let response: ReplicationResyncResponseDto = + serde_json::from_slice(&body).map_err(|error| { + Error::General(format!("Malformed replication status response: {error}")) + })?; + let targets = response + .targets + .into_iter() + .map(Self::convert_resync_status_target) + .collect::>>()?; + Ok(ReplicationResyncStatus { targets }) + } + async fn select_object_content( &self, path: &RemotePath, @@ -3458,10 +3801,14 @@ mod tests { let config = config_builder.build(); let alias = Alias::new("test", endpoint, "access-key", "secret-key"); + let xml_http_client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .expect("build redirect-disabled XML test client"); let client = S3Client { inner: aws_sdk_s3::Client::from_conf(config), presign_inner: aws_sdk_s3::Client::from_conf(presign_config), - xml_http_client: reqwest::Client::new(), + xml_http_client, alias, request_headers, }; @@ -3559,6 +3906,120 @@ mod tests { (endpoint, receiver, handle) } + fn start_replication_extension_test_server( + response: Vec, + ) -> ( + String, + mpsc::Receiver, + thread::JoinHandle<()>, + ) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + let (sender, receiver) = mpsc::channel(); + + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().expect("accept request"); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("set request timeout"); + let request = read_xml_request(&mut stream); + sender.send(request).expect("send captured request"); + stream.write_all(&response).expect("write response"); + }); + + (endpoint, receiver, handle) + } + + fn start_repeated_replication_extension_test_server( + response: Vec, + request_count: usize, + ) -> (String, thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + + let handle = thread::spawn(move || { + for _ in 0..request_count { + let (mut stream, _) = listener.accept().expect("accept request"); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .expect("set request timeout"); + let _ = read_xml_request(&mut stream); + stream.write_all(&response).expect("write response"); + } + }); + + (endpoint, handle) + } + + fn start_counting_replication_extension_test_server( + first_response: Vec, + ) -> (String, mpsc::Receiver, thread::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind test server"); + listener + .set_nonblocking(true) + .expect("configure nonblocking listener"); + let endpoint = format!("http://{}", listener.local_addr().expect("local addr")); + let (sender, receiver) = mpsc::channel(); + + let handle = thread::spawn(move || { + let first_deadline = Instant::now() + Duration::from_secs(5); + let mut first_stream = loop { + match listener.accept() { + Ok((stream, _)) => break stream, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < first_deadline, + "timed out waiting for request" + ); + thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("accept request: {error}"), + } + }; + first_stream + .set_nonblocking(false) + .expect("configure blocking request stream"); + let _ = read_xml_request(&mut first_stream); + first_stream + .write_all(&first_response) + .expect("write first response"); + drop(first_stream); + + let mut request_count = 1; + let follow_up_deadline = Instant::now() + Duration::from_millis(500); + while Instant::now() < follow_up_deadline { + match listener.accept() { + Ok((mut stream, _)) => { + request_count += 1; + stream + .set_nonblocking(false) + .expect("configure blocking follow-up stream"); + let _ = read_xml_request(&mut stream); + let success_body = + br#"{"Targets":[{"Arn":"arn:target","ResetID":"server-id"}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + success_body.len() + ); + stream + .write_all(response.as_bytes()) + .expect("write follow-up response headers"); + stream + .write_all(success_body) + .expect("write follow-up response body"); + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(10)); + } + Err(error) => panic!("accept follow-up request: {error}"), + } + } + sender.send(request_count).expect("send request count"); + }); + + (endpoint, receiver, handle) + } + fn header_value<'a>(headers: &'a [(String, String)], name: &str) -> Option<&'a str> { headers .iter() @@ -4623,6 +5084,422 @@ mod tests { server_handle.join().expect("server thread should finish"); } + #[tokio::test] + async fn replication_check_uses_signed_empty_s3_extension_request() { + let (endpoint, receiver, handle) = start_replication_extension_test_server( + b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n".to_vec(), + ); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + client + .check_bucket_replication("source-bucket") + .await + .expect("replication check should succeed"); + + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("capture replication check"); + assert_eq!(request.method, "GET"); + assert_eq!(request.target, "/source-bucket?replication-check"); + assert_eq!( + header_value(&request.headers, "x-amz-content-sha256"), + Some("e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855") + ); + assert!( + header_value(&request.headers, "authorization") + .expect("signed request") + .contains("/us-east-1/s3/aws4_request") + ); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_start_encodes_options_and_preserves_server_id() { + let body = br#"{"Targets":[{"Arn":"arn:rustfs:replication::id:dest bucket","ResetID":"server-id"}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect(); + let (endpoint, receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let result = client + .start_bucket_replication_resync( + "source-bucket", + ReplicationResyncStartOptions { + target_arn: Some("arn:rustfs:replication::id:dest bucket".to_string()), + older_than: Some(Duration::from_secs(3600)), + reset_id: None, + }, + ) + .await + .expect("start resync"); + + assert_eq!(result.target_arn, "arn:rustfs:replication::id:dest bucket"); + assert_eq!(result.reset_id, "server-id"); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("capture resync start"); + assert_eq!(request.method, "PUT"); + assert_eq!( + request.target, + "/source-bucket?replication-reset&arn=arn%3Arustfs%3Areplication%3A%3Aid%3Adest+bucket&older-than=1h" + ); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_start_encodes_caller_reset_id() { + let body = br#"{"Targets":[{"Arn":"arn:target","ResetID":"caller id"}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect(); + let (endpoint, receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let result = client + .start_bucket_replication_resync( + "source-bucket", + ReplicationResyncStartOptions { + target_arn: Some("arn:target".to_string()), + older_than: None, + reset_id: Some("caller id".to_string()), + }, + ) + .await + .expect("start resync with caller ID"); + + assert_eq!(result.reset_id, "caller id"); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("capture resync start"); + assert_eq!( + request.target, + "/source-bucket?replication-reset&arn=arn%3Atarget&reset-id=caller+id" + ); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_start_does_not_retry_put_after_server_error() { + let response = + b"HTTP/1.1 503 Service Unavailable\r\ncontent-length: 0\r\nconnection: close\r\n\r\n" + .to_vec(); + let (endpoint, count_receiver, handle) = + start_counting_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let result = client + .start_bucket_replication_resync( + "source-bucket", + ReplicationResyncStartOptions::default(), + ) + .await; + let request_count = count_receiver + .recv_timeout(Duration::from_secs(2)) + .expect("receive PUT request count"); + + assert_eq!(request_count, 1, "PUT must not be retried"); + assert!(matches!(result, Err(Error::Network(_)))); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_start_does_not_follow_redirects() { + let response = b"HTTP/1.1 307 Temporary Redirect\r\nlocation: /redirected\r\ncontent-length: 0\r\nconnection: close\r\n\r\n" + .to_vec(); + let (endpoint, count_receiver, handle) = + start_counting_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let result = client + .start_bucket_replication_resync( + "source-bucket", + ReplicationResyncStartOptions::default(), + ) + .await; + let request_count = count_receiver + .recv_timeout(Duration::from_secs(2)) + .expect("receive PUT request count"); + + assert_eq!(request_count, 1, "signed PUT must not follow redirects"); + assert!(matches!(result, Err(Error::General(_)))); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_status_preserves_partial_target_state() { + let body = br#"{"Targets":[{"Arn":"arn:target","ResetID":"reset-1","ResetBeforeDate":"2026-07-01T00:00:00Z","StartTime":"2026-07-02T00:00:00Z","EndTime":"2026-07-02T00:01:00Z","Status":"Failed","ReplicatedCount":3,"ReplicatedSize":30,"FailedCount":2,"FailedSize":20,"Bucket":"source-bucket","Object":"last.txt","Error":"target unavailable"}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect(); + let (endpoint, receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let status = client + .bucket_replication_resync_status("source-bucket", Some("arn:target")) + .await + .expect("read status"); + + assert_eq!(status.targets.len(), 1); + let target = &status.targets[0]; + assert_eq!(target.state, ReplicationResyncState::Failed); + assert_eq!(target.failed_count, 2); + assert_eq!(target.current_object.as_deref(), Some("last.txt")); + assert_eq!(target.error.as_deref(), Some("target unavailable")); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("capture status request"); + assert_eq!( + request.target, + "/source-bucket?replication-reset-status&arn=arn%3Atarget" + ); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_status_retains_multiple_unfiltered_targets() { + let body = br#"{"Targets":[{"Arn":"arn:a","ResetID":"reset-a","Status":"Pending"},{"Arn":"arn:b","ResetID":"reset-b","Status":"Completed","ReplicatedCount":2}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect(); + let (endpoint, receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let status = client + .bucket_replication_resync_status("source-bucket", None) + .await + .expect("read all target statuses"); + + assert_eq!(status.targets.len(), 2); + assert_eq!(status.targets[0].target_arn, "arn:a"); + assert_eq!(status.targets[1].target_arn, "arn:b"); + assert_eq!(status.targets[1].replicated_count, 2); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("capture unfiltered status request"); + assert_eq!(request.target, "/source-bucket?replication-reset-status"); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_resync_status_is_readable_from_fresh_clients() { + let body = br#"{"Targets":[{"Arn":"arn:target","ResetID":"reset-1","Status":"Ongoing","ReplicatedCount":3}]}"#; + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect(); + let (endpoint, handle) = start_repeated_replication_extension_test_server(response, 2); + + let (first_client, _) = test_s3_client_with_endpoint(&endpoint, None); + let first = first_client + .bucket_replication_resync_status("source-bucket", None) + .await + .expect("first client reads persisted status"); + drop(first_client); + + let (fresh_client, _) = test_s3_client_with_endpoint(&endpoint, None); + let fresh = fresh_client + .bucket_replication_resync_status("source-bucket", None) + .await + .expect("fresh client reads persisted status"); + + assert_eq!(fresh, first); + assert_eq!(fresh.targets[0].state, ReplicationResyncState::Ongoing); + handle.join().expect("server thread should finish"); + } + + #[test] + fn replication_resync_status_preserves_empty_and_future_states() { + let target = |status: &str| ReplicationResyncTargetDto { + arn: "arn:target".to_string(), + reset_id: "reset-1".to_string(), + reset_before_date: None, + start_time: None, + end_time: None, + status: Some(status.to_string()), + replicated_count: Some(0), + replicated_size: Some(0), + failed_count: Some(0), + failed_size: Some(0), + bucket: None, + object: None, + error: None, + }; + + let empty = S3Client::convert_resync_status_target(target("")) + .expect("empty server state is legitimate"); + let mut missing_status_target = target("Pending"); + missing_status_target.status = None; + let missing = S3Client::convert_resync_status_target(missing_status_target) + .expect("omitted server state uses its documented default"); + let mut empty_reset_id_target = target("Pending"); + empty_reset_id_target.reset_id.clear(); + let empty_reset_id = S3Client::convert_resync_status_target(empty_reset_id_target) + .expect("persisted status may have an empty reset ID"); + let future = S3Client::convert_resync_status_target(target("FutureState")) + .expect("future server state is preserved"); + + assert_eq!(empty.state, ReplicationResyncState::NotStarted); + assert_eq!(empty.server_state, ""); + assert_eq!(missing.state, ReplicationResyncState::NotStarted); + assert_eq!(missing.server_state, ""); + assert_eq!(empty_reset_id.reset_id, ""); + assert_eq!(future.state, ReplicationResyncState::Unknown); + assert_eq!(future.server_state, "FutureState"); + } + + #[test] + fn replication_resync_status_rejects_negative_counters() { + let target = ReplicationResyncTargetDto { + arn: "arn:target".to_string(), + reset_id: "reset-1".to_string(), + reset_before_date: None, + start_time: None, + end_time: None, + status: Some("Pending".to_string()), + replicated_count: Some(-1), + replicated_size: Some(0), + failed_count: Some(0), + failed_size: Some(0), + bucket: None, + object: None, + error: None, + }; + + let error = S3Client::convert_resync_status_target(target) + .expect_err("negative count must be malformed"); + assert!(matches!(error, Error::General(_))); + } + + #[tokio::test] + async fn replication_extension_rejects_declared_oversized_body() { + let response = + b"HTTP/1.1 200 OK\r\ncontent-length: 1048577\r\nconnection: close\r\n\r\n".to_vec(); + let (endpoint, _receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let error = client + .bucket_replication_resync_status("source-bucket", None) + .await + .expect_err("oversized response must fail"); + + assert!(matches!(error, Error::General(_))); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_extension_rejects_chunked_oversized_body() { + let body = vec![b'x'; REPLICATION_EXTENSION_BODY_LIMIT as usize + 1]; + let mut response = + b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\nconnection: close\r\n\r\n".to_vec(); + response.extend_from_slice(format!("{:x}\r\n", body.len()).as_bytes()); + response.extend_from_slice(&body); + response.extend_from_slice(b"\r\n0\r\n\r\n"); + let (endpoint, _receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let error = client + .bucket_replication_resync_status("source-bucket", None) + .await + .expect_err("chunked oversized response must fail"); + + assert!(matches!(error, Error::General(_))); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn replication_check_rejects_nonempty_success_body() { + let response = + b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok".to_vec(); + let (endpoint, _receiver, handle) = start_replication_extension_test_server(response); + let (client, _) = test_s3_client_with_endpoint(&endpoint, None); + + let error = client + .check_bucket_replication("source-bucket") + .await + .expect_err("nonempty check response must fail"); + + assert!(matches!(error, Error::General(_))); + handle.join().expect("server thread should finish"); + } + + #[test] + fn replication_extension_maps_typed_errors_and_redacts_credentials() { + let (client, _) = test_s3_client(None); + let access_denied = client.map_replication_extension_error( + reqwest::StatusCode::FORBIDDEN, + b"AccessDeniedaccess-key secret-key denied", + ); + let invalid_request = client.map_replication_extension_error( + reqwest::StatusCode::BAD_REQUEST, + b"InvalidRequesttarget versioning disabled", + ); + let missing = client.map_replication_extension_error( + reqwest::StatusCode::NOT_FOUND, + b"ReplicationConfigurationNotFoundErrormissing", + ); + let missing_bucket = client.map_replication_extension_error( + reqwest::StatusCode::NOT_FOUND, + b"NoSuchBucketmissing bucket", + ); + let missing_route = + client.map_replication_extension_error(reqwest::StatusCode::NOT_FOUND, b"not found"); + let method_not_allowed = client.map_replication_extension_error( + reqwest::StatusCode::METHOD_NOT_ALLOWED, + b"method not allowed", + ); + let unsupported = client.map_replication_extension_error( + reqwest::StatusCode::NOT_IMPLEMENTED, + b"NotImplementedunsupported", + ); + let server_error = client.map_replication_extension_error( + reqwest::StatusCode::SERVICE_UNAVAILABLE, + b"temporarily unavailable", + ); + + assert!(matches!(access_denied, Error::Auth(_))); + let message = access_denied.to_string(); + assert!(!message.contains("access-key")); + assert!(!message.contains("secret-key")); + assert!(message.contains("[REDACTED]")); + assert!(matches!(invalid_request, Error::Conflict(_))); + assert!( + invalid_request + .to_string() + .contains("target versioning disabled") + ); + assert!(matches!(missing, Error::NotFound(_))); + assert!(matches!(missing_bucket, Error::NotFound(_))); + assert!(matches!(missing_route, Error::UnsupportedFeature(_))); + assert!(matches!(method_not_allowed, Error::UnsupportedFeature(_))); + assert!(matches!(unsupported, Error::UnsupportedFeature(_))); + assert!(matches!(server_error, Error::Network(_))); + } + #[tokio::test] async fn delete_object_without_force_delete_omits_rustfs_header() { let (client, request_receiver) = test_s3_client(None); diff --git a/docs/reference/rc/bucket.md b/docs/reference/rc/bucket.md index f3fb57d0..6fdee20f 100644 --- a/docs/reference/rc/bucket.md +++ b/docs/reference/rc/bucket.md @@ -18,7 +18,7 @@ rc bucket version ... rc bucket quota ... rc bucket anonymous ... rc bucket lifecycle ... -rc bucket replication ... +rc bucket replication ... ``` ## Commands @@ -35,7 +35,7 @@ rc bucket replication ... | `quota` | Manage bucket quota. | | `anonymous` | Manage anonymous bucket or prefix access. | | `lifecycle` | Manage lifecycle rules, remote tiers, and object restore requests. | -| `replication` | Manage bucket replication rules and replication status. | +| `replication` | Manage bucket replication rules and inspect replication status or differences. | ## Parameters @@ -95,12 +95,28 @@ Check replication status: rc bucket replication status local/archive ``` +Scan for pending or failed versions below a prefix: + +```bash +rc bucket replication diff local/archive --prefix reports/2026/ +``` + +Actively validate configured targets and start a confirmed existing-object resync: + +```bash +rc bucket replication check local/archive --yes +rc bucket replication resync start local/archive --older-than 30d --yes +rc bucket replication resync status local/archive +``` + ## Behavior Prefer `rc bucket ...` for new scripts. Legacy commands such as `rc mb`, `rc rb`, `rc event`, `rc cors`, `rc version`, `rc anonymous`, `rc quota`, `rc ilm`, and `rc replicate` remain available and delegate to the same implementations. `rc bucket encryption set`, `info`, and `clear` manage only the bucket default for future writes. They do not rewrite, decrypt, or re-encrypt existing objects in the bucket. For object-level encryption flags and more detailed examples, see [Encryption workflows](encryption.md). +`rc bucket replication diff` performs a bounded, read-only scan. A truncated result is partial and cannot be resumed; narrow `--prefix` and run the command again. An empty truncated result does not prove that the bucket has no replication backlog. + Global options shown in command syntax use the same meaning everywhere: | Option | Description | diff --git a/docs/reference/rc/replicate.md b/docs/reference/rc/replicate.md index e12e7ee0..273c280f 100644 --- a/docs/reference/rc/replicate.md +++ b/docs/reference/rc/replicate.md @@ -12,9 +12,13 @@ rc replicate add [OPTIONS] --remote-bucket rc replicate list [OPTIONS] rc replicate status [OPTIONS] +rc replicate diff [--prefix ] rc replicate remove [OPTIONS] rc replicate export [OPTIONS] rc replicate import [OPTIONS] +rc bucket replication check [OPTIONS] --yes +rc bucket replication resync start [OPTIONS] --yes +rc bucket replication resync status [OPTIONS] ``` ## Parameters @@ -29,11 +33,15 @@ rc replicate import [OPTIONS] | `--storage-class` | Destination storage class override. | | `--bandwidth` | Bandwidth limit in bytes per second; `0` means unlimited. | | `--sync` | Enable synchronous replication. | -| `--prefix` | Key prefix filter. | +| `--prefix` | Key prefix filter for rules or a replication diff scan. | | `--healthcheck-seconds` | Health check interval. Defaults to `60` for new rules. | | `--disable-proxy` | Disable replication proxy. | | `--all` | Remove all rules. | | `--force` | Force operation even if capability detection fails. | +| `--target-arn` | Select one configured replication target for resync start or status. | +| `--older-than` | Resync objects older than a positive duration such as `7d10h31s`. | +| `--reset-id` | Supply a caller-controlled resync ID; omit it to use the server-generated ID. | +| `--yes` | Confirm an active target check or resync start. | ## Examples @@ -42,13 +50,33 @@ rc bucket version enable local/reports rc bucket version enable backup/reports rc bucket replication add local/reports --remote-bucket backup/reports --replicate delete,existing-objects rc replicate status local/reports +rc bucket replication diff local/reports --prefix quarterly/2026/ rc replicate export local/reports > replication.json +rc bucket replication check local/reports --yes +rc bucket replication resync start local/reports --target-arn arn:rustfs:replication::id:archive --older-than 30d --yes +rc bucket replication resync status local/reports --target-arn arn:rustfs:replication::id:archive ``` ## Behavior Replication generally requires versioning on source and destination buckets. The target bucket is resolved through its own alias, so configure both source and destination aliases before adding rules. +`replication diff` calls the RustFS Admin API to scan object versions that are pending or failed replication. It reports object keys, version IDs, delete-marker state, sizes, replication statuses, and last-modified timestamps. The request has no body and may be narrowed with `--prefix`. + +The command accepts at most 8 MiB for either a successful or error response. JSON output uses output schema v3 with the `replication` family and `diff` operation. Unknown server fields are preserved under `extensions`. + +The server may stop after its bounded scan and return `truncated: true`. Such output is explicitly partial and non-resumable. Narrow `--prefix` and run the command again; do not interpret an empty truncated result as proof that no backlog exists. The command does not provide time-range, metrics, MRF, or resume-token behavior. + +`replication check` is an active validation probe, not a read-only configuration check. RustFS writes a small temporary object to every selected target and exercises replication delete permissions before cleaning it up. A target-side failure can prevent complete cleanup, so the command requires `--yes`. A successful beta.10 response is empty; a failed response describes only the first target failure selected by the server. The CLI does not fabricate a per-target validation report. + +Use least-privilege source credentials with `s3:PutReplicationConfiguration` for `replication check`. Resync start and status require `s3:ResetBucketReplicationState`; the CLI does not call an Admin-v4 capability endpoint before these signed S3 requests. + +`resync start` uses RustFS signed S3 extension routes rather than the Admin API. Omitting `--target-arn` lets the server select the only enabled target with existing-object replication; multiple eligible targets require an explicit ARN. Omitting `--reset-id` preserves the ID generated by the server. Starting another resync for an already-running target is not an atomic conflict-checked operation in RustFS beta.10. + +`resync status` is stateless on the client and reads persisted server status. The `last_updated_at` field corresponds to RustFS beta.10's `EndTime` response field, which is populated from the latest update even while a resync is pending or ongoing. The optional server error field is returned when present, but beta.10 does not reliably aggregate or persist it; counts and sizes are the durable failure indicators. An unknown target ARN produces a successful empty target list. Bucket resync cancellation is not exposed because beta.10 has no public cancel route. + +JSON output for check, start, and status uses schema v3 with type `replication_operations`. Response and error bodies are limited to 1 MiB. + Global options shown in command syntax use the same meaning everywhere: | Option | Description | diff --git a/schemas/output_v3.json b/schemas/output_v3.json index daf3b2f2..db727e26 100644 --- a/schemas/output_v3.json +++ b/schemas/output_v3.json @@ -31,7 +31,9 @@ "watch_event", "usage", "metrics", - "admin_operations" + "admin_operations", + "replication", + "replication_operations" ] }, "pagination": { @@ -278,6 +280,122 @@ } } }, + "replicationDiffEntry": { + "type": "object", + "required": [ + "object", "version_id", "delete_marker", "size_bytes", + "replication_status", "last_modified", "extensions" + ], + "properties": { + "object": { "type": "string" }, + "version_id": { "$ref": "#/definitions/nullableString" }, + "delete_marker": { "type": "boolean" }, + "size_bytes": { "type": "integer", "minimum": 0 }, + "replication_status": { "type": "string", "minLength": 1 }, + "last_modified": { "$ref": "#/definitions/nullableTimestamp" }, + "extensions": { "type": "object" } + } + }, + "replicationDiffScan": { + "type": "object", + "required": ["scanned_versions", "truncated", "resumable"], + "properties": { + "scanned_versions": { "type": "integer", "minimum": 0 }, + "truncated": { "type": "boolean" }, + "resumable": { "const": false } + } + }, + "replicationData": { + "type": "object", + "required": [ + "operation", "bucket", "prefix", "entries", "scan", "extensions" + ], + "properties": { + "operation": { "const": "diff" }, + "bucket": { "type": "string", "minLength": 1 }, + "prefix": { "$ref": "#/definitions/nullableString" }, + "entries": { + "type": "array", + "items": { "$ref": "#/definitions/replicationDiffEntry" } + }, + "scan": { "$ref": "#/definitions/replicationDiffScan" }, + "extensions": { "type": "object" } + } + }, + "replicationTarget": { + "type": "object", + "required": [ + "target_arn", "reset_id", "reset_before", "started_at", + "last_updated_at", "state", "server_state", "replicated_count", + "replicated_size", "failed_count", "failed_size", "current_bucket", + "current_object", "error" + ], + "properties": { + "target_arn": { "type": "string", "minLength": 1 }, + "reset_id": { "type": "string" }, + "reset_before": { "$ref": "#/definitions/nullableTimestamp" }, + "started_at": { "$ref": "#/definitions/nullableTimestamp" }, + "last_updated_at": { "$ref": "#/definitions/nullableTimestamp" }, + "state": { + "type": ["string", "null"], + "enum": [ + "not_started", "pending", "ongoing", "completed", "failed", + "canceled", "unknown", null + ] + }, + "server_state": { "$ref": "#/definitions/nullableString" }, + "replicated_count": { "type": ["integer", "null"], "minimum": 0 }, + "replicated_size": { "$ref": "#/definitions/nullableBytes" }, + "failed_count": { "type": ["integer", "null"], "minimum": 0 }, + "failed_size": { "$ref": "#/definitions/nullableBytes" }, + "current_bucket": { "$ref": "#/definitions/nullableString" }, + "current_object": { "$ref": "#/definitions/nullableString" }, + "error": { "$ref": "#/definitions/nullableString" } + } + }, + "replicationStartTarget": { + "allOf": [ + { "$ref": "#/definitions/replicationTarget" }, + { + "properties": { + "reset_id": { "type": "string", "minLength": 1 } + } + } + ] + }, + "replicationOperationsData": { + "type": "object", + "required": ["operation", "bucket", "valid", "targets"], + "properties": { + "operation": { + "type": "string", + "enum": ["check", "resync_start", "resync_status"] + }, + "bucket": { "type": "string", "minLength": 1 }, + "valid": { "type": ["boolean", "null"] }, + "targets": { + "type": "array", + "items": { "$ref": "#/definitions/replicationTarget" } + } + }, + "allOf": [ + { + "if": { + "properties": { "operation": { "const": "resync_start" } } + }, + "then": { + "properties": { + "targets": { + "type": "array", + "minItems": 1, + "maxItems": 1, + "items": { "$ref": "#/definitions/replicationStartTarget" } + } + } + } + } + ] + }, "unsupportedFeatureError": { "type": "object", "required": ["type", "message", "retryable", "capability", "server"], @@ -412,6 +530,28 @@ } ] }, + { + "allOf": [ + { "$ref": "#/definitions/successEnvelope" }, + { + "properties": { + "type": { "const": "replication" }, + "data": { "$ref": "#/definitions/replicationData" } + } + } + ] + }, + { + "allOf": [ + { "$ref": "#/definitions/successEnvelope" }, + { + "properties": { + "type": { "const": "replication_operations" }, + "data": { "$ref": "#/definitions/replicationOperationsData" } + } + } + ] + }, { "$ref": "#/definitions/errorOutput" } ] }