diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 65995b95a..1f04d4330 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -202,7 +202,9 @@ pub mod bucket { } pub mod migration { - pub use crate::bucket::migration::{LegacyBlobDecryptFn, try_migrate_bucket_metadata, try_migrate_iam_config}; + pub use crate::bucket::migration::{ + LegacyBlobDecryptFn, migration_startup_error, try_migrate_bucket_metadata, try_migrate_iam_config, + }; } pub mod object_lock { diff --git a/crates/ecstore/src/bucket/migration.rs b/crates/ecstore/src/bucket/migration.rs index b5f587221..cca13fb97 100644 --- a/crates/ecstore/src/bucket/migration.rs +++ b/crates/ecstore/src/bucket/migration.rs @@ -17,7 +17,7 @@ use crate::bucket::metadata::BUCKET_METADATA_FILE; use crate::bucket::replication::ReplicationMigrationBridge; use crate::disk::{BUCKET_META_PREFIX, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; -use crate::error::Error; +use crate::error::{Error, Result, is_err_strict_not_found, is_err_strict_volume_not_found}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; use crate::storage_api_contracts::{ bucket::{BucketOperations, BucketOptions}, @@ -33,7 +33,7 @@ use rustfs_utils::path::SLASH_SEPARATOR; use serde::{Deserialize, Serialize}; use std::sync::Arc; use time::OffsetDateTime; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; /// IAM config prefix under meta bucket (e.g. config/iam/). const IAM_CONFIG_PREFIX: &str = "config/iam"; @@ -53,6 +53,39 @@ type ListObjectVersionsInfo = StorageListObjectVersionsInfo; type ObjectInfoOrErr = StorageObjectInfoOrErr; type WalkOptions = StorageWalkOptions bool>; +#[derive(Clone, Debug, thiserror::Error)] +enum MigrationMetadataError { + #[error("empty legacy metadata: {0}")] + Empty(String), + #[error("incompatible legacy metadata: {0}")] + Incompatible(String), +} + +impl From for Error { + fn from(error: MigrationMetadataError) -> Self { + let message = match &error { + MigrationMetadataError::Empty(_) => "empty legacy metadata", + MigrationMetadataError::Incompatible(_) => "incompatible legacy metadata", + }; + // Keep the record path in the typed source, not in the quorum grouping key. + Self::other_with_context(message, error) + } +} + +/// Converts a migration failure at the startup boundary, rendering the safe +/// record path while leaving storage-layer error grouping stable. +pub fn migration_startup_error(error: Error) -> std::io::Error { + if let Error::Io(io_error) = &error + && let Some(metadata_error) = io_error + .get_ref() + .and_then(|context| context.source()) + .and_then(|source| source.downcast_ref::()) + { + return std::io::Error::other(metadata_error.clone()); + } + std::io::Error::other(error) +} + /// Callback used to decrypt an at-rest config blob during MinIO -> RustFS migration. /// /// MinIO encrypts IAM identity/service-account files and the server config at rest @@ -211,7 +244,7 @@ fn normalize_bucket_meta_blob(path: &str, data: &[u8]) -> std::result::Result(store: Arc) +pub async fn try_migrate_bucket_metadata(store: Arc) -> Result<()> where S: BucketOperations + ObjectIO< @@ -231,25 +264,18 @@ where DeletedObject = DeletedObject, >, { - let buckets_list = match store + let buckets_list = store .list_bucket(&BucketOptions { no_metadata: true, ..Default::default() }) - .await - { - Ok(b) => b, - Err(e) => { - warn!("list buckets failed (skip migration): {e}"); - return; - } - }; + .await?; let buckets: Vec = buckets_list.into_iter().map(|b| b.name).collect(); if buckets.is_empty() { debug!("No migrating bucket metadata found"); - return; + return Ok(()); } debug!("Found {} migrating bucket metadata, migrating...", buckets.len()); @@ -263,26 +289,40 @@ where for bucket in buckets { let meta_path = format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{bucket}{SLASH_SEPARATOR}{BUCKET_METADATA_FILE}"); - migrate_one_if_missing(store.clone(), &opts, &h, &meta_path, &format!("bucket metadata: {bucket}")).await; + migrate_one_if_missing(store.clone(), &opts, &h, &meta_path, &format!("bucket metadata: {bucket}")).await?; let resync_path = format!( "{BUCKET_META_PREFIX}{SLASH_SEPARATOR}{bucket}{SLASH_SEPARATOR}{REPLICATION_META_DIR}{SLASH_SEPARATOR}{RESYNC_META_FILE}" ); - migrate_one_if_missing(store.clone(), &opts, &h, &resync_path, &format!("bucket replication resync: {bucket}")).await; + migrate_one_if_missing(store.clone(), &opts, &h, &resync_path, &format!("bucket replication resync: {bucket}")).await?; + } + Ok(()) +} + +async fn migration_target_exists(store: &S, path: &str) -> Result { + match store + .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default()) + .await + { + Ok(_) => Ok(true), + Err(err) if is_err_strict_not_found(&err) || is_err_strict_volume_not_found(&err) => Ok(false), + Err(err) => Err(err), } } -async fn migrate_one_if_missing(store: Arc, opts: &ObjectOptions, headers: &HeaderMap, path: &str, label: &str) +async fn migrate_one_if_missing( + store: Arc, + opts: &ObjectOptions, + headers: &HeaderMap, + path: &str, + label: &str, +) -> Result<()> where S: EcstoreObjectIO + EcstoreObjectOperations, { - if store - .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default()) - .await - .is_ok() - { + if migration_target_exists(store.as_ref(), path).await? { debug!("{label} already exists in RustFS, skip"); - return; + return Ok(()); } let mut rd = match store @@ -290,43 +330,31 @@ where .await { Ok(r) => r, - Err(e) => { - debug!("read migrating {label}: {e}"); - return; - } + // Ordinary RustFS deployments have no legacy bucket, and optional + // legacy settings (such as replication resync) may not exist. + Err(err) if is_err_strict_not_found(&err) || is_err_strict_volume_not_found(&err) => return Ok(()), + Err(err) => return Err(err), }; - let data = match rd.read_all().await { - Ok(d) if !d.is_empty() => d, - Ok(_) => return, - Err(e) => { - debug!("read migrating {label} body: {e}"); - return; - } - }; - - let data = match normalize_bucket_meta_blob(path, &data) { - Ok(Some(normalized)) => normalized, - Ok(None) => data, - Err(e) => { - warn!("skip {label} migration due to incompatible format: {e}"); - return; - } - }; + let data = rd.read_all().await?; + if data.is_empty() { + return Err(MigrationMetadataError::Empty(path.to_owned()).into()); + } + let data = normalize_bucket_meta_blob(path, &data) + .map_err(|_| MigrationMetadataError::Incompatible(path.to_owned()))? + .unwrap_or(data); let mut put_data = PutObjReader::from_vec(data); - if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await { - warn!("write {label}: {e}"); - } else { - info!("Migrated {label}"); - } + store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, opts).await?; + info!("Migrated {label}"); + Ok(()) } /// Migrates IAM config from legacy meta bucket `config/iam/` to RustFS meta bucket. /// Lists all objects under the IAM prefix in the source, copies each to the target if not present. /// Skips objects that already exist in RustFS (idempotent). -/// If list_objects_v2 on the legacy bucket fails (e.g. format differs), migration is skipped. -pub async fn try_migrate_iam_config(store: Arc, decrypt_fn: Option) +/// An absent legacy bucket is a no-op; migration errors prevent startup readiness. +pub async fn try_migrate_iam_config(store: Arc, decrypt_fn: Option) -> Result<()> where S: ListOperations< Error = crate::error::Error, @@ -366,47 +394,36 @@ where loop { let list_result = match store .clone() - .list_objects_v2(MIGRATING_META_BUCKET, &prefix, continuation, None, 500, false, None, false) + .list_objects_v2(MIGRATING_META_BUCKET, &prefix, continuation.clone(), None, 500, false, None, false) .await { Ok(r) => r, - Err(e) => { - debug!("list IAM config from legacy bucket failed (skip migration): {e}"); - return; - } + Err(err) if is_err_strict_volume_not_found(&err) => return Ok(()), + Err(err) => return Err(err), }; for obj in list_result.objects { let path = &obj.name; - if path.is_empty() || path.ends_with('/') { + // Unsupported records must not trigger target lookups, reads, or decryption. + if path != IAM_FORMAT_FILE_PATH + && !is_identity_path(path) + && !is_group_path(path) + && !is_policy_doc_path(path) + && !is_policy_mapping_path(path) + { continue; } - if store - .get_object_info(RUSTFS_META_BUCKET, path, &ObjectOptions::default()) - .await - .is_ok() - { + if migration_target_exists(store.as_ref(), path).await? { debug!("IAM config already exists in RustFS, skip: {path}"); continue; } - let mut rd = match store + let mut rd = store .get_object_reader(MIGRATING_META_BUCKET, path, None, h.clone(), &opts) - .await - { - Ok(r) => r, - Err(e) => { - debug!("read migrating IAM config {path}: {e}"); - continue; - } - }; - let data = match rd.read_all().await { - Ok(d) if !d.is_empty() => d, - Ok(_) => continue, - Err(e) => { - debug!("read migrating IAM config {path} body: {e}"); - continue; - } - }; + .await?; + let data = rd.read_all().await?; + if data.is_empty() { + return Err(MigrationMetadataError::Empty(path.to_owned()).into()); + } // MinIO encrypts IAM identity/service-account files at rest. Decrypt // before normalizing; fall back to the raw bytes when no key applies // (plaintext blobs, or nothing to decrypt) so existing behavior holds. @@ -420,22 +437,17 @@ where debug!("skip unsupported IAM config path during migration: {path}"); continue; } - Err(e) => { - warn!("skip IAM config migration due to incompatible format, path: {path}, err: {e}"); - continue; - } + // Parser errors may contain credential data. Report only the path. + Err(_) => return Err(MigrationMetadataError::Incompatible(path.to_owned()).into()), }; let mut put_data = PutObjReader::from_vec(data); - if let Err(e) = store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await { - warn!("write IAM config {path}: {e}"); - } else { - info!("Migrated IAM config: {path}"); - total_migrated += 1; - } + store.put_object(RUSTFS_META_BUCKET, path, &mut put_data, &opts).await?; + info!("Migrated IAM config: {path}"); + total_migrated += 1; } - continuation = list_result.next_continuation_token.or(list_result.continuation_token); - if !list_result.is_truncated || continuation.is_none() { + continuation = next_iam_migration_page(list_result.is_truncated, continuation, list_result.next_continuation_token)?; + if continuation.is_none() { break; } } @@ -443,10 +455,74 @@ where if total_migrated > 0 { info!("IAM migration complete: {} object(s) migrated", total_migrated); } + Ok(()) +} + +fn next_iam_migration_page(truncated: bool, previous: Option, next: Option) -> Result> { + if !truncated { + return Ok(None); + } + let next = next.filter(|token| !token.is_empty()); + if next.is_none() || next == previous { + return Err(Error::other("legacy IAM migration listing did not advance")); + } + Ok(next) } #[cfg(test)] mod tests { + #[test] + fn migration_errors_group_by_cause_and_retain_typed_record_context() { + use super::{Error, MigrationMetadataError}; + + for (make_error, message) in [ + ( + MigrationMetadataError::Empty as fn(String) -> MigrationMetadataError, + "empty legacy metadata", + ), + (MigrationMetadataError::Incompatible, "incompatible legacy metadata"), + ] { + let first: Error = make_error("buckets/first/.metadata.bin".into()).into(); + let second: Error = make_error("buckets/second/.metadata.bin".into()).into(); + assert_eq!(first, second, "record paths must not fragment error grouping"); + assert_eq!(first.clone(), second, "cloning must preserve error grouping"); + + let io_error = std::io::Error::from(first); + let detail = io_error + .get_ref() + .and_then(|context| context.source()) + .expect("record context must remain in the error source"); + assert!(detail.downcast_ref::().is_some()); + assert!(detail.to_string().contains("buckets/first/.metadata.bin")); + + let startup_error = super::migration_startup_error(make_error("buckets/startup/.metadata.bin".into()).into()); + assert!( + startup_error + .get_ref() + .is_some_and(|source| source.is::()) + ); + assert_eq!(startup_error.to_string(), format!("{message}: buckets/startup/.metadata.bin")); + } + assert_ne!( + Error::from(MigrationMetadataError::Empty("record".into())), + Error::from(MigrationMetadataError::Incompatible("record".into())), + "different migration failures must remain distinguishable" + ); + } + + #[test] + fn truncated_iam_listing_cannot_report_completed_migration() { + use super::next_iam_migration_page; + assert_eq!(next_iam_migration_page(false, Some("old".into()), None).expect("final page"), None); + assert_eq!( + next_iam_migration_page(true, Some("old".into()), Some("next".into())).expect("advancing page"), + Some("next".into()) + ); + for next in [None, Some(String::new()), Some("old".into())] { + assert!(next_iam_migration_page(true, Some("old".into()), next).is_err()); + } + } + use super::{normalize_bucket_meta_blob, normalize_iam_config_blob}; use crate::bucket::replication::{ BucketReplicationResyncStatus, ReplicationMigrationBridge, ResyncStatusType, TargetReplicationResyncStatus, @@ -659,6 +735,13 @@ mod tests { .collect(); crate::bucket::metadata_sys::init_bucket_metadata_sys(ecstore.clone(), existing).await; + super::try_migrate_bucket_metadata(ecstore.clone()) + .await + .expect("fresh stores do not require a legacy metadata bucket"); + super::try_migrate_iam_config(ecstore.clone(), None) + .await + .expect("fresh stores do not require a legacy IAM bucket"); + let meta_path = format!("{BUCKET_META_PREFIX}{SLASH_SEPARATOR}interop{SLASH_SEPARATOR}{BUCKET_METADATA_FILE}"); let put_opts = ObjectOptions::default(); @@ -680,8 +763,31 @@ mod tests { .await .expect("seed .minio.sys bucket metadata"); - // --- Run the real startup migration. --- - super::try_migrate_bucket_metadata(ecstore.clone()).await; + // A partial import must report failure, even if the main bucket + // metadata copied successfully before an incompatible resync record. + let resync_path = format!("{BUCKET_META_PREFIX}/interop/.replication/resync.bin"); + ecstore + .put_object( + MIGRATING_META_BUCKET, + &resync_path, + &mut PutObjReader::from_vec(b"invalid resync metadata".to_vec()), + &put_opts, + ) + .await + .expect("seed malformed legacy resync metadata"); + assert!( + super::try_migrate_bucket_metadata(ecstore.clone()).await.is_err(), + "incompatible native metadata must not be reported as a completed migration" + ); + ecstore + .delete_object(MIGRATING_META_BUCKET, &resync_path, ObjectOptions::default()) + .await + .expect("remove invalid optional legacy resync record"); + + // Retry the real startup migration after repairing the source. + super::try_migrate_bucket_metadata(ecstore.clone()) + .await + .expect("native bucket metadata migration completes"); // --- The migrated `.rustfs.sys` blob must carry every MinIO config, --- // byte-identical to the source (typed XML/JSON parsing of these fields is diff --git a/crates/iam/tests/minio_iam_migration_test.rs b/crates/iam/tests/minio_iam_migration_test.rs index a8d3aac57..8f6199cbe 100644 --- a/crates/iam/tests/minio_iam_migration_test.rs +++ b/crates/iam/tests/minio_iam_migration_test.rs @@ -100,6 +100,34 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() { .await; env.make_bucket(LEGACY_META_BUCKET, false).await; + for (path, body) in [ + ("config/iam/empty.json", Vec::new()), + ("config/iam/users/ignored/extra.json", b"not JSON".to_vec()), + ] { + env.put_object_bytes(LEGACY_META_BUCKET, path, body).await; + } + try_migrate_iam_config( + env.ecstore.clone(), + Some(std::sync::Arc::new(|_| panic!("unsupported IAM records must not be decrypted"))), + ) + .await + .expect("unsupported IAM records, including empty objects, must be skipped"); + + let format_path = "config/iam/format.json"; + for body in [Vec::new(), b"invalid IAM format".to_vec()] { + env.put_object_bytes(LEGACY_META_BUCKET, format_path, body).await; + let error = try_migrate_iam_config(env.ecstore.clone(), None) + .await + .expect_err("empty or incompatible supported IAM metadata must prevent startup readiness"); + let io_error = std::io::Error::from(error); + let detail = io_error + .get_ref() + .and_then(|context| context.source()) + .expect("failure must retain the supported record in its source"); + assert!(detail.to_string().contains(format_path), "failure must identify the supported record"); + } + seed_legacy_iam_object(&env, format_path, &json!({"version": 1})).await; + let regular_source = json!({ "version": 1, "credentials": { @@ -155,7 +183,12 @@ async fn minio_permanent_identities_survive_migration_and_repeated_iam_loads() { ) .await; - try_migrate_iam_config(env.ecstore.clone(), None).await; + try_migrate_iam_config(env.ecstore.clone(), None) + .await + .expect("legacy IAM migration completes after source repair"); + try_migrate_iam_config(env.ecstore.clone(), None) + .await + .expect("completed legacy IAM migration is idempotent"); let store = ObjectStore::new(env.ecstore); assert_identity_survives( diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index a85d71ab3..1f54d0b62 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -62,10 +62,10 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc, c let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); - try_migrate_bucket_metadata(store.clone()).await; + try_migrate_bucket_metadata(store.clone()).await?; init_on_demand_migration_runtime(); init_bucket_metadata_sys(store.clone(), buckets.clone()).await; - try_migrate_iam_config(store).await; + try_migrate_iam_config(store).await?; spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx.clone(), false); Ok(buckets) @@ -82,9 +82,9 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); - try_migrate_bucket_metadata(store.clone()).await; + try_migrate_bucket_metadata(store.clone()).await?; - try_migrate_iam_config(store.clone()).await; + try_migrate_iam_config(store.clone()).await?; init_on_demand_migration_runtime(); init_bucket_metadata_sys(store, buckets.clone()).await; spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx, true); diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 5c8a095ec..1e14df65b 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -1185,17 +1185,21 @@ pub(crate) fn get_global_transition_state() -> Arc { ecstore_bucket::lifecycle::bucket_lifecycle_ops::get_global_transition_state() } -pub(crate) async fn try_migrate_bucket_metadata(store: Arc) { - ecstore_bucket::migration::try_migrate_bucket_metadata(store).await; +pub(crate) async fn try_migrate_bucket_metadata(store: Arc) -> std::io::Result<()> { + ecstore_bucket::migration::try_migrate_bucket_metadata(store) + .await + .map_err(ecstore_bucket::migration::migration_startup_error) } -pub(crate) async fn try_migrate_iam_config(store: Arc) { +pub(crate) async fn try_migrate_iam_config(store: Arc) -> std::io::Result<()> { // MinIO encrypts IAM identity/service-account files at rest with a key derived // from the root credentials. Inject the IAM crate's decryption so those blobs // are decrypted before normalization instead of being skipped as "incompatible". let decrypt_fn: ecstore_bucket::migration::LegacyBlobDecryptFn = Arc::new(|data: &[u8]| rustfs_iam::try_decrypt_iam_blob(data)); - ecstore_bucket::migration::try_migrate_iam_config(store, Some(decrypt_fn)).await; + ecstore_bucket::migration::try_migrate_iam_config(store, Some(decrypt_fn)) + .await + .map_err(ecstore_bucket::migration::migration_startup_error) } pub(crate) fn init_ecstore_config() { diff --git a/rustfs/tests/native_migration_startup_test.rs b/rustfs/tests/native_migration_startup_test.rs new file mode 100644 index 000000000..65dfbabee --- /dev/null +++ b/rustfs/tests/native_migration_startup_test.rs @@ -0,0 +1,326 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#![recursion_limit = "256"] + +use reqwest::StatusCode; +use rustfs::embedded::{RustFSServerBuilder, find_available_port}; +use rustfs_ecstore::api::config::com::{delete_config, read_config}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::Stdio; +use std::time::Duration; +use tokio::process::Command; + +mod common; + +const TEST_NAME: &str = "native_migration_failure_blocks_server_startup_and_repair_preserves_records"; +const STAGE_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_STAGE"; +const ROOT_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_ROOT"; +const ADDRESS_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_ADDRESS"; +const FAILURE_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_FAILURE"; +const STOP_ENV: &str = "RUSTFS_NATIVE_MIGRATION_TEST_STOP"; +const ACCESS_KEY: &str = "native-migration-root"; +const SECRET_KEY: &str = "native-migration-root-secret"; +const LEGACY_BUCKET: &str = ".minio.sys"; +const TARGET_BUCKET: &str = ".rustfs.sys"; +const BUCKET_METADATA: &str = "buckets/interop/.metadata.bin"; +const IAM_RECORD: &str = "config/iam/groups/migration-group/members.json"; +const IAM_FORMAT: &str = "config/iam/format.json"; +const EXISTING_FORMAT: &[u8] = br#"{"version":1}"#; +const STARTUP_TIMEOUT: Duration = Duration::from_secs(60); + +#[derive(Clone, Copy, Debug)] +enum StartupMode { + Server, + Embedded, +} + +fn volumes(root: &Path) -> Vec { + (1..=4).map(|index| root.join(format!("disk{index}"))).collect() +} + +fn minio_bucket_metadata() -> Vec { + let hex: String = include_str!("../../crates/ecstore/tests/fixtures/minio/bucket_metadata.blob.hex") + .chars() + .filter(|ch| !ch.is_whitespace()) + .collect(); + hex.as_bytes() + .as_chunks::<2>() + .0 + .iter() + .map(|pair| { + u8::from_str_radix(std::str::from_utf8(pair).expect("fixture hex is UTF-8"), 16).expect("valid MinIO fixture hex") + }) + .collect() +} + +async fn prepare_or_verify_fixture(root: &Path, seed: bool) { + let env = rustfs_test_utils::TestECStoreEnv::builder() + .base_dir(root) + .disk_count(4) + .build() + .await; + if seed { + env.make_bucket("interop", false).await; + env.make_bucket(LEGACY_BUCKET, false).await; + env.put_object_bytes(LEGACY_BUCKET, BUCKET_METADATA, minio_bucket_metadata()) + .await; + env.put_object_bytes( + LEGACY_BUCKET, + IAM_RECORD, + br#"{"version":1,"status":"enabled","members":[],"updatedAt":"2026-09-10T00:00:00Z"}"#.to_vec(), + ) + .await; + env.put_object_bytes(TARGET_BUCKET, IAM_FORMAT, EXISTING_FORMAT.to_vec()) + .await; + // A completed record must be skipped before reading even a broken old copy. + env.put_object_bytes(LEGACY_BUCKET, IAM_FORMAT, b"do not overwrite the existing target".to_vec()) + .await; + delete_config(env.ecstore.clone(), BUCKET_METADATA) + .await + .expect("leave bucket metadata pending migration"); + } else { + assert_eq!( + read_config(env.ecstore.clone(), BUCKET_METADATA) + .await + .expect("migrated bucket metadata"), + minio_bucket_metadata(), + "migration must preserve the MinIO bucket settings" + ); + let group: serde_json::Value = serde_json::from_slice( + &read_config(env.ecstore.clone(), IAM_RECORD) + .await + .expect("migrated IAM group"), + ) + .expect("valid migrated IAM JSON"); + assert_eq!(group["status"], "enabled"); + assert_eq!(group["members"], serde_json::json!([])); + } + assert_eq!( + read_config(env.ecstore.clone(), IAM_FORMAT) + .await + .expect("existing IAM format"), + EXISTING_FORMAT, + "retry must not overwrite records already migrated" + ); +} + +async fn run_embedded_child(root: &Path) { + let address = std::env::var(ADDRESS_ENV).expect("embedded child address"); + let result = RustFSServerBuilder::new() + .address(address) + .access_key(ACCESS_KEY) + .secret_key(SECRET_KEY) + .volumes(volumes(root).iter().map(|path| path.to_string_lossy().into_owned()).collect()) + .build() + .await; + match result { + Ok(server) => { + let stop = PathBuf::from(std::env::var_os(STOP_ENV).expect("embedded stop path")); + while !stop.exists() { + tokio::time::sleep(Duration::from_millis(25)).await; + } + server.shutdown().await; + } + Err(error) => { + fs::write(std::env::var_os(FAILURE_ENV).expect("embedded failure path"), error.to_string()) + .expect("record the actual embedded startup error"); + } + } +} + +fn child_command(root: &Path, stage: &str, log: &Path) -> Command { + let mut command = Command::new(std::env::current_exe().expect("integration test executable")); + command + .args(["--exact", TEST_NAME, "--nocapture"]) + .env(STAGE_ENV, stage) + .env(ROOT_ENV, root); + configure_process(&mut command, log); + command +} + +fn configure_process(command: &mut Command, log: &Path) { + let output = fs::File::create(log).expect("create isolated process log"); + command + // These disposable erasure volumes intentionally share the test runner's disk. + .env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true") + .env("RUSTFS_CONSOLE_ENABLE", "false") + .env("NO_PROXY", "localhost,127.0.0.1,::1") + .env("no_proxy", "localhost,127.0.0.1,::1") + .env("RUST_LOG", "warn") + .stdin(Stdio::null()) + .stdout(Stdio::from(output.try_clone().expect("clone process log"))) + .stderr(Stdio::from(output)) + .kill_on_drop(true); +} + +async fn fixture_process(root: &Path, stage: &str) { + let log = root.join(format!("{stage}.log")); + let status = tokio::time::timeout(STARTUP_TIMEOUT, child_command(root, stage, &log).status()) + .await + .expect("fixture process must finish") + .expect("run fixture process"); + assert!(status.success(), "{stage} failed: {}", fs::read_to_string(log).expect("fixture log")); +} + +async fn check_startup(root: &Path, mode: StartupMode, failure_record: Option<&str>, label: &str) { + let ready = failure_record.is_none(); + let address = format!("127.0.0.1:{}", find_available_port().expect("free startup probe port")); + let log = root.join(format!("{label}.log")); + let failure = root.join(format!("{label}.failure")); + let stop = root.join(format!("{label}.stop")); + let mut command = match mode { + StartupMode::Server => { + let mut command = Command::new(env!("CARGO_BIN_EXE_rustfs")); + command + .args(["--address", &address, "--access-key", ACCESS_KEY, "--secret-key", SECRET_KEY]) + .args(volumes(root)); + configure_process(&mut command, &log); + command + } + StartupMode::Embedded => { + let mut command = child_command(root, "embedded", &log); + command + .env(ADDRESS_ENV, &address) + .env(FAILURE_ENV, &failure) + .env(STOP_ENV, &stop); + command + } + }; + let mut child = command.spawn().expect("start isolated server process"); + let http = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_millis(500)) + .build() + .expect("local readiness client"); + let result = tokio::time::timeout(STARTUP_TIMEOUT, async { + loop { + if let Ok(response) = http.get(format!("http://{address}/health/ready")).send().await + && response.status() == StatusCode::OK + { + assert!(ready, "{mode:?} published Ready after a migration I/O failure"); + return; + } + if let Some(status) = child.try_wait().expect("poll server process") { + let details = fs::read_to_string(&log).expect("startup log"); + assert!(!ready, "{mode:?} exited before Ready ({status}): {details}"); + let record = failure_record.expect("failed startup has an obstructed record"); + match mode { + StartupMode::Server => { + assert_eq!(status.code(), Some(1), "startup must fail: {details}"); + assert_migration_io_error(&details, record); + } + StartupMode::Embedded => { + assert!(status.success(), "embedded test process failed unexpectedly: {details}"); + let error = fs::read_to_string(&failure).expect("embedded startup returned an error"); + assert_migration_io_error(&error, record); + } + } + return; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await; + assert!( + result.is_ok(), + "{mode:?} did not reach the expected startup outcome: {}", + fs::read_to_string(&log).expect("startup diagnostics") + ); + if ready { + match mode { + StartupMode::Embedded => { + fs::write(stop, b"stop").expect("request embedded shutdown"); + assert!( + tokio::time::timeout(STARTUP_TIMEOUT, child.wait()) + .await + .expect("embedded shutdown completes") + .expect("wait for embedded shutdown") + .success() + ); + } + StartupMode::Server => child.kill().await.expect("stop the isolated server"), + } + } +} + +fn assert_migration_io_error(error: &str, record: &str) { + let lower = error.to_ascii_lowercase(); + assert!( + (lower.contains("access denied") + || lower.contains("access is denied") + || lower.contains("not a directory") + || lower.contains("not regular")) + && error.contains(&format!("{TARGET_BUCKET}/{record}")), + "startup must fail because of the obstructed metadata record, not an unrelated initialization error: {error}" + ); +} + +async fn run_startup_cases(mode: StartupMode) { + let ordinary = tempfile::TempDir::with_prefix("rustfs-no-legacy-").expect("ordinary store"); + for volume in volumes(ordinary.path()) { + fs::create_dir_all(volume).expect("ordinary volume"); + } + check_startup(ordinary.path(), mode, None, "ordinary").await; + + let control = tempfile::TempDir::with_prefix("rustfs-migration-control-").expect("control fixture"); + fixture_process(control.path(), "seed").await; + check_startup(control.path(), mode, None, "control").await; + fixture_process(control.path(), "verify").await; + + for record in [BUCKET_METADATA, IAM_RECORD] { + let target = tempfile::TempDir::with_prefix("rustfs-migration-failure-").expect("disposable migration target"); + fixture_process(target.path(), "seed").await; + let blockers: Vec<_> = volumes(target.path()) + .iter() + .map(|volume| volume.join(TARGET_BUCKET).join(record)) + .collect(); + for blocker in &blockers { + fs::create_dir_all(blocker.parent().expect("record parent")).expect("create target parent"); + assert!(!blocker.exists(), "the record must still need migration"); + // A non-directory target causes real filesystem I/O errors even when tests run as root. + fs::write(blocker, b"blocked migration target").expect("block only the destination record"); + } + check_startup(target.path(), mode, Some(record), "blocked").await; + for blocker in blockers { + fs::remove_file(blocker).expect("repair the same partially migrated target"); + } + check_startup(target.path(), mode, None, "repaired").await; + fixture_process(target.path(), "verify").await; + } +} + +#[test] +fn native_migration_failure_blocks_server_startup_and_repair_preserves_records() { + // Cold processes keep failed initialization and cached metadata out of subsequent restart attempts. + common::run_embedded_test(|| async { + match std::env::var(STAGE_ENV).ok().as_deref() { + Some("seed") => { + prepare_or_verify_fixture(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root")), true).await + } + Some("verify") => { + prepare_or_verify_fixture(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root")), false).await + } + Some("embedded") => run_embedded_child(&PathBuf::from(std::env::var_os(ROOT_ENV).expect("fixture root"))).await, + None => run_startup_cases(StartupMode::Server).await, + Some(stage) => panic!("unknown native migration test stage: {stage}"), + } + }); +} + +#[test] +fn native_migration_failure_blocks_embedded_startup_and_repair_preserves_records() { + common::run_embedded_test(|| run_startup_cases(StartupMode::Embedded)); +}