test(replication): cover backlog metrics outage e2e

Add a focused end-to-end test that exports bucket replication backlog gauges through OTLP while a slow target keeps worker backlog observable, then verifies recovery drains the current and MRF pending gauges.

Co-Authored-By: heihutu <heihutu@gmail.com>

crates/e2e_test/src/replication_extension_test.rs

@@ -29,8 +29,9 @@ use aws_sdk_s3::types::{
 use aws_sdk_s3::{Client, Config};
 use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
 use bytes::Bytes;
+use flate2::read::GzDecoder;
 use futures::{Stream, StreamExt};
-use http::header::{CONTENT_TYPE, HOST};
+use http::header::{CONTENT_ENCODING, CONTENT_TYPE, HOST};
 use http_body_util::{BodyExt, Full};
 use hyper::body::Incoming;
 use hyper::server::conn::http1;
@@ -38,6 +39,10 @@ use hyper::service::service_fn;
 use hyper::{Request, Response};
 use hyper_util::rt::TokioIo;
 use local_ip_address::local_ip;
+use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
+use opentelemetry_proto::tonic::common::v1::{KeyValue, any_value::Value as AnyValue};
+use opentelemetry_proto::tonic::metrics::v1::{Metric, metric, number_data_point};
+use prost::Message;
 use rcgen::{
     BasicConstraints, CertificateParams, CertifiedIssuer, DnType, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose,
     SanType, generate_simple_self_signed,
@@ -56,6 +61,7 @@ use sha2::{Digest, Sha256};
 use std::collections::BTreeMap;
 use std::convert::Infallible;
 use std::error::Error;
+use std::io::Read;
 use std::net::IpAddr;
 use std::path::Path;
 use std::process::Command;
@@ -64,11 +70,13 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use time::{Duration as TimeDuration, OffsetDateTime};
 use tokio::fs;
 use tokio::net::TcpListener;
-use tokio::sync::watch;
+use tokio::sync::{Mutex, watch};
+use tokio::task::JoinHandle;
 use tokio::task::JoinSet;
 use tokio::time::{Duration, sleep, timeout};
 
 type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
+type BacklogMetricPoints = Arc<Mutex<BTreeMap<String, BTreeMap<String, (u64, f64)>>>>;
 
 /// A replication source server validates the remote target endpoint, and the e2e
 /// target runs on loopback (127.0.0.1), which RustFS's SSRF egress guard rejects by
@@ -107,6 +115,250 @@ const REPL17_KMS_KEY_ID: &str = "repl17-local-key";
 const REPL17_SSEC_KEY: &str = "01234567890123456789012345678901";
 const REPLICATION_FAILED_EVENT: &str = "s3:Replication:OperationFailedReplication";
 const REPLICATION_EVENT_MAX_BUFFER_BYTES: usize = 1024 * 1024;
+const OTLP_METRICS_BODY_LIMIT: u64 = 4 * 1024 * 1024;
+const BUCKET_LABEL: &str = "bucket";
+const CURRENT_BACKLOG_COUNT_METRIC: &str = "rustfs_bucket_replication_current_backlog_count";
+const CURRENT_BACKLOG_BYTES_METRIC: &str = "rustfs_bucket_replication_current_backlog_bytes";
+const MRF_PENDING_COUNT_METRIC: &str = "rustfs_bucket_replication_mrf_pending_count";
+const MRF_PENDING_BYTES_METRIC: &str = "rustfs_bucket_replication_mrf_pending_bytes";
+
+struct ReplicationBacklogMetricCollector {
+    endpoint: String,
+    values: BacklogMetricPoints,
+    task: JoinHandle<()>,
+}
+
+impl ReplicationBacklogMetricCollector {
+    async fn start() -> Result<Self, Box<dyn Error + Send + Sync>> {
+        let listener = TcpListener::bind("127.0.0.1:0").await?;
+        let endpoint = format!("http://{}/v1/metrics", listener.local_addr()?);
+        let values = Arc::new(Mutex::new(BTreeMap::new()));
+        let task_values = values.clone();
+        let task = tokio::spawn(async move {
+            loop {
+                let Ok((stream, _)) = listener.accept().await else {
+                    break;
+                };
+                let values = task_values.clone();
+                tokio::spawn(async move {
+                    let _ = http1::Builder::new()
+                        .serve_connection(
+                            TokioIo::new(stream),
+                            service_fn(move |request| handle_backlog_metric_export(request, values.clone())),
+                        )
+                        .await;
+                });
+            }
+        });
+
+        Ok(Self { endpoint, values, task })
+    }
+
+    fn root_endpoint(&self) -> &str {
+        self.endpoint.trim_end_matches("/v1/metrics")
+    }
+
+    async fn bucket_metric_value(&self, metric: &str, bucket: &str) -> f64 {
+        self.values
+            .lock()
+            .await
+            .get(metric)
+            .and_then(|buckets| buckets.get(bucket))
+            .map(|(_, value)| *value)
+            .unwrap_or_default()
+    }
+
+    async fn wait_for_bucket_metric(
+        &self,
+        metric: &str,
+        bucket: &str,
+        expected: impl Fn(f64) -> bool,
+        description: &str,
+    ) -> Result<f64, Box<dyn Error + Send + Sync>> {
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
+        loop {
+            let value = self.bucket_metric_value(metric, bucket).await;
+            if expected(value) {
+                return Ok(value);
+            }
+            if tokio::time::Instant::now() >= deadline {
+                let snapshot = self.values.lock().await.clone();
+                return Err(format!("timed out waiting for {metric} on bucket {bucket} to satisfy {description}; last={value}, snapshot={snapshot:?}").into());
+            }
+            sleep(Duration::from_millis(200)).await;
+        }
+    }
+}
+
+impl Drop for ReplicationBacklogMetricCollector {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+async fn handle_backlog_metric_export(
+    request: Request<Incoming>,
+    values: BacklogMetricPoints,
+) -> Result<Response<Full<Bytes>>, Infallible> {
+    if request.uri().path() != "/v1/metrics" {
+        return Ok(empty_http_response(StatusCode::NOT_FOUND));
+    }
+
+    let gzip = request
+        .headers()
+        .get(CONTENT_ENCODING)
+        .and_then(|value| value.to_str().ok())
+        .is_some_and(|value| value.eq_ignore_ascii_case("gzip"));
+    let Ok(collected) = request.into_body().collect().await else {
+        return Ok(empty_http_response(StatusCode::BAD_REQUEST));
+    };
+    let body = collected.to_bytes();
+    if body.len() as u64 > OTLP_METRICS_BODY_LIMIT {
+        return Ok(empty_http_response(StatusCode::PAYLOAD_TOO_LARGE));
+    }
+    let payload = if gzip {
+        let mut decoder = GzDecoder::new(body.as_ref());
+        let mut decoded = Vec::new();
+        if decoder
+            .by_ref()
+            .take(OTLP_METRICS_BODY_LIMIT + 1)
+            .read_to_end(&mut decoded)
+            .is_err()
+            || decoded.len() as u64 > OTLP_METRICS_BODY_LIMIT
+        {
+            return Ok(empty_http_response(StatusCode::BAD_REQUEST));
+        }
+        decoded
+    } else {
+        body.to_vec()
+    };
+
+    match ExportMetricsServiceRequest::decode(payload.as_slice()) {
+        Ok(export) => {
+            let mut values = values.lock().await;
+            record_backlog_metrics(&export, &mut values);
+            Ok(empty_http_response(StatusCode::OK))
+        }
+        Err(_) => Ok(empty_http_response(StatusCode::BAD_REQUEST)),
+    }
+}
+
+fn empty_http_response(status: StatusCode) -> Response<Full<Bytes>> {
+    Response::builder()
+        .status(status)
+        .body(Full::new(Bytes::new()))
+        .expect("static HTTP response is valid")
+}
+
+fn record_backlog_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap<String, BTreeMap<String, (u64, f64)>>) {
+    for resource_metrics in &export.resource_metrics {
+        for scope_metrics in &resource_metrics.scope_metrics {
+            for metric in &scope_metrics.metrics {
+                record_backlog_metric(metric, values);
+            }
+        }
+    }
+}
+
+fn record_backlog_metric(metric: &Metric, values: &mut BTreeMap<String, BTreeMap<String, (u64, f64)>>) {
+    if ![
+        CURRENT_BACKLOG_COUNT_METRIC,
+        CURRENT_BACKLOG_BYTES_METRIC,
+        MRF_PENDING_COUNT_METRIC,
+        MRF_PENDING_BYTES_METRIC,
+    ]
+    .contains(&metric.name.as_str())
+    {
+        return;
+    }
+
+    let points = match &metric.data {
+        Some(metric::Data::Gauge(gauge)) => gauge.data_points.as_slice(),
+        Some(metric::Data::Sum(sum)) => sum.data_points.as_slice(),
+        _ => return,
+    };
+    for point in points {
+        let Some(bucket) = attribute_string(&point.attributes, BUCKET_LABEL) else {
+            continue;
+        };
+        let Some(value) = number_point_value(point.value.as_ref()) else {
+            continue;
+        };
+        values
+            .entry(metric.name.clone())
+            .or_default()
+            .entry(bucket.to_string())
+            .and_modify(|current| {
+                if point.time_unix_nano >= current.0 {
+                    *current = (point.time_unix_nano, value);
+                }
+            })
+            .or_insert((point.time_unix_nano, value));
+    }
+}
+
+fn number_point_value(value: Option<&number_data_point::Value>) -> Option<f64> {
+    match value? {
+        number_data_point::Value::AsDouble(value) => Some(*value),
+        number_data_point::Value::AsInt(value) => Some(*value as f64),
+    }
+}
+
+fn attribute_string<'a>(attributes: &'a [KeyValue], wanted_key: &str) -> Option<&'a str> {
+    attributes.iter().find_map(|attribute| {
+        if attribute.key != wanted_key {
+            return None;
+        }
+        match attribute.value.as_ref()?.value.as_ref()? {
+            AnyValue::StringValue(value) => Some(value.as_str()),
+            _ => None,
+        }
+    })
+}
+
+struct SlowReplicationTargetGuard {
+    task: Option<JoinHandle<()>>,
+}
+
+impl SlowReplicationTargetGuard {
+    async fn bind(address: &str, response_delay: Duration) -> Result<Self, Box<dyn Error + Send + Sync>> {
+        let listener = TcpListener::bind(address).await?;
+        let task = tokio::spawn(async move {
+            loop {
+                let Ok((stream, _)) = listener.accept().await else {
+                    break;
+                };
+                tokio::spawn(async move {
+                    let _ = http1::Builder::new()
+                        .serve_connection(
+                            TokioIo::new(stream),
+                            service_fn(move |_request| async move {
+                                sleep(response_delay).await;
+                                Ok::<_, Infallible>(empty_http_response(StatusCode::SERVICE_UNAVAILABLE))
+                            }),
+                        )
+                        .await;
+                });
+            }
+        });
+        Ok(Self { task: Some(task) })
+    }
+
+    async fn stop(mut self) {
+        if let Some(task) = self.task.take() {
+            task.abort();
+            let _ = task.await;
+        }
+    }
+}
+
+impl Drop for SlowReplicationTargetGuard {
+    fn drop(&mut self) {
+        if let Some(task) = self.task.take() {
+            task.abort();
+        }
+    }
+}
 
 #[derive(Debug, Clone, serde::Deserialize)]
 struct ReplicationResetStatusResponse {
@@ -3659,6 +3911,125 @@ async fn test_bucket_replication_recovers_after_target_outage() -> TestResult {
     Ok(())
 }
 
+/// backlog#1610 - black-box bucket replication backlog observability.
+///
+/// The source exports metrics through the same OTLP path production uses. A slow
+/// loopback target keeps replication workers occupied long enough for the metrics
+/// runtime to publish non-zero bucket backlog gauges; after the real target
+/// returns, replication must converge and the exported current/MRF pending gauges
+/// must settle back to zero.
+#[tokio::test]
+#[serial]
+async fn test_bucket_replication_backlog_metrics_observe_outage_and_recovery() -> TestResult {
+    init_logging();
+
+    let collector = ReplicationBacklogMetricCollector::start().await?;
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    let metric_root = collector.root_endpoint().to_string();
+    let metric_endpoint = collector.endpoint.clone();
+    let mut source_env_vars: Vec<(&str, &str)> = replication_fast_env().into_iter().collect();
+    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
+    source_env_vars.extend_from_slice(FAST_SCANNER_ENV);
+    source_env_vars.extend_from_slice(&[
+        ("RUSTFS_OBS_ENDPOINT", metric_root.as_str()),
+        ("RUSTFS_OBS_METRIC_ENDPOINT", metric_endpoint.as_str()),
+        ("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "true"),
+        ("RUSTFS_OBS_TRACES_EXPORT_ENABLED", "false"),
+        ("RUSTFS_OBS_LOGS_EXPORT_ENABLED", "false"),
+        ("RUSTFS_OBS_METER_INTERVAL", "1"),
+        ("RUSTFS_OBS_USE_STDOUT", "false"),
+        ("RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL_SEC", "1"),
+    ]);
+    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "repl-backlog-metrics-src";
+    let target_bucket = "repl-backlog-metrics-dst";
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key("before-outage.txt")
+        .body(ByteStream::from_static(b"baseline written before backlog metrics outage"))
+        .send()
+        .await?;
+    assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?;
+
+    target_env.stop_server();
+    let slow_target = SlowReplicationTargetGuard::bind(&target_env.address, Duration::from_secs(5)).await?;
+
+    let outage_keys = ["metrics-outage-1.txt", "metrics-outage-2.txt", "metrics-outage-3.txt"];
+    for key in outage_keys {
+        source_client
+            .put_object()
+            .bucket(source_bucket)
+            .key(key)
+            .body(ByteStream::from(
+                format!("written while backlog metrics target was slow: {key}").into_bytes(),
+            ))
+            .send()
+            .await?;
+    }
+
+    let current_backlog = collector
+        .wait_for_bucket_metric(
+            CURRENT_BACKLOG_COUNT_METRIC,
+            source_bucket,
+            |value| value >= 1.0,
+            "be at least 1 during outage",
+        )
+        .await?;
+    let current_bytes = collector
+        .wait_for_bucket_metric(
+            CURRENT_BACKLOG_BYTES_METRIC,
+            source_bucket,
+            |value| value > 0.0,
+            "report bytes during outage",
+        )
+        .await?;
+    assert!(
+        current_bytes >= current_backlog,
+        "backlog bytes should be at least the object count while queued; count={current_backlog}, bytes={current_bytes}"
+    );
+
+    slow_target.stop().await;
+    target_env.restart_server_preserving_data(vec![], &[]).await?;
+
+    assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?;
+    for metric in [
+        CURRENT_BACKLOG_COUNT_METRIC,
+        CURRENT_BACKLOG_BYTES_METRIC,
+        MRF_PENDING_COUNT_METRIC,
+        MRF_PENDING_BYTES_METRIC,
+    ] {
+        collector
+            .wait_for_bucket_metric(metric, source_bucket, |value| value == 0.0, "settle back to zero after recovery")
+            .await?;
+    }
+
+    let target_state = list_replication_state(&target_client, target_bucket).await?;
+    for key in ["before-outage.txt"].into_iter().chain(outage_keys) {
+        assert!(
+            target_state.iter().any(|entry| entry.key == key && !entry.delete_marker),
+            "target missing object {key} after backlog metrics recovery; state={target_state:?}"
+        );
+    }
+
+    Ok(())
+}
+
 /// backlog#1147 repl-5, scenario (b) — failure state survives a source restart
 /// (mirrors backlog#858 delete-decision re-derivation and #859 no-drop).
 ///

crates/ecstore/src/api/mod.rs

@@ -173,7 +173,8 @@ pub mod bucket {
 
     pub mod replication {
         pub use crate::bucket::replication::replication_pool::{
-            DurableMrfBacklogSummary, DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot,
+            DurableMrfBacklogSummary, DurableMrfBucketBacklog, MrfBacklogObservabilitySummary, MrfBucketBacklogObservability,
+            durable_mrf_backlog_summary_snapshot, mrf_backlog_observability_snapshot,
         };
         pub use crate::bucket::replication::{
             BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool,

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

@@ -53,6 +53,7 @@ use std::sync::LazyLock;
 use std::sync::RwLock as StdRwLock;
 use std::sync::atomic::AtomicI32;
 use std::sync::atomic::Ordering;
+use std::time::Instant;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 use tokio::sync::Mutex;
@@ -91,20 +92,46 @@ pub struct DurableMrfBacklogSummary {
     pub buckets: Vec<DurableMrfBucketBacklog>,
 }
 
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct MrfBucketBacklogObservability {
+    pub bucket: String,
+    pub pending_count: u64,
+    pub pending_bytes: u64,
+    pub dropped_count: u64,
+    pub missed_count: u64,
+    pub flush_failure_count: u64,
+    pub last_flush_duration_millis: u64,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct MrfBacklogObservabilitySummary {
+    pub buckets: Vec<MrfBucketBacklogObservability>,
+}
+
 static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock<StdRwLock<DurableMrfBacklogSummary>> =
     LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::default()));
+static MRF_BACKLOG_OBSERVABILITY: LazyLock<StdRwLock<MrfBacklogObservabilityTracker>> =
+    LazyLock::new(|| StdRwLock::new(MrfBacklogObservabilityTracker::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 {
+#[derive(Debug, Clone, Default)]
+struct DurableMrfBacklogTracker {
+    available: bool,
+    buckets: HashMap<String, DurableMrfBucketBacklog>,
+}
+
+impl DurableMrfBacklogTracker {
+    fn add_entry(&mut self, bucket_name: String, entry_size: i64) {
         let Ok(size) = u64::try_from(entry_size) else {
-            return DurableMrfBacklogSummary::default();
+            self.available = false;
+            self.buckets.clear();
+            return;
         };
 
-        let bucket = match buckets.entry(bucket_name) {
+        if !self.available {
+            return;
+        }
+
+        let bucket = match self.buckets.entry(bucket_name) {
             Entry::Occupied(entry) => entry.into_mut(),
             Entry::Vacant(entry) => {
                 let bucket = entry.key().clone();
@@ -118,12 +145,93 @@ where
         bucket.bytes = bucket.bytes.saturating_add(size);
     }
 
-    DurableMrfBacklogSummary {
-        available: true,
-        buckets: buckets.into_values().collect(),
+    fn into_summary(self) -> DurableMrfBacklogSummary {
+        if !self.available {
+            return DurableMrfBacklogSummary::default();
+        }
+
+        DurableMrfBacklogSummary {
+            available: true,
+            buckets: self.buckets.into_values().collect(),
+        }
     }
 }
 
+#[derive(Debug, Clone, Default)]
+struct MrfBacklogObservabilityTracker {
+    buckets: HashMap<String, MrfBucketBacklogObservability>,
+}
+
+impl MrfBacklogObservabilityTracker {
+    fn bucket_mut(&mut self, bucket_name: &str) -> &mut MrfBucketBacklogObservability {
+        match self.buckets.entry(bucket_name.to_string()) {
+            Entry::Occupied(entry) => entry.into_mut(),
+            Entry::Vacant(entry) => entry.insert(MrfBucketBacklogObservability {
+                bucket: bucket_name.to_string(),
+                ..Default::default()
+            }),
+        }
+    }
+
+    fn add_pending(&mut self, entry: &MrfReplicateEntry) {
+        let Ok(size) = u64::try_from(entry.size) else {
+            return;
+        };
+        let bucket = self.bucket_mut(&entry.bucket);
+        bucket.pending_count = bucket.pending_count.saturating_add(1);
+        bucket.pending_bytes = bucket.pending_bytes.saturating_add(size);
+    }
+
+    fn flush_pending_entries<'a>(&mut self, entries: impl IntoIterator<Item = &'a MrfReplicateEntry>, duration_millis: u64) {
+        for entry in entries {
+            let Ok(size) = u64::try_from(entry.size) else {
+                continue;
+            };
+            let bucket = self.bucket_mut(&entry.bucket);
+            bucket.pending_count = bucket.pending_count.saturating_sub(1);
+            bucket.pending_bytes = bucket.pending_bytes.saturating_sub(size);
+            bucket.last_flush_duration_millis = duration_millis;
+        }
+    }
+
+    fn record_drop(&mut self, entry: &MrfReplicateEntry) {
+        let bucket = self.bucket_mut(&entry.bucket);
+        bucket.dropped_count = bucket.dropped_count.saturating_add(1);
+    }
+
+    fn record_missed(&mut self, bucket_name: &str) {
+        let bucket = self.bucket_mut(bucket_name);
+        bucket.missed_count = bucket.missed_count.saturating_add(1);
+    }
+
+    fn record_flush_failure(&mut self, duration_millis: u64) {
+        for bucket in self.buckets.values_mut().filter(|bucket| bucket.pending_count > 0) {
+            bucket.flush_failure_count = bucket.flush_failure_count.saturating_add(1);
+            bucket.last_flush_duration_millis = duration_millis;
+        }
+    }
+
+    fn snapshot(&self) -> MrfBacklogObservabilitySummary {
+        MrfBacklogObservabilitySummary {
+            buckets: self.buckets.values().cloned().collect(),
+        }
+    }
+}
+
+fn durable_mrf_backlog_summary_from_sizes<I>(entries: I) -> DurableMrfBacklogSummary
+where
+    I: IntoIterator<Item = (String, i64)>,
+{
+    let mut tracker = DurableMrfBacklogTracker {
+        available: true,
+        ..Default::default()
+    };
+    for (bucket_name, entry_size) in entries {
+        tracker.add_entry(bucket_name, entry_size);
+    }
+    tracker.into_summary()
+}
+
 fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) {
     match DURABLE_MRF_BACKLOG_SUMMARY.write() {
         Ok(mut guard) => *guard = summary,
@@ -138,6 +246,40 @@ pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary {
     }
 }
 
+pub fn mrf_backlog_observability_snapshot() -> MrfBacklogObservabilitySummary {
+    match MRF_BACKLOG_OBSERVABILITY.read() {
+        Ok(guard) => guard.snapshot(),
+        Err(poisoned) => poisoned.into_inner().snapshot(),
+    }
+}
+
+fn update_mrf_backlog_observability(mut update: impl FnMut(&mut MrfBacklogObservabilityTracker)) {
+    match MRF_BACKLOG_OBSERVABILITY.write() {
+        Ok(mut guard) => update(&mut guard),
+        Err(poisoned) => update(&mut poisoned.into_inner()),
+    }
+}
+
+fn observe_mrf_pending(entry: &MrfReplicateEntry) {
+    update_mrf_backlog_observability(|tracker| tracker.add_pending(entry));
+}
+
+fn observe_mrf_pending_flushed(entries: &[MrfReplicateEntry], duration_millis: u64) {
+    update_mrf_backlog_observability(|tracker| tracker.flush_pending_entries(entries, duration_millis));
+}
+
+fn observe_mrf_drop(entry: &MrfReplicateEntry) {
+    update_mrf_backlog_observability(|tracker| tracker.record_drop(entry));
+}
+
+fn observe_mrf_missed(bucket: &str) {
+    update_mrf_backlog_observability(|tracker| tracker.record_missed(bucket));
+}
+
+fn observe_mrf_flush_failure(duration_millis: u64) {
+    update_mrf_backlog_observability(|tracker| tracker.record_flush_failure(duration_millis));
+}
+
 fn durable_mrf_backlog_from_read(result: Result<Vec<u8>, EcstoreError>) -> DurableMrfBacklog {
     match result {
         Ok(data) => match decode_mrf_file(&data) {
@@ -302,26 +444,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let handle = tokio::spawn(async move {
                 let mut rx = rx;
                 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);
-                        }
-                    }
-
-                    active_counter.fetch_sub(1, Ordering::SeqCst);
+                    let _active = ActiveWorkerGuard::new(active_counter.clone());
+                    process_replication_operation(operation, stats.clone(), storage.clone()).await;
                 }
             });
 
@@ -379,30 +503,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let handle = tokio::spawn(async move {
                 let mut rx = rx;
                 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;
-                            // Perform actual replication (placeholder)
-                            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;
-                            // Perform actual delete replication (placeholder)
-                            replicate_delete(*del_info, storage.clone()).await;
-
-                            stats.dec_q(&bucket, 0, true, op_type);
-                        }
-                    }
-
-                    active_counter.fetch_sub(1, Ordering::SeqCst);
+                    let _active = ActiveWorkerGuard::new(active_counter.clone());
+                    process_replication_operation(operation, stats.clone(), storage.clone()).await;
                 }
             });
 
@@ -448,24 +550,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                     let operation = { mrf_rx.lock().await.recv().await };
                     let Some(operation) = operation else { break };
 
-                    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);
-                        }
-                    }
-                    active_counter.fetch_sub(1, Ordering::SeqCst);
+                    let _active = ActiveWorkerGuard::new(active_counter.clone());
+                    process_replication_operation(operation, stats.clone(), storage.clone()).await;
                 }
             });
             self.task_handles.lock().await.push(handle);
