diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 202e4b7f46..c8d2783ad8 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -3876,36 +3876,13 @@ impl VolumeServer for VolumeGrpcService { } } - // Read the .ecx index to check for live entries - let ecx_path = ec_vol.ecx_file_name(); - let ecx_data = - std::fs::read(&ecx_path).map_err(|e| Status::internal(format!("read ecx: {}", e)))?; - let entry_count = ecx_data.len() / NEEDLE_MAP_ENTRY_SIZE; - - let mut has_live = false; - for i in 0..entry_count { - let start = i * NEEDLE_MAP_ENTRY_SIZE; - let (_, _, size) = - idx_entry_from_bytes(&ecx_data[start..start + NEEDLE_MAP_ENTRY_SIZE]); - if !size.is_deleted() { - has_live = true; - break; - } - } - - if !has_live { - return Err(Status::failed_precondition(format!( - "ec volume {} has no live entries", - req.volume_id - ))); - } - // Reconstruct the volume from EC shards. Use the EcVolume's // own dir for the produced .dat (matches the volume's home // disk) and its `ecx_actual_dir` for the .ecx lookup, while // reading each shard from its real on-disk location. let dat_dir = ec_vol.dir.clone(); let ecx_dir = ec_vol.ecx_actual_dir().to_string(); + let idx_dir = ec_vol.dir_idx.clone(); let collection = ec_vol.collection.clone(); let vif_dat_file_size = ec_vol.dat_file_size; let (large_block_size, small_block_size) = @@ -3918,6 +3895,28 @@ impl VolumeServer for VolumeGrpcService { .collect(); drop(store); + // Deletions journaled beside the .ecx, or collected by + // VolumeEcShardsCopy into the idx dir, count as deleted throughout. + let deleted = crate::storage::erasure_coding::ec_decoder::read_ecj_deletions( + &[&ecx_dir, &idx_dir], + &collection, + vid, + ) + .map_err(|e| Status::internal(format!("read ecj: {}", e)))?; + let has_live = crate::storage::erasure_coding::ec_decoder::has_live_needles( + &ecx_dir, + &collection, + vid, + &deleted, + ) + .map_err(|e| Status::internal(format!("HasLiveNeedles: {}", e)))?; + if !has_live { + return Err(Status::failed_precondition(format!( + "ec volume {} has no live entries", + req.volume_id + ))); + } + // Calculate .dat file size from .ecx entries (.ec00 lives on // its own disk, .ecx on the index disk). let dat_file_size = @@ -3926,6 +3925,7 @@ impl VolumeServer for VolumeGrpcService { &ecx_dir, &collection, vid, + &deleted, ) .map_err(|e| Status::internal(format!("FindDatFileSize: {}", e)))?; @@ -3950,11 +3950,15 @@ impl VolumeServer for VolumeGrpcService { ) .map_err(|e| Status::internal(format!("WriteDatFile: {}", e)))?; - // Write .idx file from .ecx and .ecj files (lives on idx dir). - crate::storage::erasure_coding::ec_decoder::write_idx_file_from_ec_index( + // Write .idx from the .ecx wherever it lives, beside the .dat where + // the mount looks first (Go moves it there after the rebuild). + crate::storage::erasure_coding::ec_decoder::write_idx_file_from_ec_index_with_dirs( + &ecx_dir, &dat_dir, &collection, vid, + &deleted, + dat_file_size, ) .map_err(|e| Status::internal(format!("WriteIdxFileFromEcIndex: {}", e)))?; @@ -11082,4 +11086,268 @@ mod tests { resp.error ); } + + /// A split -dir/-dir.idx decode target with the shards in the data dir. + /// The .ecx is moved to the idx dir, as VolumeEcShardsCopy leaves it, unless + /// `ecx_in_data_dir` (where VolumeEcShardsGenerate writes it). `journal` goes + /// into the idx-dir .ecj, where VolumeEcShardsCopy appends collected ones. + async fn make_split_idx_ec_decode_service( + journal: &[u64], + ecx_in_data_dir: bool, + ) -> (VolumeGrpcService, TempDir, String, String, u64) { + let tmp = TempDir::new().unwrap(); + let data = tmp.path().join("data").to_str().unwrap().to_string(); + let idx = tmp.path().join("idx").to_str().unwrap().to_string(); + std::fs::create_dir_all(&data).unwrap(); + std::fs::create_dir_all(&idx).unwrap(); + + let mut store = Store::new(NeedleMapKind::InMemory); + store + .add_location( + &data, + &idx, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(0.0), + Vec::new(), + ) + .unwrap(); + store + .add_volume(VolumeId(1), DiskType::HardDrive, &VolumeSpec::default()) + .unwrap(); + let mut size_before_needle_3 = 0; + { + let (_, volume) = store.find_volume_mut(VolumeId(1)).unwrap(); + for id in 1..=3u64 { + if id == 3 { + size_before_needle_3 = volume.dat_file_size().unwrap(); + } + let body = format!("split idx decode needle {id}").into_bytes(); + let mut needle = Needle { + id: NeedleId(id), + cookie: Cookie(id as u32), + data_size: body.len() as u32, + data: body, + ..Needle::default() + }; + volume.write_needle(&mut needle, true, false).unwrap(); + } + volume.sync_to_disk().unwrap(); + } + let service = split_disk_grpc_service(store); + service + .volume_ec_shards_generate(Request::new( + volume_server_pb::VolumeEcShardsGenerateRequest { + volume_id: 1, + collection: String::new(), + }, + )) + .await + .unwrap(); + + let mut store = service.state.store.write().unwrap(); + store.unmount_volume(VolumeId(1)).unwrap(); + std::fs::remove_file(format!("{data}/1.dat")).unwrap(); + std::fs::remove_file(format!("{idx}/1.idx")).unwrap(); + let ecx_dir = if ecx_in_data_dir { + data.clone() + } else { + std::fs::rename(format!("{data}/1.ecx"), format!("{idx}/1.ecx")).unwrap(); + idx.clone() + }; + let mut ecj = Vec::new(); + for id in journal { + let mut buf = [0u8; NEEDLE_ID_SIZE]; + NeedleId(*id).to_bytes(&mut buf); + ecj.extend_from_slice(&buf); + } + std::fs::write(format!("{idx}/1.ecj"), ecj).unwrap(); + for shard_id in 0..14 { + store.mount_ec_shard(VolumeId(1), "", shard_id, "").unwrap(); + } + assert_eq!( + store.find_ec_volume(VolumeId(1)).unwrap().ecx_actual_dir(), + ecx_dir, + "precondition: the mounted EC volume reads its .ecx from {ecx_dir}" + ); + drop(store); + (service, tmp, data, idx, size_before_needle_3) + } + + fn ec_shards_to_volume_request() -> Request { + let mut request = Request::new(volume_server_pb::VolumeEcShardsToVolumeRequest { + volume_id: 1, + collection: String::new(), + ..Default::default() + }); + request + .extensions_mut() + .insert(tonic::transport::server::TcpConnectInfo { + local_addr: None, + remote_addr: Some("127.0.0.1:65000".parse().unwrap()), + }); + request + } + + /// Mounts the decoded volume and checks its (file, deleted) counts, that it + /// is writable, that `live` needles read back and `deleted` ones stay deleted. + fn assert_decoded_volume( + service: &VolumeGrpcService, + counts: (i64, i64), + live: &[u64], + deleted: &[u64], + ) { + let mut store = service.state.store.write().unwrap(); + let all_shards: Vec = (0..14).collect(); + store.unmount_ec_shards(VolumeId(1), &all_shards); + store + .mount_volume(VolumeId(1), "", DiskType::HardDrive) + .unwrap(); + let (_, v) = store.find_volume(VolumeId(1)).unwrap(); + assert_eq!((v.file_count(), v.deleted_count()), counts); + assert!( + !v.is_no_write_or_delete(), + "no .idx row may point past the end of the decoded .dat" + ); + for &id in live { + let mut n = Needle { + id: NeedleId(id), + ..Needle::default() + }; + store.read_volume_needle(VolumeId(1), &mut n).unwrap(); + assert_eq!(n.data, format!("split idx decode needle {id}").into_bytes()); + } + for &id in deleted { + let mut n = Needle { + id: NeedleId(id), + ..Needle::default() + }; + assert!( + store.read_volume_needle(VolumeId(1), &mut n).is_err(), + "needle {id} was deleted through the .ecj and must stay deleted" + ); + } + let body = b"written after decode".to_vec(); + let mut n = Needle { + id: NeedleId(4), + cookie: Cookie(4), + data_size: body.len() as u32, + data: body.clone(), + ..Needle::default() + }; + store + .write_volume_needle(VolumeId(1), &mut n, true) + .unwrap(); + let mut n = Needle { + id: NeedleId(4), + ..Needle::default() + }; + store.read_volume_needle(VolumeId(1), &mut n).unwrap(); + assert_eq!(n.data, body); + } + + /// ec.decode used to read the .ecx/.ecj from the data dir when writing the + /// .idx, failing with NotFound after the .dat was already published, and + /// ignored .ecj deletions when sizing the .dat (Go folds them in first). + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_volume_ec_shards_to_volume_reads_ecx_from_split_idx_dir() { + // The lowest key (the first .idx row) and the .dat tail, each journaled + // many times over. + let journal: Vec = std::iter::repeat_n([1, 3], 500).flatten().collect(); + let (service, _tmp, data, idx, size_before_needle_3) = + make_split_idx_ec_decode_service(&journal, false).await; + let ecx = std::fs::read(format!("{idx}/1.ecx")).unwrap(); + + service + .volume_ec_shards_to_volume(ec_shards_to_volume_request()) + .await + .unwrap(); + + // The journaled tail needle is not decoded. + assert_eq!( + std::fs::metadata(format!("{data}/1.dat")).unwrap().len(), + size_before_needle_3 + ); + // The .idx sits beside the .dat. It keeps the .ecx rows the .dat holds + // and tombstones each journaled one once, however often it was journaled. + let mut expected_idx = ecx[..2 * NEEDLE_MAP_ENTRY_SIZE].to_vec(); + crate::storage::idx::write_index_entry( + &mut expected_idx, + NeedleId(1), + Offset::default(), + TOMBSTONE_FILE_SIZE, + ) + .unwrap(); + assert_eq!( + std::fs::read(format!("{data}/1.idx")).unwrap(), + expected_idx + ); + assert!(!std::path::Path::new(&format!("{idx}/1.idx")).exists()); + + // Needle 1 is still in the .dat, so it counts as garbage; needle 3 is not. + assert_decoded_volume(&service, (2, 1), &[2], &[1, 3]); + } + + /// Collected journals land in the idx dir even when the .ecx is beside the + /// shards, so the decode reads deletions from both. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_volume_ec_shards_to_volume_reads_idx_dir_ecj_with_data_dir_ecx() { + let (service, _tmp, data, _idx, size_before_needle_3) = + make_split_idx_ec_decode_service(&[3], true).await; + + service + .volume_ec_shards_to_volume(ec_shards_to_volume_request()) + .await + .unwrap(); + + assert_eq!( + std::fs::metadata(format!("{data}/1.dat")).unwrap().len(), + size_before_needle_3 + ); + assert_decoded_volume(&service, (2, 0), &[1, 2], &[3]); + } + + /// A tail needle tombstoned in the .ecx itself (Go's RebuildEcxFile) is cut + /// from the .dat the same way, so its row must not survive either. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_volume_ec_shards_to_volume_drops_sealed_tail_tombstone() { + let (service, _tmp, data, idx, size_before_needle_3) = + make_split_idx_ec_decode_service(&[], false).await; + let ecx_path = format!("{idx}/1.ecx"); + let mut ecx = std::fs::read(&ecx_path).unwrap(); + let size_at = 2 * NEEDLE_MAP_ENTRY_SIZE + NEEDLE_ID_SIZE + OFFSET_SIZE; + TOMBSTONE_FILE_SIZE.to_bytes(&mut ecx[size_at..size_at + SIZE_SIZE]); + std::fs::write(&ecx_path, &ecx).unwrap(); + + service + .volume_ec_shards_to_volume(ec_shards_to_volume_request()) + .await + .unwrap(); + + assert_eq!( + std::fs::metadata(format!("{data}/1.dat")).unwrap().len(), + size_before_needle_3 + ); + assert_eq!( + std::fs::read(format!("{data}/1.idx")).unwrap(), + ecx[..2 * NEEDLE_MAP_ENTRY_SIZE] + ); + assert_decoded_volume(&service, (2, 0), &[1, 2], &[3]); + } + + /// Deletions only in the .ecj count toward "no live entries", as after Go's + /// RebuildEcxFile, so the caller purges the shards instead of decoding. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_volume_ec_shards_to_volume_all_journaled_has_no_live_entries() { + let (service, _tmp, data, _idx, _) = + make_split_idx_ec_decode_service(&[1, 2, 3], false).await; + + let err = service + .volume_ec_shards_to_volume(ec_shards_to_volume_request()) + .await + .unwrap_err(); + assert_eq!(err.code(), tonic::Code::FailedPrecondition, "{err:?}"); + assert!(err.message().contains("has no live entries"), "{err:?}"); + assert!(!std::path::Path::new(&format!("{data}/1.dat")).exists()); + } } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs index bcc8afd8b7..94126ebb6b 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_decoder.rs @@ -3,10 +3,12 @@ //! Rebuilds the original .dat + .idx files from data shards (.ec00-.ec09) //! and the sorted index (.ecx) + deletion journal (.ecj). +use std::collections::HashSet; use std::fs::File; use std::io::{self, Read, Write}; use crate::storage::erasure_coding::ec_shard::*; +use crate::storage::erasure_coding::ec_volume::read_ecj_ids; use crate::storage::idx; use crate::storage::needle::needle::get_actual_size; use crate::storage::super_block::SUPER_BLOCK_SIZE; @@ -20,18 +22,21 @@ use crate::storage::volume::{fsync_dir, volume_file_name}; /// `dir` is used both for reading `.ec00` and `.ecx`. For split-disk /// reconciled volumes call [`find_dat_file_size_with_dirs`] instead. pub fn find_dat_file_size(dir: &str, collection: &str, volume_id: VolumeId) -> io::Result { - find_dat_file_size_with_dirs(dir, dir, collection, volume_id) + let deleted = read_ecj_deletions(&[dir], collection, volume_id)?; + find_dat_file_size_with_dirs(dir, dir, collection, volume_id, &deleted) } /// Like [`find_dat_file_size`] but lets the caller pass separate dirs /// for `.ec00` (the data shard) and `.ecx` (the sealed index). This /// is the form needed when shards are split across data dirs and the -/// `.ecx` lives on a sibling disk's idx dir (#9252). +/// `.ecx` lives on a sibling disk's idx dir (#9252). Needles in `deleted` +/// count as deleted. pub fn find_dat_file_size_with_dirs( ec00_dir: &str, ecx_dir: &str, collection: &str, volume_id: VolumeId, + deleted: &HashSet, ) -> io::Result { let ec00_base = volume_file_name(ec00_dir, collection, volume_id); let ecx_base = volume_file_name(ecx_dir, collection, volume_id); @@ -53,9 +58,9 @@ pub fn find_dat_file_size_with_dirs( for i in 0..entry_count { let start = i * NEEDLE_MAP_ENTRY_SIZE; - let (_, offset, size) = + let (key, offset, size) = idx_entry_from_bytes(&ecx_data[start..start + NEEDLE_MAP_ENTRY_SIZE]); - if size.is_deleted() { + if size.is_deleted() || deleted.contains(&key) { continue; } let entry_stop = offset.to_actual_offset() + get_actual_size(size, version); @@ -67,6 +72,48 @@ pub fn find_dat_file_size_with_dirs( Ok(dat_size) } +/// Whether the `.ecx` in `ecx_dir` indexes a needle deleted neither there nor +/// in `deleted`. +pub fn has_live_needles( + ecx_dir: &str, + collection: &str, + volume_id: VolumeId, + deleted: &HashSet, +) -> io::Result { + let ecx_base = volume_file_name(ecx_dir, collection, volume_id); + let ecx_data = std::fs::read(format!("{}.ecx", ecx_base))?; + let (entries, _) = ecx_data.as_chunks::(); + Ok(entries.iter().any(|entry| { + let (key, _, size) = idx_entry_from_bytes(entry); + !size.is_deleted() && !deleted.contains(&key) + })) +} + +/// Distinct needle ids journaled in the `.ecj` of any of `dirs`. Go folds the +/// journal into the `.ecx` (RebuildEcxFile) before a decode; reading it leaves +/// the sealed index untouched. Only NotFound means "no journal". +pub fn read_ecj_deletions( + dirs: &[&str], + collection: &str, + volume_id: VolumeId, +) -> io::Result> { + let mut ids = HashSet::new(); + for (i, dir) in dirs.iter().enumerate() { + if dirs[..i].contains(dir) { + continue; + } + let path = format!("{}.ecj", volume_file_name(dir, collection, volume_id)); + let file = match File::open(&path) { + Ok(file) => file, + Err(e) if e.kind() == io::ErrorKind::NotFound => continue, + Err(e) => return Err(e), + }; + let len = file.metadata()?.len(); + read_ecj_ids(&file, len, &mut ids)?; + } + Ok(ids) +} + /// What it takes to rebuild a volume's .dat from its EC data shards. /// /// Mirrors Go's `WriteDatFile(baseFileName, datFileSize, @@ -267,51 +314,61 @@ pub fn write_dat_file_from_shards(spec: &DatRebuild<'_>) -> io::Result<()> { /// Write .idx file from .ecx index + .ecj deletion journal. /// -/// Copies sorted .ecx entries to .idx, then appends tombstones for -/// deleted needles from .ecj. +/// See [`write_idx_file_from_ec_index_with_dirs`]; everything lives in `dir`. pub fn write_idx_file_from_ec_index( dir: &str, collection: &str, volume_id: VolumeId, ) -> io::Result<()> { - let base = volume_file_name(dir, collection, volume_id); - let ecx_path = format!("{}.ecx", base); - let ecj_path = format!("{}.ecj", base); - let idx_path = format!("{}.idx", base); + let deleted = read_ecj_deletions(&[dir], collection, volume_id)?; + let dat_file_size = find_dat_file_size_with_dirs(dir, dir, collection, volume_id, &deleted)?; + write_idx_file_from_ec_index_with_dirs(dir, dir, collection, volume_id, &deleted, dat_file_size) +} + +/// Write the `.idx` for a `.dat` decoded to `dat_file_size` bytes, from the +/// `.ecx` in `ecx_dir`, into `idx_dir`. +/// +/// Copies the `.ecx` rows, then appends one tombstone per row whose needle is +/// in `deleted`. A deleted needle at or past `dat_file_size` was cut from the +/// `.dat`, so its row is dropped: a row pointing past the end of the `.dat` +/// makes the volume load read-only. +pub fn write_idx_file_from_ec_index_with_dirs( + ecx_dir: &str, + idx_dir: &str, + collection: &str, + volume_id: VolumeId, + deleted: &HashSet, + dat_file_size: i64, +) -> io::Result<()> { + let ecx_path = format!("{}.ecx", volume_file_name(ecx_dir, collection, volume_id)); + let idx_path = format!("{}.idx", volume_file_name(idx_dir, collection, volume_id)); // Write to a temp file and atomically rename into place, so a crash // mid-write never leaves a partial .idx at the final name beside the // source shards. let tmp_path = format!("{}.tmp", idx_path); let write_result = (|| -> io::Result<()> { - // Copy .ecx to the temp .idx - std::fs::copy(&ecx_path, &tmp_path)?; - - // Append deletions from .ecj as tombstones. Read the journal directly - // and treat only NotFound as "no journal": Path::exists would also - // swallow a permission/IO error and silently skip deletions, which - // would resurrect deleted needles as live. - let mut idx_file = std::fs::OpenOptions::new().append(true).open(&tmp_path)?; - match std::fs::read(&ecj_path) { - Ok(ecj_data) => { - let count = ecj_data.len() / NEEDLE_ID_SIZE; - for i in 0..count { - let start = i * NEEDLE_ID_SIZE; - let needle_id = NeedleId::from_bytes(&ecj_data[start..start + NEEDLE_ID_SIZE]); - idx::write_index_entry( - &mut idx_file, - needle_id, - Offset::default(), - TOMBSTONE_FILE_SIZE, - )?; - } + let mut ecx_file = File::open(&ecx_path)?; + let mut idx_file = io::BufWriter::new(File::create(&tmp_path)?); + let mut tombstoned = Vec::new(); + idx::walk_index_file(&mut ecx_file, 0, |key, offset, size| { + let is_deleted = size.is_deleted() || deleted.contains(&key); + if is_deleted && offset.to_actual_offset() >= dat_file_size { + return Ok(()); } - Err(e) if e.kind() == io::ErrorKind::NotFound => {} - Err(e) => return Err(e), + idx::write_index_entry(&mut idx_file, key, offset, size)?; + if !size.is_deleted() && deleted.contains(&key) { + tombstoned.push(key); + } + Ok(()) + })?; + for key in tombstoned { + idx::write_index_entry(&mut idx_file, key, Offset::default(), TOMBSTONE_FILE_SIZE)?; } // fsync, rename, then fsync the dir so the decoded .idx is durable and // atomically published before the caller deletes the source shards. + let idx_file = idx_file.into_inner().map_err(|e| e.into_error())?; idx_file.sync_all()?; drop(idx_file); // Windows rename does not replace an existing file on every version; @@ -694,4 +751,43 @@ mod tests { .is_err() ); } + + /// A journal many chunks long that repeats a few ids reads back as those + /// ids, from each dir once, whatever its length. + #[test] + fn test_read_ecj_deletions_collects_distinct_ids_across_dirs() { + let tmp = TempDir::new().unwrap(); + let data = tmp.path().join("data"); + let idx = tmp.path().join("idx"); + let missing = tmp.path().join("missing"); + std::fs::create_dir_all(&data).unwrap(); + std::fs::create_dir_all(&idx).unwrap(); + let (data, idx, missing) = ( + data.to_str().unwrap(), + idx.to_str().unwrap(), + missing.to_str().unwrap(), + ); + + let entry = |id: u64| { + let mut buf = [0u8; NEEDLE_ID_SIZE]; + NeedleId(id).to_bytes(&mut buf); + buf + }; + // Past two load chunks of three repeating ids, then an id only in the + // last chunk and a torn trailing record. + let mut ecj = Vec::new(); + while ecj.len() <= 2 * (1 << 20) { + for id in [1, 2, 3] { + ecj.extend_from_slice(&entry(id)); + } + } + ecj.extend_from_slice(&entry(7)); + ecj.extend_from_slice(&entry(8)[..3]); + std::fs::write(format!("{idx}/1.ecj"), &ecj).unwrap(); + std::fs::write(format!("{data}/1.ecj"), entry(9)).unwrap(); + + let ids = read_ecj_deletions(&[data, idx, idx, missing], "", VolumeId(1)).unwrap(); + let expected: HashSet = [1, 2, 3, 7, 9].into_iter().map(NeedleId).collect(); + assert_eq!(ids, expected); + } } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 457ad996e0..eae9e7af41 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -48,6 +48,29 @@ pub(crate) struct ShardLocationCache { /// A multiple of `NEEDLE_ID_SIZE`; 1 MiB is 131072 entries per syscall. const ECJ_LOAD_CHUNK_BYTES: usize = 1 << 20; +/// Adds every whole needle id in the first `len` bytes of `ecj_file` to `ids`, +/// reading `ECJ_LOAD_CHUNK_BYTES` at a time; a trailing partial record is +/// ignored. +pub(crate) fn read_ecj_ids( + ecj_file: &File, + len: u64, + ids: &mut HashSet, +) -> io::Result<()> { + let mut buf = vec![0u8; std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, len) as usize]; + let mut off: u64 = 0; + while off + NEEDLE_ID_SIZE as u64 <= len { + let mut want = std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, len - off) as usize; + want -= want % NEEDLE_ID_SIZE; + // Positional read: the loader's handle is shared with journal appends. + read_exact_at(ecj_file, &mut buf[..want], off)?; + for entry in buf[..want].as_chunks::().0 { + ids.insert(NeedleId::from_bytes(entry)); + } + off += want as u64; + } + Ok(()) +} + /// An erasure-coded volume managing its local shards and index. pub struct EcVolume { pub volume_id: VolumeId, @@ -810,38 +833,7 @@ impl EcVolume { // held the `deleted_needles` write lock for the whole scan, which on a // bloated journal is the entire (unbounded) startup. let mut loaded: HashSet = HashSet::new(); - let mut buf = vec![0u8; ECJ_LOAD_CHUNK_BYTES]; - let end = self.ecj_file_size as u64; - let mut off: u64 = 0; - while off + NEEDLE_ID_SIZE as u64 <= end { - // Whole entries only; a trailing partial record is ignored, as the - // per-entry loop did by construction. - let mut want = std::cmp::min(ECJ_LOAD_CHUNK_BYTES as u64, end - off) as usize; - want -= want % NEEDLE_ID_SIZE; - if want == 0 { - break; - } - #[cfg(unix)] - { - use std::os::unix::fs::FileExt; - ecj_file.read_exact_at(&mut buf[..want], off)?; - } - #[cfg(windows)] - { - // Positional read so concurrent readers of the shared .ecj - // handle can't interleave seek/read. Mirrors the - // read_exact_at helper at the bottom of this file. - read_exact_at(ecj_file, &mut buf[..want], off)?; - } - #[cfg(not(any(unix, windows)))] - { - compile_error!("Platform not supported: only unix and windows are supported"); - } - for entry in buf[..want].chunks_exact(NEEDLE_ID_SIZE) { - loaded.insert(NeedleId::from_bytes(entry)); - } - off += want as u64; - } + read_ecj_ids(ecj_file, self.ecj_file_size as u64, &mut loaded)?; let mut set = self .deleted_needles