ci: stop archiving every dependency's unpacked source in each cache (#5566)

The four consolidated ci keys plus the build keys still do not fit the
repository's fixed 10GB Actions cache quota, so LRU keeps evicting them:
measured demand is ci-dev 2331MB + ci-feat-proto 2310MB + ci-feat-rio 1951MB +
ci-uring 1317MB + two build legs at ~1429MB + cargo-deny 844MB, and main pushes
add two more build legs. That is roughly 14.4GB against 10.24GB. The symptom is
misleading: Cache Warm reports every job successful and the restores log "full
match: true", yet ci-feat-rio and ci-uring disappear from the cache list between
runs.

cache-all-crates was the wrong default for this repository. With it set to true,
rust-cache's cleanup returns before pruning ~/.cargo/registry/src and its config
archives the whole registry, so every cache carried the unpacked source tree of
every dependency — not, as the name suggests, just a few extra crates.

Setting it to false is rust-cache's own default and loses no coverage: the
package set comes from `cargo metadata --all-features`, a strict superset of any
single lane's feature closure; -sys crates are explicitly exempt from pruning,
since their source timestamps would otherwise trigger rebuilds; and everything
pruned is re-unpacked from the .crate files still in registry/cache, whose
mtimes crates.io normalises, so cargo fingerprints stay valid.

Applied to the setup composite and to audit.yml's own rust-cache. Cache Warm
now also reports the sizes of registry/src, registry/cache, registry/index,
~/.cargo/git and target/ to the step summary, immediately before rust-cache's
post step archives them, so the size of the effect is measured rather than
assumed.

The new sizes only appear once the cache key next rotates, since rust-cache
skips the save entirely on an exact key hit. Deliberately not forcing that by
bumping prefix-key: it would invalidate every family at once and produce a
repository-wide cold build.

Refs: rustfs/backlog#1598, rustfs/backlog#1600

.config/nextest.toml

@@ -300,9 +300,6 @@ path = "junit.xml"
 # negative-path siblings of each family stay in as regression guards.
 #   * rustfs#4843 — over-limit archive entry paths hard-reject the whole
 #     archive even under ignore-errors semantics.
-#   * rustfs#4846 — distributed-lock quorum tests misclassify as timeout
-#     under parallel load (multi-node in-process clusters; natural home is
-#     ci-7's nightly cluster lane).
 [profile.e2e-full]
 default-filter = """
     package(e2e_test)
@@ -311,7 +308,6 @@ default-filter = """
     & !test(/^replication_extension_test::/)
     & !test(/^multipart_auth_test::test_signed_put_object_extract_skips_invalid_entry_when_ignore_errors_enabled$/)
     & !test(/^snowball_auto_extract_test::tests::snowball_auto_extract_(ignores_invalid_entries_when_requested|supports_standard_headers_with_combined_extract_options)$/)
