fix(cluster): clarify peer health and listing timeouts (#5086)

* fix(cluster): surface observed peer health

Refs rustfs/backlog#1387

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

* fix(admin): distinguish unreported peer health

Refs rustfs/backlog#1388

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

* fix(ecstore): decouple metacache peek timeout

Refs rustfs/backlog#1389

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

* docs(ops): diagnose metacache listing timeouts

Refs rustfs/backlog#1390

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

* refactor(cluster): clarify observed peer health

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

* fix(ecstore): route capability state through contract

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/config/src/constants/drive.rs

@@ -39,6 +39,11 @@ pub const DEFAULT_DRIVE_WALKDIR_TIMEOUT_SECS: u64 = 5;
 pub const ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS: &str = "RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS";
 pub const DEFAULT_DRIVE_WALKDIR_STALL_TIMEOUT_SECS: u64 = 5;
 
+/// Maximum time the metacache merge consumer waits for the next visible
+/// `walk_dir()` entry from a reader before detaching it from the merge.
+pub const ENV_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS: &str = "RUSTFS_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS";
+pub const DEFAULT_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS: u64 = 10;
+
 /// Interval in seconds between active health probes for local and remote drives.
 pub const ENV_DRIVE_ACTIVE_CHECK_INTERVAL_SECS: &str = "RUSTFS_DRIVE_ACTIVE_CHECK_INTERVAL_SECS";
 pub const DEFAULT_DRIVE_ACTIVE_CHECK_INTERVAL_SECS: u64 = 15;

crates/ecstore/src/cache_value/metacache_set.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::disk::disk_store::get_drive_walkdir_stall_timeout;
+use crate::disk::disk_store::{get_drive_walkdir_peek_timeout, get_drive_walkdir_stall_timeout};
 use crate::disk::error::DiskError;
 use crate::disk::{self, DiskAPI, DiskStore, WalkDirOptions};
 use metrics::counter;
@@ -175,6 +175,10 @@ pub(crate) enum TestReaderBehavior {
     ProducerError(DiskError),
     PrimaryErrorThenFallback(DiskError),
     PartialThenTimeout(Vec<MetaCacheEntry>),
+    DelayedEntries {
+        delay: Duration,
+        entries: Vec<MetaCacheEntry>,
+    },
 }
 
 #[cfg(test)]
@@ -339,6 +343,13 @@ async fn list_path_raw_inner(
                         drop(out);
                         Some(err)
                     }
+                    TestReaderBehavior::DelayedEntries { delay, entries } => {
+                        tokio::time::sleep(delay).await;
+                        let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
+                        out.write(&entries).await.expect("delayed test entries should be written");
+                        out.close().await.expect("delayed test entries should close");
+                        return Ok(());
+                    }
                 }
             } else {
                 None
@@ -473,6 +484,17 @@ async fn list_path_raw_inner(
                             record_producer_error(&producer_errs_clone, disk_idx, &err);
                             return Err(err);
                         }
+                        TestReaderBehavior::DelayedEntries { delay, entries } => {
+                            tokio::time::sleep(delay).await;
+                            let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
+                            out.write(&entries)
+                                .await
+                                .expect("delayed test fallback entries should be written");
+                            out.close().await.expect("delayed test fallback entries should close");
+                            need_fallback = false;
+                            last_err = None;
+                            continue;
+                        }
                     }
                 }
 
