fix(replication): madmin reset/diff wire compat and config validation (#5799)

* fix(admin): align replication-reset responses with madmin ResyncTargetsInfo shape The replication-reset and replication-reset-status responses serialized their shell as "Targets" and per-target fields in PascalCase, while madmin-go ResyncTargetsInfo/ResyncTarget expect the "target" shell key and lowercase field tags (arn/resetid/resyncStatus/replicationCount/ completedReplicationSize/failedReplicationCount/failedReplicationSize). Go json decoding is case-insensitive per field, but Targets vs target, Status vs resyncStatus and the size/count key names cannot match, so mc replicate resync decoded empty results. Rename the serde tags to the exact madmin wire shape, keep the ResetBeforeDate/Error RustFS extension keys (unknown keys are ignored by Go decoders), pin the shape with a snapshot unit test, and update the e2e client DTO to decode the madmin shape. * fix(admin): stream bare madmin DiffInfo documents from replication diff POST /v3/replication/diff returned a single enveloped object ({Entries, IsTruncated, ScannedVersions}) while madmin-go BucketReplicationDiff decodes the body with a json.Decoder loop over bare DiffInfo documents. The envelope decoded as exactly one DiffInfo with an empty object, so mc replicate diff printed a phantom empty row instead of the real backlog. Emit one DiffInfo JSON document per line by default, using the exact madmin json tags (object/versionId/rStatus/deletemarker/lastModified; Size stays as a RustFS extension key that Go decoders ignore). The enveloped shape moves to the opt-in ?aggregate=true RustFS extension, which remains the only carrier of scan-coverage metadata; a truncated default-mode scan is surfaced via a warn tracing event instead of in-stream. Pin both shapes with unit tests and tighten the e2e helper to reject any envelope in the stream. * feat(replication): validate replication config structure before persisting PutBucketReplication accepted structurally invalid configurations that MinIO's replication.Config.Validate rejects: empty or oversized rule lists, duplicate or negative rule priorities, over-long rule IDs, filters carrying more than one of Prefix/Tag/And, and delete marker replication enabled on tag-filtered rules. Such configs persisted silently and later produced undefined routing (e.g. ambiguous priority ties) instead of failing the PUT. Add validate_replication_config_structure as a pure function in rustfs-replication (limits documented as constants), surface it through the ecstore api facade, and run it first in the PUT capability gate so defects are named before any metadata write. Missing Priority counts as zero for the uniqueness check, matching Go's zero-value semantics. The self-target rejection deliberately stays at set-remote-target, where the endpoint is known; a config can never reference a self-pointing ARN. Document the rule-level Destination.StorageClass contract (use the remote target's storage_class instead) and renumber the acceptance matrix e2e to unique priorities, which MinIO would also require. * test(replication): pin duplicated wire types with boundary reconciliation tests rustfs-filemeta (xl.meta disk format) and rustfs-replication (MRF/resync persistence format) deliberately each own ReplicationStatusType, VersionPurgeStatusType and ReplicationState; the boundary converts between them via as_str(), whose From<&str> impls fall back to Empty on unknown tokens — a variant added on one side silently degrades to Empty on the other. Add reconciliation tests in replication_filemeta_boundary: exhaustive matches with no wildcard arm on both sides of both enums (a new variant fails compilation until the mapping is reconsidered), string-token round-trip asserts (a token the other side does not recognize fails instead of quietly becoming Empty), and a full-field ReplicationState round-trip. Cross-reference the tests from both type definitions. Struct drift was already compile-guarded by the exhaustive struct literals in the conversion functions. * docs(replication): define split completion criteria and milestone sequence The ecstore replication split plan had no completion measure — the boundary scaffolding risked ossifying because nothing said when the migration counts as done. Record the criteria in the module inventory: done means the Required Contracts table's 'Current dependency to remove' column is empty; the end state moves pool/resyncer/state into crates/replication, with the boundary micro-files dissolving as code crosses the crate line (batch-merging them beforehand is explicitly rejected — the guard scripts anchor on their file names, so merging is churn with zero functional gain; only datatypes.rs can retire early). Sequence the remaining work as M2 (resyncer pure decision logic, after the oversized function splits) → M3 (worker runtime, highest risk, last) → M4 (retire boundaries and guard entries). Refresh the stale first-step text — the event sink / runtime contracts already landed — and update the split-plan status table accordingly. * fix(replication): align structural validator with MinIO semantics after adversarial review Three interop corrections found by adversarial review of the new structural validator, plus review fallout fixes: - Delete-marker replication is now rejected only for a direct Filter.Tag, not for tags inside Filter.And — MinIO's validator 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 the stricter check rejected mc-generated configs MinIO accepts. - Rule ID length is measured in bytes (Go len semantics), not chars — a 255-char multibyte ID must not round-trip into a config MinIO rejects. - An empty <Tag/> element (no key) counts as absent, matching MinIO's Tag.IsEmpty(); console form serializers emit empty tags, which would otherwise trip the exactly-one-of and delete-marker checks. Also: repair the store-uninitialized PUT test whose empty-rules fixture now (correctly) fails structural validation before reaching the store lookup; pin the previously untested startTime madmin key in the reset-status shape test; and signal a truncated default-mode diff scan via the x-rustfs-replication-diff-truncated response header — the bare madmin stream has no envelope, so a truncated scan was otherwise indistinguishable from a complete healthy one (madmin/mc ignore unknown headers). * test(e2e): activate SSE-S3 replication contract and pin resync fail-closed path The SSE-S3 replication contract e2e was ignored under backlog#1291 (silent plaintext replication); the fail-closed gate in replication_target_boundary.rs closed that hole, so the ignore reason expired. Un-ignore the test — it now pins the current fail-closed contract (FAILED status, failure event, readable encrypted source, stable absence of all target versions), verified green. Add test_bucket_replication_sse_s3_resync_stays_fail_closed: drives the existing-object resync path (PUT ?replication-reset) over a FAILED SSE-S3 object and asserts the resync generation reaches a terminal state without ever materializing a target version, with the stays-absent window also spanning fast-scanner heal cycles. The new start_bucket_replication_reset helper doubles as the madmin ResyncTargetsInfo shape assertion (target[0].arn/resetid) for the reset-start response. Refresh the stale nextest count commentary (the module is at 20 fast + 36 nightly = 56 tests by cargo nextest list; the SSE-S3-ignored note no longer holds).
2026-08-09 06:39:25 +00:00 · 2026-08-07 22:30:12 +08:00
parent 7553715f62
commit 3792fed827
16 changed files with 861 additions and 97 deletions
@@ -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.
@@ -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<ReplicationResetStatusTarget>,
 }

 #[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::<serde_json::Value>)
+            .collect::<Result<Vec<_>, _>>()?;
+        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<dyn Error + Send + Sync>> {
+    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
  <Role></Role>
  <Rule>
    <ID>matrix-prefix</ID>
-    <Priority>100</Priority>
+    <Priority>110</Priority>
    <Status>Enabled</Status>
    <Filter><Prefix>prefix/</Prefix></Filter>
    <DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication>
@@ -3949,7 +3986,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
  </Rule>
  <Rule>
    <ID>matrix-both-prefix</ID>
-    <Priority>100</Priority>
+    <Priority>120</Priority>
    <Status>Enabled</Status>
    <Filter><Prefix>both/</Prefix></Filter>
    <DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication>
@@ -3959,7 +3996,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
  </Rule>
  <Rule>
    <ID>matrix-tag</ID>
-    <Priority>100</Priority>
+    <Priority>130</Priority>
    <Status>Enabled</Status>
    <Filter><Tag><Key>route</Key><Value>tagged</Value></Tag></Filter>
    <DeleteMarkerReplication><Status>Disabled</Status></DeleteMarkerReplication>
@@ -3969,7 +4006,7 @@ async fn test_bucket_replication_acceptance_matrix_local_dual_targets() -> TestR
  </Rule>
  <Rule>
    <ID>matrix-disabled</ID>
-    <Priority>100</Priority>
+    <Priority>140</Priority>
    <Status>Disabled</Status>
    <Filter><Prefix>disabled/</Prefix></Filter>
    <DeleteMarkerReplication><Status>Enabled</Status></DeleteMarkerReplication>
@@ -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.
@@ -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,
        };
    }

