From 51ee16527920861f79b4c8299d6fe32c07a049f6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 4 Nov 2026 09:41:21 -0800 Subject: [PATCH] ec: rebuild a needle when other shards are corrupt over its range (#12401) * storage: search corrupt shard sets when an EC needle fails to rebuild A needle whose reassembled bytes fail to parse is rebuilt with its own shards excluded. When other shards are corrupt over the same range, that decode reads them as inputs, and the retry dropped only one input at a time, so two or more extra corrupt shards left the needle unreadable even within parity. After the single pass fails, try every set of up to ParityShards shards that holds a suspect as the corrupt set, smallest first. Suspect ranges inside the set are rebuilt from the shards outside it and the rest keep their bytes, so a clean suspect can still feed the decode. Sets whose decodes read the same inputs are tried once, and the search is bounded by set count and by rebuilt bytes past the single-shard sets. Co-Authored-By: Claude Opus 5.5 * seaweed-volume: search corrupt shard sets when an EC needle fails to rebuild Mirror the Go read path: after the single pass over the suspects fails, gather each interval from every other shard once and try each set of up to parity_shards shards holding a suspect as the corrupt set, smallest first, with the same dedup and bounds. The search runs on the blocking pool, holding the recovery permit, since it can take many decodes. Co-Authored-By: Claude Opus 5.5 * ec: charge only multi-shard sets against the rebuild byte budget The budget check already let single-shard sets through, but each one was still charged, so a needle spanning many shards could spend the budget before the first multi-shard set. Co-Authored-By: Claude Opus 5.5 * ec: bound the corrupt-set gather to the recovery budget The search gathered every other shard per interval, so a large needle spanning several shards held up to Total-1 copies of itself under a permit capped at the budget. Gather only as many as the budget covers, but never fewer than the suspects-excluded gather the fallback always held, so single-suspect recovery keeps every other shard. Co-Authored-By: Claude Opus 5.5 * ec: gather at least one decode's worth for the corrupt-set search A needle over more shards than parity skips the single pass, and its suspects-excluded width can fall below DataShards, so the bounded gather left even one corrupt shard unrecoverable. Floor the width at DataShards, which the per-suspect rebuild this search replaced read. Co-Authored-By: Claude Opus 5.5 * ec: rebuild one suspect at a time when the search gather is narrow Flooring the search gather at DataShards made a large needle over more shards than parity hold ten copies of itself, where the per-suspect rebuild it replaced held one interval's fan-out. Keep that rebuild after the search, and skip the search when the budget leaves fewer inputs than one decode. Parity slack still widens the gather to every shard outside the suspects, as the dropped-input retry did. Co-Authored-By: Claude Opus 5.5 * ec: keep the dropped-input retry and run the set search within budget A narrowed search gather skipped only each interval's own shard, so it could read corrupt suspects in place of healthy shards the old retry kept, and its byte bound could stop before that retry's hypotheses. Restore the dropped-input retry as it was, move the search into its own function after it, and run the search only when every interval's full gather fits the recovery budget. Past the budget a read takes the same paths, with the same footprint, as before the search. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- seaweed-volume/src/server/store_ec.rs | 391 +++++++++++++++++++- weed/storage/store_ec.go | 187 +++++++++- weed/storage/store_ec_interval_read_test.go | 122 ++++++ 3 files changed, 690 insertions(+), 10 deletions(-) diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index e26920efa0..d56baf9c94 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -1799,10 +1799,11 @@ async fn try_ec_shard_needle_candidate( /// Repair a needle whose reassembled bytes failed to parse — Go's /// `rebuildEcShardNeedle`. Every interval's source shard is a suspect: rebuild -/// all suspect ranges at once while parity covers the suspect set, or one -/// suspect at a time when the needle spans more shards than that, and let a -/// successful parse arbitrate which copy was corrupt. A holder's deleted report -/// stops the search: the needle is gone whatever the local index says. +/// all suspect ranges at once while parity covers the suspect set, then search +/// the sets of shards that may be corrupt, then rebuild one suspect at a time, +/// and let a successful parse arbitrate which copy was corrupt. A holder's +/// deleted report stops the search: the needle is gone whatever the local index +/// says. async fn rebuild_ec_shard_needle( state: &Arc, metas: &[EcInterval], @@ -1840,9 +1841,19 @@ async fn rebuild_ec_shard_needle( return (Some(candidate), false); } } - if suspects.len() == 1 { - return (None, false); - } + } + let (candidate, is_deleted) = rebuild_ec_needle_from_corrupt_sets( + state, metas, map, corrupted, &suspects, offset, size, version, needle_id, + ) + .await; + if is_deleted { + return (None, true); + } + if let Some(candidate) = candidate { + return (Some(candidate), false); + } + if suspects.len() == 1 { + return (None, false); } for shard_id in suspects { let (candidate, is_deleted) = try_ec_shard_needle_candidate( @@ -1986,6 +1997,239 @@ async fn rebuild_ec_needle_from_dropped_inputs( (None, false) } +/// Bounds on one needle's corrupt-set search, as Go's `ecRebuildMaxSets` and +/// `ecRebuildMaxBytes`: sets enumerated, and needle bytes rebuilt and parsed +/// past the single-shard sets. 10+4 has 1470 sets of at most four shards. +const EC_REBUILD_MAX_SETS: usize = 4096; +const EC_REBUILD_MAX_BYTES: usize = 1 << 30; + +/// Handle a rebuild the retries above could not repair: more shards than one +/// can be corrupt over the same ranges and get read as reconstruction inputs. +/// Each interval is gathered once from every other shard that answers, then +/// `search_ec_corrupt_sets` tries the candidate corrupt sets. `None` when that +/// gather would outgrow the recovery budget. Mirrors Go's +/// `rebuildEcNeedleFromCorruptSets`. +async fn rebuild_ec_needle_from_corrupt_sets( + state: &Arc, + metas: &[EcInterval], + map: EcShardMap<'_>, + corrupted: &[u8], + suspects: &HashSet, + offset: i64, + size: Size, + version: Version, + needle_id: NeedleId, +) -> (Option>, bool) { + let EcShardMap { + data_shards, + parity_shards, + .. + } = map; + let total_shards = data_shards + parity_shards; + let total_size: usize = metas.iter().map(|iv| iv.size).sum(); + if total_size * (total_shards - 1) > EC_RECOVER_BUDGET { + return (None, false); + } + let Ok(permit) = EC_RECOVER_SEM + .acquire_many((total_size * total_shards).min(EC_RECOVER_BUDGET) as u32) + .await + else { + return (None, false); + }; + + let mut gathers: Vec>>> = Vec::with_capacity(metas.len()); + for iv in metas { + let (bufs, is_deleted) = + gather_ec_shard_intervals(state, *iv, map, &HashSet::from([iv.shard_id]), total_shards) + .await; + if is_deleted { + return (None, true); + } + gathers.push(bufs); + } + + // The search can run hundreds of decodes, so keep it off the runtime. The + // permit goes with it: a dropped read does not stop a blocking task. + let (metas_owned, corrupted, suspects) = (metas.to_vec(), corrupted.to_vec(), suspects.clone()); + let search = tokio::task::spawn_blocking(move || { + let _permit = permit; + let rs = ReedSolomon::new(data_shards, parity_shards).ok()?; + search_ec_corrupt_sets( + &mut gathers, + &metas_owned, + &rs, + &corrupted, + &suspects, + |candidate| { + let mut check = Needle { + id: needle_id, + ..Needle::default() + }; + check.read_bytes(candidate, offset, size, version).is_ok() && check.id == needle_id + }, + ) + }); + let Ok(Some((candidate, convicted))) = search.await else { + return (None, false); + }; + if let Some(first) = metas.first() { + for shard_id in convicted { + forget_shard_location(state, first.vid, shard_id); + } + } + (Some(candidate), false) +} + +/// Try each set of up to `parity_shards` shards holding a suspect as the +/// corrupt set, smallest first: suspect ranges inside it are rebuilt from the +/// shards outside it, the rest keep the bytes read, and `parses` arbitrates. A +/// set whose decodes read the same inputs as one already tried is skipped. +/// Returns the candidate with the shards it convicts: every differing range, +/// and every other shard the set kept out of the decodes. +fn search_ec_corrupt_sets( + gathers: &mut [Vec>>], + metas: &[EcInterval], + rs: &ReedSolomon, + corrupted: &[u8], + suspects: &HashSet, + parses: impl Fn(&[u8]) -> bool, +) -> Option<(Vec, Vec)> { + let data_shards = rs.data_shard_count(); + let total_shards = rs.total_shard_count(); + let mut candidate = corrupted.to_vec(); + let mut scratch: Vec> = vec![Vec::new(); data_shards]; + let mut inputs = vec![0u64; metas.len()]; + let mut tried: HashSet> = HashSet::new(); + + // Records in `inputs` the shards each interval's decode reads when `set` + // is excluded, 0 for an interval that keeps its bytes. False when an + // interval has too few shards left, or the same decodes were tried before. + let mut plan = |set: u64, inputs: &mut [u64], gathers: &[Vec>>]| { + for (i, iv) in metas.iter().enumerate() { + inputs[i] = 0; + if set & (1 << iv.shard_id) == 0 { + continue; + } + let mut taken = 0; + for (sid, slot) in gathers[i].iter().enumerate() { + if taken == data_shards { + break; + } + if set & (1 << sid) == 0 && slot.is_some() { + inputs[i] |= 1 << sid; + taken += 1; + } + } + if taken < data_shards { + return false; + } + } + tried.insert(inputs.to_vec()) + }; + + // Rebuilds the planned intervals into `candidate`; reset on failure. + let mut decode = + |inputs: &[u64], candidate: &mut [u8], gathers: &mut [Vec>>]| { + let mut pos = 0usize; + for (i, iv) in metas.iter().enumerate() { + let end = pos + iv.size; + if inputs[i] != 0 { + let Some(buf) = + decode_ec_interval(rs, &mut gathers[i], inputs[i], &mut scratch, iv) + else { + candidate.copy_from_slice(corrupted); + return false; + }; + candidate[pos..end].copy_from_slice(buf); + } + pos = end; + } + if parses(candidate) { + return true; + } + candidate.copy_from_slice(corrupted); + false + }; + + let (mut sets, mut spent) = (0usize, 0usize); + for k in 1..=rs.parity_shard_count().min(total_shards) { + let mut members: Vec = (0..k).collect(); + loop { + let set = members.iter().fold(0u64, |set, &sid| set | 1 << sid); + let holds_suspect = members + .iter() + .any(|&sid| suspects.contains(&(sid as ShardId))); + if holds_suspect { + sets += 1; + if sets > EC_REBUILD_MAX_SETS { + return None; + } + if plan(set, &mut inputs, gathers) { + if k > 1 { + if spent >= EC_REBUILD_MAX_BYTES { + return None; + } + spent += corrupted.len(); + } + if decode(&inputs, &mut candidate, gathers) { + let mut convicted: Vec = Vec::new(); + let mut pos = 0usize; + for iv in metas { + let end = pos + iv.size; + if candidate[pos..end] != corrupted[pos..end] { + convicted.push(iv.shard_id); + } + pos = end; + } + convicted.extend( + members + .iter() + .map(|&sid| sid as ShardId) + .filter(|sid| !suspects.contains(sid)), + ); + return Some((candidate, convicted)); + } + } + } + let Some(j) = (0..k).rev().find(|&j| members[j] != total_shards - k + j) else { + break; + }; + members[j] += 1; + for m in j + 1..k { + members[m] = members[m - 1] + 1; + } + } + } + None +} + +/// Decode `iv`'s data shard from the gathered shards in `inputs`, using +/// `scratch`, one buffer per data shard, for every data shard it lacks. +fn decode_ec_interval<'a>( + rs: &ReedSolomon, + slots: &mut [Option>], + inputs: u64, + scratch: &'a mut [Vec], + iv: &EcInterval, +) -> Option<&'a [u8]> { + let mut spare = scratch.iter_mut(); + let mut view: Vec<(&mut [u8], bool)> = slots + .iter_mut() + .enumerate() + .map(|(sid, slot)| match (slot, spare.next()) { + (Some(shard), _) if inputs & (1 << sid) != 0 => (shard.as_mut_slice(), true), + (_, Some(buf)) => { + buf.resize(iv.size, 0); + (buf.as_mut_slice(), false) + } + _ => (&mut [][..], false), + }) + .collect(); + rs.reconstruct_data(&mut view).ok()?; + drop(view); + Some(&scratch[iv.shard_id as usize]) +} + // parse_grpc_address lives in `grpc_client.rs` and is re-exported // here via the use above so this module shares a single // HTTP↔gRPC port-translation routine with grpc_server.rs. @@ -2242,6 +2486,139 @@ mod tests { ); } + const BLOCK: usize = 4096; + + /// One row of 10+4 shards holding a needle at `dat_offset`, the needle's + /// intervals, and its bytes. + fn ec_row_with_needle( + dat_offset: usize, + data_len: usize, + ) -> (Vec>, Vec, Vec, Needle) { + let mut n = Needle { + id: NeedleId(7), + data: (0..data_len).map(|i| (i * 31 + 7) as u8).collect(), + ..Needle::default() + }; + let bytes = n.write_bytes(VERSION_3); + let mut dat = vec![0u8; 10 * BLOCK]; + dat[dat_offset..dat_offset + bytes.len()].copy_from_slice(&bytes); + let mut shards: Vec> = dat.chunks(BLOCK).map(|c| c.to_vec()).collect(); + shards.resize(14, vec![0u8; BLOCK]); + ReedSolomon::new(10, 4) + .unwrap() + .encode(&mut shards) + .unwrap(); + + let mut metas = Vec::new(); + let (mut pos, end) = (dat_offset, dat_offset + bytes.len()); + while pos < end { + let size = (BLOCK - pos % BLOCK).min(end - pos); + metas.push(EcInterval { + vid: VolumeId(1), + needle_id: n.id, + shard_id: (pos / BLOCK) as ShardId, + shard_offset: (pos % BLOCK) as i64, + size, + expected_encode_ts_ns: 0, + }); + pos += size; + } + (shards, metas, bytes, n) + } + + fn corrupt(shards: &mut [Vec], shard_ids: &[ShardId], metas: &[EcInterval]) { + for &sid in shard_ids { + for iv in metas { + let start = iv.shard_offset as usize; + for b in &mut shards[sid as usize][start..start + iv.size] { + *b ^= 0xFF; + } + } + } + } + + fn search( + shards: &[Vec], + metas: &[EcInterval], + n: &Needle, + dat_offset: usize, + ) -> Option<(Vec, Vec)> { + let mut corrupted = Vec::new(); + let mut gathers = Vec::new(); + for iv in metas { + let range = iv.shard_offset as usize..iv.shard_offset as usize + iv.size; + corrupted.extend_from_slice(&shards[iv.shard_id as usize][range.clone()]); + gathers.push( + shards + .iter() + .enumerate() + .map(|(sid, shard)| { + (sid != iv.shard_id as usize).then(|| shard[range.clone()].to_vec()) + }) + .collect(), + ); + } + let suspects: HashSet = metas.iter().map(|iv| iv.shard_id).collect(); + let rs = ReedSolomon::new(10, 4).unwrap(); + search_ec_corrupt_sets( + &mut gathers, + metas, + &rs, + &corrupted, + &suspects, + |candidate| { + let mut check = Needle { + id: n.id, + ..Needle::default() + }; + check + .read_bytes(candidate, dat_offset as i64, n.size, VERSION_3) + .is_ok() + && check.id == n.id + }, + ) + } + + #[test] + fn search_ec_corrupt_sets_rebuilds_with_parity_many_shards_corrupt() { + let dat_offset = 2 * BLOCK + 100; + let (mut shards, metas, bytes, n) = ec_row_with_needle(dat_offset, 1000); + assert_eq!(metas.len(), 1); + // The needle's own shard and the lowest-numbered survivors, which every + // first decode reads. + corrupt(&mut shards, &[2, 0, 1, 3], &metas); + + let (rebuilt, mut convicted) = search(&shards, &metas, &n, dat_offset).unwrap(); + assert_eq!(rebuilt, bytes); + convicted.sort(); + assert_eq!(convicted, vec![0, 1, 2, 3]); + } + + #[test] + fn search_ec_corrupt_sets_keeps_a_clean_suspect_as_input() { + let dat_offset = BLOCK - 300; + let (mut shards, metas, bytes, n) = ec_row_with_needle(dat_offset, 1000); + assert_eq!(metas.len(), 2); + // Shard 0 is corrupt and shard 1 clean; rebuilding both would exceed + // parity with three other shards corrupt too. + corrupt(&mut shards, &[0], &metas[..1]); + corrupt(&mut shards, &[2, 3, 4], &metas); + + let (rebuilt, mut convicted) = search(&shards, &metas, &n, dat_offset).unwrap(); + assert_eq!(rebuilt, bytes); + convicted.sort(); + assert_eq!(convicted, vec![0, 2, 3, 4]); + } + + #[test] + fn search_ec_corrupt_sets_fails_past_parity() { + let dat_offset = 100; + let (mut shards, metas, _, n) = ec_row_with_needle(dat_offset, 1000); + corrupt(&mut shards, &[0, 1, 2, 3, 4], &metas); + + assert!(search(&shards, &metas, &n, dat_offset).is_none()); + } + fn locations(count: usize) -> HashMap> { (0..count) .map(|sid| (sid as ShardId, vec!["127.0.0.1:8080".to_string()])) diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 7df9bf090a..d3250c45fc 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -772,9 +772,14 @@ func (s *Store) rebuildEcShardNeedle(n *needle.Needle, needleId types.NeedleId, return candidate, false } } - if len(suspects) == 1 { - return nil, false - } + } + if candidate, isDeleted := s.rebuildEcNeedleFromCorruptSets(n, needleId, ecVolume, ecCtx, intervals, shardIds, offsets, offset, size, corrupted, suspects); isDeleted { + return nil, true + } else if candidate != nil { + return candidate, false + } + if len(suspects) == 1 { + return nil, false } for shardId := range suspects { if candidate, isDeleted := try(map[erasure_coding.ShardId]bool{shardId: true}); isDeleted { @@ -866,6 +871,182 @@ func (s *Store) rebuildEcNeedleFromDroppedInputs(n *needle.Needle, needleId type return nil, false } +// ecRebuildMaxSets bounds how many candidate corrupt sets one needle's search +// enumerates, and ecRebuildMaxBytes the needle bytes it rebuilds and parses +// past the single-shard sets. 10+4 has 1470 sets of at most four shards. +const ( + ecRebuildMaxSets = 4096 + ecRebuildMaxBytes = 1 << 30 +) + +// rebuildEcNeedleFromCorruptSets handles a rebuild the retries above could not +// repair: more shards than one can be corrupt over the same ranges and get read +// as reconstruction inputs. Each interval is gathered once from every other +// shard that answers, then each set of up to ParityShards shards holding a +// suspect is tried as the corrupt set, smallest first: suspect ranges inside it +// are rebuilt from the shards outside it, the rest keep the bytes read, and a +// needle parse arbitrates. A set whose decodes read the same inputs as one +// already tried is skipped. Nil when that gather would outgrow the recovery +// budget, or no set parses. +func (s *Store) rebuildEcNeedleFromCorruptSets(n *needle.Needle, needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, ecCtx *erasure_coding.ECContext, intervals []erasure_coding.Interval, shardIds []erasure_coding.ShardId, offsets []int64, offset int64, size types.Size, corrupted []byte, suspects map[erasure_coding.ShardId]bool) ([]byte, bool) { + var totalSize int64 + for _, interval := range intervals { + totalSize += int64(interval.Size) + } + total := ecCtx.Total() + if totalSize*int64(total-1) > ecRecoverBudget { + return nil, false + } + weight := totalSize * int64(total) + if weight > ecRecoverBudget { + weight = ecRecoverBudget + } + if err := ecRecoverSem.Acquire(context.Background(), weight); err != nil { + return nil, false + } + defer ecRecoverSem.Release(weight) + + gathers := make([][][]byte, len(intervals)) + for i, interval := range intervals { + shardIntervals, isDeleted := s.gatherEcShardIntervals(needleId, ecVolume, ecCtx, map[erasure_coding.ShardId]bool{shardIds[i]: true}, int(interval.Size), offsets[i], total) + if isDeleted { + return nil, true + } + gathers[i] = shardIntervals + } + + enc, err := reedsolomon.New(ecCtx.DataShards, ecCtx.ParityShards) + if err != nil { + return nil, false + } + candidate := slices.Clone(corrupted) + required := make([]bool, total) + view := make([][]byte, total) + inputs := make([]uint64, len(intervals)) + tried := make(map[string]bool) + + // plan records in inputs the shards each interval's decode reads when set + // is excluded, 0 for an interval that keeps its bytes. False when an + // interval has too few shards left, or the same decodes were tried before. + plan := func(set uint64) bool { + for i := range intervals { + inputs[i] = 0 + if set&(1< ecRebuildMaxSets { + return nil, false + } + if plan(set) { + if k > 1 { + if spent >= ecRebuildMaxBytes { + return nil, false + } + spent += len(corrupted) + } + if decode() { + // The verdict convicts every differing range, and every + // other shard the set kept out of the decodes. + pos := 0 + for i, interval := range intervals { + end := pos + int(interval.Size) + if !bytes.Equal(candidate[pos:end], corrupted[pos:end]) { + forgetShardId(ecVolume, shardIds[i]) + } + pos = end + } + for _, shardId := range members { + if !suspects[erasure_coding.ShardId(shardId)] { + forgetShardId(ecVolume, erasure_coding.ShardId(shardId)) + } + } + return candidate, false + } + } + } + j := k - 1 + for j >= 0 && members[j] == total-k+j { + j-- + } + if j < 0 { + break + } + members[j]++ + for m := j + 1; m < k; m++ { + members[m] = members[m-1] + 1 + } + } + } + return nil, false +} + func forgetShardId(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId) { // failed to access the source data nodes, clear it up ecVolume.ShardLocationsLock.Lock() diff --git a/weed/storage/store_ec_interval_read_test.go b/weed/storage/store_ec_interval_read_test.go index dd603cf678..9652cfcde5 100644 --- a/weed/storage/store_ec_interval_read_test.go +++ b/weed/storage/store_ec_interval_read_test.go @@ -283,6 +283,128 @@ func TestReadEcShardNeedleHealsWhenRebuildInputAlsoRotted(t *testing.T) { } } +// rotParityManyShards stores a one-interval needle and corrupts its range on +// its own shard and on the lowest-numbered survivors, as many as parity, so +// every cheap decode reads at least two corrupt inputs. +func rotParityManyShards(t *testing.T, store *Store, vid needle.VolumeId) *needle.Needle { + n := randomNeedleOfSize(99, 1024) + ecVolume := encodeAndMountEcVolume(t, store, vid, []*needle.Needle{n}) + + _, _, intervals, err := ecVolume.LocateEcShardNeedle(n.Id, ecVolume.Version) + if err != nil { + t.Fatalf("locate needle: %v", err) + } + if len(intervals) != 1 { + t.Fatalf("needle covers %d intervals, want 1", len(intervals)) + } + shardId, shardOffset := ecVolume.IntervalToShardIdAndOffset(intervals[0]) + corruptEcShardRange(t, ecVolume, shardId, shardOffset, int64(intervals[0].Size)) + corrupted := 1 + for other := erasure_coding.ShardId(0); corrupted < ecVolume.ECContext.ParityShards; other++ { + if other == shardId { + continue + } + corruptEcShardRange(t, ecVolume, other, shardOffset, int64(intervals[0].Size)) + corrupted++ + } + return n +} + +// rotSuspectNextToCleanOne stores a needle over two shards, corrupts the first +// of them, and corrupts enough other shards that rebuilding both suspects would +// exceed parity: only the clean suspect kept as an input repairs the read. +func rotSuspectNextToCleanOne(t *testing.T, store *Store, vid needle.VolumeId) *needle.Needle { + n := randomNeedleOfSize(55, erasure_coding.ErasureCodingSmallBlockSize+1234) + ecVolume := encodeAndMountEcVolume(t, store, vid, []*needle.Needle{n}) + + _, _, intervals, err := ecVolume.LocateEcShardNeedle(n.Id, ecVolume.Version) + if err != nil { + t.Fatalf("locate needle: %v", err) + } + if len(intervals) != 2 { + t.Fatalf("needle covers %d intervals, want 2", len(intervals)) + } + suspects := make(map[erasure_coding.ShardId]bool) + for _, interval := range intervals { + shardId, _ := ecVolume.IntervalToShardIdAndOffset(interval) + suspects[shardId] = true + } + shardId, shardOffset := ecVolume.IntervalToShardIdAndOffset(intervals[0]) + corruptEcShardRange(t, ecVolume, shardId, shardOffset, int64(intervals[0].Size)) + corrupted := 1 + for other := erasure_coding.ShardId(0); corrupted < ecVolume.ECContext.ParityShards; other++ { + if suspects[other] { + continue + } + for _, interval := range intervals { + _, offset := ecVolume.IntervalToShardIdAndOffset(interval) + corruptEcShardRange(t, ecVolume, other, offset, int64(interval.Size)) + } + corrupted++ + } + return n +} + +func readEcNeedleBack(t *testing.T, store *Store, vid needle.VolumeId, n *needle.Needle) error { + got := new(needle.Needle) + got.Id = n.Id + if _, err := store.ReadEcShardNeedle(vid, got, nil); err != nil { + return err + } + if !bytes.Equal(got.Data, n.Data) { + t.Fatalf("read back %d bytes, want the %d written", len(got.Data), len(n.Data)) + } + return nil +} + +func TestReadEcShardNeedleHealsWhenParityManyShardsRotted(t *testing.T) { + store := newTestStore(t, 1) + const vid = needle.VolumeId(13) + n := rotParityManyShards(t, store, vid) + if err := readEcNeedleBack(t, store, vid, n); err != nil { + t.Fatalf("read should reconstruct within parity: %v", err) + } +} + +func TestReadEcShardNeedleHealsCorruptSuspectNextToCleanOne(t *testing.T) { + store := newTestStore(t, 1) + const vid = needle.VolumeId(14) + n := rotSuspectNextToCleanOne(t, store, vid) + if err := readEcNeedleBack(t, store, vid, n); err != nil { + t.Fatalf("read should reconstruct within parity: %v", err) + } +} + +// Past the recovery budget the corrupt-set search does not run, and the read +// keeps to the retries it extends: one corrupt range in a needle over more +// shards than parity still heals one suspect at a time, while parity-many +// corrupt shards stay unreadable. +func TestReadEcShardNeedleCorruptSetSearchPastBudget(t *testing.T) { + setEcRecoverBudget(t, 1) + store := newTestStore(t, 1) + + const wideVid = needle.VolumeId(17) + wide := randomNeedleOfSize(77, 6*erasure_coding.ErasureCodingSmallBlockSize+1234) + ecVolume := encodeAndMountEcVolume(t, store, wideVid, []*needle.Needle{wide}) + _, _, intervals, err := ecVolume.LocateEcShardNeedle(wide.Id, ecVolume.Version) + if err != nil { + t.Fatalf("locate needle: %v", err) + } + if len(intervals) <= ecVolume.ECContext.ParityShards { + t.Fatalf("needle covers %d intervals, want more than %d", len(intervals), ecVolume.ECContext.ParityShards) + } + shardId, shardOffset := ecVolume.IntervalToShardIdAndOffset(intervals[1]) + corruptEcShardRange(t, ecVolume, shardId, shardOffset, int64(intervals[1].Size)) + if err := readEcNeedleBack(t, store, wideVid, wide); err != nil { + t.Fatalf("one corrupt shard should reconstruct past the budget: %v", err) + } + + n := rotParityManyShards(t, store, 15) + if err := readEcNeedleBack(t, store, 15, n); err == nil { + t.Fatalf("parity-many corrupt shards should need the search, which the budget rules out") + } +} + // When more shards than parity are corrupt in the same range no hypothesis can // gather enough clean input, so the read still fails instead of serving wrong // bytes.