diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 7464fc955..1f10d7eb6 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -7,8 +7,8 @@ use std::collections::{HashMap, HashSet}; use std::fs; use std::io; -use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering}; -use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; use tracing::warn; @@ -122,6 +122,10 @@ impl DiskLocation { // Scan for .dat files let entries = fs::read_dir(&self.directory)?; let mut dat_files: Vec<(String, VolumeId)> = Vec::new(); + // Every collection claiming an id, in scan order; open_volumes keeps + // the first that opens. + let mut to_load: Vec<(VolumeId, Vec)> = Vec::new(); + let mut queued: HashMap = HashMap::new(); let mut seen = HashSet::new(); for entry in entries { @@ -201,6 +205,7 @@ impl DiskLocation { 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 // disk) would otherwise have Volume::new write an 8-byte stub that @@ -221,30 +226,22 @@ impl DiskLocation { continue; } - 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(); - crate::metrics::VOLUME_GAUGE - .with_label_values(&[&collection, "volume"]) - .inc(); - self.volumes.insert(vid, v); - } - Err(e) => { - warn!(volume_id = vid.0, error = %e, "failed to load volume"); + 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) { + crate::metrics::VOLUME_GAUGE + .with_label_values(&[&collection, "volume"]) + .inc(); + self.volumes.insert(vid, v); + } + // After regular volumes, auto-discover EC shards on disk so a // fresh restart picks up shards without an explicit // VolumeEcShardsMount RPC. Mirrors Go's loadExistingVolumes @@ -257,6 +254,65 @@ impl DiskLocation { Ok(()) } + /// Open the volumes the directory scan selected. Opening one is dominated + /// by reading its .idx into the needle map, so a disk holding thousands + /// takes thousands of serial index reads to come up; mirrors Go's + /// concurrentLoadingVolumes down to the max(cores, 10) worker count, whose + /// floor keeps a small-core box off one-at-a-time on IO-bound work. + /// + /// 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<(VolumeId, Vec)>, + needle_map_kind: NeedleMapKind, + ) -> Vec<(String, VolumeId, Volume)> { + if to_load.is_empty() { + return Vec::new(); + } + let workers = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1) + .max(10) + .min(to_load.len()); + + let next = AtomicUsize::new(0); + let opened = Mutex::new(Vec::with_capacity(to_load.len())); + std::thread::scope(|scope| { + for _ in 0..workers { + scope.spawn(|| loop { + let i = next.fetch_add(1, Ordering::Relaxed); + let Some((vid, collections)) = to_load.get(i) else { + return; + }; + 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"); + } + } + } + }); + } + }); + opened.into_inner().unwrap_or_else(|e| e.into_inner()) + } + /// Directory pre-pass that recovers interrupted compaction commits. Collects /// every volume id that still has a .cpc commit marker or a leftover /// .cpd/.cpx temp file across the data and idx directories, then rolls the @@ -1493,6 +1549,60 @@ mod tests { assert!(ids.contains(&VolumeId(2))); } + // Two collections can name the same volume id on one disk; a candidate + // that fails to open 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, unopenable. + 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(); diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 01896350e..67a03eef7 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -1970,11 +1970,13 @@ impl Volume { } /// Verify the live needle at the .dat tail: its header must match the - /// .idx entry and the .dat must end exactly at its on-disk end. Extra - /// bytes mean an unindexed trailing record (a torn append); appending - /// after one would place the next needle at an offset the 8-byte .idx - /// encoding may not represent, so the volume is quarantined read-only - /// instead. Mirrors Go's verifyNeedleIntegrity. + /// .idx entry and, on v3, the .dat must end exactly at its on-disk end. + /// Extra bytes mean an unindexed trailing record (a torn append); + /// appending after one would place the next needle at an offset the + /// 8-byte .idx encoding may not represent, so the volume is quarantined + /// read-only instead. Mirrors Go's verifyNeedleIntegrity, whose tail + /// check is v3-only -- a v1/v2 volume Go serves read-write must not go + /// read-only here. fn verify_needle_integrity( &mut self, actual_offset: i64, @@ -2013,15 +2015,16 @@ impl Volume { ))); } - if version == VERSION_3 { - let ts_offset = - checked_offset as u64 + NEEDLE_HEADER_SIZE as u64 + size.0 as u64 + 4; // skip checksum - let mut ts_buf = [0u8; 8]; - self.read_exact_at_backend(&mut ts_buf, ts_offset)?; - let ts = u64::from_be_bytes(ts_buf); - if ts > 0 { - self.last_append_at_ns = ts; - } + if version != VERSION_3 { + return Ok(()); + } + + let ts_offset = checked_offset as u64 + NEEDLE_HEADER_SIZE as u64 + size.0 as u64 + 4; // skip checksum + let mut ts_buf = [0u8; 8]; + self.read_exact_at_backend(&mut ts_buf, ts_offset)?; + let ts = u64::from_be_bytes(ts_buf); + if ts > 0 { + self.last_append_at_ns = ts; } self.verify_dat_ends_at(checked_offset + get_actual_size(size, version)) @@ -4115,6 +4118,54 @@ mod tests { ); } + // Go's tail check is v3-only, so it serves a v1/v2 volume with an + // unindexed tail read-write. Checking it here flipped whole legacy disks + // read-only on their first boot under this server. + #[test] + fn test_integrity_skips_dat_tail_check_before_v3() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + { + let mut v = Volume::new( + dir, + dir, + "", + VolumeId(1), + NeedleMapKind::InMemory, + None, + None, + 0, + VERSION_2, + ) + .unwrap(); + for i in 1..=3u64 { + let data = format!("data {}", i); + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data: data.as_bytes().to_vec(), + data_size: data.len() as u32, + ..Needle::default() + }; + v.write_needle(&mut n, true).unwrap(); + } + v.sync_to_disk().unwrap(); + } + + let dat_path = volume_file_name(dir, "", VolumeId(1)) + ".dat"; + let mut f = OpenOptions::new().append(true).open(&dat_path).unwrap(); + f.write_all(&[0xFFu8; 96]).unwrap(); + f.sync_all().unwrap(); + drop(f); + + let v = reload_volume(dir); + assert_eq!(v.version(), VERSION_2, "volume should reload as v2"); + assert!( + !v.is_no_write_or_delete(), + "v2 volume with unindexed .dat tail must stay writable, as in Go" + ); + } + // Same torn tail, but with a deletion tombstone as the last indexed // record — the tombstone path must also verify the .dat ends at it. #[test]