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};
+use crate::common::{RustFSTestClusterEnvironment, init_logging, local_http_client, signal_process};
 use aws_sdk_s3::primitives::ByteStream;
 use http::header::HOST;
 use reqwest::StatusCode;
@@ -22,7 +22,6 @@ 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;
 
@@ -82,15 +81,6 @@ 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,6 +264,15 @@ 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,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-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 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 aws_sdk_s3::operation::RequestId;
 use aws_sdk_s3::primitives::ByteStream;
 use bytes::Bytes;
@@ -24,7 +26,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::RealtimeMetrics;
+use rustfs_madmin::metrics::{HttpMetrics, RealtimeMetrics};
 use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
 use serde_json::Value;
 use std::convert::Infallible;
@@ -126,6 +128,7 @@ 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()),
@@ -231,6 +234,186 @@ 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,18 +10682,30 @@ 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: !write_state.write_blocked && !transaction_aborted,
+            writes_ready,
+            check_timed_out: false,
             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()),
         }
     }
 
@@ -19002,6 +19014,12 @@ 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();
@@ -19018,17 +19036,39 @@ 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();
@@ -19048,6 +19088,7 @@ 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();
@@ -19066,6 +19107,12 @@ 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,24 +530,32 @@ 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,8 +41,12 @@ 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_metadata_blocked`. Metadata save-gate inspection is bounded to 100 ms;
+  `pool_meta_write_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_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.
+- 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).
 
 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,10 +45,23 @@ 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 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.
+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.
 
 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

