fix(heal): cleanup consumed MRF replay journals

Do not retain Accepted or Merged replay intents as startup anchors after they have been handed to the heal manager. Only refused or still-pending replay records keep the journal on disk until a successor snapshot can persist them.

This keeps successor snapshots limited to the pending queue, which lets successful replay remove both authoritative and legacy journal paths and restores the crash-boundary tests around successor flush.

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
(cherry picked from commit d5b8f49c9d)

crates/e2e_test/src/admin_timeout_regression_test.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::common::{RustFSTestClusterEnvironment, init_logging, local_http_client, signal_process};
+use crate::common::{RustFSTestClusterEnvironment, init_logging, local_http_client};
 use aws_sdk_s3::primitives::ByteStream;
 use http::header::HOST;
 use reqwest::StatusCode;
@@ -22,6 +22,7 @@ use rustfs_signer::sign_v4;
 use s3s::Body;
 use serde::Deserialize;
 use std::error::Error;
+use std::process::Command;
 use tokio::time::{Duration, sleep, timeout};
 use uuid::Uuid;
 
@@ -81,6 +82,15 @@ async fn parse_json_response<T: serde::de::DeserializeOwned>(
     Ok(serde_json::from_slice(&body)?)
 }
 
+fn signal_process(pid: u32, signal: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
+    let output = Command::new("kill").arg(format!("-{signal}")).arg(pid.to_string()).output()?;
+    if output.status.success() {
+        return Ok(());
+    }
+
+    Err(format!("kill -{signal} {pid} failed: {}", String::from_utf8_lossy(&output.stderr)).into())
+}
+
 fn offline_server_count(info: &InfoMessage) -> usize {
     info.servers
         .as_ref()

crates/e2e_test/src/common.rs

@@ -264,15 +264,6 @@ pub fn local_http_client() -> HttpClient {
         .expect("failed to build local reqwest client")
 }
 
-pub(crate) fn signal_process(pid: u32, signal: &str) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let output = Command::new("kill").arg(format!("-{signal}")).arg(pid.to_string()).output()?;
-    if output.status.success() {
-        return Ok(());
-    }
-
-    Err(format!("kill -{signal} {pid} failed: {}", String::from_utf8_lossy(&output.stderr)).into())
-}
-
 pub(crate) async fn signed_s3_request(
     method: http::Method,
     url: &str,

crates/e2e_test/src/distributed/observability_test.rs

@@ -12,10 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use super::harness::{
-    DistCluster, DistLayout, TestResult, assert_object_bytes, cluster_admin_ok, unique_bucket, wait_for_ready, wait_until,
-};
-use crate::common::{admin_request, init_logging, local_http_client, signal_process, signed_request};
+use super::harness::{DistCluster, DistLayout, TestResult, cluster_admin_ok, unique_bucket, wait_for_ready};
+use crate::common::{admin_request, init_logging, local_http_client};
 use aws_sdk_s3::operation::RequestId;
 use aws_sdk_s3::primitives::ByteStream;
 use bytes::Bytes;
@@ -26,7 +24,7 @@ use hyper::service::service_fn;
 use hyper::{Request, Response};
 use hyper_util::rt::TokioIo;
 use local_ip_address::local_ip;
-use rustfs_madmin::metrics::{HttpMetrics, RealtimeMetrics};
+use rustfs_madmin::metrics::RealtimeMetrics;
 use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
 use serde_json::Value;
 use std::convert::Infallible;
@@ -128,7 +126,6 @@ async fn four_node_health_inventory_metrics_and_audit_delivery_are_consistent()
     let (audit_endpoint, mut audit_entries, collector) = spawn_audit_collector().await?;
     let audit_origin = reqwest::Url::parse(&audit_endpoint)?.origin().ascii_serialization();
     let audit_env = [
-        ("RUST_LOG", "warn"),
         ("RUSTFS_AUDIT_ENABLE", "true"),
         ("RUSTFS_AUDIT_WEBHOOK_ENABLE_DISTRIBUTED", "on"),
         ("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_DISTRIBUTED", audit_endpoint.as_str()),
@@ -234,186 +231,6 @@ async fn four_node_health_inventory_metrics_and_audit_delivery_are_consistent()
         "audit entry leaked the root secret key"
     );
 
-    let result = verify_write_observations_during_peer_failure(&dist, &bucket).await;
     collector.abort();
-    result
-}
-
-async fn node_admin_body(dist: &DistCluster, node: usize, path: &str) -> TestResult<String> {
-    let (status, body) = timeout(
-        Duration::from_secs(30),
-        admin_request(
-            &dist.cluster.nodes[node].url,
-            Method::GET,
-            path,
-            None,
-            &dist.cluster.access_key,
-            &dist.cluster.secret_key,
-        ),
-    )
-    .await??;
-    assert!(status.is_success(), "node {node} admin request {path}: {status} {body}");
-    Ok(body)
-}
-
-async fn http_put_counts(dist: &DistCluster, node: usize) -> TestResult<[u64; 2]> {
-    let body = node_admin_body(dist, node, "/rustfs/admin/v3/metrics?types=512&by-host=true&n=1").await?;
-    let sample: RealtimeMetrics = serde_json::from_str(body.lines().next().ok_or("empty HTTP metrics stream")?)?;
-    assert!(sample.errors.is_empty(), "HTTP metrics returned errors: {:?}", sample.errors);
-    let http = sample.aggregated.http.ok_or("HTTP metrics missing at WARN log level")?;
-    let count = |http: &HttpMetrics, outcome: &str| {
-        http.requests
-            .iter()
-            .filter(|row| row.method == "PUT" && row.outcome == outcome)
-            .map(|row| row.total)
-            .sum::<u64>()
-    };
-    assert_eq!(sample.by_host.len(), 1, "HTTP admin metrics must remain node-local");
-    let host = sample.by_host.values().next().expect("one reporting host");
-    let host = host.http.as_ref().ok_or("by-host HTTP metrics missing")?;
-    let totals = [count(&http, "2xx"), count(&http, "5xx")];
-    assert_eq!(totals, [count(host, "2xx"), count(host, "5xx")]);
-    Ok(totals)
-}
-
-async fn observed_put(dist: &DistCluster, node: usize, bucket: &str, key: &str) -> TestResult<http::StatusCode> {
-    // One signed HTTP attempt: SDK retries must not change the expected denominator.
-    timeout(Duration::from_secs(90), async {
-        let response = signed_request(
-            Method::PUT,
-            &format!("{}/{bucket}/{key}", dist.cluster.nodes[node].url),
-            &dist.cluster.access_key,
-            &dist.cluster.secret_key,
-            Some(b"write-observation".to_vec()),
-            Some("application/octet-stream"),
-        )
-        .await?;
-        assert!(response.headers().contains_key("x-amz-request-id"), "PUT omitted correlation ID");
-        let status = response.status();
-        let body = response.text().await?;
-        assert!(!body.contains(&dist.cluster.secret_key), "PUT response leaked credentials");
-        Ok::<_, Box<dyn std::error::Error + Send + Sync>>(status)
-    })
-    .await?
-}
-
-struct SuspendedPeer<'a> {
-    // Borrowing the owned child keeps its PID from being reaped/reused before cleanup.
-    child: &'a std::process::Child,
-    suspended: bool,
-}
-
-impl<'a> SuspendedPeer<'a> {
-    fn suspend(dist: &'a DistCluster, node: usize) -> TestResult<Self> {
-        let child = dist.cluster.nodes[node].process.as_ref().ok_or("peer process missing")?;
-        signal_process(child.id(), "STOP")?;
-        Ok(Self { child, suspended: true })
-    }
-
-    fn resume(&mut self) -> TestResult {
-        signal_process(self.child.id(), "CONT")?;
-        self.suspended = false;
-        Ok(())
-    }
-}
-
-impl Drop for SuspendedPeer<'_> {
-    fn drop(&mut self) {
-        if self.suspended {
-            let _ = signal_process(self.child.id(), "CONT");
-        }
-    }
-}
-
-async fn verify_write_observations_during_peer_failure(dist: &DistCluster, bucket: &str) -> TestResult {
-    let mut baseline = Vec::new();
-    for node in 0..dist.cluster.nodes.len() {
-        let before = http_put_counts(dist, node).await?;
-        assert!(
-            observed_put(dist, node, bucket, &format!("healthy-{node}"))
-                .await?
-                .is_success()
-        );
-        let after = http_put_counts(dist, node).await?;
-        assert_eq!(after, [before[0] + 1, before[1]], "node {node} lost its successful PUT denominator");
-        baseline.push(after);
-    }
-
-    // Refresh provenance immediately before the first failed probe, within the cache age budget.
-    node_admin_body(dist, 0, "/rustfs/admin/v3/storageinfo").await?;
-    let mut suspended = [SuspendedPeer::suspend(dist, 2)?, SuspendedPeer::suspend(dist, 3)?];
-    let storage: Value = serde_json::from_str(&node_admin_body(dist, 0, "/rustfs/admin/v3/storageinfo").await?)?;
-    let observations = storage["info"]["observations"]
-        .as_array()
-        .ok_or("storageinfo omitted observations")?;
-    let disks = storage["info"]["disks"]
-        .as_array()
-        .ok_or("storageinfo omitted disks during peer failure")?;
-    assert_eq!(disks.len(), 16, "failed peers must not vanish from inventory");
-    for node in [2, 3] {
-        let endpoint = &dist.cluster.nodes[node].address;
-        let observation = observations
-            .iter()
-            .find(|item| item["endpoint"].as_str().is_some_and(|value| value.contains(endpoint)))
-            .ok_or_else(|| format!("missing failed peer observation {endpoint}: {storage}"))?;
-        assert_eq!(observation["status"], "failed", "suspension did not affect peer RPC: {observation}");
-        assert_eq!(observation["cached"], true, "first failure must identify the warm cache: {observation}");
-        assert!(observation["last_success_unix_millis"].as_u64().is_some());
-        assert!(observation["snapshot_age_seconds"].as_u64().is_some_and(|age| age < 60));
-        let peer_disks: Vec<_> = disks
-            .iter()
-            .filter(|disk| disk["endpoint"].as_str().is_some_and(|value| value.contains(endpoint)))
-            .collect();
-        assert_eq!(peer_disks.len(), 4, "failed peer lost its four drive identities: {storage}");
-        for disk in peer_disks {
-            assert_eq!(disk["state"], "unknown");
-            assert_eq!(disk["runtimeState"], "unknown");
-        }
-    }
-
-    let snapshot: Value = serde_json::from_str(&node_admin_body(dist, 0, "/rustfs/admin/v4/cluster/snapshot").await?)?;
-    let metadata = &snapshot["snapshot"]["pool_meta_write_gate"];
-    assert_eq!(
-        metadata["state"], "writable",
-        "peer probe failure must not invent a metadata latch: {snapshot}"
-    );
-    assert!(metadata.get("sinceUnixSecs").is_none());
-
-    for attempt in 0..2 {
-        let status = observed_put(dist, 0, bucket, &format!("unavailable-{attempt}")).await?;
-        assert!(status.is_server_error(), "sub-quorum write unexpectedly returned {status}");
-    }
-    assert_eq!(http_put_counts(dist, 0).await?, [baseline[0][0], baseline[0][1] + 2]);
-    assert_eq!(
-        http_put_counts(dist, 1).await?,
-        baseline[1],
-        "internal RPCs must not count as external PUTs"
-    );
-
-    for peer in &mut suspended {
-        peer.resume()?;
-    }
-    wait_until(
-        Duration::from_secs(90),
-        || async {
-            let storage: Value = serde_json::from_str(&node_admin_body(dist, 0, "/rustfs/admin/v3/storageinfo").await?)?;
-            let observations = storage["info"]["observations"]
-                .as_array()
-                .ok_or("recovery omitted observations")?;
-            Ok(observations.len() == 4
-                && observations
-                    .iter()
-                    .all(|item| item["status"] == "succeeded" && item["cached"] == false))
-        },
-        "peer probes recover to fresh successful observations",
-    )
-    .await?;
-    wait_for_ready(&dist.cluster).await?;
-    assert!(observed_put(dist, 0, bucket, "recovered").await?.is_success());
-    assert_eq!(http_put_counts(dist, 0).await?, [baseline[0][0] + 1, baseline[0][1] + 2]);
-    for node in 0..dist.cluster.nodes.len() {
-        assert_object_bytes(&dist.client(node)?, bucket, "healthy-0", b"write-observation").await?;
-        assert_object_bytes(&dist.client(node)?, bucket, "recovered", b"write-observation").await?;
-    }
     Ok(())
 }

