Merge branch 'main' into houseme/fix/pool-meta-write-gate-snapshot

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

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/core/pools.rs

@@ -65,6 +65,7 @@ use crate::storage_api_contracts::{
     namespace::NamespaceLocking as _,
     object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _},
 };
+use crate::store::PoolMetaWriteGateStatus;
 use crate::{core::sets::Sets, store::ECStore};
 use byteorder::{ByteOrder, LittleEndian, WriteBytesExt};
 use futures::{
@@ -10275,8 +10276,21 @@ impl ECStore {
     /// Read-only admission probes do not change this state; startup and real
     /// metadata transactions still latch it on unrecoverable conditions.
     pub async fn pool_meta_writes_ready(&self) -> bool {
+        self.pool_meta_write_gate_status().await.writes_ready
+    }
+
+    pub async fn pool_meta_write_gate_status(&self) -> PoolMetaWriteGateStatus {
         let write_state = self.pool_meta_save_gate.lock().await;
-        !write_state.write_blocked && !write_state.aborted_transaction.load(Ordering::SeqCst)
+        let transaction_aborted = write_state.aborted_transaction.load(Ordering::SeqCst);
+        PoolMetaWriteGateStatus {
+            writes_ready: !write_state.write_blocked && !transaction_aborted,
+            write_blocked: write_state.write_blocked,
+            transaction_aborted,
+            pool_meta_absent: write_state.pool_meta_absent,
+            identity_initialized: write_state.identity_initialized,
+            identity_needs_repair: write_state.identity_needs_repair,
+            cluster_epoch: write_state.cluster_epoch,
+        }
     }
 
     async fn load_runtime_pool_meta_observing(&self, write_state: &mut PoolMetaWriteState, operation: &str) -> Result<PoolMeta> {
@@ -16917,6 +16931,10 @@ mod tests {
             .expect("create single-pool bucket before blocking pool metadata writes");
         let incarnation = store.bucket_incarnation_id(&bucket).await.expect("load bucket incarnation");
         store.pool_meta_save_gate.lock().await.block_writes_after_fence_loss();
+        let write_gate = store.pool_meta_write_gate_status().await;
+        assert!(!write_gate.writes_ready);
+        assert!(write_gate.write_blocked);
+        assert!(!write_gate.transaction_aborted);
 
         let object = "ordinary-put.bin";
         let mut put_data = crate::object_api::PutObjReader::from_vec(b"ordinary single-pool body".to_vec());

crates/ecstore/src/lib.rs

@@ -55,6 +55,8 @@ mod set_disk;
 mod storage_api_contracts;
 mod store;
 
+pub use store::PoolMetaWriteGateStatus;
+
 // pub mod checksum;
 mod event;
 

crates/ecstore/src/store/mod.rs

@@ -527,6 +527,31 @@ impl Default for ScannerDataMovementPauseStatus {
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
+pub struct PoolMetaWriteGateStatus {
+    pub writes_ready: bool,
+    pub write_blocked: bool,
+    pub transaction_aborted: bool,
+    pub pool_meta_absent: bool,
+    pub identity_initialized: Option<bool>,
+    pub identity_needs_repair: bool,
+    pub cluster_epoch: Option<u64>,
+}
+
+impl Default for PoolMetaWriteGateStatus {
+    fn default() -> Self {
+        Self {
+            writes_ready: true,
+            write_blocked: false,
+            transaction_aborted: false,
+            pool_meta_absent: false,
+            identity_initialized: None,
+            identity_needs_repair: false,
+            cluster_epoch: None,
+        }
+    }
+}
+
 fn offset_unix_seconds(value: OffsetDateTime) -> u64 {
     u64::try_from(value.unix_timestamp()).unwrap_or(0)
 }

rustfs/src/admin/handlers/cluster_snapshot.rs

@@ -25,7 +25,8 @@ use crate::admin::{
     system,
 };
 use crate::cluster_snapshot::{
-    ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, cluster_has_actionable_pressure,
+    ClusterPoolMetaWriteGateSnapshot, ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot,
+    cluster_has_actionable_pressure,
 };
 use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason};
 use http::{HeaderMap, HeaderValue, StatusCode};
@@ -154,6 +155,7 @@ pub(crate) struct ClusterSnapshotView {
     pub observability: ObservabilitySnapshot,
     pub workload_admission: Vec<WorkloadAdmissionView>,
     pub runtime_status: ClusterRuntimeStatusView,
+    pub pool_meta_write_gate: ClusterPoolMetaWriteGateView,
     pub actionable_pressure: bool,
 }
 
@@ -182,11 +184,38 @@ impl ClusterSnapshotView {
             observability: snapshot.observability,
             workload_admission: workload_admission_views(snapshot.workload_admission),
             runtime_status: ClusterRuntimeStatusView::from(snapshot.runtime_status),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateView::from(snapshot.pool_meta_write_gate),
             actionable_pressure,
         }
     }
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct ClusterPoolMetaWriteGateView {
+    pub writes_ready: bool,
+    pub write_blocked: bool,
+    pub transaction_aborted: bool,
+    pub pool_meta_absent: bool,
+    pub identity_initialized: Option<bool>,
+    pub identity_needs_repair: bool,
+    pub cluster_epoch: Option<u64>,
+}
+
+impl From<ClusterPoolMetaWriteGateSnapshot> for ClusterPoolMetaWriteGateView {
+    fn from(snapshot: ClusterPoolMetaWriteGateSnapshot) -> Self {
+        Self {
+            writes_ready: snapshot.writes_ready,
+            write_blocked: snapshot.write_blocked,
+            transaction_aborted: snapshot.transaction_aborted,
+            pool_meta_absent: snapshot.pool_meta_absent,
+            identity_initialized: snapshot.identity_initialized,
+            identity_needs_repair: snapshot.identity_needs_repair,
+            cluster_epoch: snapshot.cluster_epoch,
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub(crate) struct ClusterComponentStatusView {
     pub storage: ClusterComponentStatus,
@@ -977,8 +1006,8 @@ mod tests {
         ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport,
     };
     use crate::cluster_snapshot::{
-        ClusterListingDiagnosticsSnapshot, ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot,
-        ClusterUsageFreshnessSnapshot,
+        ClusterListingDiagnosticsSnapshot, ClusterPoolMetaWriteGateSnapshot, ClusterReadOnlySnapshot,
+        ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, ClusterUsageFreshnessSnapshot,
     };
     use crate::shared_types::{DependencyReadiness, ReadinessDegradedReason};
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass};
@@ -1124,6 +1153,15 @@ mod tests {
             listing_diagnostics: ClusterListingDiagnosticsSnapshot {
                 internode_stall_timeouts_total: 2,
             },
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot {
+                writes_ready: false,
+                write_blocked: true,
+                transaction_aborted: true,
+                identity_initialized: Some(true),
+                identity_needs_repair: true,
+                cluster_epoch: Some(7),
+                ..Default::default()
+            },
         };
 
         let value = serde_json::to_value(ClusterSnapshotView::from_snapshot(snapshot, Some(":::9000".to_string())))
@@ -1136,6 +1174,13 @@ mod tests {
         assert_eq!(value["components"]["listing"]["source"], "workload_admission+internode_metrics");
         assert_eq!(value["components"]["listing"]["condition"], "unknown");
         assert_eq!(value["components"]["listing"]["internode_stall_timeouts_total"], 2);
+        assert_eq!(value["pool_meta_write_gate"]["writesReady"], false);
+        assert_eq!(value["pool_meta_write_gate"]["writeBlocked"], true);
+        assert_eq!(value["pool_meta_write_gate"]["transactionAborted"], true);
+        assert_eq!(value["pool_meta_write_gate"]["identityInitialized"], true);
+        assert_eq!(value["pool_meta_write_gate"]["identityNeedsRepair"], true);
+        assert_eq!(value["pool_meta_write_gate"]["clusterEpoch"], 7);
+        assert_eq!(value["actionable_pressure"], true);
         assert_eq!(value["components"]["usage"]["source"], "scanner_metrics");
         assert_eq!(value["components"]["usage"]["condition"], "stale");
         assert_eq!(value["membership"]["nodes"][0]["server_info_endpoint"], ":::9000");
@@ -1193,6 +1238,7 @@ mod tests {
             },
             usage_freshness: ClusterUsageFreshnessSnapshot::default(),
             listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
 
         let summary = ClusterSnapshotSummary::from(&snapshot);
@@ -1259,6 +1305,7 @@ mod tests {
             listing_diagnostics: ClusterListingDiagnosticsSnapshot {
                 internode_stall_timeouts_total: 0,
             },
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
 
         let view = ClusterSnapshotView::from(snapshot);
@@ -1306,6 +1353,11 @@ mod tests {
             },
             usage_freshness: ClusterUsageFreshnessSnapshot::default(),
             listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot {
+                writes_ready: false,
+                write_blocked: true,
+                ..Default::default()
+            },
         };
 
         let view = ClusterSnapshotView::from(snapshot);
@@ -1341,6 +1393,7 @@ mod tests {
             listing_diagnostics: ClusterListingDiagnosticsSnapshot {
                 internode_stall_timeouts_total: 2,
             },
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
 
         let component = super::summarize_listing_metacache(&snapshot);
@@ -1384,6 +1437,7 @@ mod tests {
                 ..Default::default()
             },
             listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
 
         let component = super::summarize_usage_freshness(&snapshot);
@@ -1421,6 +1475,7 @@ mod tests {
                 ..Default::default()
             },
             listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
 
         let component = super::summarize_usage_freshness(&snapshot);
@@ -1470,6 +1525,7 @@ mod tests {
             },
             usage_freshness: ClusterUsageFreshnessSnapshot::default(),
             listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
 
         let summary = ClusterSnapshotSummary::from(&snapshot);

rustfs/src/cluster_snapshot.rs

@@ -24,6 +24,7 @@ use crate::storage_api::cluster::control_plane::{
 };
 use crate::workload_admission::workload_admission_registry_snapshot;
 use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot};
+use rustfs_ecstore::PoolMetaWriteGateStatus;
 use rustfs_io_metrics::internode_metrics::{InternodeMetricsSnapshot, global_internode_metrics};
 use rustfs_scanner_metrics::metrics::{ScannerMetricsReport, global_metrics};
 
@@ -40,6 +41,7 @@ pub struct ClusterReadOnlySnapshot {
     pub runtime_status: ClusterRuntimeStatusSnapshot,
     pub usage_freshness: ClusterUsageFreshnessSnapshot,
     pub listing_diagnostics: ClusterListingDiagnosticsSnapshot,
+    pub pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -104,6 +106,45 @@ pub struct ClusterListingDiagnosticsSnapshot {
     pub internode_stall_timeouts_total: u64,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterPoolMetaWriteGateSnapshot {
+    pub writes_ready: bool,
+    pub write_blocked: bool,
+    pub transaction_aborted: bool,
+    pub pool_meta_absent: bool,
+    pub identity_initialized: Option<bool>,
+    pub identity_needs_repair: bool,
+    pub cluster_epoch: Option<u64>,
+}
+
+impl Default for ClusterPoolMetaWriteGateSnapshot {
+    fn default() -> Self {
+        Self {
+            writes_ready: true,
+            write_blocked: false,
+            transaction_aborted: false,
+            pool_meta_absent: false,
+            identity_initialized: None,
+            identity_needs_repair: false,
+            cluster_epoch: None,
+        }
+    }
+}
+
+impl From<PoolMetaWriteGateStatus> for ClusterPoolMetaWriteGateSnapshot {
+    fn from(status: PoolMetaWriteGateStatus) -> Self {
+        Self {
+            writes_ready: status.writes_ready,
+            write_blocked: status.write_blocked,
+            transaction_aborted: status.transaction_aborted,
+            pool_meta_absent: status.pool_meta_absent,
+            identity_initialized: status.identity_initialized,
+            identity_needs_repair: status.identity_needs_repair,
+            cluster_epoch: status.cluster_epoch,
+        }
+    }
+}
+
 impl From<InternodeMetricsSnapshot> for ClusterListingDiagnosticsSnapshot {
     fn from(snapshot: InternodeMetricsSnapshot) -> Self {
         Self {
@@ -164,6 +205,7 @@ pub fn cluster_read_only_snapshot_from_control_plane(
         runtime_status,
         usage_freshness: ClusterUsageFreshnessSnapshot::default(),
         listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+        pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
     }
 }
 
@@ -172,6 +214,7 @@ pub async fn collect_cluster_read_only_snapshot(endpoint_pools: &EndpointServerP
     let mut snapshot = cluster_read_only_snapshot_from_endpoint_pools(endpoint_pools, runtime_status);
     snapshot.usage_freshness = current_usage_freshness_snapshot().await;
     snapshot.listing_diagnostics = current_listing_diagnostics_snapshot();
+    snapshot.pool_meta_write_gate = current_pool_meta_write_gate_snapshot().await;
     Some(snapshot)
 }
 
@@ -188,8 +231,16 @@ fn current_listing_diagnostics_snapshot() -> ClusterListingDiagnosticsSnapshot {
     ClusterListingDiagnosticsSnapshot::from(metrics.snapshot())
 }
 
+async fn current_pool_meta_write_gate_snapshot() -> ClusterPoolMetaWriteGateSnapshot {
+    match crate::runtime_sources::current_object_store_handle() {
+        Some(store) => ClusterPoolMetaWriteGateSnapshot::from(store.pool_meta_write_gate_status().await),
+        None => ClusterPoolMetaWriteGateSnapshot::default(),
+    }
+}
+
 pub fn cluster_has_actionable_pressure(snapshot: &ClusterReadOnlySnapshot) -> bool {
     snapshot.runtime_status.state == ClusterRuntimeReadinessState::Degraded
+        || !snapshot.pool_meta_write_gate.writes_ready
         || snapshot
             .workload_admission
             .entries()
@@ -312,6 +363,7 @@ mod tests {
             },
             usage_freshness: ClusterUsageFreshnessSnapshot::default(),
             listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(),
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(),
         };
         assert!(!cluster_has_actionable_pressure(&no_pressure));
 
@@ -335,9 +387,19 @@ mod tests {
                 WorkloadClass::Repair,
                 AdmissionState::Unknown,
             )]),
-            ..no_pressure
+            ..no_pressure.clone()
         };
         assert!(cluster_has_actionable_pressure(&admission_pressure));
+
+        let pool_meta_pressure = ClusterReadOnlySnapshot {
+            pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot {
+                writes_ready: false,
+                write_blocked: true,
+                ..Default::default()
+            },
+            ..no_pressure
+        };
+        assert!(cluster_has_actionable_pressure(&pool_meta_pressure));
     }
 
     #[test]