Github-Mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-20 05:20:49 +02:00
Code Issues Packages Projects Releases Wiki Activity

volume server: release the store lock before scrubbing EC volumes (#11235)

Browse Source 
* 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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>
This commit is contained in:
Eliah Rusin
2026-09-08 19:12:30 -07:00
committed by GitHub
co-authored by Claude Opus 5 Chris Lu
parent 723f473f02
commit 9b12d13934
7 changed files with 1705 additions and 568 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+617 -263
View File
@@ -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<Status> {
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<u32>,
details: &mut Vec<String>,
) {
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<usize>) {
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<VolumeId>,
explicit: bool,
) -> Result<Response<volume_server_pb::ScrubVolumeResponse>, Status> {
let mode = req.mode;
let mut total_volumes: u64 = 0;
let mut total_files: u64 = 0;
let mut broken_volume_ids: Vec<u32> = Vec::new();
let mut details: Vec<String> = Vec::new();
let mut broken_vids: Vec<VolumeId> = 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<String> = 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<VolumeId>,
explicit: bool,
) -> Result<Response<volume_server_pb::ScrubEcVolumeResponse>, 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<u32> = Vec::new();
let mut broken_shard_infos: Vec<volume_server_pb::EcShardInfo> = Vec::new();
let mut details: Vec<String> = 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<u32> =
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<u32> = Vec::new();
let mut details: Vec<String> = Vec::new();
let mut broken_vids: Vec<VolumeId> = Vec::new();
let explicit = !req.volume_ids.is_empty();
let vids: Vec<VolumeId> = 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<VolumeId> = 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<String> = 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<VolumeId> = {
// 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<VolumeId> = 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<u32> = Vec::new();
let mut broken_shard_infos: Vec<volume_server_pb::EcShardInfo> = Vec::new();
let mut details: Vec<String> = 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<u32> =
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<volume_server_pb::QueriedStripe>;
@@ -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,
seaweed-volume/src/server/store_ec.rs
+143 -27
View File
@@ -259,12 +259,19 @@ pub async fn scrub_ec_volume_distributed(
force_deleted_needles_check: bool,
recover_unreadable: bool,
) -> (i64, Vec<crate::pb::volume_server_pb::EcShardInfo>, Vec<String>) {
// Phase A — under the Store read lock, run the index scrub and grab the
// 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<String>) {
let mut count: i64 = 0;
let mut needles: Vec<(NeedleId, Offset, Size)> = Vec::new();
let mut walk_errs: Vec<String> = 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<ShardId, crate::pb::volume_server_pb::EcShardInfo> = 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<ScrubSnapshot> {
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(
seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs
+16 -2
View File
@@ -505,7 +505,21 @@ pub fn verify_shard_file_blocks(
entry: &EcShardChecksums,
block_size: i64,
) -> io::Result<Vec<usize>> {
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<Vec<usize>> {
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);
}
seaweed-volume/src/storage/erasure_coding/ec_shard.rs
+15
View File
@@ -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<File> {
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 {
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+792 -272
View File
File diff suppressed because it is too large Load Diff
weed/server/volume_grpc_scrub.go
+37 -4
View File
@@ -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
weed/server/volume_grpc_scrub_test.go
+85
View File
@@ -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")
}