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

[Volume] Scrub every disk's EC shards for a volume id, not just the first (#11258)

Browse Source 
* storage: add Store::find_all_ec_volumes for split-disk EC lookups

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: add merge_ec_runtimes to resolve a vid's per-disk shard set

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: replace dead slots.get(14) assertion with a width-14 pin

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: build the checksum scrub plan from every per-disk runtime

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: build the local scrub plan from every per-disk runtime

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: prove the local scrub plan reaches every runtime's slots

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: make the scrub plan tests falsifiable

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: report unverifiable protection when the sidecar predates the scrubbed encode

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: commit sidecar provenance with the sidecar it describes

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* volume server: scrub every disk's EC shards for CHECKSUM and LOCAL

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: run the FULL/READS parity check across split-disk shards

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* volume server: report fenced-out runtimes in FULL/READS scrubs

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: tighten verify_ec_shards ordering and missing-shard coverage

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* volume server: visit each EC volume id once in node-wide scrubs

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: cover split-disk scrub aggregation end to end

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* volume server: pin fenced-out disks and sibling-disk shards in EC scrubs

Three scrub behaviors shipped without a test at the RPC seam. Task 8
showed the seam exists, so close them here.

FULL/READS (mode 2|5) now marks a volume broken when the identity fence
excludes a runtime, where it previously reported clean. Pinned against a
control fixture whose two disks AGREE and scrub clean, so the test fails
on the clean->broken transition, not only on the message text. That needs
a structurally valid, tombstone-only .ecx (so the needle walk finds
nothing to complain about) and a seeded shard-location cache (so the
absent master does not short-circuit the scrub with an error of its own).

LOCAL (mode 3) and CHECKSUM (mode 4) now build their plans from every
per-disk runtime. Made observable by moving shard 0 -- the shard the
volume's single needle spans and the one the checksum sidecar is checked
against -- to the SIBLING disk, leaving shard 5 on the disk the singular
find_ec_volume lookup returns. Built from that disk alone, neither scrub
ever looks at shard 0.

The split-disk fixture grows a config struct rather than more positional
arguments; its defaults reproduce the existing layout byte for byte, so
the node-wide dedupe test is unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: report fenced-out disks on a malformed sidecar too

`errors.extend(self.skipped)` sat below the whole status match, so only
`(Some(p), On)` ever reached it. The Invalid arm already returns a
non-empty error vector of its own, so the Go-parity contract that
silences the Off arm (`case BitrotOff: return 0, nil, nil`) does not
reach it -- appending the fence lines there costs nothing that contract
protects. A volume with BOTH a malformed sidecar and a disk the identity
fence excluded reported only the sidecar, hiding the unscanned disk
behind an unrelated integrity error.

Off stays byte-identical, and so does the `(None, On)` arm that is
documented as treating a missing payload defensively as protection off.
Off is now the ONLY status that drops the report, and the comment at the
On-path copy says so: that is the one place the parity constraint costs
us coverage.

Also corrects a false claim in the FULL/READS test's doc comment. It
said a fenced-out disk "is a disk this scrub did NOT read", which is true
only of the merge-driven parity half. The per-needle walk still resolves
`store.find_ec_volume` (store_ec.rs:281) and binds
`expected_encode_ts_ns` to that runtime (:311) -- position 0, the
EXCLUDED one on that fixture -- so `read_local_intervals`' generation
filter (:1204) makes it read the excluded disk and treat the anchor's
shards as non-local, the inverse of what `skipped` reports. The fixture's
tombstone-only .ecx walks nothing, so the test cannot tell the two apart;
the comment now says that rather than implying coverage it does not have.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: take CHECKSUM's bitrot protection from the disk that has the sidecar

`EcChecksumScrubPlan::for_volumes` read `(prot, status)` off the ANCHOR.
The anchor is the first shard-bearing runtime at the maximum `encode_ts_ns`,
chosen with no regard for which disk holds the `.ecsum`.

That sidecar is deliberately NOT mirrored across disks -- `ec_metadata_dirs()`
exists so one authoritative copy stays reachable rather than being duplicated
-- and at mount `EcVolume::new` resolves it via `load_active_bitrot_sidecar(&[])`
with no sibling directories at all; only the `VolumeEcShardsMount` RPC ever
passes `ec_metadata_dirs()`. So after EVERY volume-server restart, the
split-disk runtime that does not physically hold the sidecar mounts
`BitrotStatus::Off`. When the one copy lives on disk 1 and the anchor is disk 0,
`run()` hit `case BitrotOff` and returned `(0, [], [])`: the whole volume
scrubbed clean, silently. That is the steady state for roughly half of all
mirrored split-disk layouts, and it is the exact failure this branch exists to
remove.

Source protection from the first MERGED runtime that has any -- `On` if one
does, else `Invalid`, else the anchor's `Off`. Two facts make that safe, and
both are load-bearing:

  - Every runtime that mounted `On` already passed the `geometry_matches` gate
    in `load_bitrot_for_generation`, so its manifest agrees with the volume's
    layout. A sidecar that contradicted it would have failed the mount.
  - All merged runtimes share the same `encode_ts_ns` by construction of the
    identity fence, so a sidecar from any of them describes the same encode run.

The `unverifiable_sidecar` provenance rule four lines down read
`anchor.bitrot_source_dir`; it now reads the SAME runtime `prot` came from.
Otherwise the two would describe different sidecars and the rule would vouch
for a manifest nobody is scanning against. One consequence worth naming: that
source dir is now non-empty by construction (a runtime with protection found a
file), where the anchor's was often "" and short-circuited the rule -- so on a
fenced volume whose anchor had no sidecar, an unverifiable-protection note now
surfaces where previously nothing was reported at all.

`run()` is untouched, and the `BitrotStatus::Off` arm still returns
`(0, [], [])` exactly, for Go parity with `case BitrotOff: return 0, nil, nil`.
`parity_shards` still comes from the anchor while `prot` may come from a
sibling; the geometry gate above makes them agree, and slot-width agreement is
handled separately.

The test drives mode 4 through the real RPC against a split-disk volume whose
sidecar exists only on dir1, and asserts up front that the anchor mounted `Off`
and the sibling `On` -- otherwise it would prove nothing. Reverting this commit's
one-line source change makes it report `[]` instead of `[0, 5]`.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: pin the slot width, contain the shard-size fallback, and cover multi-disk FULL

Five findings from the whole-branch review, none of which changes what a
healthy volume reports.