@@ -902,6 +988,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             // sustained failure storm can't grow it without limit.
             const MRF_PENDING_CAP: usize = 200_000;
             let mut pending: Vec<MrfReplicateEntry> = Vec::new();
+            let mut durable_tracker = DurableMrfBacklogTracker {
+                available: true,
+                ..Default::default()
+            };
             let mut flushed_len = 0usize;
             let mut dirty = false;
             let mut capped = false;
@@ -915,6 +1005,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                     entry = rx.recv() => match entry {
                         Some(e) => {
                             if pending.len() >= MRF_PENDING_CAP {
+                                observe_mrf_drop(&e);
                                 dec_mrf_entries(stats.as_ref(), std::slice::from_ref(&e));
                                 if !capped {
                                     capped = true;
@@ -927,16 +1018,19 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                                 }
                                 continue;
                             }
+                            durable_tracker.add_entry(e.bucket.clone(), e.size);
+                            observe_mrf_pending(&e);
                             pending.push(e);
                             dirty = true;
                             // Flush eagerly once enough new entries have accumulated
                             // since the last write (measured against the flushed
                             // 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)),
-                                ));
+                            if pending.len() - flushed_len >= 1000
+                                && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await
+                            {
+                                set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary());
+                                observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis);
                                 dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                                 flushed_len = pending.len();
                                 dirty = false;
