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<VolumeServerState>,
     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<VolumeServerState>,
+    metas: &[EcInterval],
+    map: EcShardMap<'_>,
+    corrupted: &[u8],
+    suspects: &HashSet<ShardId>,
+    offset: i64,
+    size: Size,
+    version: Version,
+    needle_id: NeedleId,
+) -> (Option<Vec<u8>>, 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<Option<Vec<u8>>>> = 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<Option<Vec<u8>>>],
+    metas: &[EcInterval],
+    rs: &ReedSolomon,
+    corrupted: &[u8],
+    suspects: &HashSet<ShardId>,
+    parses: impl Fn(&[u8]) -> bool,
+) -> Option<(Vec<u8>, Vec<ShardId>)> {
+    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<u8>> = vec![Vec::new(); data_shards];
+    let mut inputs = vec![0u64; metas.len()];
+    let mut tried: HashSet<Vec<u64>> = 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<Option<Vec<u8>>>]| {
+        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<Option<Vec<u8>>>]| {
+            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<usize> = (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<ShardId> = 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<Vec<u8>>],
+    inputs: u64,
+    scratch: &'a mut [Vec<u8>],
+    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<u8>>, Vec<EcInterval>, Vec<u8>, 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<Vec<u8>> = 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<u8>], 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<u8>],
+        metas: &[EcInterval],
+        n: &Needle,
+        dat_offset: usize,
+    ) -> Option<(Vec<u8>, Vec<ShardId>)> {
+        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<ShardId> = 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<ShardId, Vec<String>> {
         (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<<shardIds[i]) == 0 {
+				continue
+			}
+			taken := 0
+			for shardId := 0; shardId < total && taken < ecCtx.DataShards; shardId++ {
+				if set&(1<<shardId) == 0 && gathers[i][shardId] != nil {
+					inputs[i] |= 1 << shardId
+					taken++
+				}
+			}
+			if taken < ecCtx.DataShards {
+				return false
+			}
+		}
+		key := fmt.Sprint(inputs)
+		if tried[key] {
+			return false
+		}
+		tried[key] = true
+		return true
+	}
+
+	// decode rebuilds the planned intervals into candidate. The parsed needle
+	// keeps slices of candidate, so it is reset only on failure.
+	decode := func() bool {
+		pos := 0
+		for i, interval := range intervals {
+			end := pos + int(interval.Size)
+			if inputs[i] != 0 {
+				for shardId := range view {
+					view[shardId] = nil
+					if inputs[i]&(1<<shardId) != 0 {
+						view[shardId] = gathers[i][shardId]
+					}
+				}
+				view[shardIds[i]] = candidate[pos:pos:end]
+				required[shardIds[i]] = true
+				err := enc.ReconstructSome(view, required)
+				required[shardIds[i]] = false
+				if err != nil {
+					copy(candidate, corrupted)
+					return false
+				}
+			}
+			pos = end
+		}
+		check := &needle.Needle{Id: needleId}
+		if check.ReadBytes(candidate, offset, size, ecVolume.Version) != nil || check.Id != needleId {
+			copy(candidate, corrupted)
+			return false
+		}
+		*n = *check
+		return true
+	}
+
+	sets, spent := 0, 0
+	members := make([]int, 0, ecCtx.ParityShards)
+	for k := 1; k <= ecCtx.ParityShards && k <= total; k++ {
+		members = members[:k]
+		for j := range members {
+			members[j] = j
+		}
+		for {
+			var set uint64
+			holdsSuspect := false
+			for _, shardId := range members {
+				set |= 1 << shardId
+				holdsSuspect = holdsSuspect || suspects[erasure_coding.ShardId(shardId)]
+			}
+			if holdsSuspect {
+				if sets++; sets > 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.