@@ -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
@@ -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::{
@@ -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,
 };
@@ -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);
+    }
+}
@@ -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.
@@ -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 <Tag/> 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<ReplicationRule>) -> 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 <Tag/> 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 {
@@ -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.
@@ -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,
@@ -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
@@ -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<String>,
    #[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<String>,
 }

-/// 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<ReplicationDiffEntry>,
+    is_truncated: bool,
+    scanned_versions: usize,
+    aggregate: bool,
+) -> Result<Vec<u8>, 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<ReplicationDiffEntry> {
+        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<serde_json::Value> = 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);
+    }
 }
@@ -123,17 +123,20 @@ enum MiscExtRoute {
    ListenNotification { bucket: Option<String> },
 }

+// 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<ReplicationResetTarget>,
 }

 #[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<String>,
 }

+// 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<ReplicationResetStatusTarget>,
 }

 #[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<OffsetDateTime>,
    #[serde(
-        rename = "StartTime",
+        rename = "startTime",
        with = "time::serde::rfc3339::option",
        skip_serializing_if = "Option::is_none"
    )]
    start_time: Option<OffsetDateTime>,
    #[serde(
-        rename = "EndTime",
+        rename = "endTime",
        with = "time::serde::rfc3339::option",
        skip_serializing_if = "Option::is_none"
    )]
    end_time: Option<OffsetDateTime>,
-    #[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<String>,
@@ -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 {
@@ -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();
@@ -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 {