Expose replication backlog gauges (#5557)

* fix(replication): count backlog at queue admission Co-Authored-By: heihutu <heihutu@gmail.com> * feat(obs): expose bucket replication backlog gauges Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs): report recent backlog from queued work Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs): preserve legacy backlog metric semantics Co-Authored-By: heihutu <heihutu@gmail.com> * feat(obs): expose durable MRF backlog gauges Co-Authored-By: heihutu <heihutu@gmail.com> * test(obs): cover replication backlog metric scope Co-Authored-By: heihutu <heihutu@gmail.com> * fix(obs): keep backlog metrics API-compatible Co-Authored-By: heihutu <heihutu@gmail.com> * refactor(obs): streamline replication backlog metrics Keep MRF backlog accounting and OBS metric collection on a single, cheaper path. Co-Authored-By: heihutu <heihutu@gmail.com> * test(kms): update aws capability snapshot Co-Authored-By: heihutu <heihutu@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com>
2026-08-24 13:16:28 +00:00 · 2026-08-01 16:48:45 +08:00
parent 8218248000
commit 62d44d10b8
13 changed files with 1025 additions and 244 deletions
@@ -172,6 +172,9 @@ pub mod bucket {
    }

    pub mod replication {
+        pub use crate::bucket::replication::replication_pool::{
+            DurableMrfBacklogSummary, DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot,
+        };
        pub use crate::bucket::replication::{
            BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool,
            MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision,
@@ -46,7 +46,11 @@ use super::replication_target_boundary::{ReplicationTargetStore, replication_obj
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
 use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
+use std::collections::HashMap;
+use std::collections::hash_map::Entry;
 use std::sync::Arc;
+use std::sync::LazyLock;
+use std::sync::RwLock as StdRwLock;
 use std::sync::atomic::AtomicI32;
 use std::sync::atomic::Ordering;
 use time::OffsetDateTime;
@@ -74,6 +78,66 @@ pub struct DurableMrfBacklog {
    pub entries: Vec<MrfReplicateEntry>,
 }

+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct DurableMrfBucketBacklog {
+    pub bucket: String,
+    pub count: u64,
+    pub bytes: u64,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct DurableMrfBacklogSummary {
+    pub available: bool,
+    pub buckets: Vec<DurableMrfBucketBacklog>,
+}
+
+static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock<StdRwLock<DurableMrfBacklogSummary>> =
+    LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::default()));
+
+fn durable_mrf_backlog_summary_from_sizes<I>(entries: I) -> DurableMrfBacklogSummary
+where
+    I: IntoIterator<Item = (String, i64)>,
+{
+    let mut buckets = HashMap::<String, DurableMrfBucketBacklog>::new();
+    for (bucket_name, entry_size) in entries {
+        let Ok(size) = u64::try_from(entry_size) else {
+            return DurableMrfBacklogSummary::default();
+        };
+
+        let bucket = match buckets.entry(bucket_name) {
+            Entry::Occupied(entry) => entry.into_mut(),
+            Entry::Vacant(entry) => {
+                let bucket = entry.key().clone();
+                entry.insert(DurableMrfBucketBacklog {
+                    bucket,
+                    ..Default::default()
+                })
+            }
+        };
+        bucket.count = bucket.count.saturating_add(1);
+        bucket.bytes = bucket.bytes.saturating_add(size);
+    }
+
+    DurableMrfBacklogSummary {
+        available: true,
+        buckets: buckets.into_values().collect(),
+    }
+}
+
+fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) {
+    match DURABLE_MRF_BACKLOG_SUMMARY.write() {
+        Ok(mut guard) => *guard = summary,
+        Err(poisoned) => *poisoned.into_inner() = summary,
+    }
+}
+
+pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary {
+    match DURABLE_MRF_BACKLOG_SUMMARY.read() {
+        Ok(guard) => guard.clone(),
+        Err(poisoned) => poisoned.into_inner().clone(),
+    }
+}
+
 fn durable_mrf_backlog_from_read(result: Result<Vec<u8>, EcstoreError>) -> DurableMrfBacklog {
    match result {
        Ok(data) => match decode_mrf_file(&data) {
@@ -233,6 +297,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

            let active_counter = self.active_lrg_workers.clone();
            let storage = self.storage.clone();
+            let stats = self.stats.clone();

            let handle = tokio::spawn(async move {
                let mut rx = rx;
@@ -241,10 +306,18 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

                    match operation {
                        ReplicationOperation::Object(obj_info) => {
+                            let bucket = obj_info.bucket.clone();
+                            let size = obj_info.size;
+                            let delete_marker = obj_info.delete_marker;
+                            let op_type = obj_info.op_type;
                            replicate_object(*obj_info, storage.clone()).await;
+                            stats.dec_q(&bucket, size, delete_marker, op_type);
                        }
                        ReplicationOperation::Delete(del_info) => {
+                            let bucket = del_info.bucket.clone();
+                            let op_type = del_info.op_type;
                            replicate_delete(*del_info, storage.clone()).await;
+                            stats.dec_q(&bucket, 0, true, op_type);
                        }
                    }

@@ -310,23 +383,22 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

                    match operation {
                        ReplicationOperation::Object(obj_info) => {
-                            stats
-                                .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                                .await;
-
+                            let bucket = obj_info.bucket.clone();
+                            let size = obj_info.size;
+                            let delete_marker = obj_info.delete_marker;
+                            let op_type = obj_info.op_type;
                            // Perform actual replication (placeholder)
-                            replicate_object(obj_info.as_ref().clone(), storage.clone()).await;
+                            replicate_object(*obj_info, storage.clone()).await;

-                            stats
-                                .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                                .await;
+                            stats.dec_q(&bucket, size, delete_marker, op_type);
                        }
                        ReplicationOperation::Delete(del_info) => {
-                            stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await;
+                            let bucket = del_info.bucket.clone();
+                            let op_type = del_info.op_type;
                            // Perform actual delete replication (placeholder)
-                            replicate_delete(del_info.as_ref().clone(), storage.clone()).await;
+                            replicate_delete(*del_info, storage.clone()).await;

-                            stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await;
+                            stats.dec_q(&bucket, 0, true, op_type);
                        }
                    }

@@ -379,16 +451,18 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                    active_counter.fetch_add(1, Ordering::SeqCst);
                    match operation {
                        ReplicationOperation::Object(obj_info) => {
-                            stats
-                                .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                                .await;
-                            replicate_object(obj_info.as_ref().clone(), storage.clone()).await;
-                            stats
-                                .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                                .await;
+                            let bucket = obj_info.bucket.clone();
+                            let size = obj_info.size;
+                            let delete_marker = obj_info.delete_marker;
+                            let op_type = obj_info.op_type;
+                            replicate_object(*obj_info, storage.clone()).await;
+                            stats.dec_q(&bucket, size, delete_marker, op_type);
                        }
                        ReplicationOperation::Delete(del_info) => {
+                            let bucket = del_info.bucket.clone();
+                            let op_type = del_info.op_type;
                            replicate_delete(*del_info, storage.clone()).await;
+                            stats.dec_q(&bucket, 0, true, op_type);
                        }
                    }
                    active_counter.fetch_sub(1, Ordering::SeqCst);
@@ -529,9 +603,13 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
            if !lrg_workers.is_empty() {
                let index = (hash as usize) % lrg_workers.len();

-                if let Some(worker) = lrg_workers.get(index)
-                    && worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_err()
-                {
+                if let Some(worker) = lrg_workers.get(index) {
+                    self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
+                    if worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() {
+                        return ReplicationQueueAdmission::Queued;
+                    }
+                    self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
+
                    // Try to add more workers if possible
                    let max_l_workers = *self.max_l_workers.read().await;
                    let existing = lrg_workers.len();
@@ -539,17 +617,13 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                    drop(lrg_workers);

                    // Queue to MRF if worker is busy.
-                    let admission =
-                        queue_mrf_save_admission(&self.mrf_save_tx, ri.to_mrf_entry(), &ri.bucket, &ri.name, "large_object")
-                            .await;
+                    let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "large_object").await;

                    if let Some(resize) = resize {
                        self.resize_lrg_workers(resize.new_count, resize.existing_count).await;
                    }
                    return admission;
                }
-
-                return ReplicationQueueAdmission::Queued;
            }
            return ReplicationQueueAdmission::Missed;
        }
@@ -562,12 +636,14 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
            return ReplicationQueueAdmission::Missed;
        };

+        self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
        if channel.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() {
            return ReplicationQueueAdmission::Queued;
        }
+        self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);

        // Queue to MRF if all workers are busy.
-        let admission = queue_mrf_save_admission(&self.mrf_save_tx, ri.to_mrf_entry(), &ri.bucket, &ri.name, "object").await;
+        let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "object").await;

        // Try to scale up workers based on priority
        self.apply_queue_backpressure("object", true, "Replication queue is backpressured")
@@ -586,18 +662,13 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
            return ReplicationQueueAdmission::Missed;
        };

+        self.stats.inc_q(&doi.bucket, 0, true, doi.op_type);
        if channel.try_send(ReplicationOperation::Delete(Box::new(doi.clone()))).is_ok() {
            return ReplicationQueueAdmission::Queued;
        }
+        self.stats.dec_q(&doi.bucket, 0, true, doi.op_type);

-        let admission = queue_mrf_save_admission(
-            &self.mrf_save_tx,
-            doi.to_mrf_entry(),
-            &doi.bucket,
-            &doi.delete_object.object_name,
-            "delete",
-        )
-        .await;
+        let admission = self.queue_mrf_save_admission(doi.to_mrf_entry(), "delete").await;

        self.apply_queue_backpressure("delete", false, "Replication delete queue is backpressured")
            .await;
@@ -607,7 +678,18 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

    /// Queues an MRF save operation
    async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
-        let _ = queue_mrf_save_admission(&self.mrf_save_tx, entry, "", "", "mrf_worker").await;
+        let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
+    }
+
+    async fn queue_mrf_save_admission(&self, entry: MrfReplicateEntry, queue_type: &'static str) -> ReplicationQueueAdmission {
+        let bucket = entry.bucket.clone();
+        let size = entry.size;
+        let is_delete = matches!(entry.op, MrfOpKind::Delete);
+        let admission = queue_mrf_save_entry(&self.mrf_save_tx, entry, queue_type).await;
+        if admission == ReplicationQueueAdmission::Queued {
+            self.stats.inc_q(&bucket, size, is_delete, ReplicationType::Heal);
+        }
+        admission
    }

    /// Starts the MRF processor — one-shot at startup.
