diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index d152c9cb0..c60526a85 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -16,6 +16,7 @@ use crate::pb::master_pb; use crate::pb::master_pb::seaweed_client::SeaweedClient; use crate::pb::volume_server_pb; use crate::pb::volume_server_pb::volume_server_server::VolumeServer; +use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT; use crate::storage::needle::needle::{self, Needle}; use crate::storage::types::*; @@ -1436,8 +1437,14 @@ impl VolumeServer for VolumeGrpcService { break; } // disk_id=0 means "unset" (protobuf default), so auto-select - // mirrors VolumeEcShardsCopy: prefer a disk already holding - // this volume's shards, then any HDD, then any disk. + // using the same primitive as volume_ec_shards_copy: prefer + // a disk that has the EC volume mounted, then a disk that + // owns the .ecx on disk (volume not yet mounted — relevant + // when shards stream in mid-rebuild before any + // VolumeEcShardsMount has happened; see #9212), then any + // HDD, then any disk. Pass the build's default data-shard + // count for free-slot maths; the helper takes it as a + // parameter so custom-ratio builds can swap it. let vid = VolumeId(info.volume_id); let dir = if info.disk_id > 0 { let count = store.locations.len(); @@ -1450,17 +1457,13 @@ impl VolumeServer for VolumeGrpcService { } Some(store.locations[info.disk_id as usize].directory.clone()) } else { - let loc_idx = store - .find_free_location_predicate(|loc| loc.has_ec_volume(vid)) - .or_else(|| { - store.find_free_location_predicate(|loc| { - loc.disk_type == DiskType::HardDrive - }) - }) - .or_else(|| { - store.find_free_location_predicate(|_| true) - }); - loc_idx.map(|i| store.locations[i].directory.clone()) + store + .find_ec_shard_target_location( + &info.collection, + vid, + DATA_SHARDS_COUNT as u32, + ) + .map(|i| store.locations[i].directory.clone()) }; drop(store); let dir = match dir { @@ -2250,9 +2253,17 @@ impl VolumeServer for VolumeGrpcService { let req = request.into_inner(); let vid = VolumeId(req.volume_id); - // Select target location matching Go's 3-tier fallback: - // When disk_id > 0: use that specific location - // When disk_id == 0 (unset): (1) location with existing EC shards, (2) any HDD, (3) any + // Select target location: + // When disk_id > 0: use that specific location. + // When disk_id == 0 (unset): auto-select via + // find_ec_shard_target_location, which prefers a disk that + // already has the EC volume mounted, then a disk that owns the + // .ecx on disk (volume not yet mounted — relevant for + // ec.rebuild, where only the first shard carries .ecx and + // subsequent shards must land on the same disk; see #9212), + // then any HDD, then any disk. Pass the build's default + // data-shard count; the helper takes it as a parameter so + // custom-ratio builds can swap it. let (dest_dir, dest_idx_dir) = { let store = self.state.store.read().unwrap(); let count = store.locations.len(); @@ -2268,20 +2279,11 @@ impl VolumeServer for VolumeGrpcService { let loc = &store.locations[req.disk_id as usize]; (loc.directory.clone(), loc.idx_directory.clone()) } else { - // Auto-select: prefer location with existing EC shards for this volume - let loc_idx = store - .find_free_location_predicate(|loc| loc.has_ec_volume(vid)) - .or_else(|| { - // Fall back to any HDD location - store.find_free_location_predicate(|loc| { - loc.disk_type == DiskType::HardDrive - }) - }) - .or_else(|| { - // Fall back to any location - store.find_free_location_predicate(|_| true) - }); - match loc_idx { + match store.find_ec_shard_target_location( + &req.collection, + vid, + DATA_SHARDS_COUNT as u32, + ) { Some(i) => { let loc = &store.locations[i]; (loc.directory.clone(), loc.idx_directory.clone()) diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index b336d0dd4..47f36a0fb 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -544,6 +544,40 @@ impl DiskLocation { self.ec_volumes.contains_key(&vid) } + /// Reports whether this disk has a sealed `.ecx` index file for the + /// given (collection, vid). Unlike [`Self::has_ec_volume`] this does + /// not require the EC volume to be mounted in memory, which makes it + /// the right primitive for placement decisions during `ec.balance` / + /// `ec.rebuild` flows where shards may arrive before any + /// `VolumeEcShardsMount` has happened on the receiving disk. Without + /// checking the on-disk state, auto-select can split shards from the + /// `.ecx` that travels with the first shard, which is the source of + /// the orphan-shard layout reported in seaweedfs/seaweedfs#9212. + /// + /// Mirrors `DiskLocation.HasEcxFileOnDisk` in + /// `weed/storage/disk_location_ec.go`. Skips entries that are + /// directories so a stray dir named `_.ecx` doesn't + /// register as a present index file. + pub fn has_ecx_file_on_disk(&self, collection: &str, vid: VolumeId) -> bool { + let idx_base = volume_file_name(&self.idx_directory, collection, vid); + let idx_path = format!("{}.ecx", idx_base); + if let Ok(meta) = fs::metadata(&idx_path) { + if !meta.is_dir() { + return true; + } + } + if self.idx_directory != self.directory { + let data_base = volume_file_name(&self.directory, collection, vid); + let data_path = format!("{}.ecx", data_base); + if let Ok(meta) = fs::metadata(&data_path) { + if !meta.is_dir() { + return true; + } + } + } + false + } + /// Remove an EC volume, returning it. pub fn remove_ec_volume(&mut self, vid: VolumeId) -> Option { self.ec_volumes.remove(&vid) diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 7fcadf48c..c56e340d8 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -197,6 +197,72 @@ impl Store { best.map(|(i, _)| i) } + /// Returns the index of the disk that should receive a new EC + /// shard / index file for `(collection, vid)`. Selection order: + /// + /// 1. a disk that already has the EC volume mounted (in-memory state), + /// 2. a disk that owns the `.ecx` file on disk (volume not yet mounted), + /// 3. any HDD with free space, + /// 4. any disk with free space. + /// + /// Step 2 is the missing primitive that pinned subsequent shards to + /// the first-shard disk during `ec.rebuild`: rebuild only sets + /// `CopyEcxFile=true` on the first shard, then relies on auto-select + /// to land later shards on the same disk. Without an on-disk check, + /// `has_ec_volume` returns false (no mount yet) and the fallback + /// picks "any HDD with free space" — which can split shards from + /// their index files across disks of the same node and lose them at + /// startup. See seaweedfs/seaweedfs#9212. + /// + /// `data_shard_count` is taken as a parameter rather than read from + /// `DATA_SHARDS_COUNT` so custom-ratio builds can swap the default + /// without touching this helper. + /// + /// Single pass over `self.locations` with tier scoring; the + /// highest-tier disk wins, ties broken by free shard-slot count. + /// Mirrors `Store.FindEcShardTargetLocation` in + /// `weed/storage/store_ec.go`. + pub fn find_ec_shard_target_location( + &self, + collection: &str, + vid: VolumeId, + data_shard_count: u32, + ) -> Option { + const TIER_ANY_DISK: u8 = 1; + const TIER_HDD: u8 = 2; + const TIER_ECX_ON_DISK: u8 = 3; + const TIER_MOUNTED: u8 = 4; + + let mut best: Option<(usize, u8, i64)> = None; + for (i, loc) in self.locations.iter().enumerate() { + if loc.is_disk_space_low.load(Ordering::Relaxed) { + continue; + } + let free = ec_free_shard_count(loc, data_shard_count); + if free <= 0 { + continue; + } + let mut tier = TIER_ANY_DISK; + if loc.disk_type == DiskType::HardDrive { + tier = TIER_HDD; + } + if loc.has_ecx_file_on_disk(collection, vid) { + tier = TIER_ECX_ON_DISK; + } + if loc.has_ec_volume(vid) { + tier = TIER_MOUNTED; + } + let better = match best { + None => true, + Some((_, b_tier, b_free)) => tier > b_tier || (tier == b_tier && free > b_free), + }; + if better { + best = Some((i, tier, free)); + } + } + best.map(|(i, _, _)| i) + } + /// Create a new volume, placing it on the location with the most free space. pub fn add_volume( &mut self, @@ -920,6 +986,48 @@ fn save_vif_volume_info(path: &str, info: &VifVolumeInfo) -> Result<(), VolumeEr Ok(()) } +/// Free EC shard capacity of `loc`, expressed in shard slots (not +/// volume-equivalent slots). `find_free_location_predicate` does similar +/// math but divides by `data_shard_count` at the end. That truncation can +/// exclude a disk that still has room for several individual shards +/// (e.g. `MaxVolumeCount=1`, `EcShardCount=1`, `data_shard_count=10` → +/// reports 0 despite 9 free shard slots), which would re-route subsequent +/// shards off the `.ecx`-owning disk and re-introduce the orphan-shard +/// layout this helper is meant to prevent (seaweedfs/seaweedfs#9212). +/// +/// `MaxVolumeCount == 0` is the "unlimited" sentinel honoured elsewhere +/// in the store; report a synthetic large free count decremented by +/// current usage so unlimited disks are eligible and tie-breaks still +/// prefer the less-loaded one. +/// +/// Mirrors `ecFreeShardCount` in `weed/storage/store_ec.go`. +fn ec_free_shard_count(loc: &DiskLocation, data_shard_count: u32) -> i64 { + if data_shard_count == 0 { + return 0; + } + let dsc = data_shard_count as i64; + let max = loc.max_volume_count.load(Ordering::Relaxed) as i64; + if max <= 0 { + // Synthetic "unlimited" capacity. Use a large but + // well-below-overflow base (1 << 60 ≈ 1.15e18) and saturating + // arithmetic so even pathological usage never wraps and + // tie-breaks among unlimited disks still meaningfully prefer + // the less-loaded one. Clamp to ≥ 1 so the disk stays eligible + // for placement no matter how loaded it is. + const UNLIMITED_FREE: i64 = 1 << 60; + let used = (loc.volumes_len() as i64) + .saturating_mul(dsc) + .saturating_add(loc.ec_shard_count() as i64); + return UNLIMITED_FREE.saturating_sub(used).max(1); + } + let mut free = (max - loc.volumes_len() as i64) * dsc; + free -= loc.ec_shard_count() as i64; + if free < 0 { + return 0; + } + free +} + // ============================================================================ // Tests // ============================================================================ @@ -928,6 +1036,7 @@ fn save_vif_volume_info(path: &str, info: &VifVolumeInfo) -> Result<(), VolumeEr mod tests { use super::*; use crate::storage::needle::needle::Needle; + use crate::storage::volume::volume_file_name; use tempfile::TempDir; fn make_test_store(dirs: &[&str]) -> Store { @@ -1301,4 +1410,151 @@ mod tests { let err = store.read_volume_needle(VolumeId(99), &mut n); assert!(matches!(err, Err(VolumeError::NotFound))); } + + /// Build a Store with N HDD disk locations under a single TempDir. + /// Returns the store and the TempDir guard so callers keep the dirs + /// alive for the test's lifetime. + fn make_ec_target_test_store(numdirs: usize) -> (Store, TempDir) { + let tmp = TempDir::new().unwrap(); + let mut store = Store::new(NeedleMapKind::InMemory); + for i in 0..numdirs { + let path = tmp.path().join(format!("data{}", i)); + std::fs::create_dir_all(&path).unwrap(); + store + .add_location( + path.to_str().unwrap(), + path.to_str().unwrap(), + 100, + DiskType::HardDrive, + MinFreeSpace::Percent(0.0), + Vec::new(), + ) + .unwrap(); + } + (store, tmp) + } + + /// Reproduces the placement half of seaweedfs/seaweedfs#9212. After + /// `ec.rebuild`'s first VolumeEcShardsCopy lands `.ecx` on disk N, + /// subsequent shards arrive with `CopyEcxFile=false` and rely on + /// auto-select. Without an on-disk check, `has_ec_volume` returns + /// false (no mount yet) and the fallback picks "any HDD with free + /// space" — splitting shards from their index files across disks. + /// `find_ec_shard_target_location` must pin to the `.ecx`-owning + /// disk via the on-disk check. + #[test] + fn test_find_ec_shard_target_location_pins_to_ecx_on_disk() { + let (store, _tmp) = make_ec_target_test_store(3); + let collection = "grafana-loki"; + let vid = VolumeId(1093); + + // Drop a sealed .ecx onto disk 2. Nothing is mounted yet — this + // is the state right after the first VolumeEcShardsCopy with + // CopyEcxFile=true and before any VolumeEcShardsMount has run. + let base = volume_file_name(&store.locations[2].idx_directory, collection, vid); + std::fs::write(format!("{}.ecx", base), vec![0u8; 20]).unwrap(); + + let got = store.find_ec_shard_target_location(collection, vid, 10); + assert_eq!( + got, + Some(2), + "placement leaked off the .ecx-owning disk; got {:?}", + got, + ); + } + + /// An already-mounted EC volume on disk 1 must win over a stray + /// `.ecx` on disk 2. Protects the post-startup steady state from + /// being perturbed by leftover index files from a prior failed move. + #[test] + fn test_find_ec_shard_target_location_prefers_mounted_over_ecx() { + let (mut store, _tmp) = make_ec_target_test_store(3); + let collection = "grafana-loki"; + let vid = VolumeId(2222); + + // Mount an EC shard on disk 1 so has_ec_volume returns true. + std::fs::write( + format!("{}/{}_{}.ec00", store.locations[1].directory, collection, vid.0), + b"shard data", + ) + .unwrap(); + store.locations[1] + .mount_ec_shards(vid, collection, &[0]) + .unwrap(); + + // Stray .ecx on disk 2 must not win. + let base = volume_file_name(&store.locations[2].idx_directory, collection, vid); + std::fs::write(format!("{}.ecx", base), vec![0u8; 20]).unwrap(); + + let got = store.find_ec_shard_target_location(collection, vid, 10); + assert_eq!(got, Some(1), "expected the mounted disk to win; got {:?}", got); + } + + /// Cold-volume case: no mount, no `.ecx` anywhere on this server. + /// Selection should still fall through to an HDD fallback. + #[test] + fn test_find_ec_shard_target_location_falls_through_to_hdd_when_nothing_matches() { + let (store, _tmp) = make_ec_target_test_store(2); + let got = store.find_ec_shard_target_location("grafana-loki", VolumeId(3333), 10); + assert!(got.is_some(), "expected an HDD fallback"); + assert_eq!(store.locations[got.unwrap()].disk_type, DiskType::HardDrive); + } + + /// `MaxVolumeCount=0` is the "unlimited disk" sentinel. The previous + /// formula returned a negative free count for unlimited disks, + /// making placement skip them entirely. The unlimited branch in + /// `ec_free_shard_count` must report a synthetic large free count. + #[test] + fn test_find_ec_shard_target_location_honours_unlimited_disk() { + let (store, _tmp) = make_ec_target_test_store(1); + store.locations[0] + .max_volume_count + .store(0, Ordering::Relaxed); + + let got = store.find_ec_shard_target_location("grafana-loki", VolumeId(4444), 10); + assert_eq!( + got, + Some(0), + "expected the only (unlimited) disk to be picked", + ); + } + + /// Truncation hazard: with `MaxVolumeCount=1, EcShardCount=1, + /// data_shard_count=10`, the old formula `(1*10 - 1) / 10 = 0` + /// would have rounded the disk to "full" and routed subsequent + /// shards elsewhere — the orphan-shard layout this PR exists to + /// prevent. Accounting in shard slots fixes it. + #[test] + fn test_find_ec_shard_target_location_tight_provisioning_keeps_ecx_disk() { + let (mut store, _tmp) = make_ec_target_test_store(2); + store.locations[0] + .max_volume_count + .store(1, Ordering::Relaxed); + store.locations[1] + .max_volume_count + .store(1, Ordering::Relaxed); + + let collection = "grafana-loki"; + let vid = VolumeId(5555); + + // Seed disk 1 with one EC shard so it owns .ecx and has 9 + // free shard slots remaining; the old formula would have + // rounded that to 0. + std::fs::write( + format!("{}/{}_{}.ec00", store.locations[1].directory, collection, vid.0), + b"shard data", + ) + .unwrap(); + store.locations[1] + .mount_ec_shards(vid, collection, &[0]) + .unwrap(); + + let got = store.find_ec_shard_target_location(collection, vid, 10); + assert_eq!( + got, + Some(1), + "expected the .ecx-owning disk (1 shard placed, 9 free shard slots) to be picked; got {:?}", + got, + ); + } }