@@ -944,20 +1038,18 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                         }
                         None => {
                             // Channel closed (pool shutting down) — final flush.
-                            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)),
-                                ));
+                            if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await {
+                                set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary());
+                                observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis);
                                 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)),
-                            ));
+                        if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await {
+                            set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary());
+                            observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis);
                             dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                             flushed_len = pending.len();
                             dirty = false;
@@ -977,30 +1069,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         stats: Arc<ReplicationStats>,
     ) {
         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;
-                    // Perform actual replication (placeholder)
-                    replicate_object(*obj_info, self.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;
-                    // Perform actual delete replication (placeholder)
-                    replicate_delete(*del_info, self.storage.clone()).await;
-
-                    stats.dec_q(&bucket, 0, true, op_type);
-                }
-            }
-
-            active_counter.fetch_sub(1, Ordering::SeqCst);
+            let _active = ActiveWorkerGuard::new(active_counter.clone());
+            process_replication_operation(operation, stats.clone(), self.storage.clone()).await;
         }
     }
 
@@ -1013,26 +1083,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         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);
-                }
-            }
-
-            active_counter.fetch_sub(1, Ordering::SeqCst);
+            let _active = ActiveWorkerGuard::new(active_counter.clone());
+            process_replication_operation(operation, stats.clone(), storage.clone()).await;
         }
     }
 
