diff --git a/crates/cli/src/commands/admin/replicate.rs b/crates/cli/src/commands/admin/replicate.rs index 298fbf2a..4cc3be31 100644 --- a/crates/cli/src/commands/admin/replicate.rs +++ b/crates/cli/src/commands/admin/replicate.rs @@ -18,8 +18,10 @@ use crate::exit_code::ExitCode; use crate::output::Formatter; use rc_core::admin::{ AdminApi, PeerSiteSpec, ReplicateEditStatus, SiteRemoveSpec, SiteReplicationInfo, - SiteReplicationPeer, SiteReplicationResyncOperation, SiteReplicationResyncStatus, + SiteReplicationPeer, SiteReplicationRepairApi, SiteReplicationRepairOperationStatus, + SiteReplicationRepairPreflight, SiteReplicationResyncOperation, SiteReplicationResyncStatus, SiteStatusOptions, validate_site_replication_ca_bundle, + validate_site_replication_repair_operation_id, validate_site_replication_repair_token, }; use rc_core::{AliasManager, Error}; use rc_s3::AdminClient; @@ -39,6 +41,9 @@ pub enum ReplicateCommands { /// Manage persisted site resync operation snapshots Resync(ResyncArgs), + /// Plan, execute, or inspect durable site replication repair + Repair(RepairArgs), + /// Show site replication status Status(StatusArgs), @@ -140,6 +145,70 @@ pub struct ResyncStatusArgs { pub site: String, } +#[derive(clap::Args, Debug)] +pub struct RepairArgs { + #[command(subcommand)] + pub command: RepairCommands, +} + +#[derive(Subcommand, Debug)] +pub enum RepairCommands { + /// Create a local-only repair plan and preflight token + DryRun(RepairDryRunArgs), + + /// Execute or resume a repair using the exact preflight token and operation ID + Execute(RepairExecuteArgs), + + /// Read a durable repair snapshot from a fresh process + Status(RepairStatusArgs), +} + +#[derive(clap::Args, Debug)] +pub struct RepairDryRunArgs { + /// Alias name of the server + pub alias: String, +} + +#[derive(clap::Args)] +pub struct RepairExecuteArgs { + /// Alias name of the server + pub alias: String, + + /// Complete server-issued preflight token from a separate dry-run + #[arg(long)] + pub preflight_token: String, + + /// Stable UUID retained and reused for retries + #[arg(long)] + pub operation_id: String, + + /// Confirm the repair mutation + #[arg(long)] + pub yes: bool, +} + +impl std::fmt::Debug for RepairExecuteArgs { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("RepairExecuteArgs") + .field("alias", &self.alias) + .field("has_preflight_token", &!self.preflight_token.is_empty()) + .field("operation_id", &self.operation_id) + .field("yes", &self.yes) + .finish() + } +} + +#[derive(clap::Args, Debug)] +pub struct RepairStatusArgs { + /// Alias name of the server + pub alias: String, + + /// Durable operation UUID + #[arg(long)] + pub operation_id: String, +} + #[derive(clap::Args, Debug)] pub struct StatusArgs { /// Alias name of the server @@ -187,6 +256,7 @@ pub async fn execute(cmd: ReplicateCommands, formatter: &Formatter) -> ExitCode ReplicateCommands::Info(args) => execute_info(args, formatter).await, ReplicateCommands::Edit(args) => execute_edit(args, formatter).await, ReplicateCommands::Resync(args) => execute_resync(args, formatter).await, + ReplicateCommands::Repair(args) => execute_repair(args, formatter).await, ReplicateCommands::Status(args) => execute_status(args, formatter).await, ReplicateCommands::Remove(args) => execute_remove(args, formatter).await, } @@ -579,6 +649,201 @@ async fn execute_resync_request( } } +async fn execute_repair(args: RepairArgs, formatter: &Formatter) -> ExitCode { + match args.command { + RepairCommands::DryRun(args) => execute_repair_dry_run(args, formatter).await, + RepairCommands::Execute(args) => execute_repair_execute(args, formatter).await, + RepairCommands::Status(args) => execute_repair_status(args, formatter).await, + } +} + +async fn repair_client( + alias: &str, + formatter: &Formatter, + operation: &'static str, +) -> Result { + let client = load_admin_client(alias) + .map_err(|error| emit_admin_error(formatter, &error, operation, false))?; + client + .site_replication_repair_capability() + .await + .map_err(|error| emit_admin_error(formatter, &error, operation, false))?; + Ok(client) +} + +async fn execute_repair_dry_run(args: RepairDryRunArgs, formatter: &Formatter) -> ExitCode { + const OPERATION: &str = "site_replication_repair_dry_run"; + let client = match repair_client(&args.alias, formatter, OPERATION).await { + Ok(client) => client, + Err(code) => return code, + }; + match client.site_replication_repair_dry_run().await { + Ok(preflight) => { + emit_repair_preflight(formatter, &args.alias, preflight); + ExitCode::Success + } + Err(error) => emit_admin_error(formatter, &error, OPERATION, false), + } +} + +async fn execute_repair_execute(args: RepairExecuteArgs, formatter: &Formatter) -> ExitCode { + const OPERATION: &str = "site_replication_repair_execute"; + if !args.yes { + return emit_admin_message_error( + formatter, + ExitCode::UsageError, + OPERATION, + false, + "Site replication repair execute requires --yes confirmation".to_string(), + ); + } + if let Err(error) = validate_site_replication_repair_token(&args.preflight_token) { + return emit_admin_error(formatter, &error, OPERATION, false); + } + if let Err(error) = validate_site_replication_repair_operation_id(&args.operation_id) { + return emit_admin_error(formatter, &error, OPERATION, false); + } + let client = match repair_client(&args.alias, formatter, OPERATION).await { + Ok(client) => client, + Err(code) => return code, + }; + match client + .site_replication_repair_execute(&args.preflight_token, &args.operation_id) + .await + { + Ok(status) => emit_repair_operation(formatter, OPERATION, &args.alias, true, status), + Err(error) => emit_admin_error( + formatter, + &error, + OPERATION, + mutation_was_attempted(true, &error), + ), + } +} + +async fn execute_repair_status(args: RepairStatusArgs, formatter: &Formatter) -> ExitCode { + const OPERATION: &str = "site_replication_repair_status"; + if let Err(error) = validate_site_replication_repair_operation_id(&args.operation_id) { + return emit_admin_error(formatter, &error, OPERATION, false); + } + let client = match repair_client(&args.alias, formatter, OPERATION).await { + Ok(client) => client, + Err(code) => return code, + }; + match client + .site_replication_repair_status(&args.operation_id) + .await + { + Ok(status) => emit_repair_operation(formatter, OPERATION, &args.alias, false, status), + Err(error) => emit_admin_error(formatter, &error, OPERATION, false), + } +} + +fn emit_repair_preflight( + formatter: &Formatter, + alias: &str, + preflight: SiteReplicationRepairPreflight, +) { + if formatter.is_json() { + formatter.json(&admin_success_output( + "site_replication_repair_dry_run", + alias.to_string(), + false, + preflight, + )); + return; + } + formatter.println("Site replication repair plan:"); + formatter.println(&format!( + " Preflight token: {}", + formatter.sanitize_text(&preflight.preflight_token) + )); + formatter.println(&format!(" Retry events: {}", preflight.retry_events)); + emit_repair_sites_human(formatter, &preflight.sites); +} + +fn emit_repair_operation( + formatter: &Formatter, + operation: &'static str, + alias: &str, + changed: bool, + status: SiteReplicationRepairOperationStatus, +) -> ExitCode { + let failed = status.has_failure(); + let state = if failed { + "failed" + } else if status.status == "success" { + "succeeded" + } else if status.status == "running" { + "running" + } else { + "unknown" + }; + if formatter.is_json() { + formatter.json(&admin_operation_output( + operation, + alias.to_string(), + state, + Some(status.operation_id.clone()), + changed, + status, + )); + } else { + formatter.println(&format!( + "Site replication repair: status={}, operation ID={}", + formatter.sanitize_text(&status.status), + formatter.sanitize_text(&status.operation_id) + )); + emit_repair_sites_human(formatter, &status.sites); + if failed { + formatter.error( + "Repair is partial or failed; retry execute with the same operation ID and preflight token", + ); + } + } + if failed { + ExitCode::GeneralError + } else { + ExitCode::Success + } +} + +fn emit_repair_sites_human( + formatter: &Formatter, + sites: &BTreeMap, +) { + for site in sites.values() { + formatter.println(&format!( + " Site {} ({})", + formatter.sanitize_text(&site.name), + formatter.sanitize_text(&site.deployment_id) + )); + for (name, family) in &site.families { + formatter.println(&format!( + " {}: planned={}, succeeded={}, failed={}, retry-events={}", + formatter.sanitize_text(name), + family.planned, + family.succeeded, + family.failed, + family.retry_events + )); + for task in &family.tasks { + let error = task + .error + .as_deref() + .map(|value| format!(", error={}", formatter.sanitize_text(value))) + .unwrap_or_default(); + formatter.println(&format!( + " {}: {}{}", + formatter.sanitize_text(&task.task_id), + formatter.sanitize_text(&task.status), + error + )); + } + } + } +} + fn mutation_was_attempted(mutation: bool, error: &Error) -> bool { mutation && !matches!(error, Error::RequestRejected(_)) } diff --git a/crates/cli/tests/admin_replicate.rs b/crates/cli/tests/admin_replicate.rs index fa0e79ad..37d9cd55 100644 --- a/crates/cli/tests/admin_replicate.rs +++ b/crates/cli/tests/admin_replicate.rs @@ -54,6 +54,130 @@ const RESYNC_START_RESPONSE: &str = r#"{ "sessionToken":"MUST-NOT-PRINT" }"#; +const REPAIR_OPERATION_ID: &str = "550e8400-e29b-41d4-a716-446655440000"; +const REPAIR_PREFLIGHT_TOKEN: &str = "abcdefghijklmnopqrstuvwxyzABCDEFGH012345678"; +const REPAIR_CAPABILITY_RESPONSE: &str = r#"{ + "summary":{ + "observability":{"state":"supported"}, + "userspace_profiling":{"state":"supported"}, + "memory_sampling":{"state":"supported"}, + "platform":{"state":"supported"}, + "topology":{"state":"supported"}, + "cluster_snapshot":{"state":"supported"} + }, + "site_replication_repair":{ + "contract_version":1, + "status":{"state":"supported"}, + "modes":["dry-run","execute"], + "execute_route":"/rustfs/admin/v3/site-replication/repair", + "status_route":"/rustfs/admin/v3/site-replication/repair/status", + "preflight_token_contract":"hmac-sha256-v1", + "operation_id_format":"uuid", + "max_retained_successful_operations":32, + "disabled_by_default":false + }, + "cluster_snapshot_path":"/rustfs/admin/v4/cluster/snapshot", + "cluster_snapshot_summary":null, + "topology_status":{"state":"supported"} +}"#; +const REPAIR_PREFLIGHT_RESPONSE: &str = r#"{ + "mode":"dry-run", + "status":"planned", + "preflightToken":"abcdefghijklmnopqrstuvwxyzABCDEFGH012345678", + "retryEvents":2, + "sites":{ + "dep-2":{ + "deploymentId":"dep-2", + "name":"secondary", + "families":{ + "iam":{ + "planned":1, + "succeeded":0, + "failed":0, + "retryEvents":2, + "tasks":[{"taskId":"abcdefghijklmnopqrstuvwxyzABCDEFGH012345678","status":"planned"}] + } + } + } + } +}"#; +const REPAIR_PARTIAL_RESPONSE: &str = r#"{ + "mode":"execute", + "operationId":"550e8400-e29b-41d4-a716-446655440000", + "status":"partial", + "sites":{ + "dep-2":{ + "deploymentId":"dep-2", + "name":"secondary", + "families":{ + "iam":{ + "planned":2, + "succeeded":1, + "failed":1, + "retryEvents":1, + "tasks":[ + {"taskId":"abcdefghijklmnopqrstuvwxyzABCDEFGH012345678","status":"succeeded"}, + {"taskId":"bbcdefghijklmnopqrstuvwxyzABCDEFGH012345678","status":"failed","error":"remote-operation-failed"} + ], + "errors":["remote-operation-failed"] + } + } + } + }, + "createdAt":"2026-07-25T00:00:00Z", + "updatedAt":"2026-07-25T00:01:00Z" +}"#; +const REPAIR_RETRY_SUCCESS_RESPONSE: &str = r#"{ + "mode":"execute", + "operationId":"550e8400-e29b-41d4-a716-446655440000", + "status":"success", + "sites":{ + "dep-2":{ + "deploymentId":"dep-2", + "name":"secondary", + "families":{ + "iam":{ + "planned":2, + "succeeded":2, + "failed":0, + "retryEvents":2, + "tasks":[ + {"taskId":"abcdefghijklmnopqrstuvwxyzABCDEFGH012345678","status":"skipped"}, + {"taskId":"bbcdefghijklmnopqrstuvwxyzABCDEFGH012345678","status":"succeeded"} + ] + } + } + } + }, + "createdAt":"2026-07-25T00:00:00Z", + "updatedAt":"2026-07-25T00:02:00Z", + "completedAt":"2026-07-25T00:02:00Z" +}"#; +const REPAIR_RUNNING_RESPONSE: &str = r#"{ + "mode":"execute", + "operationId":"550e8400-e29b-41d4-a716-446655440000", + "status":"running", + "sites":{ + "dep-2":{ + "deploymentId":"dep-2", + "name":"secondary", + "families":{ + "iam":{ + "planned":1, + "succeeded":0, + "failed":0, + "retryEvents":0, + "tasks":[ + {"taskId":"abcdefghijklmnopqrstuvwxyzABCDEFGH012345678","status":"running"} + ] + } + } + } + }, + "createdAt":"2026-07-25T00:00:00Z", + "updatedAt":"2026-07-25T00:01:00Z" +}"#; + fn run_resync_command( operation: &str, info: &'static str, @@ -1410,6 +1534,340 @@ fn replicate_resync_cancel_without_server_state_returns_conflict() { assert_eq!(requests.len(), 2); } +#[test] +fn replicate_repair_dry_run_is_capability_gated_and_never_executes() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_sequence_test_server(vec![ + ("200 OK", REPAIR_CAPABILITY_RESPONSE), + ("200 OK", REPAIR_PREFLIGHT_RESPONSE), + ]); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "dry-run", + "myalias", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run repair dry-run"); + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + let requests = (0..2) + .map(|_| { + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured repair request") + }) + .collect::>(); + handle.join().expect("admin test server finished"); + assert_eq!(requests[0].target, "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!( + requests[1].target, + "/rustfs/admin/v3/site-replication/repair" + ); + let request: serde_json::Value = + serde_json::from_slice(&requests[1].body).expect("repair request JSON"); + assert_eq!(request, serde_json::json!({"mode":"dry-run"})); + let payload: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("repair output JSON"); + assert_eq!( + payload["data"]["operations"][0]["result"]["preflightToken"], + REPAIR_PREFLIGHT_TOKEN + ); + assert_eq!( + payload["data"]["operations"][0]["result"]["sites"]["dep-2"]["families"]["iam"]["retryEvents"], + 2 + ); +} + +#[test] +fn replicate_repair_execute_requires_confirmation_before_alias_or_network() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "execute", + "missing", + "--preflight-token", + REPAIR_PREFLIGHT_TOKEN, + "--operation-id", + REPAIR_OPERATION_ID, + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .output() + .expect("run repair execute without confirmation"); + assert_eq!(output.status.code(), Some(2)); + let payload: serde_json::Value = + serde_json::from_slice(&output.stderr).expect("usage error JSON"); + assert_eq!(payload["error"]["type"], "usage_error"); +} + +#[test] +fn replicate_repair_rejects_invalid_token_before_capability_or_mutation() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "execute", + "missing", + "--preflight-token", + "truncated", + "--operation-id", + REPAIR_OPERATION_ID, + "--yes", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .output() + .expect("run repair execute with invalid token"); + assert_eq!(output.status.code(), Some(2)); + let payload: serde_json::Value = + serde_json::from_slice(&output.stderr).expect("validation error JSON"); + assert_eq!(payload["error"]["type"], "usage_error"); + assert!(!String::from_utf8_lossy(&output.stderr).contains(REPAIR_PREFLIGHT_TOKEN)); +} + +#[test] +fn replicate_repair_older_server_stops_after_capability_gate() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let unsupported = r#"{ + "summary":{ + "observability":{"state":"supported"}, + "userspace_profiling":{"state":"supported"}, + "memory_sampling":{"state":"supported"}, + "platform":{"state":"supported"}, + "topology":{"state":"supported"}, + "cluster_snapshot":{"state":"supported"} + }, + "cluster_snapshot_path":"/rustfs/admin/v4/cluster/snapshot", + "cluster_snapshot_summary":null, + "topology_status":{"state":"supported"} + }"#; + let (endpoint, receiver, handle) = start_admin_test_server(unsupported); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "dry-run", + "myalias", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run unsupported repair"); + assert_eq!(output.status.code(), Some(7)); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured capability request"); + handle.join().expect("admin test server finished"); + assert_eq!(request.target, "/rustfs/admin/v4/runtime/capabilities"); + let payload: serde_json::Value = + serde_json::from_slice(&output.stderr).expect("unsupported JSON"); + assert_eq!(payload["error"]["type"], "unsupported_feature"); +} + +#[test] +fn replicate_repair_partial_execute_retains_all_checkpoints_and_exits_nonzero() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_sequence_test_server(vec![ + ("200 OK", REPAIR_CAPABILITY_RESPONSE), + ("200 OK", REPAIR_PARTIAL_RESPONSE), + ]); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "execute", + "myalias", + "--preflight-token", + REPAIR_PREFLIGHT_TOKEN, + "--operation-id", + REPAIR_OPERATION_ID, + "--yes", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run partial repair execute"); + assert_eq!(output.status.code(), Some(1)); + assert!(output.stderr.is_empty(), "partial JSON belongs on stdout"); + let requests = (0..2) + .map(|_| { + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured repair request") + }) + .collect::>(); + handle.join().expect("admin test server finished"); + let request: serde_json::Value = + serde_json::from_slice(&requests[1].body).expect("repair request JSON"); + assert_eq!(request["mode"], "execute"); + assert_eq!(request["preflightToken"], REPAIR_PREFLIGHT_TOKEN); + assert_eq!(request["operationId"], REPAIR_OPERATION_ID); + let payload: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("partial repair JSON"); + let operation = &payload["data"]["operations"][0]; + assert_eq!(operation["state"], "failed"); + assert_eq!(operation["operation_id"], REPAIR_OPERATION_ID); + assert_eq!( + operation["result"]["sites"]["dep-2"]["families"]["iam"]["tasks"] + .as_array() + .expect("all task checkpoints") + .len(), + 2 + ); +} + +#[test] +fn replicate_repair_status_uses_only_the_durable_operation_id() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_sequence_test_server(vec![ + ("200 OK", REPAIR_CAPABILITY_RESPONSE), + ("200 OK", REPAIR_PARTIAL_RESPONSE), + ]); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "status", + "myalias", + "--operation-id", + REPAIR_OPERATION_ID, + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run repair status"); + assert_eq!(output.status.code(), Some(1)); + let requests = (0..2) + .map(|_| { + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured repair request") + }) + .collect::>(); + handle.join().expect("admin test server finished"); + assert_eq!(requests[1].method, "GET"); + assert_eq!( + requests[1].target, + format!( + "/rustfs/admin/v3/site-replication/repair/status?operation-id={REPAIR_OPERATION_ID}" + ) + ); + assert!(requests[1].body.is_empty()); +} + +#[test] +fn replicate_repair_same_id_retry_preserves_skip_and_retries_only_failed_task() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_sequence_test_server(vec![ + ("200 OK", REPAIR_CAPABILITY_RESPONSE), + ("200 OK", REPAIR_PARTIAL_RESPONSE), + ("200 OK", REPAIR_CAPABILITY_RESPONSE), + ("200 OK", REPAIR_RETRY_SUCCESS_RESPONSE), + ]); + let run = || { + Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "execute", + "myalias", + "--preflight-token", + REPAIR_PREFLIGHT_TOKEN, + "--operation-id", + REPAIR_OPERATION_ID, + "--yes", + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run repair execute") + }; + + let partial = run(); + let retried = run(); + assert_eq!(partial.status.code(), Some(1)); + assert!(retried.status.success()); + let requests = (0..4) + .map(|_| { + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured retry lifecycle request") + }) + .collect::>(); + handle.join().expect("admin test server finished"); + for request in [&requests[1], &requests[3]] { + let body: serde_json::Value = serde_json::from_slice(&request.body).expect("execute JSON"); + assert_eq!(body["operationId"], REPAIR_OPERATION_ID); + assert_eq!(body["preflightToken"], REPAIR_PREFLIGHT_TOKEN); + } + let payload: serde_json::Value = + serde_json::from_slice(&retried.stdout).expect("retry success JSON"); + let family = &payload["data"]["operations"][0]["result"]["sites"]["dep-2"]["families"]["iam"]; + assert_eq!(family["tasks"][0]["status"], "skipped"); + assert_eq!(family["tasks"][1]["status"], "succeeded"); + assert_eq!(family["retryEvents"], 2); +} + +#[test] +fn replicate_repair_running_status_remains_authoritative() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let (endpoint, receiver, handle) = start_admin_sequence_test_server(vec![ + ("200 OK", REPAIR_CAPABILITY_RESPONSE), + ("200 OK", REPAIR_RUNNING_RESPONSE), + ]); + let output = Command::new(rc_binary()) + .args([ + "--json", + "admin", + "replicate", + "repair", + "status", + "myalias", + "--operation-id", + REPAIR_OPERATION_ID, + ]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("run repair running status"); + assert!(output.status.success()); + let payload: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("running status JSON"); + assert_eq!(payload["data"]["operations"][0]["state"], "running"); + assert_eq!( + payload["data"]["operations"][0]["result"]["status"], + "running" + ); + for _ in 0..2 { + receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured status lifecycle request"); + } + handle.join().expect("admin test server finished"); +} + #[test] fn replicate_resync_start_rejects_self_before_mutation() { let config_dir = tempfile::tempdir().expect("create config dir"); diff --git a/crates/cli/tests/fixtures/output_v3/admin_operations/site_repair_partial.json b/crates/cli/tests/fixtures/output_v3/admin_operations/site_repair_partial.json new file mode 100644 index 00000000..e06f00bc --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/admin_operations/site_repair_partial.json @@ -0,0 +1,50 @@ +{ + "schema_version": 3, + "type": "admin_operations", + "status": "success", + "data": { + "operations": [ + { + "operation": "site_replication_repair_execute", + "resource": "primary", + "state": "failed", + "operation_id": "550e8400-e29b-41d4-a716-446655440000", + "changed": true, + "result": { + "mode": "execute", + "operationId": "550e8400-e29b-41d4-a716-446655440000", + "status": "partial", + "sites": { + "dep-2": { + "deploymentId": "dep-2", + "name": "secondary", + "families": { + "iam": { + "planned": 2, + "succeeded": 1, + "failed": 1, + "retryEvents": 1, + "tasks": [ + { + "taskId": "abcdefghijklmnopqrstuvwxyzABCDEFGH012345678", + "status": "succeeded" + }, + { + "taskId": "bbcdefghijklmnopqrstuvwxyzABCDEFGH012345678", + "status": "failed", + "error": "remote-operation-failed" + } + ], + "errors": ["remote-operation-failed"] + } + } + } + }, + "createdAt": "2026-07-25T00:00:00Z", + "updatedAt": "2026-07-25T00:01:00Z", + "completedAt": null + } + } + ] + } +} diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index 85461be3..22a24fde 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -665,6 +665,26 @@ fn nested_subcommand_help_contract() { usage: "Usage: rc admin replicate resync cancel [OPTIONS] --site ", expected_tokens: &["--site", "--yes"], }, + HelpCase { + args: &["admin", "replicate", "repair"], + usage: "Usage: rc admin replicate repair [OPTIONS] ", + expected_tokens: &["dry-run", "execute", "status"], + }, + HelpCase { + args: &["admin", "replicate", "repair", "dry-run"], + usage: "Usage: rc admin replicate repair dry-run [OPTIONS] ", + expected_tokens: &[], + }, + HelpCase { + args: &["admin", "replicate", "repair", "execute"], + usage: "Usage: rc admin replicate repair execute [OPTIONS] --preflight-token --operation-id ", + expected_tokens: &["--preflight-token", "--operation-id", "--yes"], + }, + HelpCase { + args: &["admin", "replicate", "repair", "status"], + usage: "Usage: rc admin replicate repair status [OPTIONS] --operation-id ", + expected_tokens: &["--operation-id"], + }, HelpCase { args: &["admin", "info", "storage"], usage: "Usage: rc admin info storage [OPTIONS] ", diff --git a/crates/cli/tests/output_schema_v3.rs b/crates/cli/tests/output_schema_v3.rs index f5436449..7a8adc50 100644 --- a/crates/cli/tests/output_schema_v3.rs +++ b/crates/cli/tests/output_schema_v3.rs @@ -157,6 +157,69 @@ fn replication_status_and_mrf_fixtures_are_valid() { } } +#[test] +fn site_repair_partial_fixture_preserves_every_checkpoint() { + let validator = load_validator(3); + let path = fixture_path("admin_operations", "site_repair_partial"); + let fixture = load_json(&path); + assert_valid(&validator, &fixture, &path.display().to_string()); + + let operation = &fixture["data"]["operations"][0]; + assert_eq!( + operation["operation_id"], + operation["result"]["operationId"] + ); + assert_eq!(operation["result"]["status"], "partial"); + assert_eq!( + operation["result"]["sites"]["dep-2"]["families"]["iam"]["tasks"] + .as_array() + .map(Vec::len), + Some(2) + ); + + let mut missing_checkpoint_id = fixture.clone(); + missing_checkpoint_id["data"]["operations"][0]["result"]["sites"]["dep-2"]["families"] + ["iam"]["tasks"][0] + .as_object_mut() + .expect("task object") + .remove("taskId"); + assert!(!validator.is_valid(&missing_checkpoint_id)); + + let mut unsafe_error = fixture; + unsafe_error["data"]["operations"][0]["result"]["sites"]["dep-2"]["families"]["iam"]["tasks"] + [1]["error"] = serde_json::json!("https://user:secret@example.test"); + assert!(!validator.is_valid(&unsafe_error)); + + let preflight = serde_json::json!({ + "schema_version": 3, + "type": "admin_operations", + "status": "success", + "data": { + "operations": [{ + "operation": "site_replication_repair_dry_run", + "resource": "primary", + "state": "succeeded", + "operation_id": null, + "changed": false, + "result": { + "mode": "dry-run", + "status": "planned", + "preflightToken": "abcdefghijklmnopqrstuvwxyzABCDEFGH012345678", + "retryEvents": 0, + "sites": {} + } + }] + } + }); + assert_valid(&validator, &preflight, "site repair preflight output"); + let mut missing_token = preflight; + missing_token["data"]["operations"][0]["result"] + .as_object_mut() + .expect("preflight result") + .remove("preflightToken"); + assert!(!validator.is_valid(&missing_token)); +} + #[test] fn multipart_partial_fixture_preserves_successes_and_per_upload_errors() { let validator = load_validator(3); diff --git a/crates/core/src/admin/capabilities.rs b/crates/core/src/admin/capabilities.rs index 273cde06..9243228f 100644 --- a/crates/core/src/admin/capabilities.rs +++ b/crates/core/src/admin/capabilities.rs @@ -92,6 +92,8 @@ pub struct RuntimeCapabilitiesSnapshot { pub summary: RuntimeCapabilitiesSummary, #[serde(default)] pub inspect_archive: Option, + #[serde(default)] + pub site_replication_repair: Option, pub cluster_snapshot_path: String, pub cluster_snapshot_summary: Option, pub topology_status: RuntimeCapabilityStatus, diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index 15511b72..ec357a9b 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -114,10 +114,16 @@ pub use replication::{ }; pub use site::{ MAX_SITE_REPLICATION_CA_CERT_BYTES, MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, - MAX_SITE_REPLICATION_REQUEST_BYTES, MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, PeerSiteSpec, - ReplicateEditStatus, ServiceActionResult, SiteRemoveSpec, SiteReplicationInfo, - SiteReplicationPeer, SiteReplicationResyncBucketStatus, SiteReplicationResyncOperation, - SiteReplicationResyncStatus, SiteStatusOptions, validate_site_replication_ca_bundle, + MAX_SITE_REPLICATION_REPAIR_RESPONSE_BYTES, MAX_SITE_REPLICATION_REQUEST_BYTES, + MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, PeerSiteSpec, ReplicateEditStatus, + SITE_REPLICATION_REPAIR_CAPABILITY, ServiceActionResult, SiteRemoveSpec, SiteReplicationInfo, + SiteReplicationPeer, SiteReplicationRepairApi, SiteReplicationRepairCapabilityContract, + SiteReplicationRepairFamilyStatus, SiteReplicationRepairOperationStatus, + SiteReplicationRepairPreflight, SiteReplicationRepairRequest, SiteReplicationRepairSiteStatus, + SiteReplicationRepairTaskStatus, SiteReplicationResyncBucketStatus, + SiteReplicationResyncOperation, SiteReplicationResyncStatus, SiteStatusOptions, + validate_site_replication_ca_bundle, validate_site_replication_repair_operation_id, + validate_site_replication_repair_token, }; pub use tier::{ ManualTransitionRunReport, ManualTransitionRunRequest, ManualTransitionRunResponse, TierAliyun, diff --git a/crates/core/src/admin/site.rs b/crates/core/src/admin/site.rs index 708823e4..c3430a46 100644 --- a/crates/core/src/admin/site.rs +++ b/crates/core/src/admin/site.rs @@ -21,6 +21,322 @@ pub const MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES: usize = 64 * 1024; pub const MAX_SITE_REPLICATION_REQUEST_BYTES: usize = 1024 * 1024; /// Maximum custom CA bundle accepted by the CLI. pub const MAX_SITE_REPLICATION_CA_CERT_BYTES: usize = 256 * 1024; +/// Stable capability name for the durable site-replication repair lifecycle. +pub const SITE_REPLICATION_REPAIR_CAPABILITY: &str = "admin.site-replication.repair"; +/// Maximum preflight/operation response accepted from the repair routes. +pub const MAX_SITE_REPLICATION_REPAIR_RESPONSE_BYTES: usize = 8 * 1024 * 1024; + +/// Validate the opaque HMAC-SHA256 v1 preflight token without inspecting its contents. +pub fn validate_site_replication_repair_token(token: &str) -> Result<()> { + if token.len() == 43 + && token + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) + { + Ok(()) + } else { + Err(Error::InvalidPath( + "Preflight token must be the complete 43-character server-issued token".to_string(), + )) + } +} + +/// Validate the canonical UUID syntax required for durable operation identifiers. +pub fn validate_site_replication_repair_operation_id(operation_id: &str) -> Result<()> { + let bytes = operation_id.as_bytes(); + let valid = bytes.len() == 36 + && bytes.iter().enumerate().all(|(index, byte)| { + if matches!(index, 8 | 13 | 18 | 23) { + *byte == b'-' + } else { + byte.is_ascii_hexdigit() + } + }); + if valid { + Ok(()) + } else { + Err(Error::InvalidPath( + "Operation ID must be a canonical UUID".to_string(), + )) + } +} + +/// Capability contract advertised by `/rustfs/admin/v4/runtime/capabilities`. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct SiteReplicationRepairCapabilityContract { + pub contract_version: u32, + pub status: super::RuntimeCapabilityStatus, + pub modes: Vec, + pub execute_route: String, + pub status_route: String, + pub preflight_token_contract: String, + pub operation_id_format: String, + pub max_retained_successful_operations: usize, + pub disabled_by_default: bool, +} + +impl SiteReplicationRepairCapabilityContract { + /// Reject incomplete or incompatible server advertisements before using a repair route. + pub fn validate(&self) -> Result<()> { + let supported = self.status.state == super::RuntimeCapabilityState::Supported; + let modes_ok = self.modes.len() == 2 + && ["dry-run", "execute"] + .iter() + .all(|required| self.modes.iter().any(|mode| mode == required)); + if self.contract_version != 1 + || !supported + || !modes_ok + || self.execute_route != "/rustfs/admin/v3/site-replication/repair" + || self.status_route != "/rustfs/admin/v3/site-replication/repair/status" + || self.preflight_token_contract != "hmac-sha256-v1" + || self.operation_id_format != "uuid" + || self.max_retained_successful_operations == 0 + || self.disabled_by_default + { + return Err(Error::UnsupportedFeature( + "RustFS did not advertise the supported durable site-replication repair contract" + .to_string(), + )); + } + Ok(()) + } +} + +/// Request body for dry-run and execute repair operations. +#[derive(Clone, Serialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct SiteReplicationRepairRequest { + pub mode: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub preflight_token: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub operation_id: Option, +} + +/// One durable task checkpoint. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SiteReplicationRepairTaskStatus { + pub task_id: String, + pub status: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +/// Per-family task counts and checkpoints. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SiteReplicationRepairFamilyStatus { + pub planned: usize, + pub succeeded: usize, + pub failed: usize, + #[serde(default)] + pub retry_events: usize, + #[serde(default)] + pub tasks: Vec, + #[serde(default)] + pub errors: Vec, +} + +/// Per-site repair plan or operation status. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SiteReplicationRepairSiteStatus { + pub deployment_id: String, + pub name: String, + pub families: BTreeMap, +} + +/// Dry-run response containing the server-issued preflight token. +#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SiteReplicationRepairPreflight { + pub mode: String, + pub status: String, + pub preflight_token: String, + pub retry_events: usize, + pub sites: BTreeMap, +} + +impl fmt::Debug for SiteReplicationRepairPreflight { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("SiteReplicationRepairPreflight") + .field("mode", &self.mode) + .field("status", &self.status) + .field("has_preflight_token", &!self.preflight_token.is_empty()) + .field("retry_events", &self.retry_events) + .field("site_count", &self.sites.len()) + .finish() + } +} + +impl SiteReplicationRepairPreflight { + pub fn validate(&self) -> Result<()> { + if self.mode != "dry-run" || self.status != "planned" { + return Err(Error::General( + "RustFS returned an inconsistent site-replication repair preflight".to_string(), + )); + } + validate_site_replication_repair_token(&self.preflight_token)?; + validate_repair_sites(&self.sites) + } +} + +/// Durable execute/status snapshot. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct SiteReplicationRepairOperationStatus { + pub mode: String, + pub operation_id: String, + pub status: String, + pub sites: BTreeMap, + #[serde(default)] + pub created_at: Option, + #[serde(default)] + pub updated_at: Option, + #[serde(default)] + pub completed_at: Option, +} + +impl SiteReplicationRepairOperationStatus { + pub fn validate(&self, expected_operation_id: &str) -> Result<()> { + validate_site_replication_repair_operation_id(&self.operation_id)?; + if self.mode != "execute" + || !self + .operation_id + .eq_ignore_ascii_case(expected_operation_id) + || !matches!( + self.status.as_str(), + "running" | "success" | "partial" | "failed" + ) + { + return Err(Error::General( + "RustFS returned an inconsistent site-replication repair operation".to_string(), + )); + } + for timestamp in [&self.created_at, &self.updated_at, &self.completed_at] + .into_iter() + .flatten() + { + timestamp.parse::().map_err(|_| { + Error::General( + "RustFS returned an invalid site-replication repair timestamp".to_string(), + ) + })?; + } + if self.created_at.is_none() || self.updated_at.is_none() { + return Err(Error::General( + "RustFS returned an incomplete site-replication repair timestamp".to_string(), + )); + } + validate_repair_sites(&self.sites) + } + + /// Terminal snapshots that must cause a non-zero CLI exit. + pub fn has_failure(&self) -> bool { + matches!( + self.status.trim().to_ascii_lowercase().as_str(), + "partial" | "failed" | "failure" + ) || self + .sites + .values() + .flat_map(|site| site.families.values()) + .any(|family| { + family.failed > 0 + || family.tasks.iter().any(|task| { + matches!( + task.status.trim().to_ascii_lowercase().as_str(), + "failed" | "failure" + ) + }) + }) + } +} + +fn validate_repair_sites(sites: &BTreeMap) -> Result<()> { + for (deployment_id, site) in sites { + if deployment_id.is_empty() + || site.deployment_id != *deployment_id + || site.name.trim().is_empty() + || site.families.is_empty() + { + return Err(Error::General( + "RustFS returned an incomplete site-replication repair site".to_string(), + )); + } + for (family_name, family) in &site.families { + let successful = family + .tasks + .iter() + .filter(|task| matches!(task.status.as_str(), "succeeded" | "skipped")) + .count(); + let failed = family + .tasks + .iter() + .filter(|task| task.status == "failed") + .count(); + let mut task_ids = std::collections::BTreeSet::new(); + if family_name.is_empty() + || family.succeeded.saturating_add(family.failed) > family.planned + || family.tasks.len() != family.planned + || successful != family.succeeded + || failed != family.failed + || family + .errors + .iter() + .any(|error| !repair_error_is_safe(error)) + || family.tasks.iter().any(|task| { + validate_site_replication_repair_token(&task.task_id).is_err() + || !task_ids.insert(task.task_id.as_str()) + || !matches!( + task.status.as_str(), + "planned" | "running" | "succeeded" | "failed" | "skipped" + ) + || task + .error + .as_deref() + .is_some_and(|error| !repair_error_is_safe(error)) + }) + { + return Err(Error::General( + "RustFS returned inconsistent site-replication repair checkpoints".to_string(), + )); + } + } + } + Ok(()) +} + +fn repair_error_is_safe(error: &str) -> bool { + matches!( + error, + "authorization-failed" + | "remote-timeout" + | "remote-dns-failed" + | "remote-tls-failed" + | "remote-connect-failed" + | "remote-operation-failed" + ) +} + +/// Durable site-replication repair transport. +#[async_trait::async_trait] +pub trait SiteReplicationRepairApi: Send + Sync { + async fn site_replication_repair_capability( + &self, + ) -> Result; + async fn site_replication_repair_dry_run(&self) -> Result; + async fn site_replication_repair_execute( + &self, + preflight_token: &str, + operation_id: &str, + ) -> Result; + async fn site_replication_repair_status( + &self, + operation_id: &str, + ) -> Result; +} /// Validate a bounded certificate-only PEM bundle for site replication edits. pub fn validate_site_replication_ca_bundle(pem: &[u8]) -> Result<()> { @@ -981,4 +1297,85 @@ mod tests { exact.push(b' '); assert!(validate_site_replication_ca_bundle(&exact).is_err()); } + + #[test] + fn repair_identifiers_are_strictly_validated() { + assert!( + validate_site_replication_repair_token("abcdefghijklmnopqrstuvwxyzABCDEFGH012345678") + .is_ok() + ); + assert!(validate_site_replication_repair_token("short").is_err()); + assert!( + validate_site_replication_repair_token("abcdefghijklmnopqrstuvwxyzABCDEFGH01234567=") + .is_err() + ); + + assert!( + validate_site_replication_repair_operation_id("550e8400-e29b-41d4-a716-446655440000") + .is_ok() + ); + assert!( + validate_site_replication_repair_operation_id("550e8400e29b41d4a716446655440000") + .is_err() + ); + + let preflight: SiteReplicationRepairPreflight = serde_json::from_value(serde_json::json!({ + "mode": "dry-run", + "status": "planned", + "preflightToken": "abcdefghijklmnopqrstuvwxyzABCDEFGH012345678", + "retryEvents": 0, + "sites": {} + })) + .expect("valid empty preflight"); + assert!(!format!("{preflight:?}").contains(&preflight.preflight_token)); + } + + #[test] + fn repair_snapshot_rejects_inconsistent_counts_and_detects_partial() { + let json = serde_json::json!({ + "mode": "execute", + "operationId": "550e8400-e29b-41d4-a716-446655440000", + "status": "partial", + "createdAt": "2026-07-25T00:00:00Z", + "updatedAt": "2026-07-25T00:01:00Z", + "sites": { + "dep-2": { + "deploymentId": "dep-2", + "name": "secondary", + "families": { + "iam": { + "planned": 1, + "succeeded": 0, + "failed": 1, + "retryEvents": 1, + "tasks": [{ + "taskId": "abcdefghijklmnopqrstuvwxyzABCDEFGH012345678", + "status": "failed", + "error": "remote-operation-failed" + }], + "errors": ["remote-operation-failed"] + } + } + } + } + }); + let status: SiteReplicationRepairOperationStatus = + serde_json::from_value(json.clone()).expect("valid repair response"); + assert!( + status + .validate("550e8400-e29b-41d4-a716-446655440000") + .is_ok() + ); + assert!(status.has_failure()); + + let mut invalid = json; + invalid["sites"]["dep-2"]["families"]["iam"]["planned"] = Value::from(0); + let status: SiteReplicationRepairOperationStatus = + serde_json::from_value(invalid).expect("syntactically valid repair response"); + assert!( + status + .validate("550e8400-e29b-41d4-a716-446655440000") + .is_err() + ); + } } diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index 8067a650..da04b537 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -37,18 +37,20 @@ use rc_core::admin::{ MAX_IAM_POLICY_ENTITIES_RESPONSE_BYTES, MAX_INSPECT_ARCHIVE_BYTES, MAX_METRICS_LINE_BYTES, MAX_METRICS_RESPONSE_BYTES, MAX_METRICS_SAMPLES, MAX_OIDC_RESPONSE_BYTES, MAX_REPLICATION_DIFF_RESPONSE_BYTES, MAX_REPLICATION_INSPECTION_RESPONSE_BYTES, - MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, MAX_SITE_REPLICATION_REQUEST_BYTES, - MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, ManualTransitionRunRequest, - ManualTransitionRunResponse, MetricsBatch, MetricsQuery, ModuleSwitches, ObservabilityApi, - OidcMutationApi, OidcMutationRequest, OidcMutationResult, OidcProvider, OidcProviderList, - OidcReadApi, OidcValidationRequest, OidcValidationResult, PeerSiteSpec, Policy, - PolicyDetachEntity, PolicyDetachRequest, PolicyDetachResult, PolicyEntitiesQuery, - PolicyEntitiesResult, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, RealtimeMetrics, - RebalanceStartResult, RebalanceStatus, ReplicateEditStatus, ReplicationDiff, + MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, MAX_SITE_REPLICATION_REPAIR_RESPONSE_BYTES, + MAX_SITE_REPLICATION_REQUEST_BYTES, MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, + ManualTransitionRunRequest, ManualTransitionRunResponse, MetricsBatch, MetricsQuery, + ModuleSwitches, ObservabilityApi, OidcMutationApi, OidcMutationRequest, OidcMutationResult, + OidcProvider, OidcProviderList, OidcReadApi, OidcValidationRequest, OidcValidationResult, + PeerSiteSpec, Policy, PolicyDetachEntity, PolicyDetachRequest, PolicyDetachResult, + PolicyEntitiesQuery, PolicyEntitiesResult, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, + RealtimeMetrics, RebalanceStartResult, RebalanceStatus, ReplicateEditStatus, ReplicationDiff, ReplicationDiffApi, ReplicationInspectionApi, ReplicationMetricScope, ReplicationMetrics, ReplicationMrf, RuntimeCapabilitiesSnapshot, RuntimeCapabilityStatus, ScannerStatus, ServiceAccount, ServiceAccountCreateResponse, ServiceActionResult, SiteRemoveSpec, - SiteReplicationInfo, SiteReplicationPeer, SiteReplicationResyncOperation, + SiteReplicationInfo, SiteReplicationPeer, SiteReplicationRepairApi, + SiteReplicationRepairCapabilityContract, SiteReplicationRepairOperationStatus, + SiteReplicationRepairPreflight, SiteReplicationRepairRequest, SiteReplicationResyncOperation, SiteReplicationResyncStatus, SiteStatusOptions, StorageInfo, UpdateGroupMembersRequest, UpdateServiceAccountRequest, User, UserStatus, }; @@ -594,7 +596,11 @@ impl AdminClient { })?; let status = response.status(); let limit = if status.is_success() { - MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES + if operation_label.starts_with("Site replication repair") { + MAX_SITE_REPLICATION_REPAIR_RESPONSE_BYTES + } else { + MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES + } } else { MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES }; @@ -1081,6 +1087,9 @@ impl AdminClient { body: &str, operation_label: &str, ) -> Error { + if operation_label.starts_with("Site replication repair") { + return map_site_replication_repair_error(status, body, operation_label); + } if matches!( status, StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED @@ -1407,13 +1416,102 @@ impl AdminClient { } } +fn map_site_replication_repair_error( + status: StatusCode, + body: &str, + operation_label: &str, +) -> Error { + if matches!( + status, + StatusCode::NOT_FOUND | StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED + ) { + return Error::UnsupportedFeature(format!( + "{operation_label} is not supported by this server" + )); + } + if matches!(status, StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED) { + return Error::Auth("Authentication failed for site-replication repair".to_string()); + } + + let message = parse_admin_error(body) + .and_then(|error| error.message) + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + match status { + StatusCode::BAD_REQUEST if message == "repair operation was not found" => Error::NotFound( + "The durable site-replication repair operation was not found".to_string(), + ), + StatusCode::BAD_REQUEST + if message == "execute requires a valid preflighttoken" + || message == "dry-run does not accept preflighttoken or operationid" => + { + Error::General( + "The site-replication repair preflight token or plan is invalid".to_string(), + ) + } + StatusCode::BAD_REQUEST if message == "operationid must be a uuid" => { + Error::General("The site-replication repair operation ID is invalid".to_string()) + } + StatusCode::CONFLICT + if message == "repair operation id is already bound to a different preflight" => + { + Error::Conflict("The repair operation ID is bound to a different preflight".to_string()) + } + StatusCode::CONFLICT if message == "another site replication repair is active" => { + Error::Conflict("Another site-replication repair operation is active".to_string()) + } + StatusCode::CONFLICT => { + Error::Conflict("Site-replication repair has an operation ID conflict".to_string()) + } + StatusCode::PRECONDITION_FAILED + if message == "site replication repair plan changed after partial execution" => + { + Error::Conflict( + "The site-replication repair plan changed after partial execution".to_string(), + ) + } + StatusCode::PRECONDITION_FAILED => Error::Conflict( + "The site-replication repair preflight or topology is stale".to_string(), + ), + status if status.is_server_error() => Error::General( + "RustFS reported an internal site-replication repair failure".to_string(), + ), + StatusCode::BAD_REQUEST | StatusCode::UNPROCESSABLE_ENTITY => { + Error::General("RustFS rejected the site-replication repair request".to_string()) + } + _ => Error::Network(format!( + "Site-replication repair request failed with HTTP {}", + status.as_u16() + )), + } +} + +fn sanitize_site_replication_repair_capability_error(error: Error) -> Error { + match error.exit_code() { + 3 => Error::Network("Site-replication repair capability discovery failed".to_string()), + 4 => Error::Auth( + "Authentication failed while checking site-replication repair capability".to_string(), + ), + 5 | 7 => Error::UnsupportedFeature( + "RustFS does not advertise durable site-replication repair".to_string(), + ), + 6 => Error::Conflict( + "Site-replication repair capability discovery conflicted with server state".to_string(), + ), + _ => Error::General( + "RustFS returned an invalid site-replication repair capability response".to_string(), + ), + } +} + fn site_replication_response_rejected( mutation_outcome_label: Option<&str>, reason: String, ) -> Error { if let Some(label) = mutation_outcome_label { Error::General(format!( - "{label} outcome is unknown because the {reason}; do not retry blindly; inspect the persisted resync snapshot and storage state" + "{label} outcome is unknown because the {reason}; do not retry blindly; inspect the persisted operation status and storage state" )) } else { Error::General(format!("Site replication {reason}")) @@ -1422,7 +1520,7 @@ fn site_replication_response_rejected( fn site_replication_response_unknown_network(label: &str, status: u16) -> Error { Error::Network(format!( - "{label} outcome is unknown after the server returned HTTP {status}; do not retry blindly; inspect the persisted resync snapshot and storage state" + "{label} outcome is unknown after the server returned HTTP {status}; do not retry blindly; inspect the persisted operation status and storage state" )) } @@ -3929,6 +4027,100 @@ impl InspectArchiveApi for AdminClient { } } +#[async_trait] +impl SiteReplicationRepairApi for AdminClient { + async fn site_replication_repair_capability( + &self, + ) -> Result { + let runtime = self + .request_v4::(Method::GET, "/runtime/capabilities") + .await + .map_err(sanitize_site_replication_repair_capability_error)?; + let contract = runtime.site_replication_repair.ok_or_else(|| { + Error::UnsupportedFeature( + "RustFS does not advertise durable site-replication repair".to_string(), + ) + })?; + contract.validate()?; + Ok(contract) + } + + async fn site_replication_repair_dry_run(&self) -> Result { + let body = serde_json::to_vec(&SiteReplicationRepairRequest { + mode: "dry-run".to_string(), + preflight_token: None, + operation_id: None, + }) + .map_err(|_| Error::General("Failed to encode site-replication repair request".into()))?; + let response: SiteReplicationRepairPreflight = self + .request_site_replication( + Method::PUT, + "/site-replication/repair", + Some(&body), + "Site replication repair dry-run", + None, + None, + ) + .await?; + response.validate()?; + Ok(response) + } + + async fn site_replication_repair_execute( + &self, + preflight_token: &str, + operation_id: &str, + ) -> Result { + rc_core::admin::validate_site_replication_repair_token(preflight_token)?; + rc_core::admin::validate_site_replication_repair_operation_id(operation_id)?; + let body = Zeroizing::new( + serde_json::to_vec(&SiteReplicationRepairRequest { + mode: "execute".to_string(), + preflight_token: Some(preflight_token.to_string()), + operation_id: Some(operation_id.to_string()), + }) + .map_err(|_| { + Error::General("Failed to encode site-replication repair request".into()) + })?, + ); + let response: SiteReplicationRepairOperationStatus = self + .request_site_replication( + Method::PUT, + "/site-replication/repair", + Some(&body), + "Site replication repair execute", + Some("Site replication repair execute"), + Some("Site replication repair execute"), + ) + .await?; + response.validate(operation_id)?; + Ok(response) + } + + async fn site_replication_repair_status( + &self, + operation_id: &str, + ) -> Result { + rc_core::admin::validate_site_replication_repair_operation_id(operation_id)?; + let path = format!( + "/site-replication/repair/status?operation-id={}", + urlencoding::encode(operation_id) + ); + let response: SiteReplicationRepairOperationStatus = self + .request_site_replication( + Method::GET, + &path, + None, + "Site replication repair status", + None, + None, + ) + .await?; + response.validate(operation_id)?; + Ok(response) + } +} + #[async_trait] impl ConfigApi for AdminClient { async fn get_config(&self, selector: &str) -> Result { @@ -8615,6 +8807,177 @@ mod tests { "generation":7 }"#; + const SITE_REPLICATION_REPAIR_TOKEN: &str = "abcdefghijklmnopqrstuvwxyzABCDEFGH012345678"; + const SITE_REPLICATION_REPAIR_ID: &str = "550e8400-e29b-41d4-a716-446655440000"; + + #[tokio::test] + async fn site_replication_repair_rejects_malformed_mutation_response_without_echoing_it() { + let (endpoint, _receiver, handle) = start_admin_test_server( + "200 OK", + r#"{"mode":"execute","credential":"MUST-NOT-ECHO"}"#, + ); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_execute( + SITE_REPLICATION_REPAIR_TOKEN, + SITE_REPLICATION_REPAIR_ID, + ) + .await + .expect_err("malformed repair response must fail"); + assert!(matches!(error, Error::General(_))); + assert!(error.to_string().contains("outcome is unknown")); + assert!(!error.to_string().contains("MUST-NOT-ECHO")); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn site_replication_repair_maps_stale_preflight_to_conflict() { + let (endpoint, _receiver, handle) = start_admin_test_server( + "412 Precondition Failed", + r#"{"Code":"PreconditionFailed","Message":"token=DO-NOT-ECHO"}"#, + ); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_execute( + SITE_REPLICATION_REPAIR_TOKEN, + SITE_REPLICATION_REPAIR_ID, + ) + .await + .expect_err("stale repair preflight must fail"); + assert!(matches!(error, Error::Conflict(_))); + assert!(!error.to_string().contains("DO-NOT-ECHO")); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn site_replication_repair_capability_errors_never_echo_server_bodies() { + for (status, expected_exit) in [ + ("500 Internal Server Error", 3), + ("400 Bad Request", 1), + ("403 Forbidden", 4), + ] { + let (endpoint, _receiver, handle) = start_admin_test_server( + status, + r#"{"Message":"https://user:secret@example.test token=MUST-NOT-ECHO"}"#, + ); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_capability() + .await + .expect_err("capability failure must be sanitized"); + assert_eq!(error.exit_code(), expected_exit); + assert!(!error.to_string().contains("MUST-NOT-ECHO")); + assert!(!error.to_string().contains("user:secret")); + handle.join().expect("server thread should finish"); + } + } + + #[tokio::test] + async fn site_replication_repair_errors_have_distinct_redacted_categories() { + for (status, body, expected_exit, expected_text) in [ + ( + "400 Bad Request", + r#"{"Message":"execute requires a valid preflightToken"}"#, + 1, + "token or plan is invalid", + ), + ( + "409 Conflict", + r#"{"Message":"repair operation ID is already bound to a different preflight"}"#, + 6, + "bound to a different preflight", + ), + ( + "409 Conflict", + r#"{"Message":"another site replication repair is active"}"#, + 6, + "operation is active", + ), + ( + "412 Precondition Failed", + r#"{"Message":"site replication repair plan changed after partial execution"}"#, + 6, + "plan changed after partial execution", + ), + ( + "500 Internal Server Error", + r#"{"Message":"secret=MUST-NOT-ECHO"}"#, + 1, + "internal site-replication repair failure", + ), + ( + "403 Forbidden", + r#"{"Message":"secret=MUST-NOT-ECHO"}"#, + 4, + "Authentication failed", + ), + ] { + let (endpoint, _receiver, handle) = start_admin_test_server(status, body); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_execute( + SITE_REPLICATION_REPAIR_TOKEN, + SITE_REPLICATION_REPAIR_ID, + ) + .await + .expect_err("repair failure must remain typed"); + assert_eq!(error.exit_code(), expected_exit); + assert!(error.to_string().contains(expected_text)); + assert!(!error.to_string().contains("MUST-NOT-ECHO")); + handle.join().expect("server thread should finish"); + } + } + + #[tokio::test] + async fn site_replication_repair_enforces_success_and_error_body_bounds() { + let success_response = format!( + "HTTP/1.1 200 OK\r\ncontent-length: {}\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n", + MAX_SITE_REPLICATION_REPAIR_RESPONSE_BYTES + 1 + ) + .into_bytes(); + let (endpoint, handle) = start_admin_raw_response_server(success_response); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_status(SITE_REPLICATION_REPAIR_ID) + .await + .expect_err("oversized repair success must fail"); + assert!(matches!(error, Error::General(_))); + handle.join().expect("server thread should finish"); + + let error_response = format!( + "HTTP/1.1 500 Internal Server Error\r\ncontent-length: {}\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n", + MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES + 1 + ) + .into_bytes(); + let (endpoint, handle) = start_admin_raw_response_server(error_response); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_status(SITE_REPLICATION_REPAIR_ID) + .await + .expect_err("oversized repair error must fail"); + assert!(matches!(error, Error::General(_))); + handle.join().expect("server thread should finish"); + } + + #[tokio::test] + async fn site_replication_repair_execute_disconnect_is_not_retried() { + let (endpoint, receiver, handle) = start_admin_disconnect_server(); + let client = admin_client_for_endpoint(&endpoint); + let error = client + .site_replication_repair_execute( + SITE_REPLICATION_REPAIR_TOKEN, + SITE_REPLICATION_REPAIR_ID, + ) + .await + .expect_err("disconnected repair execute must fail"); + assert!(matches!(error, Error::Network(_))); + assert!(error.to_string().contains("outcome is unknown")); + let request = receiver.recv().expect("one execute request"); + assert_eq!(request.method, "PUT"); + assert_eq!(request.target, "/rustfs/admin/v3/site-replication/repair"); + handle.join().expect("server thread should finish"); + } + #[tokio::test] async fn site_replication_resync_sends_exact_start_query_and_complete_peer() { let (endpoint, receiver, handle) = diff --git a/schemas/output_v3.json b/schemas/output_v3.json index bf9a442c..c7e7f072 100644 --- a/schemas/output_v3.json +++ b/schemas/output_v3.json @@ -1065,6 +1065,108 @@ { "$ref": "#/definitions/oidcDeleteData" } ] }, + "siteRepairTask": { + "type": "object", + "required": ["taskId", "status"], + "properties": { + "taskId": { + "type": "string", + "pattern": "^[A-Za-z0-9_-]{43}$" + }, + "status": { + "type": "string", + "enum": ["planned", "running", "succeeded", "failed", "skipped"] + }, + "error": { + "type": "string", + "enum": [ + "authorization-failed", "remote-timeout", "remote-dns-failed", + "remote-tls-failed", "remote-connect-failed", "remote-operation-failed" + ] + } + }, + "additionalProperties": false + }, + "siteRepairFamily": { + "type": "object", + "required": ["planned", "succeeded", "failed", "retryEvents", "tasks", "errors"], + "properties": { + "planned": { "type": "integer", "minimum": 0 }, + "succeeded": { "type": "integer", "minimum": 0 }, + "failed": { "type": "integer", "minimum": 0 }, + "retryEvents": { "type": "integer", "minimum": 0 }, + "tasks": { + "type": "array", + "items": { "$ref": "#/definitions/siteRepairTask" } + }, + "errors": { + "type": "array", + "items": { + "type": "string", + "enum": [ + "authorization-failed", "remote-timeout", "remote-dns-failed", + "remote-tls-failed", "remote-connect-failed", "remote-operation-failed" + ] + } + } + }, + "additionalProperties": false + }, + "siteRepairSite": { + "type": "object", + "required": ["deploymentId", "name", "families"], + "properties": { + "deploymentId": { "type": "string", "minLength": 1 }, + "name": { "type": "string", "minLength": 1 }, + "families": { + "type": "object", + "additionalProperties": { "$ref": "#/definitions/siteRepairFamily" } + } + }, + "additionalProperties": false + }, + "siteRepairSites": { + "type": "object", + "additionalProperties": { "$ref": "#/definitions/siteRepairSite" } + }, + "siteRepairPreflight": { + "type": "object", + "required": ["mode", "status", "preflightToken", "retryEvents", "sites"], + "properties": { + "mode": { "const": "dry-run" }, + "status": { "const": "planned" }, + "preflightToken": { + "type": "string", + "pattern": "^[A-Za-z0-9_-]{43}$" + }, + "retryEvents": { "type": "integer", "minimum": 0 }, + "sites": { "$ref": "#/definitions/siteRepairSites" } + }, + "additionalProperties": false + }, + "siteRepairOperation": { + "type": "object", + "required": [ + "mode", "operationId", "status", "sites", + "createdAt", "updatedAt", "completedAt" + ], + "properties": { + "mode": { "const": "execute" }, + "operationId": { + "type": "string", + "pattern": "^[0-9A-Fa-f]{8}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{4}-[0-9A-Fa-f]{12}$" + }, + "status": { + "type": "string", + "enum": ["running", "success", "partial", "failed"] + }, + "sites": { "$ref": "#/definitions/siteRepairSites" }, + "createdAt": { "$ref": "#/definitions/timestamp" }, + "updatedAt": { "$ref": "#/definitions/timestamp" }, + "completedAt": { "$ref": "#/definitions/nullableTimestamp" } + }, + "additionalProperties": false + }, "adminOperation": { "type": "object", "required": [ @@ -1080,7 +1182,40 @@ "operation_id": { "$ref": "#/definitions/nullableString" }, "changed": { "type": "boolean" }, "result": { "type": ["object", "null"] } - } + }, + "allOf": [ + { + "if": { + "properties": { + "operation": { "const": "site_replication_repair_dry_run" } + }, + "required": ["operation"] + }, + "then": { + "properties": { + "result": { "$ref": "#/definitions/siteRepairPreflight" } + } + } + }, + { + "if": { + "properties": { + "operation": { + "enum": [ + "site_replication_repair_execute", + "site_replication_repair_status" + ] + } + }, + "required": ["operation"] + }, + "then": { + "properties": { + "result": { "$ref": "#/definitions/siteRepairOperation" } + } + } + } + ] }, "adminOperationsData": { "type": "object",