fix(site-replication): close ILM merge review findings

Adversarial review of the lc-config merge surfaced four real defects,
all fixed here:

- Deletion tombstone regression: with the staleness axis moved to the
  in-config expiry_updated_at, a deleted lifecycle config fell back to
  UNIX_EPOCH and any delayed stale broadcast could resurrect deleted
  expiry rules. The axis now falls back to the whole-config write time
  (which survives deletion in bucket metadata as the deletion's lower
  bound), also covering legacy configs that predate the axis field.
- MinIO zero-rule documents: MinIO's delete tombstone / transition-only
  state marshals a lifecycle document with no <Rule>, which the strict
  s3s deserializer rejects — the receiver now recognizes it as the 'no
  expiry rules here' statement (delete semantics) instead of erroring
  on every MinIO heal pass.
- Inflated expiry axis at the sender: PutBucketLifecycle stamped
  expiry_updated_at unconditionally, so a transition-only edit advanced
  the axis and let this site's stale expiry subset shadow and roll back
  newer peer expiry edits fleet-wide. The stamp is now conditional
  (expiry subset present before or after the edit, MinIO parity), the
  hook item travels with the config's expiry axis (UNIX_EPOCH when the
  site has none), and the SRInfo bucket entry feeds bootstrap/repair
  the same axis instead of the whole-config write time.
- Del-marker parity: MinIO's CloneNonTransition never emits del-marker
  or abort-mpu fields, so treating del_marker_expiration as traveling
  expiry let a MinIO broadcast delete this site's del-marker-only
  rules. Both fields are now site-local on every edge: stripped from
  outbound subsets and inbound rules, restored from the local side on
  same-id merges, and never a deletion criterion.

Receiver-side validation of merged configs (object-lock / tier
constraints, MinIO runs finalLcCfg.Validate) remains a follow-up.

rustfs/src/admin/handlers/site_replication.rs