@@ -1044,27 +1096,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         stats: Arc<ReplicationStats>,
     ) {
         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.as_ref().clone(), self.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, self.storage.clone()).await;
-                    stats.dec_q(&bucket, 0, true, op_type);
-                }
-            }
-
-            active_counter.fetch_sub(1, Ordering::SeqCst);
+            let _active = ActiveWorkerGuard::new(active_counter.clone());
+            process_replication_operation(operation, stats.clone(), self.storage.clone()).await;
         }
     }
 
@@ -1395,6 +1428,76 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 }
 
+struct ActiveWorkerGuard {
+    counter: Arc<AtomicI32>,
+}
+
+impl ActiveWorkerGuard {
+    fn new(counter: Arc<AtomicI32>) -> Self {
+        counter.fetch_add(1, Ordering::SeqCst);
+        Self { counter }
+    }
+}
+
+impl Drop for ActiveWorkerGuard {
+    fn drop(&mut self) {
+        self.counter.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
+struct ReplicationBacklogGuard {
+    stats: Arc<ReplicationStats>,
+    bucket: String,
+    size: i64,
+    is_delete_marker: bool,
+    op_type: ReplicationType,
+}
+
+impl ReplicationBacklogGuard {
+    fn for_object(stats: Arc<ReplicationStats>, object: &ReplicateObjectInfo) -> Self {
+        Self {
+            stats,
+            bucket: object.bucket.clone(),
+            size: object.size,
+            is_delete_marker: object.delete_marker,
+            op_type: object.op_type,
+        }
+    }
+
+    fn for_delete(stats: Arc<ReplicationStats>, delete: &DeletedObjectReplicationInfo) -> Self {
+        Self {
+            stats,
+            bucket: delete.bucket.clone(),
+            size: 0,
+            is_delete_marker: true,
+            op_type: delete.op_type,
+        }
+    }
+}
+
+impl Drop for ReplicationBacklogGuard {
+    fn drop(&mut self) {
+        self.stats.dec_q(&self.bucket, self.size, self.is_delete_marker, self.op_type);
+    }
+}
+
+async fn process_replication_operation<S: ReplicationStorage>(
+    operation: ReplicationOperation,
+    stats: Arc<ReplicationStats>,
+    storage: Arc<S>,
+) {
+    match operation {
+        ReplicationOperation::Object(obj_info) => {
+            let _backlog = ReplicationBacklogGuard::for_object(stats, obj_info.as_ref());
+            replicate_object(*obj_info, storage).await;
+        }
+        ReplicationOperation::Delete(del_info) => {
+            let _backlog = ReplicationBacklogGuard::for_delete(stats, del_info.as_ref());
+            replicate_delete(*del_info, storage).await;
+        }
+    }
+}
+
 async fn queue_mrf_save_entry(
     tx: &Sender<MrfReplicateEntry>,
     entry: MrfReplicateEntry,
@@ -1414,6 +1517,7 @@ async fn queue_mrf_save_entry(
         queue_type = queue_type,
         "MRF save channel unavailable — replication failure entry could not be persisted for retry"
     );
+    observe_mrf_missed(&entry.bucket);
     ReplicationQueueAdmission::Missed
 }
 
@@ -1424,15 +1528,18 @@ fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) {
 }
 
 /// 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
+/// Returns the flush duration on success; on failure logs the error and returns `None`.
+/// Callers must NOT clear their in-memory buffer on `None` so the next tick
 /// can retry — otherwise a transient storage error permanently drops the batch.
-async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> bool {
+async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> Option<u64> {
+    let started = Instant::now();
     match encode_mrf_file(entries) {
         Ok(data) => {
             if let Err(e) =
                 ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await
             {
+                let duration_millis = duration_millis_u64(started.elapsed());
+                observe_mrf_flush_failure(duration_millis);
                 warn!(
                     component = LOG_COMPONENT_ECSTORE,
                     subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -1440,11 +1547,12 @@ async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry]
                     error = %e,
                     "Failed to flush MRF entries to disk"
                 );
-                return false;
+                return None;
             }
-            true
+            Some(duration_millis_u64(started.elapsed()))
         }
         Err(e) => {
+            observe_mrf_flush_failure(0);
             warn!(
                 component = LOG_COMPONENT_ECSTORE,
                 subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -1452,11 +1560,15 @@ async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry]
                 error = %e,
                 "Failed to encode MRF entries for disk flush"
             );
-            false
+            None
         }
     }
 }
 