@@ -609,47 +631,23 @@ async fn list_path_raw_inner(
         // Consumer-side peek timeout, and a caveat worth understanding
         // (rustfs/backlog#1217).
         //
-        // This budget is the SAME SOURCE and SAME VALUE as the producer-side
+        // This budget must not be stricter than the producer-side walk stall
-        // walk stall budget: both default to `walkdir_stall_timeout` and fall
+        // budget. The two measure different things:
-        // back to `get_drive_walkdir_stall_timeout()` (5s). But the two measure
-        // different things:
         //   * producer stall: bounds a single drive READ inside the walk (see
         //     `with_walk_stall_deadline` in `disk/local.rs`);
         //   * this consumer peek: bounds the interval between two ADJACENT
         //     entries arriving from a drive's reader (`peek_with_timeout`).
         //
-        // Because they are coupled to the same value, the consumer cannot wait
+        // A drive can spend one stall budget walking a dense non-listable region
-        // meaningfully longer for the next entry than the producer is allowed to
+        // before it can publish the next visible entry. Use an independent,
-        // spend producing one. When a drive walks a region dense with
+        // profile-aware consumer budget so the merge does not detach that reader
-        // non-listable internal items (many entries the producer filters out
+        // before the producer itself would have failed.
-        // before emitting the next visible one), a HEALTHY drive can take longer
+        let producer_stall_timeout = opts.walkdir_stall_timeout.unwrap_or_else(get_drive_walkdir_stall_timeout);
-        // than one budget to hand the consumer its next entry. The consumer then
+        let configured_peek_timeout = get_drive_walkdir_peek_timeout().max(producer_stall_timeout);
-        // classifies it as `PeekOutcome::TimedOut` and DETACHES that reader (it
+        #[cfg(not(test))]
-        // is replaced by a drained duplex below), dropping a good drive from the
+        let peek_timeout = configured_peek_timeout;
-        // merge. That caps the "large prefix always succeeds" guarantee: a wide
+        #[cfg(test)]
-        // enough non-listable stretch can knock healthy drives out of quorum.
+        let peek_timeout = opts.peek_timeout.unwrap_or(configured_peek_timeout);
-        //
-        // This is left documented, not decoupled. Giving the consumer peek an
-        // independent, strictly-larger budget would reduce these false detaches,
-        // but it also delays detaching a genuinely dead drive by the same amount
-        // and shifts listing tail-latency semantics; that trade-off wants soak
-        // data before it changes the default, so it is deferred to a follow-up.
-        // The constraint for any such change: the consumer peek must be >= the
-        // producer stall (never stricter), so it can never declare a drive
-        // stalled before the producer itself would have failed.
-        let peek_timeout = opts
-            .walkdir_stall_timeout
-            .or({
-                #[cfg(test)]
-                {
-                    opts.peek_timeout
-                }
-                #[cfg(not(test))]
-                {
-                    None
-                }
-            })
-            .unwrap_or_else(get_drive_walkdir_stall_timeout);
         let mut errs: Vec<Option<DiskError>> = Vec::with_capacity(readers.len());
         for _ in 0..readers.len() {
             errs.push(None);
@@ -1246,6 +1244,57 @@ mod tests {
         assert_eq!(err, DiskError::Timeout);
     }
 
+    #[tokio::test]
+    async fn list_path_raw_waits_past_producer_stall_for_slow_progressing_reader() {
+        let entry = MetaCacheEntry {
+            name: "bucket/visible-object".to_string(),
+            metadata: vec![1, 2, 3],
+            cached: None,
+            reusable: false,
+        };
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let seen_clone = seen.clone();
+
+        timeout(
+            Duration::from_secs(1),
+            list_path_raw(
+                CancellationToken::new(),
+                ListPathRawOptions {
+                    disks: vec![None, None],
+                    min_disks: 2,
+                    walkdir_stall_timeout: Some(Duration::from_millis(20)),
+                    test_reader_behaviors: vec![
+                        TestReaderBehavior::DelayedEntries {
+                            delay: Duration::from_millis(60),
+                            entries: vec![entry.clone()],
+                        },
+                        TestReaderBehavior::Entries(vec![entry]),
+                    ],
+                    partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<DiskError>]| {
+                        let seen = seen_clone.clone();
+                        Box::pin(async move {
+                            let mut names = entries.0.iter().flatten().map(|entry| entry.name.as_str());
+                            if let (Some(first), Some(second), None) = (names.next(), names.next(), names.next())
+                                && first == second
+                            {
+                                seen.lock().expect("seen mutex poisoned").push(first.to_owned());
+                            }
+                        })
+                    })),
+                    ..Default::default()
+                },
+            ),
+        )
+        .await
+        .expect("slow-progressing reader should complete inside the consumer peek budget")
+        .expect("slow-progressing reader should not be detached by the producer stall budget");
+
+        assert_eq!(
+            seen.lock().expect("seen mutex poisoned").as_slice(),
+            &["bucket/visible-object".to_string()]
+        );
+    }
+
     #[tokio::test]
     async fn list_path_raw_prefers_timeout_for_mixed_errors_beyond_quorum_budget() {
         let err = list_path_raw(

crates/ecstore/src/cluster/control_plane.rs

@@ -32,6 +32,9 @@ const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by
 const NUMA_NOT_WIRED: &str = "NUMA topology not wired into runtime";
 const PROFILING_NOT_WIRED: &str = "profiling capability not wired into ECStore";
 const PEER_HEALTH_NOT_REPORTED: &str = "peer health not reported by endpoints";
+const PEER_HEALTH_LOCAL_NODE: &str = "local node does not require peer health probing";
+const PEER_HEALTH_REACHABLE: &str = "peer marked reachable by internode health tracker";
+const PEER_HEALTH_UNREACHABLE: &str = "peer marked unreachable by internode health tracker";
 const CONTROL_RPC_SEPARATED: &str = "control RPC remains on the gRPC control plane";
 const DATA_STREAM_RPC_SEPARATED: &str = "remote disk data streams remain on the internode data transport";
 
@@ -341,12 +344,24 @@ pub fn peer_health_snapshot_from_membership(membership: &ClusterMembershipSnapsh
             .map(|node| ClusterPeerHealth {
                 node_id: node.node_id.clone(),
                 is_local: node.is_local,
-                status: CapabilityStatus::unknown().with_reason(PEER_HEALTH_NOT_REPORTED),
+                status: peer_health_status_for_node(node),
             })
             .collect(),
     }
 }
 
+fn peer_health_status_for_node(node: &ClusterNodeMembership) -> CapabilityStatus {
+    if node.is_local {
+        return CapabilityStatus::supported().with_reason(PEER_HEALTH_LOCAL_NODE);
+    }
+
+    match rustfs_io_metrics::internode_metrics::cluster_peer_observed_online_status(&node.grid_host) {
+        Some(true) => CapabilityStatus::supported().with_reason(PEER_HEALTH_REACHABLE),
+        Some(false) => CapabilityStatus::unknown().with_reason(PEER_HEALTH_UNREACHABLE),
+        None => CapabilityStatus::disabled().with_reason(PEER_HEALTH_NOT_REPORTED),
+    }
+}
+
 pub fn rpc_boundary_snapshot() -> ClusterRpcBoundarySnapshot {
     ClusterRpcBoundarySnapshot {
         control_channels: ["metadata", "lock", "health", "administrative"]
@@ -498,6 +513,7 @@ mod tests {
     use std::collections::BTreeSet;
 
     use crate::layout::endpoints::{Endpoints, PoolEndpoints};
+    use crate::storage_api_contracts::topology::CapabilityState;
 
     #[test]
     fn topology_snapshot_maps_endpoint_sets_without_local_paths() {
@@ -588,16 +604,55 @@ mod tests {
     }
 
     #[test]
-    fn peer_health_snapshot_reports_static_unknown_status() {
+    fn peer_health_snapshot_reports_observed_peer_status() {
-        let membership = membership_snapshot_from_endpoint_pools(&sample_url_endpoint_pools());
+        let online_host = "http://cluster-control-plane-peer-online-test:9000";
+        let offline_host = "http://cluster-control-plane-peer-offline-test:9000";
+        rustfs_io_metrics::internode_metrics::record_peer_reachable(online_host);
+        rustfs_io_metrics::internode_metrics::record_peer_unreachable(offline_host, 1);
+        let membership = ClusterMembershipSnapshot {
+            nodes: vec![
+                ClusterNodeMembership {
+                    node_id: LOCAL_NODE_ID.to_string(),
+                    grid_host: String::new(),
+                    is_local: true,
+                    pools: vec![0],
+                },
+                ClusterNodeMembership {
+                    node_id: "cluster-control-plane-peer-online-test:9000".to_string(),
+                    grid_host: online_host.to_string(),
+                    is_local: false,
+                    pools: vec![0],
+                },
+                ClusterNodeMembership {
+                    node_id: "cluster-control-plane-peer-offline-test:9000".to_string(),
+                    grid_host: offline_host.to_string(),
+                    is_local: false,
+                    pools: vec![0],
+                },
+                ClusterNodeMembership {
+                    node_id: "cluster-control-plane-peer-unknown-test:9000".to_string(),
+                    grid_host: "http://cluster-control-plane-peer-unknown-test:9000".to_string(),
+                    is_local: false,
+                    pools: vec![0],
+                },
+            ],
+            drives: Vec::new(),
+        };
         let snapshot = peer_health_snapshot_from_membership(&membership);
 
-        assert_eq!(snapshot.peers.len(), 2);
+        assert_eq!(snapshot.peers.len(), 4);
-        assert_eq!(snapshot.peers[0].node_id, "node1.example:9000");
+        assert_eq!(snapshot.peers[0].node_id, LOCAL_NODE_ID);
         assert!(snapshot.peers[0].is_local);
-        assert_eq!(snapshot.peers[0].status.reason.as_deref(), Some(PEER_HEALTH_NOT_REPORTED));
+        assert!(snapshot.peers[0].status.state.is_supported());
-        assert_eq!(snapshot.peers[1].node_id, "node2.example:9000");
+        assert_eq!(snapshot.peers[0].status.reason.as_deref(), Some(PEER_HEALTH_LOCAL_NODE));
+        assert_eq!(snapshot.peers[1].node_id, "cluster-control-plane-peer-online-test:9000");
         assert!(!snapshot.peers[1].is_local);
+        assert!(snapshot.peers[1].status.state.is_supported());
+        assert_eq!(snapshot.peers[1].status.reason.as_deref(), Some(PEER_HEALTH_REACHABLE));
+        assert_eq!(snapshot.peers[2].status.reason.as_deref(), Some(PEER_HEALTH_UNREACHABLE));
+        assert!(!snapshot.peers[2].status.state.is_supported());
+        assert_eq!(snapshot.peers[3].status.reason.as_deref(), Some(PEER_HEALTH_NOT_REPORTED));
+        assert_eq!(snapshot.peers[3].status.state, CapabilityState::Disabled);
     }
 
     #[test]

crates/ecstore/src/disk/disk_store.rs

@@ -209,6 +209,16 @@ pub fn get_drive_walkdir_stall_timeout() -> Duration {
     )
 }
 
+pub fn get_drive_walkdir_peek_timeout() -> Duration {
+    let stall_timeout = get_drive_walkdir_stall_timeout();
+    let configured = get_drive_timeout_duration(
+        rustfs_config::ENV_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS,
+        rustfs_config::DEFAULT_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS,
+        Some(rustfs_config::DRIVE_TIMEOUT_PROFILE_HIGH_LATENCY_SECS.saturating_mul(2)),
+    );
+    configured.max(stall_timeout)
+}
+
 pub fn get_object_disk_read_timeout() -> Duration {
     get_drive_timeout_duration(
         rustfs_config::ENV_OBJECT_DISK_READ_TIMEOUT,
@@ -1656,6 +1666,64 @@ mod tests {
         });
     }
 
+    #[test]
+    fn drive_walkdir_peek_timeout_uses_wider_default_when_unset() {
+        temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS, || {
+            temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, || {
+                temp_env::with_var_unset(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, || {
+                    temp_env::with_var_unset(rustfs_config::ENV_DRIVE_TIMEOUT_PROFILE, || {
+                        assert_eq!(
+                            get_drive_walkdir_peek_timeout(),
+                            Duration::from_secs(rustfs_config::DEFAULT_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS)
+                        );
+                    });
+                });
+            });
+        });
+    }
+
+    #[test]
+    fn drive_walkdir_peek_timeout_is_never_stricter_than_stall_timeout() {
+        temp_env::with_var(rustfs_config::ENV_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS, Some("3"), || {
+            temp_env::with_var(rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, Some("13"), || {
+                temp_env::with_var_unset(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, || {
+                    assert_eq!(get_drive_walkdir_peek_timeout(), Duration::from_secs(13));
+                });
+            });
+        });
+    }
+
+    #[test]
+    fn drive_walkdir_peek_timeout_prefers_canonical_over_legacy() {
+        temp_env::with_var(rustfs_config::ENV_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS, Some("23"), || {
+            temp_env::with_var(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("17"), || {
+                temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, || {
+                    assert_eq!(get_drive_walkdir_peek_timeout(), Duration::from_secs(23));
+                });
+            });
+        });
+    }
+
+    #[test]
+    fn drive_walkdir_peek_timeout_uses_high_latency_profile_default() {
+        temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS, || {
+            temp_env::with_var_unset(rustfs_config::ENV_DRIVE_WALKDIR_STALL_TIMEOUT_SECS, || {
+                temp_env::with_var_unset(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, || {
+                    temp_env::with_var(
+                        rustfs_config::ENV_DRIVE_TIMEOUT_PROFILE,
+                        Some(rustfs_config::DRIVE_TIMEOUT_PROFILE_HIGH_LATENCY),
+                        || {
+                            assert_eq!(
+                                get_drive_walkdir_peek_timeout(),
+                                Duration::from_secs(rustfs_config::DRIVE_TIMEOUT_PROFILE_HIGH_LATENCY_SECS * 2)
+                            );
+                        },
+                    );
+                });
+            });
+        });
+    }
+
     #[test]
     fn drive_walkdir_timeout_prefers_canonical_over_legacy() {
         temp_env::with_var(rustfs_config::ENV_DRIVE_WALKDIR_TIMEOUT_SECS, Some("11"), || {

crates/ecstore/src/storage_api_contracts/mod.rs

@@ -127,4 +127,7 @@ pub(crate) mod topology {
         CapabilityStatus, DiskCapabilities, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, TopologySet,
         TopologySnapshot,
     };
+
+    #[cfg(test)]
+    pub(crate) use rustfs_storage_api::CapabilityState;
 }

crates/io-metrics/src/internode_metrics.rs

@@ -514,6 +514,12 @@ pub fn cluster_peer_is_offline(addr: &str) -> bool {
     peers.get(normalize_peer_key(addr)).map(|peer| !peer.online).unwrap_or(false)
 }
 
+/// Return the last observed online/offline state for a peer, if this process has observed one.
+pub fn cluster_peer_observed_online_status(addr: &str) -> Option<bool> {
+    let peers = CLUSTER_PEER_HEALTH.read().unwrap_or_else(|poisoned| poisoned.into_inner());
+    peers.get(normalize_peer_key(addr)).map(|peer| peer.online)
+}
+
 /// Decide whether to fast-fail (bypass) an offline peer instead of attempting to reach it
 /// (grpc-optimization P3 offline bypass). Returns `true` to bypass.
 ///
@@ -782,6 +788,18 @@ mod tests {
         assert!(!cluster_peer_is_offline(bare), "reachable via one form must mark the shared entry online");
     }
 
+    #[test]
+    fn cluster_peer_observed_online_status_reports_known_and_unknown_peers() {
+        let addr = "http://cluster-peer-snapshot-status-test:9000";
+        assert_eq!(cluster_peer_observed_online_status(addr), None);
+
+        record_peer_reachable(addr);
+        assert_eq!(cluster_peer_observed_online_status(addr), Some(true));
+
+        record_peer_unreachable(addr, 1);
+        assert_eq!(cluster_peer_observed_online_status(addr), Some(false));
+    }
+
     #[test]
     fn cluster_servers_offline_total_name_is_stable() {
         assert_eq!(CLUSTER_SERVERS_OFFLINE_TOTAL, "rustfs_cluster_servers_offline_total");

crates/log-analyzer/src/rules/seed/disk.rs

@@ -96,6 +96,23 @@ pub(super) fn rules() -> Vec<Rule> {
                 "定位具体盘(结合 samples 的 fields),检查硬件。",
             )
         },
+        Rule {
+            evidence_fields: strings(["bucket", "path", "state", "timeout_ms", "drive"]),
+            anchors: strings(["Metacache listing quorum failed", "Metacache reader peek timed out"]),
+            min_count: 2,
+            ..base(
+                "metacache-listing-timeout",
+                P2Degraded,
+                "disk",
+                "metacache listing 多盘等待超时",
+                any([
+                    all([contains("Metacache listing quorum failed"), field("state", "quorum_failed")]),
+                    all([contains("Metacache reader peek timed out"), field("state", "peek_timed_out")]),
+                ]),
+                "对象 listing/scanner 期间 metacache reader 等待超时并可能导致 listing quorum failure;常见于宽前缀、高延迟盘或慢远端 reader。",
+                "先确认是否存在宽前缀/慢盘/高延迟远端 reader;再参考 docs/operations/drive-timeout-tuning.md 调整 RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS、RUSTFS_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS 或 RUSTFS_DRIVE_TIMEOUT_PROFILE=high_latency。",
+            )
+        },
         Rule {
             anchors: strings(["Unformatted disk"]),
             ..base(

crates/log-analyzer/src/rules/seed/tests.rs

@@ -73,7 +73,7 @@ fn msg(message: &'static str) -> Sample {
 #[test]
 fn seed_rule_set_is_valid_and_complete() {
     let set = seed_rule_set();
-    assert_eq!(set.rules().len(), 68);
+    assert_eq!(set.rules().len(), 69);
     // Every message-matching rule carries at least one anchor for the CI
     // guard; the only anchor-less rules are the two structural matchers
     // (panic kind, scanner target+level).
@@ -113,6 +113,14 @@ fn every_rule_has_a_positive_sample() {
             msg("reporting peer disks offline after consecutive storage_info failures"),
         ),
         ("drive-faulty-error", msg("remote drive is faulty")),
+        (
+            "metacache-listing-timeout",
+            Sample {
+                message: "Metacache listing quorum failed",
+                fields: &[("state", "quorum_failed"), ("bucket", "photos"), ("path", "wide/")],
+                ..Default::default()
+            },
+        ),
         ("unformatted-disk", msg("Unformatted disk found")),
         ("disk-access-denied", msg("disk access denied: /data/disk1")),
         ("inconsistent-drive", msg("inconsistent drive found")),
@@ -308,6 +316,23 @@ fn smoke_samples_hit_exact_rule_sets() {
         &["disk-marked-faulty"],
     );
     exact(&msg("erasure write quorum (required=8, achieved=5)"), &["ec-write-quorum"]);
+    exact(
+        &Sample {
+            message: "Metacache listing quorum failed",
+            fields: &[("state", "quorum_failed"), ("bucket", "photos"), ("path", "wide/")],
+            ..Default::default()
+        },
+        &["metacache-listing-timeout"],
+    );
+    exact(
+        &Sample {
+            message: "S3Error InvalidAccessKeyId: Access Key Id you provided does not exist",
+            fields: &[("code", "InvalidAccessKeyId")],
+            ..Default::default()
+        },
+        &["unknown-access-key"],
+    );
+    exact(&msg("HTTP transport failed: BrokenPipe while streaming response body"), &[]);
     // Internode auth failures legitimately hit both the internode rule and
     // the generic client signature rule.
     exact(

docs/operations/drive-timeout-tuning.md

@@ -56,6 +56,7 @@ Explicit per-operation overrides always win over the profile, so you can select
 | Environment variable | Default | `high_latency` default | Bounds |
 |---|---:|---:|---|
 | `RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS` | `5` | `60` | Max time a single walk filesystem call may go without answering during a listing walk. **The knob for listing failures on large prefixes.** |
+| `RUSTFS_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS` | `10` | `120` | Max time the metacache merge consumer waits for the next visible entry from a walk reader. Values below the resolved stall timeout are clamped up to the stall timeout. |
 | `RUSTFS_DRIVE_WALKDIR_TIMEOUT_SECS` | `5` | `60` | Total wall-clock timeout for a `walk_dir`. Retained for non-foreground callers; the foreground listing path no longer uses it (see below). |
 | `RUSTFS_DRIVE_LIST_DIR_TIMEOUT_SECS` | `5` | `60` | Timeout for a standalone `list_dir` metadata listing. |
 | `RUSTFS_DRIVE_METADATA_TIMEOUT_SECS` | `5` | `60` | Timeout for metadata reads such as `read_metadata`. |
@@ -118,6 +119,10 @@ Raise the walk stall budget, or select the high-latency profile:
 # widen just the listing walk budget
 -e RUSTFS_DRIVE_WALKDIR_STALL_TIMEOUT_SECS=60
 
+# widen the metacache reader wait budget when drives keep progressing but do
+# not publish visible entries quickly enough for the merge consumer
+-e RUSTFS_DRIVE_WALKDIR_PEEK_TIMEOUT_SECS=120
+
 # or raise every profile-aware drive default at once
 -e RUSTFS_DRIVE_TIMEOUT_PROFILE=high_latency
 ```

rustfs/src/admin/handlers/cluster_snapshot.rs

@@ -564,8 +564,17 @@ fn summarize_peer_health(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus
         .iter()
         .filter(|peer| peer.status.state == CapabilityState::Unknown)
         .count();
+    let disabled = snapshot
+        .peer_health
+        .peers
+        .iter()
+        .filter(|peer| peer.status.state == CapabilityState::Disabled)
+        .count();
     if unknown > 0 {
         CapabilityStatus::unknown().with_reason(format!("cluster peer health has {unknown} unresolved peers"))
+    } else if disabled > 0 {
+        let noun = if disabled == 1 { "peer" } else { "peers" };
+        CapabilityStatus::disabled().with_reason(format!("cluster peer health is not reported by {disabled} {noun}"))
     } else {
         CapabilityStatus::supported().with_reason("cluster peer health resolved for all peers")
     }
@@ -827,6 +836,56 @@ mod tests {
         assert_eq!(summary.actionable_pressure.state, CapabilityState::Disabled);
     }
 
+    #[test]
+    fn cluster_snapshot_summary_treats_not_reported_peer_health_as_disabled() {
+        let snapshot = ClusterReadOnlySnapshot {
+            topology: TopologySnapshot::default(),
+            membership: ClusterMembershipSnapshot {
+                nodes: vec![ClusterNodeMembership {
+                    node_id: "node-a".to_string(),
+                    grid_host: "node-a:9000".to_string(),
+                    is_local: true,
+                    pools: vec![0],
+                }],
+                drives: Vec::new(),
+            },
+            pool_state: ClusterPoolStateSnapshot::default(),
+            local_storage: ClusterLocalNodeStorageSnapshot::default(),
+            peer_health: ClusterPeerHealthSnapshot {
+                peers: vec![ClusterPeerHealth {
+                    node_id: "node-a".to_string(),
+                    is_local: true,
+                    status: CapabilityStatus::disabled().with_reason("peer health not reported by endpoints"),
+                }],
+            },
+            rpc_boundary: sample_rpc_boundary_snapshot(),
+            observability: ObservabilitySnapshot::default(),
+            workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new(
+                WorkloadClass::ForegroundRead,
+                AdmissionState::Open,
+            )]),
+            runtime_status: ClusterRuntimeStatusSnapshot {
+                readiness: DependencyReadiness {
+                    storage_ready: true,
+                    iam_ready: true,
+                    lock_quorum_ready: true,
+                    peer_health_ready: true,
+                },
+                state: ClusterRuntimeReadinessState::Ready,
+                degraded_reasons: Vec::new(),
+            },
+        };
+
+        let summary = ClusterSnapshotSummary::from(&snapshot);
+
+        assert_eq!(summary.peer_health.state, CapabilityState::Disabled);
+        assert_eq!(
+            summary.peer_health.reason.as_deref(),
+            Some("cluster peer health is not reported by 1 peer")
+        );
+        assert_eq!(summary.actionable_pressure.state, CapabilityState::Disabled);
+    }
+
     fn sample_rpc_boundary_snapshot() -> ClusterRpcBoundarySnapshot {
         ClusterRpcBoundarySnapshot {
             control_channels: vec![ClusterRpcChannelSnapshot {