@@ -2331,6 +2331,8 @@ fn append_bootstrap_bucket_items(
             bucket.expiry_lc_config_updated_at,
             |item, value| {
                 item.expiry_lc_config = Some(value);
+                // `updated_at` here is the entry's expiry axis (see the
+                // SRBucketInfo construction), not the wall clock.
                 item.expiry_updated_at = item.updated_at;
                 Ok(())
             },
@@ -4220,13 +4222,23 @@ pub async fn site_replication_delete_bucket_hook(bucket: &str, force_delete: boo
     broadcast_site_replication_json(&path, &serde_json::json!({})).await
 }
 
-pub async fn site_replication_bucket_meta_hook(item: SRBucketMeta) -> S3Result<()> {
+pub async fn site_replication_bucket_meta_hook(mut item: SRBucketMeta) -> S3Result<()> {
     let Some(runtime) = runtime_site_replication_targets().await? else {
         return Ok(());
     };
     if item.r#type == "lc-config" && !site_replication_state_replicates_ilm_expiry(&runtime.state) {
         return Ok(());
     }
+    if item.r#type == "lc-config" {
+        // Only the expiry subset travels (MinIO peers install incoming rules
+        // verbatim, so transition rules must never leave this site). An empty
+        // subset becomes a delete, which the receiver merges with the empty
+        // set — local transition rules there survive.
+        item.expiry_lc_config = item
+            .expiry_lc_config
+            .and_then(|raw| lifecycle_expiry_subset_xml(raw.as_bytes()))
+            .map(|data| String::from_utf8_lossy(&data).into_owned());
+    }
     broadcast_site_replication_json_with_runtime(
         &runtime,
         "/rustfs/admin/v3/site-replication/peer/bucket-meta",
@@ -4319,7 +4331,12 @@ async fn build_sr_info(state: &SiteReplicationState, local_peer: &PeerInfo) -> S
             entry.sse_config = raw_config_to_base64(&metadata.encryption_config_xml);
             entry.replication_config = raw_config_to_base64(&metadata.replication_config_xml);
             entry.quota_config = raw_config_to_base64(&metadata.quota_config_json);
-            entry.expiry_lc_config = raw_config_to_base64(&metadata.lifecycle_config_xml);
+            // Expiry subset only: this entry feeds both the bootstrap/repair
+            // plan (peers must not receive transition rules) and cross-site
+            // consistency views (transition rules are site-local and would
+            // read as false mismatches).
+            entry.expiry_lc_config =
+                lifecycle_expiry_subset_xml(&metadata.lifecycle_config_xml).and_then(|data| raw_config_to_base64(&data));
             entry.cors_config = raw_config_to_base64(&metadata.cors_config_xml);
             entry.policy_updated_at = maybe_time(metadata.policy_config_updated_at);
             entry.tag_config_updated_at = maybe_time(metadata.tagging_config_updated_at);
@@ -4328,7 +4345,14 @@ async fn build_sr_info(state: &SiteReplicationState, local_peer: &PeerInfo) -> S
             entry.versioning_config_updated_at = maybe_time(metadata.versioning_config_updated_at);
             entry.replication_config_updated_at = maybe_time(metadata.replication_config_updated_at);
             entry.quota_config_updated_at = maybe_time(metadata.quota_config_updated_at);
-            entry.expiry_lc_config_updated_at = maybe_time(metadata.lifecycle_config_updated_at);
+            // The expiry axis, not the whole-config write time: local
+            // transition-only edits inflate the latter, and a repair item
+            // stamped with it could out-rank a newer real expiry edit on a
+            // third site. Legacy configs without the axis fall back to the
+            // whole-config time (bounded staleness, same as the receiver).
+            entry.expiry_lc_config_updated_at = lifecycle_expiry_updated_at(&metadata.lifecycle_config_xml)
+                .map(Some)
+                .unwrap_or_else(|| maybe_time(metadata.lifecycle_config_updated_at));
             entry.cors_config_updated_at = maybe_time(metadata.cors_config_updated_at);
             entry.replication_targets_online =
                 Some(site_replication_targets_online(&bucket.name, &metadata.replication_config_xml).await);
@@ -6847,6 +6871,180 @@ fn merge_incoming_replication_config(
     Some(ReplicationConfiguration { role, rules })
 }
 
+/// Merge a peer's ILM expiry document into the local lifecycle config.
+///
+/// Mirrors MinIO's `mergeWithCurrentLCConfig` with one hardening: incoming
+/// site-local fields (transitions, abort-multipart, del-marker expiration —
+/// exactly what MinIO's `CloneNonTransition` sender never emits) are
+/// discarded outright at the trust boundary, whatever the peer sends. Local
+/// site-local fields always survive; a delete (`incoming == None`) therefore
+/// merges with the empty set instead of dropping the whole config.
+fn merge_incoming_lifecycle_config(
+    incoming: Option<s3s::dto::BucketLifecycleConfiguration>,
+    local: Option<s3s::dto::BucketLifecycleConfiguration>,
+    updated_at: Option<OffsetDateTime>,
+) -> Option<s3s::dto::BucketLifecycleConfiguration> {
+    // Incoming rules reduced to their traveling expiry side. Rules with no
+    // expiry semantics after the strip are not installed.
+    let mut incoming_by_id: HashMap<String, s3s::dto::LifecycleRule> = HashMap::new();
+    let mut incoming_order: Vec<String> = Vec::new();
+    for mut rule in incoming.into_iter().flat_map(|config| config.rules) {
+        strip_site_local_lifecycle_fields(&mut rule);
+        if !lifecycle_rule_has_expiry(&rule) {
+            continue;
+        }
+        let id = rule.id.clone().unwrap_or_default();
+        if incoming_by_id.insert(id.clone(), rule).is_none() {
+            incoming_order.push(id);
+        }
+    }
+
+    // Local order first, incoming-new appended: repeated delivery of the same
+    // document is byte-stable, so bucket metadata is written once, not on
+    // every broadcast.
+    let local_expiry_updated_at = local.as_ref().and_then(|config| config.expiry_updated_at.clone());
+    let mut rules: Vec<s3s::dto::LifecycleRule> = Vec::new();
+    for mut rule in local.into_iter().flat_map(|config| config.rules) {
+        let id = rule.id.clone().unwrap_or_default();
+        if let Some(mut incoming_rule) = incoming_by_id.remove(&id) {
+            incoming_order.retain(|pending| pending != &id);
+            // The incoming expiry side wins; the local site-local side is
+            // authoritative (MinIO CloneNonTransition + restore).
+            incoming_rule.transitions = rule.transitions.take();
+            incoming_rule.noncurrent_version_transitions = rule.noncurrent_version_transitions.take();
+            incoming_rule.abort_incomplete_multipart_upload = rule.abort_incomplete_multipart_upload.take();
+            incoming_rule.del_marker_expiration = rule.del_marker_expiration.take();
+            rules.push(incoming_rule);
+        } else if lifecycle_rule_has_expiry(&rule) {
+            // Expiry rule dropped upstream: strip only the traveling expiry
+            // side; the rule survives while any site-local action remains.
+            rule.expiration = None;
+            rule.noncurrent_version_expiration = None;
+            if lifecycle_rule_has_transition(&rule)
+                || rule.abort_incomplete_multipart_upload.is_some()
+                || rule.del_marker_expiration.is_some()
+            {
+                rules.push(rule);
+            }
+        } else {
+            // No traveling expiry semantics (transition-only / abort-mpu-only
+            // / del-marker-only): not managed by expiry replication, keep
+            // untouched.
+            rules.push(rule);
+        }
+    }
+    for id in incoming_order {
+        if let Some(rule) = incoming_by_id.remove(&id) {
+            rules.push(rule);
+        }
+    }
+
+    if rules.is_empty() {
+        return None;
+    }
+
+    Some(s3s::dto::BucketLifecycleConfiguration {
+        rules,
+        // Record the expiry axis the staleness guard compares on. (The PUT
+        // path stamps `expiry_updated_at` only when the expiry subset
+        // changes, so this axis is not inflated by transition-only edits.)
+        expiry_updated_at: updated_at.map(s3s::dto::Timestamp::from).or(local_expiry_updated_at),
+    })
+}
+
+/// True when the rule carries the expiry semantics that `replicateILMExpiry`
+/// propagates. Del-marker expiration and abort-multipart are deliberately
+/// excluded: MinIO's sender never emits them (`CloneNonTransition` drops
+/// both), so treating them as traveling state would let a MinIO peer's
+/// broadcast delete this site's del-marker-only rules.
+fn lifecycle_rule_has_expiry(rule: &s3s::dto::LifecycleRule) -> bool {
+    rule.expiration.is_some() || rule.noncurrent_version_expiration.is_some()
+}
+
+fn lifecycle_rule_has_transition(rule: &s3s::dto::LifecycleRule) -> bool {
+    rule.transitions.as_ref().is_some_and(|transitions| !transitions.is_empty())
+        || rule
+            .noncurrent_version_transitions
+            .as_ref()
+            .is_some_and(|transitions| !transitions.is_empty())
+}
+
+/// Remove the fields that never travel between sites (MinIO
+/// `CloneNonTransition` parity).
+fn strip_site_local_lifecycle_fields(rule: &mut s3s::dto::LifecycleRule) {
+    rule.transitions = None;
+    rule.noncurrent_version_transitions = None;
+    rule.abort_incomplete_multipart_upload = None;
+    rule.del_marker_expiration = None;
+}
+
+/// Reduce a lifecycle XML document to the expiry subset that is allowed to
+/// travel between sites (what MinIO's sender emits): transition fields are
+/// stripped and rules left with no expiry semantics are dropped. Returns
+/// `None` when nothing remains — the receiver then merges with the empty set,
+/// which is exactly the "no expiry rules here" statement. A document that
+/// fails to parse is forwarded unfiltered (`Some(original)`): the receiver
+/// merge strips it anyway, and turning a local parse error into a `None`
+/// would delete the peers' replicated expiry rules.
+fn lifecycle_expiry_subset_xml(raw: &[u8]) -> Option<Vec<u8>> {
+    if raw.is_empty() {
+        return None;
+    }
+    let config: s3s::dto::BucketLifecycleConfiguration = match deserialize(raw) {
+        Ok(config) => config,
+        Err(err) => {
+            warn!("failed to parse local lifecycle config for expiry replication; forwarding unfiltered: {err}");
+            return Some(raw.to_vec());
+        }
+    };
+    let expiry_updated_at = config.expiry_updated_at.clone();
+    let rules: Vec<s3s::dto::LifecycleRule> = config
+        .rules
+        .into_iter()
+        .filter_map(|mut rule| {
+            strip_site_local_lifecycle_fields(&mut rule);
+            lifecycle_rule_has_expiry(&rule).then_some(rule)
+        })
+        .collect();
+    if rules.is_empty() {
+        return None;
+    }
+    let subset = s3s::dto::BucketLifecycleConfiguration {
+        rules,
+        expiry_updated_at,
+    };
+    match serialize(&subset) {
+        Ok(data) => Some(data),
+        Err(err) => {
+            warn!("failed to serialize lifecycle expiry subset; forwarding unfiltered: {err}");
+            Some(raw.to_vec())
+        }
+    }
+}
+
+/// The expiry replication axis persisted in a lifecycle XML document, if any.
+/// Used for the SRInfo bucket entry so bootstrap/repair items carry the
+/// expiry axis instead of the whole-config write time (which local
+/// transition-only edits inflate).
+fn lifecycle_expiry_updated_at(raw: &[u8]) -> Option<OffsetDateTime> {
+    if raw.is_empty() {
+        return None;
+    }
+    deserialize::<s3s::dto::BucketLifecycleConfiguration>(raw)
+        .ok()
+        .and_then(|config| config.expiry_updated_at)
+        .map(OffsetDateTime::from)
+}
+
+/// MinIO emits a zero-rule `<LifecycleConfiguration><ExpiryUpdatedAt>…`
+/// document for its delete tombstone / transition-only states; the s3s
+/// deserializer rejects a missing `<Rule>` outright. Such a document is the
+/// "no expiry rules here" statement and must map to delete semantics rather
+/// than an error.
+fn lifecycle_document_has_rules(raw: &[u8]) -> bool {
+    raw.windows(b"<Rule".len()).any(|window| window == b"<Rule")
+}
+
 fn replication_rule_deployment_id(rule: &ReplicationRule) -> Option<String> {
     if let Some(rule_id) = rule.id.as_deref() {
         if let Some(deployment_id) = rule_id.strip_prefix("site-repl-")
@@ -7928,7 +8126,12 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
     } else {
         None
     };
-    if let Ok(bucket_meta) = metadata_sys::get(&item.bucket).await {
+    // lc-config staleness is judged on the expiry axis inside its merge block
+    // below: `lifecycle_config_updated_at` moves on local transition-only
+    // edits too, which would shadow newer peer expiry updates.
+    if item.r#type != "lc-config"
+        && let Ok(bucket_meta) = metadata_sys::get(&item.bucket).await
+    {
         let local_updated_at = bucket_meta_local_updated_at(&bucket_meta, config_file);
         if is_stale_update(local_updated_at, incoming_updated_at) {
             return Ok(());
@@ -7970,6 +8173,71 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
             None
         };
 
+    let merged_lifecycle_config = if item.r#type == "lc-config" {
+        // Receiver-side gate, symmetric with the sender hook: a peer must not
+        // install expiry rules here while `replicateILMExpiry` is off. When
+        // the state cannot be read, fall through and apply (pre-gate
+        // behavior) rather than silently dropping a legitimate update. Note
+        // the gate acks with 200 — the sender treats the item as delivered
+        // and will not retry; items skipped inside the enable-flag
+        // propagation window are healed by repair, not by retry.
+        if let Ok(state) = load_site_replication_state().await
+            && !site_replication_state_replicates_ilm_expiry(&state)
+        {
+            return Ok(());
+        }
+
+        let incoming = match item.expiry_lc_config.as_ref() {
+            Some(raw) => {
+                let data = decode_bucket_meta_wire_value(raw);
+                match deserialize::<s3s::dto::BucketLifecycleConfiguration>(&data) {
+                    Ok(config) => Some(config),
+                    // MinIO's delete tombstone / transition-only state is a
+                    // zero-rule document the strict deserializer rejects; it
+                    // means "no expiry rules here" (delete semantics).
+                    Err(_) if !lifecycle_document_has_rules(&data) => None,
+                    Err(e) => return Err(s3_error!(InvalidRequest, "invalid lifecycle config: {e}")),
+                }
+            }
+            None => None,
+        };
+        let local = match metadata_sys::get_lifecycle_config(&item.bucket).await {
+            Ok((config, _)) => Some(config),
+            Err(StorageError::ConfigNotFound) => None,
+            Err(err) => return Err(ApiError::from(err).into()),
+        };
+        // Staleness axis: the config's expiry_updated_at when present. When
+        // the config is absent (deleted) or predates the axis field, fall
+        // back to the whole-config write time — it survives deletion in
+        // bucket metadata and is the deletion's lower bound, so a delayed
+        // stale broadcast cannot resurrect deleted expiry rules.
+        let whole_config_axis = metadata_sys::get(&item.bucket)
+            .await
+            .map(|bucket_meta| bucket_meta.lifecycle_config_updated_at)
+            .unwrap_or(OffsetDateTime::UNIX_EPOCH);
+        let local_expiry_updated_at = local
+            .as_ref()
+            .and_then(|config| config.expiry_updated_at.clone())
+            .map(OffsetDateTime::from)
+            .unwrap_or(whole_config_axis);
+        if is_stale_update(local_expiry_updated_at, incoming_updated_at) {
+            return Ok(());
+        }
+        let local_absent = local.is_none();
+        match merge_incoming_lifecycle_config(incoming, local, incoming_updated_at) {
+            Some(config) => Some(
+                serialize(&config)
+                    .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize lifecycle failed: {e}")))?,
+            ),
+            None => {
+                skip_config_write = local_absent;
+                None
+            }
+        }
+    } else {
+        None
+    };
+
     let data = match item.r#type.as_str() {
         "policy" => item
             .policy
@@ -7986,7 +8254,7 @@ async fn apply_bucket_meta_item(item: SRBucketMeta) -> S3Result<()> {
         "object-lock-config" => decode_bucket_meta_wire_option(item.object_lock_config),
         "sse-config" => decode_bucket_meta_wire_option(item.sse_config),
         "replication-config" => merged_replication_config,
-        "lc-config" => decode_bucket_meta_wire_option(item.expiry_lc_config),
+        "lc-config" => merged_lifecycle_config,
         "cors-config" => decode_bucket_meta_wire_option(item.cors),
         _ => unreachable!(),
     };
@@ -14299,6 +14567,324 @@ mod tests {
         assert!(merge_incoming_replication_config(Some(site_repl_config("home")), None).is_none());
     }
 
+    fn lc_rule(id: &str, expiry_days: Option<i32>, transition_days: Option<i32>) -> s3s::dto::LifecycleRule {
+        s3s::dto::LifecycleRule {
+            id: Some(id.to_string()),
+            status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
+            prefix: Some(String::new()),
+            expiration: expiry_days.map(|days| s3s::dto::LifecycleExpiration {
+                days: Some(days),
+                ..Default::default()
+            }),
+            transitions: transition_days.map(|days| {
+                vec![s3s::dto::Transition {
+                    days: Some(days),
+                    storage_class: Some(s3s::dto::TransitionStorageClass::from_static(s3s::dto::TransitionStorageClass::GLACIER)),
+                    date: None,
+                }]
+            }),
+            abort_incomplete_multipart_upload: None,
+            del_marker_expiration: None,
+            filter: None,
+            noncurrent_version_expiration: None,
+            noncurrent_version_transitions: None,
+        }
+    }
+
+    fn lc_config(rules: Vec<s3s::dto::LifecycleRule>) -> s3s::dto::BucketLifecycleConfiguration {
+        s3s::dto::BucketLifecycleConfiguration {
+            rules,
+            expiry_updated_at: None,
+        }
+    }
+
+    fn rule_ids(config: &s3s::dto::BucketLifecycleConfiguration) -> Vec<&str> {
+        config.rules.iter().filter_map(|rule| rule.id.as_deref()).collect()
+    }
+
+    /// P1-1 red-light: an incoming expiry-only document must not erase the
+    /// receiver's local transition/tiering rules (today the receiver
+    /// overwrites the whole lifecycle config).
+    #[test]
+    fn test_merge_incoming_lifecycle_preserves_local_transition_rule() {
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![lc_rule("e1", Some(7), None)])),
+            Some(lc_config(vec![lc_rule("t1", None, Some(30))])),
+            None,
+        )
+        .expect("merge should keep rules");
+
+        let mut ids = rule_ids(&merged);
+        ids.sort_unstable();
+        assert_eq!(ids, vec!["e1", "t1"]);
+        let t1 = merged.rules.iter().find(|rule| rule.id.as_deref() == Some("t1")).unwrap();
+        assert!(t1.transitions.as_ref().is_some_and(|t| !t.is_empty()), "local transition must survive");
+    }
+
+    /// Same-id incoming rule updates the expiry side but the local transition
+    /// side is authoritative (MinIO `CloneNonTransition` + restore).
+    #[test]
+    fn test_merge_incoming_lifecycle_same_id_keeps_local_transition() {
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![lc_rule("r1", Some(7), None)])),
+            Some(lc_config(vec![lc_rule("r1", Some(1), Some(30))])),
+            None,
+        )
+        .expect("merge should keep rules");
+
+        assert_eq!(merged.rules.len(), 1);
+        let r1 = &merged.rules[0];
+        assert_eq!(r1.expiration.as_ref().and_then(|e| e.days), Some(7), "incoming expiry wins");
+        assert!(
+            r1.transitions.as_ref().is_some_and(|t| !t.is_empty()),
+            "local transition is authoritative"
+        );
+    }
+
+    /// Trust boundary: whatever the peer sends, its transition fields never
+    /// land here — a new incoming rule is stripped to its expiry parts.
+    #[test]
+    fn test_merge_incoming_lifecycle_strips_incoming_transitions() {
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![lc_rule("r1", Some(7), Some(1))])),
+            Some(lc_config(vec![lc_rule("t1", None, Some(30))])),
+            None,
+        )
+        .expect("merge should keep rules");
+
+        let r1 = merged.rules.iter().find(|rule| rule.id.as_deref() == Some("r1")).unwrap();
+        assert!(
+            r1.transitions.as_ref().is_none_or(|t| t.is_empty()),
+            "incoming transition fields must be discarded"
+        );
+    }
+
+    /// A local rule whose expiry part was dropped upstream loses only the
+    /// expiry fields; a pure-expiry rule disappears entirely.
+    #[test]
+    fn test_merge_incoming_lifecycle_dropped_rule_strips_expiry_keeps_transition() {
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![lc_rule("other", Some(3), None)])),
+            Some(lc_config(vec![
+                lc_rule("mixed", Some(1), Some(30)),
+                lc_rule("pure-expiry", Some(2), None),
+            ])),
+            None,
+        )
+        .expect("merge should keep rules");
+
+        let mut ids = rule_ids(&merged);
+        ids.sort_unstable();
+        assert_eq!(ids, vec!["mixed", "other"], "pure-expiry rule not in the incoming set is removed");
+        let mixed = merged.rules.iter().find(|rule| rule.id.as_deref() == Some("mixed")).unwrap();
+        assert!(mixed.expiration.is_none(), "expiry side cleared");
+        assert!(mixed.transitions.as_ref().is_some_and(|t| !t.is_empty()), "transition side kept");
+    }
+
+    /// Peer lifecycle delete merges with the empty set: local transition rules
+    /// survive with their expiry parts cleared; only when nothing remains does
+    /// the whole config disappear.
+    #[test]
+    fn test_merge_incoming_lifecycle_delete_merges_with_empty() {
+        let merged = merge_incoming_lifecycle_config(
+            None,
+            Some(lc_config(vec![
+                lc_rule("mixed", Some(1), Some(30)),
+                lc_rule("pure-expiry", Some(2), None),
+            ])),
+            None,
+        )
+        .expect("transition rules must survive a peer lifecycle delete");
+        assert_eq!(rule_ids(&merged), vec!["mixed"]);
+        assert!(merged.rules[0].expiration.is_none());
+
+        assert!(
+            merge_incoming_lifecycle_config(None, Some(lc_config(vec![lc_rule("pure-expiry", Some(2), None)])), None).is_none(),
+            "an all-expiry config deletes cleanly"
+        );
+    }
+
+    /// Disabled rules must survive the merge like enabled ones — the merge
+    /// must not reuse ENABLED-filtered helpers.
+    #[test]
+    fn test_merge_incoming_lifecycle_keeps_disabled_transition_rule() {
+        let mut disabled = lc_rule("t-disabled", None, Some(30));
+        disabled.status = s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::DISABLED);
+
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![lc_rule("e1", Some(7), None)])),
+            Some(lc_config(vec![disabled])),
+            None,
+        )
+        .expect("merge should keep rules");
+
+        let mut ids = rule_ids(&merged);
+        ids.sort_unstable();
+        assert_eq!(ids, vec!["e1", "t-disabled"]);
+    }
+
+    /// Abort-multipart-only rules carry no expiry semantics: local ones stay
+    /// untouched, incoming ones are not installed (they are site-local, like
+    /// MinIO's sender-side filter).
+    #[test]
+    fn test_merge_incoming_lifecycle_abort_mpu_rules_stay_local() {
+        let abort_only = |id: &str| s3s::dto::LifecycleRule {
+            id: Some(id.to_string()),
+            status: s3s::dto::ExpirationStatus::from_static(s3s::dto::ExpirationStatus::ENABLED),
+            prefix: Some(String::new()),
+            abort_incomplete_multipart_upload: Some(s3s::dto::AbortIncompleteMultipartUpload {
+                days_after_initiation: Some(3),
+            }),
+            del_marker_expiration: None,
+            expiration: None,
+            filter: None,
+            noncurrent_version_expiration: None,
+            noncurrent_version_transitions: None,
+            transitions: None,
+        };
+
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![abort_only("incoming-abort"), lc_rule("e1", Some(7), None)])),
+            Some(lc_config(vec![abort_only("local-abort")])),
+            None,
+        )
+        .expect("merge should keep rules");
+
+        let mut ids = rule_ids(&merged);
+        ids.sort_unstable();
+        assert_eq!(
+            ids,
+            vec!["e1", "local-abort"],
+            "incoming abort-mpu rule is not installed; local one survives"
+        );
+    }
+
+    /// Repeated delivery of the same document must be byte-stable (rule order
+    /// deterministic), or every broadcast rewrites bucket metadata.
+    #[test]
+    fn test_merge_incoming_lifecycle_is_idempotent() {
+        let incoming = || Some(lc_config(vec![lc_rule("e1", Some(7), None), lc_rule("e2", Some(9), None)]));
+        let local = Some(lc_config(vec![lc_rule("t1", None, Some(30))]));
+
+        let once = merge_incoming_lifecycle_config(incoming(), local, None).expect("first merge");
+        let twice = merge_incoming_lifecycle_config(incoming(), Some(once.clone()), None).expect("second merge");
+
+        assert_eq!(
+            serialize(&once).expect("serialize once"),
+            serialize(&twice).expect("serialize twice"),
+            "merge must be idempotent for identical input"
+        );
+    }
+
+    /// The merged config records the expiry axis timestamp so the staleness
+    /// guard compares expiry updates against expiry updates (a local
+    /// transition-only edit must not shadow newer peer expiry updates).
+    #[test]
+    fn test_merge_incoming_lifecycle_stamps_expiry_updated_at() {
+        let updated_at = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp");
+        let merged = merge_incoming_lifecycle_config(Some(lc_config(vec![lc_rule("e1", Some(7), None)])), None, Some(updated_at))
+            .expect("merge should keep rules");
+
+        let stamped = merged.expiry_updated_at.expect("expiry_updated_at must be stamped");
+        assert_eq!(OffsetDateTime::from(stamped).unix_timestamp(), updated_at.unix_timestamp());
+    }
+
+    /// MinIO's sender never emits del-marker-expiration rules
+    /// (CloneNonTransition drops them), so a MinIO expiry broadcast must not
+    /// delete this site's del-marker-only rules, and the local del-marker
+    /// side of a same-id rule is authoritative.
+    #[test]
+    fn test_merge_incoming_lifecycle_del_marker_rules_stay_local() {
+        let del_marker_only = |id: &str| {
+            let mut rule = lc_rule(id, None, None);
+            rule.del_marker_expiration = Some(s3s::dto::DelMarkerExpiration { days: Some(3) });
+            rule
+        };
+
+        // A local del-marker-only rule survives an incoming expiry document
+        // that does not mention it.
+        let merged = merge_incoming_lifecycle_config(
+            Some(lc_config(vec![lc_rule("e1", Some(7), None)])),
+            Some(lc_config(vec![del_marker_only("dm-local")])),
+            None,
+        )
+        .expect("merge should keep rules");
+        let mut ids = rule_ids(&merged);
+        ids.sort_unstable();
+        assert_eq!(ids, vec!["dm-local", "e1"]);
+
+        // Same-id: the incoming expiry side wins, the local del-marker /
+        // abort-mpu side is authoritative and an incoming del-marker field is
+        // discarded at the trust boundary.
+        let mut local_mixed = lc_rule("r1", Some(1), None);
+        local_mixed.del_marker_expiration = Some(s3s::dto::DelMarkerExpiration { days: Some(3) });
+        local_mixed.abort_incomplete_multipart_upload = Some(s3s::dto::AbortIncompleteMultipartUpload {
+            days_after_initiation: Some(5),
+        });
+        let mut incoming_mixed = lc_rule("r1", Some(7), None);
+        incoming_mixed.del_marker_expiration = Some(s3s::dto::DelMarkerExpiration { days: Some(9) });
+
+        let merged =
+            merge_incoming_lifecycle_config(Some(lc_config(vec![incoming_mixed])), Some(lc_config(vec![local_mixed])), None)
+                .expect("merge should keep rules");
+        let r1 = &merged.rules[0];
+        assert_eq!(r1.expiration.as_ref().and_then(|e| e.days), Some(7));
+        assert_eq!(r1.del_marker_expiration.as_ref().and_then(|d| d.days), Some(3), "local del-marker wins");
+        assert_eq!(
+            r1.abort_incomplete_multipart_upload
+                .as_ref()
+                .and_then(|a| a.days_after_initiation),
+            Some(5),
+            "local abort-mpu wins"
+        );
+    }
+
+    /// MinIO emits a zero-rule lifecycle document (delete tombstone /
+    /// transition-only state); the strict deserializer rejects it, so the
+    /// apply path must recognize it as delete semantics via the rule sniff.
+    #[test]
+    fn test_lifecycle_document_has_rules_sniff() {
+        assert!(!lifecycle_document_has_rules(
+            b"<LifecycleConfiguration><ExpiryUpdatedAt>2026-01-01T00:00:00Z</ExpiryUpdatedAt></LifecycleConfiguration>"
+        ));
+        assert!(lifecycle_document_has_rules(
+            b"<LifecycleConfiguration><Rule><ID>x</ID></Rule></LifecycleConfiguration>"
+        ));
+        assert!(lifecycle_document_has_rules(
+            b"<LifecycleConfiguration><Rule ></Rule></LifecycleConfiguration>"
+        ));
+    }
+
+    /// Sender-side filter: only the expiry subset leaves this site. MinIO
+    /// peers install incoming rules verbatim, so a full document would plant
+    /// this site's transition rules there.
+    #[test]
+    fn test_lifecycle_expiry_subset_xml_strips_transitions() {
+        let full = serialize(&lc_config(vec![lc_rule("mixed", Some(1), Some(30)), lc_rule("t-only", None, Some(7))]))
+            .expect("serialize full config");
+
+        let subset = lifecycle_expiry_subset_xml(&full).expect("expiry subset should remain");
+        let parsed: s3s::dto::BucketLifecycleConfiguration = deserialize(&subset).expect("subset should parse");
+        assert_eq!(rule_ids(&parsed), vec!["mixed"]);
+        assert!(parsed.rules[0].transitions.is_none(), "transition side must not travel");
+
+        let transition_only =
+            serialize(&lc_config(vec![lc_rule("t-only", None, Some(7))])).expect("serialize transition-only config");
+        assert!(
+            lifecycle_expiry_subset_xml(&transition_only).is_none(),
+            "a transition-only config states 'no expiry rules' (delete semantics)"
+        );
+        assert!(lifecycle_expiry_subset_xml(b"").is_none());
+    }
+
+    /// A local parse failure must forward the document unfiltered — mapping
+    /// it to `None` would delete the peers' replicated expiry rules.
+    #[test]
+    fn test_lifecycle_expiry_subset_xml_forwards_unparseable_config() {
+        let garbage = b"<LifecycleConfiguration><Rule></Broken>";
+        assert_eq!(lifecycle_expiry_subset_xml(garbage).as_deref(), Some(garbage.as_slice()));
+    }
+
     // `role` is part of the bucket's S3-visible configuration. Repairing a reverse rule must
     // drop only a sender-owned site-replication ARN, never an operator's own role — the same
     // rule the merge path applies, so both paths agree on what is ours to rewrite.