+fn duration_millis_u64(duration: std::time::Duration) -> u64 {
+    u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
+}
+
 /// Load bucket resync metadata from disk
 async fn load_bucket_resync_metadata<S: ReplicationObjectIO>(
     bucket: &str,
@@ -2664,6 +2776,132 @@ mod tests {
         assert_eq!(received.object, "second");
     }
 
+    #[tokio::test]
+    async fn mrf_save_admission_records_missed_when_channel_is_closed() {
+        let (tx, rx) = mpsc::channel(1);
+        drop(rx);
+        let bucket = "mrf-missed-hook-bucket";
+
+        let admission = queue_mrf_save_entry(
+            &tx,
+            MrfReplicateEntry {
+                bucket: bucket.to_string(),
+                object: "missed".to_string(),
+                version_id: None,
+                retry_count: 1,
+                size: 1,
+                op: MrfOpKind::Object,
+                delete_marker_version_id: None,
+                delete_marker: false,
+                delete_marker_mtime: None,
+            },
+            "test",
+        )
+        .await;
+
+        assert_eq!(admission, ReplicationQueueAdmission::Missed);
+        let snapshot = mrf_backlog_observability_snapshot();
+        let bucket = snapshot
+            .buckets
+            .iter()
+            .find(|stats| stats.bucket == "mrf-missed-hook-bucket")
+            .expect("missed MRF admission should be observable");
+        assert_eq!(bucket.missed_count, 1);
+    }
+
+    #[tokio::test]
+    async fn mrf_flush_failure_keeps_pending_backlog_observable() {
+        let shared = empty_resync_shared_state();
+        shared.fail_next_write.store(true, Ordering::SeqCst);
+        let storage = Arc::new(LoadResyncNodeStore::new("mrf-flush-failure", shared));
+        let entry = MrfReplicateEntry {
+            bucket: "mrf-flush-failure-bucket".to_string(),
+            object: "pending".to_string(),
+            version_id: None,
+            retry_count: 1,
+            size: 2048,
+            op: MrfOpKind::Object,
+            delete_marker_version_id: None,
+            delete_marker: false,
+            delete_marker_mtime: None,
+        };
+        observe_mrf_pending(&entry);
+
+        let result = flush_mrf_to_disk(std::slice::from_ref(&entry), &storage).await;
+
+        assert_eq!(result, None);
+        let snapshot = mrf_backlog_observability_snapshot();
+        let bucket = snapshot
+            .buckets
+            .iter()
+            .find(|stats| stats.bucket == "mrf-flush-failure-bucket")
+            .expect("failed MRF flush should keep the bucket observable");
+        assert_eq!(bucket.pending_count, 1);
+        assert_eq!(bucket.pending_bytes, 2048);
+        assert_eq!(bucket.flush_failure_count, 1);
+    }
+
+    #[tokio::test]
+    async fn replication_backlog_guard_decrements_on_drop() {
+        let stats = Arc::new(ReplicationStats::new());
+        stats.inc_q("guard-bucket", 256, false, ReplicationType::Object);
+
+        {
+            let object = ReplicateObjectInfo {
+                bucket: "guard-bucket".to_string(),
+                size: 256,
+                op_type: ReplicationType::Object,
+                ..Default::default()
+            };
+            let _guard = ReplicationBacklogGuard::for_object(stats.clone(), &object);
+        }
+
+        let queued = stats.get_latest_replication_stats("guard-bucket").await;
+        assert_eq!(queued.replication_stats.q_stat.curr.count, 0);
+        assert_eq!(queued.replication_stats.q_stat.curr.bytes, 0);
+    }
+
+    #[test]
+    fn mrf_observability_tracker_separates_pending_drop_miss_and_flush_failure() {
+        let first = MrfReplicateEntry {
+            bucket: "tracker-bucket".to_string(),
+            object: "first".to_string(),
+            version_id: None,
+            retry_count: 1,
+            size: 1024,
+            op: MrfOpKind::Object,
+            delete_marker_version_id: None,
+            delete_marker: false,
+            delete_marker_mtime: None,
+        };
+        let second = MrfReplicateEntry {
+            object: "second".to_string(),
+            size: 512,
+            ..first.clone()
+        };
+        let mut tracker = MrfBacklogObservabilityTracker::default();
+
+        tracker.add_pending(&first);
+        tracker.add_pending(&second);
+        tracker.record_drop(&second);
+        tracker.record_missed("tracker-bucket");
+        tracker.record_flush_failure(7);
+        tracker.flush_pending_entries([&first], 11);
+
+        let snapshot = tracker.snapshot();
+        let bucket = snapshot
+            .buckets
+            .iter()
+            .find(|stats| stats.bucket == "tracker-bucket")
+            .expect("tracker bucket should be present");
+        assert_eq!(bucket.pending_count, 1);
+        assert_eq!(bucket.pending_bytes, 512);
+        assert_eq!(bucket.dropped_count, 1);
+        assert_eq!(bucket.missed_count, 1);
+        assert_eq!(bucket.flush_failure_count, 1);
+        assert_eq!(bucket.last_flush_duration_millis, 11);
+    }
+
     #[test]
     fn auto_resume_resync_only_for_inflight_states() {
         assert!(should_auto_resume_resync(ResyncStatusType::ResyncPending));

crates/obs/src/metrics/collectors/bucket_replication.rs

@@ -20,6 +20,8 @@ use crate::metrics::schema::bucket_replication::{
     BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD,
     BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_LAST_HR_FAILED_BYTES_MD, BUCKET_REPL_LAST_HR_FAILED_COUNT_MD,
     BUCKET_REPL_LAST_MIN_FAILED_BYTES_MD, BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD, BUCKET_REPL_LATENCY_MS_MD,
+    BUCKET_REPL_MRF_DROPPED_COUNT_MD, BUCKET_REPL_MRF_FLUSH_FAILURES_MD, BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD,
+    BUCKET_REPL_MRF_MISSED_COUNT_MD, BUCKET_REPL_MRF_PENDING_BYTES_MD, BUCKET_REPL_MRF_PENDING_COUNT_MD,
     BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD,
     BUCKET_REPL_PROXIED_GET_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_GET_REQUESTS_TOTAL_MD,
     BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_TOTAL_MD,
@@ -34,7 +36,7 @@ use crate::metrics::schema::bucket_replication::{
 use std::borrow::Cow;
 
 const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25;
-const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 5;
+const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 11;
 
 #[derive(Debug, Clone, Default)]
 pub struct BucketReplicationTargetStats {
@@ -91,6 +93,12 @@ pub(crate) struct BucketReplicationBacklogStats {
     pub(crate) durable_mrf_available: bool,
     pub(crate) durable_mrf_backlog_count: u64,
     pub(crate) durable_mrf_backlog_bytes: u64,
+    pub(crate) mrf_pending_count: u64,
+    pub(crate) mrf_pending_bytes: u64,
+    pub(crate) mrf_dropped_count: u64,
+    pub(crate) mrf_missed_count: u64,
+    pub(crate) mrf_flush_failures: u64,
+    pub(crate) mrf_last_flush_duration_millis: u64,
 }
 
 pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec<PrometheusMetric> {
@@ -309,6 +317,34 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat
             PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD, stat.durable_mrf_backlog_bytes as f64)
                 .with_label(BUCKET_L, bucket_label),
         );
+        let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone());
+        metrics.push(
+            PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_PENDING_COUNT_MD, stat.mrf_pending_count as f64)
+                .with_label(BUCKET_L, bucket_label.clone()),
+        );
+        metrics.push(
+            PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_PENDING_BYTES_MD, stat.mrf_pending_bytes as f64)
+                .with_label(BUCKET_L, bucket_label.clone()),
+        );
+        metrics.push(
+            PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_DROPPED_COUNT_MD, stat.mrf_dropped_count as f64)
+                .with_label(BUCKET_L, bucket_label.clone()),
+        );
+        metrics.push(
+            PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_MISSED_COUNT_MD, stat.mrf_missed_count as f64)
+                .with_label(BUCKET_L, bucket_label.clone()),
+        );
+        metrics.push(
+            PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_FLUSH_FAILURES_MD, stat.mrf_flush_failures as f64)
+                .with_label(BUCKET_L, bucket_label.clone()),
+        );
+        metrics.push(
+            PrometheusMetric::from_descriptor(
+                &BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD,
+                stat.mrf_last_flush_duration_millis as f64,
+            )
+            .with_label(BUCKET_L, bucket_label),
+        );
     }
 
     metrics
