fix(upgrade): probe legacy IAM namespace before migration (#7816)

crates/ecstore/src/bucket/migration.rs

@@ -356,7 +356,8 @@ where
 /// An absent legacy bucket is a no-op; migration errors prevent startup readiness.
 pub async fn try_migrate_iam_config<S>(store: Arc<S>, decrypt_fn: Option<LegacyBlobDecryptFn>) -> Result<()>
 where
-    S: ListOperations<
+    S: BucketOperations<Error = crate::error::Error>
+        + ListOperations<
             Error = crate::error::Error,
             ListObjectsV2Info = ListObjectsV2Info,
             ListObjectVersionsInfo = ListObjectVersionsInfo,
@@ -381,6 +382,24 @@ where
             DeletedObject = DeletedObject,
         >,
 {
+    // Older peers abort walk_dir streams when the legacy volume is absent,
+    // which loses the typed not-found error before listing quorum resolution.
+    // Stat the namespace first; only a confirmed missing volume skips migration.
+    match store
+        .get_bucket_info(
+            MIGRATING_META_BUCKET,
+            &BucketOptions {
+                no_metadata: true,
+                ..Default::default()
+            },
+        )
+        .await
+    {
+        Ok(_) => {}
+        Err(err) if is_err_strict_volume_not_found(&err) => return Ok(()),
+        Err(err) => return Err(err),
+    }
+
     let opts = ObjectOptions {
         max_parity: true,
         no_lock: true,

crates/iam/tests/minio_iam_migration_test.rs

@@ -98,6 +98,9 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
         .base_dir(temp_dir.path())
         .build()
         .await;
+    try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect("an absent legacy namespace must not prevent startup");
     env.make_bucket(LEGACY_META_BUCKET, false).await;
 
     for (path, body) in [
@@ -128,6 +131,34 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() {
     }
     seed_legacy_iam_object(&env, format_path, &json!({"version": 1})).await;
 
+    let mapping_path = format!("{}legacy-reader.json", IAM_CONFIG_POLICY_DB_USERS_PREFIX.as_str());
+    env.put_object_bytes(LEGACY_META_BUCKET, &mapping_path, b"invalid IAM policy mapping".to_vec())
+        .await;
+    let error = try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect_err("an existing legacy namespace must reject incompatible IAM policy mappings");
+    let io_error = std::io::Error::from(error);
+    let detail = io_error
+        .get_ref()
+        .and_then(|context| context.source())
+        .expect("failure must retain the malformed policy mapping in its source");
+    assert_eq!(detail.to_string(), format!("incompatible legacy metadata: {mapping_path}"));
+    let migrated_store = ObjectStore::new(env.ecstore.clone());
+    assert!(
+        migrated_store.load_iam_config::<Value>(&mapping_path).await.is_err(),
+        "failed IAM migration must not publish a target policy mapping"
+    );
+    seed_legacy_iam_object(&env, &mapping_path, &json!({"version": 1, "policy": "readonly"})).await;
+    try_migrate_iam_config(env.ecstore.clone(), None)
+        .await
+        .expect("an existing legacy namespace must migrate supported IAM metadata after repair");
+    let mapping: Value = migrated_store
+        .load_iam_config(&mapping_path)
+        .await
+        .expect("read the migrated policy mapping");
+    assert_eq!(mapping["version"], 1);
+    assert_eq!(mapping["policy"], "readonly");
+
     let regular_source = json!({
         "version": 1,
         "credentials": {