ec: rebuild a needle's suspect shard ranges when it fails to parse (#12170)

* ec: let shard-interval recovery exclude a set of suspect shards

gatherEcShardIntervals and recoverOneRemoteEcShardInterval skipped only
the shard being rebuilt. A needle-level rebuild wants every suspect
shard kept out of the gather so a bad copy cannot poison the
reconstruction it is being compared against.

* ec: rebuild a needle's suspect shard ranges when it fails to parse

A bit-rotted data shard serves a clean-looking but wrong interval, so the
missing/unreadable-shard fallback never engages and the needle CRC check
in ReadEcShardNeedle returns the corruption to the caller. Each interval's
source shard is a suspect: rebuild the suspect ranges from the surviving
shards — all of them at once while parity covers the suspect set, or one
at a time when the needle spans more shards than that — and let a
successful needle parse arbitrate which copy was corrupt. Shards whose
range changed are dropped from the location cache so later reads go
straight to reconstruction.

* ec: mirror the corrupt-shard needle rebuild in the Rust volume server

The snapshot now keeps every interval's shard placement so a parse
failure can name its suspects. rebuild_ec_shard_needle rebuilds the
suspect ranges through recover_one_remote_ec_shard_interval with the
suspect set excluded from the gather, and the winning candidate's
differing shards are dropped via forget_shard_location.

* ec: recover suspect ranges under the requested needle id

A failed ReadBytes re-reads the id out of the possibly corrupt header, so the rebuild passed whatever it decoded to peer recovery requests; carry the requested id through instead, and reject a candidate whose parsed id does not match it.

* ec: retry a failed needle rebuild with each gathered input dropped

The gather behind a suspect-range rebuild can itself serve a second corrupt shard: Reed-Solomon then reconstructs wrong bytes and the needle parse rejects the only candidate. Gather every surviving shard once, then re-decode with each of the first decode's inputs dropped in turn; the parse arbitrates the honest subset.

* ec: mirror the dropped-input rebuild and id check in the Rust volume server

Same two guards as the Go side: a candidate that parses under a different needle id is rejected, and a failed candidate whose gather fed it a corrupt input is retried by dropping one gathered shard at a time. The gather is shared through a want-bounded gather_ec_shard_intervals so the retry pays no extra fan-out.

* ec: answer deleted when a rebuild's holder reports the needle deleted

The rebuild discarded the deleted flag its interval recovery collected, so a needle deleted on its other holders but live in the local index came back as a corruption error. A deleted report is authoritative, so it now short-circuits the hypothesis search and answers ErrorDeleted.

* ec: hydrate rebuild candidates into a scratch needle

A failed ReadBytes leaves partially parsed fields on the needle, and optional sections do not reset when absent, so a corrupt first parse could leak stale name/mime/ttl into a successful candidate. Parse each candidate into a fresh needle and adopt it only on a matching id.

* ec: corrupt parity+1 shards in the exceeds-parity read test

The loop skipped the suspect's own shard when it fell inside the parity range, leaving only ParityShards corrupt shards — a case recovery can legitimately fix. Count the additional corruptions so the test always exercises the boundary it names.

* ec: mirror the deleted-report short-circuit in the Rust volume server

A holder's deleted answer during a rebuild propagates out as EcMiss::Deleted instead of surfacing as InvalidData, matching the Go path's authoritative-delete rule.

seaweed-volume/src/server/store_ec.rs

@@ -87,6 +87,9 @@ struct Snapshot {
     offset: Offset,
     size_for_parse: Size,
     intervals: Vec<IntervalResult>,
+    /// Per-interval (shard_id, shard_offset, size) for the needle-rebuild
+    /// pass: a parse failure retries the suspect ranges through Reed-Solomon.
+    metas: Vec<EcInterval>,
     cached_locations: HashMap<ShardId, Vec<String>>,
     cache_refreshed_at: Option<Instant>,
     /// This volume's encode identity, carried to peers on remote shard reads so a
@@ -260,13 +263,51 @@ pub async fn read_ec_shard_needle_or_miss(
         id: needle_id,
         ..Needle::default()
     };
-    n.read_bytes(
+    if let Err(e) = n.read_bytes(
         &bytes,
         snapshot.offset.to_actual_offset(),
         snapshot.size_for_parse,
         snapshot.version,
-    )
-    .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?;
+    ) {
+        // A shard that served an interval can be silently corrupt: rebuild the
+        // suspect ranges from the surviving shards and let a successful parse
+        // arbitrate which copy was wrong. Mirrors Go's rebuildEcShardNeedle.
+        let (rebuilt, is_deleted) = rebuild_ec_shard_needle(
+            state,
+            &snapshot.metas,
+            EcShardMap {
+                locations: &shard_locations,
+                data_shards,
+                parity_shards,
+            },
+            &bytes,
+            snapshot.offset.to_actual_offset(),
+            snapshot.size_for_parse,
+            snapshot.version,
+            needle_id,
+        )
+        .await;
+        // A holder reporting the needle deleted is authoritative -- deletes are
+        // never invented and never undone.
+        if is_deleted {
+            return Ok(Err(EcMiss::Deleted));
+        }
+        let Some(rebuilt) = rebuilt else {
+            return Err(io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)));
+        };
+        let mut n = Needle {
+            id: needle_id,
+            ..Needle::default()
+        };
+        n.read_bytes(
+            &rebuilt,
+            snapshot.offset.to_actual_offset(),
+            snapshot.size_for_parse,
+            snapshot.version,
+        )
+        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?;
+        return Ok(Ok(n));
+    }
     Ok(Ok(n))
 }
 