@@ -427,10 +463,16 @@ mod tests {
             durable_mrf_available: true,
             durable_mrf_backlog_count: 2,
             durable_mrf_backlog_bytes: 2048,
+            mrf_pending_count: 1,
+            mrf_pending_bytes: 512,
+            mrf_dropped_count: 3,
+            mrf_missed_count: 4,
+            mrf_flush_failures: 5,
+            mrf_last_flush_duration_millis: 6,
         }];
 
         let metrics = collect_bucket_replication_backlog_metrics(&stats);
-        assert_eq!(metrics.len(), 5);
+        assert_eq!(metrics.len(), 11);
 
         let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name();
         assert!(metrics.iter().any(|metric| {
@@ -466,6 +508,48 @@ mod tests {
                 && metric.value == 2048.0
                 && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
         }));
+
+        let pending_count_name = BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == pending_count_name
+                && metric.value == 1.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
+        let pending_bytes_name = BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == pending_bytes_name
+                && metric.value == 512.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
+        let dropped_count_name = BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == dropped_count_name
+                && metric.value == 3.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
+        let missed_count_name = BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == missed_count_name
+                && metric.value == 4.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
+        let flush_failures_name = BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == flush_failures_name
+                && metric.value == 5.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
+
+        let flush_duration_name = BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name();
+        assert!(metrics.iter().any(|metric| {
+            metric.name == flush_duration_name
+                && metric.value == 6.0
+                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
+        }));
     }
 
     #[test]
