diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 2151d54b4..74a15b680 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -52,6 +52,71 @@ fn unix_now_seconds() -> f64 { /// Record scrub metrics. `broken_shards` is `Some` only for EC scrubs so the /// shard-failures family stays untouched on regular volume scrubs (matching Go). +/// How a scrub reacts to a volume that is not in the store when its turn comes. +/// +/// Nothing is held across the scan, so the volume set is free to change under a +/// node-wide scrub: the heartbeat drops a volume that reported an I/O error and +/// expires an EC volume whose destroy time passed, and a delete or an unmount +/// can land between any two volumes. Failing there would discard every result +/// gathered so far and leave every later volume unscrubbed. +/// +/// `Some(status)` means fail the request; `None` means skip this volume and +/// keep going. A volume the caller NAMED is always the former — that one is a +/// caller mistake, not a concurrent change, and callers match on the message. +fn scrub_vanished_volume(explicit: bool, kind: &str, vid: VolumeId) -> Option { + if explicit { + return Some(Status::not_found(format!("{} id {} not found", kind, vid.0))); + } + tracing::info!( + volume_id = vid.0, + "scrub: {} {} is no longer mounted, skipping", + kind, + vid.0 + ); + None +} + +/// How a scrub reacts to a `spawn_blocking` scan that failed to join. +/// +/// The scan runs on the blocking pool, so a panic in it reaches the loop as a +/// `JoinError` instead of unwinding here. Propagating it would discard every +/// volume already scanned AND skip `emit_scrub_metrics`: a panic scrubbing one +/// volume would hide real corruption found on the others and leave the +/// staleness alert firing with nothing recorded to explain it. Record it +/// against this volume and let the loop finish, the same way `scrub_volumes` +/// treats a per-volume scrub error. +/// +/// A panic is evidence about the volume, so it counts as broken. A join error +/// from runtime shutdown is not -- the volume was never scanned, and counting +/// it would put a false corruption into SCRUB_VOLUME_FAILURES. +fn record_scrub_join_failure( + e: &tokio::task::JoinError, + vid: VolumeId, + what: &str, + broken_volume_ids: &mut Vec, + details: &mut Vec, +) { + if e.is_panic() { + tracing::error!( + volume_id = vid.0, + "scrub: {} task for EC volume {} panicked: {}", + what, + vid.0, + e + ); + broken_volume_ids.push(vid.0); + details.push(format!("ecvol {}: {} task panicked: {}", vid.0, what, e)); + } else { + tracing::info!( + volume_id = vid.0, + "scrub: {} task for EC volume {} was cancelled, skipping", + what, + vid.0 + ); + details.push(format!("ecvol {}: {} task cancelled; skipped", vid.0, what)); + } +} + fn emit_scrub_metrics(mode: i32, broken_volumes: usize, broken_shards: Option) { let mode_label = scrub_mode_label(mode); crate::metrics::SCRUB_LAST_TIME_SECONDS @@ -342,6 +407,408 @@ impl VolumeGrpcService { self.notify_master_volume_readonly(&info, true).await?; Ok(()) } + + /// The scrub loop over an already-resolved volume list. + /// + /// Split out of `scrub_volume` so a test can hand it a list containing an + /// id that is not in the store: listing and lookup read the same map under + /// the same lock, so nothing outside can make an id vanish between them. + async fn scrub_volumes( + &self, + req: &volume_server_pb::ScrubVolumeRequest, + vids: Vec, + explicit: bool, + ) -> Result, Status> { + let mode = req.mode; + let mut total_volumes: u64 = 0; + let mut total_files: u64 = 0; + let mut broken_volume_ids: Vec = Vec::new(); + let mut details: Vec = Vec::new(); + let mut broken_vids: Vec = Vec::new(); + + // Scrub phase. The store read guard is taken PER VOLUME, never across the + // whole loop. + // + // Holding one guard for the entire loop meant a node-wide scrub pinned the + // store lock for the full scan of every volume it holds. The periodic + // heartbeat takes store.write(); once that writer is pending, a + // write-preferring std::sync::RwLock queues every later reader behind it, + // so every HTTP handler blocks and the node stops serving and + // heart-beating until the scrub finishes. Re-acquiring per volume lets the + // writer land between volumes. + // + // NOTE: v.scrub() still runs under the guard for the duration of ONE + // volume, which for a large volume is still a long hold. In + // scrub_ec_volume the INDEX, LOCAL and CHECKSUM arms snapshot a plan and + // scan with the guard released; FULL/READS releases it across the index + // walk but still re-takes it per needle, in + // store_ec::scrub_snapshot_under_lock. The same treatment here needs + // Volume to expose an equivalent plan and is left as a follow-up. + for vid in &vids { + // Re-resolve under a fresh guard each iteration; the volume set can + // legitimately change between volumes now that the lock is released. + let Some((scrub_result, file_count)) = ({ + let store = self.state.store.read().unwrap(); + store.find_volume(*vid).map(|(_, v)| { + // INDEX mode (1) calls scrub_index; FULL (2) and LOCAL (3) call scrub + let r = if mode == 1 { v.scrub_index() } else { v.scrub() }; + (r, v.file_count()) + }) + }) else { + if let Some(status) = scrub_vanished_volume(explicit, "volume", *vid) { + return Err(status); + } + continue; + }; + total_volumes += 1; + + match scrub_result { + Ok((files, broken)) => { + total_files += files; + if !broken.is_empty() { + broken_vids.push(*vid); + broken_volume_ids.push(vid.0); + for msg in broken { + details.push(format!("vol {}: {}", vid.0, msg)); + } + } + } + Err(e) => { + total_files += file_count.max(0) as u64; + broken_vids.push(*vid); + broken_volume_ids.push(vid.0); + details.push(format!("vol {}: scrub error: {}", vid.0, e)); + } + } + } + + // Match Go: if mark_broken_volumes_readonly, call makeVolumeReadonly on each broken volume. + // Collect errors via errors.Join semantics (return joined error if any fail). + let mut errs: Vec = Vec::new(); + if req.mark_broken_volumes_readonly { + for vid in &broken_vids { + match self.make_volume_readonly(*vid, false, true).await { + Ok(()) => { + details.push(format!("volume {} is now read-only", vid.0)); + } + // The volume was scrubbed, found broken, and then removed — + // the same concurrent teardown the scrub loop tolerates, one + // step later. There is nothing left to mark read-only, and + // failing here would throw away the whole scrub report. Go + // never reaches this at all: it passes makeVolumeReadonly the + // *storage.Volume it already holds, so a volume that left the + // store is still addressable there. + Err(e) if e.code() == tonic::Code::NotFound => { + tracing::info!( + volume_id = vid.0, + "scrub: volume {} vanished before mark-readonly, skipping", + vid.0 + ); + details.push(format!( + "volume {} vanished before mark-readonly; skipped", + vid.0 + )); + } + Err(e) => { + errs.push(e.message().to_string()); + details.push(e.message().to_string()); + } + } + } + } + + // Record metrics before the post-scrub error check so scrub failures are + // persisted even when a follow-up admin action (mark-readonly) fails. + emit_scrub_metrics(mode, broken_vids.len(), None); + + if !errs.is_empty() { + return Err(Status::internal(errs.join("\n"))); + } + + Ok(Response::new(volume_server_pb::ScrubVolumeResponse { + total_volumes, + total_files, + broken_volume_ids, + details, + })) + } + + /// The EC scrub loop over an already-resolved volume list. Same split, and + /// same vanished-volume rule, as `scrub_volumes` — see its comment. + async fn scrub_ec_volumes( + &self, + req: &volume_server_pb::ScrubEcVolumeRequest, + vids: Vec, + explicit: bool, + ) -> Result, Status> { + let mode = req.mode; + let force_deleted_needles_check = req.force_deleted_needles_check; + let mut total_volumes: u64 = 0; + let mut total_files: u64 = 0; + let mut broken_volume_ids: Vec = Vec::new(); + let mut broken_shard_infos: Vec = Vec::new(); + let mut details: Vec = Vec::new(); + + for vid in vids { + match mode { + 1 => { + // INDEX mode: check ecx index integrity only, no shard verification. + // Same shape as the CHECKSUM arm below: snapshot, release + // the store lock, then walk the index. + let Some(plan) = ({ + let store = self.state.store.read().unwrap(); + store.find_ec_volume(vid).map(|ecv| ecv.scrub_index_plan()) + }) else { + if let Some(status) = scrub_vanished_volume(explicit, "EC volume", vid) { + return Err(status); + } + continue; + }; + // Counted as attempted BEFORE the join, so a failed join + // cannot silently shrink total_volumes. + total_volumes += 1; + let (count, errs) = match tokio::task::spawn_blocking(move || plan.run()).await + { + Ok(v) => v, + Err(e) => { + record_scrub_join_failure( + &e, + vid, + "index scrub", + &mut broken_volume_ids, + &mut details, + ); + continue; + } + }; + total_files += count; + if !errs.is_empty() { + broken_volume_ids.push(vid.0); + for msg in errs { + details.push(format!("ecvol {}: {}", vid.0, msg)); + } + } + } + 2 | 5 => { + // FULL/READS: Go-parity per-needle local+remote walk, PLUS a TEMPORARY + // local Reed-Solomon parity check. The needle walk only reads + // DATA-shard intervals of LIVE needles, so on its own it can't + // catch silent bitrot in a PARITY shard or an unwalked cold + // region. Go closes that gap with a separate CHECKSUM mode over + // .ecsum, which Rust does not have yet; running both here is a + // deliberate divergence from Go FULL to preserve coverage. Drop + // verify_ec_shards from this arm once mode 4 (CHECKSUM) lands. + // + // The RS recompute needs every shard co-located, so only run it + // when this node holds all data+parity shards (single-node EC); + // on a distributed layout it would report every non-local shard + // as missing. Snapshot under a brief lock; release before await. + let Some((dir, collection, data_shards, parity_shards, all_local)) = ({ + let store = self.state.store.read().unwrap(); + store.find_ec_volume(vid).map(|ecv| { + let total = (ecv.data_shards + ecv.parity_shards) as usize; + let local = ecv.shards.iter().filter(|s| s.is_some()).count(); + ( + ecv.dir.clone(), + ecv.collection.clone(), + ecv.data_shards as usize, + ecv.parity_shards as usize, + local == total, + ) + }) + }) else { + if let Some(status) = scrub_vanished_volume(explicit, "EC volume", vid) { + return Err(status); + } + continue; + }; + total_volumes += 1; + + // (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, + force_deleted_needles_check, + mode == 5, + ) + .await; + total_files += files as u64; // count comes from the needle walk only + + // (2) Local parity check, gated on all-shards-local. Blocking RS + // verify -> spawn_blocking; inputs are owned, no lock held. + if all_local && !dir.is_empty() { + let collection_pc = collection.clone(); + let join = tokio::task::spawn_blocking(move || { + crate::storage::erasure_coding::ec_encoder::verify_ec_shards( + &dir, + &collection_pc, + vid, + data_shards, + parity_shards, + ) + }) + .await; + // Unlike the other arms this must NOT `continue`: the + // needle walk above already produced findings for this + // volume, and dropping them here would recreate the bug + // this handles, one scope down. So a panic becomes an + // error for the volume, and a cancellation -- which + // spawn_blocking only reports when the runtime is going + // down, so this response is unlikely to reach anyone -- + // records in details that the parity half did not run + // rather than inventing a corruption for it. + let (parity_broken, parity_details) = match join { + Ok(r) => r.unwrap_or_else(|e| { + (Vec::new(), vec![format!("verify_ec_shards: {}", e)]) + }), + Err(e) if e.is_panic() => ( + Vec::new(), + vec![format!("verify_ec_shards task panicked: {}", e)], + ), + Err(e) => { + details.push(format!( + "ecvol {}: verify_ec_shards task cancelled; parity check skipped ({})", + vid.0, e + )); + (Vec::new(), Vec::new()) + } + }; + + let mut seen: std::collections::HashSet = + shard_infos.iter().map(|s| s.shard_id).collect(); + for sid in parity_broken { + if seen.insert(sid) { + shard_infos.push(volume_server_pb::EcShardInfo { + shard_id: sid, + collection: collection.clone(), + volume_id: vid.0, + ..Default::default() + }); + } + } + shard_infos.sort_by_key(|s| s.shard_id); + errs.extend(parity_details); + } + + 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. + // Snapshot under a brief lock, then walk with the lock + // RELEASED: this reads every local needle's bytes, which + // under the guard stalls the node. See EcChecksumScrubPlan. + let Some(plan) = ({ + let store = self.state.store.read().unwrap(); + store.find_ec_volume(vid).map(|ecv| ecv.scrub_local_plan()) + }) else { + if let Some(status) = scrub_vanished_volume(explicit, "EC volume", vid) { + return Err(status); + } + continue; + }; + total_volumes += 1; + // Synchronous CPU + file I/O: keep it off the async workers. + let (files, shard_infos, errs) = + match tokio::task::spawn_blocking(move || plan.run()).await { + Ok(v) => v, + Err(e) => { + record_scrub_join_failure( + &e, + vid, + "local scrub", + &mut broken_volume_ids, + &mut details, + ); + continue; + } + }; + 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)); + } + } + } + 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(). + // Snapshot under a brief lock, then verify with the lock + // RELEASED: this reads every byte of every local shard, which + // under the guard stalls the node. See EcChecksumScrubPlan. + let Some((plan, collection)) = ({ + let store = self.state.store.read().unwrap(); + store + .find_ec_volume(vid) + .map(|ecv| (ecv.checksum_scrub_plan(), ecv.collection.clone())) + }) else { + if let Some(status) = scrub_vanished_volume(explicit, "EC volume", vid) { + return Err(status); + } + continue; + }; + total_volumes += 1; + // Synchronous CPU + file I/O: keep it off the async workers. + // `plan.run()`'s first return is blocks scanned, not a file + // count — Go discards it (`_, shardInfos, serrs = v.ChecksumScrub()`) + // so TotalFiles stays a needle/file count and is not inflated + // by the block count. + let (_blocks_scanned, broken, errs) = + match tokio::task::spawn_blocking(move || plan.run()).await { + Ok(v) => v, + Err(e) => { + record_scrub_join_failure( + &e, + vid, + "checksum scrub", + &mut broken_volume_ids, + &mut details, + ); + continue; + } + }; + 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 + } + } + + emit_scrub_metrics( + mode, + broken_volume_ids.len(), + Some(broken_shard_infos.len()), + ); + + Ok(Response::new(volume_server_pb::ScrubEcVolumeResponse { + total_volumes, + total_files, + broken_volume_ids, + broken_shard_infos, + details, + })) + } } #[tonic::async_trait] @@ -4428,85 +4895,14 @@ impl VolumeServer for VolumeGrpcService { } } - let mut total_volumes: u64 = 0; - let mut total_files: u64 = 0; - let mut broken_volume_ids: Vec = Vec::new(); - let mut details: Vec = Vec::new(); - let mut broken_vids: Vec = Vec::new(); + let explicit = !req.volume_ids.is_empty(); + let vids: Vec = if explicit { + req.volume_ids.iter().map(|&id| VolumeId(id)).collect() + } else { + self.state.store.read().unwrap().all_volume_ids() + }; - // Scrub phase: hold store read lock, then drop before async readonly calls. - { - let store = self.state.store.read().unwrap(); - let vids: Vec = if req.volume_ids.is_empty() { - store.all_volume_ids() - } else { - req.volume_ids.iter().map(|&id| VolumeId(id)).collect() - }; - - for vid in &vids { - let (_, v) = store - .find_volume(*vid) - .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 - let scrub_result = if mode == 1 { - v.scrub_index() - } else { - v.scrub() - }; - match scrub_result { - Ok((files, broken)) => { - total_files += files; - if !broken.is_empty() { - broken_vids.push(*vid); - broken_volume_ids.push(vid.0); - for msg in broken { - details.push(format!("vol {}: {}", vid.0, msg)); - } - } - } - Err(e) => { - total_files += v.file_count().max(0) as u64; - broken_vids.push(*vid); - broken_volume_ids.push(vid.0); - details.push(format!("vol {}: scrub error: {}", vid.0, e)); - } - } - } - } // store lock dropped here - - // Match Go: if mark_broken_volumes_readonly, call makeVolumeReadonly on each broken volume. - // Collect errors via errors.Join semantics (return joined error if any fail). - let mut errs: Vec = Vec::new(); - if req.mark_broken_volumes_readonly { - for vid in &broken_vids { - match self.make_volume_readonly(*vid, false, true).await { - Ok(()) => { - details.push(format!("volume {} is now read-only", vid.0)); - } - Err(e) => { - errs.push(e.message().to_string()); - details.push(e.message().to_string()); - } - } - } - } - - // Record metrics before the post-scrub error check so scrub failures are - // persisted even when a follow-up admin action (mark-readonly) fails. - emit_scrub_metrics(mode, broken_vids.len(), None); - - if !errs.is_empty() { - return Err(Status::internal(errs.join("\n"))); - } - - Ok(Response::new(volume_server_pb::ScrubVolumeResponse { - total_volumes, - total_files, - broken_volume_ids, - details, - })) + self.scrub_volumes(&req, vids, explicit).await } async fn scrub_ec_volume( @@ -4538,194 +4934,21 @@ impl VolumeServer for VolumeGrpcService { // 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 = { + // 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 explicit = !req.volume_ids.is_empty(); + let vids: Vec = if explicit { + req.volume_ids.iter().map(|&id| VolumeId(id)).collect() + } else { 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() - } + store + .locations + .iter() + .flat_map(|loc| loc.ec_volumes().map(|(vid, _)| *vid)) + .collect() }; - let mut total_volumes: u64 = 0; - let mut total_files: u64 = 0; - let mut broken_volume_ids: Vec = Vec::new(); - let mut broken_shard_infos: Vec = Vec::new(); - let mut details: Vec = Vec::new(); - - for vid in vids { - match mode { - 1 => { - // 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() { - broken_volume_ids.push(vid.0); - for msg in errs { - details.push(format!("ecvol {}: {}", vid.0, msg)); - } - } - } - 2 | 5 => { - // FULL/READS: Go-parity per-needle local+remote walk, PLUS a TEMPORARY - // local Reed-Solomon parity check. The needle walk only reads - // DATA-shard intervals of LIVE needles, so on its own it can't - // catch silent bitrot in a PARITY shard or an unwalked cold - // region. Go closes that gap with a separate CHECKSUM mode over - // .ecsum, which Rust does not have yet; running both here is a - // deliberate divergence from Go FULL to preserve coverage. Drop - // verify_ec_shards from this arm once mode 4 (CHECKSUM) lands. - // - // The RS recompute needs every shard co-located, so only run it - // when this node holds all data+parity shards (single-node EC); - // on a distributed layout it would report every non-local shard - // as missing. Snapshot under a brief lock; release before await. - let (dir, collection, data_shards, parity_shards, all_local) = { - 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 total = (ecv.data_shards + ecv.parity_shards) as usize; - let local = ecv.shards.iter().filter(|s| s.is_some()).count(); - ( - ecv.dir.clone(), - ecv.collection.clone(), - ecv.data_shards as usize, - ecv.parity_shards as usize, - local == total, - ) - }; - total_volumes += 1; - - // (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, - force_deleted_needles_check, - mode == 5, - ) - .await; - total_files += files as u64; // count comes from the needle walk only - - // (2) Local parity check, gated on all-shards-local. Blocking RS - // verify -> spawn_blocking; inputs are owned, no lock held. - if all_local && !dir.is_empty() { - let collection_pc = collection.clone(); - let (parity_broken, parity_details) = tokio::task::spawn_blocking(move || { - crate::storage::erasure_coding::ec_encoder::verify_ec_shards( - &dir, - &collection_pc, - vid, - data_shards, - parity_shards, - ) - }) - .await - .map_err(|e| Status::internal(format!("verify_ec_shards join: {}", e)))? - .unwrap_or_else(|e| (Vec::new(), vec![format!("verify_ec_shards: {}", e)])); - - let mut seen: std::collections::HashSet = - shard_infos.iter().map(|s| s.shard_id).collect(); - for sid in parity_broken { - if seen.insert(sid) { - shard_infos.push(volume_server_pb::EcShardInfo { - shard_id: sid, - collection: collection.clone(), - volume_id: vid.0, - ..Default::default() - }); - } - } - shard_infos.sort_by_key(|s| s.shard_id); - errs.extend(parity_details); - } - - 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; - 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)); - } - } - } - 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 - } - } - - emit_scrub_metrics( - mode, - broken_volume_ids.len(), - Some(broken_shard_infos.len()), - ); - - Ok(Response::new(volume_server_pb::ScrubEcVolumeResponse { - total_volumes, - total_files, - broken_volume_ids, - broken_shard_infos, - details, - })) + self.scrub_ec_volumes(&req, vids, explicit).await } type QueryStream = BoxStream; @@ -6911,6 +7134,137 @@ mod tests { assert!(vif.expire_at_sec <= before + ttl.to_seconds() + 5); } + /// REGRESSION: a node-wide scrub must survive a volume that legitimately + /// disappears while it runs. + /// + /// Nothing is held across the scan, so the volume set changes under the + /// loop: the heartbeat drops a volume that reported an I/O error, and a + /// delete or an unmount can land between any two volumes. Aborting there + /// discards every result gathered so far and leaves every later volume + /// unscrubbed — on a large node, one expiry costs a whole scrub pass. + /// + /// The absent id goes FIRST, so the assertion fails if the abort ever comes + /// back: the volume behind it would never be reached. + #[tokio::test] + async fn test_scrub_skips_a_vanished_volume_and_scrubs_the_rest() { + let (service, _tmp) = make_local_service_with_volume("", None); + + let resp = service + .scrub_volumes( + &volume_server_pb::ScrubVolumeRequest { + mode: 1, + volume_ids: Vec::new(), + mark_broken_volumes_readonly: false, + }, + vec![VolumeId(17), VolumeId(1)], + false, + ) + .await + .expect("a volume missing from the node's own listing must not fail the scrub") + .into_inner(); + + assert_eq!( + resp.total_volumes, 1, + "the volume behind the vanished one must still be scrubbed" + ); + } + + /// A caller who NAMES a volume that is not here is making a mistake, not + /// racing a teardown, and still gets told — callers match on the message. + #[tokio::test] + async fn test_scrub_fails_on_an_explicitly_requested_missing_volume() { + let (service, _tmp) = make_local_service_with_volume("", None); + + let err = service + .scrub_volumes( + &volume_server_pb::ScrubVolumeRequest { + mode: 1, + volume_ids: vec![17], + mark_broken_volumes_readonly: false, + }, + vec![VolumeId(17)], + true, + ) + .await + .expect_err("an explicitly requested missing volume must still fail"); + + assert_eq!(err.code(), tonic::Code::NotFound); + assert!(err.message().contains("volume id 17 not found"), "{}", err); + } + + /// Same rule for EC volumes, which the heartbeat also expires under a store + /// write: delete_expired_ec_volumes destroys a volume whose destroy time has + /// passed, and volume_ec_shards_delete unmounts one on demand. + #[tokio::test] + async fn test_ec_scrub_skips_a_vanished_volume_and_scrubs_the_rest() { + 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(); + + let resp = service + .scrub_ec_volumes( + &volume_server_pb::ScrubEcVolumeRequest { + mode: 1, + volume_ids: Vec::new(), + force_deleted_needles_check: false, + }, + vec![VolumeId(17), VolumeId(1)], + false, + ) + .await + .expect("an EC volume missing from the node's own listing must not fail the scrub") + .into_inner(); + + assert_eq!( + resp.total_volumes, 1, + "the EC volume behind the vanished one must still be scrubbed" + ); + } + + #[tokio::test] + async fn test_ec_scrub_fails_on_an_explicitly_requested_missing_volume() { + let (service, _tmp) = make_local_service_with_volume("", None); + + let err = service + .scrub_ec_volumes( + &volume_server_pb::ScrubEcVolumeRequest { + mode: 1, + volume_ids: vec![17], + force_deleted_needles_check: false, + }, + vec![VolumeId(17)], + true, + ) + .await + .expect_err("an explicitly requested missing EC volume must still fail"); + + assert_eq!(err.code(), tonic::Code::NotFound); + assert!( + err.message().contains("EC volume id 17 not found"), + "{}", + err + ); + } + async fn scrub_ec_volume_1( service: &VolumeGrpcService, mode: i32, diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 1a8715295..7bf123470 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -259,12 +259,19 @@ pub async fn scrub_ec_volume_distributed( force_deleted_needles_check: bool, recover_unreadable: bool, ) -> (i64, Vec, Vec) { - // Phase A — under the Store read lock, run the index scrub and grab the + // Phase A — under the Store read lock, snapshot the index scrub and grab the // paths/scalars + shard-location staleness; release the lock before any await. + // + // The index walk itself runs AFTER the guard: scrub_index() reads the whole + // .ecx, and doing that under store.read() parks the periodic heartbeat's + // store.write(), which a write-preferring RwLock then makes every later + // reader queue behind. See EcChecksumScrubPlan. let ( ecx_path, collection, - seed_errs, + index_plan, + ecx_walk, + encode_ts_ns, cached_locations, cache_refreshed_at, data_shards, @@ -282,7 +289,26 @@ pub async fn scrub_ec_volume_distributed( } }; // full scan means verifying the index as well - let (_, errs) = ecv.scrub_index(); + let index_plan = ecv.scrub_index_plan(); + // A SECOND .ecx descriptor, opened under the guard for the needle walk + // below. The index plan's handle is consumed by its own structural walk, + // and both seek, so a `dup` would race the cursor. Reopening by PATH + // after the guard is dropped would let a teardown that legitimately + // unlinks or replaces the .ecx (the heartbeat's + // delete_expired_ec_volumes, volume_ec_shards_delete) surface an + // intentional removal as a scrub error or mix index generations within + // one scrub — the same race the vanished-volume policy exists to hide. + // The descriptor outlives the name, the same way the checksum plan's + // shard handles do. + let ecx_walk = fs::File::open(&ecv.ecx_file_name()); + // Encode-run identity of the volume this scrub started against. The + // per-needle `scrub_snapshot_under_lock` re-resolves the volume by id + // under a fresh guard, so a teardown-and-remount of the same vid between + // two rows would otherwise apply the captured .ecx's offsets to a + // replacement volume's shards. Bind the walk to this generation: if the + // mounted volume's encode_ts_ns no longer matches, abort like a + // mid-scan unmount rather than mixing generations. + let encode_ts_ns = ecv.encode_ts_ns; // 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(); @@ -291,13 +317,47 @@ pub async fn scrub_ec_volume_distributed( ( ecv.ecx_file_name(), ecv.collection.clone(), - errs, + index_plan, + ecx_walk, + encode_ts_ns, cached_locations, cache_refreshed_at, data_shards, total_shards, ) }; + // Lock released: walk the index now, before anything else appends to errs, + // so the seeded errors keep their position in the reported details. + // + // `index_plan.run()` reads the whole .ecx synchronously, so run it in the + // blocking pool rather than on this async worker — a large index scan would + // otherwise block unrelated RPC work handled on the same executor. Same + // treatment as the CHECKSUM/LOCAL plans in the gRPC handler. + let (_, seed_errs) = match tokio::task::spawn_blocking(move || index_plan.run()).await { + Ok(v) => v, + Err(e) => { + // A panic is evidence about the volume and counts as broken; a + // cancellation is not — spawn_blocking only reports it when the + // runtime is going down, the volume was never scanned, and the + // caller (FULL/READS) would put a false corruption into + // broken_volume_ids if it reached the errs path. Match the + // record_scrub_join_failure distinction used by the handler arms. + if e.is_panic() { + return ( + 0, + Vec::new(), + vec![format!( + "EC volume {} index scrub task panicked: {}", + vid.0, e + )], + ) + } + // Cancellation: the runtime is shutting down, so this response is + // unlikely to reach anyone. Return clean rather than inventing a + // corruption for a volume that was never scanned. + return (0, Vec::new(), Vec::new()); + } + }; let mut errs = seed_errs; // Refresh the shard-location cache once up front (mirrors Go's @@ -355,28 +415,60 @@ pub async fn scrub_ec_volume_distributed( map }; - // 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; - // Skip ALL deleted entries: -1 tombstones (runtime delete folded - // into .ecx) and -originalSize entries (a needle deleted on the - // regular volume before EC encode). get_actual_size uses the raw - // signed size, so a negative would yield empty intervals - // (false-positive) or an under-16-byte buffer (parse panic). - if !size.is_deleted() { - needles.push((id, offset, size)); + // Walk the .ecx (private fd captured under the lock, no lock held) for the + // row count + live (id, offset, size). Reading through the captured + // descriptor — not a pathname reopen — keeps a concurrent teardown from + // surfacing an intentional removal as a scrub error or mixing index + // generations, the same invariant the vanished-volume policy enforces. + // + // `walk_index_file` reads the full .ecx synchronously, so run it in the + // blocking pool rather than on this async worker — same reason as + // `index_plan.run()` above. + let (count, needles, walk_errs) = + match tokio::task::spawn_blocking(move || -> (i64, Vec<(NeedleId, Offset, Size)>, Vec) { + let mut count: i64 = 0; + let mut needles: Vec<(NeedleId, Offset, Size)> = Vec::new(); + let mut walk_errs: Vec = Vec::new(); + match ecx_walk { + Ok(mut f) => { + if let Err(e) = crate::storage::idx::walk_index_file(&mut f, 0, |id, offset, size| { + count += 1; + // Skip ALL deleted entries: -1 tombstones (runtime delete folded + // into .ecx) and -originalSize entries (a needle deleted on the + // regular volume before EC encode). get_actual_size uses the raw + // signed size, so a negative would yield empty intervals + // (false-positive) or an under-16-byte buffer (parse panic). + if !size.is_deleted() { + needles.push((id, offset, size)); + } + Ok(()) + }) { + walk_errs.push(format!("walk ECX file {}: {}", ecx_path, e)); + } } - Ok(()) - }) { - errs.push(format!("walk ECX file {}: {}", ecx_path, e)); + Err(e) => walk_errs.push(format!("open ECX file {}: {}", ecx_path, e)), } - } - Err(e) => errs.push(format!("open ECX file {}: {}", ecx_path, e)), - } + (count, needles, walk_errs) + }).await { + Ok(v) => v, + Err(e) => { + // A panic is evidence about the volume and counts as broken; a + // cancellation is not — see the index_plan join above for the + // same reasoning. + if e.is_panic() { + return ( + 0, + Vec::new(), + vec![format!( + "EC volume {} ecx walk task panicked: {}", + vid.0, e + )], + ) + } + return (0, Vec::new(), Vec::new()); + } + }; + errs.extend(walk_errs); // reads for EC chunks can hit the same shard repeatedly, so dedupe broken shards let mut broken_shards: HashMap = HashMap::new(); @@ -384,10 +476,11 @@ pub async fn scrub_ec_volume_distributed( 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) { + let snapshot = match scrub_snapshot_under_lock(state, vid, offset, size, encode_ts_ns) { Ok(s) => s, - // Volume unmounted mid-scan: abort with an error rather than skipping - // every remaining needle, which would report a false-CLEAN result. + // Volume unmounted (or remounted as a different encode run) mid-scan: + // abort with an error rather than skipping every remaining needle, + // which would report a false-CLEAN result. Err(e) if e.kind() == io::ErrorKind::NotFound => { errs.push(format!("EC volume {} unmounted during scrub: {}", vid.0, e)); break; @@ -551,6 +644,7 @@ fn scrub_snapshot_under_lock( vid: VolumeId, offset: Offset, size: Size, + expected_encode_ts: i64, ) -> io::Result { let store = state.store.read().unwrap(); let ecv = match store.find_ec_volume(vid) { @@ -564,6 +658,28 @@ fn scrub_snapshot_under_lock( )) } }; + // The volume was torn down and remounted as a DIFFERENT encode run between + // two rows. The .ecx offsets captured at the start of the walk belong to + // the old generation; applying them to the replacement's shards would + // falsely report corruption. Abort like a mid-scan unmount instead of + // mixing generations within one scrub. + // + // `encode_ts_ns == 0` means the .vif carried no encode-run identity (a + // legacy or pre-feature volume). Two such volumes are NOT the same mount + // by this check alone — 0 == 0 would accept a teardown-and-remount and + // apply the old .ecx's offsets to the replacement's shards. Only treat a + // match as verified when the identity is non-zero; when it is zero, fall + // back to the pre-check behavior (no generation binding) rather than + // aborting a scrub that was already running without the guard. + if expected_encode_ts != 0 && ecv.encode_ts_ns != expected_encode_ts { + return Err(io::Error::new( + io::ErrorKind::NotFound, + format!( + "EC volume {} remounted as a different encode run during scrub (was {}, now {})", + vid.0, expected_encode_ts, ecv.encode_ts_ns + ), + )); + } let intervals = ecv.locate_ec_shard_needle_interval(offset.to_actual_offset(), size); if intervals.is_empty() { return Err(io::Error::new( diff --git a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs index 1dc9b42ee..b40bca646 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs @@ -505,7 +505,21 @@ pub fn verify_shard_file_blocks( entry: &EcShardChecksums, block_size: i64, ) -> io::Result> { - let f = File::open(path)?; + verify_shard_blocks(&File::open(path)?, entry, block_size) +} + +/// Same verification against an ALREADY-OPEN shard handle. +/// +/// Go's `ChecksumScrub` reads through `shard.ReadAt`, i.e. the handle the +/// EcVolumeShard already holds, so a concurrent teardown that unlinks the shard +/// cannot turn an intentional removal into a scrub read error. A scrub that +/// runs with the store lock released has to read the same way — see +/// `EcChecksumScrubPlan`. +pub fn verify_shard_blocks( + f: &File, + entry: &EcShardChecksums, + block_size: i64, +) -> io::Result> { let file_size = f.metadata()?.len() as i64; let want = unpack_u32_le(&entry.block_crc32c); @@ -523,7 +537,7 @@ pub fn verify_shard_file_blocks( break; } let to_read = to_read as usize; - read_full_at(&f, &mut buf[..to_read], offset as u64)?; + read_full_at(f, &mut buf[..to_read], offset as u64)?; if CRC::new(&buf[..to_read]).0 != *want_crc { mismatched.push(i); } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_shard.rs b/seaweed-volume/src/storage/erasure_coding/ec_shard.rs index 55ee72594..e7d89631f 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_shard.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_shard.rs @@ -112,6 +112,21 @@ impl EcVolumeShard { self.ecd_file_size } + /// A duplicate of the mounted shard handle, for a reader that has to + /// outlive the store guard. + /// + /// This is the same descriptor `read_at` serves from, so it carries the + /// `O_NOATIME` from `open_volume_file` and keeps pointing at the shard + /// that was mounted, whatever later happens to the path. `dup` shares the + /// kernel file offset, which is why every read through it must be + /// positional (`read_at`), never seek-based. + pub fn try_clone_file(&self) -> io::Result { + self.ecd_file + .as_ref() + .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "shard file not open"))? + .try_clone() + } + /// 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 { diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 51b85c424..f46f578d9 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -621,111 +621,36 @@ impl EcVolume { /// 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; + /// Duplicates the mounted shard handles so `run()` can scan with the store + /// guard released; see `EcChecksumScrubPlan` for why that matters. + pub fn checksum_scrub_plan(&self) -> EcChecksumScrubPlan { + let (prot, status) = self.bitrot_protection(); - 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()); - } + // Only BitrotOn reaches the shard loop in `run()`; the other statuses + // return before touching a handle, so cloning for them buys nothing. + let shards = match status { + crate::storage::erasure_coding::ec_bitrot::BitrotStatus::On => self + .shards + .iter() + .enumerate() + .filter_map(|(i, slot)| slot.as_ref().map(|s| (i as u32, s.try_clone_file()))) + .collect(), + _ => 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 - )); - } - } + EcChecksumScrubPlan { + volume_id: self.volume_id, + prot, + status, + parity_shards: self.parity_shards, + shards, } + } - // 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) + /// Convenience wrapper preserving the original call shape. Callers that + /// hold the store lock MUST use `checksum_scrub_plan()` + `run()` instead. + pub fn checksum_scrub(&self) -> (u64, Vec, Vec) { + self.checksum_scrub_plan().run() } /// Walk the .ecj journal and populate `deleted_needles`. Called once @@ -1281,187 +1206,83 @@ impl EcVolume { /// ScrubIndex verifies index integrity of an EC volume. /// Matches Go's `(ev *EcVolume) ScrubIndex()` → `idx.CheckIndexFile()`. /// Returns (entry_count, errors). - pub fn scrub_index(&self) -> (u64, Vec) { - if self.ecx_file.is_none() { - return ( - 0, - vec![format!( - "no ECX file associated with EC volume {}", - self.volume_id.0 - )], - ); - } - if self.ecx_file_size == 0 { - return ( - 0, - vec![format!("zero-size ECX file for EC volume {}", self.volume_id.0)], - ); - } - - // Walk a private fd so the structural scan never moves the shared - // ecx_file cursor (the cached handle is read positionally elsewhere). + /// Snapshot for `scrub_index`, so the index walk can run with the store + /// lock released. Same rationale as `checksum_scrub_plan`. + pub fn scrub_index_plan(&self) -> EcIndexScrubPlan { let ecx_path = self.ecx_file_name(); - let mut ecx_file = match File::open(&ecx_path) { - Ok(f) => f, - Err(e) => return (0, vec![format!("open ECX file {}: {}", ecx_path, e)]), - }; - crate::storage::idx::check_index_file(&mut ecx_file, self.ecx_file_size, self.version) + // Opened under the guard, for the same reason the checksum plan + // duplicates its shard handles. A fresh open rather than a clone of the + // cached handle: `check_index_file` seeks, and `dup` would share the + // cursor the shared handle is read from elsewhere. + let ecx_handle = File::open(&ecx_path); + EcIndexScrubPlan { + volume_id: self.volume_id, + has_ecx_file: self.ecx_file.is_some(), + ecx_path, + ecx_file_size: self.ecx_file_size, + version: self.version, + ecx_handle, + } + } + + /// Convenience wrapper preserving the original call shape. Callers holding + /// the store lock MUST use `scrub_index_plan()` + `run()` instead. + pub fn scrub_index(&self) -> (u64, Vec) { + self.scrub_index_plan().run() + } + + /// Snapshot for `scrub_local`, so the needle walk can run with the store + /// lock released. Same rationale as `checksum_scrub_plan`: this reads every + /// local needle's bytes, which is GB-scale on a real volume. + /// + /// The shard vector keeps its SLOT structure — index is the shard id, gaps + /// are the shards this node does not hold. `scrub_local` reads that + /// distinction to decide whether a needle can be reassembled locally at + /// all, so compacting it would silently change which needles get verified. + pub fn scrub_local_plan(&self) -> EcLocalScrubPlan { + EcLocalScrubPlan { + volume_id: self.volume_id, + version: self.version, + data_shards: self.data_shards, + // locate_data wants 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). + shard_size: if self.dat_file_size > 0 { + self.dat_file_size / self.data_shards as i64 + } else { + self.shard_file_size() - 1 + }, + large_block_size: self.large_block_size(), + small_block_size: self.small_block_size(), + index: self.scrub_index_plan(), + ecx_path: self.ecx_file_name(), + // A second descriptor: the index plan's is consumed by its own walk, + // and both seek. + ecx_walk: File::open(self.ecx_file_name()), + shards: self + .shards + .iter() + .map(|slot| { + slot.as_ref().map(|s| EcLocalShard { + file: s.try_clone_file(), + file_size: s.file_size(), + info: s.to_ec_shard_info(), + }) + }) + .collect(), + } } /// 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). + /// + /// Convenience wrapper preserving the original call shape. Callers that + /// hold the store lock MUST use `scrub_local_plan()` + `run()` instead. 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, _) = self.interval_to_shard_id_and_offset(iv); - 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) = self.interval_to_shard_id_and_offset(iv); - 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) { - // A delete-state disagreement between the .ecx index and the reassembled - // on-disk header (live index vs zero header size) is not corruption. - let delete_state_disagrees = matches!( - &e, - NeedleError::SizeMismatch { found, .. } if size.is_deleted() != (found.0 == 0) - ); - if !delete_state_disagrees { - 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) + self.scrub_local_plan().run() } // ---- Deletion ---- @@ -1964,6 +1785,273 @@ mod tests { assert_eq!(prot.unwrap().shards.len(), 14); } + /// REGRESSION: the scrub plans must be SELF-CONTAINED, so the handler can + /// drop the store lock before the scan runs — see `EcChecksumScrubPlan`. + /// + /// The volume is DROPPED before the plans run, and the plans are moved to + /// another thread. A plan that borrowed from `EcVolume` could do neither, so + /// this stops COMPILING if the snapshot ever regresses to a borrow. + #[test] + fn test_scrub_plans_are_self_contained_and_match_direct_call() { + 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, false).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(); + } + + // Baseline via the original call shape, with the volume still alive. + let direct_checksum = vol.checksum_scrub(); + let direct_index = vol.scrub_index(); + let direct_local = vol.scrub_local(); + assert!( + direct_checksum.0 > 0 && direct_local.0 > 0, + "fixture scanned nothing, so the equalities below would be vacuous" + ); + + let checksum_plan = vol.checksum_scrub_plan(); + let index_plan = vol.scrub_index_plan(); + let local_plan = vol.scrub_local_plan(); + + // The lock (and here the whole volume) is gone before the scan runs. + drop(vol); + + let (from_plan_checksum, from_plan_index, from_plan_local) = + std::thread::spawn(move || (checksum_plan.run(), index_plan.run(), local_plan.run())) + .join() + .expect("scrub plans must be runnable off the owning thread"); + + assert_eq!( + from_plan_checksum, direct_checksum, + "checksum scrub result changed when run from a released-lock plan" + ); + assert_eq!( + from_plan_index, direct_index, + "index scrub result changed when run from a released-lock plan" + ); + assert_eq!( + from_plan_local, direct_local, + "local scrub result changed when run from a released-lock plan" + ); + } + + /// REGRESSION: a malformed `.ecx` row must not abort the scrub TASK. + /// + /// `EcLocalScrubPlan::run()` skips only the -1 tombstone, matching Go's + /// `ScrubLocal`, so any OTHER negative size reaches the reassembly buffer + /// with a negative `get_actual_size()`. Go pays nothing for that (it + /// appends to a nil slice); Rust sizes a per-needle `Vec` from it, and + /// `Vec::with_capacity(negative as usize)` aborts the process. Now that the + /// plan runs under `spawn_blocking`, that abort comes back as a JoinError + /// and would take the whole node-wide scrub RPC down with every result + /// already collected. The row must be REPORTED, as Go reports it. + #[test] + fn test_local_scrub_plan_reports_negative_size_ecx_row() { + 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, false).unwrap(); + } + v.sync_to_disk().unwrap(); + v.close(); + crate::storage::erasure_coding::ec_encoder::write_ec_files(dir, dir, "", VolumeId(1), 10, 4) + .unwrap(); + + // Rewrite the first .ecx row's size as -1000: a negative that is NOT + // the -1 tombstone the walk skips. A scrub is what you point at an + // index you already suspect, so an arbitrary i32 in the size field is + // in-scope input, whatever wrote it. + let ecx = format!( + "{}.ecx", + crate::storage::volume::volume_file_name(dir, "", VolumeId(1)) + ); + let mut raw = std::fs::read(&ecx).unwrap(); + assert!( + raw.len() >= NEEDLE_MAP_ENTRY_SIZE, + "fixture must write at least one .ecx row" + ); + let (key, offset, _) = idx_entry_from_bytes(&raw[..NEEDLE_MAP_ENTRY_SIZE]); + idx_entry_to_bytes(&mut raw[..NEEDLE_MAP_ENTRY_SIZE], key, offset, Size(-1000)); + std::fs::write(&ecx, &raw).unwrap(); + assert!( + get_actual_size(Size(-1000), Version::current()) < 0, + "precondition: the row must drive get_actual_size negative" + ); + + 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(); + } + let plan = vol.scrub_local_plan(); + drop(vol); + + // The join IS the assertion: a panic here is the JoinError that used to + // fail the whole ScrubEcVolume RPC. + let (_count, _broken, errs) = std::thread::spawn(move || plan.run()) + .join() + .expect("a malformed .ecx row must not abort the scrub task"); + + assert!( + errs.iter() + .any(|e| e.contains(&format!("bytes for needle {}", key.0))), + "the malformed row must be reported, got {:?}", + errs + ); + } + + /// REGRESSION: a scrub running with the store lock RELEASED must not turn a + /// concurrent, intentional removal into a corruption report — see + /// `EcChecksumScrubPlan`. + /// + /// Deleting every file after the plans are built is the whole test: reads + /// that resolve a path diverge from the direct call, reads through the + /// captured descriptors are identical. + #[test] + fn test_scrub_plans_survive_files_removed_after_snapshot() { + 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, false).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(); + } + + // Baseline with every file present and the volume still mounted. + let direct_checksum = vol.checksum_scrub(); + let direct_index = vol.scrub_index(); + let direct_local = vol.scrub_local(); + assert!( + direct_checksum.0 > 0 && direct_local.0 > 0, + "fixture scanned nothing, so the equalities below would be vacuous" + ); + assert!( + direct_checksum.2.is_empty() && direct_index.1.is_empty() && direct_local.2.is_empty(), + "fixture is not clean, so a false error could not be told apart: {:?} {:?} {:?}", + direct_checksum.2, + direct_index.1, + direct_local.2 + ); + + let checksum_plan = vol.checksum_scrub_plan(); + let index_plan = vol.scrub_index_plan(); + let local_plan = vol.scrub_local_plan(); + + // The teardown a store writer would perform, after the plans exist. + drop(vol); + let base = crate::storage::volume::volume_file_name(dir, "", VolumeId(1)); + for id in 0..14u8 { + std::fs::remove_file(format!("{}.ec{:02}", base, id)).unwrap(); + } + std::fs::remove_file(format!("{}.ecx", base)).unwrap(); + assert!( + !std::path::Path::new(&format!("{}.ec00", base)).exists(), + "the removal under test did not happen" + ); + + assert_eq!( + checksum_plan.run(), + direct_checksum, + "shard files unlinked after the snapshot were reported as corruption" + ); + assert_eq!( + index_plan.run(), + direct_index, + "the .ecx unlinked after the snapshot was reported as an index error" + ); + assert_eq!( + local_plan.run(), + direct_local, + "files unlinked after the snapshot were reported as local-scrub errors" + ); + } + /// CHECKSUM scrub verifies clean shards against the sidecar and flags a shard /// whose bytes are corrupted after encode. #[test] @@ -2774,3 +2862,435 @@ mod uniform_layout_tests { assert_eq!((ds, ps, bs), (12, 4, 3 * 1024 * 1024)); } } + +/// Self-contained input for an EC checksum scrub: the sidecar plus a duplicate +/// of every mounted local shard handle, taken under the store read guard. +/// +/// Two things follow from holding descriptors rather than paths. `run()` needs +/// no store access, so the guard is released before a scan that reads every +/// byte of every local shard — under the guard that scan parks the periodic +/// heartbeat's `store.write()`, and `std::sync::RwLock` is write-preferring, so +/// every later reader queues behind it and the node stops serving. And a +/// teardown that legitimately unlinks the shards mid-scan (the heartbeat's +/// `delete_expired_ec_volumes`, `volume_ec_shards_delete`) cannot masquerade as +/// bitrot, because the descriptor outlives the name. Go reads the same way: +/// `ChecksumScrub` goes through `shard.ReadAt`. +/// +/// Not `Clone`: it owns those descriptors. +#[derive(Debug)] +pub struct EcChecksumScrubPlan { + pub volume_id: VolumeId, + pub prot: Option, + pub status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus, + pub parity_shards: u32, + /// One entry per LOCAL shard: its id and a duplicate of the mounted + /// handle (or the error to report). Private so the plan can only be built + /// by `EcVolume::checksum_scrub_plan`, which is what makes "captured under + /// the guard" an invariant rather than a convention. + /// + /// `dup` shares the kernel file offset, so every read here is positional. + shards: Vec<(u32, std::io::Result)>, +} + +impl EcChecksumScrubPlan { + /// The byte-verification pass. Touches only the filesystem — no store, no + /// lock — so it is safe to hand to `spawn_blocking`. + pub fn run(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.prot, self.status) { + (_, 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 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 (shard_id, handle) in self.shards { + 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; + }; + + // The handle was opened under the store lock; reading through it + // means a concurrent unmount/unlink cannot masquerade as bitrot. + let file = match &handle { + Ok(f) => f, + Err(e) => { + errors.push(format!( + "EC volume {} shard {} scrub read error: {}", + self.volume_id.0, shard_id, e + )); + continue; + } + }; + + let expected_blocks = entry.block_crc32c.len() / 4; + match ec_bitrot::verify_shard_blocks(file, 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) + } +} + +/// Self-contained input for an EC index scrub. +/// +/// Not `Clone`: it owns the .ecx handle opened under the store lock. +#[derive(Debug)] +pub struct EcIndexScrubPlan { + pub volume_id: VolumeId, + pub has_ecx_file: bool, + pub ecx_path: String, + pub ecx_file_size: i64, + pub version: Version, + /// The .ecx handle opened while the store lock was held. Private, so the + /// plan can only come from `EcVolume::scrub_index_plan`. + ecx_handle: std::io::Result, +} + +impl EcIndexScrubPlan { + /// Structural walk of the .ecx index. Filesystem only — no store, no lock. + pub fn run(self) -> (u64, Vec) { + if !self.has_ecx_file { + return ( + 0, + vec![format!( + "no ECX file associated with EC volume {}", + self.volume_id.0 + )], + ); + } + if self.ecx_file_size == 0 { + return ( + 0, + vec![format!("zero-size ECX file for EC volume {}", self.volume_id.0)], + ); + } + + // A private fd, so the structural scan never moves the shared ecx_file + // cursor (the cached handle is read positionally elsewhere). Checked + // after the two guards above so the error ordering is unchanged. + let mut ecx_file = match self.ecx_handle { + Ok(f) => f, + Err(e) => { + return ( + 0, + vec![format!("open ECX file {}: {}", self.ecx_path, e)], + ) + } + }; + crate::storage::idx::check_index_file(&mut ecx_file, self.ecx_file_size, self.version) + } +} + +/// One local shard as `EcLocalScrubPlan` sees it: the mounted descriptor, the +/// size the scan compares against, and the identity a broken-shard report needs. +#[derive(Debug)] +pub struct EcLocalShard { + file: std::io::Result, + /// The shard's cached size, as `scrub_local` has always compared against — + /// deliberately not a live `metadata()` call in `run()`. + file_size: i64, + info: crate::pb::volume_server_pb::EcShardInfo, +} + +impl EcLocalShard { + /// Positional read through the duplicated handle. `dup` shares the kernel + /// offset with the mounted shard, so this must never seek. + fn read_at(&self, buf: &mut [u8], offset: u64) -> io::Result { + let file = self + .file + .as_ref() + .map_err(|e| io::Error::new(e.kind(), e.to_string()))?; + #[cfg(unix)] + { + use std::os::unix::fs::FileExt; + file.read_at(buf, offset) + } + #[cfg(not(unix))] + { + use std::io::{Read, Seek, SeekFrom}; + let mut f = file.try_clone()?; + f.seek(SeekFrom::Start(offset))?; + f.read(buf) + } + } +} + +/// Self-contained input for an EC LOCAL scrub: the index snapshot, a private +/// .ecx descriptor for the needle walk, and the mounted local shard handles. +/// +/// Same reasoning as `EcChecksumScrubPlan` — `scrub_local` reads every local +/// needle's bytes, so it must not run under the store read guard. +/// +/// Not `Clone`: it owns descriptors. +#[derive(Debug)] +pub struct EcLocalScrubPlan { + volume_id: VolumeId, + version: Version, + data_shards: u32, + shard_size: i64, + large_block_size: i64, + small_block_size: i64, + index: EcIndexScrubPlan, + ecx_path: String, + ecx_walk: std::io::Result, + /// Indexed BY SHARD ID; `None` is a shard this node does not hold. + shards: Vec>, +} + +impl EcLocalScrubPlan { + /// The needle walk. Filesystem only — no store, no lock. + pub fn run(self) -> (u64, Vec, Vec) { + let EcLocalScrubPlan { + volume_id, + version, + data_shards, + shard_size, + large_block_size, + small_block_size, + index, + ecx_path, + ecx_walk, + shards, + } = self; + + // Local scan also verifies the index. + let (_, mut errs) = index.run(); + + let mut broken_shards: HashSet = HashSet::new(); + let mut count: u64 = 0; + + let mut ecx_file = match ecx_walk { + 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(()); + } + + // Go recomputes this at the size check below; hoisted because Rust + // also sizes the reassembly buffer from it. Any negative size other + // than the -1 tombstone skipped above drives it negative. + let want = get_actual_size(size, version); + + let locations = ec_locate::locate_data( + offset.to_actual_offset(), + Size(want as i32), + shard_size, + data_shards, + large_block_size, + small_block_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(data_shards, large_block_size, small_block_size); + shards.get(sid as usize).and_then(|s| s.as_ref()).is_none() + }); + let mut read: i64 = 0; + // `want <= 0` means the row is malformed. Go pays nothing for it: + // it appends to a nil slice and has no capacity hint here. Rust's + // per-needle buffer does, and `Vec::with_capacity(negative as usize)` + // aborts the process -- inside spawn_blocking that surfaces as a + // JoinError and takes the whole node-wide scrub RPC with it. Fall + // through with an empty buffer instead: locate_data already returns + // no intervals for a non-positive size, so read stays 0 and the + // `read != want` check below reports the row, exactly as Go does. + let mut data: Vec = if has_remote_chunks || want <= 0 { + Vec::new() + } else { + Vec::with_capacity(want 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(data_shards, large_block_size, small_block_size); + let ssize = iv.size; + let shard = match 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(); + + 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, 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, version) { + // A delete-state disagreement between the .ecx index and the reassembled + // on-disk header (live index vs zero header size) is not corruption. + let delete_state_disagrees = matches!( + &e, + NeedleError::SizeMismatch { found, .. } if size.is_deleted() != (found.0 == 0) + ); + if !delete_state_disagrees { + errs.push(format!( + "needle {} on volume {}, shards {:?}: {}", + id.0, 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| shards.get(*sid as usize).and_then(|s| s.as_ref())) + .map(|s| s.info.clone()) + .collect(); + broken.sort_by(|a, b| a.shard_id.cmp(&b.shard_id)); + + (count, broken, errs) + } +} diff --git a/weed/server/volume_grpc_scrub.go b/weed/server/volume_grpc_scrub.go index 351159aa7..325cde652 100644 --- a/weed/server/volume_grpc_scrub.go +++ b/weed/server/volume_grpc_scrub.go @@ -7,6 +7,7 @@ import ( "time" "github.com/prometheus/client_golang/prometheus" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage" @@ -18,7 +19,8 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S return nil, err } vids := []needle.VolumeId{} - if len(req.GetVolumeIds()) == 0 { + explicit := len(req.GetVolumeIds()) != 0 + if !explicit { for _, l := range vs.store.Locations { vids = append(vids, l.VolumeIds()...) } @@ -28,6 +30,22 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S } } + return vs.scrubVolumes(ctx, req, vids, explicit) +} + +// scrubVolumes walks an already-resolved volume list. Split out from +// ScrubVolume so the vanished-volume handling below is reachable from a test: +// the list and the store are read under the same lock nowhere, so a test +// cannot otherwise arrange for an id to disappear between the two. +// +// explicit says the caller named the volumes. A named volume that is absent is +// a caller error and still fails the request; an id that came from the node's +// own listing is not, because nothing holds a lock across the scan and the +// volume set is free to change under it — the heartbeat drops a volume with an +// I/O error, and a delete or an unmount can land at any point. Failing there +// would discard every result accumulated so far and leave every later volume +// unscrubbed, which is the opposite of what a node-wide scrub is for. +func (vs *VolumeServer) scrubVolumes(ctx context.Context, req *volume_server_pb.ScrubVolumeRequest, vids []needle.VolumeId, explicit bool) (*volume_server_pb.ScrubVolumeResponse, error) { var details []string var totalVolumes, totalFiles uint64 var brokenVolumes []*storage.Volume @@ -35,7 +53,11 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S for _, vid := range vids { v := vs.store.GetVolume(vid) if v == nil { - return nil, fmt.Errorf("volume id %d not found", vid) + if explicit { + return nil, fmt.Errorf("volume id %d not found", vid) + } + glog.V(0).Infof("scrub: volume %d is no longer mounted, skipping", vid) + continue } var files int64 @@ -103,7 +125,8 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb } vids := []needle.VolumeId{} - if len(req.GetVolumeIds()) == 0 { + explicit := len(req.GetVolumeIds()) != 0 + if !explicit { for _, l := range vs.store.Locations { vids = append(vids, l.EcVolumeIds()...) } @@ -113,6 +136,12 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb } } + return vs.scrubEcVolumes(req, vids, explicit) +} + +// scrubEcVolumes walks an already-resolved EC volume list. Same split, and same +// vanished-volume rule, as scrubVolumes — see its comment. +func (vs *VolumeServer) scrubEcVolumes(req *volume_server_pb.ScrubEcVolumeRequest, vids []needle.VolumeId, explicit bool) (*volume_server_pb.ScrubEcVolumeResponse, error) { var details []string var totalVolumes, totalFiles uint64 var brokenVolumeIds []uint32 @@ -120,7 +149,11 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb for _, vid := range vids { v, found := vs.store.FindEcVolume(vid) if !found { - return nil, fmt.Errorf("EC volume id %d not found", vid) + if explicit { + return nil, fmt.Errorf("EC volume id %d not found", vid) + } + glog.V(0).Infof("ec scrub: volume %d is no longer mounted, skipping", vid) + continue } var files int64 diff --git a/weed/server/volume_grpc_scrub_test.go b/weed/server/volume_grpc_scrub_test.go new file mode 100644 index 000000000..8b32e22c9 --- /dev/null +++ b/weed/server/volume_grpc_scrub_test.go @@ -0,0 +1,85 @@ +package weed_server + +import ( + "context" + "os" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A node-wide scrub reads the volume list without holding anything across the +// scan, so the set is free to change under it: the heartbeat drops a volume +// that reported an I/O error, and a delete or an unmount can land between two +// volumes. Failing the RPC there discards every result gathered so far and +// leaves every later volume unscrubbed, so an id that came from the node's own +// listing is skipped instead. +func TestScrubVolumeSkipsVanishedVolumeAndScrubsTheRest(t *testing.T) { + present := needle.VolumeId(1) + vs, _ := newMaintenanceModeServer(t, present, "") + + // Absent first: if the handler still aborted, the present volume behind it + // would never be scrubbed. + res, err := vs.scrubVolumes(context.Background(), + &volume_server_pb.ScrubVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX}, + []needle.VolumeId{needle.VolumeId(17), present}, false) + require.NoError(t, err, "a volume that disappeared from the node's own listing must not fail the scrub") + assert.Equal(t, uint64(1), res.GetTotalVolumes(), "the volume behind the vanished one must still be scrubbed") +} + +// A caller who names a volume that is not here gets told, unchanged. +func TestScrubVolumeFailsOnExplicitlyRequestedMissingVolume(t *testing.T) { + vs, _ := newMaintenanceModeServer(t, needle.VolumeId(1), "") + + _, err := vs.scrubVolumes(context.Background(), + &volume_server_pb.ScrubVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX}, + []needle.VolumeId{needle.VolumeId(17)}, true) + require.Error(t, err) + assert.Contains(t, err.Error(), "volume id 17 not found") +} + +// newEcScrubServer mounts one EC shard (plus the shared index) so FindEcVolume +// resolves, which is all an INDEX-mode scrub needs. +func newEcScrubServer(t *testing.T, vid needle.VolumeId, collection string) *VolumeServer { + t.Helper() + dir := t.TempDir() + store := newTraversalTestStore(dir) + t.Cleanup(store.Close) + + base := erasure_coding.EcShardFileName(collection, dir, int(vid)) + require.NoError(t, os.WriteFile(base+".ecx", make([]byte, types.NeedleMapEntrySize), 0o644)) + require.NoError(t, os.WriteFile(base+".ecj", nil, 0o644)) + require.NoError(t, os.WriteFile(base+erasure_coding.ToExt(0), []byte("s"), 0o644)) + require.NoError(t, store.MountEcShards(collection, vid, 0, types.HardDriveType.String())) + + return &VolumeServer{store: store} +} + +// Same rule for EC volumes, which the heartbeat also expires under a store +// write: delete_expired_ec_volumes destroys a volume whose destroy time has +// passed, and volume_ec_shards_delete unmounts one on demand. +func TestScrubEcVolumeSkipsVanishedVolumeAndScrubsTheRest(t *testing.T) { + present := needle.VolumeId(77) + vs := newEcScrubServer(t, present, "ec-scrub") + + res, err := vs.scrubEcVolumes( + &volume_server_pb.ScrubEcVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX}, + []needle.VolumeId{needle.VolumeId(17), present}, false) + require.NoError(t, err, "an EC volume that disappeared from the node's own listing must not fail the scrub") + assert.Equal(t, uint64(1), res.GetTotalVolumes(), "the EC volume behind the vanished one must still be scrubbed") +} + +func TestScrubEcVolumeFailsOnExplicitlyRequestedMissingVolume(t *testing.T) { + vs := newEcScrubServer(t, needle.VolumeId(77), "ec-scrub") + + _, err := vs.scrubEcVolumes( + &volume_server_pb.ScrubEcVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX}, + []needle.VolumeId{needle.VolumeId(17)}, true) + require.Error(t, err) + assert.Contains(t, err.Error(), "EC volume id 17 not found") +}