diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 39346899e..7f6d8af20 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -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 { diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index a6f46a6a5..e75aa4cc9 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -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); + } } diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 4c6d6438f..32322cccc 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -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, vid: VolumeId, force_deleted_needles_check: bool, + do_read_recovery: bool, ) -> (i64, Vec, Vec) { // 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 = 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,