fix(ecstore): read pool metadata authority without the save gate

Health inferred pool metadata write authority from acquiring the local
authority gate within 100 ms. Sharing the gate among read-only
admissions still left held writers, and writers queued behind a reader
parked on the pool.bin fence, turning into cluster-write 503s.

Each writer now publishes the authority it leaves before releasing the
gate, and health reads that published state without acquiring it.
Transaction latches stay shared with the published copy; recovery
clears the latch and its failure under one lock so inspection never
observes a half-cleared state. The inspection-timeout path that can no
longer occur (check_timed_out, the check_timeout admin state, the
pool_metadata_check_timeout reason, the node timeout cache and its
metric) is removed.

crates/ecstore/src/cluster/rpc/peer_s3_client.rs

@@ -18,7 +18,7 @@ use crate::cluster::rpc::client::{
     node_service_time_out_client,
 };
 use crate::cluster::rpc::set_tonic_mutation_body_digest;
-use crate::core::pools::{PoolMeta, PoolMetaWriteState};
+use crate::core::pools::{PoolMeta, PoolMetaSaveGate, PoolMetaSaveGateWriteGuard};
 use crate::disk::error::DiskError;
 use crate::disk::error::{Error, Result};
 use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
@@ -798,12 +798,12 @@ fn try_acquire_bucket_heal_movement_guard<'a>(
 }
 
 async fn acquire_bucket_heal_write_guard<'a>(
-    gate: Option<&'a tokio::sync::Mutex<PoolMetaWriteState>>,
-) -> Result<Option<tokio::sync::MutexGuard<'a, PoolMetaWriteState>>> {
+    gate: Option<&'a PoolMetaSaveGate>,
+) -> Result<Option<PoolMetaSaveGateWriteGuard<'a>>> {
     let Some(gate) = gate else {
         return Ok(None);
     };
-    let guard = gate.lock().await;
+    let guard = gate.write().await;
     guard
         .ensure_write_safe("bucket heal cannot run while pool metadata requires recovery")
         .map_err(|e| e.narrow_to_disk().unwrap_or_else(Error::other))?;