@@ -622,7 +704,13 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
        let handle = tokio::spawn(async move {
            let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
                Ok(d) => d,
-                Err(EcstoreError::ConfigNotFound) => return, // no file yet — normal on first start
+                Err(EcstoreError::ConfigNotFound) => {
+                    set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
+                        available: true,
+                        buckets: Vec::new(),
+                    });
+                    return;
+                }
                Err(e) => {
                    warn!(
                        component = LOG_COMPONENT_ECSTORE,
@@ -653,6 +741,9 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                    return;
                }
            };
+            set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes(
+                entries.iter().map(|entry| (entry.bucket.clone(), entry.size)),
+            ));

            let total = entries.len();
            let mut queued_count = 0usize;
@@ -765,6 +856,11 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                    error = %e,
                    "Failed to clear MRF recovery file after replay — entries may be replayed again on next restart"
                );
+            } else {
+                set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
+                    available: true,
+                    buckets: Vec::new(),
+                });
            }

            if queued_count > 0 {
@@ -793,6 +889,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
            return;
        };
        let storage = self.storage.clone();
+        let stats = self.stats.clone();

        let handle = tokio::spawn(async move {
            // The on-disk MRF file is a restart-recovery backstop: entries are
@@ -818,6 +915,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                    entry = rx.recv() => match entry {
                        Some(e) => {
                            if pending.len() >= MRF_PENDING_CAP {
+                                dec_mrf_entries(stats.as_ref(), std::slice::from_ref(&e));
                                if !capped {
                                    capped = true;
                                    warn!(
@@ -836,20 +934,31 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                            // set, not the absolute length, so a large backlog is
                            // not rewritten on every single add).
                            if pending.len() - flushed_len >= 1000 && flush_mrf_to_disk(&pending, &storage).await {
+                                set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes(
+                                    pending.iter().map(|entry| (entry.bucket.clone(), entry.size)),
+                                ));
+                                dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                                flushed_len = pending.len();
                                dirty = false;
                            }
                        }
                        None => {
                            // Channel closed (pool shutting down) — final flush.
-                            if dirty {
-                                flush_mrf_to_disk(&pending, &storage).await;
+                            if dirty && flush_mrf_to_disk(&pending, &storage).await {
+                                set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes(
+                                    pending.iter().map(|entry| (entry.bucket.clone(), entry.size)),
+                                ));
+                                dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                            }
                            break;
                        }
                    },
                    _ = interval.tick() => {
                        if dirty && flush_mrf_to_disk(&pending, &storage).await {
+                            set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes(
+                                pending.iter().map(|entry| (entry.bucket.clone(), entry.size)),
+                            ));
+                            dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                            flushed_len = pending.len();
                            dirty = false;
                        }
@@ -872,24 +981,22 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

            match operation {
                ReplicationOperation::Object(obj_info) => {
-                    stats
-                        .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                        .await;
-
+                    let bucket = obj_info.bucket.clone();
+                    let size = obj_info.size;
+                    let delete_marker = obj_info.delete_marker;
+                    let op_type = obj_info.op_type;
                    // Perform actual replication (placeholder)
-                    replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await;
+                    replicate_object(*obj_info, self.storage.clone()).await;

-                    stats
-                        .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                        .await;
+                    stats.dec_q(&bucket, size, delete_marker, op_type);
                }
                ReplicationOperation::Delete(del_info) => {
-                    stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await;
-
+                    let bucket = del_info.bucket.clone();
+                    let op_type = del_info.op_type;
                    // Perform actual delete replication (placeholder)
-                    replicate_delete(del_info.as_ref().clone(), self.storage.clone()).await;
+                    replicate_delete(*del_info, self.storage.clone()).await;

-                    stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await;
+                    stats.dec_q(&bucket, 0, true, op_type);
                }
            }

@@ -898,16 +1005,30 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
    }

    /// Worker function for handling large object replication operations
-    async fn add_large_worker(&self, mut rx: Receiver<ReplicationOperation>, active_counter: Arc<AtomicI32>, storage: Arc<S>) {
+    async fn add_large_worker(
+        &self,
+        mut rx: Receiver<ReplicationOperation>,
+        active_counter: Arc<AtomicI32>,
+        stats: Arc<ReplicationStats>,
+        storage: Arc<S>,
+    ) {
        while let Some(operation) = rx.recv().await {
            active_counter.fetch_add(1, Ordering::SeqCst);

            match operation {
                ReplicationOperation::Object(obj_info) => {
+                    let bucket = obj_info.bucket.clone();
+                    let size = obj_info.size;
+                    let delete_marker = obj_info.delete_marker;
+                    let op_type = obj_info.op_type;
                    replicate_object(*obj_info, storage.clone()).await;
+                    stats.dec_q(&bucket, size, delete_marker, op_type);
                }
                ReplicationOperation::Delete(del_info) => {
+                    let bucket = del_info.bucket.clone();
+                    let op_type = del_info.op_type;
                    replicate_delete(*del_info, storage.clone()).await;
+                    stats.dec_q(&bucket, 0, true, op_type);
                }
            }

@@ -927,18 +1048,19 @@ impl<S: ReplicationStorage> ReplicationPool<S> {

            match operation {
                ReplicationOperation::Object(obj_info) => {
-                    stats
-                        .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                        .await;
-
+                    let bucket = obj_info.bucket.clone();
+                    let size = obj_info.size;
+                    let delete_marker = obj_info.delete_marker;
+                    let op_type = obj_info.op_type;
                    replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await;

-                    stats
-                        .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type)
-                        .await;
+                    stats.dec_q(&bucket, size, delete_marker, op_type);
                }
                ReplicationOperation::Delete(del_info) => {
+                    let bucket = del_info.bucket.clone();
+                    let op_type = del_info.op_type;
                    replicate_delete(*del_info, self.storage.clone()).await;
+                    stats.dec_q(&bucket, 0, true, op_type);
                }
            }

@@ -1273,29 +1395,34 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
    }
 }

-async fn queue_mrf_save_admission(
+async fn queue_mrf_save_entry(
    tx: &Sender<MrfReplicateEntry>,
    entry: MrfReplicateEntry,
-    bucket: &str,
-    object: &str,
    queue_type: &'static str,
 ) -> ReplicationQueueAdmission {
-    if tx.send(entry).await.is_ok() {
+    let Err(error) = tx.send(entry).await else {
        return ReplicationQueueAdmission::Queued;
-    }
+    };
+    let entry = error.0;

    warn!(
        event = EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE,
        component = LOG_COMPONENT_ECSTORE,
        subsystem = LOG_SUBSYSTEM_REPLICATION,
-        bucket = %bucket,
-        object = %object,
+        bucket = %entry.bucket,
+        object = %entry.object,
        queue_type = queue_type,
        "MRF save channel unavailable — replication failure entry could not be persisted for retry"
    );
    ReplicationQueueAdmission::Missed
 }

+fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) {
+    for entry in entries {
+        stats.dec_q(&entry.bucket, entry.size, matches!(entry.op, MrfOpKind::Delete), ReplicationType::Heal);
+    }
+}
+
 /// Encodes `entries` and overwrites the MRF persistence file.
 /// Returns `true` on success; on failure logs the error and returns `false`.
 /// Callers must NOT clear their in-memory buffer on `false` so the next tick
@@ -2016,6 +2143,147 @@ mod tests {
        })
    }

+    async fn current_queue(pool: &ReplicationPool<LoadResyncNodeStore>, bucket: &str) -> (i64, i64) {
+        let stats = pool.stats.get_latest_replication_stats(bucket).await;
+        (stats.replication_stats.q_stat.curr.count, stats.replication_stats.q_stat.curr.bytes)
+    }
+
+    async fn wait_for_current_queue(pool: &ReplicationPool<LoadResyncNodeStore>, bucket: &str, expected: (i64, i64)) {
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if current_queue(pool, bucket).await == expected {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("replication queue should reach the expected state");
+    }
+
+    #[tokio::test]
+    async fn regular_worker_admission_counts_channel_backlog_before_receive() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        let (tx, _rx) = mpsc::channel(1);
+        pool.workers.write().await.push(tx);
+
+        let admission = pool
+            .queue_replica_task(ReplicateObjectInfo {
+                bucket: "admission-bucket".to_string(),
+                name: "object".to_string(),
+                size: 4096,
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        assert_eq!(current_queue(&pool, "admission-bucket").await, (1, 4096));
+    }
+
+    #[tokio::test]
+    async fn large_worker_admission_counts_channel_backlog_before_receive() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        let (tx, _rx) = mpsc::channel(1);
+        pool.lrg_workers.write().await.push(tx);
+        let size = 128 * 1024 * 1024;
+
+        let admission = pool
+            .queue_replica_task(ReplicateObjectInfo {
+                bucket: "large-admission-bucket".to_string(),
+                name: "large-object".to_string(),
+                size,
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        assert_eq!(current_queue(&pool, "large-admission-bucket").await, (1, size));
+    }
+
+    #[tokio::test]
+    async fn delete_admission_counts_channel_backlog_before_receive() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        let (tx, _rx) = mpsc::channel(1);
+        pool.workers.write().await.push(tx);
+
+        let admission = pool
+            .queue_replica_delete_task(DeletedObjectReplicationInfo {
+                bucket: "delete-admission-bucket".to_string(),
+                delete_object: ReplicationDeletedObject {
+                    object_name: "deleted-object".to_string(),
+                    ..Default::default()
+                },
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        assert_eq!(current_queue(&pool, "delete-admission-bucket").await, (1, 0));
+    }
+
+    #[tokio::test]
+    async fn regular_worker_drains_current_backlog_after_processing() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        pool.resize_workers(1, 0).await;
+
+        let admission = pool
+            .queue_replica_task(ReplicateObjectInfo {
+                bucket: "regular-drain-bucket".to_string(),
+                name: "object".to_string(),
+                size: 4096,
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        wait_for_current_queue(&pool, "regular-drain-bucket", (0, 0)).await;
+    }
+
+    #[tokio::test]
+    async fn large_worker_drains_current_backlog_after_processing() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        pool.resize_lrg_workers(1, 0).await;
+        let size = 128 * 1024 * 1024;
+
+        let admission = pool
+            .queue_replica_task(ReplicateObjectInfo {
+                bucket: "large-drain-bucket".to_string(),
+                name: "large-object".to_string(),
+                size,
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        wait_for_current_queue(&pool, "large-drain-bucket", (0, 0)).await;
+    }
+
+    #[tokio::test]
+    async fn regular_delete_worker_drains_current_backlog_after_processing() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        pool.resize_workers(1, 0).await;
+
+        let admission = pool
+            .queue_replica_delete_task(DeletedObjectReplicationInfo {
+                bucket: "delete-drain-bucket".to_string(),
+                delete_object: ReplicationDeletedObject {
+                    object_name: "deleted-object".to_string(),
+                    ..Default::default()
+                },
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        wait_for_current_queue(&pool, "delete-drain-bucket", (0, 0)).await;
+    }
+
    fn load_resync_test_metadata() -> Vec<u8> {
        let mut status = BucketReplicationResyncStatus::new();
        status.targets_map.insert(
@@ -2301,6 +2569,37 @@ mod tests {
        assert_eq!(admission, ReplicationQueueAdmission::Missed);
    }

+    #[tokio::test]
+    async fn queue_replica_task_counts_mrf_pending_backlog_when_worker_queue_is_full() {
+        let shared = empty_resync_shared_state();
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared))).await;
+        let (tx, _rx) = mpsc::channel(1);
+        tx.try_send(ReplicationOperation::Object(Box::new(ReplicateObjectInfo {
+            bucket: "runtime-backlog".to_string(),
+            name: "already-buffered".to_string(),
+            size: 1,
+            op_type: ReplicationType::Object,
+            ..Default::default()
+        })))
+        .expect("test setup should fill the worker queue");
+        pool.workers.write().await.push(tx);
+
+        let admission = pool
+            .queue_replica_task(ReplicateObjectInfo {
+                bucket: "runtime-backlog".to_string(),
+                name: "fallback-object".to_string(),
+                size: 2048,
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            })
+            .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Queued);
+        let queued = pool.stats.get_latest_replication_stats("runtime-backlog").await;
+        assert_eq!(queued.replication_stats.q_stat.curr.count, 1);
+        assert_eq!(queued.replication_stats.q_stat.curr.bytes, 2048);
+    }
+
    #[test]
    fn replicate_object_info_from_object_info_preserves_ssec_checksum() {
        let checksum = bytes::Bytes::from_static(b"ssec-checksum");
@@ -2342,7 +2641,7 @@ mod tests {

        tx.try_send(first).expect("first MRF entry should fill the test channel");

-        let admission = queue_mrf_save_admission(&tx, second, "bucket", "second", "test");
+        let admission = queue_mrf_save_entry(&tx, second, "test");
        tokio::pin!(admission);

        assert!(
@@ -2687,6 +2986,30 @@ mod tests {
        assert!(missing_file.entries.is_empty());
    }

+    #[test]
+    fn durable_mrf_summary_aggregates_entries_by_bucket_for_obs() {
+        let summary =
+            durable_mrf_backlog_summary_from_sizes([("b1".to_string(), 1024), ("b1".to_string(), 512), ("b2".to_string(), 0)]);
+
+        assert!(summary.available);
+        let buckets = summary
+            .buckets
+            .into_iter()
+            .map(|bucket| (bucket.bucket.clone(), bucket))
+            .collect::<HashMap<_, _>>();
+        assert_eq!(buckets["b1"].count, 2);
+        assert_eq!(buckets["b1"].bytes, 1536);
+        assert_eq!(buckets["b2"].count, 1);
+        assert_eq!(buckets["b2"].bytes, 0);
+    }
+
+    #[test]
+    fn durable_mrf_summary_marks_invalid_sizes_unavailable() {
+        let invalid = durable_mrf_backlog_summary_from_sizes([("bucket".to_string(), -1)]);
+        assert!(!invalid.available);
+        assert!(invalid.buckets.is_empty());
+    }
+
    #[test]
    fn durable_mrf_snapshot_marks_corrupt_or_invalid_data_unavailable() {
        let corrupt = durable_mrf_backlog_from_read(Ok(vec![0, 1, 2]));
@@ -23,8 +23,8 @@ use super::replication_stats_boundary::{
 };
 use super::runtime_boundary as runtime_sources;
 use std::collections::HashMap;
-use std::sync::Arc;
 use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::{Arc, Mutex as StdMutex};
 use std::time::{Duration, SystemTime};
 use tokio::sync::{Mutex, RwLock};
 use tokio::time::interval;
@@ -143,7 +143,7 @@ pub struct ReplicationStats {
    // Active worker statistics
    pub workers: Arc<Mutex<ActiveWorkerStat>>,
    // Queue statistics cache
-    pub q_cache: Arc<Mutex<QueueCache>>,
+    pub q_cache: Arc<StdMutex<QueueCache>>,
    // Proxy statistics cache
    pub p_cache: Arc<Mutex<ProxyStatsCache>>,
    // MRF backlog statistics (simplified)
@@ -158,7 +158,7 @@ impl ReplicationStats {
        Self {
            sr_stats: Arc::new(SRStats::new()),
            workers: Arc::new(Mutex::new(ActiveWorkerStat::new())),
-            q_cache: Arc::new(Mutex::new(QueueCache::new())),
+            q_cache: Arc::new(StdMutex::new(QueueCache::new())),
            p_cache: Arc::new(Mutex::new(ProxyStatsCache::new())),
            mrf_stats: HashMap::new(),
            cache: Arc::new(RwLock::new(HashMap::new())),
@@ -198,8 +198,9 @@ impl ReplicationStats {
            let mut interval = interval(Duration::from_secs(2));
            loop {
                interval.tick().await;
-                let mut cache = q_cache_clone.lock().await;
-                cache.update();
+                if let Ok(mut cache) = q_cache_clone.lock() {
+                    cache.update();
+                }
            }
        });
    }
@@ -391,12 +392,13 @@ impl ReplicationStats {
        drop(cache);

        {
-            let q_cache = self.q_cache.lock().await;
-            for (bucket, queue_stats) in &q_cache.bucket_stats {
-                let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new);
-                bucket_stats.q_stat = queue_stats.snapshot();
-                bucket_stats.mark_node_local_provider_available();
-                bucket_stats.queue_scope = ReplicationMetricScope::NodeLocal;
+            if let Ok(q_cache) = self.q_cache.lock() {
+                for (bucket, queue_stats) in &q_cache.bucket_stats {
+                    let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new);
+                    bucket_stats.q_stat = queue_stats.snapshot();
+                    bucket_stats.mark_node_local_provider_available();
+                    bucket_stats.queue_scope = ReplicationMetricScope::NodeLocal;
+                }
            }
        }

@@ -429,8 +431,11 @@ impl ReplicationStats {
        let boot_time = SystemTime::UNIX_EPOCH; // simplified implementation
        let uptime = SystemTime::now().duration_since(boot_time).unwrap_or_default().as_secs() as i64;

-        let q_cache = self.q_cache.lock().await;
-        let queued = q_cache.get_site_stats();
+        let queued = self
+            .q_cache
+            .lock()
+            .map(|q_cache| q_cache.get_site_stats())
+            .unwrap_or_default();

        let p_cache = self.p_cache.lock().await;
        let proxied = p_cache.get_site_stats();
@@ -633,8 +638,9 @@ impl ReplicationStats {
        drop(cache);

        {
-            let q_cache = self.q_cache.lock().await;
-            if let Some(queue_stats) = q_cache.bucket_stats.get(bucket) {
+            if let Ok(q_cache) = self.q_cache.lock()
+                && let Some(queue_stats) = q_cache.bucket_stats.get(bucket)
+            {
                replication_stats.q_stat = queue_stats.snapshot();
            }
        }
@@ -665,31 +671,17 @@ impl ReplicationStats {
    }

    /// Increase queue statistics
-    pub async fn inc_q(&self, bucket: &str, size: i64, _is_delete_repl: bool, _op_type: ReplicationType) {
-        let mut q_cache = self.q_cache.lock().await;
-        let stats = q_cache
-            .bucket_stats
-            .entry(bucket.to_string())
-            .or_insert_with(InQueueMetric::default);
-        stats.curr.now_bytes.fetch_add(size, Ordering::Relaxed);
-        stats.curr.now_count.fetch_add(1, Ordering::Relaxed);
-
-        q_cache.sr_queue_stats.curr.now_bytes.fetch_add(size, Ordering::Relaxed);
-        q_cache.sr_queue_stats.curr.now_count.fetch_add(1, Ordering::Relaxed);
+    pub fn inc_q(&self, bucket: &str, size: i64, _is_delete_repl: bool, _op_type: ReplicationType) {
+        if let Ok(mut q_cache) = self.q_cache.lock() {
+            q_cache.inc(bucket, size);
+        }
    }

    /// Decrease queue statistics
-    pub async fn dec_q(&self, bucket: &str, size: i64, _is_del_marker: bool, _op_type: ReplicationType) {
-        let mut q_cache = self.q_cache.lock().await;
-        let stats = q_cache
-            .bucket_stats
-            .entry(bucket.to_string())
-            .or_insert_with(InQueueMetric::default);
-        stats.curr.now_bytes.fetch_sub(size, Ordering::Relaxed);
-        stats.curr.now_count.fetch_sub(1, Ordering::Relaxed);
-
-        q_cache.sr_queue_stats.curr.now_bytes.fetch_sub(size, Ordering::Relaxed);
-        q_cache.sr_queue_stats.curr.now_count.fetch_sub(1, Ordering::Relaxed);
+    pub fn dec_q(&self, bucket: &str, size: i64, _is_del_marker: bool, _op_type: ReplicationType) {
+        if let Ok(mut q_cache) = self.q_cache.lock() {
+            q_cache.dec(bucket, size);
+        }
    }

    /// Increase proxy metrics
@@ -801,14 +793,14 @@ mod tests {
    async fn latest_stats_include_queue_until_drained() {
        let stats = ReplicationStats::new();

-        stats.inc_q("queued-bucket", 4096, false, ReplicationType::Object).await;
+        stats.inc_q("queued-bucket", 4096, false, ReplicationType::Object);
        let queued = stats.get_latest_replication_stats("queued-bucket").await;
        assert!(queued.replication_stats.provider_available);
        assert_eq!(queued.replication_stats.q_stat.curr.count, 1);
        assert_eq!(queued.replication_stats.q_stat.curr.bytes, 4096);
        assert_eq!(queued.replication_stats.queue_scope, ReplicationMetricScope::NodeLocal);

-        stats.dec_q("queued-bucket", 4096, false, ReplicationType::Object).await;
+        stats.dec_q("queued-bucket", 4096, false, ReplicationType::Object);
        let drained = stats.get_latest_replication_stats("queued-bucket").await;
        assert_eq!(drained.replication_stats.q_stat.curr.count, 0);
        assert_eq!(drained.replication_stats.q_stat.curr.bytes, 0);
@@ -911,7 +903,7 @@ mod tests {
        for _ in 0..32 {
            let stats = Arc::clone(&stats);
            tasks.push(tokio::spawn(async move {
-                stats.inc_q("concurrent-bucket", 7, false, ReplicationType::Object).await;
+                stats.inc_q("concurrent-bucket", 7, false, ReplicationType::Object);
            }));
        }
        for task in tasks {