//! EC encoding: convert a .dat file into 10 data + 4 parity shards. //! //! Uses Reed-Solomon erasure coding. The .dat file is split into blocks //! (1GB large, 1MB small) and encoded across 14 shard files. use std::fs::File; use std::io; #[cfg(not(unix))] use std::io::{Read, Seek, SeekFrom}; use reed_solomon_erasure::galois_8::ReedSolomon; use crate::pb::volume_server_pb::{ChecksumAlgorithm, EcBitrotProtection, EcShardChecksums}; use crate::storage::erasure_coding::ec_bitrot::{ self, DEFAULT_BITROT_BLOCK_SIZE, ShardChecksumBuilder, }; use crate::storage::erasure_coding::ec_shard::*; use crate::storage::idx; use crate::storage::types::*; use crate::storage::volume::volume_file_name; /// Encode a .dat file into EC shard files. /// /// Creates .ec00-.ec13 files in the same directory. /// Also creates a sorted .ecx index from the .idx file. /// /// Always encodes with the uniform block layout, sized for this .dat, and /// returns the block size so the caller can persist it to .vif. Mirrors Go's /// WriteEcFiles. pub fn write_ec_files( dir: &str, idx_dir: &str, collection: &str, volume_id: VolumeId, data_shards: usize, parity_shards: usize, ) -> io::Result { let base = volume_file_name(dir, collection, volume_id); let dat_path = format!("{}.dat", base); let idx_base = volume_file_name(idx_dir, collection, volume_id); let idx_path = format!("{}.idx", idx_base); // Create sorted .ecx from .idx write_sorted_ecx_from_idx(&idx_path, &format!("{}.ecx", base))?; // Encode .dat into shards let dat_file = File::open(&dat_path)?; let dat_size = dat_file.metadata()?.len() as i64; let rs = ReedSolomon::new(data_shards, parity_shards) .map_err(|e| io::Error::other(format!("reed-solomon init: {:?}", e)))?; // Create shard files let total_shards = data_shards + parity_shards; let mut shards: Vec = (0..total_shards as u8) .map(|i| EcVolumeShard::new(dir, collection, volume_id, i)) .collect(); for shard in &mut shards { shard.create()?; } // Per-shard bitrot checksum builders: accumulate a CRC32C for every // DEFAULT_BITROT_BLOCK_SIZE block of each shard's byte stream as it is // written, so the resulting `.ecsum` sidecar can later detect silent // corruption in any shard (including cold parity). let mut builders: Vec = (0..total_shards) .map(|_| ShardChecksumBuilder::new(DEFAULT_BITROT_BLOCK_SIZE as i64)) .collect(); let block_size = uniform_block_size(dat_size, data_shards); encode_dat_file( &dat_file, dat_size, &rs, &mut shards, &mut builders, EcEncodeLayout { data_shards, parity_shards, buffer_size: ENCODE_BUFFER_SIZE, large_block_size: block_size as usize, small_block_size: block_size as usize, }, )?; // Close all shards for shard in &mut shards { shard.close(); } // Write the generation-0 bitrot sidecar (`.ecsum`). Finalizing each // builder yields covered_size (== total bytes written to that shard) and // the packed little-endian CRC32C array. let mut shard_checksums: Vec = Vec::with_capacity(total_shards); for (i, builder) in builders.into_iter().enumerate() { let (covered_size, packed) = builder.finalize(); shard_checksums.push(EcShardChecksums { shard_id: i as u32, covered_size, block_crc32c: packed, }); } let prot = EcBitrotProtection { algorithm: ChecksumAlgorithm::ChecksumCrc32c as i32, block_size: DEFAULT_BITROT_BLOCK_SIZE as u32, generation: 0, ec_shard_config: Some(ec_bitrot::ec_shard_config( data_shards as u32, parity_shards as u32, block_size, )), shards: shard_checksums, encode_uuid: ec_bitrot::new_encode_uuid(), }; let sidecar_path = ec_bitrot::bitrot_sidecar_path(&base, 0); if let Err(e) = ec_bitrot::save_bitrot_sidecar(&sidecar_path, &prot) { // A failed sidecar must not fail the encode — the shards are already // written and valid. The volume simply runs with bitrot protection // off for this generation until the sidecar is regenerated. tracing::warn!( volume_id = volume_id.0, path = %sidecar_path, error = %e, "ec encode: failed to write bitrot sidecar; protection off for this generation", ); } Ok(block_size) } /// uniform_block_size returns the per-shard block size of the uniform layout /// for a .dat of the given size: ceil(dat_file_size/data_shards) rounded up to /// a whole small block. For every input this equals the legacy layout's padded /// shard size, so only the byte placement differs between the two layouts, /// never the shard length. Mirrors Go's UniformBlockSize. pub fn uniform_block_size(dat_file_size: i64, data_shards: usize) -> i64 { let small = ERASURE_CODING_SMALL_BLOCK_SIZE as i64; let per_shard = (dat_file_size + data_shards as i64 - 1) / data_shards as i64; let blocks = ((per_shard + small - 1) / small).max(1); blocks * small } /// Rebuild missing EC shard files from existing shards using Reed-Solomon reconstruct. /// /// This does not require the `.dat` file, only the existing `.ecXX` shard files. /// /// `additional_dirs` lists sibling disk locations to search for existing shards when /// they are not found in `dir`. This is required on multi-disk volume servers where /// shards for the same volume may be spread across disks. pub fn rebuild_ec_files( dir: &str, collection: &str, volume_id: VolumeId, missing_shard_ids: &[u32], data_shards: usize, parity_shards: usize, additional_dirs: &[&str], ) -> io::Result<()> { if missing_shard_ids.is_empty() { return Ok(()); } let rs = ReedSolomon::new(data_shards, parity_shards) .map_err(|e| io::Error::other(format!("reed-solomon init: {:?}", e)))?; let total_shards = data_shards + parity_shards; let mut shards: Vec = (0..total_shards as u8) .map(|i| EcVolumeShard::new(dir, collection, volume_id, i)) .collect(); // Determine the exact shard size from the first available existing shard. // When a shard is not found in `dir`, search `additional_dirs` (sibling disks // on the same volume server) before failing — mirrors Go's findShardFile logic. let mut shard_size = 0; for (i, shard) in shards.iter_mut().enumerate() { if !missing_shard_ids.contains(&(i as u32)) { if shard.open().is_ok() { let size = shard.file_size(); if size > shard_size { shard_size = size; } } else { // Try sibling disk locations before giving up. let mut found = false; for &other_dir in additional_dirs { let mut alt = EcVolumeShard::new(other_dir, collection, volume_id, i as u8); if alt.open().is_ok() { let size = alt.file_size(); if size > shard_size { shard_size = size; } *shard = alt; found = true; break; } } if !found { return Err(io::Error::new( io::ErrorKind::NotFound, format!("missing non-rebuild shard {}", i), )); } } } } if shard_size == 0 { return Err(io::Error::new( io::ErrorKind::InvalidData, "all existing shards are empty or cannot find an existing shard to determine size", )); } // Create the missing shards for writing for i in missing_shard_ids { if let Some(shard) = shards.get_mut(*i as usize) { shard.create()?; } } let block_size = ERASURE_CODING_SMALL_BLOCK_SIZE; let mut remaining = shard_size; let mut offset: u64 = 0; // Process all data in blocks while remaining > 0 { let to_process = remaining.min(block_size as i64) as usize; // Allocate buffers for all shards. Option> is required by rs.reconstruct() let mut buffers: Vec>> = vec![None; total_shards]; // Read available shards. A short read means a truncated/corrupt input // shard; treat it as an error rather than reconstructing over a // zero-padded tail and publishing the result as restored redundancy. for (i, shard) in shards.iter().enumerate() { if !missing_shard_ids.contains(&(i as u32)) { let mut buf = vec![0u8; to_process]; let n = shard.read_at(&mut buf, offset)?; if n != to_process { return Err(io::Error::new( io::ErrorKind::UnexpectedEof, format!( "ec rebuild short read shard {} at {}: got {} want {}", i, offset, n, to_process ), )); } buffers[i] = Some(buf); } } // Reconstruct missing shards rs.reconstruct(&mut buffers) .map_err(|e| io::Error::other(format!("reed-solomon reconstruct: {:?}", e)))?; // Write recovered data into the missing shards for i in missing_shard_ids { let idx = *i as usize; if let Some(buf) = buffers[idx].take() { shards[idx].write_all(&buf)?; } } offset += to_process as u64; remaining -= to_process as i64; } // Close all shards for shard in &mut shards { shard.close(); } Ok(()) } /// Reed-Solomon parity check over the locally-held shards: recompute parity from /// the data shards and flag any shard whose bytes disagree. Wired into FULL /// (scrub mode 2) as a TEMPORARY local parity/cold-region check — the per-needle /// FULL walk only reads live data-shard intervals, so on its own it can't catch /// bitrot in a parity shard or an unwalked region. Move to mode 4 (CHECKSUM) and /// drop it from mode 2 once the `.ecsum` subsystem lands. /// /// `dirs` is indexed BY SHARD ID: each entry is the directory holding that /// shard, or `None` when no disk mounts it. A reconciled volume's shards can be /// split across disks, so a single directory cannot address them all. pub fn verify_ec_shards( dirs: &[Option], collection: &str, volume_id: VolumeId, data_shards: usize, parity_shards: usize, ) -> io::Result<(Vec, Vec)> { let rs = ReedSolomon::new(data_shards, parity_shards) .map_err(|e| io::Error::other(format!("reed-solomon init: {:?}", e)))?; let total_shards = data_shards + parity_shards; let mut shards: Vec> = (0..total_shards) .map(|i| { dirs.get(i) .and_then(|d| d.as_ref()) .map(|d| EcVolumeShard::new(d, collection, volume_id, i as u8)) }) .collect(); let mut shard_size = 0; let mut broken_shards = std::collections::HashSet::new(); let mut details = Vec::new(); for (i, slot) in shards.iter_mut().enumerate() { match slot.as_mut() { // Not a match guard: a binding is immutable until the guard ends, // and `open()` needs `&mut self`. Some(shard) => { if shard.open().is_ok() { let size = shard.file_size(); if size > shard_size { shard_size = size; } } else { broken_shards.insert(i as u32); details.push(format!("failed to open or missing shard {}", i)); } } None => { broken_shards.insert(i as u32); details.push(format!("shard {} is not mounted on any disk", i)); } } } if shard_size == 0 || broken_shards.len() >= parity_shards { // Can't do much if we don't know the size or have too many missing. // Sort like the normal path below: a `HashSet` iteration order would // make this return shard ids in an arbitrary order, and enough `None` // entries in `dirs` now reach this branch for a caller to notice. let mut broken_vec: Vec = broken_shards.into_iter().collect(); broken_vec.sort_unstable(); return Ok((broken_vec, details)); } let block_size = ERASURE_CODING_SMALL_BLOCK_SIZE; let mut remaining = shard_size; let mut offset: u64 = 0; while remaining > 0 { let to_process = remaining.min(block_size as i64) as usize; let mut buffers = vec![vec![0u8; to_process]; total_shards]; let mut read_failed = false; for i in 0..total_shards { if !broken_shards.contains(&(i as u32)) { // The `None` arm is defensive and unreachable: the open loop // put every unmounted slot in `broken_shards`, which this // branch already skipped. Kept because the `Option` forces // some handling here, and an error is the only shape that // cannot quietly feed an unread buffer into the parity // comparison below. Nothing needs to cover it. let read = match shards[i].as_mut() { Some(shard) => shard.read_at(&mut buffers[i], offset), None => Err(io::Error::new(io::ErrorKind::NotFound, "shard not mounted")), }; if let Err(e) = read { broken_shards.insert(i as u32); details.push(format!("read error shard {}: {}", i, e)); read_failed = true; } } else { read_failed = true; } } // Only do verification if all shards were readable if !read_failed { // Need to convert Vec> to &[&[u8]] for rs.verify let slice_ptrs: Vec<&[u8]> = buffers.iter().map(|v| v.as_slice()).collect(); if let Ok(is_valid) = rs.verify(&slice_ptrs) && !is_valid { // Reed-Solomon verification failed. We cannot easily pinpoint which shard // is corrupted without recalculating parities or syndromes, so we just // log that this batch has corruption. Wait, we can test each parity shard! // Let's re-encode from the first `data_shards` and compare to the actual `parity_shards`. let mut verify_buffers = buffers.clone(); // Clear the parity parts for buf in &mut verify_buffers[data_shards..total_shards] { buf.fill(0); } if rs.encode(&mut verify_buffers).is_ok() { for i in 0..total_shards { if buffers[i] != verify_buffers[i] { broken_shards.insert(i as u32); details.push(format!( "parity mismatch on shard {} at offset {}", i, offset )); } } } } } offset += to_process as u64; remaining -= to_process as i64; } // Close all shards for shard in shards.iter_mut().flatten() { shard.close(); } let mut broken_vec: Vec = broken_shards.into_iter().collect(); broken_vec.sort_unstable(); Ok((broken_vec, details)) } /// Write sorted .ecx index from .idx file. pub(crate) fn write_sorted_ecx_from_idx(idx_path: &str, ecx_path: &str) -> io::Result<()> { if !std::path::Path::new(idx_path).exists() { return Err(io::Error::new( io::ErrorKind::NotFound, "idx file not found", )); } // Read all idx entries let mut idx_file = File::open(idx_path)?; let mut entries: Vec<(NeedleId, Offset, Size)> = Vec::new(); idx::walk_index_file(&mut idx_file, 0, |key, offset, size| { entries.push((key, offset, size)); Ok(()) })?; // Sort by NeedleId, then by actual offset so later entries come last entries.sort_by_key(|&(key, offset, _)| (key, offset.to_actual_offset())); // Remove duplicates (keep last/latest entry for each key). // dedup_by_key keeps the first in each run, so we reverse first, // dedup, then reverse back. entries.reverse(); entries.dedup_by_key(|entry| entry.0); entries.reverse(); // Write sorted entries to .ecx let mut ecx_file = File::create(ecx_path)?; for &(key, offset, size) in &entries { idx::write_index_entry(&mut ecx_file, key, offset, size)?; } Ok(()) } /// Rebuild the .ecx index file by walking needles in the EC data shards. /// /// This is the equivalent of Go's `RebuildEcxFile`. It reads the logical .dat /// content from the EC data shards, walks through needle headers to extract /// (needle_id, offset, size) entries, deduplicates them, and writes a sorted /// .ecx index file. /// /// `additional_dirs` lists sibling disk locations to search for data shards when /// they are not found in `dir` — mirrors the same fallback used by rebuild_ec_files, /// required on multi-disk volume servers where shards may be spread across disks. pub fn rebuild_ecx_file( dir: &str, collection: &str, volume_id: VolumeId, data_shards: usize, block_size: i64, dat_file_size: i64, additional_dirs: &[&str], ) -> io::Result<()> { use crate::storage::needle::needle::get_actual_size; use crate::storage::super_block::SUPER_BLOCK_SIZE; let base = volume_file_name(dir, collection, volume_id); let ecx_path = format!("{}.ecx", base); // Open data shards to read logical .dat content. When a shard isn't found // in `dir`, search `additional_dirs` (sibling disks on the same volume // server) before giving up. let mut shards: Vec = (0..data_shards as u8) .map(|i| EcVolumeShard::new(dir, collection, volume_id, i)) .collect(); for (i, shard) in shards.iter_mut().enumerate() { if shard.open().is_err() { let mut found = false; for &other_dir in additional_dirs { let mut alt = EcVolumeShard::new(other_dir, collection, volume_id, i as u8); if alt.open().is_ok() { *shard = alt; found = true; break; } } if !found { // If a data shard is missing, we can't rebuild ecx for s in &mut shards { s.close(); } return Err(io::Error::new( io::ErrorKind::NotFound, "cannot open data shard for ecx rebuild".to_string(), )); } } } // Determine total logical data size from shard sizes let shard_size = shards.iter().map(|s| s.file_size()).max().unwrap_or(0); let total_data_size = shard_size * data_shards as i64; // The volume's shard block layout: the .vif-recorded uniform block size, // or the legacy two-tier sizes when 0. The row count comes from the shard // length; -1 disambiguates a legacy shard that is an exact large-block // multiple (mirrors the ecdFileSize-1 fallback in the read path). let (large_block, small_block) = if block_size > 0 { (block_size, block_size) } else { ( ERASURE_CODING_LARGE_BLOCK_SIZE as i64, ERASURE_CODING_SMALL_BLOCK_SIZE as i64, ) }; // The row count the de-stripe walks with. The encode-time .dat size is the // authority — the same value the read path divides by data_shards — and the // padded extent is only a fallback: under the legacy layout a shard that is // an exact large-block multiple reads as one row too many, which // re-interprets its last large row as small blocks and scrambles the // recovered offsets. Subtracting one keeps that fallback on the safe side of // the boundary, exactly as the read path's own fallback does. let locate_shard_size = if dat_file_size > 0 { dat_file_size / data_shards as i64 } else { (shard_size - 1).max(0) }; // Read version from superblock (first byte of logical data) let mut sb_buf = [0u8; SUPER_BLOCK_SIZE]; read_from_data_shards( &shards, &mut sb_buf, 0, data_shards, locate_shard_size, large_block, small_block, )?; let version = Version(sb_buf[0]); // Walk needles starting after superblock let mut offset = SUPER_BLOCK_SIZE as i64; let header_size = NEEDLE_HEADER_SIZE; let mut entries: Vec<(NeedleId, Offset, Size)> = Vec::new(); while offset + header_size as i64 <= total_data_size { // Read needle header (cookie + needle_id + size = 16 bytes). // A read failure is NOT the end of the data — every offset in // range maps into the shards, so an error means a truncated or // unreadable shard. Publishing the entries collected so far as // a successful .ecx would hand out a silently incomplete // recovery index; propagate instead. (The scan still ends // normally on the zero-cookie tail below.) let mut header_buf = [0u8; NEEDLE_HEADER_SIZE]; if let Err(e) = read_from_data_shards( &shards, &mut header_buf, offset as u64, data_shards, locate_shard_size, large_block, small_block, ) { for s in &mut shards { s.close(); } return Err(io::Error::new( e.kind(), format!("scan needle header at offset {}: {}", offset, e), )); } let cookie = Cookie::from_bytes(&header_buf[..COOKIE_SIZE]); let needle_id = NeedleId::from_bytes(&header_buf[COOKIE_SIZE..COOKIE_SIZE + NEEDLE_ID_SIZE]); let size = Size::from_bytes(&header_buf[COOKIE_SIZE + NEEDLE_ID_SIZE..header_size]); // Validate: stop if we hit zero cookie+id (end of data) if cookie.0 == 0 && needle_id.0 == 0 { break; } // Validate size is reasonable if size.0 < 0 && !size.is_deleted() { break; } let actual_size = get_actual_size(size, version); if actual_size <= 0 || offset + actual_size > total_data_size { break; } entries.push((needle_id, Offset::from_actual_offset(offset), size)); // Advance to next needle (aligned to NEEDLE_PADDING_SIZE) offset += actual_size; let padding_rem = offset % NEEDLE_PADDING_SIZE as i64; if padding_rem != 0 { offset += NEEDLE_PADDING_SIZE as i64 - padding_rem; } } for shard in &mut shards { shard.close(); } // Sort by NeedleId, then by offset (later entries override earlier) entries.sort_by_key(|&(key, offset, _)| (key, offset.to_actual_offset())); // Deduplicate: keep latest entry per needle_id entries.reverse(); entries.dedup_by_key(|entry| entry.0); entries.reverse(); // Write sorted .ecx let mut ecx_file = File::create(&ecx_path)?; for &(key, offset, size) in &entries { idx::write_index_entry(&mut ecx_file, key, offset, size)?; } ecx_file.sync_all()?; Ok(()) } /// Read bytes from EC data shards at a logical offset in the .dat file, /// resolving the shard/offset through the volume's block layout via /// locate_data — the same mapping the read path uses. fn read_from_data_shards( shards: &[EcVolumeShard], buf: &mut [u8], logical_offset: u64, data_shards: usize, locate_shard_size: i64, large_block_size: i64, small_block_size: i64, ) -> io::Result<()> { let intervals = crate::storage::erasure_coding::ec_locate::locate_data( logical_offset as i64, Size(buf.len() as i32), locate_shard_size, data_shards as u32, large_block_size, small_block_size, ); let mut bytes_read = 0usize; for interval in &intervals { let (shard_id, shard_offset) = interval.to_shard_id_and_offset(data_shards as u32, large_block_size, small_block_size); if shard_id as usize >= data_shards { return Err(io::Error::new( io::ErrorKind::InvalidInput, "shard index out of range", )); } let to_read = interval.size as usize; let dest = &mut buf[bytes_read..bytes_read + to_read]; // Exact-read semantics: read_at may legally return fewer bytes // than requested, and treating a short read as complete leaves // the tail of `dest` as whatever the buffer held before. Loop // until filled; zero bytes inside the mapped range means the // shard is truncated — an error, not an end. let mut filled = 0usize; while filled < to_read { let n = shards[shard_id as usize] .read_at(&mut dest[filled..], shard_offset as u64 + filled as u64)?; if n == 0 { return Err(io::Error::new( io::ErrorKind::UnexpectedEof, format!( "short read from data shard {}: {} of {} bytes at offset {}", shard_id, filled, to_read, shard_offset ), )); } filled += n; } bytes_read += to_read; } if bytes_read != buf.len() { return Err(io::Error::new( io::ErrorKind::UnexpectedEof, "short read from data shards", )); } Ok(()) } /// Buffer size for one encode sub-batch per shard, mirroring Go's 256KB /// bufferSize in WriteEcFiles. A block is processed in block_size/buffer_size /// sub-batches, so memory stays at total_shards * 256KB no matter how large /// the uniform block is. const ENCODE_BUFFER_SIZE: usize = 256 * 1024; /// Shape of one encode run: the Reed-Solomon split and the block sizes that /// fix where every byte of the .dat lands in the shards. Mirrors Go's /// `ECContext`. `buffer_size` must divide both block sizes. #[derive(Clone, Copy, Debug)] pub(crate) struct EcEncodeLayout { pub(crate) data_shards: usize, pub(crate) parity_shards: usize, /// Bytes of each shard's block handled per sub-batch; bounds memory at /// `total_shards * buffer_size` however large the blocks are. pub(crate) buffer_size: usize, pub(crate) large_block_size: usize, pub(crate) small_block_size: usize, } /// Encode the .dat file data into shard files. /// /// Uses a two-phase approach matching Go's ec_encoder.go: /// 1. Process as many large blocks as possible /// 2. Process remaining data with small blocks pub(crate) fn encode_dat_file( dat_file: &File, dat_size: i64, rs: &ReedSolomon, shards: &mut [EcVolumeShard], builders: &mut [ShardChecksumBuilder], layout: EcEncodeLayout, ) -> io::Result<()> { let EcEncodeLayout { data_shards, parity_shards, buffer_size, large_block_size, small_block_size, } = layout; let total_shards = data_shards + parity_shards; let mut buffers: Vec> = (0..total_shards).map(|_| vec![0u8; buffer_size]).collect(); let mut run = EncodeRun { dat_file, rs, buffers: &mut buffers, shards, builders, data_shards, }; let mut remaining = dat_size; let mut offset: u64 = 0; // Phase 1: process whole large-block rows while enough data remains let large_row_size = large_block_size * data_shards; while remaining >= large_row_size as i64 { run.encode_row(offset, large_block_size)?; offset += large_row_size as u64; remaining -= large_row_size as i64; } // Phase 2: process remaining data with small blocks let small_row_size = small_block_size * data_shards; while remaining > 0 { let to_process = remaining.min(small_row_size as i64); run.encode_row(offset, small_block_size)?; offset += to_process as u64; remaining -= to_process; } Ok(()) } /// Everything one encode run streams through: the source .dat, the codec, a /// buffer per shard, and the per-shard file and checksum sinks. struct EncodeRun<'a> { dat_file: &'a File, rs: &'a ReedSolomon, buffers: &'a mut [Vec], shards: &'a mut [EcVolumeShard], builders: &'a mut [ShardChecksumBuilder], data_shards: usize, } impl EncodeRun<'_> { /// Encode one row of blocks, streaming it in ENCODE_BUFFER_SIZE sub-batches /// so arbitrarily large blocks never require block-sized allocations. /// Mirrors Go's encodeData. fn encode_row(&mut self, row_offset: u64, block_size: usize) -> io::Result<()> { let buffer_size = self.buffers[0].len(); if !block_size.is_multiple_of(buffer_size) { return Err(io::Error::new( io::ErrorKind::InvalidInput, format!( "unexpected block size {} buffer size {}", block_size, buffer_size ), )); } let batch_count = block_size / buffer_size; for b in 0..batch_count { self.encode_one_batch(row_offset + (b * buffer_size) as u64, block_size)?; } Ok(()) } /// Encode one sub-batch: the same buffer-sized slice of every shard's block /// in this row. Mirrors Go's encodeDataOneBatch. fn encode_one_batch(&mut self, offset: u64, block_size: usize) -> io::Result<()> { // Read data shards from the .dat file, zero-filling past EOF — the // buffers are reused across batches, so the tail must be cleared // explicitly. for (i, buf) in self.buffers[..self.data_shards].iter_mut().enumerate() { let read_offset = offset + (i * block_size) as u64; let n = read_at_most(self.dat_file, buf, read_offset)?; buf[n..].fill(0); } // Encode parity shards self.rs .encode(&mut *self.buffers) .map_err(|e| io::Error::other(format!("reed-solomon encode: {:?}", e)))?; // Write all shard buffers to files and feed the same bytes to each // shard's bitrot checksum builder, keeping covered_size == on-disk // length. for (i, buf) in self.buffers.iter().enumerate() { self.shards[i].write_all(buf)?; self.builders[i].write(buf); } Ok(()) } } /// Read into `buf` at `offset` until it is full or EOF; returns bytes read. fn read_at_most(dat_file: &File, buf: &mut [u8], offset: u64) -> io::Result { let mut n = 0; while n < buf.len() { #[cfg(unix)] let r = { use std::os::unix::fs::FileExt; dat_file.read_at(&mut buf[n..], offset + n as u64)? }; #[cfg(not(unix))] let r = { let mut f = dat_file.try_clone()?; f.seek(SeekFrom::Start(offset + n as u64))?; f.read(&mut buf[n..])? }; if r == 0 { break; } n += r; } Ok(n) } #[cfg(test)] mod tests { use super::*; use crate::storage::needle::needle::Needle; use crate::storage::needle_map::NeedleMapKind; use crate::storage::volume::{Volume, VolumeSpec}; use tempfile::TempDir; #[test] fn test_ec_encode_decode_round_trip() { let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); // Create a volume with some data let mut v = Volume::new( dir, dir, VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1..=5 { let data = format!("test data for needle {}", 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, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); // Encode to EC shards let data_shards = 10; let parity_shards = 4; let total_shards = data_shards + parity_shards; write_ec_files(dir, dir, "", VolumeId(1), data_shards, parity_shards).unwrap(); // Verify shard files exist for i in 0..total_shards { let path = format!("{}/{}.ec{:02}", dir, 1, i); assert!( std::path::Path::new(&path).exists(), "shard file {} should exist", path ); } // Verify .ecx exists let ecx_path = format!("{}/1.ecx", dir); assert!(std::path::Path::new(&ecx_path).exists()); } fn make_volume_with_needles(n: u64) -> TempDir { let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); let mut v = Volume::new( dir, dir, VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1..=n { // Larger payloads so encoded shards span multiple bitrot blocks // would require huge data; small payloads are fine for correctness. let data = format!("test data for needle {} {}", i, "x".repeat(64)); let mut needle = 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 needle, true, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); tmp } /// Encode-time capture writes a valid generation-0 `.ecsum` sidecar whose /// recorded checksums match the actual on-disk shards. #[test] fn test_encode_writes_valid_bitrot_sidecar() { use crate::storage::erasure_coding::ec_bitrot; let tmp = make_volume_with_needles(5); let dir = tmp.path().to_str().unwrap(); write_ec_files(dir, dir, "", VolumeId(1), 10, 4).unwrap(); let base = format!("{}/1", dir); let sidecar_path = ec_bitrot::bitrot_sidecar_path(&base, 0); assert!( std::path::Path::new(&sidecar_path).exists(), "generation-0 .ecsum sidecar should exist after encode" ); let prot = ec_bitrot::load_bitrot_sidecar(&sidecar_path).unwrap(); ec_bitrot::validate_manifest(&prot, 10, 4).unwrap(); assert_eq!(prot.generation, 0); assert_eq!(prot.shards.len(), 14); assert_eq!(prot.encode_uuid.len(), 16); assert_eq!(prot.block_size, ec_bitrot::DEFAULT_BITROT_BLOCK_SIZE as u32); // Every shard's recorded covered_size must equal its on-disk length and // its block CRCs must verify clean. let bs = prot.block_size as i64; for entry in &prot.shards { let path = format!("{}.ec{:02}", base, entry.shard_id); let on_disk = std::fs::metadata(&path).unwrap().len() as i64; assert_eq!( entry.covered_size, on_disk, "covered_size must equal on-disk length for shard {}", entry.shard_id ); let mm = ec_bitrot::verify_shard_file_blocks(&path, entry, bs).unwrap(); assert!( mm.is_empty(), "shard {} should verify clean, got mismatches {:?}", entry.shard_id, mm ); } } // encode_sample_volume writes a small volume and EC-encodes it, returning // the dir path so a test can drop/truncate shards and rebuild. fn encode_sample_volume(tmp: &TempDir) -> String { let dir = tmp.path().to_str().unwrap().to_string(); let mut v = Volume::new( &dir, &dir, VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1..=20 { let data = format!("test data for needle {} padded with bytes", i).repeat(64); 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, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); write_ec_files(&dir, &dir, "", VolumeId(1), 10, 4).unwrap(); dir } // A truncated/corrupt input shard must abort rebuild_ec_files with an error // rather than reconstructing over a zero-padded tail and publishing a // truncated shard as restored redundancy. #[test] fn test_rebuild_ec_files_short_read_input_errors() { let tmp = TempDir::new().unwrap(); let dir = encode_sample_volume(&tmp); // Truncate a present input shard to half its size. let victim = format!("{}/1.ec03", dir); let full = std::fs::metadata(&victim).unwrap().len(); assert!(full > 0, "encoded shard should be non-empty"); let f = std::fs::OpenOptions::new() .write(true) .open(&victim) .unwrap(); f.set_len(full / 2).unwrap(); drop(f); // Rebuild a different (genuinely missing) shard. std::fs::remove_file(format!("{}/1.ec07", dir)).unwrap(); let res = rebuild_ec_files(&dir, "", VolumeId(1), &[7], 10, 4, &[]); assert!(res.is_err(), "truncated input shard must abort the rebuild"); } // Happy path: dropping a shard and rebuilding it from the rest must succeed // and recreate the shard file (the short-read guard must not false-positive). #[test] fn test_rebuild_ec_files_happy_path() { let tmp = TempDir::new().unwrap(); let dir = encode_sample_volume(&tmp); let dropped = format!("{}/1.ec07", dir); std::fs::remove_file(&dropped).unwrap(); rebuild_ec_files(&dir, "", VolumeId(1), &[7], 10, 4, &[]).unwrap(); assert!( std::path::Path::new(&dropped).exists(), "rebuilt shard .ec07 should exist" ); assert!( std::fs::metadata(&dropped).unwrap().len() > 0, "rebuilt shard should be non-empty" ); } // Multi-disk rebuild: shards for the same volume are split across two // directories (simulating a volume server with two disk locations). The // shard to rebuild is in the primary dir; some of the input shards needed // for RS reconstruction exist only in the secondary dir. Passing the // secondary dir via `additional_dirs` must allow the rebuild to succeed // where it would previously return "missing non-rebuild shard N". #[test] fn test_rebuild_ec_files_multi_disk() { let tmp_primary = TempDir::new().unwrap(); let tmp_secondary = TempDir::new().unwrap(); let primary = tmp_primary.path().to_str().unwrap().to_string(); let secondary = tmp_secondary.path().to_str().unwrap().to_string(); // Encode into primary dir first (all 14 shards land there). let dir = encode_sample_volume(&tmp_primary); assert_eq!(dir, primary); // Simulate a multi-disk layout: move shards 0, 4, 8 to the secondary // dir, as if the master had placed them on a different disk. let moved_shards: &[u8] = &[0, 4, 8]; for &shard_id in moved_shards { let src = format!("{}/1.ec{:02}", primary, shard_id); let dst = format!("{}/1.ec{:02}", secondary, shard_id); std::fs::rename(&src, &dst).unwrap(); } // Remove shard 7 from primary — this is the shard we want to rebuild. let missing_path = format!("{}/1.ec07", primary); std::fs::remove_file(&missing_path).unwrap(); // Without additional_dirs the rebuild must fail: shards 0, 4, 8 are // not in primary and shard 7 cannot be reconstructed without them. let res = rebuild_ec_files(&primary, "", VolumeId(1), &[7], 10, 4, &[]); assert!( res.is_err(), "rebuild without additional_dirs must fail when input shards are on another disk" ); assert!( !std::path::Path::new(&missing_path).exists(), "failed rebuild must not leave a partial shard file behind" ); // With additional_dirs pointing at the secondary, the rebuild must succeed. rebuild_ec_files( &primary, "", VolumeId(1), &[7], 10, 4, &[secondary.as_str()], ) .unwrap(); assert!( std::path::Path::new(&missing_path).exists(), "rebuilt shard .ec07 should exist in primary dir" ); assert!( std::fs::metadata(&missing_path).unwrap().len() > 0, "rebuilt shard must be non-empty" ); } // Multi-disk .ecx rebuild: data shards (0-9) needed to reconstruct the // logical .dat content are split across two directories, as on a // multi-disk volume server. Passing the secondary dir via // `additional_dirs` must allow rebuild_ecx_file to find them and // succeed where it would previously fail with // "cannot open data shard for ecx rebuild". #[test] fn test_rebuild_ecx_file_multi_disk() { let tmp_primary = TempDir::new().unwrap(); let tmp_secondary = TempDir::new().unwrap(); let primary = tmp_primary.path().to_str().unwrap().to_string(); let secondary = tmp_secondary.path().to_str().unwrap().to_string(); let dir = encode_sample_volume(&tmp_primary); assert_eq!(dir, primary); // Move some data shards (0-9) to the secondary dir, simulating a // multi-disk layout where not all data shards landed on the same disk. let moved_shards: &[u8] = &[1, 3, 6]; for &shard_id in moved_shards { let src = format!("{}/1.ec{:02}", primary, shard_id); let dst = format!("{}/1.ec{:02}", secondary, shard_id); std::fs::rename(&src, &dst).unwrap(); } // Delete the existing .ecx to force a rebuild. let ecx_path = format!("{}/1.ecx", primary); std::fs::remove_file(&ecx_path).unwrap(); // Without additional_dirs, rebuild must fail: shards 1, 3, 6 are not // in primary and the full logical .dat content can't be reconstructed. let res = rebuild_ecx_file(&primary, "", VolumeId(1), 10, 0, 0, &[]); assert!( res.is_err(), "ecx rebuild without additional_dirs must fail when data shards are on another disk" ); assert!( !std::path::Path::new(&ecx_path).exists(), "failed ecx rebuild must not leave a partial .ecx file behind" ); // With additional_dirs pointing at the secondary, the rebuild must succeed. rebuild_ecx_file(&primary, "", VolumeId(1), 10, 0, 0, &[secondary.as_str()]).unwrap(); assert!( std::path::Path::new(&ecx_path).exists(), "rebuilt .ecx should exist in primary dir" ); assert!( std::fs::metadata(&ecx_path).unwrap().len() > 0, "rebuilt .ecx must be non-empty" ); } // A uniform-layout volume (block size > 1MiB) must have its .ecx rebuilt // through the recorded geometry; the legacy 1MiB mapping would scan // garbage past the first block boundary. #[test] fn test_rebuild_ecx_file_uniform_layout() { use crate::storage::needle_map::NeedleMapKind; use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap().to_string(); let mut v = Volume::new( &dir, &dir, VolumeId(2), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1u64..=12 { let data: Vec = (0..2 << 20) .map(|b| ((b as u64).wrapping_mul(2654435761).wrapping_add(i) >> 8) as u8) .collect(); let mut n = Needle { id: NeedleId(i), cookie: Cookie(i as u32), data: data.clone(), data_size: data.len() as u32, ..Needle::default() }; v.write_needle(&mut n, true, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); let block_size = write_ec_files(&dir, &dir, "", VolumeId(2), 10, 4).unwrap(); assert!( block_size > ERASURE_CODING_SMALL_BLOCK_SIZE as i64, "fixture must diverge from the legacy layout" ); let ecx_path = format!("{}/2.ecx", dir); let canonical = std::fs::read(&ecx_path).unwrap(); std::fs::remove_file(&ecx_path).unwrap(); rebuild_ecx_file(&dir, "", VolumeId(2), 10, block_size, 0, &[]).unwrap(); let rebuilt = std::fs::read(&ecx_path).unwrap(); assert_eq!( canonical, rebuilt, "rebuilt .ecx must match the encode-time .ecx" ); } // A truncated data shard must FAIL the .ecx rebuild, not publish the // entries scanned so far as a successful (silently incomplete) index. #[test] fn test_rebuild_ecx_file_fails_on_truncated_shard() { use crate::storage::needle_map::NeedleMapKind; use crate::storage::volume::{Volume, VolumeSpec}; let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap().to_string(); let mut v = Volume::new( &dir, &dir, VolumeId(3), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1u64..=12 { let data: Vec = (0..2 << 20) .map(|b| ((b as u64).wrapping_mul(2654435761).wrapping_add(i) >> 8) as u8) .collect(); let mut n = Needle { id: NeedleId(i), cookie: Cookie(i as u32), data: data.clone(), data_size: data.len() as u32, ..Needle::default() }; v.write_needle(&mut n, true, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); let block_size = write_ec_files(&dir, &dir, "", VolumeId(3), 10, 4).unwrap(); let ecx_path = format!("{}/3.ecx", dir); std::fs::remove_file(&ecx_path).unwrap(); // Truncate shard 0 to just the superblock: the scan's very first // needle-header read (offset SUPER_BLOCK_SIZE, shard 0 under the // uniform layout) lands in the missing region. The pre-fix code // broke the scan there and published an EMPTY .ecx as success. let shard_path = format!("{}/3.ec00", dir); let f = std::fs::OpenOptions::new() .write(true) .open(&shard_path) .unwrap(); f.set_len(crate::storage::super_block::SUPER_BLOCK_SIZE as u64) .unwrap(); drop(f); let res = rebuild_ecx_file(&dir, "", VolumeId(3), 10, block_size, 0, &[]); assert!(res.is_err(), "rebuild over a truncated shard must fail"); assert!( !std::path::Path::new(&ecx_path).exists(), "a failed rebuild must not leave a partial .ecx behind" ); } #[test] fn test_reed_solomon_basic() { let data_shards = 10; let parity_shards = 4; let total_shards = data_shards + parity_shards; let rs = ReedSolomon::new(data_shards, parity_shards).unwrap(); let block_size = 1024; let mut shards: Vec> = (0..total_shards) .map(|i| { if i < data_shards { vec![(i as u8).wrapping_mul(7); block_size] } else { vec![0u8; block_size] } }) .collect(); // Encode rs.encode(&mut shards).unwrap(); // Verify parity is non-zero (at least some) let parity_nonzero: bool = shards[data_shards..] .iter() .any(|s| s.iter().any(|&b| b != 0)); assert!(parity_nonzero); // Simulate losing 4 shards and reconstructing let original_0 = shards[0].clone(); let original_1 = shards[1].clone(); let mut shard_opts: Vec>> = shards.into_iter().map(Some).collect(); shard_opts[0] = None; shard_opts[1] = None; shard_opts[2] = None; shard_opts[3] = None; rs.reconstruct(&mut shard_opts).unwrap(); assert_eq!(shard_opts[0].as_ref().unwrap(), &original_0); assert_eq!(shard_opts[1].as_ref().unwrap(), &original_1); } /// EC encode must read .idx from a separate index directory when configured. #[test] fn test_ec_encode_with_separate_idx_dir() { let dat_tmp = TempDir::new().unwrap(); let idx_tmp = TempDir::new().unwrap(); let dat_dir = dat_tmp.path().to_str().unwrap(); let idx_dir = idx_tmp.path().to_str().unwrap(); // Create a volume with separate data and index directories let mut v = Volume::new( dat_dir, idx_dir, VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1..=5 { let data = format!("needle {} payload", 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, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); // Verify .dat is in data dir, .idx is in idx dir assert!(std::path::Path::new(&format!("{}/1.dat", dat_dir)).exists()); assert!(!std::path::Path::new(&format!("{}/1.idx", dat_dir)).exists()); assert!(std::path::Path::new(&format!("{}/1.idx", idx_dir)).exists()); assert!(!std::path::Path::new(&format!("{}/1.dat", idx_dir)).exists()); // EC encode with separate idx dir let data_shards = 10; let parity_shards = 4; let total_shards = data_shards + parity_shards; write_ec_files( dat_dir, idx_dir, "", VolumeId(1), data_shards, parity_shards, ) .unwrap(); // Verify all 14 shard files in data dir for i in 0..total_shards { let path = format!("{}/1.ec{:02}", dat_dir, i); assert!( std::path::Path::new(&path).exists(), "shard {} should exist in data dir", path ); } // Verify .ecx in data dir (not idx dir) assert!(std::path::Path::new(&format!("{}/1.ecx", dat_dir)).exists()); assert!(!std::path::Path::new(&format!("{}/1.ecx", idx_dir)).exists()); // Verify no shard files leaked into idx dir for i in 0..total_shards { let path = format!("{}/1.ec{:02}", idx_dir, i); assert!( !std::path::Path::new(&path).exists(), "shard {} should NOT exist in idx dir", path ); } } /// EC encode should fail gracefully when .idx is only in the data dir /// but we pass a wrong idx_dir. This guards against regressions where /// write_ec_files ignores the idx_dir parameter. #[test] fn test_ec_encode_fails_with_wrong_idx_dir() { let dat_tmp = TempDir::new().unwrap(); let idx_tmp = TempDir::new().unwrap(); let wrong_tmp = TempDir::new().unwrap(); let dat_dir = dat_tmp.path().to_str().unwrap(); let idx_dir = idx_tmp.path().to_str().unwrap(); let wrong_dir = wrong_tmp.path().to_str().unwrap(); let mut v = Volume::new( dat_dir, idx_dir, VolumeId(1), NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); let mut n = Needle { id: NeedleId(1), cookie: Cookie(1), data: b"hello".to_vec(), data_size: 5, ..Needle::default() }; v.write_needle(&mut n, true, false).unwrap(); v.sync_to_disk().unwrap(); v.close(); // Should fail: .idx is in idx_dir, not wrong_dir let result = write_ec_files(dat_dir, wrong_dir, "", VolumeId(1), 10, 4); assert!( result.is_err(), "should fail when idx_dir doesn't contain .idx" ); } /// Write a real 10+4 encoded volume into `dir`. /// /// Unlike `make_volume_with_needles` and `encode_sample_volume` this seeds /// a caller-chosen directory, which is what a split-disk test needs: the /// shards have to be scattered out of the directory they were encoded into. fn seed_encoded_volume(dir: &str, vid: VolumeId) { let mut v = Volume::new( dir, dir, vid, NeedleMapKind::InMemory, &VolumeSpec::default(), ) .unwrap(); for i in 1..=8 { let data = format!("test data for needle {} with a bit more length", 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, false).unwrap(); } v.sync_to_disk().unwrap(); v.close(); write_ec_files(dir, dir, "", vid, 10, 4).unwrap(); } /// Shards split across two directories must all be found. Passing one dir /// per shard is what lets a reconciled volume's parity be checked at all. #[test] fn test_verify_ec_shards_reads_shards_from_multiple_dirs() { let tmp = TempDir::new().unwrap(); let src = tmp.path().join("src"); let d0 = tmp.path().join("d0"); let d1 = tmp.path().join("d1"); for d in [&src, &d0, &d1] { std::fs::create_dir_all(d).unwrap(); } let src_s = src.to_str().unwrap(); seed_encoded_volume(src_s, VolumeId(1)); // Move shards 0..=6 to d0 and 7..=13 to d1. let mut dirs: Vec> = Vec::new(); for id in 0..14u8 { let target = if id < 7 { &d0 } else { &d1 }; std::fs::rename( format!("{}/1.ec{:02}", src_s, id), format!("{}/1.ec{:02}", target.to_str().unwrap(), id), ) .unwrap(); dirs.push(Some(target.to_str().unwrap().to_string())); } let (broken, details) = verify_ec_shards(&dirs, "", VolumeId(1), 10, 4).unwrap(); assert!( broken.is_empty(), "split-dir shards reported broken: {:?}", details ); } /// A shard no disk holds is a missing shard, not a panic and not a silent /// pass: it is REPORTED, by id, with a message that distinguishes "no disk /// holds this shard" from "the disk holds it but it won't open". /// /// Read the scope literally. This does NOT show that the mounted shards /// verify clean. `dirs[5] = None` puts shard 5 in `broken_shards` before /// the block loop starts, so every iteration takes the /// `else { read_failed = true; }` arm and the Reed-Solomon comparison never /// runs at all. `broken == vec![5]` therefore holds because the other 13 /// were never verified, not because they verified clean -- a parity check /// over intact shards is what /// `test_verify_ec_shards_reads_shards_from_multiple_dirs` and the /// end-to-end split-disk FULL scrub establish. #[test] fn test_verify_ec_shards_treats_a_none_dir_as_missing() { let tmp = TempDir::new().unwrap(); let dir = tmp.path().to_str().unwrap(); seed_encoded_volume(dir, VolumeId(1)); let mut dirs: Vec> = (0..14).map(|_| Some(dir.to_string())).collect(); dirs[5] = None; let (broken, details) = verify_ec_shards(&dirs, "", VolumeId(1), 10, 4).unwrap(); assert_eq!( broken, vec![5], "an unmounted shard must be reported, and only it: {:?}", details ); // "no disk holds this shard" and "the disk holds it but it won't open" // are different operator problems, which is why they carry different // messages. Asserting only the id would let one masquerade as the other. assert!( details.iter().any(|d| d.contains("not mounted")), "an unmounted shard must be distinguished from an unopenable one, got {:?}", details ); } }