@@ -484,6 +568,12 @@ mod tests {
             durable_mrf_available: true,
             durable_mrf_backlog_count: 2,
             durable_mrf_backlog_bytes: 4096,
+            mrf_pending_count: 1,
+            mrf_pending_bytes: 2,
+            mrf_dropped_count: 3,
+            mrf_missed_count: 4,
+            mrf_flush_failures: 5,
+            mrf_last_flush_duration_millis: 6,
         }];
 
         let metrics = collect_bucket_replication_backlog_metrics(&stats);
@@ -493,6 +583,12 @@ mod tests {
             BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD.get_full_metric_name(),
             BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD.get_full_metric_name(),
             BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD.get_full_metric_name(),
+            BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(),
+            BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(),
+            BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(),
+            BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(),
+            BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(),
+            BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(),
         ];
 
         for name in backlog_names {

crates/obs/src/metrics/schema/bucket_replication.rs

@@ -38,6 +38,12 @@ const CURRENT_BACKLOG_BYTES: &str = "current_backlog_bytes";
 const DURABLE_MRF_AVAILABLE: &str = "durable_mrf_available";
 const DURABLE_MRF_BACKLOG_COUNT: &str = "durable_mrf_backlog_count";
 const DURABLE_MRF_BACKLOG_BYTES: &str = "durable_mrf_backlog_bytes";
+const MRF_PENDING_COUNT: &str = "mrf_pending_count";
+const MRF_PENDING_BYTES: &str = "mrf_pending_bytes";
+const MRF_DROPPED_COUNT: &str = "mrf_dropped_count";
+const MRF_MISSED_COUNT: &str = "mrf_missed_count";
+const MRF_FLUSH_FAILURES: &str = "mrf_flush_failures";
+const MRF_LAST_FLUSH_DURATION_MILLIS: &str = "mrf_last_flush_duration_millis";
 
 pub static BUCKET_REPL_LAST_HR_FAILED_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
@@ -129,6 +135,60 @@ pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD: LazyLock<MetricDescriptor>
     )
 });
 
+pub static BUCKET_REPL_MRF_PENDING_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::from(MRF_PENDING_COUNT),
+        "Current number of MRF entries waiting to be flushed to the durable recovery file for a bucket on this node",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
+pub static BUCKET_REPL_MRF_PENDING_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::from(MRF_PENDING_BYTES),
+        "Current bytes represented by MRF entries waiting to be flushed to the durable recovery file for a bucket on this node",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
+pub static BUCKET_REPL_MRF_DROPPED_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(MRF_DROPPED_COUNT),
+        "Total number of MRF entries dropped after the bounded pending backlog cap was reached for a bucket on this node",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
+pub static BUCKET_REPL_MRF_MISSED_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(MRF_MISSED_COUNT),
+        "Total number of MRF entries that could not be admitted to the save channel for a bucket on this node",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
+pub static BUCKET_REPL_MRF_FLUSH_FAILURES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_counter_md(
+        MetricName::from(MRF_FLUSH_FAILURES),
+        "Total number of durable MRF flush failures observed while a bucket had pending MRF entries on this node",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
+pub static BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
+    new_gauge_md(
+        MetricName::from(MRF_LAST_FLUSH_DURATION_MILLIS),
+        "Duration in milliseconds of the last durable MRF flush that touched pending entries for a bucket on this node",
+        &[BUCKET_L],
+        subsystems::BUCKET_REPLICATION,
+    )
+});
+
 pub static BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ProxiedDeleteTaggingRequestsTotal,

crates/obs/src/metrics/stats_collector.rs

@@ -206,6 +206,12 @@ async fn obs_bucket_replication_stats_bundle() -> (Vec<BucketReplicationStats>,
             durable_mrf_available: stats.durable_mrf_available,
             durable_mrf_backlog_count: stats.durable_mrf_backlog_count,
             durable_mrf_backlog_bytes: stats.durable_mrf_backlog_bytes,
+            mrf_pending_count: stats.mrf_pending_count,
+            mrf_pending_bytes: stats.mrf_pending_bytes,
+            mrf_dropped_count: stats.mrf_dropped_count,
+            mrf_missed_count: stats.mrf_missed_count,
+            mrf_flush_failures: stats.mrf_flush_failures,
+            mrf_last_flush_duration_millis: stats.mrf_last_flush_duration_millis,
         });
         detail_stats.push(bucket_replication_detail_from_snapshot(stats));
     }

crates/obs/src/metrics/storage_api.rs