@@ -171,22 +171,6 @@ performance acceptance gate. Run the fake-adapter self-tests with:
 scripts/test_scanner_validation_harness.sh
 ```
 
-For CI summaries, PR evidence tables, and operator handoff, collapse the raw
-matrix into a quiet one-line verdict plus durable JSON/Markdown artifacts:
-
-```bash
-scripts/summarize_scanner_heal_perf.py \
-  --abba-dir /path/to/new-artifacts \
-  --cache-cost-log /path/to/cache-cost-profile.log \
-  --json-out /path/to/new-artifacts/perf-summary.json \
-  --markdown-out /path/to/new-artifacts/perf-summary.md
-```
-
-The command prints only `PASS scanner_heal_perf ...` for measured passing ABBA
-evidence, otherwise `FAIL scanner_heal_perf ...`. The JSON and Markdown outputs
-carry the key p99/throughput/P1/P2/cache-cost fields and artifact provenance
-hashes; raw per-cell logs remain in the original artifact tree for audit.
-
 They cover the complete 120-cell schedule, data isolation, missing builds and
 oracles, zero samples/requests, swallowed request errors, offered-load drift,
 incomplete repairs, missing metrics, noise, and P1/P2/p99 regressions. A real

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
+- 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.
 - 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,6 +193,7 @@ 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,
@@ -200,11 +201,27 @@ 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,
@@ -212,6 +229,9 @@ 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,
         }
     }
 }
@@ -686,6 +706,7 @@ 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()),
@@ -997,7 +1018,9 @@ fn summarize_named_capability_statuses<const N: usize>(
 
 #[cfg(test)]
 mod tests {
-    use super::{ClusterMembershipView, ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView};
+    use super::{
+        ClusterMembershipView, ClusterPoolMetaWriteGateView, 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::{
@@ -1064,6 +1087,64 @@ 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,6 +1995,9 @@ 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,24 +108,32 @@ 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,
         }
     }
 }
@@ -222,15 +230,22 @@ 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::default(),
+        None => ClusterPoolMetaWriteGateSnapshot {
+            writes_ready: false,
+            ..Default::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::{QuotaError, StorageError};
+use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError};
 use http::StatusCode;
 use rustfs_kms::KmsUnavailableError;
 use s3s::{S3Error, S3ErrorCode};
@@ -89,9 +89,28 @@ 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();
@@ -106,6 +125,11 @@ 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.
@@ -954,6 +978,7 @@ 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,
@@ -962,9 +987,9 @@ mod tests {
         ] {
             let error = StorageError::other(PoolMetadataError {
                 kind,
-                operation: "pool metadata test".to_owned(),
+                operation: "private operation path".to_owned(),
                 phase: "prepare_cas",
-                since: time::OffsetDateTime::now_utc(),
+                since,
                 source: Some(std::sync::Arc::new(StorageError::other("private disk failure"))),
             });
             let error = StorageError::Io(std::io::Error::new(std::io::ErrorKind::TimedOut, error));
@@ -977,6 +1002,18 @@ 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,6 +16,7 @@ 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)]
@@ -262,7 +263,29 @@ struct StorageReadinessCacheEntry {
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 struct StorageWriteReadinessStatus {
     ready: bool,
-    pool_meta_write_blocked: 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,
+            }
+        }
+    }
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -677,14 +700,12 @@ fn degraded_reasons(readiness: DependencyReadiness) -> Vec<ReadinessDegradedReas
 
 fn degraded_reasons_with_pool_meta_status(
     readiness: DependencyReadiness,
-    pool_meta_write_blocked: bool,
+    pool_metadata_reason: Option<ReadinessDegradedReason>,
 ) -> Vec<ReadinessDegradedReason> {
     let mut reasons = degraded_reasons(readiness);
-    if pool_meta_write_blocked {
+    if let Some(reason) = pool_metadata_reason {
         reasons.retain(|reason| *reason != ReadinessDegradedReason::StorageQuorumUnavailable);
-        if !reasons.contains(&ReadinessDegradedReason::PoolMetaWriteBlocked) {
-            reasons.insert(0, ReadinessDegradedReason::PoolMetaWriteBlocked);
-        }
+        reasons.insert(0, reason);
     }
     reasons
 }
@@ -717,7 +738,7 @@ fn dependency_readiness_report_from_write_status(
     storage: StorageWriteReadinessStatus,
 ) -> DependencyReadinessReport {
     DependencyReadinessReport {
-        degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_meta_write_blocked),
+        degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_metadata_reason),
         readiness,
     }
 }
@@ -846,11 +867,7 @@ 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() {
-        let ready = store.pool_meta_writes_ready().await;
-        return StorageWriteReadinessStatus {
-            ready,
-            pool_meta_write_blocked: !ready,
-        };
+        return pool_metadata_write_readiness(store.pool_meta_write_status().await);
     }
 
     StorageWriteReadinessStatus::default()
@@ -858,16 +875,14 @@ 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() {
-        if !store.pool_meta_writes_ready().await {
-            return StorageWriteReadinessStatus {
-                ready: false,
-                pool_meta_write_blocked: true,
-            };
+        let metadata = pool_metadata_write_readiness(store.pool_meta_write_status().await);
+        if !metadata.ready {
+            return metadata;
         }
         let storage_info = StorageAdminApi::storage_info(store.as_ref()).await;
         StorageWriteReadinessStatus {
             ready: storage_ready_from_runtime_state(&storage_info),
-            pool_meta_write_blocked: false,
+            pool_metadata_reason: None,
         }
     } else {
         StorageWriteReadinessStatus::default()
@@ -1900,6 +1915,48 @@ 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 {
@@ -1910,7 +1967,7 @@ mod tests {
         };
 
         assert_eq!(
-            degraded_reasons_with_pool_meta_status(readiness, true),
+            degraded_reasons_with_pool_meta_status(readiness, blocked_pool_metadata_status().pool_metadata_reason),
             vec![ReadinessDegradedReason::PoolMetaWriteBlocked]
         );
     }
@@ -1925,7 +1982,7 @@ mod tests {
         };
 
         assert_eq!(
-            degraded_reasons_with_pool_meta_status(readiness, true),
+            degraded_reasons_with_pool_meta_status(readiness, blocked_pool_metadata_status().pool_metadata_reason),
             vec![
                 ReadinessDegradedReason::PoolMetaWriteBlocked,
                 ReadinessDegradedReason::StorageAndLockUnavailable,

rustfs/src/shared_types.rs

@@ -44,6 +44,7 @@ pub enum ReadinessDegradedReason {
     ObjectReadStalled,
     ObjectWriteStalled,
     PoolMetaWriteBlocked,
+    PoolMetadataCheckTimeout,
     ClusterHealthTimeout,
     PeerHealthUnavailable,
     StartupFinalizationPending,
@@ -63,6 +64,7 @@ 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,11 +476,12 @@ pub(crate) mod ecstore_disk {
 }
 
 pub(crate) mod ecstore_error {
-    pub(crate) use rustfs_ecstore::api::error::{
-        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) 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,
+    };
 }
 
 pub(crate) mod ecstore_event {

rustfs/src/storage_api.rs

@@ -83,8 +83,9 @@ 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::{PoolMetadataError, PoolMetadataFailure};
+    pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataFailure;
     pub(crate) use crate::storage::storage_api::{QuotaError, StorageError};
 }
 

scripts/summarize_scanner_heal_perf.py

@@ -1,311 +0,0 @@
-#!/usr/bin/env python3
-"""Summarize Scanner/Heal ABBA and cache-cost profile artifacts quietly."""
-
-from __future__ import annotations
-
-import argparse
-from collections import Counter
-from decimal import Decimal
-import hashlib
-import json
-from pathlib import Path
-import sys
-from typing import Any
-
-MAX_JSON_BYTES = 1024 * 1024
-CACHE_COST_PREFIX = "CACHE_COST "
-PASS_STATES = {"pass"}
-FAIL_STATES = {"fail", "failed"}
-
-
-def require(condition: bool, message: str) -> None:
-    if not condition:
-        raise ValueError(message)
-
-
-def read_json(path: Path) -> dict[str, Any]:
-    require(path.is_file(), f"missing JSON artifact: {path}")
-    require(path.stat().st_size <= MAX_JSON_BYTES, f"oversized JSON artifact: {path}")
-    with path.open(encoding="utf-8") as stream:
-        value = json.load(stream)
-    require(isinstance(value, dict), f"expected JSON object: {path}")
-    return value
-
-
-def write_json(path: Path, value: dict[str, Any]) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8")
-
-
-def digest(path: Path) -> str:
-    with path.open("rb") as stream:
-        if hasattr(hashlib, "file_digest"):
-            return hashlib.file_digest(stream, "sha256").hexdigest()
-        hasher = hashlib.sha256()
-        while chunk := stream.read(1024 * 1024):
-            hasher.update(chunk)
-        return hasher.hexdigest()
-
-
-def number(value: Any, name: str) -> Decimal:
-    require(type(value) in (float, int), f"invalid numeric field: {name}")
-    return Decimal(str(value))
-
-
-def maybe_number(value: Any, name: str) -> Decimal | None:
-    if value is None:
-        return None
-    return number(value, name)
-
-
-def pct(value: Decimal | None) -> str:
-    if value is None:
-        return "pending"
-    return f"{float(value * Decimal('100')):.2f}%"
-
-
-def ratio(value: Decimal | None) -> str:
-    if value is None:
-        return "pending"
-    return f"{float(value):.3f}x"
-
-
-def max_decimal(values: list[Decimal | None]) -> Decimal | None:
-    present = [value for value in values if value is not None]
-    if not present:
-        return None
-    return max(present)
-
-
-def summarize_abba(abba_dir: Path) -> dict[str, Any]:
-    manifest_path = abba_dir / "manifest.json"
-    report_path = abba_dir / "report.json"
-    manifest = read_json(manifest_path)
-    report = read_json(report_path)
-    comparisons = report.get("comparisons")
-    require(isinstance(comparisons, list), "report.comparisons must be a list")
-
-    counts = Counter()
-    p99_regressions: list[Decimal] = []
-    throughput_losses: list[Decimal] = []
-    p1_rows = []
-    p2_values: list[Decimal | None] = []
-    for index, comparison in enumerate(comparisons):
-        require(isinstance(comparison, dict), f"comparison {index} must be an object")
-        state = comparison.get("status")
-        require(isinstance(state, str) and state, f"comparison {index} missing status")
-        counts[state] += 1
-        p99_regressions.append(number(comparison.get("p99_regression"), f"comparison {index} p99_regression"))
-        throughput_change = number(comparison.get("throughput_change"), f"comparison {index} throughput_change")
-        throughput_losses.append(max(Decimal("0"), -throughput_change))
-        p1 = comparison.get("p1")
-        if isinstance(p1, dict):
-            p1_rows.append({
-                "scenario": comparison.get("scenario"),
-                "comparison": comparison.get("comparison"),
-                "round": comparison.get("round"),
-                "required_reduction": float(number(p1.get("required_reduction"), "p1.required_reduction")),
-                "observed_reduction": float(number(p1.get("observed_reduction"), "p1.observed_reduction")),
-                "repeatability_drift": (
-                    None if p1.get("repeatability_drift") is None
-                    else float(number(p1.get("repeatability_drift"), "p1.repeatability_drift"))
-                ),
-            })
-        p2 = comparison.get("p2_post_stop_work_multiples")
-        if isinstance(p2, list):
-            p2_values.extend(maybe_number(value, "p2_post_stop_work_multiple") for value in p2)
-
-    report_state = report.get("status")
-    performance_state = report.get("performance")
-    require(isinstance(report_state, str) and report_state, "report.status missing")
-    require(isinstance(performance_state, str) and performance_state, "report.performance missing")
-    measured = report.get("evidence") == "measured"
-    passed = report_state in PASS_STATES and performance_state in PASS_STATES and measured
-    gate_state = "pass" if passed else "fail"
-    if report_state == "synthetic_validated":
-        reason = "synthetic evidence validates the harness only; measured performance remains pending"
-    elif report_state not in PASS_STATES:
-        reason = f"ABBA report status is {report_state}"
-    elif performance_state not in PASS_STATES:
-        reason = f"performance status is {performance_state}"
-    elif not measured:
-        reason = "measured evidence is required for a performance conclusion"
-    else:
-        reason = "measured ABBA report passed"
-
-    fixed = manifest.get("fixed", {})
-    require(isinstance(fixed, dict), "manifest.fixed must be an object")
-    return {
-        "gate_state": gate_state,
-        "reason": reason,
-        "status": report_state,
-        "performance": performance_state,
-        "evidence": report.get("evidence"),
-        "cells": report.get("cells", 0),
-        "comparisons_total": len(comparisons),
-        "comparison_status_counts": dict(sorted(counts.items())),
-        "worst_p99_regression": None if not p99_regressions else float(max(p99_regressions)),
-        "worst_throughput_loss": None if not throughput_losses else float(max(throughput_losses)),
-        "p2_worst_post_stop_work_multiple": None if max_decimal(p2_values) is None else float(max_decimal(p2_values)),
-        "p1_reductions": p1_rows,
-        "provenance": {
-            "abba_dir": str(abba_dir.resolve()),
-            "manifest_sha256": digest(manifest_path),
-            "report_sha256": digest(report_path),
-            "baseline_revision": manifest.get("baseline", {}).get("revision"),
-            "baseline_sha256": manifest.get("baseline", {}).get("sha256"),
-            "candidate_revision": manifest.get("candidate", {}).get("revision"),
-            "candidate_sha256": manifest.get("candidate", {}).get("sha256"),
-            "adapter_sha256": manifest.get("adapter_sha256"),
-            "collector_sha256": manifest.get("collector_sha256"),
-            "config_sha256": fixed.get("config_sha256"),
-            "dataset_sha256": fixed.get("dataset_sha256"),
-            "release_flags": fixed.get("release_flags"),
-            "durability": fixed.get("durability"),
-            "topology": fixed.get("topology"),
-            "offered_load_ops": fixed.get("offered_load_ops"),
-        },
-    }
-
-
-def cache_cost_records(path: Path) -> list[dict[str, Any]]:
-    require(path.is_file(), f"missing cache-cost log: {path}")
-    records = []
-    with path.open(encoding="utf-8", errors="replace") as stream:
-        for line_no, line in enumerate(stream, start=1):
-            if CACHE_COST_PREFIX not in line:
-                continue
-            payload = line.split(CACHE_COST_PREFIX, 1)[1].strip()
-            value = json.loads(payload)
-            require(isinstance(value, dict), f"cache-cost line {line_no} is not a JSON object")
-            require(value.get("schema") == 1, f"cache-cost line {line_no} has unsupported schema")
-            records.append(value)
-    require(records, f"no {CACHE_COST_PREFIX.strip()} records found in {path}")
-    return records
-
-
-def summarize_cache_cost(path: Path) -> dict[str, Any]:
-    records = cache_cost_records(path)
-    scenarios = Counter()
-    max_wire = Decimal("0")
-    max_save_amp = Decimal("0")
-    max_clone_ns = Decimal("0")
-    max_encode_ns = Decimal("0")
-    max_save_ns = Decimal("0")
-    build_sources = set()
-    for index, record in enumerate(records):
-        scenarios[str(record.get("scenario"))] += 1
-        wire = number(record.get("cache_wire_bytes"), f"cache_cost {index} cache_wire_bytes")
-        save_body = number(record.get("save_body_bytes_per_sample"), f"cache_cost {index} save_body_bytes_per_sample")
-        require(wire > 0, f"cache_cost {index} has zero wire bytes")
-        max_wire = max(max_wire, wire)
-        max_save_amp = max(max_save_amp, save_body / wire)
-        for field, target in (("clone", "max_clone_ns"), ("encode", "max_encode_ns"), ("save_inclusive", "max_save_ns")):
-            quantiles = record.get(field)
-            require(isinstance(quantiles, dict), f"cache_cost {index} missing {field} quantiles")
-            value = number(quantiles.get("max_ns"), f"cache_cost {index} {field}.max_ns")
-            if target == "max_clone_ns":
-                max_clone_ns = max(max_clone_ns, value)
-            elif target == "max_encode_ns":
-                max_encode_ns = max(max_encode_ns, value)
-            else:
-                max_save_ns = max(max_save_ns, value)
-        build = record.get("build", {})
-        require(isinstance(build, dict), f"cache_cost {index} build must be an object")
-        build_sources.add((build.get("source_revision"), build.get("source_tree"), build.get("test_opt_level_override")))
-    return {
-        "records": len(records),
-        "scenario_counts": dict(sorted(scenarios.items())),
-        "max_cache_wire_bytes": int(max_wire),
-        "max_save_body_amplification": float(max_save_amp),
-        "max_clone_ns": int(max_clone_ns),
-        "max_encode_ns": int(max_encode_ns),
-        "max_save_inclusive_ns": int(max_save_ns),
-        "build_sources": [
-            {"source_revision": revision, "source_tree": tree, "test_opt_level_override": opt}
-            for revision, tree, opt in sorted(build_sources, key=lambda item: tuple("" if part is None else str(part) for part in item))
-        ],
-        "provenance": {
-            "cache_cost_log": str(path.resolve()),
-            "cache_cost_log_sha256": digest(path),
-        },
-    }
-
-
-def markdown(summary: dict[str, Any]) -> str:
-    abba = summary["abba"]
-    p2 = None if abba["p2_worst_post_stop_work_multiple"] is None else Decimal(str(abba["p2_worst_post_stop_work_multiple"]))
-    p99 = None if abba["worst_p99_regression"] is None else Decimal(str(abba["worst_p99_regression"]))
-    throughput = None if abba["worst_throughput_loss"] is None else Decimal(str(abba["worst_throughput_loss"]))
-    lines = [
-        f"# Scanner/Heal Performance Summary",
-        "",
-        f"- verdict: {summary['verdict']}",
-        f"- reason: {summary['reason']}",
-        f"- abba: status={abba['status']} performance={abba['performance']} evidence={abba['evidence']} cells={abba['cells']} comparisons={abba['comparisons_total']}",
-        f"- worst_p99_regression: {pct(p99)}",
-        f"- worst_throughput_loss: {pct(throughput)}",
-        f"- p2_worst_post_stop_work_multiple: {ratio(p2)}",
-    ]
-    if summary.get("cache_cost") is not None:
-        cache = summary["cache_cost"]
-        lines.extend([
-            f"- cache_cost_records: {cache['records']}",
-            f"- max_cache_wire_bytes: {cache['max_cache_wire_bytes']}",
-            f"- max_save_body_amplification: {cache['max_save_body_amplification']:.3f}x",
-            f"- max_clone_ns: {cache['max_clone_ns']}",
-            f"- max_encode_ns: {cache['max_encode_ns']}",
-            f"- max_save_inclusive_ns: {cache['max_save_inclusive_ns']}",
-        ])
-    lines.extend([
-        "",
-        "## Provenance",
-        "",
-    ])
-    for key, value in abba["provenance"].items():
-        lines.append(f"- {key}: {value}")
-    if summary.get("cache_cost") is not None:
-        for key, value in summary["cache_cost"]["provenance"].items():
-            lines.append(f"- {key}: {value}")
-    return "\n".join(lines) + "\n"
-
-
-def build_summary(args: argparse.Namespace) -> dict[str, Any]:
-    abba = summarize_abba(args.abba_dir)
-    cache = summarize_cache_cost(args.cache_cost_log) if args.cache_cost_log else None
-    if args.require_cache_cost and cache is None:
-        raise ValueError("cache-cost profile log is required")
-    verdict = "PASS" if abba["gate_state"] == "pass" else "FAIL"
-    reason = abba["reason"]
-    return {"schema": 1, "verdict": verdict, "reason": reason, "abba": abba, "cache_cost": cache}
-
-
-def main() -> int:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--abba-dir", type=Path, required=True, help="Directory containing manifest.json and report.json")
-    parser.add_argument("--cache-cost-log", type=Path, help="Rust test output containing CACHE_COST JSON lines")
-    parser.add_argument("--require-cache-cost", action="store_true", help="Fail when --cache-cost-log is missing")
-    parser.add_argument("--json-out", type=Path, help="Write the normalized summary JSON artifact")
-    parser.add_argument("--markdown-out", type=Path, help="Write a compact Markdown summary artifact")
-    args = parser.parse_args()
-    try:
-        summary = build_summary(args)
-        if args.json_out:
-            write_json(args.json_out, summary)
-        if args.markdown_out:
-            args.markdown_out.parent.mkdir(parents=True, exist_ok=True)
-            args.markdown_out.write_text(markdown(summary), encoding="utf-8")
-        abba = summary["abba"]
-        print(
-            f"{summary['verdict']} scanner_heal_perf "
-            f"status={abba['status']} performance={abba['performance']} evidence={abba['evidence']} "
-            f"comparisons={abba['comparisons_total']} reason={summary['reason']}"
-        )
-        return 0 if summary["verdict"] == "PASS" else 1
-    except (ValueError, OSError, json.JSONDecodeError) as error:
-        print(f"FAIL scanner_heal_perf error={error}", file=sys.stderr)
-        return 1
-
-
-if __name__ == "__main__":
-    sys.exit(main())

scripts/test_summarize_scanner_heal_perf.py

@@ -1,171 +0,0 @@
-#!/usr/bin/env python3
-
-from __future__ import annotations
-
-import contextlib
-import hashlib
-import io
-import json
-from pathlib import Path
-import subprocess
-import sys
-import tempfile
-import unittest
-from unittest import mock
-
-sys.path.insert(0, str(Path(__file__).resolve().parent))
-import summarize_scanner_heal_perf as summary
-
-
-def sha(path: Path) -> str:
-    with path.open("rb") as stream:
-        hasher = hashlib.sha256()
-        while chunk := stream.read(1024 * 1024):
-            hasher.update(chunk)
-        return hasher.hexdigest()
-
-
-class ScannerHealPerfSummaryTest(unittest.TestCase):
-    def setUp(self):
-        self.temp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.temp.cleanup)
-        self.root = Path(self.temp.name)
-        self.abba = self.root / "abba"
-        self.abba.mkdir()
-        self.manifest = {
-            "fixed": {
-                "config_sha256": "1" * 64,
-                "dataset_sha256": "2" * 64,
-                "release_flags": "--profile production",
-                "durability": "drive-sync=on",
-                "topology": "EC8+4",
-                "offered_load_ops": 100,
-            },
-            "baseline": {"revision": "a" * 40, "sha256": "3" * 64},
-            "candidate": {"revision": "b" * 40, "sha256": "4" * 64},
-            "adapter_sha256": "5" * 64,
-            "collector_sha256": "6" * 64,
-        }
-        self.comparison = {
-            "scenario": "cold-hot",
-            "comparison": "build",
-            "round": 1,
-            "status": "pass",
-            "p99_regression": 0.02,
-            "throughput_change": -0.01,
-            "p1": {"required_reduction": 0.8, "observed_reduction": 0.82, "repeatability_drift": 0.01},
-            "p2_post_stop_work_multiples": [None, 1.1, 1.0, None],
-        }
-        self.report = {
-            "status": "pass",
-            "performance": "pass",
-            "evidence": "measured",
-            "cells": 120,
-            "comparisons": [self.comparison],
-        }
-        self.write_inputs()
-
-    def write_inputs(self):
-        (self.abba / "manifest.json").write_text(json.dumps(self.manifest), encoding="utf-8")
-        (self.abba / "report.json").write_text(json.dumps(self.report), encoding="utf-8")
-
-    def test_measured_pass_writes_quiet_artifacts(self):
-        cache_log = self.root / "cache.log"
-        cache_log.write_text(
-            "compiler noise\nCACHE_COST "
-            + json.dumps({
-                "schema": 1,
-                "scenario": "small_dirty",
-                "cache_wire_bytes": 100,
-                "save_body_bytes_per_sample": 200,
-                "clone": {"max_ns": 10},
-                "encode": {"max_ns": 20},
-                "save_inclusive": {"max_ns": 30},
-                "build": {"source_revision": "abc", "source_tree": "clean", "test_opt_level_override": "0"},
-            })
-            + "\nmore noise\n",
-            encoding="utf-8",
-        )
-        args = type("Args", (), {
-            "abba_dir": self.abba,
-            "cache_cost_log": cache_log,
-            "require_cache_cost": False,
-            "json_out": None,
-            "markdown_out": None,
-        })
-        result = summary.build_summary(args)
-        self.assertEqual(result["verdict"], "PASS")
-        self.assertEqual(result["abba"]["provenance"]["manifest_sha256"], sha(self.abba / "manifest.json"))
-        self.assertEqual(result["cache_cost"]["max_save_body_amplification"], 2.0)
-
-    def test_synthetic_report_fails_as_performance_conclusion(self):
-        self.report.update(status="synthetic_validated", performance="pending", evidence="synthetic")
-        self.write_inputs()
-        args = type("Args", (), {
-            "abba_dir": self.abba,
-            "cache_cost_log": None,
-            "require_cache_cost": False,
-            "json_out": None,
-            "markdown_out": None,
-        })
-        result = summary.build_summary(args)
-        self.assertEqual(result["verdict"], "FAIL")
-        self.assertIn("synthetic evidence", result["reason"])
-
-    def test_requires_cache_profile_when_requested(self):
-        args = type("Args", (), {
-            "abba_dir": self.abba,
-            "cache_cost_log": None,
-            "require_cache_cost": True,
-            "json_out": None,
-            "markdown_out": None,
-        })
-        with self.assertRaisesRegex(ValueError, "cache-cost profile log is required"):
-            summary.build_summary(args)
-
-    def test_cli_prints_one_line_and_exits_nonzero_for_pending_performance(self):
-        self.report.update(status="inconclusive", performance="inconclusive")
-        self.write_inputs()
-        stdout = io.StringIO()
-        stderr = io.StringIO()
-        argv = [
-            "summarize_scanner_heal_perf.py",
-            "--abba-dir",
-            str(self.abba),
-            "--json-out",
-            str(self.root / "summary.json"),
-            "--markdown-out",
-            str(self.root / "summary.md"),
-        ]
-        with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr):
-            with mock.patch.object(sys, "argv", argv):
-                code = summary.main()
-        self.assertEqual(code, 1)
-        self.assertEqual(stdout.getvalue().count("\n"), 1)
-        self.assertTrue(stdout.getvalue().startswith("FAIL scanner_heal_perf "))
-        self.assertEqual(stderr.getvalue(), "")
-        self.assertEqual(summary.read_json(self.root / "summary.json")["verdict"], "FAIL")
-        self.assertIn("worst_p99_regression", (self.root / "summary.md").read_text(encoding="utf-8"))
-
-    def test_invalid_cache_cost_lines_fail_closed(self):
-        cache_log = self.root / "cache.log"
-        cache_log.write_text("CACHE_COST {\"schema\": 2}\n", encoding="utf-8")
-        with self.assertRaisesRegex(ValueError, "unsupported schema"):
-            summary.summarize_cache_cost(cache_log)
-
-    def test_script_entrypoint_is_quiet(self):
-        script = Path(__file__).with_name("summarize_scanner_heal_perf.py")
-        process = subprocess.run(
-            [sys.executable, str(script), "--abba-dir", str(self.abba)],
-            check=False,
-            text=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
-        self.assertEqual(process.returncode, 0, process.stderr)
-        self.assertEqual(process.stdout.count("\n"), 1)
-        self.assertEqual(process.stderr, "")
-
-
-if __name__ == "__main__":
-    unittest.main()