From 9b12d13934e1cd4031608cb40fb1b8a7dd935009 Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Wed, 9 Sep 2026 05:12:30 +0300 Subject: [PATCH] volume server: release the store lock before scrubbing EC volumes (#11235) * volume server: release the store lock before scrubbing EC volumes `ec.scrub` makes a Rust volume server stop serving for the duration of the scrub, and then kills its own gRPC connection: error: rpc error: code = Unavailable desc = keepalive ping failed to receive ACK within timeout Measured on a 4.46 cluster (17 Rust volume servers on one host, ~520 volumes and 53 EC volumes, --index=redb, EC 10+4). It reproduces against a SINGLE node in 30-70s, in checksum, index and local modes, at -maxParallelization 1. ## Cause The CHECKSUM arm of scrub_ec_volume reads every byte of every local shard while holding the caller's store.read() guard: let store = self.state.store.read().unwrap(); let ecv = store.find_ec_volume(vid)...?; let (blocks, broken, errs) = ecv.checksum_scrub(); // GBs of I/O, lock held VolumeServerState::store is a std::sync::RwLock, which is write-preferring. The periodic heartbeat's collect_heartbeat_with_snapshot takes store.write() and blocks; once that writer is pending, every later store.read() queues behind it. Every HTTP handler takes store.read(), so the node serves nothing, stops heart-beating, and cannot answer the scrub RPC's own keepalive - the scrub kills the connection it is running on. The INDEX and LOCAL arms have the same shape, and the node-wide scrub_volume loop is worse: it held ONE guard across every volume on the node. ## Evidence offcputime, off-CPU stacks >1s in a 30s window during a scrub: futex_wait seaweed_volume::server::heartbeat::collect_heartbeat_with_snapshot - tokio-rt-worker 27967020 <- 27.97s blocked, of a 30s window A single HTTP /status request issued 12s into a scrub, with 180s of patience, was accepted and queued for 120 seconds, then served once the scrub released. Thread states throughout: 1 D + 48 S. One thread working, 48 idle - not executor starvation and no thread pileup, which is what a single lock holder looks like. Memory was tested and ruled out as the cause: the same scrub was run at MemoryMax 3G, 8G and unlimited. With no limit there is no reclaim at all, page cache grows freely to 22 GB, and the node still goes unresponsive at t+30s. anon stays flat at 48-86 MB in every run. ## Fix checksum_scrub, scrub_index and scrub_local gain plan types - EcChecksumScrubPlan, EcIndexScrubPlan and EcLocalScrubPlan - snapshotted from the volume under a brief guard. The handler builds a plan, drops the guard, and runs the scan in spawn_blocking, off the async workers, since it is synchronous CPU + file I/O either way. A plan captures DESCRIPTORS, not paths. Resolving a path again after the guard is dropped would let a writer that legitimately unlinks the files - the heartbeat's delete_expired_ec_volumes, which reaches EcVolume::destroy(), or volume_ec_shards_delete - surface an intentional removal as "scrub read error: No such file or directory" and put the volume in broken_volume_ids. A descriptor outlives the name. For the shards it duplicates the handle the mounted EcVolumeShard already holds (try_clone_file), which is what Go does: ChecksumScrub reads through shard.ReadAt (weed/storage/erasure_coding/ec_volume_scrub.go:71), never through a path. That also inherits open_volume_file's O_NOATIME and drops a dead branch - the old code built {base}.ec{id}.v{gen} for a non-zero generation, a name nothing in this tree writes. dup shares the kernel offset, so shard reads stay positional; the .ecx gets a fresh open instead, since check_index_file seeks. FULL/READS is unchanged here: it already released the guard across the index walk, and still re-takes it per needle in store_ec::scrub_snapshot_under_lock for that needle's local shard intervals - short holds, many of them. scrub_volume now takes the read guard PER VOLUME instead of across the whole loop, so the heartbeat can land between volumes. Its per-volume work still runs under the guard; Volume needs an equivalent plan to fix that properly, left as a follow-up and noted in the code. ## A failed scrub task must not take the whole RPC down Moving the scans into spawn_blocking changed where a panic lands. It no longer unwinds inside the handler's own future; it comes back as a JoinError at the .await, and all four join points sat behind a `?`. So one bad volume out of six hundred returned Err from the entire handler: the broken_volume_ids, broken_shard_infos and details already gathered for the other 599 were dropped, and emit_scrub_metrics - the only writer of SCRUB_LAST_TIME_SECONDS, SCRUB_VOLUME_FAILURES and SCRUB_SHARD_FAILURES - was never reached, so the staleness alert kept firing while real corruption went unreported. And there is a reachable panic behind it. EcLocalScrubPlan::run() sized its reassembly buffer with Vec::with_capacity(get_actual_size(size, version) as usize) which for any negative size that is not the -1 tombstone skipped above is a capacity-overflow abort. Mode 3 (LOCAL) is the default of `weed shell ec.scrub`, and a scrub is what you point at an index you already suspect, so an arbitrary i32 in a .ecx size field is in-scope input. The buffer is Rust-only - Go appends to a nil slice and has no capacity hint here. Guard on `want <= 0` and fall through with an empty buffer: locate_data returns no intervals for a non-positive size, read stays 0, and the existing `read != want` error reports the row exactly as Go does. Each join point now records the failure against its own volume and continues. A panic is evidence about the volume and counts as broken; a non-panic JoinError is not - spawn_blocking only reports one when the runtime is going down, the volume was never scanned, and counting it would put a false corruption into SCRUB_VOLUME_FAILURES. total_volumes moves before the join in modes 1, 3 and 4 (2|5 already counted there) so a failed join cannot silently shrink it. Mode 2|5's verify_ec_shards join is the one that must not `continue`: the needle walk above has already produced findings for that volume. The tombstone guard stays is_tombstone() on purpose. ScrubLocal in ec_volume_scrub.go:228 skips only IsTombstone(), while the distributed walk in store_ec.go:516 skips all IsDeleted() - the asymmetry is Go's, and both Rust walks mirror their own counterpart. ## Both servers: a node-wide scrub skips a volume that vanished mid-run Releasing the lock makes the volume set legitimately mutable during a scrub, so a node-wide run can reach a volume that has since been unmounted. That is not a scrub failure. A node-wide run now logs and skips it; an explicitly requested volume id still returns NotFound. The Go server is changed the same way, so both implementations answer the same shell command identically. mark_broken_volumes_readonly tolerates the same teardown one step later, instead of throwing away the whole scrub report. ## Test test_scrub_plans_are_self_contained_and_match_direct_call drops the EcVolume and runs both plans on another thread, asserting the results match the direct calls. A plan that borrowed from EcVolume could do neither, so the test stops compiling if the snapshot regresses to a borrow. test_scrub_plans_survive_files_removed_after_snapshot unlinks every shard and the .ecx after the plans are built, then asserts the results still equal the direct call. Against a path-resolving version it fails with all 14 shards reported as "No such file or directory". test_local_scrub_plan_reports_negative_size_ecx_row rewrites a .ecx row's size to -1000 and runs the local plan on another thread, so the join is the assertion - that thread is the spawn_blocking whose panic used to fail the RPC. Without the capacity guard it fails with "capacity overflow"; with it, the row is reported. The Go tests cover both halves of the vanished-volume rule for volumes and EC volumes. 517 lib tests pass, plus 34 across the other targets (`cargo test`). `go test ./weed/server -run Scrub` passes. ## Known remaining, not fixed here `ec.scrub -volumeId=N` is still fanned out to every node, and a node that holds no shard of N returns NotFound, so the shell command errors even when the nodes that do hold shards scrub cleanly. That is a shell-side fan-out question rather than a volume-server one, and both servers keep the existing behaviour for an explicitly requested id. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01DvHoW85w6SNKNBPvrqLMmK * scrub: discard checksum block count from total_files; capture .ecx fd for FULL walk Two review fixes: 1. CHECKSUM arm: plan.run() returns blocks scanned, not a file count. Go discards it (_, shardInfos, serrs = v.ChecksumScrub()) so TotalFiles stays a needle/file count. The Rust arm was adding it to total_files, inflating the count. Discard it to match Go. 2. FULL/READS (scrub_ec_volume_distributed): the needle walk reopened the .ecx by PATH after the store guard was released, so a concurrent teardown that unlinks or replaces the .ecx (heartbeat delete_expired_ec_volumes, volume_ec_shards_delete) could surface an intentional removal as a scrub error or mix index generations within one scrub. Capture a second .ecx descriptor under the guard (the index plan handle is consumed by its own structural walk, and both seek) and read through it instead -- the same descriptor-outlives-name invariant the checksum plan shard handles use. * scrub: bind FULL/READS walk to one encode generation Address Devin review: after capturing the .ecx descriptor under the guard, scrub_snapshot_under_lock still re-resolves the volume by id per needle, so a teardown-and-remount of the same vid between two rows would apply the captured .ecx offsets to a replacement volume's shards -- falsely reporting corruption. Capture the volume's encode_ts_ns (encode-run identity) in Phase A and pass it to scrub_snapshot_under_lock. If the mounted volume's encode_ts_ns no longer matches, abort the walk like a mid-scan unmount instead of mixing generations within one scrub. * scrub: run FULL/READS index scan in the blocking pool Address CodeRabbit review (5147767192): index_plan.run() reads the whole .ecx synchronously, so running it on the async executor worker could block unrelated RPC work handled on the same executor. Move it into spawn_blocking, matching the treatment the CHECKSUM/LOCAL arms already give their plans. A join failure (panic/cancellation) is reported as a seed error so the per-volume findings below are not silently dropped. * scrub: move ecx walk to blocking pool, classify join errors, guard encode_ts_ns==0 Three CodeRabbit review fixes (5148034447): 1. Move the FULL/READS needle walk (walk_index_file over the captured ecx descriptor) into spawn_blocking. It reads the full .ecx synchronously and was still running on the async executor worker, the same blocker the index_plan.run() fix in the previous commit addressed. 2. Preserve JoinError classification in both spawn_blocking join points in scrub_ec_volume_distributed. A panic is evidence about the volume and counts as broken; a cancellation only happens at runtime shutdown, the volume was never scanned, and returning it as an error would put a false corruption into broken_volume_ids (the FULL/READS arm marks the volume broken on any non-empty errs). Panics return an error; cancellations return clean. 3. Do not treat encode_ts_ns == 0 as a verified generation match. The .vif assigns 0 when it carries no encode-run identity (legacy/pre-feature volumes), so 0 == 0 would accept a teardown-and-remount and apply the old .ecx offsets to the replacement volume's shards. Only enforce the generation check when the captured 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. --------- Co-authored-by: Claude Opus 5 (1M context) Co-authored-by: Chris Lu --- seaweed-volume/src/server/grpc_server.rs | 880 ++++++++++---- seaweed-volume/src/server/store_ec.rs | 170 ++- .../src/storage/erasure_coding/ec_bitrot.rs | 18 +- .../src/storage/erasure_coding/ec_shard.rs | 15 + .../src/storage/erasure_coding/ec_volume.rs | 1064 ++++++++++++----- weed/server/volume_grpc_scrub.go | 41 +- weed/server/volume_grpc_scrub_test.go | 85 ++ 7 files changed, 1705 insertions(+), 568 deletions(-) create mode 100644 weed/server/volume_grpc_scrub_test.go 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") +}