crates/ecstore/src/core/pools.rs

@@ -10682,30 +10682,18 @@ impl ECStore {
         else {
             return PoolMetaWriteGateStatus {
                 writes_ready: false,
-                check_timed_out: true,
                 ..PoolMetaWriteGateStatus::default()
             };
         };
         let transaction_aborted = write_state.aborted_transaction.load(Ordering::SeqCst);
-        let writes_ready = !write_state.write_blocked && !transaction_aborted;
-        let failure = if writes_ready {
-            None
-        } else {
-            write_state.ensure_write_safe("pool metadata snapshot").err()
-        };
-        let context = failure.as_ref().and_then(Error::pool_metadata_failure);
         PoolMetaWriteGateStatus {
-            writes_ready,
-            check_timed_out: false,
+            writes_ready: !write_state.write_blocked && !transaction_aborted,
             write_blocked: write_state.write_blocked,
             transaction_aborted,
             pool_meta_absent: write_state.pool_meta_absent,
             identity_initialized: write_state.identity_initialized,
             identity_needs_repair: write_state.identity_needs_repair,
             cluster_epoch: write_state.cluster_epoch,
-            reason: context.map(|context| context.kind.as_str()),
-            phase: context.map(|context| context.phase),
-            since_unix_secs: context.map(|context| context.since.unix_timestamp()),
         }
     }
 