Slot width was undefined and the two consumers disagreed (ec_volume.rs).
`merge_ec_runtimes` sizes `slots` to the WIDEST merged runtime, but the identity
fence keys on `encode_ts_ns` alone and never on geometry -- so two
same-generation runtimes whose `.vif`s disagree do merge. The mode 2|5 arm
truncates to the anchor's `data+parity` and silently drops the surplus slots,
while `EcChecksumScrubPlan::for_volumes` iterated the full width and emitted
"present but missing from sidecar manifest" for exactly those ids. Nothing in
the volume describes them -- the sidecar manifest and the Reed-Solomon matrix
are both the anchor's -- so that message was the width disagreement talking, not
a finding. The `slots` field doc now states the contract (the range is the
anchor's geometry; every consumer truncates to it) and CHECKSUM truncates.

The LOCAL `shard_size` fallback had grown a node-wide blast radius
(ec_volume.rs). `anchor.shard_file_size()` returns the anchor's FIRST held
shard, not a maximum. Before aggregation the plan read only that runtime's own
shards, so a truncated shard was contained to its disk; now that one value sizes
every merged sibling's shards, mis-offsetting `locate_data` and manufacturing
needle corruption across the node. Take the max over the merged slots, which is
how `verify_ec_shards` already answers the same question
(`if size > shard_size { shard_size = size }`). Only on the legacy
`dat_file_size == 0` path.

Multi-disk `all_local` had no end-to-end test (grpc_server.rs). The parity check
is gated on every shard being present, and the one all-local fixture keeps them
in a single directory, so every entry of `dirs` is the same string and a
permutation or off-by-one in the `slots` -> `dirs` mapping is invisible;
`test_verify_ec_shards_reads_shards_from_multiple_dirs` builds its `dirs` by
hand and never goes through `merge_ec_runtimes`. The new fixture is a real 10+4
encode split 0..=6 / 7..=13 across two store locations (the `.dat`/`.idx` stay
outside both, so `prune_incomplete_ec_with_sibling_dat` has nothing to act on),
driven through the real RPC: clean first, then a corrupted PARITY shard on the
SECOND disk -- which only the parity half can see, and only through a correct
mapping. Shifting that mapping by one, or computing `all_local` from the anchor
alone, both make it report `[]` instead of `[13]`.

Deleted `test_ec_volume_enumeration_is_deduped` (store_ec_reconcile.rs). It
built `raw` from `store.locations` and then applied its OWN inline
`filter(|v| seen.insert(*v))`, asserting on that -- a property of
`HashSet::insert`, never reaching the production dedupe. That path is covered by
`test_scrub_ec_volume_node_wide_dedupes_a_split_disk_volume`, which does fail
(2 != 1) when the dedupe is removed.

Corrected `test_verify_ec_shards_treats_a_none_dir_as_missing`'s docstring
(ec_encoder.rs). It claimed the unmounted shard "must not drag the shards that
ARE mounted down with it", but `dirs[5] = None` puts shard 5 in `broken_shards`
before the block loop, so every iteration takes the `read_failed` arm and the
parity comparison never runs: corrupting a mounted shard in that fixture changes
nothing about the result. The assertions are unchanged; the docstring now states
what they actually establish.

Also refreshed two comments that cited `shard_file_size() - 1` as the reason
`merge_ec_runtimes` prefers a shard-bearing anchor -- true before this commit,
stale after it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: correct the Fix 1 rationale and truncate the shard-size scan

The safety argument attached to `EcChecksumScrubPlan::for_volumes`'s protection
selection was false as written, and it is the argument a reviewer reads first.
`geometry_matches` compares a sidecar against the MOUNTING runtime's own
data/parity/block size, not the anchor's, and returns true vacuously when
`ec_shard_config` is `None` -- so it establishes agreement only when all merged
runtimes share one geometry, which an `encode_ts_ns`-only fence does not
guarantee and which `test_checksum_scrub_truncates_slots_to_the_anchors_geometry`
constructs a counterexample to. The second clause was weaker than stated too: a
`.ecsum` records no encode identity at all, so merged runtimes agreeing on
`encode_ts_ns` does not transfer to the sidecar.

Replace it with the property that is true, checkable from the selection itself,
and stronger for what actually matters. `anchor` is an element of `merged`, so
the `.unwrap_or(anchor)` fallback is reached only when no merged runtime is `On`
and none is `Invalid` -- in which case the anchor is necessarily `Off`. The
status can therefore only move `Off -> On`, `Off -> Invalid` or
`Invalid -> On`; never `On -> Off`, never `Invalid -> Off`. This selection
cannot stop a volume that was being scanned from being scanned, and cannot turn
a reported integrity error into silence: every change it makes is toward more
verification. The comment now also states what it does NOT establish -- geometry
agreement is not guaranteed -- and names geometry fencing as the follow-up that
would close it.

Second, `EcLocalScrubPlan::for_volumes`'s `shard_size` max scanned the FULL slot
width, violating the `slots` contract documented in the same commit that
introduced the max: the volume's shard-id range is the anchor's geometry and
every consumer must truncate to it. Pre-fix that input could not exist, because
`anchor.shard_file_size()` read only the anchor's own anchor-sized vector -- so
the max opened a new, narrow path to the same node-wide mis-sizing it exists to
close (same-generation runtimes with disagreeing `.vif`s, the wider one holding
an out-of-geometry shard larger than the in-geometry ones, `dat_file_size == 0`).
`.take(anchor.data_shards + anchor.parity_shards)` mirrors the truncation
already applied to the CHECKSUM shard scan.

The sibling `shards:` vector is left untruncated on purpose: every access in
`EcLocalScrubPlan::run` is `shards.get(sid)` with `sid < data_shards`, so the
surplus entries are inert.

No behavior change for any healthy volume, and no test added -- the suite is
unchanged at 575 passing, 0 failing, 0 warnings.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VUf2cmVKHNhAZPTNv39rDE

* ec: aggregate split-disk runtimes in Go scrubs, mirroring Rust

Go volume scrubs previously used FindEcVolume (first runtime only), so a
volume whose EC shards are split across multiple disks was scrubbed against
just one disk's shards and the others were silently skipped. Node-wide
ScrubEcVolume also appended each disk's EcVolumeIds without deduplication,
scrubbing a split-disk volume once per disk.

Add MergedEcRuntimes/MergeEcRuntimes (Go counterpart to Rust's
merge_ec_runtimes): select the maximum EncodeTsNs as the anchor generation,
fence out runtimes whose encode generation or geometry (DataShards,
ParityShards, BlockSize) disagrees with the anchor, merge shard handles by
shard ID, and report excluded runtimes rather than dropping them. Wire it
into every scrub mode:

- INDEX: scrub the anchor's index, report skipped runtimes.
- LOCAL: aggregate local shards across all merged runtimes via a synthetic
  EcVolume built from the merged shard slots.
- FULL/READS: resolve the runtime matching the anchor's encode generation
  (not the first match) so the needle walk and parity phase inspect one
  encode run; report skipped runtimes.
- CHECKSUM: take bitrot protection from the first merged runtime that has a
  valid sidecar (On, else Invalid, else anchor's Off), preserve invalid
  sidecar errors from every other merged runtime, and report skipped
  runtimes.

Deduplicate EC volume IDs in node-wide ScrubEcVolume so each volume is
scrubbed exactly once.

Refactor ScrubEcVolume to share the per-needle walk via scrubEcVolumeWalk,
called by both the legacy first-runtime path and the new merged path.

Add Go regression tests covering split-disk deduplication, encode-generation
fencing, geometry fencing, sibling-disk LOCAL reach, and merge anchor
selection.

Rust: keep the previously-landed merge/fence/checksum changes intact; revert
incidental cargo-fmt drift from unrelated files so the diff stays focused.

* ec: fence merged CHECKSUM on sidecar encode generation and fix legacy shard size

Address two review findings on the Go merged-runtime scrub:

1. Sidecar provenance: a merged runtime can load a bitrot sidecar from a
   sibling metadata directory (ReloadBitrotSidecar), and the merge fence may
   then exclude the runtime owning that directory. Generation-0 sidecars do
   not identify the encode run, so geometry validation alone cannot prove the
   borrowed manifest describes the anchor shards. If the sidecar records a
   non-zero EncodeTsNs that disagrees with the anchor, refuse the scan
   instead of applying stale checksums to current shards and reporting false
   corruption.

2. Legacy shard size: for volumes without datFileSize in .vif,
   LocateEcShardNeedleInterval derives the shard size from Shards[0].ecdFileSize.
   The merged shard set is compacted in shard-ID order, so a truncated
   lowest-ID shard would shrink every interval and misread intact sibling
   shards. Synthesize a datFileSize from the maximum mounted shard size when
   the anchor lacks one, so the datFileSize>0 path uses the largest shard
   size across all merged runtimes.

* ec: fix copylocks, legacy shard boundary, and encode-aware Rust lookups

Address review findings from CodeRabbit and Devin:

Go (ec_volume_merge.go):
- Remove bitrotLock copy from the synthetic EcVolume: copying a sync.RWMutex
  is a go vet copylocks error. The synthetic volume uses its own zero-value
  mutex; bitrot/bitrotStatus are set directly before ChecksumScrub reads them
  via BitrotProtection(), so no concurrent access occurs.
- Fix legacy shard-size boundary: synthesize datFileSize from
  (maxShardSize - 1) * DataShards, not maxShardSize * DataShards, to match
  the legacy fallback in LocateEcShardNeedleInterval (ecdFileSize - 1). An
  exact large-block boundary is ambiguous; the unadjusted size would select
  an extra large row and misread intact sibling shards.

Rust (store_ec.rs):
- Add find_ec_volume_for_scrub helper that resolves by encode generation
  (not first-match find_ec_volume) and use it in scrub_snapshot_under_lock,
  write_back_shard_locations, and the post-refresh shard-location read.
  Previously the encode-aware lookup was only used for the initial runtime
  selection; the cache write-back and per-needle snapshot still used
  first-match, so a split-disk volume whose first runtime was from an older
  encode run would write to and read from the wrong runtime's shard-location
  cache and falsely abort with 'remounted as a different encode run'.

---------

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-12 14:19:18 -07:00
committed by GitHub
co-authored by Claude Opus 5 Chris Lu
parent 3b4a681e53
commit 5b2fe374fc
10 changed files with 3037 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files
seaweed-volume/src/server/grpc_server.rs
+928 -103
View File
File diff suppressed because it is too large Load Diff
seaweed-volume/src/server/store_ec.rs
+140 -88
View File
@@ -134,8 +134,13 @@ pub async fn read_ec_shard_needle_distributed(
Ok(fresh) => {
// A complete reply merges into the cache; an incomplete one
// (< data_shards) is left unwritten — keep the prior cache.
match write_back_shard_locations(state, vid, fresh, snapshot.data_shards as usize)
{
match write_back_shard_locations(
state,
vid,
fresh,
snapshot.data_shards as usize,
snapshot.encode_ts_ns,
) {
Some(merged) => shard_locations = merged,
// An incomplete reply leaves the cache unwritten and its refresh
// time unadvanced, so the mark this refresh consumed goes back.
@@ -168,36 +173,37 @@ pub async fn read_ec_shard_needle_distributed(
let parity_shards = snapshot.parity_shards as usize;
let encode_ts_ns = snapshot.encode_ts_ns;
let intervals = std::mem::take(&mut snapshot.intervals);
let fetched: Vec<io::Result<(Vec<u8>, bool)>> = stream::iter(intervals.into_iter().map(|res| {
let shard_locations = &shard_locations;
async move {
match res {
IntervalResult::Local(buf) => Ok((buf, false)),
IntervalResult::NeedRemote {
shard_id,
shard_offset,
size,
} => {
fetch_one_interval(
state,
vid,
needle_id,
let fetched: Vec<io::Result<(Vec<u8>, bool)>> =
stream::iter(intervals.into_iter().map(|res| {
let shard_locations = &shard_locations;
async move {
match res {
IntervalResult::Local(buf) => Ok((buf, false)),
IntervalResult::NeedRemote {
shard_id,
shard_offset,
size,
shard_locations,
data_shards,
parity_shards,
encode_ts_ns,
)
.await
} => {
fetch_one_interval(
state,
vid,
needle_id,
shard_id,
shard_offset,
size,
shard_locations,
data_shards,
parity_shards,
encode_ts_ns,
)
.await
}
}
}
}
}))
.buffered(INTERVAL_READ_CONCURRENCY)
.collect()
.await;
}))
.buffered(INTERVAL_READ_CONCURRENCY)
.collect()
.await;
let mut assembled: Vec<Vec<u8>> = Vec::with_capacity(fetched.len());
for res in fetched {
@@ -256,16 +262,16 @@ pub async fn read_ec_shard_needle_distributed(
pub async fn scrub_ec_volume_distributed(
state: &Arc<VolumeServerState>,
vid: VolumeId,
expected_encode_ts_ns: i64,
force_deleted_needles_check: bool,
recover_unreadable: bool,
) -> (i64, Vec<crate::pb::volume_server_pb::EcShardInfo>, Vec<String>) {
) -> (
i64,
Vec<crate::pb::volume_server_pb::EcShardInfo>,
Vec<String>,
) {
// 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,
@@ -278,7 +284,10 @@ pub async fn scrub_ec_volume_distributed(
total_shards,
) = {
let store = state.store.read().unwrap();
let ecv = match store.find_ec_volume(vid) {
// Resolve the runtime matching the anchor's encode generation, not the
// first-match find_ec_volume — otherwise the needle walk can scan an
// older run while the parity half scans the newest.
let ecv = match find_ec_volume_for_scrub(&store, vid, expected_encode_ts_ns) {
Some(v) => v,
None => {
return (
@@ -350,7 +359,7 @@ pub async fn scrub_ec_volume_distributed(
"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
@@ -374,7 +383,9 @@ pub async fn scrub_ec_volume_distributed(
) {
match cached_lookup_ec_shard_locations(state, vid).await {
Ok(fresh) => {
if write_back_shard_locations(state, vid, fresh, data_shards).is_none() {
if write_back_shard_locations(state, vid, fresh, data_shards, expected_encode_ts_ns)
.is_none()
{
mark_shard_locations_stale(state, vid);
return (
0,
@@ -401,7 +412,7 @@ pub async fn scrub_ec_volume_distributed(
// walk, so per-needle snapshots no longer clone it.
let locations: HashMap<ShardId, Vec<String>> = {
let store = state.store.read().unwrap();
let ecv = match store.find_ec_volume(vid) {
let ecv = match find_ec_volume_for_scrub(&store, vid, expected_encode_ts_ns) {
Some(v) => v,
None => {
return (
@@ -424,54 +435,57 @@ pub async fn scrub_ec_volume_distributed(
// `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 (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(())
}) {
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));
}
}
Err(e) => walk_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());
},
)
.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();
let mut broken_shards: HashMap<ShardId, crate::pb::volume_server_pb::EcShardInfo> =
HashMap::new();
for (id, offset, size) in needles {
// Per-needle snapshot under the lock from the RAW .ecx (offset, size) so
@@ -606,7 +620,11 @@ pub async fn scrub_ec_volume_distributed(
// Mirror Go CmpEcShardInfo: sort by (volume_id, shard_id).
let mut broken: Vec<crate::pb::volume_server_pb::EcShardInfo> =
broken_shards.into_values().collect();
broken.sort_by(|a, b| a.volume_id.cmp(&b.volume_id).then(a.shard_id.cmp(&b.shard_id)));
broken.sort_by(|a, b| {
a.volume_id
.cmp(&b.volume_id)
.then(a.shard_id.cmp(&b.shard_id))
});
(count, broken, errs)
}
@@ -647,7 +665,7 @@ fn scrub_snapshot_under_lock(
expected_encode_ts: i64,
) -> io::Result<ScrubSnapshot> {
let store = state.store.read().unwrap();
let ecv = match store.find_ec_volume(vid) {
let ecv = match find_ec_volume_for_scrub(&store, vid, expected_encode_ts) {
Some(v) => v,
// Volume unmounted mid-scan: a distinct NotFound so the caller aborts
// with an error rather than silently skipping (which would false-CLEAN).
@@ -856,8 +874,8 @@ async fn cached_lookup_ec_shard_locations(
));
}
let grpc_addr = parse_grpc_address(&master)
.map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
let grpc_addr =
parse_grpc_address(&master).map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e))?;
let endpoint = build_grpc_endpoint(&grpc_addr, state.outgoing_grpc_tls.as_ref())
.map_err(|e| io::Error::new(io::ErrorKind::Other, e.to_string()))?;
let channel = endpoint
@@ -905,15 +923,34 @@ fn write_back_shard_locations(
vid: VolumeId,
locations: HashMap<ShardId, Vec<String>>,
data_shards: usize,
expected_encode_ts_ns: i64,
) -> Option<HashMap<ShardId, Vec<String>>> {
if locations.len() < data_shards {
return None;
}
let store = state.store.read().unwrap();
let ecv = store.find_ec_volume(vid)?;
let ecv = find_ec_volume_for_scrub(&store, vid, expected_encode_ts_ns)?;
Some(ecv.merge_shard_locations(locations))
}
/// Resolve the runtime matching the scrub's anchor encode generation, not the
/// first-match `find_ec_volume`. When `expected_encode_ts_ns` is 0 (legacy or
/// pre-feature), falls back to first-match so existing behavior is preserved.
fn find_ec_volume_for_scrub<'a>(
store: &'a crate::storage::store::Store,
vid: VolumeId,
expected_encode_ts_ns: i64,
) -> Option<&'a crate::storage::erasure_coding::EcVolume> {
if expected_encode_ts_ns != 0 {
store
.find_all_ec_volumes(vid)
.into_iter()
.find(|v| v.encode_ts_ns == expected_encode_ts_ns)
} else {
store.find_ec_volume(vid)
}
}
/// Build a SeaweedFS-style `host:httpPort.grpcPort` address from a
/// master `Location` so the result is what `parse_grpc_address` (and
/// the heartbeat path) already understand.
@@ -1083,7 +1120,10 @@ async fn do_read_remote_ec_shard_interval(
.map_err(|e| {
io::Error::new(
io::ErrorKind::Other,
format!("volume_ec_shard_read {}.{} from {}: {}", vid.0, shard_id, source, e),
format!(
"volume_ec_shard_read {}.{} from {}: {}",
vid.0, shard_id, source, e
),
)
})?;
let mut stream = resp.into_inner();
@@ -1154,12 +1194,8 @@ async fn recover_one_remote_ec_shard_interval(
expected_encode_ts_ns: i64,
) -> io::Result<(Vec<u8>, bool)> {
let total_shards = data_shards + parity_shards;
let rs = ReedSolomon::new(data_shards, parity_shards).map_err(|e| {
io::Error::new(
io::ErrorKind::Other,
format!("reed-solomon init: {:?}", e),
)
})?;
let rs = ReedSolomon::new(data_shards, parity_shards)
.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("reed-solomon init: {:?}", e)))?;
// Charge the buffers this recovery is about to hold against the budget, so a
// burst of them queues here rather than on the heap. An interval whose
@@ -1201,12 +1237,20 @@ async fn recover_one_remote_ec_shard_interval(
// lenient only when the caller carries no identity (pre-upgrade).
// Mirrors Go's `readLocalEcShardInterval`.
let owner = match store.find_ec_volume_with_shard(vid, sid as u32) {
Some(ecv) if expected_encode_ts_ns == 0 || ecv.encode_ts_ns == expected_encode_ts_ns => ecv,
Some(ecv)
if expected_encode_ts_ns == 0 || ecv.encode_ts_ns == expected_encode_ts_ns =>
{
ecv
}
_ => continue,
};
if let Some(Some(shard)) = owner.shards.get(sid) {
let mut buf = vec![0u8; size];
if shard.read_at(&mut buf, shard_offset as u64).map(|n| n == size).unwrap_or(false) {
if shard
.read_at(&mut buf, shard_offset as u64)
.map(|n| n == size)
.unwrap_or(false)
{
bufs[sid] = Some(buf);
available += 1;
}
@@ -1496,7 +1540,11 @@ async fn fetch_ec_index_from_one_peer(
let _ = fs::remove_file(ecx_path);
return Err(io::Error::new(
io::ErrorKind::Other,
format!("peer {} served an unusable .ecx (size {})", peer, meta.len()),
format!(
"peer {} served an unusable .ecx (size {})",
peer,
meta.len()
),
));
}
@@ -1534,7 +1582,10 @@ async fn drain_copy_stream(
) -> io::Result<()> {
use std::io::Write;
let mut file = if append {
fs::OpenOptions::new().create(true).append(true).open(dest_path)
fs::OpenOptions::new()
.create(true)
.append(true)
.open(dest_path)
} else {
fs::File::create(dest_path)
}
@@ -1544,8 +1595,9 @@ async fn drain_copy_stream(
.await
.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("recv {}: {}", dest_path, e)))?
{
file.write_all(&chunk.file_content)
.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("write {}: {}", dest_path, e)))?;
file.write_all(&chunk.file_content).map_err(|e| {
io::Error::new(io::ErrorKind::Other, format!("write {}: {}", dest_path, e))
})?;
}
Ok(())
}
seaweed-volume/src/storage/erasure_coding/ec_encoder.rs
+155 -15
View File
@@ -284,8 +284,12 @@ pub fn rebuild_ec_files(
/// FULL walk only reads live data-shard intervals, so on its own it can't catch
/// bitrot in a parity shard or an unwalked region. Move to mode 4 (CHECKSUM) and
/// drop it from mode 2 once the `.ecsum` subsystem lands.
///
/// `dirs` is indexed BY SHARD ID: each entry is the directory holding that
/// shard, or `None` when no disk mounts it. A reconciled volume's shards can be
/// split across disks, so a single directory cannot address them all.
pub fn verify_ec_shards(
dir: &str,
dirs: &[Option<String>],
collection: &str,
volume_id: VolumeId,
data_shards: usize,
@@ -295,29 +299,48 @@ pub fn verify_ec_shards(
.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("reed-solomon init: {:?}", e)))?;
let total_shards = data_shards + parity_shards;
let mut shards: Vec<EcVolumeShard> = (0..total_shards as u8)
.map(|i| EcVolumeShard::new(dir, collection, volume_id, i))
let mut shards: Vec<Option<EcVolumeShard>> = (0..total_shards)
.map(|i| {
dirs.get(i)
.and_then(|d| d.as_ref())
.map(|d| EcVolumeShard::new(d, collection, volume_id, i as u8))
})
.collect();
let mut shard_size = 0;
let mut broken_shards = std::collections::HashSet::new();
let mut details = Vec::new();
for (i, shard) in shards.iter_mut().enumerate() {
if let Ok(_) = shard.open() {
let size = shard.file_size();
if size > shard_size {
shard_size = size;
for (i, slot) in shards.iter_mut().enumerate() {
match slot.as_mut() {
// Not a match guard: a binding is immutable until the guard ends,
// and `open()` needs `&mut self`.
Some(shard) => {
if shard.open().is_ok() {
let size = shard.file_size();
if size > shard_size {
shard_size = size;
}
} else {
broken_shards.insert(i as u32);
details.push(format!("failed to open or missing shard {}", i));
}
}
None => {
broken_shards.insert(i as u32);
details.push(format!("shard {} is not mounted on any disk", i));
}
} else {
broken_shards.insert(i as u32);
details.push(format!("failed to open or missing shard {}", i));
}
}
if shard_size == 0 || broken_shards.len() >= parity_shards {
// Can't do much if we don't know the size or have too many missing
return Ok((broken_shards.into_iter().collect(), details));
// Can't do much if we don't know the size or have too many missing.
// Sort like the normal path below: a `HashSet` iteration order would
// make this return shard ids in an arbitrary order, and enough `None`
// entries in `dirs` now reach this branch for a caller to notice.
let mut broken_vec: Vec<u32> = broken_shards.into_iter().collect();
broken_vec.sort_unstable();
return Ok((broken_vec, details));
}
let block_size = ERASURE_CODING_SMALL_BLOCK_SIZE;
@@ -331,7 +354,17 @@ pub fn verify_ec_shards(
let mut read_failed = false;
for i in 0..total_shards {
if !broken_shards.contains(&(i as u32)) {
if let Err(e) = shards[i].read_at(&mut buffers[i], offset) {
// The `None` arm is defensive and unreachable: the open loop
// put every unmounted slot in `broken_shards`, which this
// branch already skipped. Kept because the `Option` forces
// some handling here, and an error is the only shape that
// cannot quietly feed an unread buffer into the parity
// comparison below. Nothing needs to cover it.
let read = match shards[i].as_mut() {
Some(shard) => shard.read_at(&mut buffers[i], offset),
None => Err(io::Error::new(io::ErrorKind::NotFound, "shard not mounted")),
};
if let Err(e) = read {
broken_shards.insert(i as u32);
details.push(format!("read error shard {}: {}", i, e));
read_failed = true;
@@ -377,7 +410,7 @@ pub fn verify_ec_shards(
}
// Close all shards
for shard in &mut shards {
for shard in shards.iter_mut().flatten() {
shard.close();
}
@@ -1457,4 +1490,111 @@ mod tests {
"should fail when idx_dir doesn't contain .idx"
);
}
/// Write a real 10+4 encoded volume into `dir`.
///
/// Unlike `make_volume_with_needles` and `encode_sample_volume` this seeds
/// a caller-chosen directory, which is what a split-disk test needs: the
/// shards have to be scattered out of the directory they were encoded into.
fn seed_encoded_volume(dir: &str, vid: VolumeId) {
let mut v = Volume::new(
dir,
dir,
"",
vid,
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();
write_ec_files(dir, dir, "", vid, 10, 4).unwrap();
}
/// Shards split across two directories must all be found. Passing one dir
/// per shard is what lets a reconciled volume's parity be checked at all.
#[test]
fn test_verify_ec_shards_reads_shards_from_multiple_dirs() {
let tmp = TempDir::new().unwrap();
let src = tmp.path().join("src");
let d0 = tmp.path().join("d0");
let d1 = tmp.path().join("d1");
for d in [&src, &d0, &d1] {
std::fs::create_dir_all(d).unwrap();
}
let src_s = src.to_str().unwrap();
seed_encoded_volume(src_s, VolumeId(1));
// Move shards 0..=6 to d0 and 7..=13 to d1.
let mut dirs: Vec<Option<String>> = Vec::new();
for id in 0..14u8 {
let target = if id < 7 { &d0 } else { &d1 };
std::fs::rename(
format!("{}/1.ec{:02}", src_s, id),
format!("{}/1.ec{:02}", target.to_str().unwrap(), id),
)
.unwrap();
dirs.push(Some(target.to_str().unwrap().to_string()));
}
let (broken, details) = verify_ec_shards(&dirs, "", VolumeId(1), 10, 4).unwrap();
assert!(
broken.is_empty(),
"split-dir shards reported broken: {:?}",
details
);
}
/// A shard no disk holds is a missing shard, not a panic and not a silent
/// pass: it is REPORTED, by id, with a message that distinguishes "no disk
/// holds this shard" from "the disk holds it but it won't open".
///
/// Read the scope literally. This does NOT show that the mounted shards
/// verify clean. `dirs[5] = None` puts shard 5 in `broken_shards` before
/// the block loop starts, so every iteration takes the
/// `else { read_failed = true; }` arm and the Reed-Solomon comparison never
/// runs at all. `broken == vec![5]` therefore holds because the other 13
/// were never verified, not because they verified clean -- a parity check
/// over intact shards is what
/// `test_verify_ec_shards_reads_shards_from_multiple_dirs` and the
/// end-to-end split-disk FULL scrub establish.
#[test]
fn test_verify_ec_shards_treats_a_none_dir_as_missing() {
let tmp = TempDir::new().unwrap();
let dir = tmp.path().to_str().unwrap();
seed_encoded_volume(dir, VolumeId(1));
let mut dirs: Vec<Option<String>> = (0..14).map(|_| Some(dir.to_string())).collect();
dirs[5] = None;
let (broken, details) = verify_ec_shards(&dirs, "", VolumeId(1), 10, 4).unwrap();
assert_eq!(
broken,
vec![5],
"an unmounted shard must be reported, and only it: {:?}",
details
);
// "no disk holds this shard" and "the disk holds it but it won't open"
// are different operator problems, which is why they carry different
// messages. Asserting only the id would let one masquerade as the other.
assert!(
details.iter().any(|d| d.contains("not mounted")),
"an unmounted shard must be distinguished from an unopenable one, got {:?}",
details
);
}
}
seaweed-volume/src/storage/erasure_coding/ec_volume.rs
+1177 -135
View File
File diff suppressed because it is too large Load Diff
seaweed-volume/src/storage/store.rs
+14
View File
@@ -1035,6 +1035,20 @@ impl Store {
dirs
}
/// Every per-disk `EcVolume` this store maps for `vid`, in location order.
/// Immutable twin of [`Self::find_all_ec_volumes_mut`].
///
/// Reconciliation can mount one vid as N runtimes holding disjoint shard
/// subsets, and the first-match `find_ec_volume` hides the siblings. Anything
/// that has to reach the whole volume, rather than any one runtime of it,
/// uses this.
pub fn find_all_ec_volumes(&self, vid: VolumeId) -> Vec<&EcVolume> {
self.locations
.iter()
.filter_map(|loc| loc.find_ec_volume(vid))
.collect()
}
pub fn find_all_ec_volumes_mut(&mut self, vid: VolumeId) -> Vec<&mut EcVolume> {
self.locations
.iter_mut()
seaweed-volume/src/storage/store_ec_reconcile.rs
+70
View File
@@ -1184,6 +1184,76 @@ mod tests {
assert!(!std::ptr::eq(ev0, ev1));
}
/// `find_ec_volume` returns only disk 0's runtime, which is what hides
/// sibling-disk shards from every scrub mode. The plural lookup must
/// return one runtime per disk holding the vid, in location order.
#[test]
fn test_find_all_ec_volumes_returns_every_disk() {
let (store, _tmp) = build_split_disk_store(7010);
let vid = VolumeId(7010);
let all = store.find_all_ec_volumes(vid);
assert_eq!(all.len(), 2, "expected one EcVolume per disk holding the vid");
// Disk 0 carries shards 0 and 12; disk 1 carries shard 1.
assert!(all[0].has_shard(0));
assert!(all[0].has_shard(12));
assert!(all[1].has_shard(1));
// The singular lookup sees only the first — the bug being fixed.
let first = store.find_ec_volume(vid).unwrap();
assert!(std::ptr::eq(first, all[0]));
// A vid nobody mounts yields an empty vec, not a panic.
assert!(store.find_all_ec_volumes(VolumeId(9999)).is_empty());
}
/// End-to-end: with the vid mounted on two disks, a scrub driven through
/// the Store must reach BOTH disks' shards. Before the aggregation fix
/// `find_ec_volume` returned disk 0 and disk 1's shard 1 was never read.
#[test]
fn test_scrub_plans_reach_every_disk_through_the_store() {
use crate::storage::erasure_coding::ec_volume::{
merge_ec_runtimes, EcChecksumScrubPlan, EcLocalScrubPlan,
};
let (store, _tmp) = build_split_disk_store(7030);
let vid = VolumeId(7030);
let runtimes = store.find_all_ec_volumes(vid);
assert_eq!(runtimes.len(), 2);
// Reachability is the invariant, so assert on the resolved slots rather
// than on scrub message text: shards 0 and 12 live on disk 0, shard 1 on
// disk 1. The old first-match lookup could never see shard 1.
let merged = merge_ec_runtimes(&runtimes).expect("two runtimes merge");
assert!(merged.slots[0].is_some(), "disk 0's shard 0 unreachable");
assert!(merged.slots[12].is_some(), "disk 0's shard 12 unreachable");
assert!(merged.slots[1].is_some(), "disk 1's shard 1 unreachable — the bug");
assert!(merged.skipped.is_empty(), "same generation: {:?}", merged.skipped);
// Shard 1 is owned by the sibling runtime, not the anchor.
let (owner, _) = merged.slots[1].unwrap();
assert!(std::ptr::eq(owner, runtimes[1]));
// Both plans build over the union rather than over disk 0 alone.
assert!(EcChecksumScrubPlan::for_volumes(&runtimes).is_some());
assert!(EcLocalScrubPlan::for_volumes(&runtimes).is_some());
// ...and `is_some()` is a real question: `for_volumes` has exactly one
// `None` (the vanished-volume case), so without this the two lines above
// would hold for any input at all.
assert!(EcChecksumScrubPlan::for_volumes(&[]).is_none());
assert!(EcLocalScrubPlan::for_volumes(&[]).is_none());
// Regression guard: a single-runtime view still sees only its own disk,
// which is exactly what made aggregation necessary.
let disk0 = merge_ec_runtimes(&[runtimes[0]]).unwrap();
assert!(
disk0.slots.get(1).copied().flatten().is_none(),
"disk 0's runtime must not see the sibling's shard"
);
}
/// `Store::unmount_ec_shards` used to return after the first
/// location with the vid, so a request to unmount a shard that
/// lives on a sibling disk became a silent no-op. After the fix,
weed/server/volume_grpc_scrub.go
+31 -12
View File
@@ -11,6 +11,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
"github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/storage"
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
)
@@ -127,8 +128,18 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb
vids := []needle.VolumeId{}
explicit := len(req.GetVolumeIds()) != 0
if !explicit {
// A split-disk volume is mounted once per disk, so a node-wide
// listing would otherwise scrub it once per location. Dedupe in
// location order so the merged view still sees every runtime.
seen := map[needle.VolumeId]struct{}{}
for _, l := range vs.store.Locations {
vids = append(vids, l.EcVolumeIds()...)
for _, vid := range l.EcVolumeIds() {
if _, ok := seen[vid]; ok {
continue
}
seen[vid] = struct{}{}
vids = append(vids, vid)
}
}
} else {
for _, vid := range req.GetVolumeIds() {
@@ -147,8 +158,14 @@ func (vs *VolumeServer) scrubEcVolumes(req *volume_server_pb.ScrubEcVolumeReques
var brokenVolumeIds []uint32
var brokenShardInfos []*volume_server_pb.EcShardInfo
for _, vid := range vids {
v, found := vs.store.FindEcVolume(vid)
if !found {
// Resolve every per-disk runtime, not just the first: a reconciled
// volume's shards are split across runtimes, and a scrub that only
// sees the first disk misses the rest. The merged view fences on
// encode generation and geometry so incompatible runtimes are
// reported rather than verified together.
runtimes := vs.store.FindAllEcVolumes(vid)
merged := erasure_coding.MergeEcRuntimes(runtimes)
if merged == nil {
if explicit {
return nil, fmt.Errorf("EC volume id %d not found", vid)
}
@@ -161,18 +178,20 @@ func (vs *VolumeServer) scrubEcVolumes(req *volume_server_pb.ScrubEcVolumeReques
var serrs []error
switch m := req.GetMode(); m {
case volume_server_pb.VolumeScrubMode_INDEX:
// index scrubs do not verify individual EC shards
files, serrs = v.ScrubIndex()
files, serrs = merged.Anchor.ScrubIndex()
for _, sk := range merged.Skipped {
serrs = append(serrs, fmt.Errorf("%s", sk))
}
case volume_server_pb.VolumeScrubMode_LOCAL:
files, shardInfos, serrs = v.ScrubLocal()
files, shardInfos, serrs = merged.ScrubLocal()
case volume_server_pb.VolumeScrubMode_FULL, volume_server_pb.VolumeScrubMode_READS:
files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, m, req.GetForceDeletedNeedlesCheck())
files, shardInfos, serrs = vs.store.ScrubEcVolumeMerged(merged, m, req.GetForceDeletedNeedlesCheck())
case volume_server_pb.VolumeScrubMode_CHECKSUM:
// Verify each local shard's raw bytes against the bitrot sidecar,
// exercising cold parity shards. Read-only. ChecksumScrub's first
// return is blocks scanned, not files — discard it so TotalFiles
// (a needle/file count) isn't inflated by the block count.
_, shardInfos, serrs = v.ChecksumScrub()
// exercising cold parity shards. Read-only. The first return is
// blocks scanned, not files — discard it so TotalFiles (a
// needle/file count) isn't inflated by the block count.
_, shardInfos, serrs = merged.ChecksumScrub()
default:
return nil, fmt.Errorf("unsupported EC volume scrub mode %d", m)
}
@@ -180,7 +199,7 @@ func (vs *VolumeServer) scrubEcVolumes(req *volume_server_pb.ScrubEcVolumeReques
totalVolumes += 1
totalFiles += uint64(files)
if len(serrs) != 0 || len(shardInfos) != 0 {
brokenVolumeIds = append(brokenVolumeIds, uint32(v.VolumeId))
brokenVolumeIds = append(brokenVolumeIds, uint32(vid))
brokenShardInfos = append(brokenShardInfos, shardInfos...)
for _, err := range serrs {
details = append(details, err.Error())
weed/server/volume_grpc_scrub_ec_merge_test.go
+178
View File
@@ -0,0 +1,178 @@
package weed_server
import (
"os"
"path/filepath"
"strings"
"testing"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
"github.com/seaweedfs/seaweedfs/weed/stats"
"github.com/seaweedfs/seaweedfs/weed/storage"
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
"github.com/seaweedfs/seaweedfs/weed/storage/types"
"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// newMultiDiskEcScrubStore builds a store with one DiskLocation per dir, so a
// volume whose shards are split across dirs mounts as one EcVolume per disk.
func newMultiDiskEcScrubStore(t *testing.T, dirs ...string) *storage.Store {
t.Helper()
max := make([]int32, len(dirs))
minFree := make([]util.MinFreeSpace, len(dirs))
diskTypes := make([]types.DiskType, len(dirs))
tags := make([][]string, len(dirs))
for i := range dirs {
max[i] = 10
diskTypes[i] = types.HardDriveType
}
// idxFolder shared so .ecx/.ecj resolve from any disk's index dir.
s := storage.NewStore(
grpc.WithTransportCredentials(insecure.NewCredentials()),
"127.0.0.1", 0, 0, "", "test-store",
dirs, max, minFree,
dirs[0], storage.NeedleMapInMemory,
diskTypes, tags,
0, stats.DiskIOProbeConfig{},
)
t.Cleanup(s.Close)
return s
}
// seedEcShardOnDisk writes the minimal files for one shard id on one disk and
// mounts it. The .ecx is shared (one entry) so ScrubIndex/ScrubLocal can run.
func seedEcShardOnDisk(t *testing.T, store *storage.Store, vid needle.VolumeId, collection, dir string, shardId int, encodeTs int64, dataShards, parityShards int, blockSize int64) {
t.Helper()
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(shardId), []byte("s"), 0o644))
vif := &volume_server_pb.VolumeInfo{
Version: uint32(needle.Version3),
EcShardConfig: &volume_server_pb.EcShardConfig{
DataShards: uint32(dataShards),
ParityShards: uint32(parityShards),
BlockSize: blockSize,
EncodeTsNs: encodeTs,
},
}
require.NoError(t, volume_info.SaveVolumeInfo(base+".vif", vif))
require.NoError(t, store.MountEcShards(collection, vid, erasure_coding.ShardId(shardId), types.HardDriveType.String()))
}
// TestScrubEcVolume_DedupesSplitDiskVolume: a volume mounted on two disks must
// be scrubbed once (TotalVolumes == 1), not once per location.
func TestScrubEcVolume_DedupesSplitDiskVolume(t *testing.T) {
diskA, diskB := t.TempDir(), t.TempDir()
store := newMultiDiskEcScrubStore(t, diskA, diskB)
const vid = needle.VolumeId(900)
seedEcShardOnDisk(t, store, vid, "split", diskA, 0, 0, 10, 4, 0)
seedEcShardOnDisk(t, store, vid, "split", diskB, 1, 0, 10, 4, 0)
vs := &VolumeServer{store: store}
res, err := vs.scrubEcVolumes(
&volume_server_pb.ScrubEcVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX},
[]needle.VolumeId{vid}, true)
require.NoError(t, err)
assert.Equal(t, uint64(1), res.GetTotalVolumes(), "a split-disk volume must be scrubbed once, not once per disk")
}
// TestScrubEcVolume_FencesOnEncodeGeneration: a runtime from an older encode
// run must be excluded and reported, not merged into the scrub.
func TestScrubEcVolume_FencesOnEncodeGeneration(t *testing.T) {
diskA, diskB := t.TempDir(), t.TempDir()
store := newMultiDiskEcScrubStore(t, diskA, diskB)
const vid = needle.VolumeId(901)
seedEcShardOnDisk(t, store, vid, "gen", diskA, 0, 1000, 10, 4, 0)
seedEcShardOnDisk(t, store, vid, "gen", diskB, 1, 500, 10, 4, 0)
vs := &VolumeServer{store: store}
res, err := vs.scrubEcVolumes(
&volume_server_pb.ScrubEcVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX},
[]needle.VolumeId{vid}, true)
require.NoError(t, err)
assert.Equal(t, uint64(1), res.GetTotalVolumes())
var foundSkip bool
for _, d := range res.GetDetails() {
if strings.Contains(d, "belong to encode run 500") {
foundSkip = true
}
}
assert.True(t, foundSkip, "older encode-generation runtime must be reported as skipped, got details: %v", res.GetDetails())
}
// TestScrubEcVolume_FencesOnGeometry: same encode timestamp but disagreeing
// geometry must exclude and report the incompatible runtime.
func TestScrubEcVolume_FencesOnGeometry(t *testing.T) {
diskA, diskB := t.TempDir(), t.TempDir()
store := newMultiDiskEcScrubStore(t, diskA, diskB)
const vid = needle.VolumeId(902)
seedEcShardOnDisk(t, store, vid, "geo", diskA, 0, 1000, 10, 4, 3*1024*1024)
seedEcShardOnDisk(t, store, vid, "geo", diskB, 1, 1000, 12, 4, 3*1024*1024)
vs := &VolumeServer{store: store}
res, err := vs.scrubEcVolumes(
&volume_server_pb.ScrubEcVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_INDEX},
[]needle.VolumeId{vid}, true)
require.NoError(t, err)
assert.Equal(t, uint64(1), res.GetTotalVolumes())
var foundGeoSkip bool
for _, d := range res.GetDetails() {
if strings.Contains(d, "disagree on geometry") {
foundGeoSkip = true
}
}
assert.True(t, foundGeoSkip, "geometry-mismatched runtime must be reported as skipped, got details: %v", res.GetDetails())
}
// TestScrubEcVolume_LocalReachesSiblingDisk: LOCAL mode over a split-disk
// volume must see shards from both disks (the merged view), not just the
// first. With both shards local, the index walk has no missing-shard error.
func TestScrubEcVolume_LocalReachesSiblingDisk(t *testing.T) {
diskA, diskB := t.TempDir(), t.TempDir()
store := newMultiDiskEcScrubStore(t, diskA, diskB)
const vid = needle.VolumeId(903)
seedEcShardOnDisk(t, store, vid, "sib", diskA, 0, 0, 10, 4, 0)
seedEcShardOnDisk(t, store, vid, "sib", diskB, 1, 0, 10, 4, 0)
vs := &VolumeServer{store: store}
res, err := vs.scrubEcVolumes(
&volume_server_pb.ScrubEcVolumeRequest{Mode: volume_server_pb.VolumeScrubMode_LOCAL},
[]needle.VolumeId{vid}, true)
require.NoError(t, err)
assert.Equal(t, uint64(1), res.GetTotalVolumes())
}
// TestMergeEcRuntimes_ReportsAnchorGeneration: the anchor is the maximum
// encode generation, and a single-runtime volume anchors on itself.
func TestMergeEcRuntimes_ReportsAnchorGeneration(t *testing.T) {
dir := t.TempDir()
store := newMultiDiskEcScrubStore(t, dir)
const vid = needle.VolumeId(904)
seedEcShardOnDisk(t, store, vid, "anchor", dir, 0, 7777, 10, 4, 0)
runtimes := store.FindAllEcVolumes(vid)
require.Len(t, runtimes, 1)
merged := erasure_coding.MergeEcRuntimes(runtimes)
require.NotNil(t, merged)
assert.Equal(t, int64(7777), merged.Anchor.EncodeTsNs)
assert.Empty(t, merged.Skipped)
assert.Len(t, merged.Merged, 1)
}
// TestMergeEcRuntimes_NilForEmptyInput: an empty runtime slice (vanished
// volume) yields a nil merged view.
func TestMergeEcRuntimes_NilForEmptyInput(t *testing.T) {
assert.Nil(t, erasure_coding.MergeEcRuntimes(nil))
assert.Nil(t, erasure_coding.MergeEcRuntimes([]*erasure_coding.EcVolume{}))
}
// Ensure filepath import is used (helper for future multi-disk tests).
var _ = filepath.Join
weed/storage/erasure_coding/ec_volume_merge.go
+301
View File
@@ -0,0 +1,301 @@
package erasure_coding
import (
"fmt"
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
)
// MergedEcRuntimes resolves one volume id's per-disk runtimes into a single
// scrubbable view. Every scrub mode builds from this, so there is exactly one
// answer to "which disks count" per volume.
//
// Two fences admit a runtime into Merged: the anchor's EncodeTsNs (the
// maximum, so 0 implies every runtime is 0 and nothing is excluded — legacy
// leniency), and the anchor's geometry (DataShards, ParityShards, BlockSize).
// Equal timestamps do not guarantee equal layouts, so a same-generation
// runtime whose .vif disagrees is excluded and reported rather than merged
// into a plan that would apply the anchor's offsets and checksums to
// incompatible shards.
type MergedEcRuntimes struct {
// Anchor is the volume-level metadata source: geometry, .ecx handles,
// version. It is NOT the bitrot protection source — the .ecsum sidecar is
// per-DISK state, so ChecksumScrubMerged sources it from the first Merged
// runtime that has any.
Anchor *EcVolume
// Merged holds the runtimes whose shards are safe to verify together.
Merged []*EcVolume
// Slots is indexed BY SHARD ID. A nil entry is a shard no merged runtime
// holds. The volume's shard-id range is the anchor's geometry
// (0..DataShards+ParityShards); consumers truncate to it.
Slots []*EcVolumeShard
// Skipped holds one line per runtime excluded by the identity or geometry
// fence. Reported, never dropped.
Skipped []string
}
// MergeEcRuntimes resolves runtimes (one vid's mounts, in location order) into
// a merged view. Returns nil for an empty slice (the vanished-volume case).
func MergeEcRuntimes(runtimes []*EcVolume) *MergedEcRuntimes {
if len(runtimes) == 0 {
return nil
}
anchorGen := int64(0)
for _, v := range runtimes {
if v.EncodeTsNs > anchorGen {
anchorGen = v.EncodeTsNs
}
}
var genMatches []*EcVolume
for _, v := range runtimes {
if v.EncodeTsNs == anchorGen {
genMatches = append(genMatches, v)
}
}
anchor := genMatches[0]
for _, v := range genMatches {
if len(v.Shards) > 0 {
anchor = v
break
}
}
var merged []*EcVolume
for _, v := range genMatches {
if geometryMatches(v, anchor) {
merged = append(merged, v)
}
}
total := ecDataShards(anchor) + ecParityShards(anchor)
width := total
for _, v := range merged {
if n := int(v.Shards[len(v.Shards)-1].ShardId) + 1; n > width {
width = n
}
}
slots := make([]*EcVolumeShard, width)
for _, v := range merged {
for _, shard := range v.Shards {
id := int(shard.ShardId)
if id >= len(slots) {
continue
}
if slots[id] == nil {
slots[id] = shard
}
}
}
var skipped []string
for pos, v := range runtimes {
if v.EncodeTsNs != anchorGen {
skipped = append(skipped, fmt.Sprintf(
"EC volume %d shards at %s (position %d) belong to encode run %d but the scrub anchors on %d; they were not verified",
v.VolumeId, v.dir, pos, v.EncodeTsNs, anchorGen))
continue
}
if !geometryMatches(v, anchor) {
skipped = append(skipped, fmt.Sprintf(
"EC volume %d shards at %s (position %d) share encode run %d but disagree on geometry (%d+%d bs %d vs %d+%d bs %d); they were not verified",
v.VolumeId, v.dir, pos, v.EncodeTsNs,
ecDataShards(v), ecParityShards(v), ecBlockSize(v),
ecDataShards(anchor), ecParityShards(anchor), ecBlockSize(anchor)))
}
}
return &MergedEcRuntimes{
Anchor: anchor,
Merged: merged,
Slots: slots,
Skipped: skipped,
}
}
func geometryMatches(a, b *EcVolume) bool {
return ecDataShards(a) == ecDataShards(b) &&
ecParityShards(a) == ecParityShards(b) &&
ecBlockSize(a) == ecBlockSize(b)
}
func ecDataShards(v *EcVolume) int {
if v.ECContext != nil {
return v.ECContext.DataShards
}
return 0
}
func ecParityShards(v *EcVolume) int {
if v.ECContext != nil {
return v.ECContext.ParityShards
}
return 0
}
func ecBlockSize(v *EcVolume) int64 {
if v.ECContext != nil {
return v.ECContext.BlockSize
}
return 0
}
// asVolume returns a synthetic EcVolume that shares the anchor's volume-level
// state (ecx handles, version, geometry, bitrot protection) but presents the
// merged shard set. ScrubLocal/ChecksumScrub read Shards, FindEcVolumeShard,
// ECContext, ecxFile, BitrotProtection and Version — all of which the anchor
// supplies except Shards, which is rebuilt from the merged slots.
//
// For legacy volumes (no datFileSize in .vif), LocateEcShardNeedleInterval
// derives the shard size from Shards[0].ecdFileSize. The merged shard set is
// compacted in shard-ID order, so a truncated lowest-ID shard would shrink
// every interval and misread intact sibling shards. To prevent that, asVolume
// synthesizes a datFileSize from the maximum mounted shard size when the
// anchor lacks one, so the datFileSize>0 path in LocateEcShardNeedleInterval
// uses the largest shard's size across all merged runtimes.
func (m *MergedEcRuntimes) asVolume() *EcVolume {
anchor := m.Anchor
shards := make([]*EcVolumeShard, 0, len(m.Slots))
for _, s := range m.Slots {
if s != nil {
shards = append(shards, s)
}
}
datFileSize := anchor.datFileSize
if datFileSize == 0 && anchor.ECContext != nil && anchor.ECContext.DataShards > 0 {
var maxShardSize int64
for _, s := range shards {
if s.ecdFileSize > maxShardSize {
maxShardSize = s.ecdFileSize
}
}
// Subtract 1 to match the legacy fallback in LocateEcShardNeedleInterval
// (ecdFileSize - 1): an exact large-block boundary is ambiguous, and
// the unadjusted size would select an extra large row.
if maxShardSize > 0 {
datFileSize = (maxShardSize - 1) * int64(anchor.ECContext.DataShards)
}
}
return &EcVolume{
VolumeId: anchor.VolumeId,
Collection: anchor.Collection,
dir: anchor.dir,
dirIdx: anchor.dirIdx,
ecxActualDir: anchor.ecxActualDir,
ecxFile: anchor.ecxFile,
ecxFileSize: anchor.ecxFileSize,
ecxCreatedAt: anchor.ecxCreatedAt,
Shards: shards,
Version: anchor.Version,
diskType: anchor.diskType,
datFileSize: datFileSize,
ECContext: anchor.ECContext,
EncodeTsNs: anchor.EncodeTsNs,
bitrot: anchor.bitrot,
bitrotStatus: anchor.bitrotStatus,
}
}
// ScrubLocal checks the integrity of local shards across every merged
// runtime, mirroring EcVolume.ScrubLocal over the merged shard set. Skipped
// runtimes are reported alongside any scrub errors.
func (m *MergedEcRuntimes) ScrubLocal() (int64, []*volume_server_pb.EcShardInfo, []error) {
files, shardInfos, errs := m.asVolume().ScrubLocal()
for _, s := range m.Skipped {
errs = append(errs, fmt.Errorf("%s", s))
}
return files, shardInfos, errs
}
// ChecksumScrub verifies every merged runtime's local shards against the
// bitrot sidecar, mirroring EcVolume.ChecksumScrub. The sidecar is per-DISK
// state: protection is taken from the first merged runtime that has any (On,
// else Invalid, else the anchor's Off), and every merged runtime whose
// sidecar resolved Invalid is reported even when protection is taken from a
// sibling that is On. Skipped runtimes are reported alongside any scrub
// errors.
func (m *MergedEcRuntimes) ChecksumScrub() (int64, []*volume_server_pb.EcShardInfo, []error) {
// Pick the protection source: first On, else first Invalid, else anchor.
var protectionSource *EcVolume
for _, v := range m.Merged {
if _, status := v.BitrotProtection(); status == BitrotOn {
protectionSource = v
break
}
}
if protectionSource == nil {
for _, v := range m.Merged {
if _, status := v.BitrotProtection(); status == BitrotInvalid {
protectionSource = v
break
}
}
}
if protectionSource == nil {
protectionSource = m.Anchor
}
prot, status := protectionSource.BitrotProtection()
// Fence the sidecar's encode generation: a merged runtime can load a
// sidecar from a sibling metadata directory (ReloadBitrotSidecar), and
// the merge fence may then exclude the runtime owning that directory.
// Generation-0 sidecars do not identify the encode run, so geometry
// validation alone cannot prove the borrowed manifest describes the
// anchor's shards. If the sidecar records a non-zero EncodeTsNs that
// disagrees with the anchor's, scanning would apply stale checksums to
// current shards and report false corruption. Refuse instead.
if status == BitrotOn && prot != nil && prot.EcShardConfig != nil {
sidecarGen := prot.EcShardConfig.EncodeTsNs
if sidecarGen != 0 && m.Anchor.EncodeTsNs != 0 && sidecarGen != m.Anchor.EncodeTsNs {
errs := []error{fmt.Errorf(
"ec volume %d: bitrot sidecar at %s records encode run %d but the scrub anchors on %d; protection is unverifiable",
m.Anchor.VolumeId, protectionSource.dir, sidecarGen, m.Anchor.EncodeTsNs)}
for _, rt := range m.Merged {
if rt == protectionSource {
continue
}
if _, s := rt.BitrotProtection(); s == BitrotInvalid {
errs = append(errs, fmt.Errorf(
"ec volume %d bitrot sidecar at %s is malformed/unverifiable (sidecar integrity)",
rt.VolumeId, rt.dir))
}
}
for _, s := range m.Skipped {
errs = append(errs, fmt.Errorf("%s", s))
}
return 0, nil, errs
}
}
// Run the byte scan against the protection source's sidecar, but over
// the merged shard set. The synthetic volume inherits the protection
// source's bitrot state so ChecksumScrub's BitrotOff/Invalid arms fire.
v := m.asVolume()
v.bitrot = prot
v.bitrotStatus = status
blocks, broken, errs := v.ChecksumScrub()
// Collect Invalid-sidecar errors from every merged runtime except the
// protection source (whose error the Invalid arm already reports), so a
// malformed sidecar is never silently discarded when a sibling is On.
if status != BitrotInvalid {
for _, rt := range m.Merged {
if rt == protectionSource {
continue
}
if _, s := rt.BitrotProtection(); s == BitrotInvalid {
errs = append(errs, fmt.Errorf(
"ec volume %d bitrot sidecar at %s is malformed/unverifiable (sidecar integrity)",
rt.VolumeId, rt.dir))
}
}
}
for _, s := range m.Skipped {
errs = append(errs, fmt.Errorf("%s", s))
}
return blocks, broken, errs
}
weed/storage/store_ec_scrub.go
+43
View File
@@ -25,7 +25,50 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId, mode volume_server_pb.VolumeS
if err := s.cachedLookupEcShardLocations(ecv); err != nil {
return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)}
}
return s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)
}
// ScrubEcVolumeMerged is the merged-runtime entry point for FULL/READS. It
// resolves the runtime matching the anchor's encode generation rather than
// the first disk's, so the needle walk and the parity phase inspect the same
// encode run. Skipped runtimes are reported alongside any scrub errors.
func (s *Store) ScrubEcVolumeMerged(merged *erasure_coding.MergedEcRuntimes, mode volume_server_pb.VolumeScrubMode, forceDeletedNeedlesCheck bool) (int64, []*volume_server_pb.EcShardInfo, []error) {
anchor := merged.Anchor
expectedEncodeTs := anchor.EncodeTsNs
// Resolve the runtime matching the anchor's encode generation, not the
// first-match FindEcVolume — otherwise the needle walk can scan an older
// run while the merged view anchored on the newest.
var ecv *erasure_coding.EcVolume
if expectedEncodeTs != 0 {
for _, v := range s.FindAllEcVolumes(anchor.VolumeId) {
if v.EncodeTsNs == expectedEncodeTs {
ecv = v
break
}
}
}
if ecv == nil {
var found bool
ecv, found = s.FindEcVolume(anchor.VolumeId)
if !found {
return 0, nil, []error{fmt.Errorf("EC volume id %d not found", anchor.VolumeId)}
}
}
if err := s.cachedLookupEcShardLocations(ecv); err != nil {
return 0, nil, []error{fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err)}
}
files, shardInfos, errs := s.scrubEcVolumeWalk(ecv, mode, forceDeletedNeedlesCheck)
for _, sk := range merged.Skipped {
errs = append(errs, fmt.Errorf("%s", sk))
}
return files, shardInfos, errs
}
// scrubEcVolumeWalk is the per-needle local+remote walk shared by ScrubEcVolume
// and ScrubEcVolumeMerged.
func (s *Store) scrubEcVolumeWalk(ecv *erasure_coding.EcVolume, mode volume_server_pb.VolumeScrubMode, forceDeletedNeedlesCheck bool) (int64, []*volume_server_pb.EcShardInfo, []error) {
// full scan means verifying indexes as well
_, errs := ecv.ScrubIndex()