diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index e9cdd5be3..34382f429 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -15729,7 +15729,7 @@ mod test { assert!(slots[0][0][1].is_some()); drop(slots); drop(entries); - let (disks, errors) = crate::store::init_format::init_disks(&endpoints, &options).await; + let (disks, errors) = crate::store::init_format::init_disks(&ctx, &endpoints, &options).await; assert!(disks[0].is_none()); assert!(disks[1].is_some()); assert_eq!(errors[0], Some(DiskError::FileAccessDenied)); diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index e60297250..e0031ca06 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -540,6 +540,7 @@ impl ECStore { // failures can still spawn recovery probes for peers that come up // after this node. let (mut disks, errs) = init_format::init_disks( + &instance_ctx, &pool_eps.endpoints, &DiskOption { cleanup: true, diff --git a/crates/ecstore/src/store/init_format.rs b/crates/ecstore/src/store/init_format.rs index 8de59ec40..c93d17612 100644 --- a/crates/ecstore/src/store/init_format.rs +++ b/crates/ecstore/src/store/init_format.rs @@ -44,12 +44,32 @@ const LEGACY_FORMAT_READ_CONCURRENCY: usize = 16; const LEGACY_FORMAT_BASE_MAX_BYTES: usize = 16 * 1024; const LEGACY_FORMAT_MAX_BYTES_PER_DISK: usize = 64; -pub async fn init_disks(eps: &Endpoints, opt: &DiskOption) -> (Vec>, Vec>) { +pub async fn init_disks( + instance_ctx: &InstanceContext, + eps: &Endpoints, + opt: &DiskOption, +) -> (Vec>, Vec>) { let mut futures = Vec::with_capacity(eps.as_ref().len()); + let local_disk_map = instance_ctx.local_disk_map(); + let local_disks = local_disk_map.read().await; for ep in eps.as_ref().iter() { - futures.push(new_disk(ep, opt)); + // Foundation disks may already serve RPC writes. Opening them again + // with startup cleanup can move live staging out from under a writer. + let registered = if ep.is_local { + local_disks.get(&ep.to_string()).cloned().flatten() + } else { + None + }; + futures.push(async move { + match registered { + Some(disk) if disk.endpoint() == *ep => Ok(disk), + Some(_) => Err(DiskError::InconsistentDisk), + None => new_disk(ep, opt).await, + } + }); } + drop(local_disks); let mut res = Vec::with_capacity(eps.as_ref().len()); let mut errors = Vec::with_capacity(eps.as_ref().len()); @@ -1030,6 +1050,7 @@ mod tests { } let (disks, errors) = init_disks( + &InstanceContext::new(), &Endpoints::from(endpoints), &DiskOption { cleanup: false, @@ -1049,6 +1070,91 @@ mod tests { (temp_dir, disks) } + #[tokio::test] + async fn init_disks_preserves_live_staging_on_registered_local_disks() { + let (temp_dir, disks) = local_disks(1).await; + let registered = disks[0].as_ref().expect("registered disk should exist"); + let instance_ctx = Arc::new(InstanceContext::new()); + record_local_disks(&instance_ctx, vec![registered.clone()]).await; + let staging = temp_dir.path().join("disk-0/.rustfs.sys/tmp/in-flight/xl.meta"); + tokio::fs::create_dir_all(staging.parent().expect("staging has a parent")) + .await + .expect("create staging directory"); + tokio::fs::write(&staging, b"in-flight pool metadata") + .await + .expect("stage metadata through the registered disk"); + + let (initialized, errors) = init_disks( + &instance_ctx, + &Endpoints::from(vec![registered.endpoint()]), + &DiskOption { + cleanup: true, + health_check: false, + }, + ) + .await; + + assert!(errors.iter().all(Option::is_none), "disk initialization failed: {errors:?}"); + assert_eq!( + tokio::fs::read(&staging) + .await + .expect("startup must preserve live RPC staging"), + b"in-flight pool metadata" + ); + assert!(Arc::ptr_eq(registered, initialized[0].as_ref().expect("initialized disk should exist"))); + } + + #[tokio::test] + async fn init_disks_cleans_stale_staging_without_a_registered_disk() { + let temp_dir = tempfile::tempdir().expect("create disk root"); + let endpoint = + Endpoint::try_from(temp_dir.path().to_str().expect("disk root should be UTF-8")).expect("endpoint should parse"); + let staging = temp_dir.path().join(".rustfs.sys/tmp/stale/xl.meta"); + tokio::fs::create_dir_all(staging.parent().expect("staging has a parent")) + .await + .expect("create stale staging directory"); + tokio::fs::write(&staging, b"stale pool metadata") + .await + .expect("write stale staging"); + + let (disks, errors) = init_disks( + &InstanceContext::new(), + &Endpoints::from(vec![endpoint]), + &DiskOption { + cleanup: true, + health_check: false, + }, + ) + .await; + + assert!(errors.iter().all(Option::is_none), "disk initialization failed: {errors:?}"); + assert!(disks[0].is_some()); + assert!(!staging.exists(), "a fresh disk open must still clear stale staging"); + } + + #[tokio::test] + async fn init_disks_rejects_registered_disk_topology_mismatch() { + let (_temp_dir, disks) = local_disks(1).await; + let registered = disks[0].as_ref().expect("registered disk should exist"); + let instance_ctx = Arc::new(InstanceContext::new()); + record_local_disks(&instance_ctx, vec![registered.clone()]).await; + let mut endpoint = registered.endpoint(); + endpoint.set_pool_index(1); + + let (initialized, errors) = init_disks( + &instance_ctx, + &Endpoints::from(vec![endpoint]), + &DiskOption { + cleanup: true, + health_check: false, + }, + ) + .await; + + assert!(initialized[0].is_none(), "a registered disk must not be reopened in a different slot"); + assert_eq!(errors[0], Some(DiskError::InconsistentDisk)); + } + async fn format_bytes(disks: &[Option]) -> Vec>> { let mut snapshots = Vec::with_capacity(disks.len()); for disk in disks { @@ -1758,6 +1864,7 @@ mod tests { disk.set_disk_id(None).await.expect("test disk ID should be reset"); } let (prewarm_disks, prewarm_errors) = init_disks( + &InstanceContext::new(), &prewarm_endpoints, &DiskOption { cleanup: false, diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 549ed536b..d6d345183 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -3295,6 +3295,7 @@ mod tests { init_local_disks(endpoint_pools.clone()).await.expect("init local disks"); let (mut disks, errs) = init_disks( + &crate::runtime::global::current_ctx(), &endpoint_pools.as_ref().first().expect("pool endpoints").endpoints, &DiskOption { cleanup: true,