@@ -847,6 +888,7 @@ pub async fn scrub_ec_volume_distributed(
                                     data_shards,
                                     parity_shards: total_shards - data_shards,
                                 },
+                                &HashSet::from([iv.shard_id]),
                             )
                             .await
                             {
@@ -933,7 +975,7 @@ fn snapshot_under_lock(
             EcMiss::NotFound
         }));
     };
-    build_snapshot(ecv, offset, size, &intervals).map(Ok)
+    build_snapshot(ecv, vid, needle_id, offset, size, &intervals).map(Ok)
 }
 
 /// Like `snapshot_under_lock`, but locates intervals from the RAW .ecx
@@ -1046,6 +1088,8 @@ fn read_local_intervals(
 /// read path can drop the store lock before awaiting remote reads.
 fn build_snapshot(
     ecv: &crate::storage::erasure_coding::ec_volume::EcVolume,
+    vid: VolumeId,
+    needle_id: NeedleId,
     offset: Offset,
     size: Size,
     intervals: &[crate::storage::erasure_coding::ec_locate::Interval],
@@ -1058,6 +1102,20 @@ fn build_snapshot(
     }
     let actual = get_actual_size(size, ecv.version);
     let interval_results = read_local_intervals(ecv, intervals);
+    let metas = intervals
+        .iter()
+        .map(|iv| {
+            let (shard_id, shard_offset) = ecv.interval_to_shard_id_and_offset(iv);
+            EcInterval {
+                vid,
+                needle_id,
+                shard_id,
+                shard_offset,
+                size: iv.size as usize,
+                expected_encode_ts_ns: ecv.encode_ts_ns,
+            }
+        })
+        .collect();
     let (cached_locations, cache_refreshed_at) = ecv.shard_locations_snapshot();
 
     Ok(Snapshot {
@@ -1068,6 +1126,7 @@ fn build_snapshot(
         offset,
         size_for_parse: size,
         intervals: interval_results,
+        metas,
         cached_locations,
         cache_refreshed_at,
         encode_ts_ns: ecv.encode_ts_ns,
@@ -1297,7 +1356,7 @@ async fn fetch_one_interval(
 
     // Reconstruct: fan-out reads to every other shard at the same
     // (shard_offset, size). Mirrors `recoverOneRemoteEcShardInterval`.
-    recover_one_remote_ec_shard_interval(state, iv, map).await
+    recover_one_remote_ec_shard_interval(state, iv, map, &HashSet::from([iv.shard_id])).await
 }
 
 async fn read_remote_ec_shard_interval(
@@ -1422,18 +1481,26 @@ async fn do_read_remote_ec_shard_interval(
     Ok((out, false))
 }
 
-async fn recover_one_remote_ec_shard_interval(
+/// Gather `iv`'s interval from every shard outside `exclude`, returning one
+/// buffer per shard and `None` where the shard could not be read, plus whether
+/// any holder reported the needle deleted. Stops once `want` buffers are in
+/// hand: reconstruction consumes no more than `data_shards`, so callers pass
+/// that unless a retry needs the wider set.
+/// Mirrors Go's `gatherEcShardIntervals`.
+async fn gather_ec_shard_intervals(
     state: &Arc<VolumeServerState>,
     iv: EcInterval,
     map: EcShardMap<'_>,
-) -> io::Result<(Vec<u8>, bool)> {
+    exclude: &HashSet<ShardId>,
+    want: usize,
+) -> (Vec<Option<Vec<u8>>>, bool) {
     let EcInterval {
         vid,
         needle_id,
-        shard_id: shard_id_to_recover,
         shard_offset,
         size,
         expected_encode_ts_ns,
+        ..
     } = iv;
     let EcShardMap {
         locations: shard_locations,
@@ -1441,23 +1508,6 @@ async fn recover_one_remote_ec_shard_interval(
         parity_shards,
     } = map;
     let total_shards = data_shards + parity_shards;
-    let rs = ReedSolomon::new(data_shards, parity_shards)
-        .map_err(|e| io::Error::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
-    // fan-out outgrows the whole budget takes all of it and so runs alone,
-    // rather than waiting on permits that can never be granted.
-    let _permit = EC_RECOVER_SEM
-        .acquire_many((size * data_shards).min(EC_RECOVER_BUDGET) as u32)
-        .await
-        .map_err(|e| {
-            io::Error::other(format!(
-                "ec recover budget for shard {}.{}: {}",
-                vid.0, shard_id_to_recover, e
-            ))
-        })?;
-
     let mut bufs: Vec<Option<Vec<u8>>> = vec![None; total_shards];
 
     // Phase 0: seed bufs from LOCALLY mounted shards. If this node
@@ -1469,12 +1519,9 @@ async fn recover_one_remote_ec_shard_interval(
     {
         let store = state.store.read().unwrap();
         for (sid, slot) in bufs.iter_mut().enumerate() {
-            if available >= data_shards {
+            if available >= want {
                 break;
             }
-            if sid as ShardId == shard_id_to_recover {
-                continue;
-            }
             // Resolve the shard together with the EcVolume on the disk that owns
             // it: a reconciled volume has its shards split across data dirs. A
             // shard from a different encode run must not be fed to Reed-Solomon;
@@ -1483,6 +1530,9 @@ async fn recover_one_remote_ec_shard_interval(
             let Ok(sid_shard) = ShardId::try_from(sid) else {
                 continue;
             };
+            if exclude.contains(&sid_shard) {
+                continue;
+            }
             let owner = match store.find_ec_volume_with_shard(vid, sid_shard) {
                 Some(ecv)
                     if expected_encode_ts_ns == 0 || ecv.encode_ts_ns == expected_encode_ts_ns =>
@@ -1513,7 +1563,7 @@ async fn recover_one_remote_ec_shard_interval(
     let mut candidates: Vec<(ShardId, Vec<String>)> = shard_locations
         .iter()
         .filter(|(sid, locs)| {
-            **sid != shard_id_to_recover
+            !exclude.contains(*sid)
                 && (**sid as usize) < total_shards
                 && !locs.is_empty()
                 && bufs[**sid as usize].is_none()
@@ -1522,8 +1572,8 @@ async fn recover_one_remote_ec_shard_interval(
         .collect();
 
     let mut any_deleted = false;
-    while available < data_shards && !candidates.is_empty() {
-        let rest = candidates.split_off((data_shards - available).min(candidates.len()));
+    while available < want && !candidates.is_empty() {
+        let rest = candidates.split_off((want - available).min(candidates.len()));
         let wave = std::mem::replace(&mut candidates, rest);
         let results = join_all(wave.into_iter().map(|(sid, locs)| {
             let state = state.clone();
@@ -1573,6 +1623,51 @@ async fn recover_one_remote_ec_shard_interval(
         }
     }
 
+    (bufs, any_deleted)
+}
+
+/// Rebuild `iv.shard_id`'s interval from the shards outside `exclude`, which
+/// always contains `iv.shard_id` itself. Go's `recoverEcShardInterval`: the
+/// needle-level rebuild passes the whole suspect set so a bad copy cannot
+/// poison the reconstruction it is being compared against.
+async fn recover_one_remote_ec_shard_interval(
+    state: &Arc<VolumeServerState>,
+    iv: EcInterval,
+    map: EcShardMap<'_>,
+    exclude: &HashSet<ShardId>,
+) -> io::Result<(Vec<u8>, bool)> {
+    let EcInterval {
+        vid,
+        shard_id: shard_id_to_recover,
+        size,
+        ..
+    } = iv;
+    let EcShardMap {
+        data_shards,
+        parity_shards,
+        ..
+    } = map;
+    let rs = ReedSolomon::new(data_shards, parity_shards)
+        .map_err(|e| io::Error::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
+    // fan-out outgrows the whole budget takes all of it and so runs alone,
+    // rather than waiting on permits that can never be granted.
+    let _permit = EC_RECOVER_SEM
+        .acquire_many((size * data_shards).min(EC_RECOVER_BUDGET) as u32)
+        .await
+        .map_err(|e| {
+            io::Error::other(format!(
+                "ec recover budget for shard {}.{}: {}",
+                vid.0, shard_id_to_recover, e
+            ))
+        })?;
+
+    let (mut bufs, any_deleted) =
+        gather_ec_shard_intervals(state, iv, map, exclude, data_shards).await;
+    let available = bufs.iter().filter(|b| b.is_some()).count();
+
     if available < data_shards {
         // A holder reporting the needle deleted is authoritative -- deletes are
         // never invented and never undone -- so answer that rather than the
@@ -1602,6 +1697,263 @@ async fn recover_one_remote_ec_shard_interval(
     }
 }
 
+/// Drop a confirmed-corrupt shard's cached location so later reads go straight
+/// to reconstruction — Go's `forgetShardId` reached through the volume.
+fn forget_shard_location(state: &Arc<VolumeServerState>, vid: VolumeId, shard_id: ShardId) {
+    let store = state.store.read().unwrap();
+    if let Some(ecv) = store.find_ec_volume(vid) {
+        ecv.forget_shard_location(shard_id);
+    }
+}
+
+/// Rebuild the `metas` ranges served by shards in `exclude`, keeping the
+/// original bytes elsewhere, and return the candidate only when it parses as
+/// the needle. On success every shard whose range changed is confirmed corrupt
+/// and dropped from the location cache. The flag is a holder's deleted report:
+/// deletes are never invented, so it ends the search whatever the local index
+/// says.
+async fn try_ec_shard_needle_candidate(
+    state: &Arc<VolumeServerState>,
+    metas: &[EcInterval],
+    map: EcShardMap<'_>,
+    corrupted: &[u8],
+    exclude: &HashSet<ShardId>,
+    offset: i64,
+    size: Size,
+    version: Version,
+    needle_id: NeedleId,
+) -> (Option<Vec<u8>>, bool) {
+    let mut candidate = vec![0u8; corrupted.len()];
+    let mut pos = 0usize;
+    for iv in metas {
+        let end = pos + iv.size;
+        if exclude.contains(&iv.shard_id) {
+            let Ok((buf, is_deleted)) =
+                recover_one_remote_ec_shard_interval(state, *iv, map, exclude).await
+            else {
+                return (None, false);
+            };
+            if is_deleted {
+                return (None, true);
+            }
+            if buf.len() != end - pos {
+                return (None, false);
+            }
+            candidate[pos..end].copy_from_slice(&buf);
+        } else {
+            candidate[pos..end].copy_from_slice(&corrupted[pos..end]);
+        }
+        pos = end;
+    }
+    let mut check = Needle {
+        id: needle_id,
+        ..Needle::default()
+    };
+    // A candidate whose header decodes a different id than was asked for is no
+    // proof the bytes are right, whatever the checksum said.
+    if check.read_bytes(&candidate, offset, size, version).is_err() || check.id != needle_id {
+        return (None, false);
+    }
+    let mut pos = 0usize;
+    for iv in metas {
+        let end = pos + iv.size;
+        if candidate[pos..end] != corrupted[pos..end] {
+            forget_shard_location(state, iv.vid, iv.shard_id);
+        }
+        pos = end;
+    }
+    (Some(candidate), false)
+}
+
+/// 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.
+async fn rebuild_ec_shard_needle(
+    state: &Arc<VolumeServerState>,
+    metas: &[EcInterval],
+    map: EcShardMap<'_>,
+    corrupted: &[u8],
+    offset: i64,
+    size: Size,
+    version: Version,
+    needle_id: NeedleId,
+) -> (Option<Vec<u8>>, bool) {
+    let suspects: HashSet<ShardId> = metas.iter().map(|iv| iv.shard_id).collect();
+    if suspects.len() <= map.parity_shards {
+        let (candidate, is_deleted) = try_ec_shard_needle_candidate(
+            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);
+        }
+        // The gather behind a failed candidate can itself have read a second
+        // corrupt shard as reconstruction input. With slack left in the parity
+        // budget, retry while dropping one gathered input at a time.
+        if suspects.len() < map.parity_shards {
+            let (candidate, is_deleted) = rebuild_ec_needle_from_dropped_inputs(
+                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(
+            state,
+            metas,
+            map,
+            corrupted,
+            &HashSet::from([shard_id]),
+            offset,
+            size,
+            version,
+            needle_id,
+        )
+        .await;
+        if is_deleted {
+            return (None, true);
+        }
+        if let Some(candidate) = candidate {
+            return (Some(candidate), false);
+        }
+    }
+    (None, false)
+}
+
+/// Retry a rebuild whose every suspect range was recovered yet the candidate
+/// still failed to parse: the gather behind it can itself have read a second
+/// corrupt shard as reconstruction input. Each interval is gathered once —
+/// every shard that answers, not just the `data_shards` a decode consumes —
+/// then the decode is retried with one gathered input dropped at a time and a
+/// needle parse arbitrates. The hypotheses tried are the inputs the first
+/// decode consumed, the only shards whose corruption can explain the failed
+/// candidate. Mirrors Go's `rebuildEcNeedleFromDroppedInputs`.
+async fn rebuild_ec_needle_from_dropped_inputs(
+    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();
+    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());
+    let mut picked: HashSet<ShardId> = HashSet::new();
+    for iv in metas {
+        let (bufs, is_deleted) =
+            gather_ec_shard_intervals(state, *iv, map, suspects, total_shards).await;
+        if is_deleted {
+            return (None, true);
+        }
+        let mut taken = 0;
+        for (sid, slot) in bufs.iter().enumerate() {
+            if taken >= data_shards {
+                break;
+            }
+            if slot.is_some() {
+                picked.insert(sid as ShardId);
+                taken += 1;
+            }
+        }
+        gathers.push(bufs);
+    }
+
+    let Ok(rs) = ReedSolomon::new(data_shards, parity_shards) else {
+        return (None, false);
+    };
+    for drop in 0..total_shards as ShardId {
+        if !picked.contains(&drop) {
+            continue;
+        }
+        let mut candidate = corrupted.to_vec();
+        let mut pos = 0usize;
+        let mut ok = true;
+        for (i, iv) in metas.iter().enumerate() {
+            let bufs = &mut gathers[i];
+            let saved = bufs[drop as usize].take();
+            let present: Vec<bool> = bufs.iter().map(|slot| slot.is_some()).collect();
+            let out = rs
+                .reconstruct(bufs)
+                .ok()
+                .and_then(|_| bufs[iv.shard_id as usize].take());
+            // reconstruct fills every missing slot; keep only what the gather
+            // produced so the next variant decodes from the same inputs.
+            for (sid, slot) in bufs.iter_mut().enumerate() {
+                if !present[sid] {
+                    *slot = None;
+                }
+            }
+            bufs[drop as usize] = saved;
+            let Some(buf) = out else {
+                ok = false;
+                break;
+            };
+            if buf.len() != iv.size {
+                ok = false;
+                break;
+            }
+            candidate[pos..pos + iv.size].copy_from_slice(&buf);
+            pos += iv.size;
+        }
+        if !ok {
+            continue;
+        }
+        let mut check = Needle {
+            id: needle_id,
+            ..Needle::default()
+        };
+        if check.read_bytes(&candidate, offset, size, version).is_err() || check.id != needle_id {
+            continue;
+        }
+        // The verdict convicts every differing range — and the dropped input,
+        // without which the same decode kept failing to parse.
+        let mut pos = 0usize;
+        for iv in metas {
+            let end = pos + iv.size;
+            if candidate[pos..end] != corrupted[pos..end] {
+                forget_shard_location(state, iv.vid, iv.shard_id);
+            }
+            pos = end;
+        }
+        if let Some(first) = metas.first() {
+            forget_shard_location(state, first.vid, drop);
+        }
+        return (Some(candidate), false);
+    }
+    (None, false)
+}
+
 // 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.

seaweed-volume/src/storage/erasure_coding/ec_volume.rs

@@ -1472,6 +1472,16 @@ impl EcVolume {
         self.shard_locations.write().unwrap().stale = true;
     }
 
+    /// Drop one shard's cached locations and mark the map for a prompt
+    /// re-check — Go's `forgetShardId`. A read that disproved the entry (a
+    /// failed fetch, or a range the needle parse convicted as corrupt) must
+    /// not aim at it again until the map is re-learned.
+    pub(crate) fn forget_shard_location(&self, shard_id: ShardId) {
+        let mut cache = self.shard_locations.write().unwrap();
+        cache.locations.remove(&shard_id);
+        cache.stale = true;
+    }
+
     /// Put the stale mark to `decide`, and clear it only if `decide` says the
     /// master lookup is going ahead — both in one critical section, so the mark
     /// is consumed exactly once by the refresh that answers for it and a mark

weed/storage/store_ec.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -538,7 +539,10 @@ func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle, onReadS
 			if len(intervals) > 1 {
 				glog.V(3).Infof("ReadEcShardNeedle needle id %s intervals:%+v", n.String(), intervals)
 			}
-			bytes, isDeleted, err := s.readEcShardIntervals(n.Id, localEcVolume, intervals)
+			// ReadBytes re-reads the id out of a possibly corrupt header, so the
+			// requested id is kept aside for the rebuild path to recover under.
+			needleId := n.Id
+			bytes, isDeleted, err := s.readEcShardIntervals(needleId, localEcVolume, intervals)
 			// A holder reporting the needle deleted is authoritative -- deletes are
 			// never invented and never undone -- so answer that ahead of whatever
 			// error the shards it could not gather produced.
@@ -551,6 +555,11 @@ func (s *Store) ReadEcShardNeedle(vid needle.VolumeId, n *needle.Needle, onReadS
 
 			err = n.ReadBytes(bytes, offset.ToActualOffset(), size, localEcVolume.Version)
 			if err != nil {
+				if rebuilt, isDeleted := s.rebuildEcShardNeedle(n, needleId, localEcVolume, intervals, offset.ToActualOffset(), size, bytes); isDeleted {
+					return 0, ErrorDeleted
+				} else if rebuilt != nil {
+					return len(rebuilt), nil
+				}
 				return 0, fmt.Errorf("ec volume %d needle %s offset %d size %d: %w", vid, n.String(), offset.ToActualOffset(), size, err)
 			}
 
@@ -668,6 +677,181 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur
 	return
 }
 
+// rebuildEcShardNeedle repairs a needle whose reassembled bytes failed to
+// parse: a shard that served an interval can be silently corrupt, so each
+// interval's source shard is a suspect. Rebuilding the suspect ranges from the
+// surviving shards — with the suspects excluded from the gather, so a bad copy
+// cannot poison its own reconstruction — produces the bytes Reed-Solomon says
+// are right, and a successful needle parse arbitrates them correct. On success
+// the parse has already hydrated n, every shard whose range changed is dropped
+// from the location cache, and the repaired bytes are returned. Nil when no
+// hypothesis parses (index or needle corrupt in all shards alike). A deleted
+// report from a holder stops the search: deletes are never invented, so the
+// needle is gone whatever the local index says.
+func (s *Store) rebuildEcShardNeedle(n *needle.Needle, needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval, offset int64, size types.Size, corrupted []byte) ([]byte, bool) {
+	ecCtx := ecVolume.ECContext
+	if ecCtx == nil {
+		ecCtx = erasure_coding.NewDefaultECContext(ecVolume.Collection, ecVolume.VolumeId)
+	}
+
+	shardIds := make([]erasure_coding.ShardId, len(intervals))
+	offsets := make([]int64, len(intervals))
+	suspects := make(map[erasure_coding.ShardId]bool)
+	for i, interval := range intervals {
+		shardIds[i], offsets[i] = ecVolume.IntervalToShardIdAndOffset(interval)
+		suspects[shardIds[i]] = true
+	}
+
+	try := func(exclude map[erasure_coding.ShardId]bool) ([]byte, bool) {
+		candidate := make([]byte, len(corrupted))
+		pos := 0
+		for i, interval := range intervals {
+			buf := candidate[pos : pos+int(interval.Size)]
+			pos += int(interval.Size)
+			if !exclude[shardIds[i]] {
+				copy(buf, corrupted[pos-int(interval.Size):pos])
+				continue
+			}
+			_, isDeleted, err := s.recoverEcShardInterval(needleId, ecVolume, shardIds[i], buf, offsets[i], exclude)
+			if isDeleted {
+				return nil, true
+			}
+			if err != nil {
+				return nil, false
+			}
+		}
+		// A candidate whose header decodes a different id than was asked for is
+		// no proof the bytes are right, whatever the checksum said. Parse into a
+		// scratch needle: a failed earlier parse can leave stale fields in n.
+		check := &needle.Needle{Id: needleId}
+		if check.ReadBytes(candidate, offset, size, ecVolume.Version) != nil || check.Id != needleId {
+			return nil, false
+		}
+		// The parse verdict makes every differing range a confirmed corruption.
+		pos = 0
+		for i, interval := range intervals {
+			if !bytes.Equal(candidate[pos:pos+int(interval.Size)], corrupted[pos:pos+int(interval.Size)]) {
+				forgetShardId(ecVolume, shardIds[i])
+			}
+			pos += int(interval.Size)
+		}
+		*n = *check
+		return candidate, false
+	}
+
+	// When parity still covers every suspect, one pass excludes them all and
+	// repairs any subset of them being corrupt at once. Otherwise one suspect
+	// is rebuilt at a time, which fixes a single corrupt suspect.
+	if len(suspects) <= ecCtx.ParityShards {
+		if candidate, isDeleted := try(suspects); isDeleted {
+			return nil, true
+		} else if candidate != nil {
+			return candidate, false
+		}
+		// The gather behind a failed candidate can itself have read a second
+		// corrupt shard as reconstruction input. With slack left in the parity
+		// budget, retry while dropping one gathered input at a time.
+		if len(suspects) < ecCtx.ParityShards {
+			if candidate, isDeleted := s.rebuildEcNeedleFromDroppedInputs(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 {
+			return nil, true
+		} else if candidate != nil {
+			return candidate, false
+		}
+	}
+	return nil, false
+}
+
+// rebuildEcNeedleFromDroppedInputs handles a rebuild whose every suspect range
+// was recovered yet the candidate still failed to parse: the gather behind it
+// can itself have read a second corrupt shard as reconstruction input. Each
+// interval is gathered once — keeping every shard that answers rather than
+// stopping at DataShards — then the decode is retried with one gathered input
+// dropped at a time and a needle parse arbitrates. The hypotheses tried are
+// the inputs the first decode consumed, the only shards whose corruption can
+// explain the failed candidate. Nil when no dropped-input subset parses.
+func (s *Store) rebuildEcNeedleFromDroppedInputs(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)
+	}
+	weight := totalSize * int64(ecCtx.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))
+	picked := make(map[erasure_coding.ShardId]bool)
+	for i, interval := range intervals {
+		shardIntervals, isDeleted := s.gatherEcShardIntervals(needleId, ecVolume, ecCtx, suspects, int(interval.Size), offsets[i], ecCtx.Total())
+		if isDeleted {
+			return nil, true
+		}
+		taken := 0
+		for shardId := 0; shardId < ecCtx.Total() && taken < ecCtx.DataShards; shardId++ {
+			if shardIntervals[shardId] != nil {
+				picked[erasure_coding.ShardId(shardId)] = true
+				taken++
+			}
+		}
+		gathers[i] = shardIntervals
+	}
+
+	for drop := erasure_coding.ShardId(0); int(drop) < ecCtx.Total(); drop++ {
+		if !picked[drop] {
+			continue
+		}
+		candidate := make([]byte, len(corrupted))
+		copy(candidate, corrupted)
+		pos := 0
+		ok := true
+		for i, interval := range intervals {
+			view := slices.Clone(gathers[i])
+			view[drop] = nil
+			if err := reconstructEcShardInterval(ecVolume, ecCtx, view, shardIds[i]); err != nil {
+				ok = false
+				break
+			}
+			copy(candidate[pos:pos+int(interval.Size)], view[shardIds[i]])
+			pos += int(interval.Size)
+		}
+		if !ok {
+			continue
+		}
+		check := &needle.Needle{Id: needleId}
+		if check.ReadBytes(candidate, offset, size, ecVolume.Version) != nil || check.Id != needleId {
+			continue
+		}
+		// The verdict convicts every differing range — and the dropped input,
+		// without which the same decode kept failing to parse.
+		pos = 0
+		for i, interval := range intervals {
+			if !bytes.Equal(candidate[pos:pos+int(interval.Size)], corrupted[pos:pos+int(interval.Size)]) {
+				forgetShardId(ecVolume, shardIds[i])
+			}
+			pos += int(interval.Size)
+		}
+		forgetShardId(ecVolume, drop)
+		*n = *check
+		return candidate, false
+	}
+	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()
@@ -869,17 +1053,18 @@ func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, nee
 	return
 }
 
-// gatherEcShardIntervals collects the same interval from every shard but shardIdToRecover,
+// gatherEcShardIntervals collects the same interval from every shard outside skip,
 // returning one buffer per shard and nil where the shard could not be gathered. It stops
-// once DataShards of them are in hand: reconstruction consumes no more than that.
-func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, ecCtx *erasure_coding.ECContext, shardIdToRecover erasure_coding.ShardId, size int, offset int64) (shardIntervals [][]byte, isDeleted bool) {
+// once want of them are in hand: reconstruction consumes no more than DataShards, so
+// callers pass that unless a retry needs the wider set.
+func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, ecCtx *erasure_coding.ECContext, skip map[erasure_coding.ShardId]bool, size int, offset int64, want int) (shardIntervals [][]byte, isDeleted bool) {
 	shardIntervals = make([][]byte, ecCtx.Total())
 
 	// A shard this server already holds costs no round trip and no peer buffer,
 	// so seed those before asking peers for the rest.
 	available := 0
-	for shardId := erasure_coding.ShardId(0); int(shardId) < ecCtx.Total() && available < ecCtx.DataShards; shardId++ {
-		if shardId == shardIdToRecover {
+	for shardId := erasure_coding.ShardId(0); int(shardId) < ecCtx.Total() && available < want; shardId++ {
+		if skip[shardId] {
 			continue
 		}
 		if _, _, found := s.FindEcVolumeWithShard(ecVolume.VolumeId, shardId); !found {
@@ -899,8 +1084,8 @@ func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasur
 	ecVolume.ShardLocationsLock.RLock()
 	for shardId, locations := range ecVolume.ShardLocations {
 
-		// skip the shard being recovered, one already seeded locally, or an empty shard
-		if shardId == shardIdToRecover || int(shardId) >= ecCtx.Total() || shardIntervals[shardId] != nil {
+		// skip a shard being rebuilt, one already seeded locally, or an empty shard
+		if skip[shardId] || int(shardId) >= ecCtx.Total() || shardIntervals[shardId] != nil {
 			continue
 		}
 		if len(locations) == 0 {
@@ -920,8 +1105,8 @@ func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasur
 	// Reconstruction consumes DataShards buffers, so reading every remaining shard
 	// holds a third more than that and asks a third more of peers that are, by
 	// then, already struggling. Widen only when a wave falls short.
-	for len(candidates) > 0 && available < ecCtx.DataShards {
-		wave := min(ecCtx.DataShards-available, len(candidates))
+	for len(candidates) > 0 && available < want {
+		wave := min(want-available, len(candidates))
 		var wg sync.WaitGroup
 		var fetched atomic.Int64
 		for _, shardId := range candidates[:wave] {
@@ -1018,6 +1203,12 @@ func reconstructEcShardInterval(ecVolume *erasure_coding.EcVolume, ecCtx *erasur
 }
 
 func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, shardIdToRecover erasure_coding.ShardId, buf []byte, offset int64) (n int, is_deleted bool, err error) {
+	return s.recoverEcShardInterval(needleId, ecVolume, shardIdToRecover, buf, offset, map[erasure_coding.ShardId]bool{shardIdToRecover: true})
+}
+
+// recoverEcShardInterval rebuilds shardIdToRecover's interval from the shards
+// outside exclude, which always contains shardIdToRecover itself.
+func (s *Store) recoverEcShardInterval(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, shardIdToRecover erasure_coding.ShardId, buf []byte, offset int64, exclude map[erasure_coding.ShardId]bool) (n int, is_deleted bool, err error) {
 	glog.V(3).Infof("recover ec shard %d.%d from other locations", ecVolume.VolumeId, shardIdToRecover)
 
 	// Reconstruct with the volume's OWN EC ratio (loaded from its .vif), not the
@@ -1049,7 +1240,7 @@ func (s *Store) recoverOneRemoteEcShardInterval(needleId types.NeedleId, ecVolum
 	}
 	defer ecRecoverSem.Release(weight)
 
-	shardIntervals, is_deleted := s.gatherEcShardIntervals(needleId, ecVolume, ecCtx, shardIdToRecover, len(buf), offset)
+	shardIntervals, is_deleted := s.gatherEcShardIntervals(needleId, ecVolume, ecCtx, exclude, len(buf), offset, ecCtx.DataShards)
 	if err := reconstructEcShardInterval(ecVolume, ecCtx, shardIntervals, shardIdToRecover); err != nil {
 		return 0, is_deleted, err
 	}

weed/storage/store_ec_interval_read_test.go

@@ -175,3 +175,142 @@ func TestReadEcShardNeedleUniformLayout(t *testing.T) {
 		}
 	}
 }
+
+func corruptEcShardRange(t *testing.T, ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId, offset int64, size int64) {
+	t.Helper()
+	path := ecVolume.DataBaseFileName() + erasure_coding.ToExt(int(shardId))
+	f, err := os.OpenFile(path, os.O_WRONLY, 0)
+	if err != nil {
+		t.Fatalf("open shard %s: %v", path, err)
+	}
+	defer f.Close()
+	if _, err := f.WriteAt(bytes.Repeat([]byte{0xA5}, int(size)), offset); err != nil {
+		t.Fatalf("corrupt shard %s: %v", path, err)
+	}
+}
+
+// A shard that serves corrupt bytes fails the needle CRC check after the
+// intervals are assembled. The read should rebuild the suspect range from the
+// surviving shards and return the correct needle instead of an error.
+func TestReadEcShardNeedleHealsBitRottedShard(t *testing.T) {
+	store := newTestStore(t, 1)
+	const vid = needle.VolumeId(9)
+
+	n := randomNeedleOfSize(77, 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))
+
+	got := new(needle.Needle)
+	got.Id = n.Id
+	if _, err := store.ReadEcShardNeedle(vid, got, nil); err != nil {
+		t.Fatalf("read should reconstruct around the corrupt shard: %v", err)
+	}
+	if !bytes.Equal(got.Data, n.Data) {
+		t.Fatalf("read back %d bytes, want the %d written", len(got.Data), len(n.Data))
+	}
+}
+
+// Same for a needle striped over several shards: only the rotted interval is
+// rebuilt; the good shards' bytes must survive untouched.
+func TestReadEcShardNeedleHealsBitRottedShardInSpanningNeedle(t *testing.T) {
+	store := newTestStore(t, 1)
+	const vid = needle.VolumeId(10)
+
+	n := randomNeedleOfSize(42, 3*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 interval(s), want it split over several", len(intervals))
+	}
+	shardId, shardOffset := ecVolume.IntervalToShardIdAndOffset(intervals[0])
+	corruptEcShardRange(t, ecVolume, shardId, shardOffset, int64(intervals[0].Size))
+
+	got := new(needle.Needle)
+	got.Id = n.Id
+	if _, err := store.ReadEcShardNeedle(vid, got, nil); err != nil {
+		t.Fatalf("read should reconstruct around the corrupt shard: %v", err)
+	}
+	if !bytes.Equal(got.Data, n.Data) {
+		t.Fatalf("read back %d bytes, want the %d written", len(got.Data), len(n.Data))
+	}
+}
+
+// The gather that rebuilds a suspect range can itself pick up a second corrupt
+// shard as reconstruction input. The first candidate then fails the needle
+// parse, and the read should retry the decode with that input dropped.
+func TestReadEcShardNeedleHealsWhenRebuildInputAlsoRotted(t *testing.T) {
+	store := newTestStore(t, 1)
+	const vid = needle.VolumeId(12)
+
+	n := randomNeedleOfSize(77, 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))
+
+	// Every shard is local, so the rebuild gather takes the lowest-numbered
+	// surviving shards: the first parity shard is always one of the inputs.
+	parityInput := erasure_coding.ShardId(ecVolume.ECContext.DataShards)
+	corruptEcShardRange(t, ecVolume, parityInput, shardOffset, int64(intervals[0].Size))
+
+	got := new(needle.Needle)
+	got.Id = n.Id
+	if _, err := store.ReadEcShardNeedle(vid, got, nil); err != nil {
+		t.Fatalf("read should reconstruct past a corrupt rebuild input: %v", err)
+	}
+	if !bytes.Equal(got.Data, n.Data) {
+		t.Fatalf("read back %d bytes, want the %d written", len(got.Data), len(n.Data))
+	}
+}
+
+// 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.
+func TestReadEcShardNeedleFailsWhenCorruptionExceedsParity(t *testing.T) {
+	store := newTestStore(t, 1)
+	const vid = needle.VolumeId(11)
+
+	n := randomNeedleOfSize(88, 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)
+	}
+	shardId, shardOffset := ecVolume.IntervalToShardIdAndOffset(intervals[0])
+	corruptEcShardRange(t, ecVolume, shardId, shardOffset, int64(intervals[0].Size))
+	corrupted := 0
+	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++
+	}
+
+	got := new(needle.Needle)
+	got.Id = n.Id
+	if _, err := store.ReadEcShardNeedle(vid, got, nil); err == nil {
+		t.Fatal("read should fail when corruption exceeds what parity can rebuild")
+	}
+}