fix(ecstore): bootstrap verified MinIO adoption metadata (#7020)

crates/ecstore/src/core/pools.rs

@@ -4075,23 +4075,54 @@ pub(crate) struct PoolMetaWriteState {
     expected_cluster_id: Option<uuid::Uuid>,
     cluster_epoch: Option<u64>,
     pool_meta_absent: bool,
-    fresh_bootstrap_proven: bool,
+    bootstrap_authority: PoolMetaBootstrapAuthority,
     identity_initialized: Option<bool>,
     identity_fresh_bootstrap_nonce: Option<uuid::Uuid>,
     identity_needs_repair: bool,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub(crate) enum PoolMetaBootstrapAuthority {
+    #[default]
+    None,
+    Fresh,
+    LegacyAdoption,
+}
+
+impl PoolMetaBootstrapAuthority {
+    pub(crate) fn combine_across_pools(self, other: Self) -> Self {
+        if self == other { self } else { Self::None }
+    }
+
+    fn is_proven(self) -> bool {
+        !matches!(self, Self::None)
+    }
+}
+
 impl PoolMetaWriteState {
+    #[cfg(any(test, feature = "test-util"))]
     pub(crate) fn for_startup(cluster_id: uuid::Uuid, fresh_bootstrap_proven: bool) -> Self {
+        let bootstrap_authority = if fresh_bootstrap_proven {
+            PoolMetaBootstrapAuthority::Fresh
+        } else {
+            PoolMetaBootstrapAuthority::None
+        };
+        Self::for_startup_with_bootstrap_authority(cluster_id, bootstrap_authority)
+    }
+
+    pub(crate) fn for_startup_with_bootstrap_authority(
+        cluster_id: uuid::Uuid,
+        bootstrap_authority: PoolMetaBootstrapAuthority,
+    ) -> Self {
         Self {
             expected_cluster_id: Some(cluster_id),
-            fresh_bootstrap_proven,
+            bootstrap_authority,
             ..Default::default()
         }
     }
 
-    pub(crate) fn fresh_bootstrap_proven(&self) -> bool {
-        self.fresh_bootstrap_proven
+    pub(crate) fn bootstrap_identity_proven(&self) -> bool {
+        self.bootstrap_authority.is_proven()
     }
 
     pub(crate) fn identity_is_pending(&self) -> bool {
@@ -4113,7 +4144,7 @@ impl PoolMetaWriteState {
     #[cfg(any(test, feature = "test-util"))]
     fn for_test_bootstrap() -> Self {
         Self {
-            fresh_bootstrap_proven: true,
+            bootstrap_authority: PoolMetaBootstrapAuthority::Fresh,
             identity_initialized: Some(false),
             identity_fresh_bootstrap_nonce: Some(uuid::Uuid::new_v4()),
             ..Default::default()
@@ -4174,7 +4205,7 @@ impl PoolMetaWriteState {
         self.identity_fresh_bootstrap_nonce = selection.identity.and_then(|identity| identity.fresh_bootstrap_nonce);
         if let Some(identity) = selection.identity {
             if identity.initialized {
-                self.fresh_bootstrap_proven = false;
+                self.bootstrap_authority = PoolMetaBootstrapAuthority::None;
             }
             if let Some(metadata_epoch) = self.cluster_epoch
                 && metadata_epoch != identity.epoch
@@ -4198,11 +4229,11 @@ impl PoolMetaWriteState {
             return Ok(());
         }
         match self.identity_initialized {
-            Some(false) if self.fresh_bootstrap_proven && self.identity_fresh_bootstrap_nonce.is_some() => Ok(()),
+            Some(false) if self.bootstrap_identity_proven() && self.identity_fresh_bootstrap_nonce.is_some() => Ok(()),
             Some(false) => {
                 self.block_writes();
                 Err(Error::other(
-                    "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof",
+                    "pool metadata recovery required: pending cluster identity exists but this startup has no verified fresh-bootstrap proof or legacy-adoption proof",
                 ))
             }
             Some(true) => {
@@ -5184,10 +5215,10 @@ where
                 ..identity
             },
             Some(identity) => identity,
-            None if !initialized && !write_state.fresh_bootstrap_proven() => {
+            None if !initialized && !write_state.bootstrap_identity_proven() => {
                 write_state.block_writes();
                 return Err(Error::other(
-                    "pool metadata recovery required: cannot create a pending cluster identity without verified fresh-bootstrap proof",
+                    "pool metadata recovery required: cannot create a pending cluster identity without verified fresh-bootstrap proof or legacy-adoption proof",
                 ));
             }
             None => PersistedPoolMetaIdentity {

crates/ecstore/src/store/init.rs

@@ -14,8 +14,8 @@
 
 use super::*;
 use crate::core::pools::{
-    PoolMetaReplicaState, PoolMetaWriteState, load_pool_meta_identity_observing, local_decommission_queue_prefix,
-    persist_pool_meta_identity_for_startup, pool_meta_has_active_decommission,
+    PoolMetaBootstrapAuthority, PoolMetaReplicaState, PoolMetaWriteState, load_pool_meta_identity_observing,
+    local_decommission_queue_prefix, persist_pool_meta_identity_for_startup, pool_meta_has_active_decommission,
 };
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
@@ -173,7 +173,7 @@ async fn establish_pool_meta_bootstrap_identity_if_proven<S>(
 where
     S: EcstoreObjectIO,
 {
-    if elected_writer && write_state.fresh_bootstrap_proven() {
+    if elected_writer && write_state.bootstrap_identity_proven() {
         persist_pool_meta_identity_for_startup(pools, write_state, false).await?;
     }
     Ok(())
@@ -215,7 +215,7 @@ where
     }
     let mut committed = meta.clone();
     if should_write {
-        if write_state.fresh_bootstrap_proven() || write_state.identity_is_pending() {
+        if write_state.bootstrap_identity_proven() || write_state.identity_is_pending() {
             persist_pool_meta_identity_for_startup(pools.clone(), write_state, false)
                 .await
                 .map_err(|err| Error::other(format!("store init failed during prepare_pool_meta_identity: {err}")))?;
@@ -402,7 +402,7 @@ impl ECStore {
         preflight_startup_rpc_secret(&endpoint_pools)?;
 
         let mut deployment_id = None;
-        let mut fresh_bootstrap_proven = true;
+        let mut pool_meta_bootstrap_authority = None;
 
         // let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?;
 
@@ -518,7 +518,12 @@ impl ECStore {
                     }
                 }
             }?;
-            fresh_bootstrap_proven &= loaded_format.fresh_bootstrap_proven;
+            pool_meta_bootstrap_authority = Some(pool_meta_bootstrap_authority.map_or(
+                loaded_format.pool_meta_bootstrap_authority,
+                |authority: PoolMetaBootstrapAuthority| {
+                    authority.combine_across_pools(loaded_format.pool_meta_bootstrap_authority)
+                },
+            ));
             let fm = loaded_format.format;
 
             // Format loading succeeded, enable health monitoring on all disks
@@ -559,6 +564,10 @@ impl ECStore {
         let peer_sys = S3PeerSys::new_with_instance_ctx(&endpoint_pools, instance_ctx.clone());
         let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default());
         pool_meta.dont_save = true;
+        let pool_meta_write_state = PoolMetaWriteState::for_startup_with_bootstrap_authority(
+            deployment_id,
+            pool_meta_bootstrap_authority.unwrap_or_default(),
+        );
 
         let decommission_cancelers = RwLock::new(vec![None; pools.len()]);
         let ec = Arc::new(ECStore {
@@ -570,7 +579,7 @@ impl ECStore {
             rebalance_meta: RwLock::new(None),
             decommission_cancelers,
             start_gate: Mutex::new(()),
-            pool_meta_save_gate: Mutex::new(PoolMetaWriteState::for_startup(deployment_id, fresh_bootstrap_proven)),
+            pool_meta_save_gate: Mutex::new(pool_meta_write_state),
             decommission_capacity_entry_gate: Mutex::default(),
             // Adopt the caller's context (the process bootstrap one on the
             // legacy path) so startup writes (erasure type recorded before
@@ -791,6 +800,7 @@ mod tests {
         run_local_decommission_watchdog, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
         should_defer_rebalance_auto_start, should_retry_format_load, wait_for_local_decommission_resume_delay,
     };
+    use crate::core::pools::PoolMetaBootstrapAuthority;
     #[cfg(feature = "test-util")]
     use crate::disk::DiskAPI;
     #[cfg(feature = "test-util")]
@@ -1151,6 +1161,65 @@ mod tests {
             .expect("successful startup publication must disarm the transaction guard");
     }
 
+    #[tokio::test]
+    async fn test_legacy_adoption_pool_meta_bootstrap_reaches_v3_cas() {
+        let deployment_id = Uuid::new_v4();
+        let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let mut write_state =
+            PoolMetaWriteState::for_startup_with_bootstrap_authority(deployment_id, PoolMetaBootstrapAuthority::LegacyAdoption);
+
+        establish_pool_meta_bootstrap_identity_if_proven(vec![storage.clone()], &mut write_state, true)
+            .await
+            .expect("verified legacy adoption should persist a nonce-bound identity before loading pool metadata");
+        let (loaded, replica_state) = load_pool_meta_for_startup(vec![storage.clone()], &mut write_state)
+            .await
+            .expect("verified legacy adoption should authorize initially missing pool metadata");
+        assert!(loaded.pools.is_empty());
+
+        let committed = persist_pool_meta_for_startup_if_safe(
+            &init_test_pool_meta(None),
+            vec![storage.clone()],
+            replica_state,
+            &mut write_state,
+            true,
+            true,
+        )
+        .await
+        .expect("verified legacy adoption should publish initial pool metadata");
+        assert_eq!(committed.pools[0].cmd_line, "pool-0");
+
+        let objects = storage.objects.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        assert!(objects.contains_key(POOL_META_NAME));
+        let identity = objects
+            .get(POOL_META_IDENTITY_NAME)
+            .map(|(payload, _)| payload.clone())
+            .expect("legacy adoption should commit the bootstrap identity");
+        assert!(pool_meta_identity_initialized_for_test(&identity).expect("decode committed identity"));
+    }
+
+    #[tokio::test]
+    async fn test_non_elected_legacy_adoption_cannot_initialize_pool_meta() {
+        let storage = Arc::new(StartupPoolMetaStorage::new(Vec::new()));
+        let mut write_state =
+            PoolMetaWriteState::for_startup_with_bootstrap_authority(Uuid::new_v4(), PoolMetaBootstrapAuthority::LegacyAdoption);
+
+        establish_pool_meta_bootstrap_identity_if_proven(vec![storage.clone()], &mut write_state, false)
+            .await
+            .expect("a non-elected distributed node must not create legacy adoption authority");
+        let err = load_pool_meta_for_startup(vec![storage.clone()], &mut write_state)
+            .await
+            .expect_err("legacy adoption still requires the elected writer to create a durable identity");
+        assert!(err.to_string().contains("no durable bootstrap identity"));
+        assert!(
+            !storage
+                .objects
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .contains_key(POOL_META_NAME),
+            "classification must not create pool metadata on non-elected nodes"
+        );
+    }
+
     #[tokio::test]
     async fn test_unproven_pending_identity_cannot_authorize_all_missing_pool_meta() {
         let deployment_id = Uuid::new_v4();

crates/ecstore/src/store/init_format.rs

@@ -14,6 +14,7 @@
 
 use crate::cluster::rpc::client::is_network_like_disk_error;
 use crate::config::storageclass;
+use crate::core::pools::PoolMetaBootstrapAuthority;
 use crate::disk::error_reduce::{count_errs, reduce_write_quorum_errs};
 use crate::disk::{self, DiskAPI};
 use crate::error::{Error, Result};
@@ -84,7 +85,7 @@ pub async fn connect_load_init_formats(
 
 pub(crate) struct LoadedFormat {
     pub(crate) format: FormatV3,
-    pub(crate) fresh_bootstrap_proven: bool,
+    pub(crate) pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority,
 }
 
 pub(crate) async fn connect_load_init_formats_with_instance_ctx(
@@ -133,7 +134,7 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
                 retain_format_quorum_members(instance_ctx, disks, &format, &quorum_members, set_drive_count).await?;
                 return Ok(LoadedFormat {
                     format: *format,
-                    fresh_bootstrap_proven: false,
+                    pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority::LegacyAdoption,
                 });
             }
             Ok(LegacyFormatOutcome::Incompatible) => {
@@ -153,7 +154,7 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
             let fm = init_format_erasure(instance_ctx, disks, set_count, set_drive_count, deployment_id).await?;
             return Ok(LoadedFormat {
                 format: fm,
-                fresh_bootstrap_proven: true,
+                pool_meta_bootstrap_authority: PoolMetaBootstrapAuthority::Fresh,
             });
         }
     }
@@ -182,10 +183,29 @@ pub(crate) async fn connect_load_init_formats_with_instance_ctx(
 
     Ok(LoadedFormat {
         format: fm,
-        fresh_bootstrap_proven: false,
+        pool_meta_bootstrap_authority: verified_legacy_adoption_source(disks, &formats, set_count, set_drive_count).await?,
     })
 }
 
+async fn verified_legacy_adoption_source(
+    disks: &[Option<DiskStore>],
+    rustfs_formats: &[Option<FormatV3>],
+    set_count: usize,
+    set_drive_count: usize,
+) -> Result<PoolMetaBootstrapAuthority> {
+    match try_migrate_format(disks, rustfs_formats, set_count, set_drive_count).await {
+        Ok(LegacyFormatOutcome::Migrated { .. }) => Ok(PoolMetaBootstrapAuthority::LegacyAdoption),
+        Ok(LegacyFormatOutcome::None | LegacyFormatOutcome::Incompatible) => Ok(PoolMetaBootstrapAuthority::None),
+        Err(err) => {
+            debug!(
+                error = %err,
+                "legacy adoption proof skipped because legacy format verification failed"
+            );
+            Ok(PoolMetaBootstrapAuthority::None)
+        }
+    }
+}
+
 async fn retain_format_quorum_members(
     instance_ctx: &Arc<InstanceContext>,
     disks: &mut [Option<DiskStore>],
@@ -1311,10 +1331,7 @@ mod tests {
         let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 3, None)
             .await
             .expect("fresh disks should receive a storage format");
-        assert!(
-            loaded.fresh_bootstrap_proven,
-            "every configured disk explicitly reporting unformatted should establish fresh topology proof"
-        );
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::Fresh);
         let format = loaded.format;
 
         let (formats, errors) = load_format_erasure_all(&disks, false).await;
@@ -1358,12 +1375,11 @@ mod tests {
 
         let mut expected = legacy;
         expected.erasure.this = Uuid::nil();
-        assert_eq!(
-            connect_load_init_formats(true, &mut disks, 1, 3, None)
-                .await
-                .expect("compatible legacy format should migrate"),
-            expected
-        );
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 3, None)
+            .await
+            .expect("compatible legacy format should migrate");
+        assert_eq!(loaded.format, expected);
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::LegacyAdoption);
         let (formats, errors) = load_format_erasure_all(&disks, false).await;
         assert!(
             errors.iter().all(Option::is_none),
@@ -1378,6 +1394,51 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn compatible_single_drive_legacy_format_marks_adoption_proof() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let legacy = FormatV3::new(1, 1);
+        write_legacy_majority(&disks, &legacy).await;
+
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("single-drive MinIO format should migrate");
+
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::LegacyAdoption);
+    }
+
+    #[tokio::test]
+    async fn existing_migrated_format_keeps_legacy_adoption_proof() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let legacy = FormatV3::new(1, 1);
+        write_legacy_majority(&disks, &legacy).await;
+
+        connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("first run should migrate the MinIO format");
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("retry after a partial adoption should reload the migrated RustFS format");
+
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::LegacyAdoption);
+    }
+
+    #[tokio::test]
+    async fn existing_rustfs_format_without_legacy_source_is_not_legacy_adoption() {
+        let (_temp_dir, mut disks) = local_disks(1).await;
+        let mut format = FormatV3::new(1, 1);
+        format.erasure.this = format.erasure.sets[0][0];
+        save_format_file(&disks[0], &Some(format))
+            .await
+            .expect("existing RustFS format should be written");
+
+        let loaded = connect_load_init_formats_with_instance_ctx(&current_ctx(), true, &mut disks, 1, 1, None)
+            .await
+            .expect("existing RustFS format should load");
+
+        assert_eq!(loaded.pool_meta_bootstrap_authority, PoolMetaBootstrapAuthority::None);
+    }
+
     #[tokio::test]
     async fn compatible_legacy_format_migrates_when_the_file_is_missing() {
         let (_temp_dir, mut disks) = local_disks(3).await;