fix(health): report storage quorum in node readiness (#7566)

crates/e2e_test/src/namespace_lock_quorum_test.rs

@@ -17,6 +17,7 @@ use aws_sdk_s3::Client;
 use aws_sdk_s3::error::SdkError;
 use bytes::Bytes;
 use std::sync::Arc;
+use std::time::{Duration, Instant};
 use tokio::sync::Barrier;
 use tracing::{info, warn};
 
@@ -175,6 +176,9 @@ async fn test_concurrent_cluster_overwrites_do_not_fail_namespace_lock_quorum()
     // Keep the regression focused on false quorum-loss errors, not ordinary lock
     // wait exhaustion under a heavily contended same-key overwrite workload.
     cluster.set_env("RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT", "20");
+    cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", "EC:2");
+    cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false");
+    cluster.set_env("RUSTFS_HEALTH_MINIMAL_RESPONSE_ENABLE", "false");
     cluster.start().await?;
     cluster.create_test_bucket(BUCKET).await?;
 
@@ -233,6 +237,181 @@ async fn test_concurrent_cluster_overwrites_do_not_fail_namespace_lock_quorum()
     );
 
     clients[0].delete_object().bucket(BUCKET).key(KEY).send().await?;
+    assert_node_readiness_tracks_quorum(&mut cluster).await?;
+    Ok(())
+}
+
+async fn assert_node_readiness_tracks_quorum(cluster: &mut RustFSTestClusterEnvironment) -> TestResult {
+    let clients: Vec<_> = cluster
+        .create_all_clients()?
+        .into_iter()
+        .map(|client| {
+            Client::from_conf(
+                client
+                    .config()
+                    .to_builder()
+                    .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1))
+                    .build(),
+            )
+        })
+        .collect();
+    let http = reqwest::Client::builder()
+        .no_proxy()
+        .timeout(Duration::from_secs(3))
+        .build()?;
+    let seed_key = "readiness-seed";
+    let seed_body = b"readiness quorum regression";
+    clients[0]
+        .put_object()
+        .bucket(BUCKET)
+        .key(seed_key)
+        .body(Bytes::from_static(seed_body).into())
+        .send()
+        .await?;
+
+    for (phase, survivors) in [4, 3, 2, 1, 4].into_iter().enumerate() {
+        if phase == 4 {
+            cluster.stop();
+            cluster.start().await?;
+        } else if survivors < 4 {
+            cluster.stop_node(survivors)?;
+        }
+        let write_ready = survivors >= 3;
+        let read_quorum = survivors >= 2;
+        let expected_status = if write_ready { 200 } else { 503 };
+        for (idx, client) in clients.iter().enumerate().take(survivors) {
+            let url = &cluster.nodes[idx].url;
+            let deadline = Instant::now() + Duration::from_secs(30);
+            // Poll health before issuing S3 I/O: idle remote disk handles must
+            // not remain evidence of quorum after their host becomes unreachable.
+            let payload = loop {
+                let response = http.get(format!("{url}/health/ready")).send().await?;
+                let status = response.status().as_u16();
+                let payload: serde_json::Value = response.json().await?;
+                if status == expected_status
+                    && payload["ready"] == write_ready
+                    && payload["details"]["storage"]["ready"] == write_ready
+                    && payload["details"]["storage"]["readQuorum"] == read_quorum
+                    && payload["details"]["storage"]["writeQuorum"] == write_ready
+                    && payload["details"]["poolMetadata"]["ready"] == true
+                    && payload["details"]["iam"]["ready"] == true
+                    && payload["details"]["lock"]["ready"] == write_ready
+                {
+                    break payload;
+                }
+                assert!(Instant::now() < deadline, "node {idx}, survivors={survivors}: HTTP {status}, {payload}");
+                tokio::time::sleep(Duration::from_millis(200)).await;
+            };
+            assert_eq!(payload["details"]["storage"]["readinessScope"], "write_quorum_and_pool_metadata");
+            assert_eq!(payload["details"]["storage"]["source"], "local_runtime");
+            assert_eq!(
+                payload["details"]["storage"]["status"],
+                if write_ready { "connected" } else { "disconnected" }
+            );
+            if !write_ready {
+                assert!(
+                    payload["degradedReasons"]
+                        .as_array()
+                        .expect("degraded reasons")
+                        .iter()
+                        .any(|reason| reason == "storage_and_lock_unavailable")
+                );
+            }
+
+            for path in ["/health/ready", "/minio/health/ready"] {
+                let head = http.head(format!("{url}{path}")).send().await?;
+                assert_eq!(head.status().as_u16(), expected_status, "HEAD {path}, survivors={survivors}");
+                assert!(head.bytes().await?.is_empty());
+                let response = http.get(format!("{url}{path}")).send().await?;
+                assert_eq!(response.status().as_u16(), expected_status);
+                let body: serde_json::Value = response.json().await?;
+                assert_eq!(body["details"]["storage"], payload["details"]["storage"]);
+                assert_eq!(body["details"]["poolMetadata"], payload["details"]["poolMetadata"]);
+            }
+            let live = http.get(format!("{url}/health/live")).send().await?;
+            assert_eq!(live.status().as_u16(), 200);
+            assert!(live.json::<serde_json::Value>().await?.get("details").is_none());
+            for (path, storage_ready, scope) in [
+                ("/minio/health/cluster", write_ready, "write_quorum_and_pool_metadata"),
+                ("/minio/health/cluster/read", read_quorum, "read_quorum"),
+            ] {
+                let deadline = Instant::now() + Duration::from_secs(30);
+                // Cluster read/write reports have independent caches; allow
+                // each observation to expire before comparing stable states.
+                let body = loop {
+                    let response = http.get(format!("{url}{path}")).send().await?;
+                    let status = response.status().as_u16();
+                    let body: serde_json::Value = response.json().await?;
+                    if status == expected_status
+                        && body["details"]["storage"]["ready"] == storage_ready
+                        && body["details"]["lock"]["ready"] == write_ready
+                    {
+                        break body;
+                    }
+                    assert!(Instant::now() < deadline, "{path}, survivors={survivors}: HTTP {status}, {body}");
+                    tokio::time::sleep(Duration::from_millis(200)).await;
+                };
+                assert_eq!(body["details"]["storage"]["readinessScope"], scope);
+            }
+
+            let put = client
+                .put_object()
+                .bucket(BUCKET)
+                .key(format!("readiness-phase-{phase}-node-{idx}"))
+                .body(Bytes::from_static(seed_body).into())
+                .send()
+                .await;
+            let put_status = if write_ready {
+                put.expect("a ready node must accept the PUT");
+                200
+            } else {
+                let error = put.expect_err("subquorum node must reject PUT");
+                assert!(
+                    error.raw_response().is_some_and(|response| response.status().as_u16() >= 500),
+                    "unexpected PUT failure: {error:?}"
+                );
+                error.raw_response().expect("PUT error response").status().as_u16()
+            };
+            let get = client.get_object().bucket(BUCKET).key(seed_key).send().await;
+            let get_status = match get {
+                Ok(object) => {
+                    assert_eq!(object.body.collect().await?.into_bytes().as_ref(), seed_body);
+                    200
+                }
+                Err(error) => {
+                    assert!(!write_ready, "GET must succeed on a ready cluster: {error:?}");
+                    let status = error
+                        .raw_response()
+                        .expect("GET should have an HTTP response")
+                        .status()
+                        .as_u16();
+                    assert!(status >= 500, "unexpected GET failure: {error:?}");
+                    status
+                }
+            };
+            let list = client.list_objects_v2().bucket(BUCKET).send().await;
+            let list_status = match list {
+                Ok(result) => {
+                    assert!(result.contents().iter().any(|object| object.key() == Some(seed_key)));
+                    200
+                }
+                Err(error) => {
+                    assert!(!write_ready, "listing must succeed on a ready cluster: {error:?}");
+                    let status = error
+                        .raw_response()
+                        .expect("LIST should have an HTTP response")
+                        .status()
+                        .as_u16();
+                    assert!(status >= 500, "unexpected listing failure: {error:?}");
+                    status
+                }
+            };
+            eprintln!(
+                "readiness matrix: survivors={survivors}, node={idx}, ready={write_ready}, read_quorum={read_quorum}, PUT={put_status}, GET={get_status}, LIST={list_status}"
+            );
+        }
+    }
+    cluster.stop();
     Ok(())
 }
 