@@ -18,7 +18,8 @@ use std::time::Duration;
 pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor;
 pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config;
 use rustfs_ecstore::api::bucket::replication::{
-    DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, get_global_replication_stats,
+    DurableMrfBucketBacklog, MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, get_global_replication_stats,
+    mrf_backlog_observability_snapshot,
 };
 pub(crate) use rustfs_ecstore::api::capacity::{
     get_total_usable_capacity as obs_get_total_usable_capacity,
@@ -76,6 +77,12 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot {
     pub(crate) durable_mrf_available: bool,
     pub(crate) durable_mrf_backlog_count: u64,
     pub(crate) durable_mrf_backlog_bytes: u64,
+    pub(crate) mrf_pending_count: u64,
+    pub(crate) mrf_pending_bytes: u64,
+    pub(crate) mrf_dropped_count: u64,
+    pub(crate) mrf_missed_count: u64,
+    pub(crate) mrf_flush_failures: u64,
+    pub(crate) mrf_last_flush_duration_millis: u64,
     pub(crate) targets: Vec<ObsBucketReplicationTargetStatsSnapshot>,
 }
 
@@ -152,6 +159,7 @@ fn bucket_replication_stats_snapshot_from_parts(
     proxy: ObsBucketReplicationProxySnapshot,
     durable_mrf_available: bool,
     durable_bucket: DurableMrfBucketBacklog,
+    mrf_observability: MrfBucketBacklogObservability,
 ) -> ObsBucketReplicationStatsSnapshot {
     ObsBucketReplicationStatsSnapshot {
         bucket,
@@ -185,6 +193,12 @@ fn bucket_replication_stats_snapshot_from_parts(
         durable_mrf_available,
         durable_mrf_backlog_count: durable_bucket.count,
         durable_mrf_backlog_bytes: durable_bucket.bytes,
+        mrf_pending_count: mrf_observability.pending_count,
+        mrf_pending_bytes: mrf_observability.pending_bytes,
+        mrf_dropped_count: mrf_observability.dropped_count,
+        mrf_missed_count: mrf_observability.missed_count,
+        mrf_flush_failures: mrf_observability.flush_failure_count,
+        mrf_last_flush_duration_millis: mrf_observability.last_flush_duration_millis,
         targets: runtime.targets,
     }
 }
@@ -196,7 +210,8 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
     } else {
         HashMap::new()
     };
-    let durable_mrf_summary = if obs_resolve_object_store_handle().is_some() {
+    let replication_storage_available = obs_resolve_object_store_handle().is_some();
+    let durable_mrf_summary = if replication_storage_available {
         durable_mrf_backlog_summary_snapshot()
     } else {
         Default::default()
@@ -207,7 +222,22 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
         .into_iter()
         .map(|bucket| (bucket.bucket.clone(), bucket))
         .collect::<HashMap<String, DurableMrfBucketBacklog>>();
-    let mut bucket_names = Vec::with_capacity(all_bucket_stats.len().saturating_add(durable_buckets.len()));
+    let mrf_observability = if replication_storage_available {
+        mrf_backlog_observability_snapshot()
+    } else {
+        Default::default()
+    };
+    let mrf_observability_buckets = mrf_observability
+        .buckets
+        .into_iter()
+        .map(|bucket| (bucket.bucket.clone(), bucket))
+        .collect::<HashMap<String, MrfBucketBacklogObservability>>();
+    let mut bucket_names = Vec::with_capacity(
+        all_bucket_stats
+            .len()
+            .saturating_add(durable_buckets.len())
+            .saturating_add(mrf_observability_buckets.len()),
+    );
     bucket_names.extend(all_bucket_stats.keys().cloned());
     bucket_names.extend(
         durable_buckets
@@ -215,6 +245,12 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
             .filter(|bucket| !all_bucket_stats.contains_key(*bucket))
             .cloned(),
     );
+    bucket_names.extend(
+        mrf_observability_buckets
+            .keys()
+            .filter(|bucket| !all_bucket_stats.contains_key(*bucket) && !durable_buckets.contains_key(*bucket))
+            .cloned(),
+    );
     let mut buckets = Vec::with_capacity(bucket_names.len());
 
     for bucket in bucket_names {
@@ -291,12 +327,14 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
             runtime.current_backlog_bytes = i64_to_u64_floor_zero(bucket_stats.q_stat.curr.bytes);
         }
         let durable_bucket = durable_buckets.get(&bucket).cloned().unwrap_or_default();
+        let mrf_observability = mrf_observability_buckets.get(&bucket).cloned().unwrap_or_default();
         buckets.push(bucket_replication_stats_snapshot_from_parts(
             bucket,
             runtime,
             proxy,
             durable_mrf_available,
             durable_bucket,
+            mrf_observability,
         ));
     }
 
@@ -394,6 +432,15 @@ mod tests {
                 count: 5,
                 bytes: 8192,
             },
+            MrfBucketBacklogObservability {
+                bucket: "runtime-bucket".to_string(),
+                pending_count: 1,
+                pending_bytes: 512,
+                dropped_count: 2,
+                missed_count: 3,
+                flush_failure_count: 4,
+                last_flush_duration_millis: 5,
+            },
         );
 
         assert_eq!(snapshot.bucket, "runtime-bucket");
@@ -402,6 +449,12 @@ mod tests {
         assert!(snapshot.durable_mrf_available);
         assert_eq!(snapshot.durable_mrf_backlog_count, 5);
         assert_eq!(snapshot.durable_mrf_backlog_bytes, 8192);
+        assert_eq!(snapshot.mrf_pending_count, 1);
+        assert_eq!(snapshot.mrf_pending_bytes, 512);
+        assert_eq!(snapshot.mrf_dropped_count, 2);
+        assert_eq!(snapshot.mrf_missed_count, 3);
+        assert_eq!(snapshot.mrf_flush_failures, 4);
+        assert_eq!(snapshot.mrf_last_flush_duration_millis, 5);
         assert_eq!(snapshot.resync_failed_count, 2);
         assert_eq!(snapshot.proxied_get_requests_total, 7);
         assert_eq!(snapshot.proxied_get_requests_failures, 1);
@@ -419,6 +472,7 @@ mod tests {
                 count: 11,
                 bytes: 2048,
             },
+            MrfBucketBacklogObservability::default(),
         );
 
         assert_eq!(snapshot.bucket, "durable-only");
@@ -428,6 +482,39 @@ mod tests {
         assert_eq!(snapshot.durable_mrf_backlog_bytes, 2048);
     }
 
+    #[test]
+    fn bucket_replication_snapshot_reports_mrf_observability_only_bucket() {
+        let snapshot = bucket_replication_stats_snapshot_from_parts(
+            "mrf-observability-only".to_string(),
+            ObsBucketReplicationRuntimeSnapshot::default(),
+            ObsBucketReplicationProxySnapshot::default(),
+            true,
+            DurableMrfBucketBacklog::default(),
+            MrfBucketBacklogObservability {
+                bucket: "mrf-observability-only".to_string(),
+                pending_count: 13,
+                pending_bytes: 4096,
+                dropped_count: 1,
+                missed_count: 2,
+                flush_failure_count: 3,
+                last_flush_duration_millis: 4,
+            },
+        );
+
+        assert_eq!(snapshot.bucket, "mrf-observability-only");
+        assert_eq!(snapshot.current_backlog_count, 0);
+        assert_eq!(snapshot.current_backlog_bytes, 0);
+        assert!(snapshot.durable_mrf_available);
+        assert_eq!(snapshot.durable_mrf_backlog_count, 0);
+        assert_eq!(snapshot.durable_mrf_backlog_bytes, 0);
+        assert_eq!(snapshot.mrf_pending_count, 13);
+        assert_eq!(snapshot.mrf_pending_bytes, 4096);
+        assert_eq!(snapshot.mrf_dropped_count, 1);
+        assert_eq!(snapshot.mrf_missed_count, 2);
+        assert_eq!(snapshot.mrf_flush_failures, 3);
+        assert_eq!(snapshot.mrf_last_flush_duration_millis, 4);
+    }
+
     #[test]
     fn bucket_replication_snapshot_preserves_durable_mrf_unavailable_state() {
         let snapshot = bucket_replication_stats_snapshot_from_parts(
@@ -440,6 +527,7 @@ mod tests {
             ObsBucketReplicationProxySnapshot::default(),
             false,
             DurableMrfBucketBacklog::default(),
+            MrfBucketBacklogObservability::default(),
         );
 
         assert_eq!(snapshot.current_backlog_count, 1);