fix(replication): space periodic MRF replay across nodes

Every node runs the MRF replay loop and the recovery leader lock only
serializes passes, so an N-node cluster replayed the same retained
backlog N times per interval while a target stayed down.

Record the start time of a pass that finds entries in a shared replay
stamp and skip a periodic pass when any node started one within the
interval. The startup pass still always runs.

CHANGELOG.md

@@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - IAM snapshot/deletion replay, target-assigned delete-marker purges, timestamp ordering, and best-effort peer broadcast now close the control-plane gaps found by the R6 review ([#7195](https://github.com/rustfs/rustfs/pull/7195)).
 - A `set-remote-target?update=true&creds=true` request on a site-replication peer's bucket target is now rejected with `400 InvalidRequest`, naming the target and why it counts as peer-owned. It used to return success while keeping the stored endpoint, credentials, and TLS settings, so an endpoint edit from the Console's bucket replication form appeared to apply but did not. Change a peer's address with a site-replication edit instead; see "Changing a site's address" in `docs/operations/site-replication-operations.md`. The `sync`, `bandwidth`, `proxy`, and `path` update groups are unchanged.
 - Site-replication edits that omit the site name, such as `mc admin replicate update --deployment-id <ID> --endpoint <URL>`, now keep the stored name on the editing site and on every receiver instead of renaming the site after the new endpoint host. No persisted format changes.
+- The persisted replication MRF journal is now replayed every 5 minutes while the server runs, not only at startup. Delete-marker purges whose targets were unreachable, and replications that overflowed the worker queues, now converge within one replay interval after the target returns instead of waiting for a restart. A target that stays down is re-attempted once per retained entry per interval across the cluster: the start time of the last pass is shared through a small `config/replication/mrf.bin.replay` object, so nodes do not each replay once per interval. Older versions ignore that object; existing persisted formats are unchanged.
 - Upgrade and rollback: upgrade every node in one site consecutively and verify reconciliation before moving to the next site; do not intentionally run a site mixed-version. Target-version ledger keys are harmless on rollback, although old code cannot use their routing. Before rolling back past [#7307](https://github.com/rustfs/rustfs/pull/7307), drain or repair every pending version purge: older code can free a retained version's data directory before its remote purge is acknowledged. See `docs/operations/site-replication-operations.md`.
 
 ### Security

Cargo.lock

@@ -11058,7 +11058,6 @@ dependencies = [
 name = "rustfs-utils"
 version = "1.0.2"
 dependencies = [
- "aws-lc-rs",
  "base64-simd",
  "blake2",
  "brotli 9.1.1",

Cargo.toml

@@ -212,7 +212,6 @@ serde_urlencoded = "0.7.1"
 # releases.
 aes-gcm = { version = "0.11.1" }
 argon2 = { version = "0.6.0" }
-aws-lc-rs = { version = "1.18.0", default-features = false }
 blake2 = "0.11.0"
 chacha20poly1305 = { version = "0.11.0" }
 crc-fast = "1.10.0"

crates/ecstore/Cargo.toml

@@ -31,7 +31,7 @@ workspace = true
 
 [features]
 default = []
-fips = ["rustfs-s3-client/fips", "rustfs-utils/fips"]
+fips = ["rustfs-s3-client/fips"]
 diagnostic-read-version-caller = ["rustfs-io-metrics/diagnostic-read-version-caller"]
 # Sampled wall/CPU diagnostics only; does not change the read backend.
 diagnostic-local-batch = ["rustix/time"]

crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs

@@ -33,6 +33,7 @@ pub(crate) struct ReplicationMetadataStore;
 impl ReplicationMetadataStore {
     pub(crate) const MRF_REPLICATION_FILE: &'static str = "config/replication/mrf.bin";
     pub(crate) const MRF_REPLICATION_RECOVERY_LOCK: &'static str = "config/replication/mrf.bin.recovery";
+    pub(crate) const MRF_REPLICATION_REPLAY_STAMP: &'static str = "config/replication/mrf.bin.replay";
     pub(crate) const FORCE_DELETE_REPLICATION_FILE: &'static str = "config/replication/force-delete.bin";
     pub(crate) const FORCE_DELETE_REPLICATION_TRANSACTION_LOCK: &'static str = "config/replication/force-delete.bin.transaction";
 
@@ -116,6 +117,10 @@ mod tests {
             ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
             "config/replication/mrf.bin.recovery"
         );
+        assert_eq!(
+            ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP,
+            "config/replication/mrf.bin.replay"
+        );
         assert_eq!(
             ReplicationMetadataStore::FORCE_DELETE_REPLICATION_FILE,
             "config/replication/force-delete.bin"

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -86,6 +86,10 @@ const MRF_CORRUPT_FILE_PREFIX: &str = "config/replication/mrf.corrupt";
 const MRF_PENDING_CAP: usize = 200_000;
 const MRF_RETRY_INITIAL_DELAY: Duration = Duration::from_millis(100);
 const MRF_RETRY_MAX_DELAY: Duration = Duration::from_secs(5);
+/// Interval between runtime replays of the MRF journal, cluster-wide. A pass
+/// re-attempts every retained entry, so a target that stays down costs one
+/// attempt per entry per interval.
+const MRF_REPLAY_INTERVAL: Duration = Duration::from_secs(300);
 
 #[derive(Debug, Default)]
 pub struct DurableMrfBacklog {
@@ -615,7 +619,7 @@ async fn read_mrf_entries<S: ReplicationObjectIO>(storage: Arc<S>) -> Result<Vec
 }
 
 /// Acknowledge only the generation read by the recovery leader. Entries appended
-/// after that generation are retained as a suffix and replayed on the next startup.
+/// after that generation are retained as a suffix and replayed on the next pass.
 /// Lock order: recovery leader lock -> MRF journal object lock.
 async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
     storage: Arc<S>,
@@ -667,7 +671,7 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
     Err(EcstoreError::PreconditionFailed)
 }
 
-/// Acquires the MRF recovery leader lock for the startup replay.
+/// Acquires the MRF recovery leader lock for one journal replay pass.
 /// Returns `None` (after logging) when the lock cannot be created or another
 /// node is already processing the backlog.
 async fn acquire_mrf_recovery_guard<S: ReplicationStorage>(storage: &Arc<S>) -> Option<rustfs_lock::NamespaceLockGuard> {
@@ -1018,7 +1022,7 @@ fn admitted_mrf_replicate_object(oi: ObjectInfo, entry: &MrfReplicateEntry, op_t
 }
 
 /// Acknowledges the replayed MRF prefix and returns the retained backlog.
-/// On acknowledgement failure the backlog is preserved for the next startup and
+/// On acknowledgement failure the backlog is preserved for the next replay pass and
 /// re-read (falling back to the replayed snapshot) so the published summary stays accurate.
 async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
     storage: &Arc<S>,
@@ -1033,7 +1037,7 @@ async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
                 component = LOG_COMPONENT_ECSTORE,
                 subsystem = LOG_SUBSYSTEM_REPLICATION,
                 error = %error,
-                "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
+                "Failed to acknowledge the MRF recovery prefix; preserving it for the next replay pass"
             );
             match read_mrf_entries(storage.clone()).await {
                 Ok(current) => current,
@@ -1051,6 +1055,120 @@ async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
     }
 }
 
+fn unix_millis_now() -> i64 {
+    i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX)
+}
+
+/// Whether a periodic pass may run, given the start time of the last pass on
+/// any node. A stamp more than one interval ahead of this node's clock is skew
+/// or a bad value, so it does not suppress replay.
+fn mrf_replay_due(last_started_ms: Option<i64>, now_ms: i64, interval: Duration) -> bool {
+    let Some(last_started_ms) = last_started_ms else {
+        return true;
+    };
+    let interval_ms = i64::try_from(interval.as_millis()).unwrap_or(i64::MAX);
+    now_ms.saturating_sub(last_started_ms) >= interval_ms || last_started_ms.saturating_sub(now_ms) >= interval_ms
+}
+
+async fn read_mrf_replay_stamp<S: ReplicationStorage>(storage: &Arc<S>) -> Option<i64> {
+    let data = ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP)
+        .await
+        .ok()?;
+    std::str::from_utf8(&data).ok()?.trim().parse().ok()
+}
+
+async fn write_mrf_replay_stamp<S: ReplicationStorage>(storage: &Arc<S>, started_ms: i64) {
+    if let Err(error) = ReplicationConfigStore::save(
+        storage.clone(),
+        ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP,
+        started_ms.to_string().into_bytes(),
+    )
+    .await
+    {
+        warn!(
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION,
+            error = %error,
+            "Failed to record the MRF replay start time"
+        );
+    }
+}
+
+/// One MRF journal replay pass. The recovery leader reads the journal,
+/// re-drives every entry, and acknowledges the replayed prefix; entries that
+/// could not be admitted, and entries appended meanwhile, stay for the next pass.
+///
+/// Every node runs the periodic loop, and the leader lock only serializes
+/// passes. With `min_interval`, a pass is skipped when a pass on any node
+/// started within that interval, read from the shared replay stamp, so a
+/// cluster replays once per interval rather than once per node. The startup
+/// pass passes `None` and always runs. A pass that finds entries records its
+/// start time. Returns whether the journal was read.
+/// Lock order: recovery leader lock -> replay stamp object lock.
+async fn replay_mrf_journal<S: ReplicationStorage>(storage: &Arc<S>, min_interval: Option<Duration>) -> bool {
+    let Some(recovery_guard) = acquire_mrf_recovery_guard(storage).await else {
+        return false;
+    };
+    let started_ms = unix_millis_now();
+    if let Some(interval) = min_interval
+        && !mrf_replay_due(read_mrf_replay_stamp(storage).await, started_ms, interval)
+    {
+        return false;
+    }
+
+    let Some(entries) = load_mrf_recovery_entries(storage).await else {
+        return true;
+    };
+    if !entries.is_empty() {
+        write_mrf_replay_stamp(storage, started_ms).await;
+    }
+    set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
+
+    let total = entries.len();
+    let mut queued_count = 0usize;
+    let mut retry_entries = Vec::new();
+
+    for entry in entries.iter() {
+        let Some(admission) = replay_mrf_entry(entry, storage, &mut retry_entries).await else {
+            continue;
+        };
+
+        if admission == ReplicationQueueAdmission::Missed {
+            retry_entries.push(entry.clone());
+        } else if admission == ReplicationQueueAdmission::Queued {
+            queued_count += 1;
+        }
+    }
+
+    let retained = resolve_retained_mrf_entries(storage, &recovery_guard, &entries, &retry_entries).await;
+    let retained_count = retained.len();
+    set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
+
+    if queued_count > 0 {
+        info!(
+            component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_REPLICATION,
+            recovered = queued_count,
+            total,
+            retained = retained_count,
+            "Replayed MRF entries admitted for retry"
+        );
+    }
+    true
+}
+
+/// Re-runs [`replay_mrf_journal`] every `period`. The recovery leader lock keeps
+/// one pass at a time across the cluster, including the startup pass, and the
+/// shared replay stamp spaces passes from different nodes by `period`.
+fn spawn_mrf_replay_loop<S: ReplicationStorage>(storage: Arc<S>, period: Duration) -> JoinHandle<()> {
+    tokio::spawn(async move {
+        loop {
+            tokio::time::sleep(period).await;
+            replay_mrf_journal(&storage, Some(period)).await;
+        }
+    })
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")]
 struct ResyncActiveConflictError {
@@ -1171,6 +1289,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         // Start background tasks
         pool.start_mrf_persister().await;
         pool.start_mrf_processor().await;
+        pool.start_mrf_replayer().await;
         pool.start_force_delete_processor().await;
 
         pool
@@ -1639,55 +1758,26 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
 
     /// Starts the MRF processor — one-shot at startup.
     ///
-    /// Reads the on-disk MRF file, re-injects admitted entries as Heal operations, and
-    /// rewrites any entries that could not be admitted for a later startup retry.
+    /// Replays the on-disk MRF journal once; [`Self::start_mrf_replayer`] repeats
+    /// the pass while the server runs.
     async fn start_mrf_processor(&self) {
         let storage = self.storage.clone();
-
         let handle = tokio::spawn(async move {
-            let Some(recovery_guard) = acquire_mrf_recovery_guard(&storage).await else {
-                return;
-            };
-
-            let Some(entries) = load_mrf_recovery_entries(&storage).await else {
-                return;
-            };
-            set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
-
-            let total = entries.len();
-            let mut queued_count = 0usize;
-            let mut retry_entries = Vec::new();
-
-            for entry in entries.iter() {
-                let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
-                    continue;
-                };
-
-                if admission == ReplicationQueueAdmission::Missed {
-                    retry_entries.push(entry.clone());
-                } else if admission == ReplicationQueueAdmission::Queued {
-                    queued_count += 1;
-                }
-            }
-
-            let retained = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
-            let retained_count = retained.len();
-            set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
-
-            if queued_count > 0 {
-                info!(
-                    component = LOG_COMPONENT_ECSTORE,
-                    subsystem = LOG_SUBSYSTEM_REPLICATION,
-                    recovered = queued_count,
-                    total,
-                    retained = retained_count,
-                    "Replayed MRF entries admitted for retry"
-                );
-            }
+            replay_mrf_journal(&storage, None).await;
         });
         self.task_handles.lock().await.push(handle);
     }
 
+    /// Starts the runtime MRF replayer — ongoing background task.
+    ///
+    /// Entries reach the journal while the server runs (worker-queue overflow,
+    /// delete-marker purge intents whose watch window drained with targets still
+    /// dirty). Replaying only at startup left them waiting for a restart.
+    async fn start_mrf_replayer(&self) {
+        let handle = spawn_mrf_replay_loop(self.storage.clone(), MRF_REPLAY_INTERVAL);
+        self.task_handles.lock().await.push(handle);
+    }
+
     async fn start_force_delete_processor(&self) {
         let storage = self.storage.clone();
         let handle =
@@ -1755,7 +1845,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     /// writes them to the on-disk MRF file every flush interval (default 10s,
     /// overridable via `RUSTFS_REPL_MRF_FLUSH_INTERVAL_MS`) or when 1 000 new
     /// entries accumulate. Runtime writers append under the journal lock; only the
-    /// startup recovery leader may remove an acknowledged prefix.
+    /// recovery leader of a replay pass may remove an acknowledged prefix.
     async fn start_mrf_persister(&self) {
         let Some(mut rx) = self.mrf_save_rx.lock().await.take() else {
             return;
@@ -3422,6 +3512,7 @@ mod tests {
         read_count: AtomicUsize,
         delete_storage_calls: AtomicUsize,
         delete_error: StdMutex<EcstoreError>,
+        replay_stamp: StdMutex<Option<Vec<u8>>>,
         reported_size: StdMutex<Option<i64>>,
         stream_read_bytes: Arc<AtomicUsize>,
         write_count: AtomicUsize,
@@ -3489,6 +3580,24 @@ mod tests {
             _h: Self::HeaderMap,
             _opts: &Self::ObjectOptions,
         ) -> Result<Self::GetObjectReader, Self::Error> {
+            if object.ends_with(ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP) {
+                let stamp = self
+                    .shared
+                    .replay_stamp
+                    .lock()
+                    .expect("test replay stamp lock should not be poisoned")
+                    .clone()
+                    .ok_or(EcstoreError::FileNotFound)?;
+                return Ok(Self::GetObjectReader {
+                    object_info: ObjectInfo {
+                        size: i64::try_from(stamp.len()).expect("test stamp length should fit i64"),
+                        ..Default::default()
+                    },
+                    stream: Box::new(Cursor::new(stamp)),
+                    buffered_body: None,
+                    body_source: Default::default(),
+                });
+            }
             if !object.ends_with("/.replication/resync.bin")
                 && !object.ends_with("config/replication/mrf.bin")
                 && !object.ends_with("config/replication/force-delete.bin")
@@ -3552,6 +3661,16 @@ mod tests {
                 let lock = self.new_ns_lock(bucket, object).await?;
                 Some(lock.get_write_lock(Duration::from_secs(10)).await?)
             };
+            if object.ends_with(ReplicationMetadataStore::MRF_REPLICATION_REPLAY_STAMP) {
+                let mut stamp = Vec::new();
+                data.stream.read_to_end(&mut stamp).await.map_err(EcstoreError::from)?;
+                *self
+                    .shared
+                    .replay_stamp
+                    .lock()
+                    .expect("test replay stamp lock should not be poisoned") = Some(stamp);
+                return Ok(ObjectInfo::default());
+            }
             if opts.http_preconditions.is_some()
                 && let Some(replacement) = self
                     .shared
@@ -4178,6 +4297,7 @@ mod tests {
             read_count: AtomicUsize::new(0),
             delete_storage_calls: AtomicUsize::new(0),
             delete_error: StdMutex::new(EcstoreError::NotImplemented),
+            replay_stamp: StdMutex::new(None),
             reported_size: StdMutex::new(None),
             stream_read_bytes: Arc::new(AtomicUsize::new(0)),
             write_count: AtomicUsize::new(0),
@@ -5409,6 +5529,7 @@ mod tests {
                 read_count: AtomicUsize::new(0),
                 delete_storage_calls: AtomicUsize::new(0),
                 delete_error: StdMutex::new(EcstoreError::NotImplemented),
+                replay_stamp: StdMutex::new(None),
                 reported_size: StdMutex::new(None),
                 stream_read_bytes: Arc::new(AtomicUsize::new(0)),
                 write_count: AtomicUsize::new(0),
@@ -6373,6 +6494,108 @@ mod tests {
         assert_eq!(retained.target_arns, entry.target_arns);
     }
 
+    #[tokio::test]
+    async fn mrf_replay_loop_drains_entries_persisted_after_startup() {
+        let shared = empty_resync_shared_state();
+        let storage = Arc::new(LoadResyncNodeStore::new("mrf-runtime-replay", shared.clone()));
+        // A force-delete intent from a superseded generation replays as
+        // Skipped, so a pass that reaches it acknowledges it off the journal.
+        let stale_intent = |object: &str| MrfReplicateEntry {
+            bucket: "mrf-runtime-replay".to_string(),
+            object: object.to_string(),
+            op: MrfOpKind::Delete,
+            force_delete_id: Some(uuid::Uuid::new_v4()),
+            force_delete_local_commit: false,
+            target_arns: vec!["arn:rustfs:replication:target-a".to_string()],
+            ..Default::default()
+        };
+        let journal_drained = || async {
+            tokio::time::timeout(Duration::from_secs(5), async {
+                while !shared.data.lock().expect("test data lock should not be poisoned").is_empty() {
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await
+            .is_ok()
+        };
+
+        let handle = spawn_mrf_replay_loop(storage, Duration::from_millis(20));
+        for object in ["first", "second"] {
+            *shared.data.lock().expect("test data lock should not be poisoned") =
+                encode_mrf_file(&[stale_intent(object)]).expect("MRF entry should encode");
+            assert!(
+                journal_drained().await,
+                "an MRF entry persisted while the server runs must be replayed without a restart ({object})"
+            );
+        }
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn mrf_periodic_replay_runs_once_per_interval_across_nodes() {
+        let shared = empty_resync_shared_state();
+        let node_a = Arc::new(LoadResyncNodeStore::new("mrf-node-a", shared.clone()));
+        let node_b = Arc::new(LoadResyncNodeStore::new("mrf-node-b", shared.clone()));
+        let interval = Duration::from_secs(1);
+        let persist_backlog = |object: &str| {
+            let entry = MrfReplicateEntry {
+                bucket: "mrf-node-replay".to_string(),
+                object: object.to_string(),
+                op: MrfOpKind::Delete,
+                force_delete_id: Some(uuid::Uuid::new_v4()),
+                force_delete_local_commit: false,
+                target_arns: vec!["arn:rustfs:replication:target-a".to_string()],
+                ..Default::default()
+            };
+            *shared.data.lock().expect("test data lock should not be poisoned") =
+                encode_mrf_file(&[entry]).expect("MRF entry should encode");
+        };
+        let journal_is_empty = || shared.data.lock().expect("test data lock should not be poisoned").is_empty();
+
+        persist_backlog("first");
+        assert!(replay_mrf_journal(&node_a, Some(interval)).await, "node A's periodic pass should replay");
+        assert!(journal_is_empty(), "node A's pass should acknowledge the replayed entry");
+
+        // Node B's timer fires within the same interval with a backlog present.
+        persist_backlog("second");
+        assert!(
+            !replay_mrf_journal(&node_b, Some(interval)).await,
+            "a node must not replay within one interval of another node's pass"
+        );
+        assert!(!journal_is_empty(), "the backlog should wait for the next interval");
+
+        assert!(
+            replay_mrf_journal(&node_b, None).await,
+            "the startup pass should replay regardless of the shared stamp"
+        );
+        assert!(journal_is_empty(), "the startup pass should acknowledge the replayed entry");
+
+        persist_backlog("third");
+        tokio::time::sleep(interval).await;
+        assert!(
+            replay_mrf_journal(&node_a, Some(interval)).await,
+            "a periodic pass should replay once the interval has elapsed"
+        );
+        assert!(journal_is_empty(), "the due pass should acknowledge the replayed entry");
+    }
+
+    #[test]
+    fn mrf_replay_due_tolerates_missing_and_skewed_stamps() {
+        let interval = Duration::from_secs(300);
+        let now = 1_000_000_000;
+        assert!(mrf_replay_due(None, now, interval));
+        assert!(!mrf_replay_due(Some(now - 299_999), now, interval));
+        assert!(mrf_replay_due(Some(now - 300_000), now, interval));
+        assert!(
+            !mrf_replay_due(Some(now + 1_000), now, interval),
+            "small clock skew should still space passes"
+        );
+        assert!(
+            mrf_replay_due(Some(now + 300_000), now, interval),
+            "a stamp an interval ahead must not suppress replay"
+        );
+    }
+
     #[test]
     fn mrf_entry_delete_marker_roundtrip() {
         let dm_vid = Uuid::new_v4();

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -3015,7 +3015,7 @@ const DELETE_MARKER_PURGE_WATCH_INTERVAL: TokioDuration = TokioDuration::from_se
 /// If the marker disappears (deleted before or while the replica landed),
 /// purge the replicated marker from the targets, retrying failed targets on
 /// later rounds. When the window drains with targets still dirty, persist the
-/// purge intent as a durable MRF entry so the next startup replays it through
+/// purge intent as a durable MRF entry so the next MRF replay pass re-drives it through
 /// `purge_stale_delete_marker_targets`.
 async fn watch_and_purge_source_delete_marker<S: ReplicationStorage>(
     bucket: String,
@@ -3106,7 +3106,7 @@ async fn enqueue_delete_marker_purge_mrf(dobj: &DeletedObjectReplicationInfo, fa
 /// Heal callers retry for the full watch window because the startup MRF
 /// processor runs before bucket metadata (and thus target clients) finishes
 /// initializing — the first pass can see `target_client_missing` and a later
-/// round resolves the client; the replay loop is serial and startup-only, so
+/// round resolves the client; replay passes are serial background work, so
 /// blocking it for up to the window per dirty entry is acceptable. Live
 /// callers run on replication workers where a down target would pin a worker
 /// for the whole window, so they attempt once and lean on the durable intent

crates/ecstore/src/cluster/rpc/http_auth.rs

@@ -32,6 +32,7 @@ use crate::storage_api_contracts::internode::{
     PUT_FILE_AUTH_TRAILER_MAGIC, PUT_FILE_CAPABILITY_VERSION,
 };
 
+use hmac::{Hmac, KeyInit, Mac};
 use http::uri::Authority;
 use http::{HeaderMap, HeaderValue, Method, Uri};
 #[cfg(test)]
@@ -46,7 +47,6 @@ use rustfs_io_metrics::internode_metrics::{
     global_internode_metrics,
 };
 use rustfs_object_data_cache::{MemoryBasis, resolve_effective_memory};
-use rustfs_utils::crypto::{HmacSha256, KeyInit, Mac};
 use rustfs_utils::get_env_bool;
 use sha2::Digest as _;
 use sha2::Sha256;
@@ -58,6 +58,8 @@ use time::OffsetDateTime;
 use tracing::{error, info, warn};
 use uuid::Uuid;
 
+type HmacSha256 = Hmac<Sha256>;
+
 const SIGNATURE_HEADER: &str = "x-rustfs-signature";
 pub const RPC_BACKGROUND_ADMISSION_HEADER: &str = "x-rustfs-rpc-background-admission";
 const RPC_BACKGROUND_ADMISSION_V1: &str = "v1";

crates/utils/Cargo.toml

@@ -26,7 +26,6 @@ categories = ["web-programming", "development-tools", "cryptography"]
 
 [dependencies]
 hotpath.workspace = true
-aws-lc-rs = { workspace = true, optional = true }
 base64-simd = { workspace = true, optional = true }
 blake2 = { workspace = true, optional = true }
 brotli = { workspace = true, optional = true }
@@ -90,7 +89,6 @@ path = []  # path manipulation features
 compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"]
 string = ["dep:regex"]
 crypto = ["dep:base64-simd", "dep:hex-simd", "dep:hmac", "dep:hyper", "dep:sha1", "dep:sha2"]
-fips = ["crypto", "dep:aws-lc-rs", "aws-lc-rs/fips"]
 hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake2", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:crc-fast"]
 os = ["dep:rustix", "dep:tempfile", "dep:windows"]  # operating system utilities
 integration = []  # integration test features

crates/utils/src/crypto.rs

@@ -13,77 +13,12 @@
 // limitations under the License.
 
 use hex_simd::{AsOut, AsciiCase};
-use hmac::Hmac;
-#[cfg(not(feature = "fips"))]
-pub use hmac::{KeyInit, Mac};
+use hmac::{Hmac, KeyInit, Mac};
 use hyper::body::Bytes;
-#[cfg(not(feature = "fips"))]
 use sha1::Sha1;
 use sha2::{Digest, Sha256};
 use std::mem::MaybeUninit;
 
-#[cfg(not(feature = "fips"))]
-pub type HmacSha256 = Hmac<Sha256>;
-
-#[cfg(feature = "fips")]
-pub struct HmacSha256Output([u8; 32]);
-
-#[cfg(feature = "fips")]
-impl HmacSha256Output {
-    #[must_use]
-    pub fn into_bytes(self) -> hmac::digest::Output<Hmac<Sha256>> {
-        self.0.into()
-    }
-}
-
-#[cfg(feature = "fips")]
-pub trait KeyInit: Sized {
-    fn new_from_slice(key: &[u8]) -> Result<Self, hmac::digest::InvalidLength>;
-}
-
-#[cfg(feature = "fips")]
-pub trait Mac: Sized {
-    fn update(&mut self, data: &[u8]);
-    fn finalize(self) -> HmacSha256Output;
-    fn verify_slice(self, tag: &[u8]) -> Result<(), hmac::digest::MacError>;
-}
-
-#[cfg(feature = "fips")]
-#[derive(Clone)]
-pub struct HmacSha256(aws_lc_rs::hmac::Context);
-
-#[cfg(feature = "fips")]
-impl HmacSha256 {
-    pub fn new_from_slice(key: &[u8]) -> Result<Self, hmac::digest::InvalidLength> {
-        let signing_key = aws_lc_rs::hmac::Key::new(aws_lc_rs::hmac::HMAC_SHA256, key);
-        Ok(Self(aws_lc_rs::hmac::Context::with_key(&signing_key)))
-    }
-}
-
-#[cfg(feature = "fips")]
-impl KeyInit for HmacSha256 {
-    fn new_from_slice(key: &[u8]) -> Result<Self, hmac::digest::InvalidLength> {
-        Self::new_from_slice(key)
-    }
-}
-
-#[cfg(feature = "fips")]
-impl Mac for HmacSha256 {
-    fn update(&mut self, data: &[u8]) {
-        self.0.update(data);
-    }
-
-    fn finalize(self) -> HmacSha256Output {
-        let tag = self.0.sign();
-        HmacSha256Output(tag.as_ref().try_into().expect("HMAC-SHA256 tag is 32 bytes"))
-    }
-
-    fn verify_slice(self, tag: &[u8]) -> Result<(), hmac::digest::MacError> {
-        let computed = self.0.sign();
-        aws_lc_rs::constant_time::verify_slices_are_equal(computed.as_ref(), tag).map_err(|_| hmac::digest::MacError)
-    }
-}
-
 /// Base64 URL safe encoding without padding
 /// `base64_encode_url_safe_no_pad(input)`
 ///
@@ -149,18 +84,9 @@ pub fn is_sha256_checksum(s: &str) -> bool {
 /// A 20-byte array containing the HMAC-SHA1 hash of the input data using the provided key
 ///
 pub fn hmac_sha1(key: impl AsRef<[u8]>, data: impl AsRef<[u8]>) -> [u8; 20] {
-    #[cfg(feature = "fips")]
-    {
-        let signing_key = aws_lc_rs::hmac::Key::new(aws_lc_rs::hmac::HMAC_SHA1_FOR_LEGACY_USE_ONLY, key.as_ref());
-        let tag = aws_lc_rs::hmac::sign(&signing_key, data.as_ref());
-        tag.as_ref().try_into().expect("HMAC-SHA1 tag is 20 bytes")
-    }
-    #[cfg(not(feature = "fips"))]
-    {
-        let mut m = <Hmac<Sha1>>::new_from_slice(key.as_ref()).expect("operation should succeed");
-        m.update(data.as_ref());
-        m.finalize().into_bytes().into()
-    }
+    let mut m = <Hmac<Sha1>>::new_from_slice(key.as_ref()).expect("operation should succeed");
+    m.update(data.as_ref());
+    m.finalize().into_bytes().into()
 }
 
 /// HMAC-SHA256 hashing
@@ -174,7 +100,7 @@ pub fn hmac_sha1(key: impl AsRef<[u8]>, data: impl AsRef<[u8]>) -> [u8; 20] {
 /// A 32-byte array containing the HMAC-SHA256 hash of the input data using the provided key
 ///
 pub fn hmac_sha256(key: impl AsRef<[u8]>, data: impl AsRef<[u8]>) -> [u8; 32] {
-    let mut m = HmacSha256::new_from_slice(key.as_ref()).expect("operation should succeed");
+    let mut m = Hmac::<Sha256>::new_from_slice(key.as_ref()).expect("operation should succeed");
     m.update(data.as_ref());
     m.finalize().into_bytes().into()
 }
@@ -241,20 +167,3 @@ fn sha256_checksum_rejects_wrong_length_or_non_lowercase_hex() {
     assert!(!is_sha256_checksum("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdeg"));
     assert!(!is_sha256_checksum("0123456789ABCDEF0123456789abcdef0123456789abcdef0123456789abcdef"));
 }
-
-#[test]
-fn test_hmac_sha256_rfc4231() {
-    #[cfg(feature = "fips")]
-    aws_lc_rs::try_fips_mode().expect("aws-lc-rs FIPS mode must be active when `rustfs-utils/fips` is enabled");
-
-    let key = [0x0b_u8; 20];
-    let data = b"Hi There";
-    assert_eq!(hex(hmac_sha1(key, data)), "b617318655057264e28bc0b6fb378c8ef146be00");
-    let mac = hmac_sha256(key, data);
-    assert_eq!(hex(mac), "b0344c61d8db38535ca8afceaf0bf12b881dc200c9833da726e9376c2e32cff7");
-
-    let mut verifier = <HmacSha256 as KeyInit>::new_from_slice(&key).expect("valid key");
-    verifier.update(data);
-    assert!(verifier.clone().verify_slice(&mac).is_ok());
-    assert!(verifier.verify_slice(&[0_u8; 32]).is_err());
-}

docs/operations/kms-cryptographic-compliance.md

@@ -14,7 +14,6 @@ This document records where RustFS stands on cryptographic module validation and
 | Layer | Where | Implementation | Primitives | Validated module |
 | --- | --- | --- | --- | --- |
 | TLS using the process-wide rustls provider | Provider installed or checked by `install_default_crypto_provider` in `rustfs/src/startup_runtime_hooks.rs` | `rustls` with the process-wide provider (AWS-LC is installed when none exists) | TLS 1.2/1.3 suites, `prefer-post-quantum` hybrid key exchange (restricted to FIPS-approved suites and groups when `--features fips` is enabled) | No in the default build (`aws-lc-sys`); opt-in `--features fips` enables `aws-lc-fips-sys` and checks the installed provider at startup; module validation depends on the deployment |
-| Internode RPC HMAC authentication | `crates/ecstore/src/cluster/rpc/http_auth.rs` via `rustfs_utils::crypto::HmacSha256` | RustCrypto `hmac` + `sha2` in the default build; `aws-lc-rs` when `fips` is enabled | HMAC-SHA256 (`x-rustfs-signature`, `x-rustfs-rpc-signature-v2`, `x-rustfs-rpc-signature-v3`) | No in the default build; opt-in `--features fips` enables `rustfs-utils/fips` (`aws-lc-fips-sys`) |
 | Object data path AEAD (SSE) | `crates/kms/src/encryption/ciphers.rs`, `crates/rio/src/encrypt_reader.rs`, `crates/rio-v2/src/encrypt_reader.rs` | RustCrypto `aes-gcm`, `chacha20poly1305` | AES-256-GCM, ChaCha20-Poly1305 | No |
 | DEK wrapping | `crates/kms/src/encryption/dek.rs` | RustCrypto `aes-gcm` | AES-256-GCM | No |
 | Local KMS backend master key | `crates/kms/src/backends/local.rs` | RustCrypto `argon2`, `aes-gcm` | Argon2id KDF, AES-256-GCM | No |

docs/operations/site-replication-operations.md

@@ -131,7 +131,10 @@ persist until the refresh settles.
 Control-plane retry begins on the 30-second drain, while heavyweight snapshots,
 pending topology operations, and bucket wiring are revisited on the 600-second
 pass. Object MRF entries are persisted every 10 seconds by default and target
-health is probed every 5 seconds. These are scheduling bounds, not delivery
+health is probed every 5 seconds. The persisted MRF journal is replayed at
+startup and then once every 5 minutes across the cluster, so a delete-marker purge or an overflowed
+replication whose target was down converges within one replay interval after
+the target returns. These are scheduling bounds, not delivery
 SLAs: network timeouts and the amount of queued work add to them.
 
 Objects that must be rediscovered by the scanner have this conservative upper

rustfs/Cargo.toml

@@ -58,7 +58,7 @@ required-features = ["swift"]
 
 [features]
 default = ["ftps", "webdav", "gcs"]
-fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips", "rustfs-ecstore/fips", "rustfs-protocols/fips", "rustfs-s3-client/fips", "rustfs-targets/fips", "rustfs-utils/fips"]
+fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips", "rustfs-ecstore/fips", "rustfs-protocols/fips", "rustfs-s3-client/fips", "rustfs-targets/fips"]
 gcs = ["rustfs-ecstore/gcs", "dep:google-cloud-auth"]
 metrics-gpu = ["rustfs-obs/gpu"]
 ftps = ["rustfs-protocols/ftps"]

rustfs/src/app/object/delete.rs

@@ -295,32 +295,16 @@ fn enrich_delete_replication_state_if_needed(
     }
 }
 
-/// The current peer set is read only when a rule has the derived shape. If it
-/// cannot be read, no rule is skipped: a redundant echo is safer than a chained
-/// target missing the delete.
-async fn replica_delete_fanout_state(
+fn should_schedule_replica_delete_replication(
     snapshot: &DeleteReplicationConfigSnapshot,
-    bucket: &str,
     replication_source: &ObjectInfo,
     version_id: Option<Uuid>,
-) -> Option<ReplicationState> {
-    let config = snapshot.replication_config()?;
-    let site_peer_deployment_ids = if has_site_replication_shaped_rule(config) {
-        match crate::site_replication::site_replication_remote_peer_deployment_ids().await {
-            Ok(peers) => peers,
-            Err(error) => {
-                warn!(
-                    bucket = %bucket,
-                    error = %error,
-                    "failed to load site replication peers; forwarding replica delete along every rule"
-                );
-                HashSet::new()
-            }
-        }
-    } else {
-        HashSet::new()
+) -> bool {
+    let Some(config) = snapshot.replication_config() else {
+        return false;
     };
-    replica_delete_fanout_state_from_config(config, replication_source, version_id, &site_peer_deployment_ids)
+
+    delete_replication_state_from_config(config, replication_source, version_id, true).is_some()
 }
 
 fn validate_undo_delete_version(expected: Option<&str>, requested: Option<&str>) -> S3Result<()> {
@@ -1363,21 +1347,12 @@ impl DefaultObjectUsecase {
         let deleted_delete_marker_version = deleted_replication_info.is_some_and(|info| info.delete_marker);
 
         let delete_replication_version_id = delete_replication_version_id(deleted_object_source, deleted_delete_marker_version);
-        let replica_fanout_state = if opts.replication_request && replica {
-            Some(
-                replica_delete_fanout_state(
-                    &delete_config_snapshot,
-                    &bucket,
-                    replication_state_source,
-                    delete_replication_version_id,
-                )
-                .await,
+        let schedule_delete_replication = if opts.replication_request && replica {
+            should_schedule_replica_delete_replication(
+                &delete_config_snapshot,
+                replication_state_source,
+                delete_replication_version_id,
             )
-        } else {
-            None
-        };
-        let schedule_delete_replication = if let Some(fanout_state) = &replica_fanout_state {
-            fanout_state.is_some()
         } else {
             should_schedule_delete_replication(
                 &opts,
@@ -1406,12 +1381,8 @@ impl DefaultObjectUsecase {
                 replication_state: None,
                 ..Default::default()
             };
-            if let Some(Some(fanout_state)) = &replica_fanout_state {
-                set_deleted_object_replication_state(&mut deleted_object, fanout_state);
-            } else {
-                set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state());
-                enrich_delete_replication_state_if_needed(&delete_config_snapshot, &mut deleted_object, replication_state_source);
-            }
+            set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state());
+            enrich_delete_replication_state_if_needed(&delete_config_snapshot, &mut deleted_object, replication_state_source);
             schedule_replication_delete(deleted_object, bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await;
         }
 
@@ -3612,83 +3583,6 @@ mod tests {
         assert!(state.purge_targets.contains_key(&arn));
     }
 
-    #[test]
-    fn replica_delete_fanout_skips_site_replication_peers_but_keeps_chain_targets() {
-        let peer_arn = "arn:rustfs:replication::peer-dep:bucket";
-        let chain_arn = "arn:aws:s3:::downstream-bucket".to_string();
-        let site_rule = crate::site_replication::build_site_replication_rule(peer_arn, 10, "site-repl-peer-dep");
-        let mut chain_rule = site_rule.clone();
-        chain_rule.id = Some("chain".to_string());
-        chain_rule.priority = Some(1);
-        chain_rule.destination.bucket = chain_arn.clone();
-        let site_only = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![site_rule.clone()],
-        };
-        let site_and_chain = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![site_rule, chain_rule],
-        };
-        let marker = ObjectInfo {
-            bucket: "bucket".to_string(),
-            name: "object".to_string(),
-            delete_marker: true,
-            replication_status: ReplicationStatusType::Replica,
-            ..Default::default()
-        };
-        let version = ObjectInfo {
-            delete_marker: false,
-            ..marker.clone()
-        };
-        // An ordinary bucket target ARN carries its target id, so an operator
-        // rule named `site-repl-<target id>` has the derived shape.
-        let shaped_operator_arn = "arn:minio:replication:us-east-1:target-1:downstream".to_string();
-        let mut shaped_operator_rule =
-            crate::site_replication::build_site_replication_rule(&shaped_operator_arn, 1, "site-repl-target-1");
-        shaped_operator_rule.priority = Some(1);
-        let shaped_operator_only = ReplicationConfiguration {
-            role: String::new(),
-            rules: vec![shaped_operator_rule],
-        };
-        assert!(has_site_replication_shaped_rule(&shaped_operator_only));
-        let site_peers = HashSet::from(["peer-dep".to_string()]);
-        let no_site_peers = HashSet::new();
-        let removed_peer_only = HashSet::from(["other-dep".to_string()]);
-        let purged_version = Some(Uuid::new_v4());
-        let fanout_targets = |state: ReplicationState, version_id: Option<Uuid>| {
-            if version_id.is_some() {
-                state.purge_targets.keys().cloned().collect::<Vec<_>>()
-            } else {
-                state.targets.keys().cloned().collect::<Vec<_>>()
-            }
-        };
-
-        for (source, version_id) in [(&marker, None), (&marker, purged_version), (&version, purged_version)] {
-            assert!(
-                delete_replication_state_from_config(&site_only, source, version_id, true).is_some(),
-                "the site-replication rule alone would forward the replica delete"
-            );
-            assert!(
-                replica_delete_fanout_state_from_config(&site_only, source, version_id, &site_peers).is_none(),
-                "a replica delete must not echo back across the site-replication mesh"
-            );
-
-            let state = replica_delete_fanout_state_from_config(&site_and_chain, source, version_id, &site_peers)
-                .expect("a replica delete must still follow a chained operator rule");
-            assert!(!state.replicate_decision_str.contains(peer_arn));
-            assert_eq!(fanout_targets(state, version_id), vec![chain_arn.clone()]);
-
-            for peers in [&no_site_peers, &removed_peer_only, &site_peers] {
-                let state = replica_delete_fanout_state_from_config(&shaped_operator_only, source, version_id, peers)
-                    .expect("an operator rule naming no current site peer must keep forwarding replica deletes");
-                assert_eq!(fanout_targets(state, version_id), vec![shaped_operator_arn.clone()]);
-            }
-            let state = replica_delete_fanout_state_from_config(&site_only, source, version_id, &no_site_peers)
-                .expect("with site replication disabled a site-repl rule is an operator rule");
-            assert_eq!(fanout_targets(state, version_id), vec![peer_arn.to_string()]);
-        }
-    }
-
     #[test]
     fn replica_delete_enrichment_must_not_reuse_upstream_targets() {
         let upstream_state = ReplicationState {

rustfs/src/app/object/mod.rs

@@ -29,9 +29,8 @@ use super::storage_api::object_usecase::access::{
 #[cfg(test)]
 use super::storage_api::object_usecase::bucket::quota::BucketQuota;
 use super::storage_api::object_usecase::bucket::quota::checker::QuotaChecker;
-use super::storage_api::object_usecase::bucket::replication::ReplicationState;
 #[cfg(test)]
-use super::storage_api::object_usecase::bucket::replication::replication_statuses_map;
+use super::storage_api::object_usecase::bucket::replication::{ReplicationState, replication_statuses_map};
 use super::storage_api::object_usecase::bucket::{
     VersioningConfigExt as _,
     lifecycle::{
@@ -50,11 +49,10 @@ use super::storage_api::object_usecase::bucket::{
     replication::{
         DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, commit_force_delete_intent,
         delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete,
-        force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, has_site_replication_shaped_rule,
-        load_delete_config_snapshot, must_replicate_object, persist_force_delete_intent, record_replication_proxy,
-        replica_delete_fanout_state_from_config, schedule_object_replication, schedule_replication_delete,
-        schedule_replication_deletes, set_deleted_object_replication_state, should_schedule_delete_replication,
-        should_use_existing_delete_replication_info,
+        force_delete_target_set, get_read_proxy_targets, has_active_delete_rule, load_delete_config_snapshot,
+        must_replicate_object, persist_force_delete_intent, record_replication_proxy, schedule_object_replication,
+        schedule_replication_delete, schedule_replication_deletes, set_deleted_object_replication_state,
+        should_schedule_delete_replication, should_use_existing_delete_replication_info,
     },
     tagging::decode_tags,
     validate_restore_request,

rustfs/src/app/storage_api.rs

@@ -708,6 +708,7 @@ pub(crate) mod bucket {
         pub(crate) type DeleteReplicationConfigSnapshot =
             crate::storage::storage_api::ecstore_bucket::replication::DeleteReplicationConfigSnapshot;
         pub(crate) type ReplicateDecision = replication_contracts::ReplicateDecision;
+        #[cfg(test)]
         pub(crate) type ReplicationState = replication_contracts::ReplicationState;
         pub(crate) type ReplicationStatusType = replication_contracts::ReplicationStatusType;
         pub(crate) type ReplicationTargetValidationError = replication_contracts::ReplicationTargetValidationError;
@@ -958,52 +959,6 @@ pub(crate) mod bucket {
             replication_contracts::delete_replication_state_from_config(config, &source)
         }
 
-        /// Replication state for forwarding a replica delete (an inbound
-        /// replication request) to this site's own targets.
-        ///
-        /// Operator rules with ReplicaModifications keep forwarding, so chained
-        /// replication A -> B -> C propagates deletes (rustfs#2609). Derived
-        /// site-replication rules never do: site replication is a full mesh
-        /// where the origin delivers, and MRF-retries, every delete to each
-        /// peer itself. Forwarding along a `site-repl-<deploymentId>` rule only
-        /// echoes the delete back to the origin, where it removes the origin's
-        /// pending-purge version before its own delivery state is written back,
-        /// and duplicates it to every other peer (rustfs#9121). MinIO never
-        /// re-replicates a replication request at all.
-        ///
-        /// Rule ids and target ARN shapes are not reserved: an ordinary bucket
-        /// target ARN also carries a target id, so an operator rule can match
-        /// the derived shape. Only a rule naming a current remote peer in
-        /// `site_peer_deployment_ids` is skipped; with site replication
-        /// disabled the set is empty and every rule forwards.
-        pub(crate) fn replica_delete_fanout_state_from_config(
-            config: &s3s::dto::ReplicationConfiguration,
-            obj_info: &crate::storage::storage_api::StorageObjectInfo,
-            version_id: Option<Uuid>,
-            site_peer_deployment_ids: &HashSet<String>,
-        ) -> Option<replication_contracts::ReplicationState> {
-            let is_site_peer_rule = |rule: &s3s::dto::ReplicationRule| {
-                replication_contracts::site_replication_rule_deployment_id(rule)
-                    .is_some_and(|deployment_id| site_peer_deployment_ids.contains(deployment_id))
-            };
-            if !config.rules.iter().any(is_site_peer_rule) {
-                return delete_replication_state_from_config(config, obj_info, version_id, true);
-            }
-
-            let mut operator_config = config.clone();
-            operator_config.rules.retain(|rule| !is_site_peer_rule(rule));
-            delete_replication_state_from_config(&operator_config, obj_info, version_id, true)
-        }
-
-        /// Whether `config` has a rule with the derived site-replication shape,
-        /// so the replica-delete fan-out needs the current peer set.
-        pub(crate) fn has_site_replication_shaped_rule(config: &s3s::dto::ReplicationConfiguration) -> bool {
-            config
-                .rules
-                .iter()
-                .any(|rule| replication_contracts::site_replication_rule_deployment_id(rule).is_some())
-        }
-
         pub(crate) fn replication_target_arns(config: &s3s::dto::ReplicationConfiguration) -> HashSet<String> {
             replication_contracts::replication_target_arns(config)
         }

rustfs/src/site_replication/state.rs

@@ -368,18 +368,6 @@ pub(crate) async fn site_replication_deployment_ids() -> S3Result<HashSet<String
     Ok(state.peers.values().map(|peer| peer.deployment_id.clone()).collect())
 }
 
-/// Deployment ids of the remote peers this site's reconciler derives a
-/// `site-repl-<id>` rule for; empty when site replication is not enabled.
-/// Read on the replica-delete path so only rules naming a current peer are
-/// treated as reconciler-owned.
-pub(crate) async fn site_replication_remote_peer_deployment_ids() -> S3Result<HashSet<String>> {
-    let state = load_site_replication_state().await?;
-    if !state.enabled() {
-        return Ok(HashSet::new());
-    }
-    Ok(remote_peer_deployment_ids(&state, &current_local_runtime_peer(&state)))
-}
-
 pub(crate) async fn load_site_replication_state_no_lock(store: Arc<ECStore>) -> S3Result<SiteReplicationState> {
     match read_config_no_lock(store, SITE_REPLICATION_STATE_PATH).await {
         Ok(data) => parse_site_replication_state(&data),