fix(site-replication): admit same-generation peer-edit fan-out bodies (#6007)

The peer-edit delivery fence from #5882 treated an equal applied generation as stale. One edit legitimately fans out one delivery per peer record under a single generation (the ILM-expiry edit sends every peer's record), so the receiver applied only the first body, raised its high-water mark, and silently acked-success while dropping the rest — enableILMExpiryReplication never converged on receiving sites and the three-node nightly e2e failed deterministically (issue #5767).

Only a strictly newer applied generation is stale now. Equal generation implies the same logical edit and re-applying a delivery is idempotent (update_peer overwrites the peer record; the mark is raised with max), while strictly older deliveries — the cross-node ordering case the fence exists for — stay rejected.

Adds a composed unit test driving three same-generation bodies through the receiver's fenced sequence, and widens the replication e2e's two site-replication wait helpers from a 10s polling ceiling to the 30s deadline the file's other waits use.

crates/e2e_test/src/replication_extension_test.rs

@@ -2401,15 +2401,20 @@ async fn wait_for_site_replication_info<F>(
 where
     F: Fn(&SiteReplicationInfo) -> bool,
 {
-    for _ in 0..40 {
+    // 30s to match wait_for_replication_state: the three-node site tests run
+    // several full rustfs processes on one runner, so peer-state propagation
+    // can take well over 10s under CI load.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+    loop {
         let info = site_replication_info(env).await?;
         if predicate(&info) {
             return Ok(info);
         }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!("site replication info did not reach expected state on {}", env.address).into());
+        }
         sleep(Duration::from_millis(250)).await;
     }
-
-    Err(format!("site replication info did not reach expected state on {}", env.address).into())
 }
 
 async fn wait_for_site_replication_status<F>(
@@ -2420,15 +2425,19 @@ async fn wait_for_site_replication_status<F>(
 where
     F: Fn(&SRStatusInfo) -> bool,
 {
-    for _ in 0..40 {
+    // Same 30s ceiling as wait_for_site_replication_info: the status probes
+    // fan out to every peer, so they see the same multi-process CI load.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+    loop {
         let status = site_replication_status(env, query).await?;
         if predicate(&status) {
             return Ok(status);
         }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!("site replication status did not reach expected state on {}", env.address).into());
+        }
         sleep(Duration::from_millis(250)).await;
     }
-
-    Err(format!("site replication status did not reach expected state on {}", env.address).into())
 }
 
 async fn wait_for_replication_reset_target<F>(

rustfs/src/admin/handlers/site_replication.rs

@@ -5932,15 +5932,18 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
     Some((origin.clone(), generation))
 }
 
-/// True when a newer edit from the same origin site already landed here. The
+/// True when a strictly newer edit from the same origin site already landed
-/// process mutex on the sending node cannot order deliveries issued by two
+/// here. The process mutex on the sending node cannot order deliveries issued
-/// nodes of that site, so ordering is decided here, on the generation the
+/// by two nodes of that site, so ordering is decided here, on the generation
-/// sender allocated under the distributed lock.
+/// the sender allocated under the distributed lock. Equal generations are NOT
+/// stale: one edit legitimately fans out several deliveries under a single
+/// generation (the ILM-expiry edit sends every peer's record), and a replay of
+/// an applied delivery re-applies the same edit idempotently.
 fn peer_edit_delivery_is_stale(state: &SiteReplicationState, origin: &str, generation: u64) -> bool {
     state
         .applied_edit_generations
         .get(origin)
-        .is_some_and(|applied| *applied >= generation)
+        .is_some_and(|applied| *applied > generation)
 }
 
 fn record_applied_peer_edit_generation(state: &mut SiteReplicationState, origin: &str, generation: u64) {
@@ -12783,8 +12786,11 @@ mod tests {
 
         // The delivery that lost the race carries the older generation.
         assert!(peer_edit_delivery_is_stale(&state, "origin-site", 6));
-        // A replay of the generation already applied is stale too.
+        // The generation already applied is NOT stale: one edit fans out one
-        assert!(peer_edit_delivery_is_stale(&state, "origin-site", 7));
+        // delivery per peer record under a single generation (the ILM-expiry
+        // edit), so an equal-generation delivery is the same edit's next body
+        // (or an idempotent replay) and must apply.
+        assert!(!peer_edit_delivery_is_stale(&state, "origin-site", 7));
         // The next edit from that origin still applies...
         assert!(!peer_edit_delivery_is_stale(&state, "origin-site", 8));
         // ...and another origin site is ordered independently.
@@ -12798,6 +12804,65 @@ mod tests {
         assert!(peer_edit_fence(&HashMap::new()).is_none());
     }
 
+    /// One edit fans out one delivery per peer record under a single
+    /// generation (the ILM-expiry edit sends every peer's record). The
+    /// receiver's fenced sequence — staleness check, apply, raise the
+    /// high-water mark — must therefore accept every body of that fan-out,
+    /// not just the first, while a strictly older delivery stays rejected.
+    #[test]
+    fn peer_edit_fence_admits_every_body_of_one_edits_fan_out() {
+        let local = PeerInfo {
+            deployment_id: "site-a".to_string(),
+            ..peer("site-a", "https://site-a.example.com")
+        };
+        let mut state = SiteReplicationState {
+            peers: BTreeMap::from([
+                ("site-a".to_string(), local.clone()),
+                (
+                    "site-b".to_string(),
+                    PeerInfo {
+                        deployment_id: "site-b".to_string(),
+                        ..peer("site-b", "https://site-b.example.com")
+                    },
+                ),
+                (
+                    "site-c".to_string(),
+                    PeerInfo {
+                        deployment_id: "site-c".to_string(),
+                        ..peer("site-c", "https://site-c.example.com")
+                    },
+                ),
+            ]),
+            ..Default::default()
+        };
+        let origin = "origin-site";
+        let generation = 2;
+
+        let bodies: Vec<PeerInfo> = state
+            .peers
+            .values()
+            .map(|peer| PeerInfo {
+                replicate_ilm_expiry: true,
+                ..peer.clone()
+            })
+            .collect();
+        for body in bodies {
+            assert!(
+                !peer_edit_delivery_is_stale(&state, origin, generation),
+                "a same-generation fan-out body must not be fenced out"
+            );
+            state = apply_internal_peer_edit(state, &local, body, None).expect("fan-out body applies");
+            record_applied_peer_edit_generation(&mut state, origin, generation);
+        }
+
+        assert!(
+            state.peers.values().all(|peer| peer.replicate_ilm_expiry),
+            "every peer record from the fan-out must be applied: {:?}",
+            state.peers
+        );
+        assert!(peer_edit_delivery_is_stale(&state, origin, generation - 1));
+    }
+
     /// P1-15 review follow-up: a site that leaves the mesh drops below two
     /// peers, which clears its state object and restarts its generation
     /// counter at zero. A mark left over from its previous membership would