@@ -2388,8 +2388,8 @@ mod tests {
 
     #[tokio::test]
     async fn local_bucket_heal_refuses_receiver_with_unsafe_pool_metadata() {
-        let gate = tokio::sync::Mutex::new(PoolMetaWriteState::default());
-        gate.lock().await.observe_replicas(PoolMetaReplicaState {
+        let gate = PoolMetaSaveGate::default();
+        gate.write().await.observe_replicas(PoolMetaReplicaState {
             needs_repair: true,
             repair_write_safe: false,
         });

crates/ecstore/src/core/pool_meta_native_integration_tests.rs

@@ -628,7 +628,7 @@ async fn pool_meta_native_physical_runtime_fork_after_apply_latches_original_gat
     let fork = identity_epoch_fork(&fixture).await;
     let since = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed original failure timestamp should be valid");
     {
-        let mut state = fixture.store.pool_meta_save_gate.lock().await;
+        let mut state = fixture.store.pool_meta_save_gate.write().await;
         assert_eq!(state.active_transactions.load(Ordering::SeqCst), 0);
         state.write_blocked = false;
         state.block_context = None;
@@ -685,7 +685,7 @@ async fn pool_meta_native_physical_runtime_fork_after_apply_latches_original_gat
     );
     assert_committed(&fixture).await;
     {
-        let state = fixture.store.pool_meta_save_gate.lock().await;
+        let state = fixture.store.pool_meta_save_gate.write().await;
         assert!(
             state.write_blocked,
             "candidate validation must propagate its permanent latch to the original state"
@@ -706,7 +706,7 @@ async fn pool_meta_native_physical_runtime_fork_after_apply_latches_original_gat
             .expect("permanently blocked recovery should return without mutation"),
         "ordinary-readable repaired heads must not clear the original known-fork latch"
     );
-    let state = fixture.store.pool_meta_save_gate.lock().await;
+    let state = fixture.store.pool_meta_save_gate.write().await;
     assert!(state.write_blocked);
     assert_eq!(
         state
@@ -979,7 +979,7 @@ async fn pool_meta_native_heal_v1_with_all_upgrade_gates_disabled_keeps_exact_ra
         async {
             let (fixture, raw) = legacy_native_owner_fixture(POOL_META_V1_VERSION, false).await;
             let guard = owner(&fixture.store).await;
-            let mut state = fixture.store.pool_meta_save_gate.lock().await;
+            let mut state = fixture.store.pool_meta_save_gate.write().await;
             let result = tokio::time::timeout(
                 std::time::Duration::from_secs(30),
                 fixture.store.heal_pool_meta_replica_under_fence(0, guard, &mut state),
@@ -1036,7 +1036,7 @@ async fn pool_meta_native_heal_rejects_legacy_source_with_trailing_bytes_without
                 let before = snapshots(&fixture.store).await;
                 let identities = owner_repair_identity_heads(&fixture.store).await;
                 let guard = owner(&fixture.store).await;
-                let mut state = fixture.store.pool_meta_save_gate.lock().await;
+                let mut state = fixture.store.pool_meta_save_gate.write().await;
                 let result = tokio::time::timeout(
                     std::time::Duration::from_secs(30),
                     fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
@@ -1110,7 +1110,7 @@ async fn pool_meta_native_heal_mixed_v1_v2_preserves_observed_format_floor() {
                     assert_eq!(selection.revision.version, POOL_META_VERSION, "the observed V2 format floor must be retained");
                     assert_eq!(selection.canonical.as_deref(), Some(semantic_v2.as_slice()));
                     let guard = owner(&fixture.store).await;
-                    let mut state = fixture.store.pool_meta_save_gate.lock().await;
+                    let mut state = fixture.store.pool_meta_save_gate.write().await;
                     tokio::time::timeout(std::time::Duration::from_secs(30),
                         fixture.store.heal_pool_meta_replica_under_fence(target_pool, guard, &mut state))
                         .await.expect("mixed legacy heal must terminate")
@@ -1199,7 +1199,7 @@ async fn pool_meta_native_heal_rejects_v1_and_v3_prepare_in_either_pool_order()
                 let before = snapshots(&fixture.store).await;
                 let identities = owner_repair_identity_heads(&fixture.store).await;
                 let guard = owner(&fixture.store).await;
-                let mut state = fixture.store.pool_meta_save_gate.lock().await;
+                let mut state = fixture.store.pool_meta_save_gate.write().await;
                 let result = tokio::time::timeout(
                     std::time::Duration::from_secs(30),
                     fixture
@@ -1301,7 +1301,7 @@ async fn pool_meta_native_heal_rejects_v1_legacy_dont_save_true_without_writes()
             let before = snapshots(&fixture.store).await;
             let identities = owner_repair_identity_heads(&fixture.store).await;
             let guard = owner(&fixture.store).await;
-            let mut state = fixture.store.pool_meta_save_gate.lock().await;
+            let mut state = fixture.store.pool_meta_save_gate.write().await;
             let result = tokio::time::timeout(
                 std::time::Duration::from_secs(30),
                 fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),
@@ -1410,7 +1410,7 @@ async fn pool_meta_native_heal_rejects_target_v1_dont_save_true_without_writes()
             let before = snapshots(&fixture.store).await;
             let identities = owner_repair_identity_heads(&fixture.store).await;
             let guard = owner(&fixture.store).await;
-            let mut state = fixture.store.pool_meta_save_gate.lock().await;
+            let mut state = fixture.store.pool_meta_save_gate.write().await;
             let result = tokio::time::timeout(
                 std::time::Duration::from_secs(30),
                 fixture.store.heal_pool_meta_replica_under_fence(1, guard, &mut state),

crates/ecstore/src/core/pools_test.rs

@@ -527,8 +527,8 @@ mod decommission_lock_order_tests {
             rebalance_meta: tokio::sync::RwLock::new(other_store.rebalance_meta.read().await.clone()),
             decommission_cancelers: tokio::sync::RwLock::new(vec![None; other_store.pools.len()]),
             start_gate: tokio::sync::Mutex::new(()),
-            pool_meta_save_gate: tokio::sync::Mutex::new(
-                other_store.pool_meta_save_gate.lock().await.independent_clone_for_test(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(
+                other_store.pool_meta_save_gate.write().await.independent_clone_for_test(),
             ),
             ctx,
             bucket_fence_registry: Arc::default(),

crates/ecstore/src/data_usage/mod.rs

@@ -3436,7 +3436,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: TokioMutex::new(()),
-            pool_meta_save_gate: TokioMutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
         })

crates/ecstore/src/services/rebalance/control.rs

@@ -593,7 +593,7 @@ impl ECStore {
         #[cfg(test)]
         crate::core::pools::observe_pool_activation_start_attempt(crate::core::pools::PoolActivationStartKind::Rebalance);
         let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
-        let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
+        let mut pool_meta_guard = self.pool_meta_save_gate.write().await;
         pool_meta_guard.ensure_write_safe(stage)?;
         let activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), fleet_proof).await?;
         let pool_meta = self
@@ -625,7 +625,7 @@ impl ECStore {
     {
         // Lock order: pool_meta_save_gate -> rebalance.bin -> pool.bin. These
         // shared fences admit a peer without waiting for the run's own entries.
-        let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
+        let mut pool_meta_guard = self.pool_meta_save_gate.write().await;
         pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
         let mut worker_fence = acquire_pool_rebalance_worker_locks(pool.clone()).await?;
         let pool_meta = self
@@ -654,7 +654,7 @@ impl ECStore {
     {
         // The admission read fences must already be dropped. Reacquire in the
         // original order and revalidate every durable boundary across that gap.
-        let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
+        let mut pool_meta_guard = self.pool_meta_save_gate.write().await;
         pool_meta_guard.ensure_write_safe("rebalance worker metadata update")?;
         let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
         let pool_meta = self
@@ -1714,7 +1714,7 @@ mod tests {
         persist_pool_meta_identity_for_startup(store.pools.clone(), &mut write_state, true)
             .await
             .expect("initialized pool metadata identity should persist");
-        *store.pool_meta_save_gate.lock().await = write_state;
+        *store.pool_meta_save_gate.write().await = write_state;
         // Simulate physical loss; raw deletion of authoritative pool.bin is
         // deliberately rejected by the native metadata mutation guard.
         for pool in &store.pools {

crates/ecstore/src/services/rebalance/mod.rs

@@ -100,7 +100,7 @@ pub async fn test_store_with_persisted_rebalance_meta(
         rebalance_meta: tokio::sync::RwLock::new(Some(meta)),
         decommission_cancelers: tokio::sync::RwLock::new(vec![None]),
         start_gate: tokio::sync::Mutex::new(()),
-        pool_meta_save_gate: tokio::sync::Mutex::default(),
+        pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
         ctx,
         bucket_fence_registry: std::sync::Arc::default(),
     });
@@ -335,7 +335,7 @@ async fn test_pool_stores_with_contexts(
             rebalance_meta: tokio::sync::RwLock::new(rebalance_meta.clone()),
             decommission_cancelers: tokio::sync::RwLock::new(vec![None; pool_count]),
             start_gate: tokio::sync::Mutex::new(()),
-            pool_meta_save_gate: tokio::sync::Mutex::new(pool_meta_write_state.independent_clone_for_test()),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(pool_meta_write_state.independent_clone_for_test()),
             ctx: store_ctx,
             bucket_fence_registry: std::sync::Arc::default(),
         })

crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs

@@ -3412,7 +3412,7 @@ fn test_store_with_rebalance_meta(meta: RebalanceMeta) -> Arc<crate::store::ECSt
         rebalance_meta: tokio::sync::RwLock::new(Some(meta)),
         decommission_cancelers: tokio::sync::RwLock::new(Vec::new()),
         start_gate: tokio::sync::Mutex::new(()),
-        pool_meta_save_gate: tokio::sync::Mutex::default(),
+        pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
         ctx: crate::runtime::instance::bootstrap_ctx(),
         bucket_fence_registry: std::sync::Arc::default(),
     })

crates/ecstore/src/store/heal.rs

@@ -527,7 +527,7 @@ impl ECStore {
     async fn acquire_heal_format_fence(
         &self,
     ) -> Result<(
-        tokio::sync::MutexGuard<'_, PoolMetaWriteState>,
+        crate::core::pools::PoolMetaSaveGateWriteGuard<'_>,
         NamespaceLockGuard,
         NamespaceLockGuard,
         PoolMeta,
@@ -538,7 +538,7 @@ impl ECStore {
             .first()
             .cloned()
             .ok_or_else(|| Error::other("heal format requires at least one storage pool"))?;
-        let mut write_state = self.pool_meta_save_gate.lock().await;
+        let mut write_state = self.pool_meta_save_gate.write().await;
         write_state.ensure_write_safe("heal format fence failed")?;
 
         // Metadata fence order is part of the decommission/rebalance protocol:
@@ -1079,7 +1079,7 @@ impl ECStore {
     pub(super) async fn handle_heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
         let movement_gate = self.ctx.data_movement_operation_gate();
         let _movement_guard = movement_gate.read().await;
-        let save_guard = self.pool_meta_save_gate.lock().await;
+        let save_guard = self.pool_meta_save_gate.write().await;
         save_guard.ensure_write_safe("bucket heal cannot run while pool metadata requires recovery")?;
         let mut fenced_pools = BTreeSet::new();
         {
@@ -2527,7 +2527,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         }
@@ -3550,10 +3550,14 @@ mod tests {
     async fn full_format_heal_preblocked_pool_metadata_never_writes_format() {
         let (temp_dir, store, shutdown) = multi_pool_heal_store().await;
         remove_heal_test_format(&temp_dir, &store, 0, 3).await;
-        store.pool_meta_save_gate.lock().await.observe_replicas(PoolMetaReplicaState {
-            needs_repair: true,
-            repair_write_safe: false,
-        });
+        store
+            .pool_meta_save_gate
+            .write()
+            .await
+            .observe_replicas(PoolMetaReplicaState {
+                needs_repair: true,
+                repair_write_safe: false,
+            });
 
         let err = store
             .handle_heal_format(false)
@@ -3688,7 +3692,7 @@ mod tests {
         let target = remove_heal_test_format(&temp_dir, &store, 0, 3).await;
         store.pools[0].disk_set[0].disks.write().await[3] = None;
         let barrier = HealFormatAfterSaveBarrier::install(&store.pools[0], 3);
-        let recovery_latch = store.pool_meta_save_gate.lock().await.aborted_transaction_latch_for_test();
+        let recovery_latch = store.pool_meta_save_gate.write().await.aborted_transaction_latch_for_test();
         let mut heal = tokio::spawn({
             let store = Arc::clone(&store);
             async move { store.handle_heal_format(false).await }
@@ -3848,7 +3852,7 @@ mod tests {
             }
         }
         let endpoints: EndpointServerPools = pools.iter().map(|pool| pool.endpoints.clone()).collect::<Vec<_>>().into();
-        let state = source.pool_meta_save_gate.lock().await.independent_clone_for_test();
+        let state = source.pool_meta_save_gate.write().await.independent_clone_for_test();
         Arc::new(ECStore {
             id: Uuid::new_v4(),
             disk_map: Default::default(),
@@ -3858,7 +3862,7 @@ mod tests {
             rebalance_meta: tokio::sync::RwLock::new(None),
             decommission_cancelers: tokio::sync::RwLock::new(vec![None; 2]),
             start_gate: tokio::sync::Mutex::new(()),
-            pool_meta_save_gate: tokio::sync::Mutex::new(state),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(state),
             ctx: Arc::clone(&source.ctx),
             bucket_fence_registry: Arc::default(),
         })
@@ -4040,7 +4044,7 @@ mod tests {
         let missing = remove_pool_meta_shard(&store, 0).await;
         let (guard, admissions, mut state) = store.acquire_pool_meta_object_heal_fence(&[0]).await.expect("heal authority");
         assert!(admissions.into_iter().all(|admission| admission.is_ok()));
-        assert!(store.pool_meta_save_gate.try_lock().is_err(), "heal must retain the earlier save gate");
+        assert!(store.pool_meta_save_gate.try_write().is_err(), "heal must retain the earlier save gate");
         let writer_store = Arc::clone(&store);
         let writer = tokio::spawn(async move { writer_store.save_current_pool_meta_for_test(&[0]).await });
         tokio::task::yield_now().await;
@@ -4687,10 +4691,14 @@ mod tests {
     #[tokio::test]
     async fn bucket_heal_blocks_before_dispatch_after_unreadable_pool_meta_replica() {
         let store = minimal_heal_store().await;
-        store.pool_meta_save_gate.lock().await.observe_replicas(PoolMetaReplicaState {
-            needs_repair: true,
-            repair_write_safe: false,
-        });
+        store
+            .pool_meta_save_gate
+            .write()
+            .await
+            .observe_replicas(PoolMetaReplicaState {
+                needs_repair: true,
+                repair_write_safe: false,
+            });
 
         let err = store
             .handle_heal_bucket("bucket", &HealOpts::default())
@@ -4724,7 +4732,7 @@ mod tests {
             .await
             .expect("peer dispatch should start");
         assert!(
-            store.pool_meta_save_gate.try_lock().is_ok(),
+            store.pool_meta_save_gate.try_write().is_ok(),
             "coordinator must release its local save gate before waiting for peers"
         );
         assert!(
@@ -5116,7 +5124,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         };
@@ -5136,7 +5144,7 @@ mod tests {
             .save_for_startup(store.pools.clone())
             .await
             .expect("pool metadata should be persisted before format heal");
-        *store.pool_meta_save_gate.lock().await = PoolMetaWriteState::default();
+        *store.pool_meta_save_gate.write().await = crate::core::pools::PoolMetaWriteState::default();
 
         let replacement_endpoint = store.pools[1].endpoints.endpoints.as_ref()[2].clone();
         tokio::fs::remove_file(

crates/ecstore/src/store/init.rs

@@ -676,7 +676,7 @@ impl ECStore {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers,
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::new(pool_meta_write_state),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::new(pool_meta_write_state),
             // Adopt the caller's context (the process bootstrap one on the
             // legacy path) so startup writes (erasure type recorded before
             // this point) and later reads share one cell.
@@ -725,7 +725,7 @@ impl ECStore {
         // Lock order is save gate -> distributed metadata writer, matching
         // runtime mutations. RPC locks are available before ECStore publication
         // and do not depend on the later fleet proof.
-        let mut startup_write_state = self.pool_meta_save_gate.lock().await;
+        let mut startup_write_state = self.pool_meta_save_gate.write().await;
         let startup_pool_meta_guard = if should_persist_pool_meta {
             let pool = self
                 .pools

crates/ecstore/src/store/mod.rs

@@ -34,7 +34,7 @@ use crate::bucket::utils::check_put_object_part_args;
 use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
 use crate::cluster::rpc::{RemoteClient, S3PeerSys};
 use crate::config::storageclass;
-use crate::core::pools::{DecommissionCanceler, PoolMeta, PoolMetaWriteState};
+use crate::core::pools::{DecommissionCanceler, PoolMeta, PoolMetaSaveGate};
 use crate::disk::endpoint::{Endpoint, EndpointType};
 use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
 use crate::error::{Error, Result};
@@ -480,9 +480,13 @@ pub struct ECStore {
     /// Serializes full-document pool metadata saves and retains a fail-closed
     /// write block after startup observes an unreadable replica.
     ///
+    /// Immutable admission proofs share authority; publication, recovery and
+    /// authority changes retain exclusive access through their existing I/O.
+    /// Each writer publishes its final state on release, and health
+    /// inspection reads that committed state without acquiring the gate.
     /// Lock order: acquire `pool_meta_save_gate`, then the distributed
     /// `pool.bin` fence, then clone `pool_meta` under a short read lock.
-    pub(crate) pool_meta_save_gate: Mutex<PoolMetaWriteState>,
+    pub(crate) pool_meta_save_gate: PoolMetaSaveGate,
     /// Per-instance runtime state (Phase 5, backlog#939).
     ///
     /// Carries this instance's identity/runtime out of the process globals so
@@ -555,7 +559,6 @@ impl Default for ScannerDataMovementPauseStatus {
 #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
 pub struct PoolMetaWriteGateStatus {
     pub writes_ready: bool,
-    pub check_timed_out: bool,
     pub write_blocked: bool,
     pub transaction_aborted: bool,
     pub pool_meta_absent: bool,
@@ -571,7 +574,6 @@ impl Default for PoolMetaWriteGateStatus {
     fn default() -> Self {
         Self {
             writes_ready: true,
-            check_timed_out: false,
             write_blocked: false,
             transaction_aborted: false,
             pool_meta_absent: false,
@@ -1936,7 +1938,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
         };
@@ -2011,7 +2013,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx,
             bucket_fence_registry: Arc::default(),
         })

crates/ecstore/src/store/multipart.rs

@@ -2133,7 +2133,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         }

crates/ecstore/src/store/object.rs

@@ -10003,7 +10003,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx: crate::runtime::instance::bootstrap_ctx(),
             bucket_fence_registry: std::sync::Arc::default(),
         }
@@ -10095,7 +10095,7 @@ mod tests {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers: RwLock::new(Vec::new()),
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::default(),
+            pool_meta_save_gate: crate::core::pools::PoolMetaSaveGate::default(),
             ctx,
             bucket_fence_registry: std::sync::Arc::default(),
         }

docs/architecture/readiness-matrix.md

@@ -45,18 +45,9 @@ In a distributed deployment, startup publishes `FullReady` only after that local
 - Liveness reports process availability and must not depend on storage, IAM,
   lock quorum, or peer health.
 - Node readiness (`/health/ready`, `/minio/health/ready`) reports whether this process can serve reads: storage read quorum, IAM, and shared-lock quorum. It is the Kubernetes Service membership signal. Losing write quorum, the pool-metadata writer, or exclusive-lock quorum does not remove a node that can still read. The public health handler still returns 503 until startup publishes `FullReady`, so a process that has never opened S3 admission does not join the Service. Runtime storage diagnostics do not change that startup gate.
-- A blocked pool metadata writer degrades cluster-write readiness with
-  `pool_meta_write_blocked`. It does not withdraw `/health/ready`, and the node
-  probe omits that reason from `degradedReasons`. Metadata save-gate inspection
-  is bounded to 100 ms; contention reports `pool_metadata_check_timeout` on the
-  cluster-write probe without installing a block. The node probe keeps the last
-  confirmed writable observation for `details.poolMetadata` during a transient
-  inspection timeout, while preserving a confirmed block and failing that
-  detail closed when no fresh observation exists.
-- The authenticated cluster snapshot extends its existing node-local metadata
-  gate inspection with safe reason, failure phase, and original block time. It
-  distinguishes timeout from a block and changes no admission or recovery
-  decision. Runtime readiness and gate status are separate bounded observations.
+- Immutable pool metadata admission proofs share the local authority gate as readers. Publication, recovery, identity changes and operations requiring serialization retain exclusive ownership and their existing distributed namespace fences. A multi-phase operation releases its reader before acquiring the writer and revalidates under exclusive ownership; it never upgrades a held reader. The gate is fair, so a writer queued behind a reader that is waiting on namespace I/O also delays later admissions until that reader finishes. Every writer publishes the authority it leaves before releasing exclusive ownership. Health inspection reads that published authority without acquiring the gate, so neither a held nor a queued writer delays it. A writer's changes become visible when it releases the gate; a latch recorded by a transaction outcome after its writer released is shared with the published authority and visible immediately. Mutating paths still check authority under the gate.
+- A blocked pool metadata writer degrades cluster-write readiness with `pool_meta_write_blocked`. It does not withdraw `/health/ready`, and the node probe omits that reason from `degradedReasons`. Metadata authority inspection reads the state published by the last local writer and does not wait for the save gate, so a held or queued metadata writer does not degrade either probe; a block that writer installs appears when it releases the gate.
+- The authenticated cluster snapshot extends its existing node-local metadata gate inspection with safe reason, failure phase, and original block time. It changes no admission or recovery decision. Runtime readiness and gate status are separate observations.
 - Cluster write readiness (`/minio/health/cluster`) requires storage write quorum, the pool-metadata write gate, and exclusive-lock quorum. Use it when a caller must know that new writes can commit. It is not the Service membership probe.
 - 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 `n - n/2`, so the cluster read probe is conservative: a four-client set can still admit reads with two clients while that probe returns 503. Node `/health/ready` uses the shared-lock threshold instead, because that is what object reads acquire.
 - `HEAD` health probes keep header/status semantics and do not require response
@@ -91,7 +82,7 @@ For a healthy IAM and metadata writer in a four-node, one-drive-per-node EC 2+2
 
 With two of four nodes up, GET, HEAD, and List of a bucket that this process has not yet cached still succeed. PUT and other mutations fail because write quorum and exclusive locks are gone.
 
-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; a timeout is counted by `rustfs_pool_metadata_check_timeouts_total` and uses the last confirmed node-local gate state only within the same cache TTL. These observations are not an atomic cluster snapshot and do not bypass the existing lock-probe timing or cache policy.
+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 reads the authority published by the last local writer and does not wait for the save gate. 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.
 

docs/operations/pool-metadata-recovery.md

@@ -77,7 +77,7 @@ An unreadable replica, lost fence, or conditional-write conflict leaves the orig
 
 - The first block emits `decommission_state` with `state=pool_metadata_blocked`, `reason`, `phase`, and `blocked_since`. A change in recovery failure classification emits `state=pool_metadata_recovery_pending`; successful recovery emits `state=pool_metadata_recovered` with the original timestamp.
 - `rustfs_pool_metadata_blocks_total{reason}` and `rustfs_pool_metadata_recoveries_total` count block and recovery transitions. The original cause and phase remain attached to local typed errors; storage/RPC error numbers and on-disk formats are unchanged.
-- Node and cluster-write readiness include `pool_meta_write_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md).
+- Cluster-write readiness reports `pool_meta_write_blocked`; node readiness reports the writer only through `details.poolMetadata`. Readiness reads the authority published by the last local metadata writer instead of waiting for the save gate, so an active or queued metadata transaction does not degrade readiness by itself. A block installed by a writer appears when that writer releases the gate. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md).
 
 Pool metadata CAS explicitly selects `TailDrainedWithOwnerRepair`: it waits for the complete rename result and returns repair responsibility to its existing deployment and transaction owner when write quorum commits but a replica fails or signatures diverge. Cache invalidation and the committed cleanup handoff precede that error. Ordinary system metadata retains quorum-success semantics under `Quorum` and `TailDrained`; draining a tail does not require every disk to succeed. A repair-required completion does not enter rollback cleanup or wait for another temporary-data deletion. The committed bytes are not rolled back. Generic MRF and heal-channel requests cannot supply the deployment and transaction authority needed to repair system metadata; pool CAS retains the uncertain outcome and its existing recovery fence. Ordinary object repair keeps its bucket-incarnation fencing.
 

docs/operations/rolling-restart.md

@@ -97,7 +97,7 @@ When the whole cluster (or several nodes) went down and nodes are brought back o
 
 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. Once storage read quorum is available, pending nodes can finish IAM bootstrap on the next retry. `/health/ready` returns `200` when startup has published `FullReady` and the node can still serve reads (storage read quorum, IAM, and shared-lock quorum). Write quorum is reported separately and is required by `/minio/health/cluster`, not by Service membership. 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` follows read quorum (`readinessScope` is `read_quorum`); `storage.readQuorum` and `storage.writeQuorum` show the two observations. `poolMetadata.ready` is the metadata writer and does not by itself withdraw `/health/ready`. `degradedReasons` on this probe names read-path causes such as `storage_quorum_unavailable` and `storage_and_lock_unavailable`. `pool_meta_write_blocked` and `pool_metadata_check_timeout` stay on `/minio/health/cluster`. See the [storage detail contract](../architecture/readiness-matrix.md#storage-detail-contract) for probe scopes and sampling limits:
+4. Check readiness detail while waiting. `/health/ready` (and `/minio/health/ready`) separate `storage` / `poolMetadata` / `iam` / `lock` readiness. `storage.ready` follows read quorum (`readinessScope` is `read_quorum`); `storage.readQuorum` and `storage.writeQuorum` show the two observations. `poolMetadata.ready` is the metadata writer and does not by itself withdraw `/health/ready`. `degradedReasons` on this probe names read-path causes such as `storage_quorum_unavailable` and `storage_and_lock_unavailable`. `pool_meta_write_blocked` stays on `/minio/health/cluster`. 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

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

@@ -45,15 +45,14 @@ The probe round timeout is configured independently; see [Admin peer probe timeo
 
 ## Correlate bounded diagnostics
 
-Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while runtime readiness diagnostics distinguish `pool_meta_write_blocked`, `pool_metadata_check_timeout`, and insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green.
+Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while runtime readiness diagnostics distinguish `pool_meta_write_blocked` from insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green.
 
-Query the authenticated `/rustfs/admin/v4/cluster/snapshot` endpoint on the affected node. `snapshot.pool_meta_write_gate` describes that node's metadata writer, not a fleet-wide aggregate. Its existing booleans are preserved; `state` and optional block details extend the same bounded, read-only gate inspection. This starts no recovery, disk reads, or additional RPCs. Runtime readiness and the gate are inspected separately, so a whole cluster snapshot is not atomic across sections.
+Query the authenticated `/rustfs/admin/v4/cluster/snapshot` endpoint on the affected node. `snapshot.pool_meta_write_gate` describes that node's metadata writer, not a fleet-wide aggregate. Its existing booleans are preserved; `state` and optional block details extend the same read-only inspection, which reads the authority published by the node's last metadata writer without waiting for the save gate. A block that an active writer installs appears when that writer releases the gate. Nodes running earlier releases can also report `check_timeout`, the `pool_metadata_check_timeout` readiness reason, and `rustfs_pool_metadata_check_timeouts_total` when their 100 ms gate inspection expired; current nodes emit none of them. This starts no recovery, disk reads, or additional RPCs. Runtime readiness and the gate are inspected separately, so a whole cluster snapshot is not atomic across sections.
 
 | `state` | Meaning |
 | --- | --- |
 | `writable` | No metadata write block was observed. Other dependencies may still make the node unready. |
 | `blocked` | A metadata write block was observed; `reason`, `phase`, and `sinceUnixSecs` describe its typed failure context. |
-| `check_timeout` | The existing 100 ms inspection budget expired. Readiness remains false, but a write block is not asserted. |
 | `unavailable` | The object store or a recognized metadata observation was unavailable; no block details are invented. |
 
 For `blocked`, `reason` reuses the metadata failure classification, `phase` identifies the operation stage that caused the block (not live recovery-worker progress), and `sinceUnixSecs` is the original block time in Unix seconds. Polling does not reset that time. Recovery replaces the observation with `writable` and removes the previous block details. An unavailable store reports `writesReady=false` with `state="unavailable"`. Older responses may omit `state` and block details; missing fields alone do not prove a healthy writer. Operation text, raw replica errors, disk paths, and credentials are excluded. See [Pool metadata recovery](pool-metadata-recovery.md) for recovery and escalation boundaries.

rustfs/src/admin/handlers/cluster_snapshot.rs

@@ -211,9 +211,7 @@ pub(crate) struct ClusterPoolMetaWriteGateView {
 
 impl From<ClusterPoolMetaWriteGateSnapshot> for ClusterPoolMetaWriteGateView {
     fn from(snapshot: ClusterPoolMetaWriteGateSnapshot) -> Self {
-        let state = if snapshot.check_timed_out {
-            "check_timeout"
-        } else if snapshot.writes_ready {
+        let state = if snapshot.writes_ready {
             "writable"
         } else if snapshot.write_blocked || snapshot.transaction_aborted {
             "blocked"
@@ -706,7 +704,6 @@ fn summarize_storage_readiness(snapshot: &ClusterReadOnlySnapshot) -> Capability
         .filter_map(|reason| match reason {
             ReadinessDegradedReason::StorageQuorumUnavailable
             | ReadinessDegradedReason::PoolMetaWriteBlocked
-            | ReadinessDegradedReason::PoolMetadataCheckTimeout
             | ReadinessDegradedReason::StorageAndIamUnavailable
             | ReadinessDegradedReason::StorageAndLockUnavailable
             | ReadinessDegradedReason::StorageIamAndLockUnavailable => Some(reason.as_str()),
@@ -1120,14 +1117,6 @@ mod tests {
 
         for (snapshot, state) in [
             (ClusterPoolMetaWriteGateSnapshot::default(), "writable"),
-            (
-                ClusterPoolMetaWriteGateSnapshot {
-                    writes_ready: false,
-                    check_timed_out: true,
-                    ..Default::default()
-                },
-                "check_timeout",
-            ),
             (
                 ClusterPoolMetaWriteGateSnapshot {
                     writes_ready: false,

rustfs/src/cluster_snapshot.rs

@@ -108,7 +108,6 @@ pub struct ClusterListingDiagnosticsSnapshot {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ClusterPoolMetaWriteGateSnapshot {
     pub writes_ready: bool,
-    pub check_timed_out: bool,
     pub write_blocked: bool,
     pub transaction_aborted: bool,
     pub pool_meta_absent: bool,
@@ -124,7 +123,6 @@ impl Default for ClusterPoolMetaWriteGateSnapshot {
     fn default() -> Self {
         Self {
             writes_ready: true,
-            check_timed_out: false,
             write_blocked: false,
             transaction_aborted: false,
             pool_meta_absent: false,
@@ -230,7 +228,6 @@ async fn current_pool_meta_write_gate_snapshot() -> ClusterPoolMetaWriteGateSnap
             let status = store.pool_meta_write_gate_status().await;
             ClusterPoolMetaWriteGateSnapshot {
                 writes_ready: status.writes_ready,
-                check_timed_out: status.check_timed_out,
                 write_blocked: status.write_blocked,
                 transaction_aborted: status.transaction_aborted,
                 pool_meta_absent: status.pool_meta_absent,

rustfs/src/server/health.rs

@@ -614,22 +614,6 @@ mod tests {
         });
     }
 
-    #[test]
-    #[serial]
-    fn writable_node_readiness_keeps_200_for_transient_pool_metadata_timeout() {
-        let mut report = ready_report();
-        report.degraded_reasons = vec![ReadinessDegradedReason::PoolMetadataCheckTimeout];
-
-        with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false"), || {
-            let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&report), "rustfs", None, None);
-
-            assert_eq!(parts.status_code, StatusCode::OK);
-            let payload = parts.payload.expect("GET readiness body");
-            assert_eq!(payload["ready"], true);
-            assert_eq!(payload["degradedReasons"], json!(["pool_metadata_check_timeout"]));
-        });
-    }
-
     #[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

@@ -37,7 +37,7 @@ use std::task::{Context, Poll};
 use std::time::Duration;
 use std::{
     collections::{HashMap, HashSet},
-    sync::{Mutex as StdMutex, OnceLock},
+    sync::OnceLock,
     time::Instant,
 };
 use tokio::sync::Mutex;
@@ -80,7 +80,6 @@ const LOG_COMPONENT_SERVER: &str = "server";
 const LOG_SUBSYSTEM_READINESS: &str = "readiness";
 const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready";
 const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total";
-const METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL: &str = "rustfs_pool_metadata_check_timeouts_total";
 
 pub use crate::shared_types::{
     DependencyReadiness, DependencyReadinessReport, MissingReadinessDrive, ReadinessDegradedReason, StorageReadinessDetails,
@@ -293,66 +292,6 @@ struct StorageWriteReadinessStatus {
     pool_metadata_reason: Option<ReadinessDegradedReason>,
 }
 
-#[derive(Debug, Clone, Copy, Default)]
-struct NodePoolMetadataReadinessCache {
-    entry: Option<NodePoolMetadataReadinessCacheEntry>,
-}
-
-#[derive(Debug, Clone, Copy)]
-struct NodePoolMetadataReadinessCacheEntry {
-    cached_at: Instant,
-    observed_at: Instant,
-    status: StorageWriteReadinessStatus,
-}
-
-impl NodePoolMetadataReadinessCache {
-    fn observe(
-        &mut self,
-        observed: StorageWriteReadinessStatus,
-        observed_at: Instant,
-        now: Instant,
-        ttl: Duration,
-    ) -> StorageWriteReadinessStatus {
-        if ttl.is_zero() {
-            self.entry = None;
-            return observed;
-        }
-
-        if observed.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) {
-            let Some(previous) = self.entry else {
-                return observed;
-            };
-            if now.saturating_duration_since(previous.cached_at) > ttl {
-                return observed;
-            }
-
-            // A confirmed write block must never be hidden by a later timeout.
-            if previous.status.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetaWriteBlocked) {
-                return previous.status;
-            }
-
-            // A transient inspection outage is diagnostic, not proof that the
-            // writer became unsafe. Keep the last confirmed writable state.
-            if previous.status.ready {
-                return StorageWriteReadinessStatus {
-                    ready: true,
-                    pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-                };
-            }
-            return observed;
-        }
-
-        if self.entry.is_none_or(|entry| entry.observed_at <= observed_at) {
-            self.entry = Some(NodePoolMetadataReadinessCacheEntry {
-                cached_at: now,
-                observed_at,
-                status: observed,
-            });
-        }
-        observed
-    }
-}
-
 fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWriteReadinessStatus {
     match result {
         Ok(()) => StorageWriteReadinessStatus {
@@ -360,13 +299,9 @@ fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWri
             pool_metadata_reason: None,
         },
         Err(error) => {
-            let pool_metadata_reason = if error.pool_metadata_failure().is_some() {
-                Some(ReadinessDegradedReason::PoolMetaWriteBlocked)
-            } else if matches!(error, StorageError::Timeout) {
-                Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)
-            } else {
-                None
-            };
+            let pool_metadata_reason = error
+                .pool_metadata_failure()
+                .map(|_| ReadinessDegradedReason::PoolMetaWriteBlocked);
             StorageWriteReadinessStatus {
                 ready: false,
                 pool_metadata_reason,
@@ -434,11 +369,6 @@ fn health_cluster_timeout() -> Duration {
     )
 }
 
-fn node_pool_metadata_readiness_cache() -> &'static StdMutex<NodePoolMetadataReadinessCache> {
-    static CACHE: OnceLock<StdMutex<NodePoolMetadataReadinessCache>> = OnceLock::new();
-    CACHE.get_or_init(|| StdMutex::new(NodePoolMetadataReadinessCache::default()))
-}
-
 fn storage_readiness_cache() -> &'static Mutex<Option<StorageReadinessCacheEntry>> {
     static CACHE: OnceLock<Mutex<Option<StorageReadinessCacheEntry>>> = OnceLock::new();
     CACHE.get_or_init(|| Mutex::new(None))
@@ -543,26 +473,6 @@ async fn update_storage_readiness_cache(status: StorageWriteReadinessStatus) {
     });
 }
 
-fn apply_node_pool_metadata_timeout_policy(
-    observed: StorageWriteReadinessStatus,
-    observed_at: Instant,
-) -> StorageWriteReadinessStatus {
-    if observed.pool_metadata_reason == Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) {
-        counter!(METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL).increment(1);
-    }
-    let mut cache = node_pool_metadata_readiness_cache()
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner);
-    cache.observe(observed, observed_at, Instant::now(), health_readiness_cache_ttl())
-}
-
-#[cfg(test)]
-fn reset_node_pool_metadata_readiness_cache() {
-    *node_pool_metadata_readiness_cache()
-        .lock()
-        .unwrap_or_else(std::sync::PoisonError::into_inner) = NodePoolMetadataReadinessCache::default();
-}
-
 async fn load_cached_lock_quorum_status() -> Option<LockQuorumObservation> {
     let ttl = health_readiness_cache_ttl();
     if ttl.is_zero() {
@@ -855,9 +765,6 @@ fn record_readiness_report(report: &DependencyReadinessReport) {
         && report.readiness.peer_health_ready;
     gauge!(METRIC_RUNTIME_READINESS_READY).set(if ready { 1.0 } else { 0.0 });
     for reason in &report.degraded_reasons {
-        if ready && *reason == ReadinessDegradedReason::PoolMetadataCheckTimeout {
-            continue;
-        }
         counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1);
     }
 }
@@ -944,7 +851,7 @@ pub async fn collect_node_readiness_report() -> DependencyReadinessReport {
     let mut storage_check_timed_out = false;
     if let Some(store) = runtime_sources::current_object_store_handle() {
         let pool_metadata_status = store.pool_meta_write_status().await;
-        storage = apply_node_pool_metadata_timeout_policy(pool_metadata_write_readiness(pool_metadata_status), Instant::now());
+        storage = pool_metadata_write_readiness(pool_metadata_status);
         match node_storage_snapshot(store.as_ref(), &lock_observation.online_hosts).await {
             Ok((info, unavailable_drives, missing_drives)) => {
                 details.unavailable_drives = unavailable_drives;
@@ -2927,7 +2834,7 @@ mod tests {
     }
 
     #[test]
-    fn pool_metadata_observation_distinguishes_block_timeout_and_unavailable() {
+    fn pool_metadata_observation_distinguishes_block_and_unavailable() {
         let blocked = blocked_pool_metadata_status();
         assert!(!blocked.ready);
         assert_eq!(blocked.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetaWriteBlocked));
@@ -2935,227 +2842,76 @@ mod tests {
         let writable = pool_metadata_write_readiness(Ok(()));
         assert!(writable.ready);
         assert_eq!(writable.pool_metadata_reason, None);
-        let timed_out = pool_metadata_write_readiness(Err(StorageError::Timeout));
-        assert!(!timed_out.ready);
-        assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
         let unavailable = pool_metadata_write_readiness(Err(StorageError::other("pool metadata writes remain blocked")));
         assert!(!unavailable.ready);
         assert_eq!(
             unavailable.pool_metadata_reason, None,
             "error text alone must not be interpreted as a typed write block"
         );
+    }
 
-        let readiness = DependencyReadiness {
-            storage_ready: false,
+    #[test]
+    fn readiness_report_records_ready_gauge_and_degraded_reasons() {
+        let record = |report: DependencyReadinessReport| {
+            let recorder = DebuggingRecorder::new();
+            let snapshotter = recorder.snapshotter();
+            metrics::with_local_recorder(&recorder, || record_readiness_report(&report));
+            snapshotter.snapshot().into_vec()
+        };
+        let ready = DependencyReadiness {
+            storage_ready: true,
             iam_ready: true,
             lock_quorum_ready: true,
             peer_health_ready: true,
         };
-        let report = dependency_readiness_report_from_write_status(readiness, timed_out);
-        assert!(!report.readiness.storage_ready, "inspection timeout must remain fail-closed");
-        assert_eq!(report.degraded_reasons, vec![ReadinessDegradedReason::PoolMetadataCheckTimeout]);
-        assert_eq!(report.degraded_reasons[0].as_str(), "pool_metadata_check_timeout");
-    }
 
-    #[test]
-    #[serial]
-    fn sticky_pool_metadata_timeout_uses_dedicated_metric_without_degraded_count() {
-        reset_node_pool_metadata_readiness_cache();
-        let recorder = DebuggingRecorder::new();
-        let snapshotter = recorder.snapshotter();
-
-        metrics::with_local_recorder(&recorder, || {
-            with_var(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"), || {
-                let writable = StorageWriteReadinessStatus {
-                    ready: true,
-                    pool_metadata_reason: None,
-                };
-                assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable);
-                let sticky = apply_node_pool_metadata_timeout_policy(
-                    StorageWriteReadinessStatus {
-                        ready: false,
-                        pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-                    },
-                    Instant::now(),
-                );
-                assert!(sticky.ready);
-                record_readiness_report(&DependencyReadinessReport {
-                    readiness: DependencyReadiness {
-                        storage_ready: true,
-                        iam_ready: true,
-                        lock_quorum_ready: true,
-                        peer_health_ready: true,
-                    },
-                    degraded_reasons: vec![ReadinessDegradedReason::PoolMetadataCheckTimeout],
-                    storage_details: None,
-                });
-            });
+        let entries = record(DependencyReadinessReport {
+            readiness: ready,
+            degraded_reasons: Vec::new(),
+            storage_details: None,
         });
-        reset_node_pool_metadata_readiness_cache();
-
-        let entries = snapshotter.snapshot().into_vec();
-        let ready = entries.iter().find_map(|(composite, _, _, value)| {
+        let gauge = entries.iter().find_map(|(composite, _, _, value)| {
             (composite.kind() == MetricKind::Gauge && composite.key().name() == METRIC_RUNTIME_READINESS_READY).then_some(value)
         });
-        assert!(matches!(ready, Some(DebugValue::Gauge(value)) if value.into_inner() == 1.0));
+        assert!(
+            matches!(gauge, Some(DebugValue::Gauge(value)) if value.into_inner() == 1.0),
+            "a ready report must set the readiness gauge: {gauge:?}"
+        );
+        assert!(
+            entries
+                .iter()
+                .all(|(composite, _, _, _)| composite.key().name() != METRIC_RUNTIME_READINESS_DEGRADED_TOTAL),
+            "a ready report without reasons must not count degradation"
+        );
 
+        let entries = record(DependencyReadinessReport {
+            readiness: DependencyReadiness {
+                storage_ready: false,
+                ..ready
+            },
+            degraded_reasons: vec![ReadinessDegradedReason::PoolMetaWriteBlocked],
+            storage_details: None,
+        });
+        let gauge = entries.iter().find_map(|(composite, _, _, value)| {
+            (composite.kind() == MetricKind::Gauge && composite.key().name() == METRIC_RUNTIME_READINESS_READY).then_some(value)
+        });
+        assert!(
+            matches!(gauge, Some(DebugValue::Gauge(value)) if value.into_inner() == 0.0),
+            "a degraded report must clear the readiness gauge: {gauge:?}"
+        );
         let degraded = entries.iter().find_map(|(composite, _, _, value)| {
             (composite.kind() == MetricKind::Counter
                 && composite.key().name() == METRIC_RUNTIME_READINESS_DEGRADED_TOTAL
                 && composite
                     .key()
                     .labels()
-                    .any(|label| label.key() == "reason" && label.value() == "pool_metadata_check_timeout"))
+                    .any(|label| label.key() == "reason" && label.value() == "pool_meta_write_blocked"))
             .then_some(value)
         });
-        assert!(degraded.is_none());
-
-        let timeout = entries.iter().find_map(|(composite, _, _, value)| {
-            (composite.kind() == MetricKind::Counter && composite.key().name() == METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL)
-                .then_some(value)
-        });
-        assert!(matches!(timeout, Some(DebugValue::Counter(1))));
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn node_pool_metadata_timeout_preserves_recent_writable_observation() {
-        reset_node_pool_metadata_readiness_cache();
-        async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
-            let writable = StorageWriteReadinessStatus {
-                ready: true,
-                pool_metadata_reason: None,
-            };
-            assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable);
-
-            let timed_out = apply_node_pool_metadata_timeout_policy(
-                StorageWriteReadinessStatus {
-                    ready: false,
-                    pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-                },
-                Instant::now(),
-            );
-
-            assert!(timed_out.ready);
-            assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
-        })
-        .await;
-        reset_node_pool_metadata_readiness_cache();
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn node_pool_metadata_timeout_fails_closed_with_empty_cache() {
-        reset_node_pool_metadata_readiness_cache();
-        async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
-            let timed_out = apply_node_pool_metadata_timeout_policy(
-                StorageWriteReadinessStatus {
-                    ready: false,
-                    pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-                },
-                Instant::now(),
-            );
-
-            assert!(!timed_out.ready);
-            assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
-        })
-        .await;
-        reset_node_pool_metadata_readiness_cache();
-    }
-
-    #[test]
-    fn node_pool_metadata_timeout_fails_closed_after_cache_ttl() {
-        let mut cache = NodePoolMetadataReadinessCache::default();
-        let observed_at = Instant::now();
-        let ttl = Duration::from_secs(60);
-        let writable = StorageWriteReadinessStatus {
-            ready: true,
-            pool_metadata_reason: None,
-        };
-        assert_eq!(cache.observe(writable, observed_at, observed_at, ttl), writable);
-
-        let timed_out = cache.observe(
-            StorageWriteReadinessStatus {
-                ready: false,
-                pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-            },
-            observed_at + Duration::from_secs(61),
-            observed_at + Duration::from_secs(61),
-            ttl,
+        assert!(
+            matches!(degraded, Some(DebugValue::Counter(1))),
+            "each degraded reason must be counted once: {degraded:?}"
         );
-
-        assert!(!timed_out.ready);
-        assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn node_pool_metadata_timeout_never_hides_confirmed_block() {
-        reset_node_pool_metadata_readiness_cache();
-        async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
-            let blocked = StorageWriteReadinessStatus {
-                ready: false,
-                pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetaWriteBlocked),
-            };
-            let newer_observation = Instant::now();
-            assert_eq!(apply_node_pool_metadata_timeout_policy(blocked, newer_observation), blocked);
-            assert_eq!(
-                apply_node_pool_metadata_timeout_policy(
-                    StorageWriteReadinessStatus {
-                        ready: true,
-                        pool_metadata_reason: None,
-                    },
-                    newer_observation
-                        .checked_sub(Duration::from_secs(1))
-                        .expect("test instant has elapsed time"),
-                ),
-                StorageWriteReadinessStatus {
-                    ready: true,
-                    pool_metadata_reason: None,
-                }
-            );
-
-            let timed_out = apply_node_pool_metadata_timeout_policy(
-                StorageWriteReadinessStatus {
-                    ready: false,
-                    pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-                },
-                Instant::now(),
-            );
-
-            assert_eq!(timed_out, blocked);
-        })
-        .await;
-        reset_node_pool_metadata_readiness_cache();
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn node_pool_metadata_timeout_fails_closed_without_fresh_observation() {
-        reset_node_pool_metadata_readiness_cache();
-        async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
-            let writable = StorageWriteReadinessStatus {
-                ready: true,
-                pool_metadata_reason: None,
-            };
-            assert_eq!(apply_node_pool_metadata_timeout_policy(writable, Instant::now()), writable);
-
-            async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("0"))], async {
-                let timed_out = apply_node_pool_metadata_timeout_policy(
-                    StorageWriteReadinessStatus {
-                        ready: false,
-                        pool_metadata_reason: Some(ReadinessDegradedReason::PoolMetadataCheckTimeout),
-                    },
-                    Instant::now(),
-                );
-
-                assert!(!timed_out.ready);
-                assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout));
-            })
-            .await;
-        })
-        .await;
-        reset_node_pool_metadata_readiness_cache();
     }
 
     #[test]

rustfs/src/shared_types.rs

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