docs/architecture/readiness-matrix.md

@@ -39,7 +39,7 @@ an unknown or unsupported peer-health snapshot degrades readiness with
 
 - Liveness reports process availability and must not depend on storage, IAM,
   lock quorum, or peer health.
-- Node readiness reports local dependency readiness.
+- Node readiness reports the node's observed storage write quorum, local pool metadata write gate, IAM, and lock quorum. Runtime storage diagnostics do not change the existing startup `FullReady` publication gate or S3 request admission.
 - 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;
   contention reports `pool_metadata_check_timeout` without installing a block.
@@ -49,11 +49,37 @@ an unknown or unsupported peer-health snapshot degrades readiness with
   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
-  timeout behavior.
+- Cluster read readiness uses the storage read-quorum path and cluster-health timeout behavior. Its lock dependency still uses the per-set majority/write-lock health check. Actual shared namespace locks require `ceil(lock_clients / 2)`, so the cluster read probe is conservative: a four-client set can still admit some reads with two clients while the probe returns 503. This diagnostic change does not lower that probe's lock threshold.
 - `HEAD` health probes keep header/status semantics and do not require response
   bodies.
 
+### Storage Detail Contract
+
+The existing `details.storage.ready` boolean and `connected` / `disconnected` status values are retained. `readinessScope` states the condition they summarize:
+
+| Probe | `readinessScope` | `source` |
+| --- | --- | --- |
+| `/health/ready`, `/minio/health/ready` | `write_quorum_and_pool_metadata` | `local_runtime` |
+| `/minio/health/cluster` | `write_quorum_and_pool_metadata` | `storage_inventory` |
+| `/minio/health/cluster/read` | `read_quorum` | `storage_inventory` |
+
+Node readiness additionally reports `details.storage.readQuorum`, `details.storage.writeQuorum`, and `details.poolMetadata.ready`. The metadata component's status is `writable` or `unavailable`; existing typed degradation reasons distinguish a write block from an inspection timeout. A healthy metadata writer alone no longer makes the storage component ready.
+
+Node storage quorum uses configured drives per set, all configured pools/sets, and their Standard storage-class data/parity layout. Missing, duplicate, unreachable, or unhealthy disk observations cannot supply extra quorum votes. The read quorum is the data-drive count; the write quorum is that count plus one when data and parity counts are equal. These are observations of available storage slots, not guarantees that a particular object's metadata, shards, or required locks are available.
+
+For a healthy IAM and metadata writer in a four-node, one-drive-per-node EC 2+2 set:
+
+| Surviving nodes | Storage read quorum | Storage write quorum | Pool metadata ready | Node HTTP / top-level ready |
+| --- | --- | --- | --- | --- |
+| 4 or 3 | true | true | true | 200 / true |
+| 2 | true | false | true | 503 / false |
+| 1 | false | false | true | 503 / false |
+| All restored | true | true | true | 200 / true |
+
+The node path reads local disk-handle health and reuses the same reachable-host observation as its lock dependency, including the existing `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` cache. Only `Online` drives count; a reachable host with a `Returning` drive does not yet prove data I/O has recovered. It does not call cluster `storage_info`, local `disk_info`, or add disk-info RPCs. The entire storage inventory snapshot has a separate 100 ms wait budget; expiry reports `storage_readiness_check_timeout` and fails closed. Pool metadata inspection retains its own 100 ms budget. These observations are not an atomic cluster snapshot and do not bypass the existing lock-probe timing or cache policy.
+
+The new fields are additive. Their absence in an older response is not evidence of storage quorum. Minimal responses still contain only the existing top-level fields, liveness remains dependency-independent, and HEAD responses remain bodyless.
+
 ## Preservation Rules
 
 - Do not move peer-health checks into the S3 data hot path.

