diff --git a/.config/nextest.toml b/.config/nextest.toml index e5d7e9d68..d17398000 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -211,7 +211,7 @@ test-group = 'ecstore-serial-flaky' # the nightly profile derives its set as "the replication module MINUS this # allowlist", so any new replication test lands in nightly by default (never # silently unrun) until it is explicitly blessed as fast here. Keep the two -# regexes byte-identical. Count invariant: 20 here + 28 nightly = 48 total +# regexes byte-identical. Count invariant: 20 here + 36 nightly = 56 total # (authority: `cargo nextest list`; docs/testing/e2e-suite-inventory.md). # HISTORY (2026-07-11): the 20 fast tests were briefly pulled out of this lane # (#4724) because they set a loopback (127.0.0.1) replication target that the @@ -273,10 +273,10 @@ slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" } # tests that are unfit for the per-PR e2e-smoke gate: # # * 2 remote-target TLS validation tests. -# * 12 bucket-replication data-plane/helper tests — they PUT/delete objects -# and poll until source and target converge; two replicate over HTTPS, two -# pin active SSE failure contracts, and one guards event/history observers. -# The SSE-S3 contract remains ignored under backlog#1291. +# * 13 bucket-replication data-plane/helper tests — they PUT/delete objects +# and poll until source and target converge; two replicate over HTTPS, +# four pin active SSE fail-closed contracts (SSE-C, SSE-S3, SSE-KMS, and +# the SSE-S3 resync path), and one guards event/history observers. # * 12 `_real_dual_node` site-replication tests — each spawns TWO full rustfs # servers and drives the cross-process site-replication control plane. # * 1 `_real_three_node` site-replication test. diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index f0f467927..c5023d64c 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -363,19 +363,21 @@ impl Drop for SlowReplicationTargetGuard { } } +// Mirrors madmin-go `ResyncTargetsInfo`/`ResyncTarget` json tags — the same +// shape `mc replicate resync status` decodes. #[derive(Debug, Clone, serde::Deserialize)] struct ReplicationResetStatusResponse { - #[serde(rename = "Targets", default)] + #[serde(rename = "target", default)] targets: Vec, } #[derive(Debug, Clone, serde::Deserialize)] struct ReplicationResetStatusTarget { - #[serde(rename = "Arn", default)] + #[serde(rename = "arn", default)] arn: String, - #[serde(rename = "ResetID", default)] + #[serde(rename = "resetid", default)] reset_id: String, - #[serde(rename = "Status", default)] + #[serde(rename = "resyncStatus", default)] status: String, } @@ -1655,19 +1657,30 @@ async fn wait_for_source_delete_marker_replication_failed( if response.status() != StatusCode::OK { return Err(format!("replication diff failed with status {}", response.status()).into()); } - let diff: serde_json::Value = response.json().await?; - let failed = diff["Entries"].as_array().is_some_and(|entries| { - entries.iter().any(|entry| { - entry["Object"].as_str() == Some(key) - && entry["IsDeleteMarker"].as_bool() == Some(true) - && entry["ReplicationStatus"].as_str() == Some("FAILED") - }) + // The default diff response is a madmin-style stream of bare DiffInfo + // JSON documents (one per line) with no envelope; assert the envelope + // is gone so an aggregate-shaped regression fails loudly here. + let body = response.text().await?; + let entries = body + .lines() + .filter(|line| !line.trim().is_empty()) + .map(serde_json::from_str::) + .collect::, _>>()?; + for entry in &entries { + if entry.get("Entries").is_some() { + return Err(format!("replication diff must stream bare DiffInfo documents, got envelope: {entry}").into()); + } + } + let failed = entries.iter().any(|entry| { + entry["object"].as_str() == Some(key) + && entry["deletemarker"].as_bool() == Some(true) + && entry["rStatus"].as_str() == Some("FAILED") }); if failed { return Ok(()); } if tokio::time::Instant::now() >= deadline { - return Err(format!("source delete marker {key} never reported FAILED; last diff={diff}").into()); + return Err(format!("source delete marker {key} never reported FAILED; last diff={body}").into()); } sleep(Duration::from_millis(200)).await; } @@ -2276,6 +2289,30 @@ async fn site_replication_state_edit( Ok(()) } +/// Start a bucket-level replication resync (`PUT ?replication-reset`) and +/// return the target `(arn, reset_id)`, asserting the response carries the +/// madmin `ResyncTargetsInfo` shape (`target[0].arn` / `target[0].resetid`) +/// that `mc replicate resync start` decodes. +async fn start_bucket_replication_reset( + env: &RustFSTestEnvironment, + bucket: &str, +) -> Result<(String, String), Box> { + let url = format!("{}/{bucket}?replication-reset", env.url); + let response = signed_request(http::Method::PUT, &url, &env.access_key, &env.secret_key, None, None).await?; + if response.status() != StatusCode::OK { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(format!("replication reset start failed: {status} {body}").into()); + } + let payload: serde_json::Value = response.json().await?; + let arn = payload["target"][0]["arn"].as_str().unwrap_or_default().to_string(); + let reset_id = payload["target"][0]["resetid"].as_str().unwrap_or_default().to_string(); + if arn.is_empty() || reset_id.is_empty() { + return Err(format!("replication reset response missing madmin target[0].arn/resetid: {payload}").into()); + } + Ok((arn, reset_id)) +} + async fn get_replication_reset_status( env: &RustFSTestEnvironment, bucket: &str, @@ -3938,7 +3975,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR matrix-prefix - 100 + 110 Enabled prefix/ Enabled @@ -3949,7 +3986,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR matrix-both-prefix - 100 + 120 Enabled both/ Enabled @@ -3959,7 +3996,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR matrix-tag - 100 + 130 Enabled routetagged Disabled @@ -3969,7 +4006,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR matrix-disabled - 100 + 140 Disabled disabled/ Enabled @@ -4451,16 +4488,76 @@ async fn test_bucket_replication_sse_c_contract() -> TestResult { } /// backlog#1147 repl-17 / backlog#1291: SSE-S3 must fail closed until managed -/// encryption is supported on the target. The current plaintext replication is -/// a known security bug, so this pins the required contract without blessing it. +/// encryption is supported on the target. The silent plaintext replication +/// that originally kept this test ignored was fixed by the fail-closed gate in +/// `crates/ecstore/src/bucket/replication/replication_target_boundary.rs` +/// (all replication modes route through it), so this now pins the current +/// fail-closed contract: FAILED status, failure event, readable source, and a +/// stable absence of all target versions. #[tokio::test] #[serial] -#[ignore = "backlog#1291: SSE-S3 replication silently drops encryption"] async fn test_bucket_replication_sse_s3_contract() -> TestResult { init_logging(); assert_managed_sse_replication_fails_explicitly("sse-s3", false).await } +/// P1-22 stage 0: the existing-object resync path must fail closed for +/// managed-SSE objects exactly like inline replication (which +/// `test_bucket_replication_sse_s3_contract` pins, including the scanner heal +/// re-drive). Resync re-drives every object version through the same +/// fail-closed target boundary, so a resync over an encrypted bucket must +/// terminate without ever materializing a plaintext (or unreadable) replica; +/// the post-resync stays-absent window also spans further fast-scanner heal +/// cycles. +#[tokio::test] +#[serial] +async fn test_bucket_replication_sse_s3_resync_stays_fail_closed() -> TestResult { + init_logging(); + + let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("sse-resync", true).await?; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + let key = "sse-resync-contract.txt"; + let body = b"repl-22 sse resync payload".to_vec(); + + source_client + .put_object() + .bucket(&source_bucket) + .key(key) + .body(ByteStream::from(body.clone())) + .server_side_encryption(ServerSideEncryption::Aes256) + .send() + .await?; + wait_for_source_replication_status(&source_client, &source_bucket, key, "FAILED", false).await?; + + // Resync: drive the existing-object resync path over the failed object. + let (target_arn, reset_id) = start_bucket_replication_reset(&source_env, &source_bucket).await?; + let terminal = wait_for_replication_reset_target(&source_env, &source_bucket, &target_arn, |target| { + target.reset_id == reset_id && matches!(target.status.as_str(), "Completed" | "Failed") + }) + .await?; + assert_eq!(terminal.reset_id, reset_id); + + // The resync pass must have failed closed: still no target version (the + // window also spans further scanner heal cycles), and the source object + // stays readable and encrypted. + assert_failed_replication_stays_absent_for( + &source_client, + &source_bucket, + &target_client, + &target_bucket, + key, + false, + Duration::from_secs(5), + ) + .await?; + let source = source_client.get_object().bucket(&source_bucket).key(key).send().await?; + assert_eq!(source.server_side_encryption(), Some(&ServerSideEncryption::Aes256)); + assert_eq!(source.body.collect().await?.into_bytes().as_ref(), body.as_slice()); + + Ok(()) +} + /// backlog#1147 repl-17: SSE-KMS currently fails closed rather than creating an /// unreadable replica; the shared helper verifies FAILED, the failure event, /// source readability, and a stable absence of all target versions. diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 84caf7f26..f4938ea95 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -185,19 +185,19 @@ pub mod bucket { MustReplicateOptions, ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATE_INCOMING_DELETE, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, ReplicateDecision, ReplicateObjectInfo, - ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigurationExt, ReplicationDeleteScheduleInput, - ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, - ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, - ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, - ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus, - VersionPurgeStatusType, commit_force_delete_intent, complete_force_delete_intent, + ReplicationBatchAdmission, ReplicationConfig, ReplicationConfigStructureError, ReplicationConfigurationExt, + ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, + ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, + ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, + ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, + TargetReplicationResyncStatus, VersionPurgeStatusType, commit_force_delete_intent, complete_force_delete_intent, delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, get_global_replication_stats, init_background_replication, invalid_replication_config_status_field, persist_force_delete_intent, read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns, resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, unsupported_replication_config_field, - validate_replication_config_target_arns, version_purge_status_to_filemeta, + validate_replication_config_structure, validate_replication_config_target_arns, version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index 1d07e99c7..68f43a32b 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -100,11 +100,41 @@ paths. behind the ECStore replication facade; only `rustfs/src/app/storage_api.rs` may retain direct object/delete replication helper calls. -## First Code-Bearing Step +## Completion Criteria -Start with `ReplicationRuntime` or `ReplicationEventSink`. Both can be added as -narrow internal contracts while keeping current queue, MRF, resync, and target -behavior unchanged. Do not start with a crate move. +The split is complete when the "Current dependency to remove" column in the +Required Contracts table above is empty: every row is either deleted because +the dependency is gone, or reduced to "none". No other signal — file count, +boundary count, line count — measures completion. + +Target end state: + +- `replication_pool.rs`, `replication_resyncer.rs`, and `replication_state.rs` + move into `crates/replication` behind the contracts above; +- the `*_boundary.rs` and `*_bridge.rs` micro-files dissolve naturally as the + code they fence moves across the crate boundary. They are the mechanical + seams of the migration ratchet — the architecture guard scripts anchor on + their file names — so batch-merging them beforehand is explicitly rejected: + it forces synchronized guard-script/mod/import churn with zero functional + gain; +- the only module that can retire early is `datatypes.rs`: delete it once its + facade consumers import the resync status enums through `rustfs-replication` + directly. + +## Milestones + +| Milestone | Scope | Status | +|---|---|---| +| M0 | Record the completion criteria and end state (this section). | Done | +| M1 | Contract extraction: resync/queue/stats/object-decision/filemeta/storage wire contracts owned by `crates/replication`; ECStore imports concentrated in `*_boundary.rs`; event sink and runtime access behind local contracts. | Done — see Required Contracts | +| M2 | Move resyncer pure decision logic (no IO) into `crates/replication`. | Pending; sequence after splitting the oversized resyncer/pool functions (`resync_bucket`, `replicate_all`, `start_mrf_processor`) so moves stay mechanical | +| M3 | Move the worker runtime (`replication_pool.rs`, the IO paths of `replication_resyncer.rs`, `replication_state.rs`) once the contract traits are stable. Highest-risk step of the whole plan; do it last. | Pending | +| M4 | Retire the boundary modules together with their guard-script entries; delete `datatypes.rs`. | Pending | + +The original first code-bearing step (narrow `ReplicationEventSink` / +`ReplicationRuntime` contracts) has landed — `replication_event_sink.rs` +exists and runtime access goes through local boundary aliases — so new work +starts from M2. Current compatibility guard: `crates/ecstore/tests/replication_facade_compat_test.rs` keeps the ECStore replication facade types covered while architecture rules diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 19cb5f702..c1ae1345c 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -47,9 +47,9 @@ pub use datatypes::ResyncStatusType; pub use replication_config_boundary::{ ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, - ReplicationConfigurationExt, ReplicationTargetValidationError, invalid_replication_config_status_field, - replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, - validate_replication_config_target_arns, + ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, + invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, + unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; pub use replication_filemeta_boundary::{ diff --git a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs index 04d53883f..484fe7a25 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs @@ -15,7 +15,7 @@ pub use rustfs_replication::{ ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, - ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, invalid_replication_config_status_field, - replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, - validate_replication_config_target_arns, + ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationRuleExt, ReplicationTargetValidationError, + invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, + unsupported_replication_config_field, validate_replication_config_structure, validate_replication_config_target_arns, }; diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index ae56fec17..5a89e0fb2 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -89,3 +89,116 @@ pub fn replication_state_to_filemeta(state: &ReplicationState) -> rustfs_filemet target_delete_marker_version_ids_corrupt: state.target_delete_marker_version_ids_corrupt, } } + +// Reconciliation tests for the deliberately duplicated wire types. +// +// `rustfs-filemeta` (xl.meta disk format) and `rustfs-replication` (MRF/resync +// persistence format) each own a copy of `ReplicationStatusType`, +// `VersionPurgeStatusType` and `ReplicationState`; the conversions above hop +// between them via `as_str()`, whose `From<&str>` impls fall back to `Empty` +// on any unknown token. That fallback silently degrades data the moment one +// side gains a variant the other lacks, so these tests pin the two sides +// together: +// +// - the `match` statements are exhaustive with no `_` arm — adding a variant +// on either side fails compilation here until the mapping is reconsidered; +// - the round-trips assert the string token survives both directions — a +// variant whose token the other side does not recognize fails the assert +// instead of quietly becoming `Empty`. +// +// Struct-shaped drift on `ReplicationState` is already compile-guarded by the +// exhaustive struct literals in the two conversion functions above; the +// round-trip test below additionally pins value fidelity for every field. +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashMap; + + #[test] + fn replication_status_variants_round_trip_across_boundary() { + use rustfs_replication::ReplicationStatusType as Repl; + + let all = [ + Repl::Pending, + Repl::Completed, + Repl::CompletedLegacy, + Repl::Failed, + Repl::Replica, + Repl::Empty, + ]; + for status in all { + // Exhaustive on the replication side: a new variant breaks this match. + match status { + Repl::Pending | Repl::Completed | Repl::CompletedLegacy | Repl::Failed | Repl::Replica | Repl::Empty => {} + } + let filemeta = replication_status_to_filemeta(status.clone()); + assert_eq!( + filemeta.as_str(), + status.as_str(), + "replication->filemeta conversion must not degrade {status:?} (unknown tokens fall back to Empty)" + ); + assert_eq!(replication_status_from_filemeta(filemeta), status); + } + + // Exhaustive on the filemeta side: a new variant breaks this match. + fn _filemeta_side_is_covered(status: rustfs_filemeta::ReplicationStatusType) { + use rustfs_filemeta::ReplicationStatusType as Meta; + match status { + Meta::Pending | Meta::Completed | Meta::CompletedLegacy | Meta::Failed | Meta::Replica | Meta::Empty => {} + } + } + } + + #[test] + fn version_purge_status_variants_round_trip_across_boundary() { + use rustfs_replication::VersionPurgeStatusType as Repl; + + let all = [Repl::Pending, Repl::Complete, Repl::Failed, Repl::Empty]; + for status in all { + // Exhaustive on the replication side: a new variant breaks this match. + match status { + Repl::Pending | Repl::Complete | Repl::Failed | Repl::Empty => {} + } + let filemeta = version_purge_status_to_filemeta(status.clone()); + assert_eq!( + filemeta.as_str(), + status.as_str(), + "replication->filemeta conversion must not degrade {status:?} (unknown tokens fall back to Empty)" + ); + assert_eq!(version_purge_status_from_filemeta(filemeta), status); + } + + // Exhaustive on the filemeta side: a new variant breaks this match. + fn _filemeta_side_is_covered(status: rustfs_filemeta::VersionPurgeStatusType) { + use rustfs_filemeta::VersionPurgeStatusType as Meta; + match status { + Meta::Pending | Meta::Complete | Meta::Failed | Meta::Empty => {} + } + } + } + + #[test] + fn replication_state_round_trips_every_field_across_boundary() { + let timestamp = time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid timestamp"); + let state = ReplicationState { + replica_timestamp: Some(timestamp), + replica_status: ReplicationStatusType::Replica, + delete_marker: true, + replication_timestamp: Some(timestamp), + replication_status_internal: Some("arn:a=PENDING;".to_string()), + version_purge_status_internal: Some("arn:a=FAILED;".to_string()), + replicate_decision_str: "arn:a=true;false;;".to_string(), + targets: HashMap::from([ + ("arn:a".to_string(), ReplicationStatusType::Completed), + ("arn:b".to_string(), ReplicationStatusType::Failed), + ]), + purge_targets: HashMap::from([("arn:a".to_string(), VersionPurgeStatusType::Pending)]), + reset_statuses_map: HashMap::from([("reset-arn:a".to_string(), "reset-id;ts".to_string())]), + target_delete_marker_version_ids: HashMap::from([("arn:a".to_string(), "version-1".to_string())]), + target_delete_marker_version_ids_corrupt: true, + }; + + let round_tripped = replication_state_from_filemeta(&replication_state_to_filemeta(&state)); + assert_eq!(round_tripped, state); + } +} diff --git a/crates/filemeta/src/replication.rs b/crates/filemeta/src/replication.rs index 3b008bc4b..b34c01751 100644 --- a/crates/filemeta/src/replication.rs +++ b/crates/filemeta/src/replication.rs @@ -48,6 +48,14 @@ pub const REPLICATE_HEAL: &str = "replicate:heal"; pub const REPLICATE_HEAL_DELETE: &str = "replicate:heal:delete"; /// StatusType of Replication for x-amz-replication-status header +/// +/// NOTE: `rustfs-replication` owns a sibling copy of this enum (plus +/// `VersionPurgeStatusType` and `ReplicationState`) bound to the MRF/resync +/// persistence format, while this copy is bound to the xl.meta disk format. +/// When adding or renaming a variant here, reconcile the sibling and the +/// conversion layer — the reconciliation tests in +/// `crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs` +/// fail to compile until both sides agree. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Hash)] pub enum ReplicationStatusType { /// Pending - replication is pending. diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index 2c79534d3..baeaa993c 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -32,6 +32,12 @@ pub const REPLICATION_CAPABILITY_CONTRACT_VERSION: u32 = 1; // clients should keep omitting it, but the validator tolerates an explicit // `STANDARD` as a no-op (see `unsupported_replication_config_field`) because the // console's rule form always sends it. +// +// Contract note: rule-level `Destination.StorageClass` is never consumed by the +// replication engine (MinIO's engine likewise reads only the target-level +// storage class). To control the storage class of replicated objects, set the +// remote target's `storage_class` field (set-remote-target API), which RustFS +// does apply on replication PUTs. pub const REPLICATION_WRITABLE_FIELDS: &[&str] = &[ "Role", "Rule.ID", @@ -297,6 +303,119 @@ pub fn should_remove_replication_target( is_replication_target && config_target_arns.contains(target_arn) } +/// Maximum number of rules accepted in one replication configuration, +/// matching MinIO's `replication.Config.Validate` limit. +pub const REPLICATION_CONFIG_MAX_RULES: usize = 1000; + +/// Maximum length of a replication rule ID, matching the S3 schema. +pub const REPLICATION_CONFIG_MAX_RULE_ID_LEN: usize = 255; + +/// A structural defect in a replication configuration, detected before the +/// configuration is persisted. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReplicationConfigStructureError { + NoRules, + TooManyRules, + NegativeRulePriority, + DuplicateRulePriority, + RuleIdTooLong, + AmbiguousRuleFilter, + TagFilterWithDeleteMarkerReplication, +} + +impl ReplicationConfigStructureError { + pub fn message(self) -> &'static str { + match self { + Self::NoRules => "replication configuration must contain at least one rule", + Self::TooManyRules => "replication configuration cannot contain more than 1000 rules", + Self::NegativeRulePriority => "replication rule Priority must be zero or a positive integer", + Self::DuplicateRulePriority => "replication rule Priority must be unique across rules", + Self::RuleIdTooLong => "replication rule ID cannot be longer than 255 characters", + Self::AmbiguousRuleFilter => "replication rule Filter must specify only one of Prefix, Tag or And", + Self::TagFilterWithDeleteMarkerReplication => { + "delete marker replication cannot be enabled on a rule with a Tag filter" + } + } + } +} + +fn filter_and_operator_is_set(and: &s3s::dto::ReplicationRuleAndOperator) -> bool { + and.prefix.as_ref().is_some_and(|prefix| !prefix.is_empty()) || and.tags.as_ref().is_some_and(|tags| !tags.is_empty()) +} + +/// Structural validation of a replication configuration, mirroring the checks +/// MinIO's `replication.Config.Validate` performs before persisting: at least +/// one rule, at most [`REPLICATION_CONFIG_MAX_RULES`], non-negative and unique +/// per-rule priorities (a missing Priority counts as 0, like Go's zero value), +/// rule IDs within [`REPLICATION_CONFIG_MAX_RULE_ID_LEN`] bytes, a Filter +/// carrying only one of Prefix/Tag/And, and delete marker replication +/// disabled on rules with a direct `Filter.Tag`. Tags inside `Filter.And` do +/// NOT trigger the delete-marker check — MinIO only inspects the direct tag, +/// and `mc replicate add --tags "k1=v1&k2=v2"` (delete-marker replication on +/// by default) puts multiple tags into `And.Tags`, so rejecting that shape +/// would break mc-generated configs that MinIO accepts. +/// +/// This is shape-only validation: capability gating lives in +/// [`unsupported_replication_config_field`]/[`invalid_replication_config_status_field`], +/// and the self-target ("same target") rejection is enforced when the remote +/// target itself is created, so a config can never reference a self-pointing +/// ARN. +pub fn validate_replication_config_structure( + config: &ReplicationConfiguration, +) -> std::result::Result<(), ReplicationConfigStructureError> { + if config.rules.is_empty() { + return Err(ReplicationConfigStructureError::NoRules); + } + if config.rules.len() > REPLICATION_CONFIG_MAX_RULES { + return Err(ReplicationConfigStructureError::TooManyRules); + } + + let mut priorities = HashSet::new(); + for rule in &config.rules { + let priority = rule.priority.unwrap_or(0); + if priority < 0 { + return Err(ReplicationConfigStructureError::NegativeRulePriority); + } + if !priorities.insert(priority) { + return Err(ReplicationConfigStructureError::DuplicateRulePriority); + } + + // Byte length, matching Go's `len(r.ID) > 255` in MinIO. + if rule + .id + .as_ref() + .is_some_and(|id| id.len() > REPLICATION_CONFIG_MAX_RULE_ID_LEN) + { + return Err(ReplicationConfigStructureError::RuleIdTooLong); + } + + if let Some(filter) = &rule.filter { + let has_and = filter.and.as_ref().is_some_and(filter_and_operator_is_set); + let has_prefix = filter.prefix.as_ref().is_some_and(|prefix| !prefix.is_empty()); + // An empty element (no key) counts as absent, matching + // MinIO's Tag.IsEmpty(); console form serializers emit empty tags. + let has_tag = filter + .tag + .as_ref() + .is_some_and(|tag| tag.key.as_ref().is_some_and(|key| !key.is_empty())); + if usize::from(has_and) + usize::from(has_prefix) + usize::from(has_tag) > 1 { + return Err(ReplicationConfigStructureError::AmbiguousRuleFilter); + } + + let delete_marker_replication_enabled = rule + .delete_marker_replication + .as_ref() + .and_then(|delete_marker| delete_marker.status.as_ref()) + .is_some_and(|status| status.as_str() == DeleteMarkerReplicationStatus::ENABLED); + if delete_marker_replication_enabled && has_tag { + return Err(ReplicationConfigStructureError::TagFilterWithDeleteMarkerReplication); + } + } + } + + Ok(()) +} + impl ReplicationConfigurationExt for ReplicationConfiguration { /// Check whether any object-replication rules exist fn has_existing_object_replication(&self, arn: &str) -> (bool, bool) { @@ -565,6 +684,202 @@ mod tests { } } + fn structure_config(rules: Vec) -> ReplicationConfiguration { + ReplicationConfiguration { + role: String::new(), + rules, + } + } + + fn tag_filter() -> s3s::dto::ReplicationRuleFilter { + s3s::dto::ReplicationRuleFilter { + tag: Some(s3s::dto::Tag { + key: Some("k".to_string()), + value: Some("v".to_string()), + }), + ..Default::default() + } + } + + #[test] + fn structure_validation_accepts_multi_rule_config_with_unique_priorities() { + let mut second = replication_rule("rule-2", "arn:target:a"); + second.priority = Some(2); + second.filter = Some(s3s::dto::ReplicationRuleFilter { + and: Some(s3s::dto::ReplicationRuleAndOperator { + prefix: Some("photos/".to_string()), + tags: Some(vec![s3s::dto::Tag { + key: Some("k".to_string()), + value: Some("v".to_string()), + }]), + }), + ..Default::default() + }); + let config = structure_config(vec![replication_rule("rule-1", "arn:target:a"), second]); + + assert_eq!(validate_replication_config_structure(&config), Ok(())); + } + + #[test] + fn structure_validation_rejects_empty_rule_list() { + let config = structure_config(Vec::new()); + + assert_eq!( + validate_replication_config_structure(&config), + Err(ReplicationConfigStructureError::NoRules) + ); + } + + #[test] + fn structure_validation_rejects_more_than_max_rules() { + let rules = (0..=REPLICATION_CONFIG_MAX_RULES as i32) + .map(|priority| { + let mut rule = replication_rule(&format!("rule-{priority}"), "arn:target:a"); + rule.priority = Some(priority); + rule + }) + .collect(); + + assert_eq!( + validate_replication_config_structure(&structure_config(rules)), + Err(ReplicationConfigStructureError::TooManyRules) + ); + } + + #[test] + fn structure_validation_rejects_duplicate_priorities() { + let config = structure_config(vec![ + replication_rule("rule-1", "arn:target:a"), + replication_rule("rule-2", "arn:target:a"), + ]); + + assert_eq!( + validate_replication_config_structure(&config), + Err(ReplicationConfigStructureError::DuplicateRulePriority) + ); + } + + #[test] + fn structure_validation_treats_missing_priority_as_zero_for_uniqueness() { + let mut first = replication_rule("rule-1", "arn:target:a"); + first.priority = None; + let mut second = replication_rule("rule-2", "arn:target:a"); + second.priority = None; + + assert_eq!( + validate_replication_config_structure(&structure_config(vec![first, second])), + Err(ReplicationConfigStructureError::DuplicateRulePriority) + ); + } + + #[test] + fn structure_validation_rejects_negative_priority() { + let mut rule = replication_rule("rule-1", "arn:target:a"); + rule.priority = Some(-1); + + assert_eq!( + validate_replication_config_structure(&structure_config(vec![rule])), + Err(ReplicationConfigStructureError::NegativeRulePriority) + ); + } + + #[test] + fn structure_validation_rejects_rule_id_longer_than_255_chars() { + let mut rule = replication_rule(&"x".repeat(REPLICATION_CONFIG_MAX_RULE_ID_LEN + 1), "arn:target:a"); + rule.priority = Some(1); + + assert_eq!( + validate_replication_config_structure(&structure_config(vec![rule])), + Err(ReplicationConfigStructureError::RuleIdTooLong) + ); + } + + #[test] + fn structure_validation_rejects_filter_with_both_prefix_and_tag() { + let mut rule = replication_rule("rule-1", "arn:target:a"); + let mut filter = tag_filter(); + filter.prefix = Some("photos/".to_string()); + rule.filter = Some(filter); + + assert_eq!( + validate_replication_config_structure(&structure_config(vec![rule])), + Err(ReplicationConfigStructureError::AmbiguousRuleFilter) + ); + } + + #[test] + fn structure_validation_rejects_delete_marker_replication_on_tag_filtered_rule() { + let mut rule = replication_rule("rule-1", "arn:target:a"); + rule.delete_marker_replication = Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }); + rule.filter = Some(tag_filter()); + + assert_eq!( + validate_replication_config_structure(&structure_config(vec![rule])), + Err(ReplicationConfigStructureError::TagFilterWithDeleteMarkerReplication) + ); + } + + #[test] + fn structure_validation_treats_empty_tag_element_as_absent() { + // MinIO's Tag.IsEmpty() ignores an empty element; the console's + // form serializer emits them, so prefix + empty tag must stay valid + // and an empty tag must not trip the delete-marker check. + let mut rule = replication_rule("rule-1", "arn:target:a"); + rule.delete_marker_replication = Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }); + rule.filter = Some(s3s::dto::ReplicationRuleFilter { + prefix: Some("photos/".to_string()), + tag: Some(s3s::dto::Tag { key: None, value: None }), + ..Default::default() + }); + + assert_eq!(validate_replication_config_structure(&structure_config(vec![rule])), Ok(())); + } + + #[test] + fn structure_validation_allows_delete_marker_replication_with_and_tags() { + // mc `replicate add --tags "k1=v1&k2=v2"` puts multiple tags into + // Filter.And.Tags and enables delete-marker replication by default; + // MinIO's validator only inspects the direct Filter.Tag, so this + // shape must stay accepted for mc interop. + let mut rule = replication_rule("rule-1", "arn:target:a"); + rule.delete_marker_replication = Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }); + rule.filter = Some(s3s::dto::ReplicationRuleFilter { + and: Some(s3s::dto::ReplicationRuleAndOperator { + prefix: None, + tags: Some(vec![ + s3s::dto::Tag { + key: Some("k1".to_string()), + value: Some("v1".to_string()), + }, + s3s::dto::Tag { + key: Some("k2".to_string()), + value: Some("v2".to_string()), + }, + ]), + }), + ..Default::default() + }); + + assert_eq!(validate_replication_config_structure(&structure_config(vec![rule])), Ok(())); + } + + #[test] + fn structure_validation_allows_tag_filter_when_delete_marker_replication_disabled() { + let mut rule = replication_rule("rule-1", "arn:target:a"); + rule.delete_marker_replication = Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)), + }); + rule.filter = Some(tag_filter()); + + assert_eq!(validate_replication_config_structure(&structure_config(vec![rule])), Ok(())); + } + #[test] fn filter_target_arns_uses_role_when_role_is_present() { let config = ReplicationConfiguration { diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index e6a57cac6..68064dc19 100644 --- a/crates/replication/src/filemeta.rs +++ b/crates/replication/src/filemeta.rs @@ -48,6 +48,14 @@ pub const REPLICATE_HEAL: &str = "replicate:heal"; pub const REPLICATE_HEAL_DELETE: &str = "replicate:heal:delete"; /// StatusType of Replication for x-amz-replication-status header +/// +/// NOTE: `rustfs-filemeta` owns a sibling copy of this enum (plus +/// `VersionPurgeStatusType` and `ReplicationState`) bound to the xl.meta disk +/// format, while this copy is bound to the MRF/resync persistence format. +/// When adding or renaming a variant here, reconcile the sibling and the +/// conversion layer — the reconciliation tests in +/// `crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs` +/// fail to compile until both sides agree. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Hash)] pub enum ReplicationStatusType { /// Pending - replication is pending. diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 6c4ec5620..bb0c08385 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -31,9 +31,10 @@ pub mod tagging; pub use config::{ ObjectOpts, REMOTE_TARGET_CAPABILITY_CONTRACT_VERSION, REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, REPLICATION_CAPABILITY_CONTRACT_VERSION, REPLICATION_READ_ONLY_HISTORICAL_FIELDS, REPLICATION_WRITABLE_FIELDS, - ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, - invalid_replication_config_status_field, replication_target_arns, should_remove_replication_target, - unsupported_replication_config_field, validate_replication_config_target_arns, + ReplicationConfigStructureError, ReplicationConfigurationExt, ReplicationTargetValidationError, + active_replication_rule_destination_arns, invalid_replication_config_status_field, replication_target_arns, + should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure, + validate_replication_config_target_arns, }; pub use delete::{ DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication, diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 913cea48b..4db943dba 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -10,7 +10,7 @@ and rollback steps. | Area | Current owner | Size | Split status | |---|---|---:|---| | Bucket lifecycle | `crates/lifecycle/` + `crates/ecstore/src/bucket/lifecycle/` | core contracts + ECStore runtime | Core contract extracted | -| Bucket replication | `crates/ecstore/src/bucket/replication/` | 8,730 lines | Proposal only | +| Bucket replication | `crates/ecstore/src/bucket/replication/` | 8,730 lines | Contracts extracted; runtime move pending | | Set disks | `crates/ecstore/src/set_disk/` | state carrier plus operation modules | Keep in ECStore | | Public ECStore facade | `crates/ecstore/src/api/mod.rs` | broad compatibility surface | Shrink only through guarded PRs | @@ -117,6 +117,14 @@ Focused verification for the first code-bearing lifecycle PR: status wire format. The remaining `bucket/replication` worker runtime is not ready for a full standalone crate yet. +The completion criteria and milestone sequence for this candidate (when the +split counts as done, the target end state, and the order of the remaining +moves) live in the module inventory: +`crates/ecstore/src/bucket/replication/README.md`, sections "Completion +Criteria" and "Milestones". The originally proposed first code-bearing step +(event sink / runtime contracts) has landed; remaining work starts from moving +resyncer pure decision logic. + Current coupling: - replication workers depend on `ReplicationStorage`, ECStore object APIs and diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index be7d4a8ba..196c0df3b 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -886,30 +886,37 @@ const REPLICATION_DIFF_MAX_SCAN: usize = 10_000; const REPLICATION_DIFF_PAGE_SIZE: i32 = 1_000; /// A single object version whose replication is not yet complete, reported by -/// `POST /v3/replication/diff`. Field names mirror MinIO's `madmin.DiffInfo` -/// so MinIO-compatible admin clients can parse the response. +/// `POST /v3/replication/diff`. Field names are the exact json tags of +/// madmin-go `DiffInfo` (replication-api.go), which `mc replicate diff` +/// decodes one JSON document at a time from the response body. `Size` is a +/// RustFS extension key with no madmin counterpart; Go decoders ignore +/// unknown keys. #[derive(Debug, Serialize)] struct ReplicationDiffEntry { - #[serde(rename = "Object")] + #[serde(rename = "object")] object: String, - #[serde(rename = "VersionID", skip_serializing_if = "Option::is_none")] + #[serde(rename = "versionId", skip_serializing_if = "Option::is_none")] version_id: Option, #[serde(rename = "Size")] size: i64, - #[serde(rename = "IsDeleteMarker")] + #[serde(rename = "deletemarker")] is_delete_marker: bool, - #[serde(rename = "ReplicationStatus")] + #[serde(rename = "rStatus")] replication_status: String, - #[serde(rename = "LastModified", skip_serializing_if = "Option::is_none")] + #[serde(rename = "lastModified", skip_serializing_if = "Option::is_none")] last_modified: Option, } -/// Response body for `POST /v3/replication/diff`. +/// Aggregate response body for `POST /v3/replication/diff?aggregate=true`. /// -/// `entries` lists object versions with a `PENDING` or `FAILED` replication -/// status. `is_truncated` indicates the on-demand scan hit -/// [`REPLICATION_DIFF_MAX_SCAN`] before reaching the end of the bucket, so the -/// diff is partial and should be re-run with a narrower prefix. +/// This shell is a deliberate RustFS extension: madmin streams bare +/// `DiffInfo` documents with no envelope (see [`render_replication_diff`]), +/// so the scan-coverage metadata (`is_truncated`, `scanned_versions`) is only +/// representable in this opt-in aggregate shape. `entries` lists object +/// versions with a `PENDING` or `FAILED` replication status. `is_truncated` +/// indicates the on-demand scan hit [`REPLICATION_DIFF_MAX_SCAN`] before +/// reaching the end of the bucket, so the diff is partial and should be +/// re-run with a narrower prefix. #[derive(Debug, Serialize)] struct ReplicationDiffResponse { #[serde(rename = "Entries")] @@ -920,6 +927,39 @@ struct ReplicationDiffResponse { scanned_versions: usize, } +/// Render the diff scan result as a response body. +/// +/// Default (madmin-compatible) mode emits one `DiffInfo` JSON document per +/// line with no envelope — madmin's `BucketReplicationDiff` reads the body +/// with a `json.Decoder` loop, so any envelope object would decode as a +/// single entry with an empty `object` (a phantom row in `mc replicate +/// diff`). Scan-truncation info is not representable in that stream and is +/// reported via the tracing event in the handler instead. +/// +/// `aggregate=true` (RustFS extension) keeps the enveloped shape including +/// `IsTruncated`/`ScannedVersions`. +fn render_replication_diff( + entries: Vec, + is_truncated: bool, + scanned_versions: usize, + aggregate: bool, +) -> Result, serde_json::Error> { + if aggregate { + return serde_json::to_vec(&ReplicationDiffResponse { + entries, + is_truncated, + scanned_versions, + }); + } + + let mut data = Vec::new(); + for entry in &entries { + serde_json::to_writer(&mut data, entry)?; + data.push(b'\n'); + } + Ok(data) +} + #[derive(Debug, Default, Deserialize)] #[serde(deny_unknown_fields)] struct ReplicationDiffRequest { @@ -1036,16 +1076,31 @@ impl Operation for ReplicationDiffHandler { truncated = is_truncated, "computed replication diff" ); + let aggregate = queries.get("aggregate").map(String::as_str) == Some("true"); + if is_truncated && !aggregate { + // The madmin stream has no envelope to carry truncation info, so + // surface the partial-scan condition here instead. + tracing::warn!( + bucket = %bucket, + prefix = %prefix, + scanned = scanned_versions, + max_scan = REPLICATION_DIFF_MAX_SCAN, + "replication diff scan truncated; stream response is partial — re-run with a narrower prefix or use aggregate=true" + ); + } - let response = ReplicationDiffResponse { - entries, - is_truncated, - scanned_versions, - }; - let data = serde_json::to_vec(&response) + let data = render_replication_diff(entries, is_truncated, scanned_versions, aggregate) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?; let mut headers = HeaderMap::new(); headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + // The madmin stream has no envelope to carry truncation info; a + // truncated scan would otherwise be indistinguishable from a complete + // one (an empty diff on a >MAX_SCAN bucket reads as "healthy"). Signal + // it out-of-band so RustFS-aware clients can detect the partial scan; + // madmin/mc ignore unknown headers. + if is_truncated { + headers.insert("x-rustfs-replication-diff-truncated", HeaderValue::from_static("true")); + } Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), headers)) } } @@ -1247,8 +1302,8 @@ impl Operation for ReplicationMrfHandler { mod tests { use super::{ REMOTE_TARGET_UNSUPPORTED_FIELDS, REMOTE_TARGET_WRITABLE_FIELDS, RemoteTargetCredentialsRequest, RemoteTargetRequest, - SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params, parse_remote_target_update_ops, - unique_replication_peers, validate_remote_target_tls_settings, + ReplicationDiffEntry, SUPPORTED_REMOTE_TARGET_API, TargetUpdateOp, build_mrf_response, extract_query_params, + parse_remote_target_update_ops, render_replication_diff, unique_replication_peers, validate_remote_target_tls_settings, }; use crate::admin::storage_api::bucket::target::{BucketTarget, LatencyStat}; use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry}; @@ -1950,4 +2005,71 @@ mod tests { assert!(target.secure); assert_eq!(target.credentials.expect("credentials should be present").access_key, "access"); } + + fn sample_diff_entries() -> Vec { + vec![ + ReplicationDiffEntry { + object: "a.txt".to_string(), + version_id: Some("v1".to_string()), + size: 42, + is_delete_marker: false, + replication_status: "PENDING".to_string(), + last_modified: Some("2026-01-01T00:00:00Z".to_string()), + }, + ReplicationDiffEntry { + object: "b.txt".to_string(), + version_id: None, + size: 0, + is_delete_marker: true, + replication_status: "FAILED".to_string(), + last_modified: None, + }, + ] + } + + #[test] + fn replication_diff_stream_emits_bare_madmin_diff_info_documents() { + let data = render_replication_diff(sample_diff_entries(), true, 2, false).expect("stream must render"); + let text = std::str::from_utf8(&data).expect("stream must be utf-8"); + + // madmin decodes the body with a json.Decoder loop over DiffInfo — one + // bare document per entry, no envelope keys anywhere in the stream. + let lines: Vec = text + .lines() + .map(|line| serde_json::from_str(line).expect("each line must be a JSON document")) + .collect(); + assert_eq!(lines.len(), 2); + assert_eq!(lines[0]["object"], "a.txt"); + assert_eq!(lines[0]["versionId"], "v1"); + assert_eq!(lines[0]["rStatus"], "PENDING"); + assert_eq!(lines[0]["deletemarker"], false); + assert_eq!(lines[0]["lastModified"], "2026-01-01T00:00:00Z"); + assert_eq!(lines[1]["object"], "b.txt"); + assert_eq!(lines[1]["deletemarker"], true); + assert_eq!(lines[1]["rStatus"], "FAILED"); + for line in &lines { + assert!(line.get("Entries").is_none()); + assert!(line.get("IsTruncated").is_none()); + assert!(line.get("ScannedVersions").is_none()); + } + } + + #[test] + fn replication_diff_stream_renders_empty_body_for_no_entries() { + let data = render_replication_diff(Vec::new(), false, 0, false).expect("stream must render"); + // An empty body makes madmin's json.Decoder loop end immediately with + // io.EOF — zero rows, not one phantom empty row. + assert!(data.is_empty()); + } + + #[test] + fn replication_diff_aggregate_keeps_enveloped_extension_shape() { + let data = render_replication_diff(sample_diff_entries(), true, 7, true).expect("aggregate must render"); + let payload: serde_json::Value = serde_json::from_slice(&data).expect("aggregate must be one JSON object"); + + assert_eq!(payload["Entries"].as_array().map(Vec::len), Some(2)); + assert_eq!(payload["Entries"][0]["object"], "a.txt"); + assert_eq!(payload["IsTruncated"], true); + assert_eq!(payload["ScannedVersions"], 7); + } } diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 5c81e0e5b..92b6aee47 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -123,17 +123,20 @@ enum MiscExtRoute { ListenNotification { bucket: Option }, } +// Wire shape mirrors madmin-go `ResyncTargetsInfo`/`ResyncTarget` json tags so +// `mc replicate resync` can decode the response (Go json decoding is +// case-insensitive per field, but the `target` shell key must match exactly). #[derive(Debug, Clone, serde::Serialize, Default)] struct ReplicationResetResponse { - #[serde(rename = "Targets")] + #[serde(rename = "target")] targets: Vec, } #[derive(Debug, Clone, serde::Serialize, Default)] struct ReplicationResetTarget { - #[serde(rename = "Arn")] + #[serde(rename = "arn")] arn: String, - #[serde(rename = "ResetID")] + #[serde(rename = "resetid")] reset_id: String, } @@ -149,17 +152,21 @@ struct ReplicationResetStatusRequest { arn: Option, } +// Wire shape mirrors madmin-go `ResyncTargetsInfo`/`ResyncTarget` json tags +// (see `ReplicationResetResponse`). `ResetBeforeDate` and `Error` are RustFS +// extension keys with no madmin counterpart; Go decoders ignore unknown keys, +// so they coexist with madmin/mc clients at zero cost. #[derive(Debug, Clone, serde::Serialize, Default)] struct ReplicationResetStatusResponse { - #[serde(rename = "Targets")] + #[serde(rename = "target")] targets: Vec, } #[derive(Debug, Clone, serde::Serialize, Default)] struct ReplicationResetStatusTarget { - #[serde(rename = "Arn")] + #[serde(rename = "arn")] arn: String, - #[serde(rename = "ResetID")] + #[serde(rename = "resetid")] reset_id: String, #[serde( rename = "ResetBeforeDate", @@ -168,30 +175,30 @@ struct ReplicationResetStatusTarget { )] reset_before_date: Option, #[serde( - rename = "StartTime", + rename = "startTime", with = "time::serde::rfc3339::option", skip_serializing_if = "Option::is_none" )] start_time: Option, #[serde( - rename = "EndTime", + rename = "endTime", with = "time::serde::rfc3339::option", skip_serializing_if = "Option::is_none" )] end_time: Option, - #[serde(rename = "Status")] + #[serde(rename = "resyncStatus")] status: String, - #[serde(rename = "ReplicatedCount")] + #[serde(rename = "replicationCount")] replicated_count: i64, - #[serde(rename = "ReplicatedSize")] + #[serde(rename = "completedReplicationSize")] replicated_size: i64, - #[serde(rename = "FailedCount")] + #[serde(rename = "failedReplicationCount")] failed_count: i64, - #[serde(rename = "FailedSize")] + #[serde(rename = "failedReplicationSize")] failed_size: i64, - #[serde(rename = "Bucket", skip_serializing_if = "String::is_empty")] + #[serde(rename = "bucket", skip_serializing_if = "String::is_empty")] bucket: String, - #[serde(rename = "Object", skip_serializing_if = "String::is_empty")] + #[serde(rename = "object", skip_serializing_if = "String::is_empty")] object: String, #[serde(rename = "Error", skip_serializing_if = "Option::is_none")] error: Option, @@ -3205,6 +3212,22 @@ mod tests { assert!(error.message().unwrap_or_default().contains("run-active")); } + #[test] + fn replication_reset_response_matches_madmin_resync_targets_info_shape() { + let payload = serde_json::to_value(ReplicationResetResponse { + targets: vec![ReplicationResetTarget { + arn: "arn:minio:replication::depl:bucket".to_string(), + reset_id: "rid-1".to_string(), + }], + }) + .expect("reset response must serialize"); + + // madmin-go `ResyncTargetsInfo` json tags: shell `target`, fields `arn`/`resetid`. + assert_eq!(payload["target"][0]["arn"], "arn:minio:replication::depl:bucket"); + assert_eq!(payload["target"][0]["resetid"], "rid-1"); + assert!(payload.get("Targets").is_none()); + } + #[test] fn build_replication_reset_status_response_serializes_sorted_targets() { let mut status = BucketReplicationResyncStatus::new(); @@ -3225,6 +3248,7 @@ mod tests { "arn:a".to_string(), crate::admin::storage_api::bucket::replication::TargetReplicationResyncStatus { resync_id: "rid-a".to_string(), + start_time: Some(datetime!(2025-01-01 00:00 UTC)), last_update: Some(datetime!(2025-01-02 00:00 UTC)), resync_status: crate::admin::storage_api::bucket::replication::ResyncStatusType::ResyncCompleted, replicated_count: 3, @@ -3240,15 +3264,21 @@ mod tests { .to_bytes(); let payload: serde_json::Value = serde_json::from_slice(&bytes).expect("response must be json"); - assert_eq!(payload["Targets"][0]["Arn"], "arn:a"); - assert_eq!(payload["Targets"][0]["Bucket"], "bucket-a"); - assert_eq!(payload["Targets"][0]["Status"], "Completed"); - assert_eq!(payload["Targets"][0]["EndTime"], "2025-01-02T00:00:00Z"); - assert_eq!(payload["Targets"][1]["Arn"], "arn:z"); - assert_eq!(payload["Targets"][1]["Bucket"], "bucket-z"); - assert_eq!(payload["Targets"][1]["Status"], "Failed"); - assert_eq!(payload["Targets"][1]["EndTime"], "2025-01-03T00:00:00Z"); - assert_eq!(payload["Targets"][1]["Error"], "boom"); + assert_eq!(payload["target"][0]["arn"], "arn:a"); + assert_eq!(payload["target"][0]["resetid"], "rid-a"); + assert_eq!(payload["target"][0]["bucket"], "bucket-a"); + assert_eq!(payload["target"][0]["resyncStatus"], "Completed"); + assert_eq!(payload["target"][0]["startTime"], "2025-01-01T00:00:00Z"); + assert_eq!(payload["target"][0]["endTime"], "2025-01-02T00:00:00Z"); + assert_eq!(payload["target"][0]["replicationCount"], 3); + assert_eq!(payload["target"][0]["completedReplicationSize"], 9); + assert_eq!(payload["target"][1]["arn"], "arn:z"); + assert_eq!(payload["target"][1]["bucket"], "bucket-z"); + assert_eq!(payload["target"][1]["resyncStatus"], "Failed"); + assert_eq!(payload["target"][1]["endTime"], "2025-01-03T00:00:00Z"); + assert_eq!(payload["target"][1]["failedReplicationCount"], 2); + assert_eq!(payload["target"][1]["failedReplicationSize"], 4); + assert_eq!(payload["target"][1]["Error"], "boom"); } #[test] @@ -3286,12 +3316,12 @@ mod tests { .to_bytes(); let payload: serde_json::Value = serde_json::from_slice(&bytes).expect("response must be json"); - assert_eq!(payload["Targets"].as_array().map(Vec::len), Some(1)); - assert_eq!(payload["Targets"][0]["Arn"], "arn:z"); - assert_eq!(payload["Targets"][0]["Bucket"], "bucket-z"); - assert_eq!(payload["Targets"][0]["Status"], "Failed"); - assert_eq!(payload["Targets"][0]["EndTime"], "2025-02-03T00:00:00Z"); - assert_eq!(payload["Targets"][0]["Error"], "boom"); + assert_eq!(payload["target"].as_array().map(Vec::len), Some(1)); + assert_eq!(payload["target"][0]["arn"], "arn:z"); + assert_eq!(payload["target"][0]["bucket"], "bucket-z"); + assert_eq!(payload["target"][0]["resyncStatus"], "Failed"); + assert_eq!(payload["target"][0]["endTime"], "2025-02-03T00:00:00Z"); + assert_eq!(payload["target"][0]["Error"], "boom"); } fn replication_check_target(arn: &str, status: &str, error: Option<&str>) -> ReplicationCheckTargetStatus { diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index ee160a7f4..8f045dbfe 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -39,7 +39,8 @@ use super::storage_api::bucket_usecase::bucket::{ policy_sys::PolicySys, replication::{ ReplicationTargetValidationError, invalid_replication_config_status_field, replication_target_arns, - should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, + should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_structure, + validate_replication_config_target_arns, }, target::{BucketTargetType, BucketTargets}, utils::serialize, @@ -584,6 +585,9 @@ fn validate_replication_config_targets(targets: &BucketTargets, config: &Replica } fn validate_replication_config_capabilities(config: &ReplicationConfiguration) -> S3Result<()> { + if let Err(err) = validate_replication_config_structure(config) { + return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, err.message())); + } if let Some(field) = invalid_replication_config_status_field(config) { return Err(S3Error::with_message( S3ErrorCode::InvalidRequest, @@ -3175,6 +3179,24 @@ mod tests { assert!(!err.to_string().contains(destination_key_id)); } + #[test] + fn validate_replication_config_capabilities_rejects_structural_defects_before_write() { + let mut first = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket"); + first.priority = Some(1); + let mut second = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket"); + second.priority = Some(1); + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![first, second], + }; + + let err = validate_replication_config_capabilities(&config) + .expect_err("duplicate rule priorities must be rejected before persistence"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert!(err.to_string().contains("Priority must be unique")); + } + #[test] fn validate_replication_config_capabilities_rejects_invalid_status_before_write() { let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket"); @@ -4448,11 +4470,13 @@ mod tests { #[tokio::test] async fn execute_put_bucket_replication_returns_internal_error_when_store_uninitialized() { + // The config must clear the structural/capability validators so the + // request actually reaches the store lookup this test pins. let input = PutBucketReplicationInput::builder() .bucket("test-bucket".to_string()) .replication_configuration(ReplicationConfiguration { role: "arn:aws:iam::123456789012:role/test".to_string(), - rules: vec![], + rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket")], }) .build() .unwrap(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 3536aaac6..38d96fbb1 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -896,6 +896,14 @@ pub(crate) mod bucket { ) -> Option<&'static str> { replication_contracts::invalid_replication_config_status_field(config) } + + pub(crate) type ReplicationConfigStructureError = replication_contracts::ReplicationConfigStructureError; + + pub(crate) fn validate_replication_config_structure( + config: &s3s::dto::ReplicationConfiguration, + ) -> Result<(), ReplicationConfigStructureError> { + replication_contracts::validate_replication_config_structure(config) + } } pub(crate) mod tagging {