rustfs/src/app/bucket_usecase.rs

@@ -1158,6 +1158,19 @@ fn lifecycle_has_expiry_rules(config: &BucketLifecycleConfiguration) -> bool {
     })
 }
 
+/// Status-independent presence of the expiry subset that site replication
+/// propagates (`replicateILMExpiry`): expiration / noncurrent-version
+/// expiration only. Distinct from [`lifecycle_has_expiry_rules`], which
+/// filters on ENABLED for scanner scheduling — editing a Disabled expiry rule
+/// must still advance the replication axis. Del-marker expiration and
+/// abort-multipart are site-local and never travel.
+fn lifecycle_rules_have_expiry(config: &BucketLifecycleConfiguration) -> bool {
+    config
+        .rules
+        .iter()
+        .any(|rule| rule.expiration.is_some() || rule.noncurrent_version_expiration.is_some())
+}
+
 fn lifecycle_has_abort_multipart_rules(config: &BucketLifecycleConfiguration) -> bool {
     config.rules.iter().any(|rule| {
         rule.status == ExpirationStatus::from_static(ExpirationStatus::ENABLED)
@@ -2186,7 +2199,24 @@ impl DefaultBucketUsecase {
             return Err(s3_error!(InvalidArgument, "{err}"));
         }
 
-        input_cfg.expiry_updated_at = Some(Timestamp::from(time::OffsetDateTime::now_utc()));
+        // Stamp the expiry axis only when the expiry subset can have changed
+        // (MinIO: HasExpiry() || expiryRuleRemoved). Site-replication peers
+        // judge lc-config staleness on this axis; a transition-only edit that
+        // advanced it would let this site's stale expiry subset shadow — and
+        // roll back — a newer peer expiry edit fleet-wide.
+        let previous_expiry_updated_at = match metadata_sys::get_lifecycle_config(&bucket).await {
+            Ok((previous, _)) => {
+                if lifecycle_rules_have_expiry(&input_cfg) || lifecycle_rules_have_expiry(&previous) {
+                    Some(Timestamp::from(time::OffsetDateTime::now_utc()))
+                } else {
+                    previous.expiry_updated_at
+                }
+            }
+            // No previous config (or unreadable): stamping is the
+            // conservative pre-existing behavior.
+            Err(_) => lifecycle_rules_have_expiry(&input_cfg).then(|| Timestamp::from(time::OffsetDateTime::now_utc())),
+        };
+        input_cfg.expiry_updated_at = previous_expiry_updated_at;
         let data = serialize_config(&input_cfg)?;
         update_bucket_config_for_incarnation(&bucket, BUCKET_LIFECYCLE_CONFIG, data, expected_incarnation_id)
             .await
@@ -2197,7 +2227,14 @@ impl DefaultBucketUsecase {
         let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
         item.expiry_lc_config =
             Some(serialize_config(&input_cfg).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
-        item.expiry_updated_at = item.updated_at;
+        // The item travels with the expiry axis, not the wall clock: a site
+        // whose expiry knowledge is old (or absent — UNIX_EPOCH) must not
+        // out-rank newer peer expiry state at the receivers.
+        item.expiry_updated_at = input_cfg
+            .expiry_updated_at
+            .clone()
+            .map(time::OffsetDateTime::from)
+            .or(Some(time::OffsetDateTime::UNIX_EPOCH));
         if let Err(err) = site_replication_bucket_meta_hook(item).await {
             warn!(bucket = %bucket, error = ?err, "site replication bucket lifecycle hook failed");
         }