diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index bd708ce8c..b863cad09 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -3957,8 +3957,9 @@ impl VolumeServer for VolumeGrpcService { } } } - 2 | 3 => { - // FULL (2) / LOCAL (3): verify EC shard data + 2 => { + // FULL: Reed-Solomon parity verification over the local shards. + // (Cross-server needle verification arrives in a follow-up.) let files = ecv.walk_ecx_stats().map(|(f, _, _)| f).unwrap_or(0); // After cross-disk reconciliation, an EcVolume can @@ -4013,6 +4014,19 @@ impl VolumeServer for VolumeGrpcService { } } } + 3 => { + // LOCAL: verify each needle against the locally-held shards. + total_volumes += 1; + let (files, shard_infos, errs) = ecv.scrub_local(); + total_files += files; + if !errs.is_empty() || !shard_infos.is_empty() { + broken_volume_ids.push(vid.0); + broken_shard_infos.extend(shard_infos); + for msg in errs { + details.push(format!("ecvol {}: {}", vid.0, msg)); + } + } + } _ => unreachable!(), // validated above } } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_shard.rs b/seaweed-volume/src/storage/erasure_coding/ec_shard.rs index 6a6a8d6ea..5f246adcf 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_shard.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_shard.rs @@ -109,6 +109,17 @@ impl EcVolumeShard { self.ecd_file_size } + /// Protobuf descriptor for this shard. Mirrors Go's ToEcShardInfo. + pub fn to_ec_shard_info(&self) -> crate::pb::volume_server_pb::EcShardInfo { + crate::pb::volume_server_pb::EcShardInfo { + shard_id: self.shard_id as u32, + size: self.file_size(), + collection: self.collection.clone(), + volume_id: self.volume_id.0, + ..Default::default() + } + } + /// Close the shard file. pub fn close(&mut self) { if let Some(ref file) = self.ecd_file { diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 294927aa9..b6f1d6001 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -546,6 +546,26 @@ impl EcVolume { } /// Locate the EC shard intervals needed to read a needle. + /// Locate the EC shard intervals covering a needle at `actual_offset` whose + /// index size is `size`. Mirrors Go's EcVolume.LocateEcShardNeedleInterval. + pub fn locate_ec_shard_needle_interval( + &self, + actual_offset: i64, + size: Size, + ) -> Vec { + // shardSize = datFileSize / DataShards when known, else ecdFileSize - 1 + // (shards are padded to the small block size; the -1 avoids an + // off-by-one in the large-block row count). + let shard_size = if self.dat_file_size > 0 { + self.dat_file_size / self.data_shards as i64 + } else { + self.shard_file_size() - 1 + }; + // locate_data wants the on-disk size (header+body+checksum+timestamp+padding). + let actual = get_actual_size(size, self.version); + ec_locate::locate_data(actual_offset, Size(actual as i32), shard_size, self.data_shards) + } + pub fn locate_needle( &self, needle_id: NeedleId, @@ -559,25 +579,7 @@ impl EcVolume { return Ok(None); } - // Match Go's LocateEcShardNeedleInterval: shardSize = shard.ecdFileSize - 1 - // Shards are usually padded to ErasureCodingSmallBlockSize, so subtract 1 - // to avoid off-by-one in large block row count calculation. - // If datFileSize is known, use datFileSize / DataShards instead. - let shard_size = if self.dat_file_size > 0 { - self.dat_file_size / self.data_shards as i64 - } else { - self.shard_file_size() - 1 - }; - // Pass the actual on-disk size (header+body+checksum+timestamp+padding) - // to locate_data, matching Go: types.Size(needle.GetActualSize(size, version)) - let actual = get_actual_size(size, self.version); - let intervals = ec_locate::locate_data( - offset.to_actual_offset(), - Size(actual as i32), - shard_size, - self.data_shards, - ); - + let intervals = self.locate_ec_shard_needle_interval(offset.to_actual_offset(), size); Ok(Some((offset, size, intervals))) } @@ -734,6 +736,154 @@ impl EcVolume { crate::storage::idx::check_index_file(&mut ecx_file, self.ecx_file_size, self.version) } + /// ScrubLocal verifies each needle against the LOCAL shards only; it cannot + /// CRC-check a needle whose intervals span shards held on other servers. + /// Mirrors Go's EcVolume.ScrubLocal. Returns (rows walked, broken shards, errors). + pub fn scrub_local( + &self, + ) -> (u64, Vec, Vec) { + // Local scan also verifies the index. + let (_, mut errs) = self.scrub_index(); + + let mut broken_shards: HashSet = HashSet::new(); + let mut count: u64 = 0; + + let ecx_path = self.ecx_file_name(); + let mut ecx_file = match File::open(&ecx_path) { + Ok(f) => f, + Err(e) => { + errs.push(format!("open ECX file {}: {}", ecx_path, e)); + return (count, Vec::new(), errs); + } + }; + + // Reused across every needle/chunk to avoid a per-chunk allocation. + let mut chunk_buf: Vec = Vec::new(); + let walk = crate::storage::idx::walk_index_file(&mut ecx_file, 0, |id, offset, size| { + count += 1; + if size.is_tombstone() { + return Ok(()); + } + + let locations = self.locate_ec_shard_needle_interval(offset.to_actual_offset(), size); + // A needle is verifiable locally only if every shard it spans is local; + // when any is remote, skip the reassembly buffer entirely. + let has_remote_chunks = locations.iter().any(|iv| { + let (sid, _) = iv.to_shard_id_and_offset(self.data_shards); + self.shards.get(sid as usize).and_then(|s| s.as_ref()).is_none() + }); + let mut read: i64 = 0; + let mut data: Vec = if has_remote_chunks { + Vec::new() + } else { + Vec::with_capacity(get_actual_size(size, self.version) as usize) + }; + let mut local_shard_ids: Vec = Vec::new(); + + for (i, iv) in locations.iter().enumerate() { + let (sid, soffset) = iv.to_shard_id_and_offset(self.data_shards); + let ssize = iv.size; + let shard = match self.shards.get(sid as usize).and_then(|s| s.as_ref()) { + Some(s) => s, + None => { + // Shard is not local; we can't verify it without decoding. + read += ssize; + continue; + } + }; + local_shard_ids.push(sid); + + if soffset + ssize > shard.file_size() { + broken_shards.insert(sid); + errs.push(format!( + "local shard {} for needle {} is too short ({}), cannot read chunk {}/{}", + sid, + id.0, + shard.file_size(), + i + 1, + locations.len() + )); + continue; + } + + chunk_buf.resize(ssize as usize, 0); + match shard.read_at(&mut chunk_buf, soffset as u64) { + Err(e) => { + broken_shards.insert(sid); + errs.push(format!( + "failed to read chunk {}/{} for needle {} from local shard {} at offset {}: {}", + i + 1, + locations.len(), + id.0, + sid, + soffset, + e + )); + continue; + } + Ok(got) if got as i64 != ssize => { + broken_shards.insert(sid); + errs.push(format!( + "expected {} bytes for chunk {}/{} for needle {} from local shard {}, got {}", + ssize, + i + 1, + locations.len(), + id.0, + sid, + got + )); + continue; + } + Ok(_) => {} + } + + if !has_remote_chunks { + data.extend_from_slice(&chunk_buf); + } + read += ssize; + } + + local_shard_ids.sort_unstable(); + + let want = get_actual_size(size, self.version); + if read != want { + // Like Go, returning from the walk callback aborts the scan. + return Err(io::Error::new( + io::ErrorKind::Other, + format!( + "expected {} bytes for needle {} on volume {}, got {}", + want, id.0, self.volume_id.0, read + ), + )); + } + + // Only a fully-local needle can be reassembled and CRC-checked. + if !has_remote_chunks { + let mut n = Needle::default(); + if let Err(e) = n.read_bytes(&data, 0, size, self.version) { + errs.push(format!( + "needle {} on volume {}, shards {:?}: {}", + id.0, self.volume_id.0, local_shard_ids, e + )); + } + } + Ok(()) + }); + if let Err(e) = walk { + // Go appends the walk/callback error verbatim. + errs.push(e.to_string()); + } + + let mut broken: Vec = broken_shards + .iter() + .filter_map(|sid| self.shards.get(*sid as usize).and_then(|s| s.as_ref())) + .map(|s| s.to_ec_shard_info()) + .collect(); + broken.sort_by(|a, b| a.shard_id.cmp(&b.shard_id)); + + (count, broken, errs) + } + // ---- Deletion ---- /// Write `TOMBSTONE_FILE_SIZE` over the Size field of an existing .ecx @@ -1272,4 +1422,45 @@ mod tests { assert_eq!(vol.data_shards, 16); assert_eq!(vol.parity_shards, 4); } + + #[test] + fn test_scrub_local_skips_tombstones() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let entries = vec![(NeedleId(1), Offset::from_actual_offset(0), Size(-1))]; + write_ecx_file(dir, "", VolumeId(1), &entries); + + let vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + let (count, broken, errs) = vol.scrub_local(); + assert_eq!(count, 1); + assert!(broken.is_empty(), "{:?}", broken); + assert!(errs.is_empty(), "{:?}", errs); + } + + #[test] + fn test_scrub_local_clean_when_no_local_shards() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let entries = vec![(NeedleId(1), Offset::from_actual_offset(0), Size(100))]; + write_ecx_file(dir, "", VolumeId(1), &entries); + + // dat_file_size makes the shard-size math well-defined. + let vif = crate::storage::volume::VifVolumeInfo { + dat_file_size: 14000, + ..Default::default() + }; + let base = crate::storage::volume::volume_file_name(dir, "", VolumeId(1)); + std::fs::write( + format!("{}.vif", base), + serde_json::to_string_pretty(&vif).unwrap(), + ) + .unwrap(); + + // No shard files present: nothing to verify locally, so no errors. + let vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + let (count, broken, errs) = vol.scrub_local(); + assert_eq!(count, 1); + assert!(broken.is_empty(), "{:?}", broken); + assert!(errs.is_empty(), "{:?}", errs); + } }