-    & !test(/^reliant::lock::test_distributed_lock_(2_nodes_grpc_read_survives_failed_node|4_nodes_grpc_read_write_quorum_split_with_two_failed_nodes)$/)
 """
 fail-fast = false
 

.github/actions/setup/action.yml

@@ -111,7 +111,18 @@ runs:
     - name: Setup Rust cache
       uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
       with:
-        cache-all-crates: true
+        # false is rust-cache's own default. With true, cleanup.ts returns
+        # *before* pruning ~/.cargo/registry/src, and config.ts archives the
+        # whole registry — so every cache carried the unpacked source tree of
+        # every dependency, not just "a few extra crates".
+        #
+        # No coverage is lost: getPackages runs `cargo metadata --all-features`,
+        # a strict superset of any single lane's feature closure, and -sys crates
+        # are explicitly exempted from pruning (their src timestamps would
+        # otherwise trigger rebuilds). Anything pruned is re-unpacked from the
+        # .crate files still in registry/cache, whose mtimes crates.io
+        # normalises, so cargo fingerprints stay valid.
+        cache-all-crates: false
         cache-on-failure: true
         shared-key: ${{ inputs.cache-shared-key }}
         save-if: ${{ inputs.cache-save-if }}

.github/workflows/audit.yml

@@ -100,7 +100,9 @@ jobs:
       - name: Setup Rust cache
         uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
         with:
-          cache-all-crates: true
+          # Same reasoning as the setup composite: true archives every
+          # dependency's unpacked source tree.
+          cache-all-crates: false
           cache-on-failure: true
           shared-key: rustfs-cargo-deny
           save-if: ${{ github.ref == 'refs/heads/main' }}

.github/workflows/cache-warm.yml

@@ -107,10 +107,6 @@ jobs:
           cache-save-if: 'true'
           install-build-packaging-tools: 'false'
 
-      # --all-targets covers the test binaries nextest builds, including
-      # e2e_test, which test-and-lint's own run excludes. The second build adds
-      # the e2e-test-hooks feature resolution that build-rustfs-debug-binary uses
-      # and that no lint lane enables.
       # rustfs/backlog#1601 gate. sccache can only cache compilation units whose
       # --emit includes link, so it covers workspace rlibs and nothing else:
       # clippy is metadata-only, and the ~100 test binaries, the rustfs bin and
@@ -138,6 +134,10 @@ jobs:
           retention-days: 30
           if-no-files-found: error
 
+      # --all-targets covers the test binaries nextest builds, including
+      # e2e_test, which test-and-lint's own run excludes. The second build adds
+      # the e2e-test-hooks feature resolution that build-rustfs-debug-binary uses
+      # and that no lint lane enables.
       - name: Build ci-dev superset
         env:
           # Same limit ci.yml puts on its nextest step: this builds the same
@@ -148,6 +148,21 @@ jobs:
           cargo build --workspace --all-targets
           cargo build -p rustfs --bins --features e2e-test-hooks
 
+      # Runs before rust-cache's post step, so these are the sizes it is about
+      # to archive. Reported so the cache-all-crates decision stays evidence-led:
+      # registry/src is what that flag prunes, registry/cache is what the pruned
+      # sources are re-unpacked from. See rustfs/backlog#1600.
+      - name: Report cache input sizes
+        if: always()
+        run: |
+          {
+            echo "### Cache input sizes (ci-dev)"
+            echo '```'
+            du -sh ~/.cargo/registry/src ~/.cargo/registry/cache ~/.cargo/registry/index \
+                   ~/.cargo/git target 2>/dev/null || true
+            echo '```'
+          } >> "$GITHUB_STEP_SUMMARY"
+
   # Readers: test-and-lint-rio-v2, build-rustfs-debug-binary-rio-v2.
   warm-ci-feat-rio:
     name: Warm ci-feat-rio

.github/workflows/runner-hygiene.yml

@@ -0,0 +1,81 @@
+# Copyright 2026 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Asserts that the self-hosted runners are still ephemeral — one job per pod.
+#
+# This repository is public and its pull_request jobs run on those runners,
+# executing the PR's own build.rs, proc-macros and tests. The only thing keeping
+# that code from reaching a later job is that each ARC pod handles exactly one
+# job and is then destroyed. That guarantee lives in the ARC scale-set
+# configuration, outside this repository, where it can be changed without any PR
+# — so it is asserted here from the outside, against real run data, instead of
+# being assumed.
+#
+# Monthly rather than per-PR: the property changes only when someone
+# reconfigures the scale set, and the check costs a few dozen API calls.
+# See docs/ci/runners.md and rustfs/backlog#1602.
+
+name: Runner Hygiene
+
+on:
+  schedule:
+    - cron: "0 6 1 * *" # Monthly, 1st at 06:00 UTC (after the daily audit cron)
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: runner-hygiene
+  cancel-in-progress: false
+
+jobs:
+  check-ephemerality:
+    name: Check runner ephemerality
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+
+      # Exit 2 (inconclusive / broken) is deliberately not a pass: a window
+      # where every sm-* job was still queued would otherwise look identical to
+      # a clean bill of health.
+      - name: Assert one job per self-hosted runner
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: ./scripts/ci/check_runner_ephemerality.sh 40
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    needs: [check-ephemerality]
+    # Same ci-8 mechanism as coverage.yml, audit.yml and the nightly lanes:
+    # scheduled runs file a tracking issue, manual dispatch stays quiet so
+    # debugging never produces a spurious alert.
+    if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

crates/e2e_test/src/reliant/lock.rs

@@ -15,9 +15,7 @@
 
 use super::{grpc_lock_client::GrpcLockClient, grpc_lock_server::spawn_lock_server};
 use rustfs_lock::client::{LockClient, local::LocalClient};
-use rustfs_lock::{
-    GlobalLockManager, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey,
-};
+use rustfs_lock::{GlobalLockManager, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -35,7 +33,11 @@ struct FailingClient;
 #[async_trait::async_trait]
 impl rustfs_lock::LockClient for FailingClient {
     async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result<LockResponse> {
-        Err(LockError::internal("simulated gRPC node failure"))
+        // Match RemoteClient's transport-failure response so the coordinator can count this node toward quorum loss.
+        Ok(LockResponse::failure(
+            "Remote lock RPC failed: simulated gRPC node failure",
+            Duration::ZERO,
+        ))
     }
 
     async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {

crates/e2e_test/src/replication_extension_test.rs

@@ -29,9 +29,8 @@ 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_ENCODING, CONTENT_TYPE, HOST};
+use http::header::{CONTENT_TYPE, HOST};
 use http_body_util::{BodyExt, Full};
 use hyper::body::Incoming;
 use hyper::server::conn::http1;
@@ -39,10 +38,6 @@ 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,
@@ -61,7 +56,6 @@ 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;
@@ -70,13 +64,11 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use time::{Duration as TimeDuration, OffsetDateTime};
 use tokio::fs;
 use tokio::net::TcpListener;
-use tokio::sync::{Mutex, watch};
-use tokio::task::JoinHandle;
+use tokio::sync::watch;
 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
@@ -115,250 +107,6 @@ 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 {
@@ -3911,125 +3659,6 @@ 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,8 +173,7 @@ pub mod bucket {
 
     pub mod replication {
         pub use crate::bucket::replication::replication_pool::{
-            DurableMrfBacklogSummary, DurableMrfBucketBacklog, MrfBacklogObservabilitySummary, MrfBucketBacklogObservability,
-            durable_mrf_backlog_summary_snapshot, mrf_backlog_observability_snapshot,
+            DurableMrfBacklogSummary, DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot,
         };
         pub use crate::bucket::replication::{
             BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool,

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

@@ -53,7 +53,6 @@ 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;
@@ -92,46 +91,20 @@ 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()));
 
-#[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) {
+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 {
-            self.available = false;
-            self.buckets.clear();
-            return;
+            return DurableMrfBacklogSummary::default();
         };
 
-        if !self.available {
-            return;
-        }
-
-        let bucket = match self.buckets.entry(bucket_name) {
+        let bucket = match buckets.entry(bucket_name) {
             Entry::Occupied(entry) => entry.into_mut(),
             Entry::Vacant(entry) => {
                 let bucket = entry.key().clone();
@@ -145,91 +118,10 @@ impl DurableMrfBacklogTracker {
         bucket.bytes = bucket.bytes.saturating_add(size);
     }
 
-    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 {
+    DurableMrfBacklogSummary {
         available: true,
-        ..Default::default()
-    };
-    for (bucket_name, entry_size) in entries {
-        tracker.add_entry(bucket_name, entry_size);
+        buckets: buckets.into_values().collect(),
     }
-    tracker.into_summary()
 }
 
 fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) {
@@ -246,40 +138,6 @@ 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) {
@@ -444,8 +302,26 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let handle = tokio::spawn(async move {
                 let mut rx = rx;
                 while let Some(operation) = rx.recv().await {
-                    let _active = ActiveWorkerGuard::new(active_counter.clone());
-                    process_replication_operation(operation, stats.clone(), storage.clone()).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);
                 }
             });
 
@@ -503,8 +379,30 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let handle = tokio::spawn(async move {
                 let mut rx = rx;
                 while let Some(operation) = rx.recv().await {
-                    let _active = ActiveWorkerGuard::new(active_counter.clone());
-                    process_replication_operation(operation, stats.clone(), storage.clone()).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);
                 }
             });
 
@@ -550,8 +448,24 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                     let operation = { mrf_rx.lock().await.recv().await };
                     let Some(operation) = operation else { break };
 
-                    let _active = ActiveWorkerGuard::new(active_counter.clone());
-                    process_replication_operation(operation, stats.clone(), storage.clone()).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);
                 }
             });
             self.task_handles.lock().await.push(handle);
@@ -988,10 +902,6 @@ 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;
@@ -1005,7 +915,6 @@ 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;
@@ -1018,19 +927,16 @@ 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
-                                && 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);
+                            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;
@@ -1038,18 +944,20 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                         }
                         None => {
                             // Channel closed (pool shutting down) — final flush.
-                            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);
+                            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 && 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);
+                        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;
@@ -1069,8 +977,30 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         stats: Arc<ReplicationStats>,
     ) {
         while let Some(operation) = rx.recv().await {
-            let _active = ActiveWorkerGuard::new(active_counter.clone());
-            process_replication_operation(operation, stats.clone(), self.storage.clone()).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);
         }
     }
 
@@ -1083,8 +1013,26 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         storage: Arc<S>,
     ) {
         while let Some(operation) = rx.recv().await {
-            let _active = ActiveWorkerGuard::new(active_counter.clone());
-            process_replication_operation(operation, stats.clone(), storage.clone()).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);
         }
     }
 
@@ -1096,8 +1044,27 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         stats: Arc<ReplicationStats>,
     ) {
         while let Some(operation) = rx.recv().await {
-            let _active = ActiveWorkerGuard::new(active_counter.clone());
-            process_replication_operation(operation, stats.clone(), self.storage.clone()).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);
         }
     }
 
@@ -1428,76 +1395,6 @@ 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,
@@ -1517,7 +1414,6 @@ 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
 }
 
@@ -1528,18 +1424,15 @@ fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) {
 }
 
 /// Encodes `entries` and overwrites the MRF persistence file.
-/// 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
+/// 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
 /// can retry — otherwise a transient storage error permanently drops the batch.
-async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> Option<u64> {
-    let started = Instant::now();
+async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> bool {
     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,
@@ -1547,12 +1440,11 @@ async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry]
                     error = %e,
                     "Failed to flush MRF entries to disk"
                 );
-                return None;
+                return false;
             }
-            Some(duration_millis_u64(started.elapsed()))
+            true
         }
         Err(e) => {
-            observe_mrf_flush_failure(0);
             warn!(
                 component = LOG_COMPONENT_ECSTORE,
                 subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -1560,15 +1452,11 @@ async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry]
                 error = %e,
                 "Failed to encode MRF entries for disk flush"
             );
-            None
+            false
         }
     }
 }
 
-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,
@@ -2776,132 +2664,6 @@ 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,8 +20,6 @@ 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,
@@ -36,7 +34,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 = 11;
+const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 5;
 
 #[derive(Debug, Clone, Default)]
 pub struct BucketReplicationTargetStats {
@@ -93,12 +91,6 @@ 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> {
@@ -317,34 +309,6 @@ 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
@@ -463,16 +427,10 @@ 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(), 11);
+        assert_eq!(metrics.len(), 5);
 
         let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name();
         assert!(metrics.iter().any(|metric| {
@@ -508,48 +466,6 @@ 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]
@@ -568,12 +484,6 @@ 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);
@@ -583,12 +493,6 @@ 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,12 +38,6 @@ 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(
@@ -135,60 +129,6 @@ 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,12 +206,6 @@ 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,8 +18,7 @@ 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, MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, get_global_replication_stats,
-    mrf_backlog_observability_snapshot,
+    DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, get_global_replication_stats,
 };
 pub(crate) use rustfs_ecstore::api::capacity::{
     get_total_usable_capacity as obs_get_total_usable_capacity,
@@ -77,12 +76,6 @@ 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>,
 }
 
@@ -159,7 +152,6 @@ fn bucket_replication_stats_snapshot_from_parts(
     proxy: ObsBucketReplicationProxySnapshot,
     durable_mrf_available: bool,
     durable_bucket: DurableMrfBucketBacklog,
-    mrf_observability: MrfBucketBacklogObservability,
 ) -> ObsBucketReplicationStatsSnapshot {
     ObsBucketReplicationStatsSnapshot {
         bucket,
@@ -193,12 +185,6 @@ 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,
     }
 }
@@ -210,8 +196,7 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
     } else {
         HashMap::new()
     };
-    let replication_storage_available = obs_resolve_object_store_handle().is_some();
-    let durable_mrf_summary = if replication_storage_available {
+    let durable_mrf_summary = if obs_resolve_object_store_handle().is_some() {
         durable_mrf_backlog_summary_snapshot()
     } else {
         Default::default()
@@ -222,22 +207,7 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
         .into_iter()
         .map(|bucket| (bucket.bucket.clone(), bucket))
         .collect::<HashMap<String, DurableMrfBucketBacklog>>();
-    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()),
-    );
+    let mut bucket_names = Vec::with_capacity(all_bucket_stats.len().saturating_add(durable_buckets.len()));
     bucket_names.extend(all_bucket_stats.keys().cloned());
     bucket_names.extend(
         durable_buckets
@@ -245,12 +215,6 @@ 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 {
@@ -327,14 +291,12 @@ 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,
         ));
     }
 
@@ -432,15 +394,6 @@ 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");
@@ -449,12 +402,6 @@ 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);
@@ -472,7 +419,6 @@ mod tests {
                 count: 11,
                 bytes: 2048,
             },
-            MrfBucketBacklogObservability::default(),
         );
 
         assert_eq!(snapshot.bucket, "durable-only");
@@ -482,39 +428,6 @@ 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(
@@ -527,7 +440,6 @@ mod tests {
             ObsBucketReplicationProxySnapshot::default(),
             false,
             DurableMrfBucketBacklog::default(),
-            MrfBucketBacklogObservability::default(),
         );
 
         assert_eq!(snapshot.current_backlog_count, 1);

scripts/ci/check_runner_ephemerality.sh

@@ -0,0 +1,77 @@
+#!/usr/bin/env bash
+# Assert that self-hosted runners are ephemeral: one job per runner, ever.
+#
+# WHY THIS IS CHECKED CONTINUOUSLY RATHER THAN ONCE
+#
+# rustfs/rustfs is public, and pull_request jobs run on self-hosted runners —
+# they execute the PR's own build.rs, proc-macros and tests, which is arbitrary
+# code. What keeps that from reaching the release build is that each runner is a
+# fresh ARC pod that handles exactly one job and is destroyed. Nothing in this
+# repository enforces it: it is a property of the ARC scale-set configuration,
+# which lives outside the repo and can be changed without any PR. So it is
+# asserted from the outside, on real data.
+#
+# A repeated runner_name means a runner served two jobs, i.e. state survived
+# between them, i.e. a poisoned PR job could leave something behind for whatever
+# runs next — including, while the release build shares the sm-standard-2 pool,
+# a release build.
+#
+# Usage: scripts/ci/check_runner_ephemerality.sh [run-count]
+#   run-count defaults to 40. MIN_SAMPLE (default 10) is the number of observed
+#   assignments below which the window is reported INCONCLUSIVE rather than OK.
+#
+# Exit codes: 0 ephemeral, 1 a runner served two jobs, 2 inconclusive or broken.
+set -euo pipefail
+
+REPO="${GITHUB_REPOSITORY:-rustfs/rustfs}"
+LIMIT="${1:-40}"
+# Below this many observed assignments the window says nothing useful: with the
+# pool saturated, most sm-* jobs sit queued with runner_name still null, and a
+# handful of samples would pass by coincidence rather than by evidence.
+MIN_SAMPLE="${MIN_SAMPLE:-10}"
+
+command -v gh >/dev/null || { echo "ERROR: gh CLI not found" >&2; exit 2; }
+
+runs="$(gh run list --repo "$REPO" --limit "$LIMIT" --json databaseId --jq '.[].databaseId')"
+[ -n "$runs" ] || { echo "ERROR: no runs returned for $REPO" >&2; exit 2; }
+
+names="$(
+    for run in $runs; do
+        gh api "repos/${REPO}/actions/runs/${run}/jobs" --paginate \
+            --jq '.jobs[] | select(.runner_name != null) | select(.runner_name | startswith("sm-")) | .runner_name' 2>/dev/null || true
+    done
+)"
+
+total="$(printf '%s\n' "$names" | grep -c . || true)"
+if [ "${total:-0}" -eq 0 ]; then
+    echo "ERROR: no self-hosted (sm-*) runner names found across $LIMIT runs." >&2
+    echo "  Either the label scheme changed or the API shape did — this check" >&2
+    echo "  must not silently pass by finding nothing." >&2
+    exit 2
+fi
+
+if [ "$total" -lt "$MIN_SAMPLE" ]; then
+    echo "INCONCLUSIVE: only ${total} self-hosted job assignments across ${LIMIT} runs" >&2
+    echo "  (need ${MIN_SAMPLE}). Most sm-* jobs are probably still queued, so" >&2
+    echo "  runner_name is null. Re-run with a larger window when the pool drains." >&2
+    exit 2
+fi
+
+dupes="$(printf '%s\n' "$names" | sort | uniq -d)"
+
+if [ -n "$dupes" ]; then
+    echo "ERROR: self-hosted runners served more than one job — not ephemeral:" >&2
+    printf '%s\n' "$dupes" | while read -r name; do
+        count="$(printf '%s\n' "$names" | grep -c "^${name}$")"
+        echo "  ${name}: ${count} jobs" >&2
+    done
+    echo "" >&2
+    echo "State can survive between jobs on those runners. Because this is a" >&2
+    echo "public repository whose pull_request jobs run PR-authored code, and" >&2
+    echo "because build.yml's release legs share the sm-standard-2 pool, a" >&2
+    echo "poisoned PR job could reach a release build. Check the ARC scale-set" >&2
+    echo "configuration (rustfs/backlog#1602)." >&2
+    exit 1
+fi
+
+echo "OK: ${total} self-hosted job assignments across ${LIMIT} runs, all on distinct runners"