@@ -19014,12 +19002,6 @@ mod tests {
         drop(outstanding);
         assert!(store.recover_pool_meta_transaction().await.unwrap());
         assert!(store.pool_meta_writes_ready().await);
-        let recovered = store.pool_meta_write_gate_status().await;
-        assert!(recovered.writes_ready);
-        assert!(!recovered.check_timed_out);
-        assert!(!recovered.write_blocked);
-        assert!(!recovered.transaction_aborted);
-        assert_eq!((recovered.reason, recovered.phase, recovered.since_unix_secs), (None, None, None));
         assert_eq!(store.pool_meta.read().await.pools[0].last_update, requested.pools[0].last_update);
         assert!(!store.recover_pool_meta_transaction().await.unwrap());
         store.pool_meta_save_gate.lock().await.block_writes_after_fence_loss();
@@ -19036,39 +19018,17 @@ mod tests {
             .await
             .unwrap();
         assert!(matches!(result, Err(Error::Timeout)));
-        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(1), store.pool_meta_write_gate_status())
-            .await
-            .unwrap();
-        assert!(!snapshot.writes_ready);
-        assert!(snapshot.check_timed_out);
-        assert!(!snapshot.write_blocked);
-        assert!(!snapshot.transaction_aborted);
-        assert_eq!((snapshot.reason, snapshot.phase, snapshot.since_unix_secs), (None, None, None));
     }
 
     #[tokio::test]
     #[serial_test::serial]
     async fn pool_meta_cancelled_recovery_and_new_integrity_block_never_clear_original_gate() {
         let (_dirs, store, _peer) = crate::services::rebalance::test_two_pool_stores(None).await;
-        let since = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap();
         {
             let state = store.pool_meta_save_gate.lock().await;
             let mut arm = state.arm_transaction();
             arm.phase = Some("publication");
-            drop(arm);
-            // An old fixed timestamp detects polling-time resets without sleeping.
-            state.transaction_failure.lock().unwrap().as_mut().unwrap().since = since;
-            *state.block_started_at.lock().unwrap() = Some(since);
         }
-        let original = store.pool_meta_write_gate_status().await;
-        assert!(!original.writes_ready);
-        assert!(!original.check_timed_out);
-        assert!(!original.write_blocked);
-        assert!(original.transaction_aborted);
-        assert_eq!(original.reason, Some("transaction_unknown"));
-        assert_eq!(original.phase, Some("publication"));
-        assert_eq!(original.since_unix_secs, Some(since.unix_timestamp()));
-        assert_eq!(store.pool_meta_write_gate_status().await, original);
         let publication_guard = store.pool_meta.write().await;
         let recovery = tokio::spawn({
             let store = store.clone();
@@ -19088,7 +19048,6 @@ mod tests {
         assert!(recovery.await.unwrap_err().is_cancelled());
         drop(publication_guard);
         assert!(!store.pool_meta_writes_ready().await);
-        assert_eq!(store.pool_meta_write_gate_status().await, original);
         let start_guard = store.start_gate.lock().await;
         let recovery = tokio::spawn({
             let store = store.clone();
@@ -19107,12 +19066,6 @@ mod tests {
                 .kind,
             crate::error::PoolMetadataFailure::FenceLost
         );
-        let fenced = store.pool_meta_write_gate_status().await;
-        assert!(!fenced.writes_ready);
-        assert!(fenced.write_blocked);
-        assert_eq!(fenced.reason, Some("fence_lost"));
-        assert_eq!(fenced.phase, Some("format_heal"));
-        assert_eq!(fenced.since_unix_secs, original.since_unix_secs);
     }
 
     #[test]

crates/ecstore/src/store/mod.rs

@@ -530,32 +530,24 @@ impl Default for ScannerDataMovementPauseStatus {
 #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
 pub struct PoolMetaWriteGateStatus {
     pub writes_ready: bool,
-    pub check_timed_out: bool,
     pub write_blocked: bool,
     pub transaction_aborted: bool,
     pub pool_meta_absent: bool,
     pub identity_initialized: Option<bool>,
     pub identity_needs_repair: bool,
     pub cluster_epoch: Option<u64>,
-    pub reason: Option<&'static str>,
-    pub phase: Option<&'static str>,
-    pub since_unix_secs: Option<i64>,
 }
 
 impl Default for PoolMetaWriteGateStatus {
     fn default() -> Self {
         Self {
             writes_ready: true,
-            check_timed_out: false,
             write_blocked: false,
             transaction_aborted: false,
             pool_meta_absent: false,
             identity_initialized: None,
             identity_needs_repair: false,
             cluster_epoch: None,
-            reason: None,
-            phase: None,
-            since_unix_secs: None,
         }
     }
 }

docs/architecture/readiness-matrix.md

@@ -41,12 +41,8 @@ an unknown or unsupported peer-health snapshot degrades readiness with
   lock quorum, or peer health.
 - Node readiness reports local dependency readiness.
 - A blocked pool metadata writer degrades node and cluster-write readiness with
-  `pool_meta_write_blocked`. Metadata save-gate inspection is bounded to 100 ms;
+  `pool_metadata_blocked`. Metadata save-gate inspection is bounded to 100 ms;
   contention reports `pool_metadata_check_timeout` without installing a block.
-- The authenticated cluster snapshot extends its existing node-local metadata
-  gate inspection with safe reason, failure phase, and original block time. It
-  distinguishes timeout from a block and changes no admission or recovery
-  decision. Runtime readiness and gate status are separate bounded observations.
 - Cluster write readiness requires write quorum and the runtime dependency
   readiness used by `FullReady`.
 - Cluster read readiness may use the read-quorum path and cluster-health

docs/operations/pool-metadata-recovery.md

@@ -49,7 +49,7 @@ An unreadable replica, lost fence, or conditional-write conflict leaves the orig
 
 - The first block emits `decommission_state` with `state=pool_metadata_blocked`, `reason`, `phase`, and `blocked_since`. A change in recovery failure classification emits `state=pool_metadata_recovery_pending`; successful recovery emits `state=pool_metadata_recovered` with the original timestamp.
 - `rustfs_pool_metadata_blocks_total{reason}` and `rustfs_pool_metadata_recoveries_total` count block and recovery transitions. The original cause and phase remain attached to local typed errors; storage/RPC error numbers and on-disk formats are unchanged.
-- Node and cluster-write readiness include `pool_meta_write_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md).
+- Node and cluster-write readiness include `pool_metadata_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged.
 
 If a block persists, inspect the first block and subsequent recovery phase, restore disk/peer readability, and verify every metadata and identity copy before restarting. Do not delete metadata to make readiness green.
 

docs/operations/s3-write-failure-diagnostics.md

@@ -45,23 +45,10 @@ The probe round timeout is configured independently; see [Admin peer probe timeo
 
 ## Correlate bounded diagnostics
 
-Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while runtime readiness diagnostics distinguish `pool_meta_write_blocked`, `pool_metadata_check_timeout`, and insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green.
-
-Query the authenticated `/rustfs/admin/v4/cluster/snapshot` endpoint on the affected node. `snapshot.pool_meta_write_gate` describes that node's metadata writer, not a fleet-wide aggregate. Its existing booleans are preserved; `state` and optional block details extend the same bounded, read-only gate inspection. This starts no recovery, disk reads, or additional RPCs. Runtime readiness and the gate are inspected separately, so a whole cluster snapshot is not atomic across sections.
-
-| `state` | Meaning |
-| --- | --- |
-| `writable` | No metadata write block was observed. Other dependencies may still make the node unready. |
-| `blocked` | A metadata write block was observed; `reason`, `phase`, and `sinceUnixSecs` describe its typed failure context. |
-| `check_timeout` | The existing 100 ms inspection budget expired. Readiness remains false, but a write block is not asserted. |
-| `unavailable` | The object store or a recognized metadata observation was unavailable; no block details are invented. |
-
-For `blocked`, `reason` reuses the metadata failure classification, `phase` identifies the operation stage that caused the block (not live recovery-worker progress), and `sinceUnixSecs` is the original block time in Unix seconds. Polling does not reset that time. Recovery replaces the observation with `writable` and removes the previous block details. An unavailable store reports `writesReady=false` with `state="unavailable"`. Older responses may omit `state` and block details; missing fields alone do not prove a healthy writer. Operation text, raw replica errors, disk paths, and credentials are excluded. See [Pool metadata recovery](pool-metadata-recovery.md) for recovery and escalation boundaries.
+Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while existing runtime readiness diagnostics distinguish `pool_meta_write_blocked` from insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green.
 
 PUT storage failures retain their typed source chain internally and emit bounded S3/storage error codes, I/O kinds, and RPC status codes alongside the existing request ID, bucket, and key. Raw nested error strings and RPC metadata are not logged by this diagnostic. A repeated PUT diagnostic is limited to one event per five seconds; HTTP server-error logs are limited per status code over the same interval for accounted S3 traffic. `suppressed_errors` reports suppressed events at the next emitted event; use the HTTP counter, not log-line counts, to measure failures. HTTP server-error URI diagnostics omit query strings, including presigned credentials.
 
-Typed pool metadata failures additionally carry `pool_metadata_reason`, `pool_metadata_phase`, and `pool_metadata_since_unix_secs` in the PUT diagnostic. These describe the request's failure context: `read_unavailable` before write dispatch is retryable and does not by itself imply a latched write block. Use the current admin snapshot to distinguish that case from a persistent block. Public S3 errors remain sanitized `503 ServiceUnavailable` responses.
-
 Storage inventory emits a WARN event on the first failed probe and an INFO event on recovery, using `event="storage_info_probe"`. A recovery event confirms the RPC succeeded, not that every reported disk is healthy. Bucket metadata load/retry errors include the bucket and a bounded error code, so one failing bucket can be identified without dumping its metadata.
 
 No new environment variable, admin authorization action, or recovery command is required.

docs/operations/scanner-benchmark-runbook.md

@@ -125,6 +125,11 @@ and `metrics`. All metrics must be finite nonnegative numbers: `p99_ms`,
 `throughput_ops`, `rss_bytes`, `cpu_seconds`, `iops`, `rpc_count`,
 `cache_clone_bytes`, `encode_bytes`, `save_bytes`, `oldest_age_seconds`,
 `walk_objects`, `cold_walk_objects`, `healed_objects`, `errors`, and `requests`.
+The adapter also reports the measurement-window delta of
+`rustfs_heal_mainline_throttle_total{source="admin",result="delayed"}` as
+`heal_mainline_throttle_delayed`; a cumulative process-lifetime value is not a
+valid input.
+The clone, encode, and save byte fields are also deltas from the same window.
 Requests, throughput, and p99 must be positive; errors must be zero. Repair
 counts must match the manifest when background work is on. Keep underlying
 request samples, counter reset checks, profiler captures, and per-node telemetry
@@ -132,6 +137,19 @@ in the cell artifact directory; aggregate values alone do not establish their
 measurement provenance. Missing production instrumentation is a pending gate,
 not permission to report a fabricated zero.
 
+Each comparison records a `w22` section with clone, encode, and save bytes per
+walked object, clone/encode and save/encode byte ratios, and candidate changes.
+These are traffic amplification indicators, not allocation attribution or an
+fsync profile. The `running-heal` build comparison also records a `w10`
+section. `status=observed` requires sampled high foreground pressure, at least
+one admin pacing delay in the same window, and an improvement in either
+foreground p99 or throughput. `no_measured_benefit` means pacing ran but neither
+foreground metric improved; `pending` means the run did not prove that pacing
+engaged; `inconclusive` means ABBA repeatability failed. Baseline and candidate
+delay counts are both retained so an operator can reject unrelated or
+process-lifetime counter contamination. Correct repair oracles and the existing
+regression limits still apply in every case.
+
 For P2, `measure.convergence` contains booleans `writes_stopped`,
 `last_mutation_observed`, `first_complete_publication`; numeric
 `last_mutation_time`, `last_mutation_observed_time`, `writes_stopped_time`, `window_start`, `window_end`,

docs/testing/distributed-e2e.md

@@ -29,7 +29,7 @@ The expansion fixture is an all-current-binary fleet, so it initializes pool met
 - Object Lock COMPLIANCE, GOVERNANCE and bypass, legal hold, bucket default retention, and non-lock bucket rejection
 - Versioning, exact historical reads, delete-marker removal, and suspended null-version overwrite semantics
 - Bucket replication between two 4-node clusters, including metadata/tags and target-outage retry; hard quota admission and absence of rejected keys
-- Ready/live probes on every node, exact 4-server/16-disk inventory, realtime metrics on every node, and correlated audit-webhook delivery. The observability case runs at WARN and suspends two node processes to model nonresponsive peers: cached drives become unknown immediately, exact per-node HTTP PUT deltas expose sub-quorum failures, the local metadata snapshot does not invent a write latch, and resumed peers allow new writes and byte-identical reads from every node. This models stalled processes, not a physical network partition.
+- Ready/live probes on every node, exact 4-server/16-disk inventory, realtime metrics on every node, and correlated audit-webhook delivery
 - Pool expand, decommission, rebalance, checksum integrity, versioned and multipart data, and S3 during active movement
 - Bidirectional site-replication convergence plus enabled/synchronized peer state on both sites
 - A 24-worker mixed PUT/HEAD/GET/COPY/DELETE workload; concurrent PUT during active decommission

rustfs/src/admin/handlers/cluster_snapshot.rs

@@ -193,7 +193,6 @@ impl ClusterSnapshotView {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct ClusterPoolMetaWriteGateView {
-    pub state: &'static str,
     pub writes_ready: bool,
     pub write_blocked: bool,
     pub transaction_aborted: bool,
@@ -201,27 +200,11 @@ pub(crate) struct ClusterPoolMetaWriteGateView {
     pub identity_initialized: Option<bool>,
     pub identity_needs_repair: bool,
     pub cluster_epoch: Option<u64>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub reason: Option<&'static str>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub phase: Option<&'static str>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub since_unix_secs: Option<i64>,
 }
 
 impl From<ClusterPoolMetaWriteGateSnapshot> for ClusterPoolMetaWriteGateView {
     fn from(snapshot: ClusterPoolMetaWriteGateSnapshot) -> Self {
-        let state = if snapshot.check_timed_out {
-            "check_timeout"
-        } else if snapshot.writes_ready {
-            "writable"
-        } else if snapshot.write_blocked || snapshot.transaction_aborted {
-            "blocked"
-        } else {
-            "unavailable"
-        };
         Self {
-            state,
             writes_ready: snapshot.writes_ready,
             write_blocked: snapshot.write_blocked,
             transaction_aborted: snapshot.transaction_aborted,
@@ -229,9 +212,6 @@ impl From<ClusterPoolMetaWriteGateSnapshot> for ClusterPoolMetaWriteGateView {
             identity_initialized: snapshot.identity_initialized,
             identity_needs_repair: snapshot.identity_needs_repair,
             cluster_epoch: snapshot.cluster_epoch,
-            reason: snapshot.reason,
-            phase: snapshot.phase,
-            since_unix_secs: snapshot.since_unix_secs,
         }
     }
 }
@@ -706,7 +686,6 @@ fn summarize_storage_readiness(snapshot: &ClusterReadOnlySnapshot) -> Capability
         .filter_map(|reason| match reason {
             ReadinessDegradedReason::StorageQuorumUnavailable
             | ReadinessDegradedReason::PoolMetaWriteBlocked
-            | ReadinessDegradedReason::PoolMetadataCheckTimeout
             | ReadinessDegradedReason::StorageAndIamUnavailable
             | ReadinessDegradedReason::StorageAndLockUnavailable
             | ReadinessDegradedReason::StorageIamAndLockUnavailable => Some(reason.as_str()),
@@ -1018,9 +997,7 @@ fn summarize_named_capability_statuses<const N: usize>(
 
 #[cfg(test)]
 mod tests {
-    use super::{
-        ClusterMembershipView, ClusterPoolMetaWriteGateView, ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView,
-    };
+    use super::{ClusterMembershipView, ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView};
     use crate::admin::storage_api::cluster::CapabilityState;
     use crate::admin::storage_api::cluster::{CapabilityStatus, ObservabilitySnapshot, TopologySnapshot};
     use crate::admin::storage_api::cluster::{
@@ -1087,64 +1064,6 @@ mod tests {
         assert_eq!(value, serde_json::json!({ "snapshot": null }));
     }
 
-    #[test]
-    fn pool_meta_write_gate_details_are_additive_and_clear_when_not_blocked() {
-        #[derive(serde::Deserialize)]
-        #[serde(rename_all = "camelCase")]
-        struct LegacyGate {
-            writes_ready: bool,
-            write_blocked: bool,
-            transaction_aborted: bool,
-        }
-
-        let blocked = ClusterPoolMetaWriteGateSnapshot {
-            writes_ready: false,
-            transaction_aborted: true,
-            reason: Some("transaction_unknown"),
-            phase: Some("publication"),
-            since_unix_secs: Some(1_700_000_000),
-            ..Default::default()
-        };
-        let value = serde_json::to_value(ClusterPoolMetaWriteGateView::from(blocked)).unwrap();
-        assert_eq!(value["state"], "blocked");
-        assert_eq!(value["reason"], "transaction_unknown");
-        assert_eq!(value["phase"], "publication");
-        assert_eq!(value["sinceUnixSecs"], 1_700_000_000);
-        assert!(value.get("since_unix_secs").is_none());
-        assert!(value.get("operation").is_none());
-        assert!(value.get("source").is_none());
-        let legacy: LegacyGate = serde_json::from_value(value).unwrap();
-        assert!(!legacy.writes_ready);
-        assert!(!legacy.write_blocked);
-        assert!(legacy.transaction_aborted);
-
-        for (snapshot, state) in [
-            (ClusterPoolMetaWriteGateSnapshot::default(), "writable"),
-            (
-                ClusterPoolMetaWriteGateSnapshot {
-                    writes_ready: false,
-                    check_timed_out: true,
-                    ..Default::default()
-                },
-                "check_timeout",
-            ),
-            (
-                ClusterPoolMetaWriteGateSnapshot {
-                    writes_ready: false,
-                    ..Default::default()
-                },
-                "unavailable",
-            ),
-        ] {
-            let value = serde_json::to_value(ClusterPoolMetaWriteGateView::from(snapshot)).unwrap();
-            assert_eq!(value["state"], state);
-            assert_eq!(value["writesReady"], state == "writable");
-            for field in ["reason", "phase", "sinceUnixSecs"] {
-                assert!(value.get(field).is_none(), "{state} must not retain {field}: {value}");
-            }
-        }
-    }
-
     #[tokio::test]
     async fn cluster_snapshot_discovery_reports_path_without_snapshot() {
         let response = super::build_cluster_snapshot_discovery_response().await;

rustfs/src/app/object/put.rs

@@ -1995,9 +1995,6 @@ impl DefaultObjectUsecase {
                                 storage_error_code = ?diagnostic.storage_code,
                                 io_error_kind = ?diagnostic.io_kind,
                                 rpc_error_code = ?diagnostic.rpc_code,
-                                pool_metadata_reason = diagnostic.pool_metadata.map(|context| context.reason),
-                                pool_metadata_phase = diagnostic.pool_metadata.map(|context| context.phase),
-                                pool_metadata_since_unix_secs = diagnostic.pool_metadata.map(|context| context.since_unix_secs),
                                 source_chain_truncated = diagnostic.truncated,
                                 suppressed_errors,
                                 "PutObject store write returned"

rustfs/src/cluster_snapshot.rs

@@ -108,32 +108,24 @@ pub struct ClusterListingDiagnosticsSnapshot {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ClusterPoolMetaWriteGateSnapshot {
     pub writes_ready: bool,
-    pub check_timed_out: bool,
     pub write_blocked: bool,
     pub transaction_aborted: bool,
     pub pool_meta_absent: bool,
     pub identity_initialized: Option<bool>,
     pub identity_needs_repair: bool,
     pub cluster_epoch: Option<u64>,
-    pub reason: Option<&'static str>,
-    pub phase: Option<&'static str>,
-    pub since_unix_secs: Option<i64>,
 }
 
 impl Default for ClusterPoolMetaWriteGateSnapshot {
     fn default() -> Self {
         Self {
             writes_ready: true,
-            check_timed_out: false,
             write_blocked: false,
             transaction_aborted: false,
             pool_meta_absent: false,
             identity_initialized: None,
             identity_needs_repair: false,
             cluster_epoch: None,
-            reason: None,
-            phase: None,
-            since_unix_secs: None,
         }
     }
 }
@@ -230,22 +222,15 @@ async fn current_pool_meta_write_gate_snapshot() -> ClusterPoolMetaWriteGateSnap
             let status = store.pool_meta_write_gate_status().await;
             ClusterPoolMetaWriteGateSnapshot {
                 writes_ready: status.writes_ready,
-                check_timed_out: status.check_timed_out,
                 write_blocked: status.write_blocked,
                 transaction_aborted: status.transaction_aborted,
                 pool_meta_absent: status.pool_meta_absent,
                 identity_initialized: status.identity_initialized,
                 identity_needs_repair: status.identity_needs_repair,
                 cluster_epoch: status.cluster_epoch,
-                reason: status.reason,
-                phase: status.phase,
-                since_unix_secs: status.since_unix_secs,
             }
         }
-        None => ClusterPoolMetaWriteGateSnapshot {
-            writes_ready: false,
-            ..Default::default()
-        },
+        None => ClusterPoolMetaWriteGateSnapshot::default(),
     }
 }
 

rustfs/src/error.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::storage_api::error::contract::{StorageErrorCode, range::HTTPRangeError};
-use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError};
+use crate::storage_api::error::{QuotaError, StorageError};
 use http::StatusCode;
 use rustfs_kms::KmsUnavailableError;
 use s3s::{S3Error, S3ErrorCode};
@@ -89,28 +89,9 @@ pub(crate) struct ApiErrorDiagnostic {
     pub storage_code: Option<StorageErrorCode>,
     pub io_kind: Option<std::io::ErrorKind>,
     pub rpc_code: Option<tonic::Code>,
-    pub pool_metadata: Option<PoolMetadataDiagnostic>,
     pub truncated: bool,
 }
 
-/// Safe projection of local metadata failure context, without operation text or sources.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub(crate) struct PoolMetadataDiagnostic {
-    pub reason: &'static str,
-    pub phase: &'static str,
-    pub since_unix_secs: i64,
-}
-
-impl From<&PoolMetadataError> for PoolMetadataDiagnostic {
-    fn from(context: &PoolMetadataError) -> Self {
-        Self {
-            reason: context.kind.as_str(),
-            phase: context.phase,
-            since_unix_secs: context.since.unix_timestamp(),
-        }
-    }
-}
-
 impl ApiError {
     pub(crate) fn diagnostic(&self) -> ApiErrorDiagnostic {
         let mut diagnostic = ApiErrorDiagnostic::default();
@@ -125,11 +106,6 @@ impl ApiError {
             if let Some(status) = error.downcast_ref::<tonic::Status>() {
                 diagnostic.rpc_code = Some(status.code());
             }
-            if diagnostic.pool_metadata.is_none()
-                && let Some(context) = error.downcast_ref::<PoolMetadataError>()
-            {
-                diagnostic.pool_metadata = Some(context.into());
-            }
             current = if let Some(io) = error.downcast_ref::<std::io::Error>() {
                 diagnostic.io_kind = Some(io.kind());
                 // io::Error::source can skip the wrapped error itself.
@@ -978,7 +954,6 @@ mod tests {
     #[test]
     fn pool_metadata_failures_map_to_503_and_preserve_typed_private_context() {
         use crate::storage_api::error::{PoolMetadataError, PoolMetadataFailure};
-        let since = time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed failure time");
         for kind in [
             PoolMetadataFailure::ReadUnavailable,
             PoolMetadataFailure::RecoveryRequired,
@@ -987,9 +962,9 @@ mod tests {
         ] {
             let error = StorageError::other(PoolMetadataError {
                 kind,
-                operation: "private operation path".to_owned(),
+                operation: "pool metadata test".to_owned(),
                 phase: "prepare_cas",
-                since,
+                since: time::OffsetDateTime::now_utc(),
                 source: Some(std::sync::Arc::new(StorageError::other("private disk failure"))),
             });
             let error = StorageError::Io(std::io::Error::new(std::io::ErrorKind::TimedOut, error));
@@ -1002,18 +977,6 @@ mod tests {
             assert!(!api.message.contains("private"));
             let source = api.source.as_ref().unwrap().downcast_ref::<StorageError>().unwrap();
             assert_eq!(source.pool_metadata_failure().unwrap().kind, kind);
-            let diagnostic = api.diagnostic();
-            assert_eq!(
-                diagnostic.pool_metadata,
-                Some(PoolMetadataDiagnostic {
-                    reason: kind.as_str(),
-                    phase: "prepare_cas",
-                    since_unix_secs: since.unix_timestamp(),
-                })
-            );
-            assert!(!diagnostic.truncated);
-            assert!(!format!("{diagnostic:?}").contains("private"));
-            assert_eq!(api.diagnostic(), diagnostic, "inspection must preserve the original failure time");
         }
     }
 

rustfs/src/server/readiness.rs

@@ -16,7 +16,6 @@ use crate::server::runtime_sources;
 use crate::server::{ServiceState, ServiceStateManager};
 use crate::server::{has_path_prefix, is_table_catalog_path};
 use crate::storage_api::cluster::control_plane::ClusterControlPlane;
-use crate::storage_api::error::StorageError;
 use crate::storage_api::server::readiness::contract::admin::StorageAdminApi;
 use crate::storage_api::server::readiness::{Endpoint, EndpointServerPools, is_dist_erasure};
 #[cfg(test)]
@@ -263,29 +262,7 @@ struct StorageReadinessCacheEntry {
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 struct StorageWriteReadinessStatus {
     ready: bool,
-    pool_metadata_reason: Option<ReadinessDegradedReason>,
-}
-
-fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWriteReadinessStatus {
-    match result {
-        Ok(()) => StorageWriteReadinessStatus {
-            ready: true,
-            pool_metadata_reason: None,
-        },
-        Err(error) => {
-            let pool_metadata_reason = if error.pool_metadata_failure().is_some() {
-                Some(ReadinessDegradedReason::PoolMetaWriteBlocked)
-            } else if matches!(error, StorageError::Timeout) {
-                Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)
-            } else {
-                None
-            };
-            StorageWriteReadinessStatus {
-                ready: false,
-                pool_metadata_reason,
-            }
-        }
-    }
+    pool_meta_write_blocked: bool,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -700,12 +677,14 @@ fn degraded_reasons(readiness: DependencyReadiness) -> Vec<ReadinessDegradedReas
 
 fn degraded_reasons_with_pool_meta_status(
     readiness: DependencyReadiness,
-    pool_metadata_reason: Option<ReadinessDegradedReason>,
+    pool_meta_write_blocked: bool,
 ) -> Vec<ReadinessDegradedReason> {
     let mut reasons = degraded_reasons(readiness);
-    if let Some(reason) = pool_metadata_reason {
+    if pool_meta_write_blocked {
         reasons.retain(|reason| *reason != ReadinessDegradedReason::StorageQuorumUnavailable);
-        reasons.insert(0, reason);
+        if !reasons.contains(&ReadinessDegradedReason::PoolMetaWriteBlocked) {
+            reasons.insert(0, ReadinessDegradedReason::PoolMetaWriteBlocked);
+        }
     }
     reasons
 }
@@ -738,7 +717,7 @@ fn dependency_readiness_report_from_write_status(
     storage: StorageWriteReadinessStatus,
 ) -> DependencyReadinessReport {
     DependencyReadinessReport {
-        degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_metadata_reason),
+        degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_meta_write_blocked),
         readiness,
     }
 }
@@ -867,7 +846,11 @@ async fn collect_lock_quorum_status() -> LockQuorumStatus {
 
 async fn node_pool_meta_write_readiness() -> StorageWriteReadinessStatus {
     if let Some(store) = runtime_sources::current_object_store_handle() {
-        return pool_metadata_write_readiness(store.pool_meta_write_status().await);
+        let ready = store.pool_meta_writes_ready().await;
+        return StorageWriteReadinessStatus {
+            ready,
+            pool_meta_write_blocked: !ready,
+        };
     }
 
     StorageWriteReadinessStatus::default()
@@ -875,14 +858,16 @@ async fn node_pool_meta_write_readiness() -> StorageWriteReadinessStatus {
 
 async fn collect_storage_write_readiness_uncached() -> StorageWriteReadinessStatus {
     if let Some(store) = runtime_sources::current_object_store_handle() {
-        let metadata = pool_metadata_write_readiness(store.pool_meta_write_status().await);
-        if !metadata.ready {
-            return metadata;
+        if !store.pool_meta_writes_ready().await {
+            return StorageWriteReadinessStatus {
+                ready: false,
+                pool_meta_write_blocked: true,
+            };
         }
         let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
         StorageWriteReadinessStatus {
             ready: storage_ready_from_runtime_state(&storage_info),
-            pool_metadata_reason: None,
+            pool_meta_write_blocked: false,
         }
     } else {
         StorageWriteReadinessStatus::default()
@@ -1915,48 +1900,6 @@ mod tests {
         );
     }
 
-    fn blocked_pool_metadata_status() -> StorageWriteReadinessStatus {
-        use crate::storage_api::error::{PoolMetadataError, PoolMetadataFailure};
-        pool_metadata_write_readiness(Err(StorageError::other(PoolMetadataError {
-            kind: PoolMetadataFailure::TransactionUnknown,
-            operation: "private operation".to_owned(),
-            phase: "prepare_cas",
-            since: time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed block time"),
-            source: Some(Arc::new(StorageError::other("private replica failure"))),
-        })))
-    }
-
-    #[test]
-    fn pool_metadata_observation_distinguishes_block_timeout_and_unavailable() {
-        let blocked = blocked_pool_metadata_status();
-        assert!(!blocked.ready);
-        assert_eq!(blocked.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetaWriteBlocked));
-        assert!(!format!("{blocked:?}").contains("private"));
-        let writable = pool_metadata_write_readiness(Ok(()));
-        assert!(writable.ready);
-        assert_eq!(writable.pool_metadata_reason, None);
-        let timed_out = pool_metadata_write_readiness(Err(StorageError::Timeout));
-        assert!(!timed_out.ready);
-        assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
-        let unavailable = pool_metadata_write_readiness(Err(StorageError::other("pool metadata writes remain blocked")));
-        assert!(!unavailable.ready);
-        assert_eq!(
-            unavailable.pool_metadata_reason, None,
-            "error text alone must not be interpreted as a typed write block"
-        );
-
-        let readiness = DependencyReadiness {
-            storage_ready: false,
-            iam_ready: true,
-            lock_quorum_ready: true,
-            peer_health_ready: true,
-        };
-        let report = dependency_readiness_report_from_write_status(readiness, timed_out);
-        assert!(!report.readiness.storage_ready, "inspection timeout must remain fail-closed");
-        assert_eq!(report.degraded_reasons, vec![ReadinessDegradedReason::PoolMetadataCheckTimeout]);
-        assert_eq!(report.degraded_reasons[0].as_str(), "pool_metadata_check_timeout");
-    }
-
     #[test]
     fn degraded_reasons_report_pool_meta_write_blocked() {
         let readiness = DependencyReadiness {
@@ -1967,7 +1910,7 @@ mod tests {
         };
 
         assert_eq!(
-            degraded_reasons_with_pool_meta_status(readiness, blocked_pool_metadata_status().pool_metadata_reason),
+            degraded_reasons_with_pool_meta_status(readiness, true),
             vec![ReadinessDegradedReason::PoolMetaWriteBlocked]
         );
     }
@@ -1982,7 +1925,7 @@ mod tests {
         };
 
         assert_eq!(
-            degraded_reasons_with_pool_meta_status(readiness, blocked_pool_metadata_status().pool_metadata_reason),
+            degraded_reasons_with_pool_meta_status(readiness, true),
             vec![
                 ReadinessDegradedReason::PoolMetaWriteBlocked,
                 ReadinessDegradedReason::StorageAndLockUnavailable,

rustfs/src/shared_types.rs

@@ -44,7 +44,6 @@ pub enum ReadinessDegradedReason {
     ObjectReadStalled,
     ObjectWriteStalled,
     PoolMetaWriteBlocked,
-    PoolMetadataCheckTimeout,
     ClusterHealthTimeout,
     PeerHealthUnavailable,
     StartupFinalizationPending,
@@ -64,7 +63,6 @@ impl ReadinessDegradedReason {
             ReadinessDegradedReason::ObjectReadStalled => "object_read_stalled",
             ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled",
             ReadinessDegradedReason::PoolMetaWriteBlocked => "pool_meta_write_blocked",
-            ReadinessDegradedReason::PoolMetadataCheckTimeout => "pool_metadata_check_timeout",
             ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout",
             ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable",
             ReadinessDegradedReason::StartupFinalizationPending => "startup_finalization_pending",

rustfs/src/storage/storage_api.rs

@@ -476,12 +476,11 @@ pub(crate) mod ecstore_disk {
 }
 
 pub(crate) mod ecstore_error {
-    #[cfg(test)]
-    pub(crate) use rustfs_ecstore::api::error::PoolMetadataFailure;
     pub(crate) use rustfs_ecstore::api::error::{
-        Error, PoolMetadataError, Result, StorageError, is_err_bucket_not_found, is_err_object_not_found,
-        is_err_version_not_found,
+        Error, Result, StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
     };
+    #[cfg(test)]
+    pub(crate) use rustfs_ecstore::api::error::{PoolMetadataError, PoolMetadataFailure};
 }
 
 pub(crate) mod ecstore_event {

rustfs/src/storage_api.rs

@@ -83,9 +83,8 @@ pub(crate) mod error {
         }
     }
 
-    pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataError;
     #[cfg(test)]
-    pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataFailure;
+    pub(crate) use crate::storage::storage_api::ecstore_error::{PoolMetadataError, PoolMetadataFailure};
     pub(crate) use crate::storage::storage_api::{QuotaError, StorageError};
 }
 

scripts/scanner_abba.py

@@ -23,6 +23,7 @@ METRICS = (
     "cache_clone_bytes", "encode_bytes", "save_bytes", "oldest_age_seconds",
     "walk_objects", "cold_walk_objects", "healed_objects", "errors", "requests",
     "foreground_pressure_samples", "foreground_pressure_high_samples",
+    "heal_mainline_throttle_delayed",
     "heal_lock_wait_p99_ms", "heal_attempts", "heal_attempt_failures",
     "heal_retry_attempts",
 )
@@ -59,6 +60,12 @@ def relative_change(current, baseline, name):
     return ratio(current, baseline, name) - Decimal("1")
 
 
+def relative_change_or_none(current, baseline, name):
+    if decimal_number(baseline, f"{name} baseline") == 0:
+        return None
+    return relative_change(current, baseline, name)
+
+
 def repeatability_change(first, second, name):
     first = decimal_number(first, name)
     second = decimal_number(second, name)
@@ -283,6 +290,62 @@ def pressure_high_ratio(metrics):
                  metrics["foreground_pressure_samples"], "foreground pressure high samples")
 
 
+def scanner_cache_cost(metrics):
+    walked = decimal_number(metrics["walk_objects"], "walk_objects")
+    encoded = decimal_number(metrics["encode_bytes"], "encode_bytes")
+    return {
+        "clone_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["cache_clone_bytes"], walked, "clone bytes per walk object")),
+        "encode_bytes_per_walk_object": None if walked == 0 else float(ratio(encoded, walked, "encode bytes per walk object")),
+        "save_bytes_per_walk_object": None if walked == 0 else float(ratio(metrics["save_bytes"], walked, "save bytes per walk object")),
+        "clone_to_encode_byte_ratio": None if encoded == 0 else float(ratio(metrics["cache_clone_bytes"], encoded, "clone to encode bytes")),
+        "save_to_encode_byte_amplification": None if encoded == 0 else float(ratio(metrics["save_bytes"], encoded, "save to encode bytes")),
+    }
+
+
+def scanner_cache_cost_change(candidate, baseline):
+    changes = {}
+    for key in ("cache_clone_bytes", "encode_bytes", "save_bytes"):
+        change = relative_change_or_none(candidate[key], baseline[key], key)
+        changes[f"{key}_change"] = None if change is None else float(change)
+    return changes
+
+
+def running_heal_pacing(group, baseline, candidate, p99, throughput, noisy):
+    if group[0]["scenario"] != "running-heal" or group[0]["comparison"] != "build":
+        return None
+    baseline_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[0], group[3])) / 2
+    candidate_seconds = sum(decimal_number(cell["result"]["elapsed_seconds"], "elapsed_seconds") for cell in (group[1], group[2])) / 2
+    baseline_rate = ratio(baseline["heal_attempts"], baseline_seconds, "baseline heal attempt rate")
+    candidate_rate = ratio(candidate["heal_attempts"], candidate_seconds, "candidate heal attempt rate")
+    rate_change = relative_change_or_none(candidate_rate, baseline_rate, "heal attempt rate")
+    candidate_high_ratio = pressure_high_ratio(candidate)
+    baseline_delayed = decimal_number(baseline["heal_mainline_throttle_delayed"], "baseline pacing delays")
+    delayed = decimal_number(candidate["heal_mainline_throttle_delayed"], "candidate pacing delays")
+    pacing_observed = candidate_high_ratio > 0 and delayed > 0
+    foreground_improved = p99 < 0 or throughput > 0
+    status = (
+        "inconclusive"
+        if noisy
+        else "observed"
+        if pacing_observed and foreground_improved
+        else "no_measured_benefit"
+        if pacing_observed
+        else "pending"
+    )
+    return {
+        "status": status,
+        "pacing_observed": pacing_observed,
+        "candidate_pressure_high_ratio": float(candidate_high_ratio),
+        "baseline_delay_events": float(baseline_delayed),
+        "candidate_delay_events": float(delayed),
+        "baseline_heal_attempts_per_second": float(baseline_rate),
+        "candidate_heal_attempts_per_second": float(candidate_rate),
+        "heal_attempt_rate_change": None if rate_change is None else float(rate_change),
+        "foreground_p99_change": float(p99),
+        "foreground_throughput_change": float(throughput),
+    }
+
+
 def convergence(result):
     window = result.get("convergence")
     if not window or window.get("writes_stopped") is not True or window.get("last_mutation_observed") is not True or window.get("first_complete_publication") is not True:
@@ -342,6 +405,7 @@ def evaluate(cells):
         candidate_attempt_costs = [
             value for cell, value in zip(group, attempt_costs) if cell["leg"].startswith("B") and value is not None
         ]
+        w10 = running_heal_pacing(group, a, b, p99, throughput, noise)
         inconclusive |= noise or p2_pending
         if not noise and not passed:
             failed = True
@@ -352,6 +416,12 @@ def evaluate(cells):
                             "thresholds": {key: float(value) for key, value in thresholds.items()},
                             "p1": p1, "p2_max_work_multiple": float(P2_WORK_MULTIPLE_LIMIT),
                             "p2_post_stop_work_multiples": p2_report,
+                            "w22": {
+                                "baseline": scanner_cache_cost(a),
+                                "candidate": scanner_cache_cost(b),
+                                "candidate_vs_baseline": scanner_cache_cost_change(b, a),
+                            },
+                            "w10": w10,
                             "w10_w11": {
                                 "foreground_pressure_high_sample_ratios": [
                                     float(pressure_high_ratio(cell["result"]["metrics"])) for cell in group

scripts/test_scanner_abba.py

@@ -94,6 +94,8 @@ def fake_adapter():
                 result["metrics"].update(walk_objects=100, cold_walk_objects=0)
         elif fault == "missing-metric":
             del result["metrics"]["save_bytes"]
+        elif fault == "missing-pacing-metric":
+            del result["metrics"]["heal_mainline_throttle_delayed"]
         elif fault == "incomplete-repair":
             result["metrics"]["healed_objects"] = 0
         elif fault == "zero-pressure-samples":
@@ -102,6 +104,12 @@ def fake_adapter():
             result["metrics"]["foreground_pressure_high_samples"] = result["metrics"]["foreground_pressure_samples"] + 1
         elif fault == "attempt-accounting":
             result["metrics"]["heal_attempt_failures"] = result["metrics"]["heal_attempts"] + 1
+        elif fault == "pacing-benefit" and request["scenario"] == "running-heal" \
+                and request["comparison"] == "build" and request["leg"].startswith("B"):
+            result["metrics"].update(p99_ms=9, heal_mainline_throttle_delayed=5)
+        elif fault == "pacing-pending" and request["scenario"] == "running-heal" \
+                and request["comparison"] == "build" and request["leg"].startswith("B"):
+            result["metrics"]["heal_mainline_throttle_delayed"] = 0
     harness.write_json(Path(output_path), result)
     return 0
 
@@ -278,6 +286,31 @@ class ScannerAbbaTest(unittest.TestCase):
                     self.assertEqual({r["leg"] for r in legs}, set(harness.LEGS))
         self.assertTrue(all(c["p2_max_work_multiple"] == 1.2 for c in report["comparisons"]))
         for comparison in report["comparisons"]:
+            w22 = comparison["w22"]
+            self.assertEqual(w22["baseline"]["save_to_encode_byte_amplification"], 1.0)
+            self.assertEqual(w22["candidate"]["clone_to_encode_byte_ratio"], 1.0)
+            self.assertEqual(
+                w22["candidate_vs_baseline"],
+                {"cache_clone_bytes_change": 0.0, "encode_bytes_change": 0.0, "save_bytes_change": 0.0},
+            )
+            if comparison["scenario"] == "running-heal" and comparison["comparison"] == "build":
+                self.assertEqual(
+                    comparison["w10"],
+                    {
+                        "status": "no_measured_benefit",
+                        "pacing_observed": True,
+                        "candidate_pressure_high_ratio": 1.0,
+                        "baseline_delay_events": 10.0,
+                        "candidate_delay_events": 10.0,
+                        "baseline_heal_attempts_per_second": 10.0,
+                        "candidate_heal_attempts_per_second": 10.0,
+                        "heal_attempt_rate_change": 0.0,
+                        "foreground_p99_change": 0.0,
+                        "foreground_throughput_change": 0.0,
+                    },
+                )
+            else:
+                self.assertIsNone(comparison["w10"])
             w10_w11 = comparison["w10_w11"]
             self.assertEqual(w10_w11["foreground_pressure_high_sample_ratios"], [1.0, 1.0, 1.0, 1.0])
             self.assertEqual(w10_w11["heal_lock_wait_p99_ms"], [10, 10, 10, 10])
@@ -288,7 +321,8 @@ class ScannerAbbaTest(unittest.TestCase):
     def test_fail_closed_adapter_and_data_errors(self):
         for fault in ("measure-exit", "oracle-exit", "missing-oracle", "oracle-mismatch", "zero-samples",
                       "zero-requests", "request-errors", "load-drift", "missing-metric", "incomplete-repair",
-                      "zero-pressure-samples", "pressure-sample-order", "attempt-accounting"):
+                      "zero-pressure-samples", "pressure-sample-order", "attempt-accounting",
+                      "missing-pacing-metric"):
             with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
                 self.root = Path(directory)
                 with self.assertRaises((ValueError, OSError, subprocess.SubprocessError)):
@@ -302,6 +336,36 @@ class ScannerAbbaTest(unittest.TestCase):
             self.assertEqual(self.run_harness("noise"), 3)
         self.assertEqual(harness.read_json(self.root / "out/report.json")["status"], "inconclusive")
 
+    def test_noisy_running_heal_does_not_claim_pacing_benefit(self):
+        with patch.object(harness, "SCENARIOS", ("running-heal",)):
+            self.assertEqual(self.run_harness("noise"), 3)
+        comparisons = harness.read_json(self.root / "out/report.json")["comparisons"]
+        build = next(comparison for comparison in comparisons if comparison["comparison"] == "build")
+        self.assertEqual(build["w10"]["status"], "inconclusive")
+
+    def test_idle_cache_window_reports_unavailable_ratios(self):
+        metrics = dict.fromkeys(harness.METRICS, 0)
+        self.assertEqual(
+            harness.scanner_cache_cost(metrics),
+            {
+                "clone_bytes_per_walk_object": None,
+                "encode_bytes_per_walk_object": None,
+                "save_bytes_per_walk_object": None,
+                "clone_to_encode_byte_ratio": None,
+                "save_to_encode_byte_amplification": None,
+            },
+        )
+
+    def test_running_heal_pacing_status_requires_engagement_and_benefit(self):
+        for fault, expected in (("pacing-benefit", "observed"), ("pacing-pending", "pending")):
+            with self.subTest(fault=fault), tempfile.TemporaryDirectory() as directory:
+                self.root = Path(directory)
+                with patch.object(harness, "SCENARIOS", ("running-heal",)):
+                    self.assertEqual(self.run_harness(fault), 0)
+                comparisons = harness.read_json(self.root / "out/report.json")["comparisons"]
+                build = next(comparison for comparison in comparisons if comparison["comparison"] == "build")
+                self.assertEqual(build["w10"]["status"], expected)
+
     def test_missing_first_publication_is_inconclusive(self):
         with patch.object(harness, "SCENARIOS", ("cold-hot",)):
             self.assertEqual(self.run_harness("no-publication"), 3)