diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 8db9f15e6..da1b941bd 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -442,6 +442,7 @@ message VolumeEcShardsCopyRequest { bool copy_ecj_file = 6; bool copy_vif_file = 7; uint32 disk_id = 8; // Target disk ID for storing EC shards + bool copy_ecsum_file = 9; // copy the bitrot checksum sidecar (.ecsum) when present; tolerant of a missing source (no-op), since this non-2PC path has no Prepare backstop } message VolumeEcShardsCopyResponse { } @@ -587,6 +588,29 @@ message EcShardConfig { uint32 parity_shards = 2; // Number of parity shards (e.g., 4) int64 encode_ts_ns = 3; // encode time (unix nanos); a read served from a shard of a different encode run is rejected } +// EcBitrotProtection is the entire content of a bitrot checksum sidecar +// (.ecsum for the legacy generation, .ecsum.v for vacuum +// generation N): per-shard, per-block CRC32C so a CHECKSUM scrub can detect +// silent bit rot in any shard (including parity) without decoding. Field +// numbers and types match weed/pb/volume_server.proto byte-for-byte so the +// serialized sidecar payload is wire-identical across the Go and Rust binaries. +message EcBitrotProtection { + ChecksumAlgorithm algorithm = 1; // CRC32C (Castagnoli) + uint32 block_size = 2; // bytes per checksum block; default 16777216 (16 MiB), a power-of-two multiple of 1 MiB + uint32 generation = 3; // EC vacuum generation these checksums describe (0 = legacy/fresh); must match the sidecar filename version + EcShardConfig ec_shard_config = 4; // data/parity shard counts at encode time + repeated EcShardChecksums shards = 5; // one entry per shard id in the active layout + bytes encode_uuid = 6; // random per-encode identity, for stale-sidecar detection across in-place re-encodes +} +message EcShardChecksums { + uint32 shard_id = 1; // 0..MaxShardCount-1 (custom EC ratios go up to 32) + int64 covered_size = 2; // shard byte length these checksums cover (must equal the on-disk shard length) + bytes block_crc32c = 3; // packed little-endian uint32[] = ceil(covered_size/block_size) entries +} +enum ChecksumAlgorithm { + CHECKSUM_NONE = 0; + CHECKSUM_CRC32C = 1; +} message OldVersionVolumeInfo { repeated RemoteFile files = 1; uint32 version = 2; @@ -664,6 +688,7 @@ enum VolumeScrubMode { INDEX = 1; FULL = 2; LOCAL = 3; + CHECKSUM = 4; // EC only: verify each local shard's raw bytes against the bitrot checksum sidecar } message ScrubVolumeRequest { diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index e1535f7b1..1371a4820 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -37,6 +37,7 @@ fn scrub_mode_label(mode: i32) -> &'static str { 1 => "INDEX", 2 => "FULL", 3 => "LOCAL", + 4 => "CHECKSUM", _ => "UNKNOWN", } } @@ -2765,6 +2766,58 @@ impl VolumeServer for VolumeGrpcService { } } + // Copy the generation-0 bitrot checksum sidecar (.ecsum) when requested, so + // protection travels with the shards. Tolerant of a missing source (no-op): + // this non-2PC path has no Prepare backstop, and an unprotected source + // simply leaves the copy unprotected. + if req.copy_ecsum_file { + let copy_req = volume_server_pb::CopyFileRequest { + volume_id: req.volume_id, + collection: req.collection.clone(), + is_ec_volume: true, + ext: ".ecsum".to_string(), + compaction_revision: u32::MAX, + stop_offset: i64::MAX as u64, + ignore_source_file_not_found: true, + ..Default::default() + }; + let mut stream = client + .copy_file(copy_req) + .await + .map_err(|e| { + Status::internal(format!( + "VolumeEcShardsCopy volume {} copy .ecsum: {}", + vid, e + )) + })? + .into_inner(); + + let file_path = { + let base = + crate::storage::volume::volume_file_name(&dest_dir, &req.collection, vid); + format!("{}.ecsum", base) + }; + let mut file = std::fs::File::create(&file_path) + .map_err(|e| Status::internal(format!("create {}: {}", file_path, e)))?; + let mut written: u64 = 0; + while let Some(chunk) = stream + .message() + .await + .map_err(|e| Status::internal(format!("recv .ecsum: {}", e)))? + { + use std::io::Write; + file.write_all(&chunk.file_content) + .map_err(|e| Status::internal(format!("write {}: {}", file_path, e)))?; + written += chunk.file_content.len() as u64; + } + // A missing source yields an empty stream; drop the 0-byte file so mount + // sees no sidecar (protection Off) rather than a truncated/invalid one. + if written == 0 { + drop(file); + let _ = std::fs::remove_file(&file_path); + } + } + Ok(Response::new( volume_server_pb::VolumeEcShardsCopyResponse {}, )) @@ -3911,7 +3964,7 @@ impl VolumeServer for VolumeGrpcService { // Validate mode let mode = req.mode; match mode { - 1 | 2 | 3 => {} // INDEX=1, FULL=2, LOCAL=3 + 1 | 2 | 3 | 4 => {} // INDEX=1, FULL=2, LOCAL=3, CHECKSUM=4 _ => { return Err(Status::invalid_argument(format!( "unsupported EC volume scrub mode {}", @@ -4058,6 +4111,36 @@ impl VolumeServer for VolumeGrpcService { } } } + 4 => { + // CHECKSUM: verify each local shard's raw bytes against the + // bitrot checksum sidecar, exercising cold parity shards. + // Read-only. Mirrors Go's v.ChecksumScrub(). + let (blocks_scanned, broken, errs, collection) = { + let store = self.state.store.read().unwrap(); + let ecv = store.find_ec_volume(vid).ok_or_else(|| { + Status::not_found(format!("EC volume id {} not found", vid.0)) + })?; + let collection = ecv.collection.clone(); + let (blocks, broken, errs) = ecv.checksum_scrub(); + (blocks, broken, errs, collection) + }; + total_volumes += 1; + total_files += blocks_scanned; + if !errs.is_empty() || !broken.is_empty() { + broken_volume_ids.push(vid.0); + for b in broken { + broken_shard_infos.push(volume_server_pb::EcShardInfo { + volume_id: vid.0, + collection: collection.clone(), + shard_id: b, + ..Default::default() + }); + } + for msg in errs { + details.push(format!("ecvol {}: {}", vid.0, msg)); + } + } + } _ => unreachable!(), // validated above } } diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 86d7fbadd..a8458fb6b 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -401,6 +401,12 @@ impl DiskLocation { for i in 0..MAX_SHARD_COUNT { rm_if_present(format!("{}.ec{:02}", base, i))?; } + + // Remove the bitrot checksum sidecars from both the data and idx dirs. + remove_bitrot_sidecars(&base)?; + if self.idx_directory != self.directory { + remove_bitrot_sidecars(&idx_base)?; + } Ok(()) } @@ -1089,6 +1095,49 @@ fn rm_if_present(path: String) -> io::Result<()> { } } +/// Remove the bitrot checksum sidecars for a base file name: the legacy +/// `.ecsum` (generation 0) and any versioned `.ecsum.v`. +/// Already-gone is success; returns the first real removal failure (and surfaces +/// a directory-scan error) so a stale sidecar left behind is not reported as +/// cleaned. Mirrors Go's removeBitrotSidecars. +fn remove_bitrot_sidecars(base: &str) -> io::Result<()> { + use crate::storage::erasure_coding::ec_bitrot::BITROT_SIDECAR_EXT; + let rm = |path: std::path::PathBuf| -> io::Result<()> { + match fs::remove_file(&path) { + Err(e) if e.kind() != io::ErrorKind::NotFound => Err(e), + _ => Ok(()), + } + }; + let mut first_err: Option = None; + let mut record = |res: io::Result<()>| { + if let Err(e) = res { + if first_err.is_none() { + first_err = Some(e); + } + } + }; + record(rm(format!("{}{}", base, BITROT_SIDECAR_EXT).into())); + let path = std::path::Path::new(base); + if let (Some(parent), Some(fname)) = (path.parent(), path.file_name()) { + let prefix = format!("{}{}.v", fname.to_string_lossy(), BITROT_SIDECAR_EXT); + match fs::read_dir(parent) { + Ok(entries) => { + for entry in entries.flatten() { + if entry.file_name().to_string_lossy().starts_with(&prefix) { + record(rm(entry.path())); + } + } + } + Err(e) if e.kind() != io::ErrorKind::NotFound => record(Err(e)), + Err(_) => {} + } + } + match first_err { + Some(e) => Err(e), + None => Ok(()), + } +} + fn ec_data_shards_from_vif(directory: &str, idx_directory: &str, collection: &str, vid: VolumeId) -> usize { for dir in [directory, idx_directory] { let vif = format!("{}.vif", volume_file_name(dir, collection, vid)); diff --git a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs new file mode 100644 index 000000000..06a31167f --- /dev/null +++ b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs @@ -0,0 +1,827 @@ +//! EC bitrot detection — checksum sidecar. +//! +//! A per-volume sidecar file stores a CRC32C (Castagnoli) checksum for every +//! fixed-size block of every EC shard, so a scrub (and the reconstruction path) +//! can detect silent disk corruption in any shard — including cold parity shards +//! that are never read during normal serving. +//! +//! The sidecar is OPTIONAL: an absent or generation-mismatched sidecar simply +//! means "feature off" for that generation, so old binaries, JSON-only nodes, +//! and rollback deployments ignore it and degrade gracefully. +//! +//! On-disk layout of `.ecsum` (legacy/generation 0) and `.ecsum.v`: +//! +//! ```text +//! [ magic(4) | format_version(2) | payload_len(4) | payload_crc32c(4) ] [ proto payload ] +//! ``` +//! +//! All header fields are BIG-ENDIAN. The header's `payload_crc32c` lets a loader +//! detect corruption of the sidecar itself BEFORE trusting any contents, so a +//! rotted sidecar can never be mistaken for shard corruption. This format is +//! byte-identical to the Go implementation in +//! `weed/storage/erasure_coding/ec_bitrot.go`. + +use std::fs::File; +use std::io::{self, Read, Write}; + +use prost::Message; + +use crate::pb::volume_server_pb::{ + ChecksumAlgorithm, EcBitrotProtection, EcShardChecksums, EcShardConfig, +}; +use crate::storage::erasure_coding::ec_shard::MAX_SHARD_COUNT; +use crate::storage::needle::crc::CRC; + +/// Canonical extension for the checksum sidecar. Generation 0 (legacy/fresh +/// encode) uses `.ecsum`; vacuum generation N uses `.ecsum.v`, +/// mirroring the `.vif`/`.ecx` versioned convention. +pub const BITROT_SIDECAR_EXT: &str = ".ecsum"; + +/// Default checksum granularity (16 MiB). It is a power-of-two multiple of +/// `ERASURE_CODING_SMALL_BLOCK_SIZE` (1 MiB) and keeps the sidecar tiny +/// (~11 KB for a 30 GB volume) while localizing corruption to a 16 MiB region. +pub const DEFAULT_BITROT_BLOCK_SIZE: usize = 16 * 1024 * 1024; + +/// Caps the block granularity so a loaded sidecar cannot force a huge +/// scrub/verify scratch buffer. Power-of-two multiple of 1 MiB. +pub const MAX_BITROT_BLOCK_SIZE: u32 = 64 * 1024 * 1024; + +/// Magic "ECSU". +pub const BITROT_MAGIC: u32 = 0x4543_5355; +/// On-disk format version. +pub const BITROT_FORMAT_VERSION: u16 = 1; +/// Header size: magic(4) + version(2) + payload_len(4) + payload_crc32c(4). +pub const BITROT_HEADER_SIZE: usize = 14; + +/// Resolved protection state of an EC volume's active generation after loading +/// and validating its sidecar. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BitrotStatus { + /// No sidecar, or a sidecar that does not describe the active generation. + /// The generation is unprotected; this is NOT corruption. + Off, + /// A complete, well-formed, generation-matching sidecar is loaded. + On, + /// A generation-matching sidecar that is malformed, incomplete, or + /// self-integrity-failed. The generation is unprotected pending repair of + /// the sidecar, and an integrity alarm should fire. The rebuild path treats + /// this as fail-closed. + Invalid, +} + +impl std::fmt::Display for BitrotStatus { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let s = match self { + BitrotStatus::On => "on", + BitrotStatus::Invalid => "invalid", + BitrotStatus::Off => "off", + }; + f.write_str(s) + } +} + +/// Error returned by [`load_bitrot_sidecar`]. A self-integrity failure is a +/// sidecar-integrity problem (the caller maps it to [`BitrotStatus::Invalid`]), +/// never a shard-corruption signal. +#[derive(Debug)] +pub enum BitrotLoadError { + /// File missing or other underlying I/O error. + Io(io::Error), + /// File shorter than the fixed header. + TooShort(usize), + /// Header magic did not match. + BadMagic(u32), + /// Header format version is unsupported. + UnsupportedVersion(u16), + /// Header payload_len disagrees with the actual payload length. + LengthMismatch { header: u32, actual: usize }, + /// Header payload_crc32c disagrees with the computed CRC32C of the payload. + CrcMismatch { header: u32, computed: u32 }, + /// Protobuf payload failed to decode. + Decode(prost::DecodeError), +} + +impl std::fmt::Display for BitrotLoadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BitrotLoadError::Io(e) => write!(f, "bitrot sidecar io error: {}", e), + BitrotLoadError::TooShort(n) => { + write!(f, "bitrot sidecar too short ({} bytes)", n) + } + BitrotLoadError::BadMagic(m) => write!(f, "bitrot sidecar bad magic {:#x}", m), + BitrotLoadError::UnsupportedVersion(v) => { + write!(f, "bitrot sidecar unsupported format version {}", v) + } + BitrotLoadError::LengthMismatch { header, actual } => write!( + f, + "bitrot sidecar length mismatch: header {}, actual {}", + header, actual + ), + BitrotLoadError::CrcMismatch { header, computed } => write!( + f, + "bitrot sidecar self-integrity CRC mismatch: header {:#x}, computed {:#x}", + header, computed + ), + BitrotLoadError::Decode(e) => write!(f, "unmarshal bitrot sidecar: {}", e), + } + } +} + +impl std::error::Error for BitrotLoadError {} + +impl From for BitrotLoadError { + fn from(e: io::Error) -> Self { + BitrotLoadError::Io(e) + } +} + +/// Returns the sidecar path for a base file name and EC generation. Generation +/// 0 is the un-suffixed legacy path; generation N>0 is the versioned path, +/// consistent with how `.vif`/`.ecx` are versioned by the 2PC switch. +pub fn bitrot_sidecar_path(base: &str, generation: u32) -> String { + if generation == 0 { + format!("{}{}", base, BITROT_SIDECAR_EXT) + } else { + format!("{}{}.v{}", base, BITROT_SIDECAR_EXT, generation) + } +} + +/// Returns a fresh random per-encode identity used to detect a stale sidecar +/// left behind by an in-place re-encode. +pub fn new_encode_uuid() -> Vec { + use rand::RngCore; + let mut b = vec![0u8; 16]; + rand::thread_rng().fill_bytes(&mut b); + b +} + +/// Reports whether `block_size` is a power of two in [1 MiB, MAX_BITROT_BLOCK_SIZE]. +pub fn is_pow2_multiple_of_1mib(block_size: u32) -> bool { + block_size >= (1 << 20) && block_size <= MAX_BITROT_BLOCK_SIZE && block_size.count_ones() == 1 +} + +/// Returns ceil(covered_size / block_size). +fn expected_block_count(covered_size: i64, block_size: i64) -> usize { + if block_size <= 0 { + return 0; + } + ((covered_size + block_size - 1) / block_size) as usize +} + +/// Packs a slice of u32 into little-endian bytes. +fn pack_u32_le(vals: &[u32]) -> Vec { + let mut out = Vec::with_capacity(vals.len() * 4); + for v in vals { + out.extend_from_slice(&v.to_le_bytes()); + } + out +} + +/// Unpacks little-endian bytes into a Vec. Trailing bytes (len % 4) are +/// ignored, matching the Go implementation's `len(b)/4` truncation. +fn unpack_u32_le(b: &[u8]) -> Vec { + let n = b.len() / 4; + let mut out = Vec::with_capacity(n); + for i in 0..n { + out.push(u32::from_le_bytes([ + b[i * 4], + b[i * 4 + 1], + b[i * 4 + 2], + b[i * 4 + 3], + ])); + } + out +} + +/// Accumulates the per-block CRC32C of a single shard's byte stream as it is +/// written. Tolerates arbitrary chunk sizes that cross block boundaries, so it +/// works for both small encode buffers and the larger rebuild buffers. +pub struct ShardChecksumBuilder { + block_size: i64, + cur: CRC, + cur_len: i64, + total: i64, + blocks: Vec, +} + +impl ShardChecksumBuilder { + /// Creates a builder over a given block size in bytes. + pub fn new(block_size: i64) -> Self { + ShardChecksumBuilder { + block_size, + cur: CRC(0), + cur_len: 0, + total: 0, + blocks: Vec::new(), + } + } + + /// Feeds a chunk of arbitrary size, splitting it across block boundaries. + pub fn write(&mut self, mut p: &[u8]) { + while !p.is_empty() { + let room = self.block_size - self.cur_len; + let n = (p.len() as i64).min(room) as usize; + self.cur = self.cur.update(&p[..n]); + self.cur_len += n as i64; + self.total += n as i64; + p = &p[n..]; + if self.cur_len == self.block_size { + self.blocks.push(self.cur.0); + self.cur = CRC(0); + self.cur_len = 0; + } + } + } + + /// Flushes any partial last block and returns the covered size and the + /// packed little-endian u32 CRC array. + pub fn finalize(mut self) -> (i64, Vec) { + if self.cur_len > 0 { + self.blocks.push(self.cur.0); + self.cur = CRC(0); + self.cur_len = 0; + } + (self.total, pack_u32_le(&self.blocks)) + } +} + +/// Atomically writes `prot` to `path`, wrapped in the on-disk header with a +/// CRC32C over the serialized payload (temp file + rename). +pub fn save_bitrot_sidecar(path: &str, prot: &EcBitrotProtection) -> io::Result<()> { + let payload = prot.encode_to_vec(); + let mut buf = Vec::with_capacity(BITROT_HEADER_SIZE + payload.len()); + buf.extend_from_slice(&BITROT_MAGIC.to_be_bytes()); + buf.extend_from_slice(&BITROT_FORMAT_VERSION.to_be_bytes()); + buf.extend_from_slice(&(payload.len() as u32).to_be_bytes()); + buf.extend_from_slice(&CRC::new(&payload).0.to_be_bytes()); + buf.extend_from_slice(&payload); + + let tmp = format!("{}.tmp", path); + { + let mut f = File::create(&tmp)?; + f.write_all(&buf)?; + f.sync_all()?; + } + if let Err(e) = std::fs::rename(&tmp, path) { + let _ = std::fs::remove_file(&tmp); + return Err(e); + } + Ok(()) +} + +/// Reads and self-integrity-checks a sidecar file. Returns the parsed message, +/// or an error if the file is missing, truncated, has a bad magic/version, or +/// fails the payload CRC. A self-integrity failure is a sidecar-integrity +/// problem (caller maps it to [`BitrotStatus::Invalid`]), never a shard +/// corruption signal. +pub fn load_bitrot_sidecar(path: &str) -> Result { + let mut data = Vec::new(); + File::open(path)?.read_to_end(&mut data)?; + if data.len() < BITROT_HEADER_SIZE { + return Err(BitrotLoadError::TooShort(data.len())); + } + let magic = u32::from_be_bytes([data[0], data[1], data[2], data[3]]); + if magic != BITROT_MAGIC { + return Err(BitrotLoadError::BadMagic(magic)); + } + let ver = u16::from_be_bytes([data[4], data[5]]); + if ver != BITROT_FORMAT_VERSION { + return Err(BitrotLoadError::UnsupportedVersion(ver)); + } + let payload_len = u32::from_be_bytes([data[6], data[7], data[8], data[9]]); + let want_crc = u32::from_be_bytes([data[10], data[11], data[12], data[13]]); + let payload = &data[BITROT_HEADER_SIZE..]; + if payload_len as usize != payload.len() { + return Err(BitrotLoadError::LengthMismatch { + header: payload_len, + actual: payload.len(), + }); + } + let got = CRC::new(payload).0; + if got != want_crc { + return Err(BitrotLoadError::CrcMismatch { + header: want_crc, + computed: got, + }); + } + EcBitrotProtection::decode(payload).map_err(BitrotLoadError::Decode) +} + +/// Performs the disk-free manifest/syntax checks that every loader runs: +/// supported algorithm, valid block size, exactly one entry per shard id in the +/// active layout (no duplicates, no out-of-range ids), positive covered_size, +/// and a packed-CRC count consistent with covered_size. +/// +/// It does NOT compare covered_size against on-disk shard lengths — that is a +/// per-node physical check done only for locally-held shards. +pub fn validate_manifest( + prot: &EcBitrotProtection, + data_shards: usize, + parity_shards: usize, +) -> Result<(), String> { + if prot.algorithm != ChecksumAlgorithm::ChecksumCrc32c as i32 { + return Err(format!("unsupported checksum algorithm {}", prot.algorithm)); + } + if !is_pow2_multiple_of_1mib(prot.block_size) { + return Err(format!( + "invalid block_size {} (must be a power-of-two multiple of 1 MiB, at most {})", + prot.block_size, MAX_BITROT_BLOCK_SIZE + )); + } + let bs = prot.block_size as i64; + let total = data_shards + parity_shards; + if total == 0 || total > MAX_SHARD_COUNT { + return Err(format!( + "invalid active layout: data={} parity={}", + data_shards, parity_shards + )); + } + if prot.shards.len() != total { + return Err(format!( + "incomplete manifest: {} shard entries, expected {}", + prot.shards.len(), + total + )); + } + let mut seen = vec![false; MAX_SHARD_COUNT]; + for s in &prot.shards { + if s.shard_id >= total as u32 { + return Err(format!( + "shard id {} out of range [0,{})", + s.shard_id, total + )); + } + if seen[s.shard_id as usize] { + return Err(format!("duplicate shard id {}", s.shard_id)); + } + seen[s.shard_id as usize] = true; + if s.covered_size <= 0 { + return Err(format!( + "shard {} has non-positive covered_size {}", + s.shard_id, s.covered_size + )); + } + let want_count = expected_block_count(s.covered_size, bs); + if s.block_crc32c.len() != want_count * 4 { + return Err(format!( + "shard {} crc count mismatch: {} bytes, expected {} (covered_size={} block_size={})", + s.shard_id, + s.block_crc32c.len(), + want_count * 4, + s.covered_size, + prot.block_size + )); + } + } + Ok(()) +} + +/// Resolves the protection status of a loaded-or-missing sidecar against the +/// active generation and layout. `loaded` is the result of attempting to load +/// the sidecar at the active generation's path. +/// +/// - missing sidecar (NotFound) => [`BitrotStatus::Off`] +/// - load/self-integrity failure => [`BitrotStatus::Invalid`] +/// - generation mismatch => [`BitrotStatus::Off`] +/// - manifest validation failure => [`BitrotStatus::Invalid`] +/// - otherwise => [`BitrotStatus::On`] +pub fn resolve_status( + loaded: &Result, + active_generation: u32, + data_shards: usize, + parity_shards: usize, +) -> BitrotStatus { + match loaded { + Err(BitrotLoadError::Io(e)) if e.kind() == io::ErrorKind::NotFound => BitrotStatus::Off, + Err(_) => BitrotStatus::Invalid, + Ok(prot) => { + if prot.generation != active_generation { + return BitrotStatus::Off; + } + if validate_manifest(prot, data_shards, parity_shards).is_err() { + return BitrotStatus::Invalid; + } + BitrotStatus::On + } + } +} + +/// Returns the [`EcShardChecksums`] entry for a shard id, or `None`. +pub fn shard_checksums(prot: &EcBitrotProtection, shard_id: u32) -> Option<&EcShardChecksums> { + prot.shards.iter().find(|s| s.shard_id == shard_id) +} + +/// Reads a shard file at `path` in `block_size` chunks and compares each block's +/// CRC32C against the manifest entry. Returns the list of mismatching block +/// indices (empty == clean), or a fatal `io::Error` for genuine I/O problems. +/// +/// A length mismatch (truncation or unexpected trailing bytes) is itself shard +/// corruption: every block index is reported as mismatched so the caller treats +/// the shard as bad. It does not interpret the result — the caller (scrub / +/// rebuild) arbitrates shard-vs-sidecar via Reed-Solomon before acting. +pub fn verify_shard_file_blocks( + path: &str, + entry: &EcShardChecksums, + block_size: i64, +) -> io::Result> { + let f = File::open(path)?; + let file_size = f.metadata()?.len() as i64; + let want = unpack_u32_le(&entry.block_crc32c); + + if file_size != entry.covered_size { + // Length drift is shard corruption: report every block as mismatched. + return Ok((0..want.len()).collect()); + } + + let mut mismatched = Vec::new(); + let mut buf = vec![0u8; block_size.max(1) as usize]; + let mut offset: i64 = 0; + for (i, want_crc) in want.iter().enumerate() { + let to_read = (entry.covered_size - offset).min(block_size); + if to_read <= 0 { + break; + } + let to_read = to_read as usize; + read_full_at(&f, &mut buf[..to_read], offset as u64)?; + if CRC::new(&buf[..to_read]).0 != *want_crc { + mismatched.push(i); + } + offset += to_read as i64; + } + Ok(mismatched) +} + +/// Reads exactly `buf.len()` bytes from `f` at `offset`, erroring on early EOF. +fn read_full_at(f: &File, buf: &mut [u8], offset: u64) -> io::Result<()> { + let mut total = 0usize; + while total < buf.len() { + #[cfg(unix)] + let n = { + use std::os::unix::fs::FileExt; + f.read_at(&mut buf[total..], offset + total as u64)? + }; + #[cfg(not(unix))] + let n = { + use std::io::{Read, Seek, SeekFrom}; + let mut fc = f.try_clone()?; + fc.seek(SeekFrom::Start(offset + total as u64))?; + fc.read(&mut buf[total..])? + }; + if n == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "short read on shard block", + )); + } + total += n; + } + Ok(()) +} + +/// Builds the `EcShardConfig` proto for the given layout. The bitrot sidecar +/// carries its own top-level encode_uuid, so the nested config leaves it empty. +pub fn ec_shard_config(data_shards: u32, parity_shards: u32) -> EcShardConfig { + EcShardConfig { + data_shards, + parity_shards, + encode_ts_ns: 0, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Cross-binary byte-exact proof: CANONICAL_HEX equals `canonicalInteropHex` + /// in weed/storage/erasure_coding/ec_bitrot_interop_test.go (the Go reference + /// asserts the same constant). Both binaries port the same format, so a + /// sidecar written by either must be byte-identical; this pins the Rust side + /// of that guarantee. If you change the format, regenerate and update BOTH. + #[test] + fn test_byte_exact_go_interop() { + const CANONICAL_HEX: &str = "45435355000100000039cc1b826a080110808080082204080a10042a0a108080401a04040302012a0c0801108080401a04080706053210000102030405060708090a0b0c0d0e0f"; + let prot = EcBitrotProtection { + algorithm: ChecksumAlgorithm::ChecksumCrc32c as i32, + block_size: DEFAULT_BITROT_BLOCK_SIZE as u32, + generation: 0, + ec_shard_config: Some(EcShardConfig { + data_shards: 10, + parity_shards: 4, + encode_ts_ns: 0, + }), + shards: vec![ + EcShardChecksums { + shard_id: 0, + covered_size: 1024 * 1024, + block_crc32c: pack_u32_le(&[0x0102_0304]), + }, + EcShardChecksums { + shard_id: 1, + covered_size: 1024 * 1024, + block_crc32c: pack_u32_le(&[0x0506_0708]), + }, + ], + encode_uuid: (0..16u8).collect(), + }; + let dir = std::env::temp_dir().join(format!("ecsum_interop_{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("v1.ecsum"); + let path = path.to_str().unwrap(); + save_bitrot_sidecar(path, &prot).unwrap(); + let bytes = std::fs::read(path).unwrap(); + let hex: String = bytes.iter().map(|b| format!("{:02x}", b)).collect(); + assert_eq!(hex, CANONICAL_HEX, "Rust .ecsum bytes drifted from the Go canonical form"); + let _ = std::fs::remove_file(path); + } + + #[test] + fn test_sidecar_path_generations() { + assert_eq!(bitrot_sidecar_path("/d/1", 0), "/d/1.ecsum"); + assert_eq!(bitrot_sidecar_path("/d/1", 3), "/d/1.ecsum.v3"); + } + + #[test] + fn test_is_pow2_multiple_of_1mib() { + assert!(is_pow2_multiple_of_1mib(1 << 20)); // 1 MiB + assert!(is_pow2_multiple_of_1mib(16 * 1024 * 1024)); // 16 MiB default + assert!(is_pow2_multiple_of_1mib(1 << 25)); + assert!(is_pow2_multiple_of_1mib(MAX_BITROT_BLOCK_SIZE)); // 64 MiB boundary + assert!(!is_pow2_multiple_of_1mib(0)); + assert!(!is_pow2_multiple_of_1mib(1 << 19)); // 512 KiB, too small + assert!(!is_pow2_multiple_of_1mib(3 << 20)); // 3 MiB, not pow2 + assert!(!is_pow2_multiple_of_1mib(128 * 1024 * 1024)); // pow2 but > MAX_BITROT_BLOCK_SIZE + assert!(!is_pow2_multiple_of_1mib(DEFAULT_BITROT_BLOCK_SIZE as u32 + 1)); + } + + #[test] + fn test_expected_block_count() { + assert_eq!(expected_block_count(0, 16), 0); + assert_eq!(expected_block_count(1, 16), 1); + assert_eq!(expected_block_count(16, 16), 1); + assert_eq!(expected_block_count(17, 16), 2); + assert_eq!(expected_block_count(32, 16), 2); + assert_eq!(expected_block_count(100, 0), 0); + } + + #[test] + fn test_pack_unpack_roundtrip() { + let vals = vec![0x0102_0304u32, 0xdead_beef, 0, u32::MAX]; + let packed = pack_u32_le(&vals); + assert_eq!(packed.len(), 16); + // Verify little-endian byte order of the first entry. + assert_eq!(&packed[0..4], &[0x04, 0x03, 0x02, 0x01]); + assert_eq!(unpack_u32_le(&packed), vals); + } + + #[test] + fn test_builder_block_boundaries() { + // block_size = 4; feed 10 bytes in chunks that cross boundaries. + let mut b = ShardChecksumBuilder::new(4); + let data = b"0123456789"; + b.write(&data[0..3]); // partial block 0 + b.write(&data[3..7]); // completes block 0 (idx 3), fills block 1 + b.write(&data[7..10]); // partial block 2 + let (covered, packed) = b.finalize(); + assert_eq!(covered, 10); + let crcs = unpack_u32_le(&packed); + // ceil(10/4) = 3 blocks + assert_eq!(crcs.len(), 3); + // Compare against direct per-block CRCs. + assert_eq!(crcs[0], CRC::new(&data[0..4]).0); + assert_eq!(crcs[1], CRC::new(&data[4..8]).0); + assert_eq!(crcs[2], CRC::new(&data[8..10]).0); + } + + #[test] + fn test_builder_exact_block_multiple() { + let mut b = ShardChecksumBuilder::new(4); + b.write(b"01234567"); // exactly 2 blocks, no partial + let (covered, packed) = b.finalize(); + assert_eq!(covered, 8); + assert_eq!(unpack_u32_le(&packed).len(), 2); + } + + #[test] + fn test_save_load_roundtrip() { + let tmp = tempfile::TempDir::new().unwrap(); + let path = tmp + .path() + .join("vol.ecsum") + .to_str() + .unwrap() + .to_string(); + + let mut builder = ShardChecksumBuilder::new(DEFAULT_BITROT_BLOCK_SIZE as i64); + builder.write(b"hello world"); + let (covered, packed) = builder.finalize(); + + let prot = EcBitrotProtection { + algorithm: ChecksumAlgorithm::ChecksumCrc32c as i32, + block_size: DEFAULT_BITROT_BLOCK_SIZE as u32, + generation: 0, + ec_shard_config: Some(ec_shard_config(10, 4)), + shards: vec![EcShardChecksums { + shard_id: 0, + covered_size: covered, + block_crc32c: packed, + }], + encode_uuid: new_encode_uuid(), + }; + + save_bitrot_sidecar(&path, &prot).unwrap(); + let loaded = load_bitrot_sidecar(&path).unwrap(); + assert_eq!(loaded, prot); + } + + #[test] + fn test_load_rejects_bad_magic() { + let tmp = tempfile::TempDir::new().unwrap(); + let path = tmp.path().join("bad.ecsum").to_str().unwrap().to_string(); + std::fs::write(&path, vec![0u8; BITROT_HEADER_SIZE + 4]).unwrap(); + match load_bitrot_sidecar(&path) { + Err(BitrotLoadError::BadMagic(_)) => {} + other => panic!("expected BadMagic, got {:?}", other), + } + } + + #[test] + fn test_load_rejects_corrupted_payload() { + let tmp = tempfile::TempDir::new().unwrap(); + let path = tmp + .path() + .join("corrupt.ecsum") + .to_str() + .unwrap() + .to_string(); + let prot = EcBitrotProtection { + algorithm: ChecksumAlgorithm::ChecksumCrc32c as i32, + block_size: DEFAULT_BITROT_BLOCK_SIZE as u32, + generation: 0, + ec_shard_config: Some(ec_shard_config(10, 4)), + shards: vec![EcShardChecksums { + shard_id: 0, + covered_size: 5, + block_crc32c: pack_u32_le(&[CRC::new(b"hello").0]), + }], + encode_uuid: vec![0u8; 16], + }; + save_bitrot_sidecar(&path, &prot).unwrap(); + + // Flip a byte in the payload (after the 14-byte header). + let mut data = std::fs::read(&path).unwrap(); + let last = data.len() - 1; + data[last] ^= 0xff; + std::fs::write(&path, &data).unwrap(); + + match load_bitrot_sidecar(&path) { + Err(BitrotLoadError::CrcMismatch { .. }) => {} + other => panic!("expected CrcMismatch, got {:?}", other), + } + } + + #[test] + fn test_load_missing_is_notfound() { + let res = load_bitrot_sidecar("/nonexistent/path/x.ecsum"); + match res { + Err(BitrotLoadError::Io(e)) => assert_eq!(e.kind(), io::ErrorKind::NotFound), + other => panic!("expected Io(NotFound), got {:?}", other), + } + } + + fn good_manifest() -> EcBitrotProtection { + let mut shards = Vec::new(); + for id in 0..14u32 { + shards.push(EcShardChecksums { + shard_id: id, + covered_size: 5, + block_crc32c: pack_u32_le(&[CRC::new(b"hello").0]), + }); + } + EcBitrotProtection { + algorithm: ChecksumAlgorithm::ChecksumCrc32c as i32, + block_size: DEFAULT_BITROT_BLOCK_SIZE as u32, + generation: 0, + ec_shard_config: Some(ec_shard_config(10, 4)), + shards, + encode_uuid: vec![0u8; 16], + } + } + + #[test] + fn test_validate_manifest_ok() { + assert!(validate_manifest(&good_manifest(), 10, 4).is_ok()); + } + + #[test] + fn test_validate_rejects_wrong_algorithm() { + let mut m = good_manifest(); + m.algorithm = ChecksumAlgorithm::ChecksumNone as i32; + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_validate_rejects_bad_block_size() { + let mut m = good_manifest(); + m.block_size = 3 << 20; + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_validate_rejects_incomplete() { + let mut m = good_manifest(); + m.shards.pop(); + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_validate_rejects_duplicate_shard_id() { + let mut m = good_manifest(); + m.shards[1].shard_id = 0; + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_validate_rejects_out_of_range_id() { + let mut m = good_manifest(); + m.shards[13].shard_id = 14; + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_validate_rejects_nonpositive_covered_size() { + let mut m = good_manifest(); + m.shards[0].covered_size = 0; + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_validate_rejects_crc_count_mismatch() { + let mut m = good_manifest(); + m.shards[0].block_crc32c = vec![0u8; 8]; // 2 entries but covered_size=5 => want 1 + assert!(validate_manifest(&m, 10, 4).is_err()); + } + + #[test] + fn test_resolve_status() { + // Missing => Off. + let notfound: Result = Err(BitrotLoadError::Io( + io::Error::new(io::ErrorKind::NotFound, "x"), + )); + assert_eq!(resolve_status(¬found, 0, 10, 4), BitrotStatus::Off); + + // Integrity failure => Invalid. + let bad: Result = + Err(BitrotLoadError::BadMagic(0)); + assert_eq!(resolve_status(&bad, 0, 10, 4), BitrotStatus::Invalid); + + // Generation mismatch => Off. + let mut m = good_manifest(); + m.generation = 7; + let ok: Result = Ok(m); + assert_eq!(resolve_status(&ok, 0, 10, 4), BitrotStatus::Off); + + // Matching + valid => On. + let ok2: Result = Ok(good_manifest()); + assert_eq!(resolve_status(&ok2, 0, 10, 4), BitrotStatus::On); + + // Matching generation but invalid manifest => Invalid. + let mut bad_m = good_manifest(); + bad_m.shards.pop(); + let ok3: Result = Ok(bad_m); + assert_eq!(resolve_status(&ok3, 0, 10, 4), BitrotStatus::Invalid); + } + + #[test] + fn test_verify_shard_file_blocks() { + let tmp = tempfile::TempDir::new().unwrap(); + let path = tmp.path().join("s.ec00").to_str().unwrap().to_string(); + let block_size: i64 = 4; + let data = b"0123456789"; // 10 bytes, 3 blocks + std::fs::write(&path, data).unwrap(); + + let mut b = ShardChecksumBuilder::new(block_size); + b.write(data); + let (covered, packed) = b.finalize(); + let entry = EcShardChecksums { + shard_id: 0, + covered_size: covered, + block_crc32c: packed, + }; + + // Clean file => no mismatches. + let mm = verify_shard_file_blocks(&path, &entry, block_size).unwrap(); + assert!(mm.is_empty()); + + // Corrupt block index 1 (bytes 4..8). + let mut corrupt = data.to_vec(); + corrupt[5] ^= 0xff; + std::fs::write(&path, &corrupt).unwrap(); + let mm = verify_shard_file_blocks(&path, &entry, block_size).unwrap(); + assert_eq!(mm, vec![1]); + + // Truncation => all blocks mismatched. + std::fs::write(&path, b"012").unwrap(); + let mm = verify_shard_file_blocks(&path, &entry, block_size).unwrap(); + assert_eq!(mm, vec![0, 1, 2]); + } +} diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs index ad01f2012..c66e63c07 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs @@ -10,6 +10,12 @@ 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, ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE, +}; use crate::storage::erasure_coding::ec_shard::*; use crate::storage::idx; use crate::storage::types::*; @@ -52,12 +58,21 @@ pub fn write_ec_files( 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(); + // Encode in large blocks, then small blocks encode_dat_file( &dat_file, dat_size, &rs, &mut shards, + &mut builders, data_shards, parity_shards, )?; @@ -67,6 +82,42 @@ pub fn write_ec_files( 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, + )), + 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(()) } @@ -536,6 +587,7 @@ fn encode_dat_file( dat_size: i64, rs: &ReedSolomon, shards: &mut [EcVolumeShard], + builders: &mut [ShardChecksumBuilder], data_shards: usize, parity_shards: usize, ) -> io::Result<()> { @@ -553,6 +605,7 @@ fn encode_dat_file( large_block_size, rs, shards, + builders, data_shards, parity_shards, )?; @@ -572,6 +625,7 @@ fn encode_dat_file( small_block_size, rs, shards, + builders, data_shards, parity_shards, )?; @@ -589,6 +643,7 @@ fn encode_one_batch( block_size: usize, rs: &ReedSolomon, shards: &mut [EcVolumeShard], + builders: &mut [ShardChecksumBuilder], data_shards: usize, parity_shards: usize, ) -> io::Result<()> { @@ -642,9 +697,11 @@ fn encode_one_batch( ) })?; - // Write all shard buffers to files + // 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 buffers.iter().enumerate() { shards[i].write_all(buf)?; + builders[i].write(buf); } Ok(()) @@ -712,6 +769,84 @@ mod tests { 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, + None, + None, + 0, + Version::current(), + ) + .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).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 { diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index eff6d3047..e67418a8b 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -61,6 +61,13 @@ pub struct EcVolume { /// served from a shard of a different encode run is rejected (server- and /// client-side); 0 for a pre-feature volume, which is treated leniently. pub encode_ts_ns: i64, + /// Active-generation EC bitrot checksum sidecar (`.ecsum`), loaded and + /// validated at mount. `None` unless `bitrot_status == On`. + pub(crate) bitrot: Option, + /// Resolved protection state of the loaded sidecar. Cached alongside `bitrot` + /// so `bitrot_protection()` can return the `Off`/`Invalid` distinction without + /// re-reading, mirroring Go's `EcVolume.bitrotStatus`. + pub(crate) bitrot_status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus, } /// Locate the `.vif` for a (collection, vid) by preferring the data dir @@ -191,6 +198,8 @@ impl EcVolume { shard_locations_refresh_time: std::sync::Mutex::new(None), expire_at_sec, encode_ts_ns, + bitrot: None, + bitrot_status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus::Off, }; // Open .ecx file (sorted index) in read/write mode for in-place deletion marking. @@ -237,9 +246,189 @@ impl EcVolume { // Seed the in-memory deleted set from the journal. vol.load_deleted_needles_from_ecj()?; + // Load the generation-0 EC bitrot checksum sidecar (optional; best-effort). + vol.load_active_bitrot_sidecar(); + Ok(vol) } + /// Load the generation-0 checksum sidecar into `self.bitrot`/`self.bitrot_status`. + /// OSS only produces generation-0 (fresh-encode) sidecars, mirroring Go's + /// `loadActiveBitrotSidecar`. + fn load_active_bitrot_sidecar(&mut self) { + self.load_bitrot_for_generation(0); + } + + /// Load and validate the sidecar describing `generation`, setting + /// `self.bitrot`/`self.bitrot_status`. Absent or generation/config-mismatched + /// => `Off` (protection off, not corruption); self-integrity or manifest + /// failure => `Invalid` with a warning (protection off pending repair); usable + /// => `On`. Mirrors Go's `loadBitrotForGeneration`. + fn load_bitrot_for_generation(&mut self, generation: u32) { + use crate::storage::erasure_coding::ec_bitrot; + let base = self.base_name(); + let path = ec_bitrot::bitrot_sidecar_path(&base, generation); + let loaded = ec_bitrot::load_bitrot_sidecar(&path); + let status = ec_bitrot::resolve_status( + &loaded, + generation, + self.data_shards as usize, + self.parity_shards as usize, + ); + self.bitrot = None; + self.bitrot_status = status; + match status { + ec_bitrot::BitrotStatus::On => self.bitrot = loaded.ok(), + ec_bitrot::BitrotStatus::Off => {} + ec_bitrot::BitrotStatus::Invalid => { + tracing::warn!( + volume_id = self.volume_id.0, + path = %path, + generation, + "ec volume: bitrot sidecar present but invalid; protection off pending repair", + ); + } + } + } + + /// The active-generation bitrot protection AND its status (cached at mount), + /// mirroring Go's `EcVolume.BitrotProtection()`. Preserves the distinction + /// `checksum_scrub` needs: an absent/generation-mismatched sidecar is `Off` + /// (a clean no-op), a present-but-malformed one is `Invalid` (a real integrity + /// error). Returns `Some(prot)` only for `On`. + pub(crate) fn bitrot_protection( + &self, + ) -> ( + Option, + crate::storage::erasure_coding::ec_bitrot::BitrotStatus, + ) { + (self.bitrot.clone(), self.bitrot_status) + } + + /// Read-only EC bitrot checksum scrub of the LOCAL shards of this volume's + /// active generation. + /// + /// Loads the active-generation `.ecsum` sidecar and, for each locally-held + /// shard, reads the on-disk shard file in `block_size` chunks and compares + /// each block's CRC32C against the sidecar. Returns + /// `(blocks_scanned, mismatched_shards, errors)`. + /// + /// If more than `parity_shards` shards mismatch wholesale (i.e. every block + /// of those shards is wrong — the signature of a stale/wrong sidecar rather + /// than localized disk rot), the result is classified as a suspect sidecar: + /// `mismatched_shards` is cleared and an integrity note is added to `errors` + /// instead. A genuine multi-shard disk failure of that magnitude is + /// already unrecoverable, so treating it as a sidecar-integrity issue avoids + /// raising false shard-corruption alarms. + /// + /// This method NEVER deletes or mutates anything — it is purely diagnostic. + pub fn checksum_scrub(&self) -> (u64, Vec, Vec) { + use crate::storage::erasure_coding::ec_bitrot; + use crate::storage::erasure_coding::ec_bitrot::BitrotStatus; + + let mut errors: Vec = Vec::new(); + + // Resolve the active-generation protection AND its status, mirroring + // Go's `ChecksumScrub` (`prot, status := ecv.BitrotProtection()`): + // - BitrotOff => sidecars are OPTIONAL; an absent (or generation/ + // config-mismatched) sidecar simply means protection is not enabled + // for this generation. Return a CLEAN, EMPTY result — NOT an error — + // so legacy/intentionally-unprotected volumes are never reported + // broken. (Go: `case BitrotOff: return 0, nil, nil`.) + // - BitrotInvalid => the sidecar is PRESENT but malformed/unverifiable + // (self-integrity or manifest failure). That is the only status that + // yields an integrity error here. + // - BitrotOn => scan local shards against it. + let prot = match self.bitrot_protection() { + (_, BitrotStatus::Off) => { + // Unprotected generation: nothing to verify. Not an error. + return (0, Vec::new(), Vec::new()); + } + (_, BitrotStatus::Invalid) => { + return ( + 0, + Vec::new(), + vec![format!( + "EC volume {} bitrot sidecar is malformed/unverifiable (sidecar integrity)", + self.volume_id.0 + )], + ); + } + (Some(p), BitrotStatus::On) => p, + (None, BitrotStatus::On) => { + // Unreachable: BitrotOn always carries a loaded sidecar. Treat a + // missing payload defensively as protection off (clean no-op). + return (0, Vec::new(), Vec::new()); + } + }; + + let block_size = prot.block_size as i64; + let generation = prot.generation; + let base = self.base_name(); + + let mut blocks_scanned: u64 = 0; + let mut mismatched_shards: Vec = Vec::new(); + // Track shards whose blocks ALL mismatch (wholesale) to detect a + // stale/wrong sidecar. + let mut wholesale_mismatch = 0usize; + + for (i, slot) in self.shards.iter().enumerate() { + if slot.is_none() { + continue; // not local + } + let shard_id = i as u32; + let Some(entry) = ec_bitrot::shard_checksums(&prot, shard_id) else { + errors.push(format!( + "EC volume {} shard {} present but missing from sidecar manifest", + self.volume_id.0, shard_id + )); + continue; + }; + + // Resolve the on-disk shard file path for the active generation, + // mirroring EcVolumeShard::reopen_against_generation's convention. + let path = if generation == 0 { + format!("{}.ec{:02}", base, shard_id) + } else { + format!("{}.ec{:02}.v{}", base, shard_id, generation) + }; + + let expected_blocks = entry.block_crc32c.len() / 4; + match ec_bitrot::verify_shard_file_blocks(&path, entry, block_size) { + Ok(mismatched) => { + blocks_scanned += expected_blocks as u64; + if !mismatched.is_empty() { + mismatched_shards.push(shard_id); + if expected_blocks > 0 && mismatched.len() == expected_blocks { + wholesale_mismatch += 1; + } + } + } + Err(e) => { + errors.push(format!( + "EC volume {} shard {} scrub read error: {}", + self.volume_id.0, shard_id, e + )); + } + } + } + + // If more shards mismatch wholesale than parity can mask, the sidecar + // itself is the likely culprit (stale generation / wrong volume), so + // suppress the shard-corruption verdict and flag a sidecar-integrity + // issue instead. + if wholesale_mismatch > self.parity_shards as usize { + errors.push(format!( + "EC volume {}: {} shards mismatch wholesale (> {} parity); suspect stale/wrong sidecar, not shard corruption", + self.volume_id.0, wholesale_mismatch, self.parity_shards + )); + mismatched_shards.clear(); + } + + mismatched_shards.sort_unstable(); + (blocks_scanned, mismatched_shards, errors) + } + /// Walk the .ecj journal and populate `deleted_needles`. Called once /// from `new()` under exclusive ownership of the just-constructed /// EcVolume, so locking is not strictly required — but we take the @@ -1274,6 +1463,118 @@ mod tests { use super::*; use tempfile::TempDir; + /// Mounting an EC volume loads and validates its generation-0 `.ecsum` + /// sidecar, so `bitrot_protection()` reports `On` with the parsed manifest. + #[test] + fn test_mount_loads_bitrot_sidecar() { + use crate::storage::erasure_coding::ec_bitrot::BitrotStatus; + use crate::storage::needle_map::NeedleMapKind; + use crate::storage::volume::Volume; + + 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::current(), + ) + .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).unwrap(); + } + v.sync_to_disk().unwrap(); + v.close(); + crate::storage::erasure_coding::ec_encoder::write_ec_files(dir, dir, "", VolumeId(1), 10, 4) + .unwrap(); + + let vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + assert!( + vol.bitrot.is_some(), + "mount should load the generation-0 .ecsum sidecar" + ); + let (prot, status) = vol.bitrot_protection(); + assert_eq!(status, BitrotStatus::On); + assert_eq!(prot.unwrap().shards.len(), 14); + } + + /// CHECKSUM scrub verifies clean shards against the sidecar and flags a shard + /// whose bytes are corrupted after encode. + #[test] + fn test_checksum_scrub_clean_and_detects_corruption() { + use crate::storage::needle_map::NeedleMapKind; + use crate::storage::volume::Volume; + + 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::current(), + ) + .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).unwrap(); + } + v.sync_to_disk().unwrap(); + v.close(); + crate::storage::erasure_coding::ec_encoder::write_ec_files(dir, dir, "", VolumeId(1), 10, 4) + .unwrap(); + + let mut vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + for id in 0..14u8 { + vol.add_shard(EcVolumeShard::new(dir, "", VolumeId(1), id)) + .unwrap(); + } + + // Clean scrub: no mismatches, no errors, blocks scanned > 0. + let (scanned, broken, errs) = vol.checksum_scrub(); + assert!(errs.is_empty(), "unexpected scrub errors: {:?}", errs); + assert!(broken.is_empty(), "unexpected mismatches: {:?}", broken); + assert!(scanned > 0, "scrub should scan at least one block"); + + // Corrupt one byte of shard 3 on disk; re-scrub must report shard 3. + let shard3 = format!("{}/1.ec03", dir); + let mut bytes = std::fs::read(&shard3).unwrap(); + assert!(!bytes.is_empty()); + bytes[0] ^= 0xFF; + std::fs::write(&shard3, &bytes).unwrap(); + + let (_, broken2, _) = vol.checksum_scrub(); + assert!( + broken2.contains(&3), + "corrupted shard 3 should be flagged, got {:?}", + broken2 + ); + } + fn write_ecx_file( dir: &str, collection: &str, diff --git a/seaweed-volume/src/storage/erasure_coding/mod.rs b/seaweed-volume/src/storage/erasure_coding/mod.rs index b6c07b450..fd1dab928 100644 --- a/seaweed-volume/src/storage/erasure_coding/mod.rs +++ b/seaweed-volume/src/storage/erasure_coding/mod.rs @@ -3,6 +3,7 @@ //! Encodes a volume's .dat file into 10 data + 4 parity shards using //! Reed-Solomon erasure coding. Can reconstruct from any 10 of 14 shards. +pub mod ec_bitrot; pub mod ec_decoder; pub mod ec_encoder; pub mod ec_locate; diff --git a/weed/storage/erasure_coding/ec_bitrot_interop_test.go b/weed/storage/erasure_coding/ec_bitrot_interop_test.go new file mode 100644 index 000000000..df5ea15c5 --- /dev/null +++ b/weed/storage/erasure_coding/ec_bitrot_interop_test.go @@ -0,0 +1,60 @@ +package erasure_coding + +import ( + "encoding/hex" + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" +) + +// interopSample is the fully-deterministic sidecar shared with the Rust port's +// ec_bitrot.rs `sample()`. Both binaries must serialize it to identical bytes. +func interopSample() *volume_server_pb.EcBitrotProtection { + return &volume_server_pb.EcBitrotProtection{ + Algorithm: volume_server_pb.ChecksumAlgorithm_CHECKSUM_CRC32C, + BlockSize: uint32(DefaultBitrotBlockSize), + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: 10, + ParityShards: 4, + }, + Shards: []*volume_server_pb.EcShardChecksums{ + {ShardId: 0, CoveredSize: 1024 * 1024, BlockCrc32C: packUint32LE([]uint32{0x01020304})}, + {ShardId: 1, CoveredSize: 1024 * 1024, BlockCrc32C: packUint32LE([]uint32{0x05060708})}, + }, + EncodeUuid: []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}, + } +} + +// canonicalInteropHex is the exact on-disk bytes of interopSample(). The Rust +// port's ec_bitrot.rs test_byte_exact_go_interop pins the SAME constant, so a +// format change in EITHER binary (new always-serialized field, renumber, +// EcShardConfig default flip, protobuf canonicalization change) fails that +// binary's test instead of silently desyncing a Go-written sidecar from a +// Rust-written one. If you intentionally change the format, regenerate this +// (run with -v to print the hex) and update the Rust constant in lock-step. +const canonicalInteropHex = "45435355000100000039cc1b826a080110808080082204080a10042a0a108080401a04040302012a0c0801108080401a04080706053210000102030405060708090a0b0c0d0e0f" + +// TestBitrotSidecarBytes_RustInterop pins the on-disk bytes so a Go-side format +// drift fails here, and the Rust port asserts the same constant (cross-binary +// sidecar interop). +func TestBitrotSidecarBytes_RustInterop(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "v1.ecsum") + if err := SaveBitrotSidecar(path, interopSample()); err != nil { + t.Fatalf("save: %v", err) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if got := hex.EncodeToString(data); got != canonicalInteropHex { + t.Fatalf("bitrot sidecar bytes drifted from the cross-binary canonical form;\n got=%s\nwant=%s\n(regenerate and update the Rust constant in lock-step)", got, canonicalInteropHex) + } + + // Round-trips through the loader. + if _, err := LoadBitrotSidecar(path); err != nil { + t.Fatalf("load: %v", err) + } +}