diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index bf973d7fa5..d11486fd47 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -87,6 +87,9 @@ struct Snapshot { offset: Offset, size_for_parse: Size, intervals: Vec, + /// Per-interval (shard_id, shard_offset, size) for the needle-rebuild + /// pass: a parse failure retries the suspect ranges through Reed-Solomon. + metas: Vec, cached_locations: HashMap>, cache_refreshed_at: Option, /// 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, iv: EcInterval, map: EcShardMap<'_>, -) -> io::Result<(Vec, bool)> { + exclude: &HashSet, + want: usize, +) -> (Vec>>, 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>> = 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)> = 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, + iv: EcInterval, + map: EcShardMap<'_>, + exclude: &HashSet, +) -> io::Result<(Vec, 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, 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, + metas: &[EcInterval], + map: EcShardMap<'_>, + corrupted: &[u8], + exclude: &HashSet, + offset: i64, + size: Size, + version: Version, + needle_id: NeedleId, +) -> (Option>, 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, + metas: &[EcInterval], + map: EcShardMap<'_>, + corrupted: &[u8], + offset: i64, + size: Size, + version: Version, + needle_id: NeedleId, +) -> (Option>, bool) { + let suspects: HashSet = 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, + 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(); + 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()); + let mut picked: HashSet = 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 = 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. diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index ac0496a3be..338d5d935c 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/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 diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 6c306046b3..34142e7252 100644 --- a/weed/storage/store_ec.go +++ b/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 } diff --git a/weed/storage/store_ec_interval_read_test.go b/weed/storage/store_ec_interval_read_test.go index dedb57a335..dd603cf678 100644 --- a/weed/storage/store_ec_interval_read_test.go +++ b/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") + } +}