docs/operations/rolling-restart.md

@@ -56,8 +56,8 @@ When the whole cluster (or several nodes) went down and nodes are brought back o
    | `startup_finalization` | Last startup steps are being published. |
 
 2. Logs say what the node waits for. The IAM recovery loop retries with backoff and logs `event="iam_bootstrap_retry_failed"` with an actionable `hint` field (for example, "storage read quorum not met yet; waiting for enough cluster nodes/disks to come online"). After repeated failures the level escalates from WARN to ERROR; this still does not kill the process.
-3. Recovery is automatic. As soon as enough peers are online for the storage read quorum, the pending nodes finish IAM bootstrap on the next retry and flip `/health/ready` to `200` on their own. Restarting them does not speed anything up.
-4. Check readiness detail while waiting. `/health/ready` (and `/minio/health/ready`) return per-dependency detail during degradation; the `details` object shows `storage` / `iam` / `lock` readiness and `degradedReasons` lists machine-readable causes such as `storage_quorum_unavailable` or `lock_quorum_unavailable`:
+3. Recovery is automatic. Once storage read quorum is available, pending nodes can finish IAM bootstrap on the next retry. `/health/ready` returns `200` when storage write quorum, the metadata write gate, IAM, and lock readiness are satisfied. Restarting pending nodes does not speed this up.
+4. Check readiness detail while waiting. `/health/ready` (and `/minio/health/ready`) separate `storage` / `poolMetadata` / `iam` / `lock` readiness. `storage.ready` summarizes write quorum plus the metadata write gate; `storage.readQuorum` and `storage.writeQuorum` show the separate quorum observations. A healthy `poolMetadata.ready` does not imply storage quorum. `degradedReasons` lists machine-readable causes such as `storage_quorum_unavailable`, `storage_and_lock_unavailable`, or `pool_metadata_check_timeout`. See the [storage detail contract](../architecture/readiness-matrix.md#storage-detail-contract) for probe scopes and sampling limits:
 
    ```bash
    curl -s http://<node>:9000/health/ready | jq

rustfs/src/admin/handlers/health.rs

@@ -211,6 +211,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: vec![crate::shared_types::ReadinessDegradedReason::StorageQuorumUnavailable],
+            storage_details: None,
         };
         let parts = build_health_response_parts(
             Method::GET,
@@ -233,6 +234,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: Vec::new(),
+            storage_details: None,
         };
         let parts = build_health_response_parts(
             Method::GET,
@@ -255,6 +257,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: vec![crate::shared_types::ReadinessDegradedReason::StorageAndIamUnavailable],
+            storage_details: None,
         };
         let parts = build_health_response_parts(
             Method::GET,
@@ -284,6 +287,7 @@ mod tests {
                     peer_health_ready: true,
                 },
                 degraded_reasons: vec![crate::shared_types::ReadinessDegradedReason::LockQuorumUnavailable],
+                storage_details: None,
             };
 
             let liveness = build_health_response_parts(
@@ -328,6 +332,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: vec![crate::shared_types::ReadinessDegradedReason::StorageAndIamUnavailable],
+            storage_details: None,
         };
         let parts = build_health_response_parts(
             Method::HEAD,
@@ -398,6 +403,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: Vec::new(),
+            storage_details: None,
         };
         let parts =
             build_health_response_parts(Method::HEAD, HealthProbe::Readiness, Some(&report), "rustfs-endpoint", None, None);
@@ -417,6 +423,7 @@ mod tests {
                     peer_health_ready: true,
                 },
                 degraded_reasons: vec![crate::shared_types::ReadinessDegradedReason::StorageQuorumUnavailable],
+                storage_details: None,
             };
             let parts =
                 build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs-endpoint", None, None);
@@ -440,6 +447,7 @@ mod tests {
                     peer_health_ready: true,
                 },
                 degraded_reasons: Vec::new(),
+                storage_details: None,
             };
             let parts = build_health_response_parts(
                 Method::GET,

rustfs/src/cluster_snapshot.rs

@@ -277,6 +277,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: Vec::new(),
+            storage_details: None,
         });
         assert_eq!(ready.state, ClusterRuntimeReadinessState::Ready);
 
@@ -288,6 +289,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: vec![ReadinessDegradedReason::StorageQuorumUnavailable],
+            storage_details: None,
         });
         assert_eq!(degraded.state, ClusterRuntimeReadinessState::Degraded);
         assert_eq!(degraded.degraded_reasons, vec![ReadinessDegradedReason::StorageQuorumUnavailable]);
@@ -304,6 +306,7 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: vec![ReadinessDegradedReason::StorageAndLockUnavailable],
+            storage_details: None,
         });
 
         let snapshot = cluster_read_only_snapshot_from_endpoint_pools(&endpoint_pools, runtime_status);

rustfs/src/server/health.rs

@@ -343,7 +343,7 @@ pub(crate) fn build_health_response_parts(
     let payload = if method == Method::HEAD {
         None
     } else {
-        Some(build_health_payload(HealthPayloadContext {
+        let mut payload = build_health_payload(HealthPayloadContext {
             probe,
             health,
             storage_ready,
@@ -354,7 +354,18 @@ pub(crate) fn build_health_response_parts(
             uptime,
             kms_ready,
             include_dependency_details,
-        }))
+        });
+        if let Some(details) = readiness_report.and_then(|report| report.storage_details)
+            && payload.get("details").is_some()
+        {
+            payload["details"]["storage"]["readQuorum"] = json!(details.read_quorum_ready);
+            payload["details"]["storage"]["writeQuorum"] = json!(details.write_quorum_ready);
+            payload["details"]["poolMetadata"] = json!({
+                "ready": details.pool_metadata_write_ready,
+                "status": if details.pool_metadata_write_ready { "writable" } else { "unavailable" },
+            });
+        }
+        Some(payload)
     };
 
     HealthResponseParts {
@@ -390,6 +401,14 @@ pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value {
 
     if ctx.include_dependency_details {
         payload["details"] = build_component_details(ctx.storage_ready, ctx.iam_ready, ctx.lock_quorum_ready, ctx.kms_ready);
+        payload["details"]["storage"]["readinessScope"] = json!(match ctx.probe {
+            HealthProbe::ClusterRead => "read_quorum",
+            _ => "write_quorum_and_pool_metadata",
+        });
+        payload["details"]["storage"]["source"] = json!(match ctx.probe {
+            HealthProbe::Readiness => "local_runtime",
+            _ => "storage_inventory",
+        });
         payload["degradedReasons"] = build_degraded_reasons(ctx.degraded_reasons);
     }
 
@@ -432,9 +451,86 @@ mod tests {
                 peer_health_ready: true,
             },
             degraded_reasons: Vec::new(),
+            storage_details: None,
         }
     }
 
+    #[test]
+    #[serial]
+    fn node_storage_details_separate_quorum_from_pool_metadata() {
+        with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"), || {
+            for (read_quorum, write_quorum, metadata_ready, lock_ready) in [
+                (true, true, true, true),
+                (true, false, true, false),
+                (false, false, true, false),
+                (true, true, false, true),
+                (true, true, true, true),
+            ] {
+                let mut report = ready_report();
+                report.readiness.storage_ready = write_quorum && metadata_ready;
+                report.readiness.lock_quorum_ready = lock_ready;
+                report.storage_details = Some(crate::shared_types::StorageReadinessDetails {
+                    read_quorum_ready: read_quorum,
+                    write_quorum_ready: write_quorum,
+                    pool_metadata_write_ready: metadata_ready,
+                });
+                let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs", None, None);
+                let expected_ready = write_quorum && metadata_ready && lock_ready;
+                assert_eq!(
+                    parts.status_code,
+                    if expected_ready {
+                        StatusCode::OK
+                    } else {
+                        StatusCode::SERVICE_UNAVAILABLE
+                    }
+                );
+                let payload = parts.payload.expect("GET readiness body");
+                assert_eq!(payload["ready"], expected_ready);
+                assert_eq!(payload["details"]["storage"]["ready"], write_quorum && metadata_ready);
+                assert_eq!(
+                    payload["details"]["storage"]["status"],
+                    if write_quorum && metadata_ready {
+                        "connected"
+                    } else {
+                        "disconnected"
+                    }
+                );
+                assert_eq!(payload["details"]["storage"]["readQuorum"], read_quorum);
+                assert_eq!(payload["details"]["storage"]["writeQuorum"], write_quorum);
+                assert_eq!(payload["details"]["poolMetadata"]["ready"], metadata_ready);
+                assert_eq!(payload["details"]["storage"]["source"], "local_runtime");
+                assert_eq!(payload["details"]["storage"]["readinessScope"], "write_quorum_and_pool_metadata");
+                assert!(
+                    build_health_response_parts(Method::HEAD, HealthProbe::Readiness, Some(&report), "rustfs", None, None)
+                        .payload
+                        .is_none()
+                );
+            }
+        });
+    }
+
+    #[test]
+    #[serial]
+    fn node_storage_details_do_not_expand_minimal_or_liveness_payloads() {
+        let mut report = ready_report();
+        report.storage_details = Some(crate::shared_types::StorageReadinessDetails {
+            read_quorum_ready: true,
+            write_quorum_ready: true,
+            pool_metadata_write_ready: true,
+        });
+        with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || {
+            let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs", None, None);
+            assert_eq!(parts.payload, Some(json!({ "status": "ok", "ready": true })));
+        });
+        with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"), || {
+            let parts = build_health_response_parts(Method::GET, HealthProbe::Liveness, Some(&report), "rustfs", None, None);
+            assert_eq!(parts.status_code, StatusCode::OK);
+            let payload = parts.payload.expect("liveness GET body");
+            assert!(payload.get("details").is_none());
+            assert!(payload.get("ready").is_none());
+        });
+    }
+
     #[tokio::test]
     async fn readiness_collects_object_stalls_and_recovers_on_completion() {
         let object_traffic_health = ObjectTrafficHealth::enabled_for_test(Duration::ZERO);

rustfs/src/server/readiness.rs

@@ -17,8 +17,8 @@ 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};
+use crate::storage_api::server::readiness::contract::admin::{DiskSetSelector, StorageAdminApi};
+use crate::storage_api::server::readiness::{DiskStore, Endpoint, EndpointServerPools, disk_endpoint_snapshot, is_dist_erasure};
 #[cfg(test)]
 use crate::storage_api::server::readiness::{Endpoints, PoolEndpoints};
 use crate::storage_api::startup::shutdown::mark_get_metadata_read_version_coalescing_service_ready;
@@ -30,7 +30,7 @@ use http_body_util::{BodyExt, Full};
 use hyper::body::Incoming;
 use metrics::{counter, gauge};
 use rustfs_common::GlobalReadiness;
-use rustfs_madmin::{Disk, StorageInfo};
+use rustfs_madmin::{BackendInfo, Disk, StorageInfo};
 use std::future::Future;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -76,7 +76,7 @@ fn startup_runtime_readiness_max_wait() -> Duration {
 const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready";
 const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total";
 
-pub use crate::shared_types::{DependencyReadiness, DependencyReadinessReport, ReadinessDegradedReason};
+pub use crate::shared_types::{DependencyReadiness, DependencyReadinessReport, ReadinessDegradedReason, StorageReadinessDetails};
 
 /// ReadinessGateLayer ensures that the system components (IAM, Storage)
 /// are fully initialized before allowing any request to proceed.
@@ -288,10 +288,16 @@ fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWri
     }
 }
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone)]
 struct LockQuorumCacheEntry {
     captured_at: Instant,
+    observation: LockQuorumObservation,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+struct LockQuorumObservation {
     status: LockQuorumStatus,
+    online_hosts: HashSet<String>,
 }
 
 #[derive(Debug, Clone)]
@@ -317,6 +323,7 @@ pub struct LockQuorumStatus {
 const DISK_STATE_OK: &str = "ok";
 const DISK_STATE_UNFORMATTED: &str = "unformatted";
 const RUNTIME_STATE_RETURNING: &str = "returning";
+const NODE_STORAGE_READINESS_TIMEOUT: Duration = Duration::from_millis(100);
 
 fn health_readiness_cache_ttl() -> Duration {
     Duration::from_millis(rustfs_utils::get_env_u64(
@@ -439,7 +446,7 @@ async fn update_storage_readiness_cache(status: StorageWriteReadinessStatus) {
     });
 }
 
-async fn load_cached_lock_quorum_status() -> Option<LockQuorumStatus> {
+async fn load_cached_lock_quorum_status() -> Option<LockQuorumObservation> {
     let ttl = health_readiness_cache_ttl();
     if ttl.is_zero() {
         return None;
@@ -448,13 +455,13 @@ async fn load_cached_lock_quorum_status() -> Option<LockQuorumStatus> {
     let cache = lock_quorum_status_cache().lock().await;
     let entry = cache.as_ref()?;
     if entry.captured_at.elapsed() <= ttl {
-        return Some(entry.status);
+        return Some(entry.observation.clone());
     }
 
     None
 }
 
-async fn update_lock_quorum_status_cache(status: LockQuorumStatus) {
+async fn update_lock_quorum_status_cache(observation: LockQuorumObservation) {
     if health_readiness_cache_ttl().is_zero() {
         return;
     }
@@ -462,7 +469,7 @@ async fn update_lock_quorum_status_cache(status: LockQuorumStatus) {
     let mut cache = lock_quorum_status_cache().lock().await;
     *cache = Some(LockQuorumCacheEntry {
         captured_at: Instant::now(),
-        status,
+        observation,
     });
 }
 
@@ -730,6 +737,7 @@ fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) ->
     DependencyReadinessReport {
         degraded_reasons: degraded_reasons(readiness),
         readiness,
+        storage_details: None,
     }
 }
 
@@ -740,6 +748,7 @@ fn dependency_readiness_report_from_write_status(
     DependencyReadinessReport {
         degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_metadata_reason),
         readiness,
+        storage_details: None,
     }
 }
 
@@ -774,18 +783,95 @@ pub async fn collect_cluster_read_health_report() -> DependencyReadinessReport {
 }
 
 pub async fn collect_node_readiness_report() -> DependencyReadinessReport {
-    let storage = node_pool_meta_write_readiness().await;
+    let lock_observation = collect_lock_quorum_observation().await;
+    let mut storage = StorageWriteReadinessStatus::default();
+    let mut details = StorageReadinessDetails::default();
+    let mut storage_check_timed_out = false;
+    if let Some(store) = runtime_sources::current_object_store_handle() {
+        storage = pool_metadata_write_readiness(store.pool_meta_write_status().await);
+        details.pool_metadata_write_ready = storage.ready;
+        match node_storage_snapshot(store.as_ref(), &lock_observation.online_hosts).await {
+            Ok(info) => {
+                details.read_quorum_ready = storage_read_ready_from_runtime_state(&info);
+                details.write_quorum_ready = storage_ready_from_runtime_state(&info);
+            }
+            Err(StorageError::Timeout) => storage_check_timed_out = true,
+            Err(_) => {}
+        }
+    }
+    storage.ready &= details.write_quorum_ready;
     let readiness = DependencyReadiness {
         storage_ready: storage.ready,
         iam_ready: runtime_sources::current_iam_ready(),
-        lock_quorum_ready: collect_lock_quorum_status().await.ready,
+        lock_quorum_ready: lock_observation.status.ready,
         peer_health_ready: collect_peer_health_readiness(),
     };
-    let report = dependency_readiness_report_from_write_status(readiness, storage);
+    let mut report = dependency_readiness_report_from_write_status(readiness, storage);
+    report.storage_details = Some(details);
+    if storage_check_timed_out {
+        report
+            .degraded_reasons
+            .push(ReadinessDegradedReason::StorageReadinessCheckTimeout);
+    }
     record_readiness_report(&report);
     report
 }
 
+async fn node_storage_snapshot<S>(store: &S, online_hosts: &HashSet<String>) -> Result<StorageInfo, StorageError>
+where
+    S: StorageAdminApi<BackendInfo = BackendInfo, Disk = DiskStore, Error = StorageError>,
+{
+    tokio::time::timeout(NODE_STORAGE_READINESS_TIMEOUT, async {
+        let mut info = StorageInfo {
+            backend: store.backend_info().await,
+            ..Default::default()
+        };
+        if configured_readiness_topology(&info).is_none() {
+            return Ok(info);
+        }
+
+        // Inventory and runtime health are local snapshots. Reuse the lock probe's
+        // peer reachability so an idle remote disk cannot outlive its failed host.
+        // Do not perform disk-info RPCs or filesystem probes in node readiness.
+        for (pool_idx, &set_count) in info.backend.total_sets.iter().enumerate() {
+            for set_idx in 0..set_count {
+                for disk in store
+                    .disk_set_inventory(DiskSetSelector::new(pool_idx, set_idx))
+                    .await?
+                    .into_iter()
+                    .flatten()
+                {
+                    info.disks.push(node_disk_snapshot(
+                        disk_endpoint_snapshot(&disk),
+                        disk.runtime_state().as_str(),
+                        online_hosts,
+                    ));
+                }
+            }
+        }
+        Ok(info)
+    })
+    .await
+    .map_err(|_| StorageError::Timeout)?
+}
+
+fn node_disk_snapshot(endpoint: Endpoint, runtime_state: &str, online_hosts: &HashSet<String>) -> Disk {
+    let reachable = endpoint.is_local || online_hosts.contains(&endpoint.host_port());
+    // Returning drives can still reject data I/O as faulty. Without a fresh
+    // disk-info probe, only an Online runtime observation can supply quorum.
+    let online = reachable && runtime_state == rustfs_madmin::ITEM_ONLINE;
+    Disk {
+        endpoint: endpoint.to_string(),
+        drive_path: endpoint.get_file_path(),
+        pool_index: endpoint.pool_idx,
+        set_index: endpoint.set_idx,
+        disk_index: endpoint.disk_idx,
+        state: if online { DISK_STATE_OK } else { "offline" }.to_string(),
+        runtime_state: Some(runtime_state.to_string()),
+        ..Default::default()
+    }
+}
+
 async fn collect_cluster_health_report_with<LoadFn, Fut>(
     kind: ClusterHealthProbeKind,
     mut load_report: LoadFn,
@@ -820,11 +906,22 @@ fn cluster_health_timeout_report() -> DependencyReadinessReport {
             peer_health_ready: false,
         },
         degraded_reasons: vec![ReadinessDegradedReason::ClusterHealthTimeout],
+        storage_details: None,
     }
 }
 
 async fn collect_node_readiness() -> DependencyReadiness {
-    collect_node_readiness_report().await.readiness
+    // Startup publication retains its local dependency gate. Runtime probes
+    // additionally report storage quorum without changing S3 admission.
+    let storage = node_pool_meta_write_readiness().await;
+    let readiness = DependencyReadiness {
+        storage_ready: storage.ready,
+        iam_ready: runtime_sources::current_iam_ready(),
+        lock_quorum_ready: collect_lock_quorum_status().await.ready,
+        peer_health_ready: collect_peer_health_readiness(),
+    };
+    record_readiness_report(&dependency_readiness_report_from_write_status(readiness, storage));
+    readiness
 }
 
 pub async fn collect_cluster_read_dependency_readiness_report() -> DependencyReadinessReport {
@@ -856,11 +953,15 @@ pub(crate) async fn snapshot_dependency_readiness_report() -> DependencyReadines
 }
 
 async fn collect_lock_quorum_status() -> LockQuorumStatus {
+    collect_lock_quorum_observation().await.status
+}
+
+async fn collect_lock_quorum_observation() -> LockQuorumObservation {
     if let Some(cached) = load_cached_lock_quorum_status().await {
         cached
     } else {
-        let computed = collect_lock_quorum_status_uncached().await;
-        update_lock_quorum_status_cache(computed).await;
+        let computed = collect_lock_quorum_observation_uncached().await;
+        update_lock_quorum_status_cache(computed.clone()).await;
         computed
     }
 }
@@ -972,20 +1073,27 @@ fn aggregate_lock_quorum_status(pool_endpoints: &EndpointServerPools, online_hos
 }
 
 async fn collect_lock_quorum_status_uncached() -> LockQuorumStatus {
+    collect_lock_quorum_observation_uncached().await.status
+}
+
+async fn collect_lock_quorum_observation_uncached() -> LockQuorumObservation {
     if !is_dist_erasure().await {
-        return LockQuorumStatus {
-            ready: true,
-            connected_clients: 1,
-            total_clients: 1,
-            required_quorum: 1,
+        return LockQuorumObservation {
+            status: LockQuorumStatus {
+                ready: true,
+                connected_clients: 1,
+                total_clients: 1,
+                required_quorum: 1,
+            },
+            online_hosts: HashSet::new(),
         };
     }
 
     let Some(pool_endpoints) = runtime_sources::current_endpoints_handle() else {
-        return LockQuorumStatus::default();
+        return LockQuorumObservation::default();
     };
     let Some(lock_clients) = runtime_sources::current_lock_clients_handle() else {
-        return LockQuorumStatus::default();
+        return LockQuorumObservation::default();
     };
 
     let online_hosts = futures::future::join_all(lock_clients.iter().map(|(host, client)| {
@@ -998,7 +1106,10 @@ async fn collect_lock_quorum_status_uncached() -> LockQuorumStatus {
     .filter_map(|(host, online)| online.then_some(host))
     .collect::<HashSet<_>>();
 
-    aggregate_lock_quorum_status(&pool_endpoints, &online_hosts)
+    LockQuorumObservation {
+        status: aggregate_lock_quorum_status(&pool_endpoints, &online_hosts),
+        online_hosts,
+    }
 }
 
 pub async fn wait_for_runtime_readiness_with<F, Fut, ReadyFn>(
@@ -1048,12 +1159,206 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::storage_api::server::readiness::{DiskOption, new_disk};
     use rustfs_madmin::{BackendInfo, Disk};
     use serial_test::serial;
     use std::future;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use temp_env::{async_with_vars, with_var};
 
+    #[derive(Debug, Default)]
+    struct RuntimeInventory {
+        backend: BackendInfo,
+        disks: HashMap<DiskSetSelector, Vec<Option<DiskStore>>>,
+        pending_set: Option<DiskSetSelector>,
+        failed_set: Option<DiskSetSelector>,
+    }
+
+    #[async_trait::async_trait]
+    impl StorageAdminApi for RuntimeInventory {
+        type BackendInfo = BackendInfo;
+        type StorageInfo = StorageInfo;
+        type Disk = DiskStore;
+        type Error = StorageError;
+
+        async fn backend_info(&self) -> BackendInfo {
+            self.backend.clone()
+        }
+
+        async fn storage_info(&self) -> StorageInfo {
+            panic!("node readiness must not perform a cluster storage-info probe")
+        }
+
+        async fn local_storage_info(&self) -> StorageInfo {
+            panic!("node readiness must not perform local disk-info I/O")
+        }
+
+        async fn disk_set_inventory(&self, selector: DiskSetSelector) -> Result<Vec<Option<DiskStore>>, StorageError> {
+            if self.pending_set == Some(selector) {
+                return future::pending().await;
+            }
+            if self.failed_set == Some(selector) {
+                return Err(StorageError::other("inventory unavailable"));
+            }
+            Ok(self.disks.get(&selector).cloned().unwrap_or_default())
+        }
+
+        fn set_drive_counts(&self) -> Vec<usize> {
+            self.backend.drives_per_set.clone()
+        }
+    }
+
+    async fn runtime_inventory(layouts: &[(usize, usize, usize)]) -> RuntimeInventory {
+        let mut store = RuntimeInventory::default();
+        for (pool_idx, &(set_count, drive_count, parity)) in layouts.iter().enumerate() {
+            store.backend.total_sets.push(set_count);
+            store.backend.drives_per_set.push(drive_count);
+            store.backend.standard_sc_data.push(drive_count - parity);
+            store.backend.standard_sc_parities.push(parity);
+            for set_idx in 0..set_count {
+                let mut disks = Vec::new();
+                for disk_idx in 0..drive_count {
+                    let endpoint = Endpoint {
+                        url: url::Url::parse(&format!("http://node-{disk_idx}:9000/pool-{pool_idx}-set-{set_idx}"))
+                            .expect("valid test endpoint"),
+                        is_local: false,
+                        pool_idx: i32::try_from(pool_idx).expect("test pool index"),
+                        set_idx: i32::try_from(set_idx).expect("test set index"),
+                        disk_idx: i32::try_from(disk_idx).expect("test disk index"),
+                    };
+                    disks.push(Some(
+                        new_disk(&endpoint, &DiskOption::default())
+                            .await
+                            .expect("create runtime disk handle"),
+                    ));
+                }
+                store.disks.insert(DiskSetSelector::new(pool_idx, set_idx), disks);
+            }
+        }
+        store
+    }
+
+    #[tokio::test]
+    async fn node_storage_snapshot_tracks_read_write_quorum_and_recovery() {
+        for (drive_count, parity) in [(4, 2), (4, 1), (8, 2)] {
+            let store = runtime_inventory(&[(1, drive_count, parity)]).await;
+            let data = drive_count - parity;
+            let write_quorum = data + usize::from(data == parity);
+            for survivors in (0..=drive_count).rev().chain(std::iter::once(drive_count)) {
+                let online_hosts = (0..survivors).map(|idx| format!("node-{idx}:9000")).collect();
+                let info = node_storage_snapshot(&store, &online_hosts)
+                    .await
+                    .expect("read local runtime inventory");
+                assert_eq!(info.disks.len(), drive_count, "offline members retain their topology slots");
+                assert_eq!(
+                    storage_read_ready_from_runtime_state(&info),
+                    survivors >= data,
+                    "layout={drive_count}/{parity}, survivors={survivors}"
+                );
+                assert_eq!(
+                    storage_ready_from_runtime_state(&info),
+                    survivors >= write_quorum,
+                    "layout={drive_count}/{parity}, survivors={survivors}"
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn node_disk_snapshot_requires_online_runtime_before_counting_recovered_drives() {
+        for is_local in [false, true] {
+            for reachable in [false, true] {
+                let online_hosts = if reachable {
+                    HashSet::from(["node-0:9000".to_string()])
+                } else {
+                    HashSet::new()
+                };
+                for runtime_state in ["online", "suspect", "offline", "returning", "unknown"] {
+                    let endpoint = Endpoint {
+                        url: url::Url::parse("http://node-0:9000/data").expect("valid test endpoint"),
+                        is_local,
+                        pool_idx: 0,
+                        set_idx: 0,
+                        disk_idx: 2,
+                    };
+                    let mut disks = online_readiness_disks(0, 2);
+                    disks.push(node_disk_snapshot(endpoint, runtime_state, &online_hosts));
+                    let info = StorageInfo {
+                        backend: BackendInfo {
+                            total_sets: vec![1],
+                            drives_per_set: vec![4],
+                            standard_sc_data: vec![2],
+                            standard_sc_parities: vec![2],
+                            ..Default::default()
+                        },
+                        disks,
+                        ..Default::default()
+                    };
+                    assert!(storage_read_ready_from_runtime_state(&info));
+                    assert_eq!(
+                        storage_ready_from_runtime_state(&info),
+                        runtime_state == "online" && (is_local || reachable),
+                        "runtime_state={runtime_state}, local={is_local}, reachable={reachable}"
+                    );
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn node_storage_snapshot_requires_each_configured_set_and_distinct_drives() {
+        let mut store = runtime_inventory(&[(2, 4, 2), (1, 8, 2)]).await;
+        let online_hosts = (0..8).map(|idx| format!("node-{idx}:9000")).collect();
+        let info = node_storage_snapshot(&store, &online_hosts)
+            .await
+            .expect("healthy mixed layout");
+        assert!(storage_ready_from_runtime_state(&info));
+
+        let selector = DiskSetSelector::new(0, 1);
+        let original = store.disks.remove(&selector).expect("second configured set");
+        let info = node_storage_snapshot(&store, &online_hosts)
+            .await
+            .expect("missing set snapshot");
+        assert!(!storage_read_ready_from_runtime_state(&info));
+        assert!(!storage_ready_from_runtime_state(&info));
+
+        store.disks.insert(selector, vec![original[0].clone(); 4]);
+        let info = node_storage_snapshot(&store, &online_hosts)
+            .await
+            .expect("duplicate drive snapshot");
+        assert!(!storage_read_ready_from_runtime_state(&info));
+        assert!(!storage_ready_from_runtime_state(&info));
+
+        store.disks.insert(selector, original);
+        assert!(storage_ready_from_runtime_state(
+            &node_storage_snapshot(&store, &online_hosts)
+                .await
+                .expect("restored inventory")
+        ));
+    }
+
+    #[tokio::test]
+    async fn node_storage_snapshot_bounds_waits_and_propagates_inventory_failure() {
+        let mut store = runtime_inventory(&[(1, 4, 2)]).await;
+        let online_hosts = (0..4).map(|idx| format!("node-{idx}:9000")).collect();
+        let selector = DiskSetSelector::new(0, 0);
+        store.pending_set = Some(selector);
+        let result = tokio::time::timeout(Duration::from_secs(1), node_storage_snapshot(&store, &online_hosts))
+            .await
+            .expect("node storage inspection must honor its own 100 ms budget");
+        assert!(matches!(result, Err(StorageError::Timeout)));
+
+        store.pending_set = None;
+        store.failed_set = Some(selector);
+        assert!(node_storage_snapshot(&store, &online_hosts).await.is_err());
+        store.failed_set = None;
+        assert!(storage_ready_from_runtime_state(
+            &node_storage_snapshot(&store, &online_hosts)
+                .await
+                .expect("inspection recovered")
+        ));
+    }
+
     fn online_readiness_disks(set_idx: i32, count: i32) -> Vec<Disk> {
         (0..count)
             .map(|disk_index| Disk {
@@ -1157,6 +1462,7 @@ mod tests {
                         peer_health_ready: true,
                     },
                     degraded_reasons: Vec::new(),
+                    storage_details: None,
                 };
 
                 let first_calls = calls.clone();
@@ -1213,6 +1519,7 @@ mod tests {
                             peer_health_ready: true,
                         },
                         degraded_reasons: Vec::new(),
+                        storage_details: None,
                     }
                 })
                 .await;
@@ -2013,24 +2320,17 @@ mod tests {
                 *guard = None;
             }
 
-            update_lock_quorum_status_cache(LockQuorumStatus {
-                ready: true,
-                connected_clients: 2,
-                total_clients: 3,
-                required_quorum: 2,
-            })
-            .await;
-
-            let cached = load_cached_lock_quorum_status().await;
-            assert_eq!(
-                cached,
-                Some(LockQuorumStatus {
+            let observation = LockQuorumObservation {
+                status: LockQuorumStatus {
                     ready: true,
                     connected_clients: 2,
                     total_clients: 3,
                     required_quorum: 2,
-                })
-            );
+                },
+                online_hosts: HashSet::from(["node-a:9000".to_owned(), "node-b:9000".to_owned()]),
+            };
+            update_lock_quorum_status_cache(observation.clone()).await;
+            assert_eq!(load_cached_lock_quorum_status().await, Some(observation));
         })
         .await;
     }

rustfs/src/shared_types.rs

@@ -45,6 +45,7 @@ pub enum ReadinessDegradedReason {
     ObjectWriteStalled,
     PoolMetaWriteBlocked,
     PoolMetadataCheckTimeout,
+    StorageReadinessCheckTimeout,
     ClusterHealthTimeout,
     PeerHealthUnavailable,
     StartupFinalizationPending,
@@ -65,6 +66,7 @@ impl ReadinessDegradedReason {
             ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled",
             ReadinessDegradedReason::PoolMetaWriteBlocked => "pool_meta_write_blocked",
             ReadinessDegradedReason::PoolMetadataCheckTimeout => "pool_metadata_check_timeout",
+            ReadinessDegradedReason::StorageReadinessCheckTimeout => "storage_readiness_check_timeout",
             ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout",
             ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable",
             ReadinessDegradedReason::StartupFinalizationPending => "startup_finalization_pending",
@@ -80,6 +82,14 @@ impl ReadinessDegradedReason {
 pub struct DependencyReadinessReport {
     pub readiness: DependencyReadiness,
     pub degraded_reasons: Vec<ReadinessDegradedReason>,
+    pub storage_details: Option<StorageReadinessDetails>,
+}
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct StorageReadinessDetails {
+    pub read_quorum_ready: bool,
+    pub write_quorum_ready: bool,
+    pub pool_metadata_write_ready: bool,
 }
 
 pub(crate) fn convert_ecstore_object_info(object: StorageObjectInfo) -> NotifyObjectInfo {

rustfs/src/storage/storage_api.rs

@@ -475,6 +475,8 @@ pub(crate) mod ecstore_disk {
         RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, SnapshotLeaseToken,
         UpdateMetadataOpts, VolumeInfo, WalkDirOptions, get_object_disk_read_timeout, validate_batch_read_version_item_count,
     };
+    #[cfg(test)]
+    pub(crate) use rustfs_ecstore::api::disk::{DiskOption, new_disk};
     pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce};
 }
 

rustfs/src/storage_api.rs

@@ -211,12 +211,19 @@ pub(crate) mod server {
     pub(crate) mod readiness {
         pub(crate) mod contract {
             pub(crate) mod admin {
-                pub(crate) use super::super::super::super::storage_contracts::StorageAdminApi;
+                pub(crate) use super::super::super::super::storage_contracts::{DiskSetSelector, StorageAdminApi};
             }
         }
 
+        pub(crate) use crate::storage::storage_api::ecstore_disk::DiskStore;
         pub(crate) use crate::storage::storage_api::{Endpoint, EndpointServerPools, is_dist_erasure};
 
+        pub(crate) fn disk_endpoint_snapshot(disk: &DiskStore) -> Endpoint {
+            crate::storage::storage_api::ecstore_disk::DiskAPI::endpoint(disk.as_ref())
+        }
+
+        #[cfg(test)]
+        pub(crate) use crate::storage::storage_api::ecstore_disk::{DiskOption, new_disk};
         #[cfg(test)]
         pub(crate) use crate::storage::storage_api::{Endpoints, PoolEndpoints};
     }