mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-29 19:25:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

rust: support the READS scrub mode

Browse Source
This commit is contained in:
Chris Lu
2026-08-07 20:51:18 -07:00
parent 457277ec9a
commit 7dc534961c
3 changed files with 176 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/proto/volume_server.proto
+13 -1
View File
@@ -699,10 +699,22 @@ message FetchAndWriteNeedleResponse {
enum VolumeScrubMode {
UNKNOWN = 0;
// Perform a quick scrub of volume index contents only.
INDEX = 1;
// Performs a full scrub of the volume, reading and verifying every needle contained in it.
FULL = 2;
// Performs a local scrub of volume contents hosted in this server. For regular volumes,
// this operation is equivalent to FULL; for EC volumes, only needles fully contained
// in local shards are verified.
LOCAL = 3;
CHECKSUM = 4; // EC only: verify each local shard's raw bytes against the bitrot checksum sidecar
// Performs an EC checksum scrub, verifying each local shard's raw bytes against the bitrot
// checksum sidecar. Available only for EC volumes.
CHECKSUM = 4;
// Performs a full scrub of the volume, but attempting to reconstruct data from missing/damaged
// shards from other shards in the cluster when necessary. This check ensures that EC volume
// contents are readable by Seaweed, even on a degraded storage state, by exercising parity data.
// For regular volumes, this operation is equivalent to FULL.
READS = 5;
}
message ScrubVolumeRequest {
seaweed-volume/src/server/grpc_server.rs
+124 -5
View File
@@ -38,6 +38,7 @@ fn scrub_mode_label(mode: i32) -> &'static str {
2 => "FULL",
3 => "LOCAL",
4 => "CHECKSUM",
5 => "READS",
_ => "UNKNOWN",
}
}
@@ -3998,7 +3999,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 | 5 => {} // INDEX=1, FULL=2, LOCAL=3, READS=5
_ => {
return Err(Status::invalid_argument(format!(
"unsupported volume scrub mode {}",
@@ -4028,7 +4029,9 @@ impl VolumeServer for VolumeGrpcService {
.ok_or_else(|| Status::not_found(format!("volume id {} not found", vid.0)))?;
total_volumes += 1;
// INDEX mode (1) calls scrub_index; FULL (2) and LOCAL (3) call scrub
// INDEX mode (1) calls scrub_index; FULL (2), LOCAL (3) and
// READS (5) call scrub — they are all equivalent to FULL for
// regular volumes
let scrub_result = if mode == 1 {
v.scrub_index()
} else {
@@ -4097,7 +4100,7 @@ impl VolumeServer for VolumeGrpcService {
// Validate mode
let mode = req.mode;
match mode {
1 | 2 | 3 | 4 => {} // INDEX=1, FULL=2, LOCAL=3, CHECKSUM=4
1 | 2 | 3 | 4 | 5 => {} // INDEX=1, FULL=2, LOCAL=3, CHECKSUM=4, READS=5
_ => {
return Err(Status::invalid_argument(format!(
"unsupported EC volume scrub mode {}",
@@ -4180,8 +4183,13 @@ impl VolumeServer for VolumeGrpcService {
// (1) Per-needle local+remote walk (Go ScrubEcVolume parity).
let (files, mut shard_infos, mut errs) =
crate::server::store_ec::scrub_ec_volume_distributed(&self.state, vid, false)
.await;
crate::server::store_ec::scrub_ec_volume_distributed(
&self.state,
vid,
false,
false,
)
.await;
total_files += files as u64; // count comes from the needle walk only
// (2) Local parity check, gated on all-shards-local. Blocking RS
@@ -4274,6 +4282,35 @@ impl VolumeServer for VolumeGrpcService {
}
}
}
5 => {
// READS: the FULL per-needle walk, but reconstruct
// unreadable intervals from the remaining shards before
// flagging them, exercising parity data. Mirrors Go's
// READS mode.
{
let store = self.state.store.read().unwrap();
store.find_ec_volume(vid).ok_or_else(|| {
Status::not_found(format!("EC volume id {} not found", vid.0))
})?;
}
total_volumes += 1;
let (files, shard_infos, errs) =
crate::server::store_ec::scrub_ec_volume_distributed(
&self.state,
vid,
false,
true,
)
.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));
}
}
}
_ => unreachable!(), // validated above
}
}
@@ -5964,4 +6001,86 @@ mod tests {
assert!(vif.expire_at_sec >= before + ttl.to_seconds());
assert!(vif.expire_at_sec <= before + ttl.to_seconds() + 5);
}
async fn scrub_ec_volume_1(
service: &VolumeGrpcService,
mode: i32,
) -> volume_server_pb::ScrubEcVolumeResponse {
service
.scrub_ec_volume(Request::new(volume_server_pb::ScrubEcVolumeRequest {
mode,
volume_ids: vec![1],
force_deleted_needles_check: false,
}))
.await
.unwrap()
.into_inner()
}
#[tokio::test]
async fn test_scrub_ec_volume_reads_mode_reconstructs_missing_shard() {
let (service, _tmp) = make_local_service_with_volume("", None);
service
.volume_ec_shards_generate(Request::new(
volume_server_pb::VolumeEcShardsGenerateRequest {
volume_id: 1,
collection: String::new(),
},
))
.await
.unwrap();
service
.volume_ec_shards_mount(Request::new(
volume_server_pb::VolumeEcShardsMountRequest {
volume_id: 1,
collection: String::new(),
shard_ids: (0..14).collect(),
source_disk_type: String::new(),
recover_missing_index: false,
},
))
.await
.unwrap();
// Seed the shard-location cache so the scrub skips the master lookup.
// The explicit-gRPC-port form targets port 1, so remote reads fail fast
// with connection-refused instead of resolving to a live local port.
{
let store = service.state.store.read().unwrap();
let ecv = store.find_ec_volume(VolumeId(1)).unwrap();
let mut locs = ecv.shard_locations.write().unwrap();
for sid in 0u8..14 {
locs.insert(sid, vec!["127.0.0.1:255.1".to_string()]);
}
*ecv.shard_locations_refresh_time.lock().unwrap() =
Some(std::time::Instant::now());
}
// All shards local: FULL is clean.
let resp = scrub_ec_volume_1(&service, 2).await;
assert!(resp.broken_volume_ids.is_empty(), "{:?}", resp.details);
assert_eq!(resp.total_files, 1);
service
.volume_ec_shards_unmount(Request::new(
volume_server_pb::VolumeEcShardsUnmountRequest {
volume_id: 1,
shard_ids: vec![0],
encode_ts_ns: 0,
},
))
.await
.unwrap();
// With a shard unreadable, FULL flags it broken...
let resp = scrub_ec_volume_1(&service, 2).await;
assert_eq!(resp.broken_volume_ids, vec![1]);
assert!(resp.broken_shard_infos.iter().any(|s| s.shard_id == 0));
// ...while READS reconstructs the interval from the local survivors.
let resp = scrub_ec_volume_1(&service, 5).await;
assert!(resp.broken_volume_ids.is_empty(), "{:?}", resp.details);
assert!(resp.broken_shard_infos.is_empty());
assert_eq!(resp.total_files, 1);
}
}
seaweed-volume/src/server/store_ec.rs
+39 -11
View File
@@ -204,19 +204,22 @@ 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.
/// FULL/READS EC scrub: verify every needle's bytes across local AND remote
/// shards. Mirrors Go's `Store.ScrubEcVolume`. Returns (rows walked, broken
/// shards, errors). `force_deleted_needles_check` disables the benign
/// delete-state size-mismatch suppression. `do_read_recovery` (READS mode)
/// reconstructs unreadable intervals from the remaining shards — exercising
/// parity data — before flagging them; FULL reports genuine shard faults
/// rather than healing around them.
///
/// 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`,
/// `scrub_snapshot_under_lock` + lock-drop + `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,
do_read_recovery: 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.
@@ -340,8 +343,9 @@ pub async fn scrub_ec_volume_distributed(
}
};
// Read each interval local-then-remote WITHOUT reconstructing: we verify
// the shards are valid, we do not heal them. Locations refreshed above.
// Read each interval local-then-remote. FULL does NOT reconstruct — it
// verifies the shards are valid, it does not heal around them; READS
// falls back to reconstruction below. 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() {
@@ -371,10 +375,34 @@ pub async fn scrub_ec_volume_distributed(
// -> the delete-state suppression (mirrors Go's pre-zeroed buffer).
Ok((_, true)) => data.resize(data.len() + *ssize, 0),
Ok((buf, false)) => data.extend_from_slice(&buf),
Err(_) => {
Err(read_err) => {
let mut last_err = read_err;
if do_read_recovery {
// ...then reconstruct it from the other shards.
match recover_one_remote_ec_shard_interval(
state,
vid,
id,
*shard_id,
*shard_offset,
*ssize,
&locations,
data_shards,
total_shards - data_shards,
snapshot.encode_ts_ns,
)
.await
{
Ok(buf) => {
data.extend_from_slice(&buf);
continue;
}
Err(e) => last_err = e,
}
}
errs.push(format!(
"failed to read EC shard {} for needle {} on volume {} (interval {}/{})",
shard_id, id.0, vid.0, i + 1, n_intervals
"failed to read EC shard {} for needle {} on volume {} (interval {}/{}): {}",
shard_id, id.0, vid.0, i + 1, n_intervals, last_err
));
broken_shards.insert(
*shard_id,