From 12c8df3405e5cd80afa76226c62eace64efa21f4 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 6 Aug 2026 15:22:36 -0700 Subject: [PATCH] rust: let a failed volume open fall back to the next candidate Two collections can name the same volume id on one disk. Deduping the load queue by id claimed the id for whichever candidate the scan saw first, so a corrupt one shadowed a good one behind it; the serial loader this replaced only claimed an id once a volume had actually opened. Carry every claiming collection per id and try them in scan order until one loads. --- seaweed-volume/src/storage/disk_location.rs | 121 +++++++++++++++----- 1 file changed, 94 insertions(+), 27 deletions(-) diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 35fc41402..0f7183e31 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -122,8 +122,11 @@ impl DiskLocation { // Scan for .dat files let entries = fs::read_dir(&self.directory)?; let mut dat_files: Vec<(String, VolumeId)> = Vec::new(); - let mut to_load: Vec<(String, VolumeId)> = Vec::new(); - let mut queued: HashSet = HashSet::new(); + // One entry per volume id, holding every collection name that claims it + // in scan order -- the loader tries them in turn and keeps the first + // that opens, so a corrupt candidate does not shadow a good one. + let mut to_load: Vec<(VolumeId, Vec)> = Vec::new(); + let mut queued: HashMap = HashMap::new(); let mut seen = HashSet::new(); for entry in entries { @@ -203,11 +206,6 @@ impl DiskLocation { continue; } - // Skip volumes already queued under a different collection name for - // the same id -- the loader below keys results by id. - if !queued.insert(vid) { - continue; - } // Load existing data only; never create a phantom `.dat`. A lone // `.vif`/`.idx` (e.g. an EC sidecar whose `.ecx` is on a sibling @@ -229,7 +227,13 @@ impl DiskLocation { continue; } - to_load.push((collection, vid)); + match queued.get(&vid) { + Some(&i) => to_load[i].1.push(collection), + None => { + queued.insert(vid, to_load.len()); + to_load.push((vid, vec![collection])); + } + } } for (collection, vid, v) in self.open_volumes(to_load, needle_map_kind) { @@ -257,9 +261,13 @@ impl DiskLocation { /// serial index reads to come up. Mirrors Go's concurrentLoadingVolumes, /// including its max(cores, 10) worker count -- the work is IO-bound, so /// the floor keeps a small-core box from loading one volume at a time. + /// + /// Each id carries every collection name that claims it, tried in scan + /// order until one opens: an id is only spoken for once a volume actually + /// loads, so a corrupt `colA_5.dat` still leaves `colB_5.dat` a chance. fn open_volumes( &self, - to_load: Vec<(String, VolumeId)>, + to_load: Vec<(VolumeId, Vec)>, needle_map_kind: NeedleMapKind, ) -> Vec<(String, VolumeId, Volume)> { if to_load.is_empty() { @@ -277,26 +285,29 @@ impl DiskLocation { for _ in 0..workers { scope.spawn(|| loop { let i = next.fetch_add(1, Ordering::Relaxed); - let Some((collection, vid)) = to_load.get(i) else { + let Some((vid, collections)) = to_load.get(i) else { return; }; - match Volume::new( - &self.directory, - &self.idx_directory, - collection, - *vid, - needle_map_kind, - None, // replica placement read from superblock - None, // TTL read from superblock - 0, // no preallocate on load - Version::current(), - ) { - Ok(mut v) => { - v.location_disk_space_low = self.is_disk_space_low.clone(); - opened.lock().unwrap().push((collection.clone(), *vid, v)); - } - Err(e) => { - warn!(volume_id = vid.0, error = %e, "failed to load volume"); + for collection in collections { + match Volume::new( + &self.directory, + &self.idx_directory, + collection, + *vid, + needle_map_kind, + None, // replica placement read from superblock + None, // TTL read from superblock + 0, // no preallocate on load + Version::current(), + ) { + Ok(mut v) => { + v.location_disk_space_low = self.is_disk_space_low.clone(); + opened.lock().unwrap().push((collection.clone(), *vid, v)); + break; + } + Err(e) => { + warn!(volume_id = vid.0, error = %e, "failed to load volume"); + } } } }); @@ -1541,6 +1552,62 @@ mod tests { assert!(ids.contains(&VolumeId(2))); } + // Two collections can name the same volume id on one disk. The id is only + // spoken for once a volume actually opens, so a candidate that fails to + // load must not shadow a good one behind it. + #[test] + fn test_open_volumes_falls_back_past_a_corrupt_candidate() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + + { + let mut loc = DiskLocation::new( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + loc.create_volume( + VolumeId(9), + "good", + NeedleMapKind::InMemory, + None, + None, + 0, + Version::current(), + ) + .unwrap(); + loc.close(); + } + + // Same id under another collection, with a superblock this build + // cannot read -- Volume::new fails on it. + let mut bad = vec![0u8; 16]; + bad[0] = 9; // unsupported version + std::fs::write(format!("{}/bad_9.dat", dir), &bad).unwrap(); + + let loc = DiskLocation::new( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + let opened = loc.open_volumes( + vec![(VolumeId(9), vec!["bad".to_string(), "good".to_string()])], + NeedleMapKind::InMemory, + ); + + assert_eq!(opened.len(), 1, "the good candidate should still open"); + assert_eq!(opened[0].0, "good"); + assert_eq!(opened[0].1, VolumeId(9)); + } + #[test] fn test_disk_location_delete_volume() { let tmp = TempDir::new().unwrap();