mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-19 05:36:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(scrub): EC FULL scrub — distributed local+remote needle walk (#10149)

Browse Source 
* feat(ec): add scrub_ec_volume_distributed (FULL EC scrub, local+remote)

Ports Go's Store.ScrubEcVolume: walk the raw .ecx, verify every needle across
local AND remote shards without decoding (report faults, don't heal), with the
#10130 deleted-needle size-mismatch suppression gated on a force flag. Reuses
the read path's lock-drop + no-reconstruct read_remote_ec_shard_interval so no
!Send store guard is held across an .await.

Walks the unmasked index (scrub_snapshot_under_lock locates from the raw
(offset, size), not locate_needle) so logically-deleted-but-present needles are
still byte-verified, matching Go. Refreshes shard locations once up front and
hard-fails on a master-lookup error rather than retrying per needle.

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo

* feat(scrub): dispatch EC FULL (mode 2) to the distributed needle walk

FULL ran a local-only Reed-Solomon parity check; route it to the per-needle
local+remote walk instead, mirroring Go. The handler collects vids under a brief
lock then releases it: FULL self-locks per needle (it awaits remote reads),
INDEX/LOCAL re-acquire a brief lock. verify_ec_shards is retained but no longer
wired to a mode.

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo
This commit is contained in:
Chris Lu
2026-06-30 03:29:46 -07:00
committed by GitHub
parent 53087cb237
commit d18b85ef61
3 changed files with 277 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+49 -71
View File
@@ -3920,15 +3920,19 @@ impl VolumeServer for VolumeGrpcService {
}
}
let store = self.state.store.read().unwrap();
let vids: Vec<VolumeId> = if req.volume_ids.is_empty() {
store
.locations
.iter()
.flat_map(|loc| loc.ec_volumes().map(|(vid, _)| *vid))
.collect()
} else {
req.volume_ids.iter().map(|&id| VolumeId(id)).collect()
// Collect the volume ids under a brief lock, then release it: FULL (mode 2)
// reads remote shards and must not hold the !Send store guard across .await.
let vids: Vec<VolumeId> = {
let store = self.state.store.read().unwrap();
if req.volume_ids.is_empty() {
store
.locations
.iter()
.flat_map(|loc| loc.ec_volumes().map(|(vid, _)| *vid))
.collect()
} else {
req.volume_ids.iter().map(|&id| VolumeId(id)).collect()
}
};
let mut total_volumes: u64 = 0;
@@ -3937,17 +3941,17 @@ impl VolumeServer for VolumeGrpcService {
let mut broken_shard_infos: Vec<volume_server_pb::EcShardInfo> = Vec::new();
let mut details: Vec<String> = Vec::new();
for vid in &vids {
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();
for vid in vids {
match mode {
1 => {
// INDEX mode: check ecx index integrity only, no shard verification
// Matches Go's v.ScrubIndex() → idx.CheckIndexFile()
let (count, errs) = ecv.scrub_index();
// INDEX mode: check ecx index integrity only, no shard verification.
let (count, errs) = {
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))
})?;
ecv.scrub_index()
};
total_volumes += 1;
total_files += count;
if !errs.is_empty() {
@@ -3958,66 +3962,40 @@ impl VolumeServer for VolumeGrpcService {
}
}
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
// legitimately have ecv.dir != ecv.dir_idx (shards
// on one disk, .ecx / .ecj / .vif on a sibling).
// Use the EcVolume's own dirs rather than collapsing
// both args to find_ec_dir's single answer, otherwise
// read_ec_shard_config falls back to the wrong .vif
// location for split-disk volumes (#9252).
let dir = ecv.dir.clone();
let idx_dir = ecv.dir_idx.clone();
if dir.is_empty() {
continue;
}
total_volumes += 1;
total_files += files;
let (data_shards, parity_shards) =
crate::storage::erasure_coding::ec_volume::read_ec_shard_config(
&dir,
&idx_dir,
&collection,
*vid,
);
match crate::storage::erasure_coding::ec_encoder::verify_ec_shards(
&dir,
&collection,
*vid,
data_shards as usize,
parity_shards as usize,
) {
Ok((broken, msgs)) => {
if !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 msgs {
details.push(format!("ecvol {}: {}", vid.0, msg));
}
// FULL: verify every needle across local AND remote shards.
{
// Match Go: a missing requested volume is a hard error.
let store = self.state.store.read().unwrap();
if store.find_ec_volume(vid).is_none() {
return Err(Status::not_found(format!(
"EC volume id {} not found",
vid.0
)));
}
Err(e) => {
broken_volume_ids.push(vid.0);
details.push(format!("ecvol {}: scrub error: {}", vid.0, e));
}
total_volumes += 1;
let (files, shard_infos, errs) =
crate::server::store_ec::scrub_ec_volume_distributed(&self.state, vid, false)
.await;
total_files += files as u64;
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));
}
}
}
3 => {
// LOCAL: verify each needle against the locally-held shards.
let (files, shard_infos, errs) = {
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))
})?;
ecv.scrub_local()
};
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);
seaweed-volume/src/server/store_ec.rs
+222 -16
View File
@@ -44,7 +44,7 @@ use crate::server::grpc_client::{build_grpc_endpoint, parse_grpc_address, GRPC_M
use crate::server::request_id::outgoing_request_id_interceptor;
use crate::server::volume_server::{to_http_address, VolumeServerState};
use crate::storage::erasure_coding::ec_shard::ShardId;
use crate::storage::needle::needle::{get_actual_size, Needle};
use crate::storage::needle::needle::{get_actual_size, Needle, NeedleError};
use crate::storage::store_ec_reconcile::EcVolumeMissingIndex;
use crate::storage::types::*;
use crate::storage::volume::volume_file_name;
@@ -193,6 +193,188 @@ pub async fn read_ec_shard_needle_distributed(
Ok(Some(n))
}
/// FULL EC scrub: verify every needle's bytes across local AND remote shards,
/// without decoding (so genuine shard faults are reported rather than healed).
/// Mirrors Go's `Store.ScrubEcVolume`. Returns (rows walked, broken shards,
/// errors). `force_deleted_needles_check` disables the benign delete-state
/// size-mismatch suppression.
///
/// Shard locations are refreshed once up front. Each needle is then processed via
/// `scrub_snapshot_under_lock` + lock-drop + no-reconstruct `read_remote_ec_shard_interval`,
/// so no `!Send` store guard is held across an `.await`.
pub async fn scrub_ec_volume_distributed(
state: &Arc<VolumeServerState>,
vid: VolumeId,
force_deleted_needles_check: bool,
) -> (i64, Vec<crate::pb::volume_server_pb::EcShardInfo>, Vec<String>) {
// Phase A — under the Store read lock, run the index scrub and grab the
// paths/scalars + shard-location staleness; release the lock before any await.
let (ecx_path, collection, seed_errs, cached_locations, cache_refreshed_at, data_shards, total_shards) = {
let store = state.store.read().unwrap();
let ecv = match store.find_ec_volume(vid) {
Some(v) => v,
None => {
return (
0,
Vec::new(),
vec![format!("EC volume id {} not found", vid.0)],
)
}
};
// full scan means verifying the index as well
let (_, errs) = ecv.scrub_index();
// Bind to locals so the inner RwLock/Mutex guards drop before the block ends.
let cached_locations = ecv.shard_locations.read().unwrap().clone();
let cache_refreshed_at = *ecv.shard_locations_refresh_time.lock().unwrap();
let data_shards = ecv.data_shards as usize;
let total_shards = (ecv.data_shards + ecv.parity_shards) as usize;
(
ecv.ecx_file_name(),
ecv.collection.clone(),
errs,
cached_locations,
cache_refreshed_at,
data_shards,
total_shards,
)
};
let mut errs = seed_errs;
// Refresh the shard-location cache once up front (mirrors Go's
// cachedLookupEcShardLocations); a failed lookup is a hard error rather than
// a per-needle connect storm against a down master.
if needs_refresh(&cached_locations, cache_refreshed_at, data_shards, total_shards) {
match cached_lookup_ec_shard_locations(state, vid).await {
Ok(fresh) => write_back_shard_locations(state, vid, fresh),
Err(e) => {
return (
0,
Vec::new(),
vec![format!("failed to locate shard via master grpc: {}", e)],
)
}
}
}
// Walk the .ecx (private fd, no lock) for the row count + live (id, offset, size).
let mut count: i64 = 0;
let mut needles: Vec<(NeedleId, Offset, Size)> = Vec::new();
match fs::File::open(&ecx_path) {
Ok(mut f) => {
if let Err(e) = crate::storage::idx::walk_index_file(&mut f, 0, |id, offset, size| {
count += 1;
if !size.is_tombstone() {
needles.push((id, offset, size));
}
Ok(())
}) {
errs.push(format!("walk ECX file {}: {}", ecx_path, e));
}
}
Err(e) => errs.push(format!("open ECX file {}: {}", ecx_path, e)),
}
// reads for EC chunks can hit the same shard repeatedly, so dedupe broken shards
let mut broken_shards: HashMap<ShardId, crate::pb::volume_server_pb::EcShardInfo> = HashMap::new();
for (id, offset, size) in needles {
// Per-needle snapshot under the lock from the RAW .ecx (offset, size) so
// logically-deleted needles are still verified; lock dropped before await.
let snapshot = match scrub_snapshot_under_lock(state, vid, offset, size) {
Ok(Some(s)) => s,
Ok(None) => continue, // volume gone since the walk
Err(e) => {
errs.push(format!("needle {} on EC volume {}: {}", id.0, vid.0, e));
continue;
}
};
// Read each interval local-then-remote WITHOUT reconstructing: we verify
// the shards are valid, we do not heal them. Locations refreshed above.
let n_intervals = snapshot.intervals.len();
let mut data: Vec<u8> = Vec::with_capacity(snapshot.actual_size);
for (i, res) in snapshot.intervals.iter().enumerate() {
match res {
IntervalResult::Local(buf) => data.extend_from_slice(buf),
IntervalResult::NeedRemote {
shard_id,
shard_offset,
size: ssize,
} => {
let sources = snapshot
.cached_locations
.get(shard_id)
.cloned()
.unwrap_or_default();
match read_remote_ec_shard_interval(
state,
&sources,
vid,
id,
*shard_id,
*shard_offset,
*ssize,
snapshot.encode_ts_ns,
)
.await
{
Ok(buf) => data.extend_from_slice(&buf),
Err(_) => {
errs.push(format!(
"failed to read EC shard {} for needle {} on volume {} (interval {}/{})",
shard_id, id.0, vid.0, i + 1, n_intervals
));
broken_shards.insert(
*shard_id,
crate::pb::volume_server_pb::EcShardInfo {
shard_id: *shard_id as u32,
size: *ssize as i64,
collection: collection.clone(),
volume_id: vid.0,
..Default::default()
},
);
break;
}
}
}
}
}
// Also fires when a chunk read broke out above (data is short).
if data.len() != snapshot.actual_size {
errs.push(format!(
"expected {} bytes for needle {}, got {}",
snapshot.actual_size,
id.0,
data.len()
));
continue;
}
let mut n = Needle::default();
if let Err(e) = n.read_bytes(&data, 0, snapshot.size_for_parse, snapshot.version) {
// A delete-state disagreement between the index and the reassembled
// header (live index vs zero header size) is not corruption.
let delete_state_disagrees = matches!(
&e,
NeedleError::SizeMismatch { found, .. }
if snapshot.size_for_parse.is_deleted() != (found.0 == 0)
);
if !delete_state_disagrees || force_deleted_needles_check {
errs.push(format!("needle {} on EC volume {}: {}", id.0, vid.0, e));
}
}
}
// Mirror Go CmpEcShardInfo: sort by (volume_id, shard_id).
let mut broken: Vec<crate::pb::volume_server_pb::EcShardInfo> =
broken_shards.into_values().collect();
broken.sort_by(|a, b| a.volume_id.cmp(&b.volume_id).then(a.shard_id.cmp(&b.shard_id)));
(count, broken, errs)
}
fn snapshot_under_lock(
state: &Arc<VolumeServerState>,
vid: VolumeId,
@@ -207,11 +389,44 @@ fn snapshot_under_lock(
// Reuse EcVolume::locate_needle for offset/size resolution AND
// the per-needle shard-interval math — it's the same routine the
// local-only read path uses, so we stay byte-identical on the
// shard-size + interval boundaries.
// shard-size + interval boundaries. locate_needle applies the runtime
// delete mask, which is correct for serving reads.
let (offset, size, intervals) = match ecv.locate_needle(needle_id)? {
Some(v) => v,
None => return Ok(None),
};
build_snapshot(ecv, offset, size, &intervals).map(Some)
}
/// Like `snapshot_under_lock`, but locates intervals from the RAW .ecx
/// (offset, size) the FULL-scrub walk supplies — NOT `locate_needle`, which
/// masks runtime-deleted needles. EC deletes are logical (the shard bytes stay
/// until re-encode), so the scrub must still byte-verify them, matching Go's
/// `Store.ScrubEcVolume` which walks the unmasked index.
fn scrub_snapshot_under_lock(
state: &Arc<VolumeServerState>,
vid: VolumeId,
offset: Offset,
size: Size,
) -> io::Result<Option<Snapshot>> {
let store = state.store.read().unwrap();
let ecv = match store.find_ec_volume(vid) {
Some(v) => v,
None => return Ok(None),
};
let intervals = ecv.locate_ec_shard_needle_interval(offset.to_actual_offset(), size);
build_snapshot(ecv, offset, size, &intervals).map(Some)
}
/// Read any locally-held shard intervals and assemble the rest as `NeedRemote`,
/// then snapshot the scalars + shard-location cache so the caller can drop the
/// store lock before awaiting remote reads. Shared by the read and scrub paths.
fn build_snapshot(
ecv: &crate::storage::erasure_coding::ec_volume::EcVolume,
offset: Offset,
size: Size,
intervals: &[crate::storage::erasure_coding::ec_locate::Interval],
) -> io::Result<Snapshot> {
if intervals.is_empty() {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
@@ -220,25 +435,16 @@ fn snapshot_under_lock(
}
let actual = get_actual_size(size, ecv.version);
// Phase A.local: for each interval read the local shard if we
// hold it; otherwise fall through to remote. We accumulate the
// results in interval order so the assembly step is just a
// concat.
let mut interval_results = Vec::with_capacity(intervals.len());
for interval in &intervals {
for interval in intervals {
let (shard_id, shard_offset) = interval.to_shard_id_and_offset(ecv.data_shards);
let buf_size = interval.size as usize;
let local = ecv
.shards
.get(shard_id as usize)
.and_then(|s| s.as_ref());
let local = ecv.shards.get(shard_id as usize).and_then(|s| s.as_ref());
match local {
Some(shard) => {
let mut buf = vec![0u8; buf_size];
match shard.read_at(&mut buf, shard_offset as u64) {
Ok(n) if n == buf_size => {
interval_results.push(IntervalResult::Local(buf));
}
Ok(n) if n == buf_size => interval_results.push(IntervalResult::Local(buf)),
_ => interval_results.push(IntervalResult::NeedRemote {
shard_id,
shard_offset,
@@ -257,7 +463,7 @@ fn snapshot_under_lock(
let cached_locations = ecv.shard_locations.read().unwrap().clone();
let cache_refreshed_at = *ecv.shard_locations_refresh_time.lock().unwrap();
Ok(Some(Snapshot {
Ok(Snapshot {
data_shards: ecv.data_shards,
parity_shards: ecv.parity_shards,
version: ecv.version,
@@ -268,7 +474,7 @@ fn snapshot_under_lock(
cached_locations,
cache_refreshed_at,
encode_ts_ns: ecv.encode_ts_ns,
}))
})
}
/// Master `LookupEcVolume` freshness rules — match Go's
seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
+6
View File
@@ -208,6 +208,12 @@ pub fn rebuild_ec_files(
}
/// Verify EC shards by computing parity against the existing data and identifying corrupted shards.
///
/// No longer wired to a scrub mode: FULL is now Go's per-needle local+remote walk
/// (`store_ec::scrub_ec_volume_distributed`). Retained as a standalone Reed-Solomon
/// parity check — the only parity/cold-region verification available until the
/// `.ecsum` bitrot CHECKSUM path lands.
#[allow(dead_code)]
pub fn verify_ec_shards(
dir: &str,
collection: &str,