diff --git a/crates/e2e_test/src/namespace_lock_quorum_test.rs b/crates/e2e_test/src/namespace_lock_quorum_test.rs index 47b180cd7..8ff961321 100644 --- a/crates/e2e_test/src/namespace_lock_quorum_test.rs +++ b/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::().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(()) } diff --git a/docs/architecture/readiness-matrix.md b/docs/architecture/readiness-matrix.md index 75b180bb6..a6fccc77c 100644 --- a/docs/architecture/readiness-matrix.md +++ b/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. diff --git a/docs/operations/rolling-restart.md b/docs/operations/rolling-restart.md index ce185e2a2..6dbd4297a 100644 --- a/docs/operations/rolling-restart.md +++ b/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://:9000/health/ready | jq diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index cecbd9292..57d405a74 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/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, diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index 9d38302a8..fa72d8fbd 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/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); diff --git a/rustfs/src/server/health.rs b/rustfs/src/server/health.rs index 47f4357c1..ee1747263 100644 --- a/rustfs/src/server/health.rs +++ b/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); diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index e7122ba9c..27f5cd018 100644 --- a/rustfs/src/server/readiness.rs +++ b/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, } #[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 { +async fn load_cached_lock_quorum_status() -> Option { let ttl = health_readiness_cache_ttl(); if ttl.is_zero() { return None; @@ -448,13 +455,13 @@ async fn load_cached_lock_quorum_status() -> Option { 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(store: &S, online_hosts: &HashSet) -> Result +where + S: StorageAdminApi, +{ + 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) -> 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( 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::>(); - 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( @@ -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>>, + pending_set: Option, + failed_set: Option, + } + + #[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>, 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 { + 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 { (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; } diff --git a/rustfs/src/shared_types.rs b/rustfs/src/shared_types.rs index 7406a0978..3a872eb8c 100644 --- a/rustfs/src/shared_types.rs +++ b/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, + pub storage_details: Option, +} + +#[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 { diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 59ddade5e..4449c0dd7 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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}; } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 9aea57c03..088e61b03 100644 --- a/rustfs/src/storage_api.rs +++ b/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}; }