diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 1183d7c18..29c76765d 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -1049,6 +1049,10 @@ impl LocalDiskWrapper { Ok(()) } + pub(crate) async fn set_disk_id_state(&self, id: Option) { + *self.disk_id.write().await = id; + } + /// Get the current disk ID pub async fn get_current_disk_id(&self) -> Option { *self.disk_id.read().await diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 4ac894373..a8074ddd9 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -132,6 +132,18 @@ pub enum Disk { Remote(Box), } +impl Disk { + pub(crate) async fn set_disk_id_state(&self, id: Option) -> Result<()> { + match self { + Disk::Local(local_disk) => { + local_disk.set_disk_id_state(id).await; + Ok(()) + } + Disk::Remote(remote_disk) => remote_disk.set_disk_id(id).await, + } + } +} + #[async_trait::async_trait] impl DiskAPI for Disk { fn to_string(&self) -> String { @@ -1552,6 +1564,72 @@ mod tests { let _ = fs::remove_dir_all(&test_dir).await; } + #[tokio::test] + #[serial_test::serial] + async fn local_disk_id_state_does_not_publish_to_the_process_registry() { + let local_dir = tempfile::tempdir().expect("local disk tempdir should be created"); + let mut endpoint = + Endpoint::try_from(local_dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + let local_disk = LocalDisk::new(&endpoint, false).await.expect("local disk should initialize"); + let disk = Disk::Local(Box::new(LocalDiskWrapper::new(Arc::new(local_disk), false))); + let disk_id = Uuid::new_v4(); + + disk.set_disk_id_state(Some(disk_id)) + .await + .expect("local wrapper state should accept a disk ID"); + + let Disk::Local(local_disk) = &disk else { + panic!("test disk should remain local"); + }; + assert_eq!(local_disk.get_current_disk_id().await, Some(disk_id)); + assert!( + !crate::runtime::global::current_ctx() + .local_disk_id_map() + .read() + .await + .contains_key(&disk_id), + "state-only startup publication must not update the process disk-ID registry" + ); + + disk.set_disk_id_state(None) + .await + .expect("local wrapper state should clear a disk ID"); + assert_eq!(local_disk.get_current_disk_id().await, None); + } + + #[tokio::test] + async fn remote_disk_id_state_delegates_some_and_none() { + let mut endpoint = Endpoint::try_from("http://remote-server:9000/data").expect("remote endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + let remote_disk = RemoteDisk::new( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + Arc::new(crate::cluster::rpc::TcpHttpInternodeDataTransport), + ) + .await + .expect("remote disk should initialize"); + let disk = Disk::Remote(Box::new(remote_disk)); + let disk_id = Uuid::new_v4(); + + disk.set_disk_id_state(Some(disk_id)) + .await + .expect("remote state should accept a disk ID"); + assert_eq!(disk.get_disk_id().await.expect("remote disk ID should be readable"), Some(disk_id)); + + disk.set_disk_id_state(None) + .await + .expect("remote state should clear a disk ID"); + assert_eq!(disk.get_disk_id().await.expect("remote disk ID should be readable"), None); + } + #[tokio::test] async fn reset_health_for_store_init_retry_delegates_to_disk_variants() { let local_dir = tempfile::tempdir().unwrap(); diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index f2df75d2a..8a16f0c26 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -27,7 +27,7 @@ use crate::{ bucket::replication::{DynReplicationPool, ReplicationStats}, config::{get_global_storage_class, get_global_storage_class_snapshot, set_global_storage_class, storageclass}, disk::{DiskAPI, DiskOption, DiskStore, new_disk}, - error::Result, + error::{Error, Result}, layout::endpoints::{EndpointServerPools, SetupType}, runtime::global::{ GLOBAL_BOOT_TIME, GLOBAL_LIFECYCLE_SYS, GLOBAL_LOCAL_NODE_NAME_FALLBACK, GLOBAL_ROOT_DISK_THRESHOLD, @@ -417,10 +417,6 @@ pub(crate) async fn clear_local_disk_id_map_for_test() { local_disk_id_map_handle().write().await.clear(); } -pub(crate) async fn record_local_disk_id(instance_ctx: &Arc, disk_id: Uuid, endpoint: String) { - instance_ctx.local_disk_id_map().write().await.insert(disk_id, endpoint); -} - pub(crate) async fn replace_local_disk_id(previous: Option, current: Option, endpoint: String) { let id_map = local_disk_id_map_handle(); let mut disk_id_map = id_map.write().await; @@ -436,6 +432,53 @@ pub(crate) async fn replace_local_disk_id(previous: Option, current: Optio } } +pub(crate) async fn reconcile_local_disk_ids( + instance_ctx: &InstanceContext, + pool_endpoints: &[String], + selected: &[(Uuid, String)], +) { + let pool_endpoints = pool_endpoints.iter().map(String::as_str).collect::>(); + let disk_id_map = instance_ctx.local_disk_id_map(); + let mut disk_ids = disk_id_map.write().await; + disk_ids.retain(|_, registered_endpoint| !pool_endpoints.contains(registered_endpoint.as_str())); + disk_ids.extend(selected.iter().cloned()); +} + +pub(crate) async fn quarantine_local_disks(instance_ctx: &InstanceContext, endpoints: &[Endpoint]) -> Result<()> { + let slots = endpoints + .iter() + .map(|endpoint| { + Ok(( + usize::try_from(endpoint.pool_idx).map_err(|_| Error::CorruptedFormat)?, + usize::try_from(endpoint.set_idx).map_err(|_| Error::CorruptedFormat)?, + usize::try_from(endpoint.disk_idx).map_err(|_| Error::CorruptedFormat)?, + )) + }) + .collect::>>()?; + + let local_disk_map = instance_ctx.local_disk_map(); + let mut local_disks = local_disk_map.write().await; + for endpoint in endpoints { + local_disks.insert(endpoint.to_string(), None); + } + drop(local_disks); + + let set_drives = instance_ctx.local_disk_set_drives(); + let mut local_set_drives = set_drives.write().await; + if local_set_drives.is_empty() { + return Ok(()); + } + for (pool_idx, set_idx, disk_idx) in slots { + let disk = local_set_drives + .get_mut(pool_idx) + .and_then(|sets| sets.get_mut(set_idx)) + .and_then(|disks| disks.get_mut(disk_idx)) + .ok_or(Error::CorruptedFormat)?; + *disk = None; + } + Ok(()) +} + pub(crate) async fn record_local_disks(instance_ctx: &Arc, disks: Vec) { let map = instance_ctx.local_disk_map(); let mut global_local_disk_map = map.write().await; @@ -563,8 +606,8 @@ pub(crate) async fn init_tier_config_mgr(store: Arc) -> Result<()> { #[cfg(test)] mod tests { use super::{ - LockRegistry, clear_local_disk_id_map_for_test, local_disk_path_by_id, local_node_name, replace_local_disk_id, - set_local_node_name, + LockRegistry, clear_local_disk_id_map_for_test, local_disk_path_by_id, local_node_name, reconcile_local_disk_ids, + replace_local_disk_id, set_local_node_name, }; use crate::disk::endpoint::Endpoint; use rustfs_lock::{LocalClient, LockClient}; @@ -628,4 +671,65 @@ mod tests { assert_eq!(local_disk_path_by_id(&disk_id).await, Some("endpoint-a".to_string())); clear_local_disk_id_map_for_test().await; } + + #[tokio::test] + #[serial_test::serial] + async fn reconciling_pool_disk_ids_preserves_other_endpoints() { + let instance_ctx = Arc::new(crate::runtime::instance::InstanceContext::new()); + let process_ctx = crate::runtime::global::current_ctx(); + let bootstrap_ctx = crate::runtime::instance::bootstrap_ctx(); + let retained_id = Uuid::new_v4(); + let removed_id = Uuid::new_v4(); + let selected_id = Uuid::new_v4(); + let process_sentinel = Uuid::new_v4(); + let bootstrap_sentinel = Uuid::new_v4(); + instance_ctx.local_disk_id_map().write().await.extend([ + (retained_id, "endpoint-a".to_string()), + (removed_id, "endpoint-b".to_string()), + ]); + process_ctx + .local_disk_id_map() + .write() + .await + .insert(process_sentinel, "endpoint-b".to_string()); + bootstrap_ctx + .local_disk_id_map() + .write() + .await + .insert(bootstrap_sentinel, "endpoint-b".to_string()); + + reconcile_local_disk_ids( + &instance_ctx, + &["endpoint-b".to_string(), "endpoint-c".to_string()], + &[(selected_id, "endpoint-c".to_string())], + ) + .await; + + let disk_ids = instance_ctx.local_disk_id_map(); + let disk_ids = disk_ids.read().await; + assert_eq!(disk_ids.get(&retained_id).map(String::as_str), Some("endpoint-a")); + assert_eq!(disk_ids.get(&removed_id), None); + assert_eq!(disk_ids.get(&selected_id).map(String::as_str), Some("endpoint-c")); + drop(disk_ids); + assert_eq!( + process_ctx + .local_disk_id_map() + .read() + .await + .get(&process_sentinel) + .map(String::as_str), + Some("endpoint-b") + ); + assert_eq!( + bootstrap_ctx + .local_disk_id_map() + .read() + .await + .get(&bootstrap_sentinel) + .map(String::as_str), + Some("endpoint-b") + ); + process_ctx.local_disk_id_map().write().await.remove(&process_sentinel); + bootstrap_ctx.local_disk_id_map().write().await.remove(&bootstrap_sentinel); + } } diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 4e7128188..c89320908 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -313,7 +313,8 @@ impl ECStore { let mut times = 0; let mut interval = 1; loop { - match init_format::connect_load_init_formats( + match init_format::connect_load_init_formats_with_instance_ctx( + &instance_ctx, pool_first_is_local, &mut disks, pool_eps.set_count, @@ -1259,6 +1260,16 @@ mod tests { let registered: Vec = instance_ctx.local_disk_map().read().await.keys().cloned().collect(); assert_eq!(registered.len(), 4, "the passed context must register all four local disks"); + let registered_disk_ids = instance_ctx.local_disk_id_map(); + let registered_disk_ids = registered_disk_ids.read().await; + assert_eq!(registered_disk_ids.len(), 4, "the passed context must publish all four disk IDs"); + for endpoint in registered_disk_ids.values() { + assert!( + registered.contains(endpoint), + "every disk ID in the passed context must resolve to one of its registered endpoints" + ); + } + drop(registered_disk_ids); let bootstrap = crate::runtime::instance::bootstrap_ctx(); assert_ne!( bootstrap.deployment_id(), @@ -1273,6 +1284,15 @@ mod tests { "the bootstrap context must not absorb the fresh store's disks" ); } + drop(bootstrap_map); + let bootstrap_disk_ids = bootstrap.local_disk_id_map(); + let bootstrap_disk_ids = bootstrap_disk_ids.read().await; + for endpoint in bootstrap_disk_ids.values() { + assert!( + !registered.contains(endpoint), + "the bootstrap context must not absorb the fresh store's disk IDs" + ); + } } #[tokio::test] diff --git a/crates/ecstore/src/store/init_format.rs b/crates/ecstore/src/store/init_format.rs index bd4ccb301..95cf17c66 100644 --- a/crates/ecstore/src/store/init_format.rs +++ b/crates/ecstore/src/store/init_format.rs @@ -16,6 +16,8 @@ use crate::config::storageclass; use crate::disk::error_reduce::{count_errs, reduce_write_quorum_errs}; use crate::disk::{self, DiskAPI}; use crate::error::{Error, Result}; +use crate::runtime::instance::InstanceContext; +use crate::runtime::sources::{quarantine_local_disks, reconcile_local_disk_ids}; use crate::{ disk::{ DiskInfoOptions, DiskOption, DiskStore, FORMAT_CONFIG_FILE, MIGRATING_META_BUCKET, RUSTFS_META_BUCKET, @@ -26,7 +28,10 @@ use crate::{ layout::endpoints::Endpoints, }; use futures::{future::join_all, stream, stream::StreamExt}; -use std::collections::{HashMap, HashSet}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; use tokio::io::AsyncReadExt; use tracing::{debug, error, info, warn}; use uuid::Uuid; @@ -62,12 +67,25 @@ pub async fn init_disks(eps: &Endpoints, opt: &DiskOption) -> (Vec], set_count: usize, set_drive_count: usize, deployment_id: Option, +) -> Result { + let instance_ctx = crate::runtime::global::current_ctx(); + connect_load_init_formats_with_instance_ctx(&instance_ctx, first_disk, disks, set_count, set_drive_count, deployment_id).await +} + +pub(crate) async fn connect_load_init_formats_with_instance_ctx( + instance_ctx: &Arc, + first_disk: bool, + disks: &mut [Option], + set_count: usize, + set_drive_count: usize, + deployment_id: Option, ) -> Result { let (formats, errs) = load_format_erasure_all(disks, false).await; @@ -98,7 +116,7 @@ pub async fn connect_load_init_formats( match try_migrate_format(disks, &formats, set_count, set_drive_count).await { Ok(LegacyFormatOutcome::Migrated { format, quorum_members }) => { info!("Migrated format from MinIO config"); - retain_format_quorum_members(disks, &format, &quorum_members, set_drive_count).await?; + retain_format_quorum_members(instance_ctx, disks, &format, &quorum_members, set_drive_count).await?; return Ok(*format); } Ok(LegacyFormatOutcome::Incompatible) => { @@ -115,7 +133,7 @@ pub async fn connect_load_init_formats( Err(e) => return Err(e), } if all_unformatted { - let fm = init_format_erasure(disks, set_count, set_drive_count, deployment_id).await?; + let fm = init_format_erasure(instance_ctx, disks, set_count, set_drive_count, deployment_id).await?; return Ok(fm); } } @@ -140,12 +158,13 @@ pub async fn connect_load_init_formats( None => select_format_erasure_in_quorum(&formats, 0)?, }; check_format_erasure_value_for_topology(&fm, formats.len(), set_drive_count)?; - retain_format_quorum_members(disks, &fm, &quorum_members, set_drive_count).await?; + retain_format_quorum_members(instance_ctx, disks, &fm, &quorum_members, set_drive_count).await?; Ok(fm) } async fn retain_format_quorum_members( + instance_ctx: &Arc, disks: &mut [Option], format: &FormatV3, quorum_members: &[bool], @@ -154,26 +173,73 @@ async fn retain_format_quorum_members( if set_drive_count == 0 || quorum_members.len() != disks.len() { return Err(Error::CorruptedFormat); } - for (disk, belongs_to_quorum) in disks.iter().zip(quorum_members) { - if !belongs_to_quorum && let Some(disk) = disk { - disk.set_disk_id(None).await?; + + let pool_idx = disks + .iter() + .flatten() + .map(|disk| disk.endpoint().pool_idx) + .find_map(|pool_idx| usize::try_from(pool_idx).ok()); + let registered_endpoints = if let Some(pool_idx) = pool_idx { + instance_ctx + .local_disk_set_drives() + .read() + .await + .get(pool_idx) + .map(|sets| { + sets.iter() + .flat_map(|set| set.iter()) + .map(|disk| disk.as_ref().map(|disk| disk.endpoint())) + .collect::>() + }) + .filter(|endpoints| endpoints.len() == disks.len()) + } else { + None + }; + let endpoints = disks + .iter() + .enumerate() + .map(|(index, disk)| { + disk.as_ref() + .map(|disk| disk.endpoint()) + .or_else(|| registered_endpoints.as_ref()?.get(index)?.clone()) + }) + .collect::>(); + + let mut member_disk_ids = vec![None; disks.len()]; + let mut local_pool_endpoints = Vec::new(); + let mut selected_local_disk_ids = Vec::new(); + let mut quarantined_endpoints = Vec::new(); + for (index, ((disk, endpoint), belongs_to_quorum)) in disks.iter().zip(&endpoints).zip(quorum_members).enumerate() { + if let Some(endpoint) = endpoint.as_ref().filter(|endpoint| endpoint.is_local) { + local_pool_endpoints.push(endpoint.to_string()); + if !belongs_to_quorum { + quarantined_endpoints.push(endpoint.clone()); + } } - } - for (index, (disk, belongs_to_quorum)) in disks.iter().zip(quorum_members).enumerate() { if !belongs_to_quorum { continue; } + let disk = disk.as_ref().ok_or(Error::CorruptedFormat)?; let disk_id = format .erasure .sets .get(index / set_drive_count) .and_then(|set| set.get(index % set_drive_count)) + .copied() .ok_or(Error::CorruptedFormat)?; - disk.as_ref() - .ok_or(Error::CorruptedFormat)? - .set_disk_id(Some(*disk_id)) - .await?; + member_disk_ids[index] = Some(disk_id); + if disk.is_local() { + selected_local_disk_ids.push((disk_id, disk.endpoint().to_string())); + } } + quarantine_local_disks(instance_ctx, &quarantined_endpoints).await?; + + for (disk, disk_id) in disks.iter().zip(member_disk_ids) { + if let Some(disk) = disk { + disk.set_disk_id_state(disk_id).await?; + } + } + reconcile_local_disk_ids(instance_ctx, &local_pool_endpoints, &selected_local_disk_ids).await; for (disk, belongs_to_quorum) in disks.iter_mut().zip(quorum_members) { if !belongs_to_quorum { *disk = None; @@ -207,7 +273,8 @@ pub fn check_disk_fatal_errs(errs: &[Option]) -> disk::error::Result< } async fn init_format_erasure( - disks: &[Option], + instance_ctx: &Arc, + disks: &mut [Option], set_count: usize, set_drive_count: usize, deployment_id: Option, @@ -227,9 +294,11 @@ async fn init_format_erasure( } } - save_format_file_all(disks, &fms).await?; - - get_format_erasure_in_quorum(&fms, 0) + write_format_file_all(disks, &fms).await?; + let format = get_format_erasure_in_quorum(&fms, 0)?; + let quorum_members = vec![true; disks.len()]; + retain_format_quorum_members(instance_ctx, disks, &format, &quorum_members, set_drive_count).await?; + Ok(format) } /// Outcome of attempting to migrate an on-disk MinIO `format.json`. @@ -348,44 +417,59 @@ async fn try_migrate_format( .iter() .zip(&formats_to_write) .zip(rustfs_formats) - .filter(|&((disk, _), existing)| disk.is_some() && existing.is_none()) - .map(|((disk, format), _)| save_format_file(disk, format)); + .filter_map(|((disk, format), existing)| { + if existing.is_some() { + return None; + } + Some(write_format_file(disk.as_ref()?, format.as_ref()?)) + }); let mut write_error = None; for result in join_all(writes).await { if let Err(error) = result { write_error.get_or_insert(error); } } - for disk in disks.iter().flatten() { - disk.set_disk_id(None).await?; - } - let (persisted_formats, persisted_errors) = load_format_erasure_all(disks, false).await; + let Some((persisted_format, quorum_members)) = + select_persisted_migration_format(&persisted_formats, persisted_errors, &format, set_drive_count, write_error)? + else { + return Ok(LegacyFormatOutcome::Incompatible); + }; + + Ok(LegacyFormatOutcome::Migrated { + format: Box::new(persisted_format), + quorum_members, + }) +} + +fn select_persisted_migration_format( + persisted_formats: &[Option], + persisted_errors: Vec>, + reference: &FormatV3, + set_drive_count: usize, + write_error: Option, +) -> Result)>> { + if !formats_match_reference_slots(persisted_formats, reference, 0) { + return Ok(None); + } + let quorum_error = match select_format_erasure_in_quorum(persisted_formats, 0) { + Ok((format, quorum_members)) => { + check_format_erasure_value_for_topology(&format, persisted_formats.len(), set_drive_count)?; + return Ok(Some((format, quorum_members))); + } + Err(error) => error, + }; if let Some(error) = persisted_errors .into_iter() .flatten() .find(|error| !matches!(error, DiskError::UnformattedDisk | DiskError::DiskNotFound)) { return match error { - DiskError::CorruptedFormat | DiskError::CorruptedBackend | DiskError::InconsistentDisk => { - Ok(LegacyFormatOutcome::Incompatible) - } + DiskError::CorruptedFormat | DiskError::CorruptedBackend | DiskError::InconsistentDisk => Ok(None), error => Err(error.into()), }; } - if !formats_match_reference_slots(&persisted_formats, &format, 0) { - return Ok(LegacyFormatOutcome::Incompatible); - } - let (persisted_format, quorum_members) = match select_format_erasure_in_quorum(&persisted_formats, 0) { - Ok(selected) => selected, - Err(error) => return Err(write_error.map_or(error, Into::into)), - }; - check_format_erasure_value_for_topology(&persisted_format, disks.len(), set_drive_count)?; - - Ok(LegacyFormatOutcome::Migrated { - format: Box::new(persisted_format), - quorum_members, - }) + Err(write_error.map_or(quorum_error, Into::into)) } fn legacy_format_max_bytes(disk_count: usize) -> Result { @@ -676,13 +760,12 @@ pub async fn load_format_erasure(disk: &DiskStore, heal: bool) -> disk::error::R Ok(fm) } -async fn save_format_file_all(disks: &[Option], formats: &[Option]) -> disk::error::Result<()> { - let mut futures = Vec::with_capacity(disks.len()); - - for (i, disk) in disks.iter().enumerate() { - futures.push(save_format_file(disk, &formats[i])); - } - +async fn write_format_file_all(disks: &[Option], formats: &[Option]) -> disk::error::Result<()> { + let futures = disks.iter().zip(formats).map(|(disk, format)| async move { + let disk = disk.as_ref().ok_or(DiskError::DiskNotFound)?; + let format = format.as_ref().ok_or_else(|| DiskError::other("format is none"))?; + write_format_file(disk, format).await + }); let mut errors = Vec::with_capacity(disks.len()); let results = join_all(futures).await; @@ -713,6 +796,13 @@ pub async fn save_format_file(disk: &Option, format: &Option disk::error::Result<()> { let json_data = format.to_json()?; let tmpfile = Uuid::new_v4().to_string(); @@ -723,8 +813,6 @@ pub async fn save_format_file(disk: &Option, format: &Option Result { mod tests { use super::*; use crate::layout::endpoint::Endpoint; - use crate::runtime::sources::{clear_local_disk_id_map_for_test, local_disk_path_by_id}; + use crate::runtime::global::{current_ctx, reset_local_disk_test_state}; + use crate::runtime::sources::{local_disk_path_by_id, record_local_disks}; + use crate::store::peer::{find_local_disk_by_ref, prewarm_local_disk_id_map_with_instance_ctx}; use serial_test::serial; async fn local_disks(count: usize) -> (tempfile::TempDir, Vec>) { @@ -1045,7 +1135,7 @@ mod tests { #[tokio::test] #[serial] async fn existing_format_load_publishes_only_validated_quorum_disk_ids() { - clear_local_disk_id_map_for_test().await; + reset_local_disk_test_state().await; let (_temp_dir, mut disks) = local_disks(5).await; let canonical = FormatV3::new(1, 5); for (index, disk) in disks.iter().enumerate().take(3) { @@ -1055,19 +1145,56 @@ mod tests { .await .expect("canonical format should be written"); } - let mut wrong_slot = canonical.clone(); - wrong_slot.erasure.this = wrong_slot.erasure.sets[0][0]; - save_format_file(&disks[3], &Some(wrong_slot)) - .await - .expect("wrong-slot format should be written"); + let canonical_id = canonical.erasure.sets[0][0]; let mut foreign = FormatV3::new(1, 5); foreign.erasure.this = foreign.erasure.sets[0][4]; let foreign_id = foreign.erasure.this; save_format_file(&disks[4], &Some(foreign)) .await .expect("foreign format should be written"); - let canonical_id = canonical.erasure.sets[0][0]; + let mut collision = FormatV3::new(1, 5); + collision.erasure.sets[0][3] = canonical_id; + collision.erasure.this = canonical_id; + save_format_file(&disks[3], &Some(collision)) + .await + .expect("foreign collision format should be written"); let canonical_endpoint = disks[0].as_ref().expect("canonical disk should exist").endpoint().to_string(); + let collision_endpoint = disks[3].as_ref().expect("collision disk should exist").endpoint().to_string(); + let foreign_endpoint = disks[4].as_ref().expect("foreign disk should exist").endpoint().to_string(); + let prewarm_endpoints = Endpoints::from( + disks + .iter() + .map(|disk| disk.as_ref().expect("test disk should exist").endpoint()) + .collect::>(), + ); + for disk in disks.iter().flatten() { + disk.set_disk_id(None).await.expect("test disk ID should be reset"); + } + let (prewarm_disks, prewarm_errors) = init_disks( + &prewarm_endpoints, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await; + assert!(prewarm_errors.iter().all(Option::is_none)); + let prewarm_disks = prewarm_disks + .into_iter() + .map(|disk| disk.expect("prewarm disk should exist")) + .collect::>(); + let instance_ctx = current_ctx(); + record_local_disks(&instance_ctx, prewarm_disks.clone()).await; + *instance_ctx.local_disk_set_drives().write().await = vec![vec![prewarm_disks.into_iter().map(Some).collect()]]; + prewarm_local_disk_id_map_with_instance_ctx(&instance_ctx).await; + instance_ctx + .local_disk_id_map() + .write() + .await + .insert(canonical_id, collision_endpoint.clone()); + assert!(local_disk_path_by_id(&foreign_id).await.is_some(), "foreign ID should be prewarmed"); + assert_eq!(local_disk_path_by_id(&canonical_id).await, Some(collision_endpoint)); + disks[4] = None; connect_load_init_formats(true, &mut disks, 1, 5, None) .await @@ -1075,9 +1202,51 @@ mod tests { assert!(disks[..3].iter().all(Option::is_some)); assert!(disks[3..].iter().all(Option::is_none)); - assert_eq!(local_disk_path_by_id(&canonical_id).await, Some(canonical_endpoint)); + assert_eq!(local_disk_path_by_id(&canonical_id).await, Some(canonical_endpoint.clone())); assert_eq!(local_disk_path_by_id(&foreign_id).await, None); - clear_local_disk_id_map_for_test().await; + assert!( + instance_ctx + .local_disk_map() + .read() + .await + .get(&foreign_endpoint) + .is_some_and(Option::is_none) + ); + assert!(instance_ctx.local_disk_set_drives().read().await[0][0][4].is_none()); + assert!(find_local_disk_by_ref(&foreign_id.to_string()).await.is_none()); + assert_eq!( + find_local_disk_by_ref(&canonical_id.to_string()) + .await + .map(|disk| disk.endpoint().to_string()), + Some(canonical_endpoint) + ); + reset_local_disk_test_state().await; + } + + #[test] + fn persisted_migration_quorum_ignores_nonmember_read_errors() { + for drive_count in [3, 4] { + let reference = FormatV3::new(1, drive_count); + let required = drive_count / 2 + 1; + let mut formats = vec![None; drive_count]; + let mut errors = (0..drive_count).map(|_| None).collect::>>(); + for (index, format) in formats.iter_mut().enumerate().take(required) { + let mut disk_format = reference.clone(); + disk_format.erasure.this = reference.erasure.sets[0][index]; + *format = Some(disk_format); + } + errors[drive_count - 1] = Some(DiskError::FaultyDisk); + + let (selected, members) = + select_persisted_migration_format(&formats, errors, &reference, drive_count, Some(DiskError::FaultyDisk)) + .expect("a persisted strict majority must outrank a nonmember read error") + .expect("the persisted formats should be compatible"); + + assert_eq!(selected.shared_identity(), reference.shared_identity()); + assert_eq!(members.iter().filter(|member| **member).count(), required); + assert!(members[..required].iter().all(|member| *member)); + assert!(members[required..].iter().all(|member| !*member)); + } } #[tokio::test] diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index e827c7fa4..d8ba54d1f 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -28,8 +28,26 @@ async fn remember_local_disk_id(disk: &DiskStore) -> Option { async fn remember_local_disk_id_with_instance_ctx(instance_ctx: &Arc, disk: &DiskStore) -> Option { let disk_id = disk.get_disk_id().await.ok().flatten()?; - runtime_sources::record_local_disk_id(instance_ctx, disk_id, disk.endpoint().to_string()).await; - Some(disk_id) + record_local_disk_id_if_active(instance_ctx, disk, disk_id) + .await + .then_some(disk_id) +} + +async fn record_local_disk_id_if_active(instance_ctx: &Arc, disk: &DiskStore, disk_id: Uuid) -> bool { + let endpoint = disk.endpoint().to_string(); + let local_disk_map = instance_ctx.local_disk_map(); + let local_disks = local_disk_map.read().await; + let Some(active_disk) = local_disks.get(&endpoint).and_then(Option::as_ref) else { + return false; + }; + if !Arc::ptr_eq(active_disk, disk) { + return false; + } + + // Lock order is local_disk_map -> local_disk_id_map so quarantine is the + // linearization point for rejecting an in-flight stale disk snapshot. + instance_ctx.local_disk_id_map().write().await.insert(disk_id, endpoint); + true } pub async fn find_local_disk(disk_path: &str) -> Option { @@ -228,6 +246,7 @@ pub async fn get_disk_infos(disks: &[Option]) -> Vec #[cfg(test)] mod tests { use super::*; + use crate::disk::new_disk; use crate::layout::endpoints::{Endpoints, PoolEndpoints}; fn single_local_disk_pools(dir: &std::path::Path) -> EndpointServerPools { @@ -314,4 +333,56 @@ mod tests { ); } } + + #[tokio::test] + async fn stale_local_disk_snapshot_cannot_repopulate_the_id_registry() { + let temp_dir = tempfile::tempdir().expect("create temp disk dir"); + let endpoint_pools = single_local_disk_pools(temp_dir.path()); + let instance_ctx = Arc::new(InstanceContext::new()); + init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools) + .await + .expect("local disk should be registered"); + let disk = instance_ctx + .local_disk_map() + .read() + .await + .values() + .find_map(|disk| disk.clone()) + .expect("registered local disk"); + let endpoint = disk.endpoint().to_string(); + let disk_id = Uuid::new_v4(); + + let local_disk_map = instance_ctx.local_disk_map(); + let mut quarantine = local_disk_map.write().await; + let replacement = new_disk( + &disk.endpoint(), + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("replacement disk should initialize"); + assert!(!Arc::ptr_eq(&disk, &replacement)); + let task_ctx = instance_ctx.clone(); + let task_disk = disk.clone(); + let remember = tokio::spawn(async move { record_local_disk_id_if_active(&task_ctx, &task_disk, disk_id).await }); + tokio::task::yield_now().await; + quarantine.insert(endpoint.clone(), Some(replacement.clone())); + drop(quarantine); + + assert!(!remember.await.expect("stale lookup task should complete")); + assert!(!instance_ctx.local_disk_id_map().read().await.contains_key(&disk_id)); + let active = instance_ctx + .local_disk_map() + .read() + .await + .get(&endpoint) + .cloned() + .flatten() + .expect("replacement disk should remain registered"); + assert!(Arc::ptr_eq(&active, &replacement)); + assert!(record_local_disk_id_if_active(&instance_ctx, &replacement, disk_id).await); + assert_eq!(instance_ctx.local_disk_id_map().read().await.get(&disk_id), Some(&endpoint)); + } }