diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index bde0994e5c..c354d05a10 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -56,6 +56,24 @@ fn needle_not_found(needle_id: NeedleId) -> Status { Status::not_found(format!("needle not found {}", needle_id.0)) } +/// The disks holding `shard_id` of `vid`. +fn ec_shard_disks( + store: &crate::storage::store::Store, + vid: VolumeId, + shard_id: ShardId, +) -> Vec { + store + .locations + .iter() + .enumerate() + .filter(|(_, loc)| { + loc.find_ec_volume(vid) + .is_some_and(|ecv| ecv.has_shard(shard_id)) + }) + .map(|(disk, _)| disk) + .collect() +} + /// Mirrors Go's `isNotLeaderErr`: the master returns raft's NotLeaderError, /// which arrives as an Unknown-status message. fn is_not_leader_status(e: &Status) -> bool { @@ -4326,13 +4344,40 @@ impl VolumeServer for VolumeGrpcService { // Mount one shard at a time, returning error on first failure. // Matches Go: for _, shardId := range req.ShardIds { err = vs.store.MountEcShards(...) } + // A failed shard stops the batch, but the shards mounted before it still + // get the sidecar reload and the heartbeat notify below. let mut store = self.state.store.write().unwrap(); + let mut mount_err = None; for &shard_id in &validated { - store - .mount_ec_shard(vid, &req.collection, shard_id, &req.source_disk_type) - .map_err(|e| { - Status::internal(format!("mount {}.{}: {}", req.volume_id, shard_id, e)) - })?; + let held_before = ec_shard_disks(&store, vid, shard_id); + let disk = + match store.mount_ec_shard(vid, &req.collection, shard_id, &req.source_disk_type) { + Ok(disk) => disk, + Err(e) => { + mount_err = Some(Status::internal(format!( + "mount {}.{}: {}", + req.volume_id, shard_id, e + ))); + break; + } + }; + // Mirrors Go's MountEcShards: a caller moving the shard here keeps + // its source copy, and the heartbeat never announces it. A shard + // this call mounted is dropped again, so it does not draw later + // copies to the same disk. + if store.locations[disk] + .find_ec_volume(vid) + .is_some_and(|ecv| ecv.get_io_error_state().2) + { + if !held_before.contains(&disk) { + store.locations[disk].unmount_ec_shards(vid, &[shard_id]); + } + mount_err = Some(Status::internal(format!( + "mount {}.{}: ec volume is quarantined after IO errors, shard not announced", + req.volume_id, shard_id + ))); + break; + } } // A delivery can bring the checksum manifest alongside the shards, but // the receive path only writes the file. When this server already had @@ -4357,6 +4402,9 @@ impl VolumeServer for VolumeGrpcService { } drop(store); self.state.volume_state_notify.notify_one(); + if let Some(status) = mount_err { + return Err(status); + } Ok(Response::new( volume_server_pb::VolumeEcShardsMountResponse {}, @@ -4432,6 +4480,12 @@ impl VolumeServer for VolumeGrpcService { req.volume_id, req.shard_id ))); } + if ec_vol.get_io_error_state().2 { + return Err(Status::unavailable(format!( + "ec shard {}.{} is on a disk quarantined after IO errors", + req.volume_id, req.shard_id + ))); + } // Identity of the shard actually served, echoed on every response chunk so // the client can reject a different encode run even from a pre-upgrade server. @@ -13005,16 +13059,16 @@ mod tests { .unwrap(); } - /// Leaves `service` holding only EC shard 1 of volume 1, with shard 0 (the - /// one with needle 11's bytes) served by a peer; `peer_deletes` journals - /// the needle's delete on the peer alone. + /// Leaves `service` holding `local_shards` of EC volume 1, with shard 0 (the + /// one with needle 11's bytes) also served by a peer; `peer_deletes` + /// journals the needle's delete on the peer alone. async fn put_ec_1_needle_shard_on_a_peer( service: &VolumeGrpcService, tmp: &TempDir, + local_shards: Vec, peer_deletes: bool, ) -> (TempDir, tokio::sync::oneshot::Sender<()>) { - // One local shard, not the needle's: only a peer read can answer. - generate_and_mount_ec_1(service, vec![1]).await; + generate_and_mount_ec_1(service, local_shards).await; let (peer, peer_tmp) = make_local_service_with_volume("", None); peer.state @@ -13070,7 +13124,7 @@ mod tests { async fn test_volume_needle_status_reads_an_ec_needle_from_a_peer_shard() { let (service, tmp) = make_local_service_with_volume("", None); let want = needle_status(&service, 11).await.unwrap(); - let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, false).await; + let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, vec![1], false).await; let got = needle_status(&service, 11) .await @@ -13083,13 +13137,258 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_volume_needle_status_reports_a_peer_reported_ec_deletion() { let (service, tmp) = make_local_service_with_volume("", None); - let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, true).await; + let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, vec![1], true).await; let err = needle_status(&service, 11).await.unwrap_err(); assert_eq!(err.code(), tonic::Code::Unknown, "{err}"); assert_eq!(err.message(), "already deleted"); } + fn quarantine_ec_1(service: &VolumeGrpcService) { + service + .state + .store + .read() + .unwrap() + .find_ec_volume(VolumeId(1)) + .unwrap() + .mark_io_quarantined(); + } + + /// A quarantined disk is not read while a peer holds the shard: the peer's + /// answer, here that the needle is deleted, replaces the local copy's. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_ec_read_skips_quarantined_local_shards() { + let (service, tmp) = make_local_service_with_volume("", None); + let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, (0..14).collect(), true).await; + needle_status(&service, 11) + .await + .expect("healthy local shards answer first"); + + quarantine_ec_1(&service); + let miss = crate::server::store_ec::read_ec_shard_needle_or_miss( + &service.state, + VolumeId(1), + NeedleId(11), + ) + .await + .unwrap() + .unwrap_err(); + assert_eq!(miss, crate::server::store_ec::EcMiss::Deleted); + } + + /// Leaves `service` holding `local_shards` of EC volume 1 on a quarantined + /// disk, with every shard's other holder unreachable. + async fn quarantine_ec_1_without_peers(service: &VolumeGrpcService, local_shards: Vec) { + generate_and_mount_ec_1(service, local_shards).await; + { + let mut store = service.state.store.write().unwrap(); + store.unmount_volume(VolumeId(1)).unwrap(); + store + .find_ec_volume(VolumeId(1)) + .unwrap() + .merge_shard_locations( + (0u8..14) + .map(|sid| (sid, vec!["127.0.0.1:255.1".to_string()])) + .collect(), + ); + } + quarantine_ec_1(service); + } + + /// When no healthy copy answers, the quarantined shards still serve the read. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_ec_read_falls_back_to_quarantined_shards() { + let (service, _tmp) = make_local_service_with_volume("", None); + let want = needle_status(&service, 11).await.unwrap(); + quarantine_ec_1_without_peers(&service, (0..14).collect()).await; + + let got = needle_status(&service, 11) + .await + .expect("only the quarantined disk holds the shards"); + assert_eq!(got, want); + } + + /// Reconstruction falls back to quarantined local shards when the healthy + /// sources cannot supply enough of them. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_ec_read_reconstructs_from_quarantined_shards() { + let (service, _tmp) = make_local_service_with_volume("", None); + let want = needle_status(&service, 11).await.unwrap(); + quarantine_ec_1_without_peers(&service, (4..14).collect()).await; + + let got = needle_status(&service, 11) + .await + .expect("the needle's shard is gone and only quarantined shards can rebuild it"); + assert_eq!(got, want); + } + + /// Peers move on to the next holder or reconstruct when a shard sits on a + /// quarantined disk. + #[tokio::test] + async fn test_volume_ec_shard_read_refuses_a_quarantined_shard() { + let (service, _tmp) = make_local_service_with_volume("", None); + generate_and_mount_ec_1(&service, vec![0]).await; + let read = || { + service.volume_ec_shard_read(Request::new( + volume_server_pb::VolumeEcShardReadRequest { + volume_id: 1, + shard_id: 0, + offset: 0, + size: 8, + ..Default::default() + }, + )) + }; + assert!(read().await.is_ok()); + + quarantine_ec_1(&service); + match read().await { + Ok(_) => panic!("a quarantined shard was served"), + Err(status) => assert_eq!(status.code(), tonic::Code::Unavailable, "{status}"), + } + } + + /// A shard mounted onto a quarantined EC volume fails the mount, so a + /// caller moving it here keeps its source copy. + #[tokio::test] + async fn test_volume_ec_shards_mount_refuses_a_quarantined_volume() { + let (service, _tmp) = make_local_service_with_volume("", None); + generate_and_mount_ec_1(&service, vec![0]).await; + quarantine_ec_1(&service); + + let mount = |shard_id: u32| { + service.volume_ec_shards_mount(Request::new( + volume_server_pb::VolumeEcShardsMountRequest { + volume_id: 1, + shard_ids: vec![shard_id], + ..Default::default() + }, + )) + }; + let shard_mounted = |shard_id: ShardId| { + service + .state + .store + .read() + .unwrap() + .find_ec_volume_with_shard(VolumeId(1), shard_id) + .is_some() + }; + + let err = mount(1).await.unwrap_err(); + assert!(err.message().contains("quarantined"), "{err}"); + assert!(!shard_mounted(1), "the refused shard stayed mounted"); + + let err = mount(0).await.unwrap_err(); + assert!(err.message().contains("quarantined"), "{err}"); + assert!( + shard_mounted(0), + "a refused remount unmounted a shard that was already mounted" + ); + } + + /// While a quarantined sibling holds the same shard, the result follows + /// the disk the store picks for the shard: the healthy disk mounts it, + /// the quarantined one refuses and leaves no copy on the healthy disk. + #[tokio::test] + async fn test_volume_ec_shards_mount_beside_a_quarantined_sibling() { + let (service, tmp) = make_local_service_with_volume("", None); + generate_and_mount_ec_1(&service, vec![]).await; + let sibling = TempDir::new().unwrap(); + let dir = sibling.path().to_str().unwrap(); + let picked = { + let mut store = service.state.store.write().unwrap(); + store + .add_location( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(1.0), + Vec::new(), + ) + .unwrap(); + store.locations[0].unmount_ec_shards(VolumeId(1), &[1]); + for name in ["1.ec01", "1.ecx", "1.ecj", "1.vif"] { + let src = tmp.path().join(name); + if src.exists() { + std::fs::copy(&src, sibling.path().join(name)).unwrap(); + } + } + store.locations[1] + .mount_ec_shards_with_idx_dir(VolumeId(1), "", &[1], dir, "") + .unwrap(); + store.locations[1] + .find_ec_volume(VolumeId(1)) + .unwrap() + .mark_io_quarantined(); + let picked = store.mount_ec_shard(VolumeId(1), "", 1, "").unwrap(); + if picked == 0 { + store.locations[0].unmount_ec_shards(VolumeId(1), &[1]); + } + picked + }; + + let mounted = service + .volume_ec_shards_mount(Request::new(volume_server_pb::VolumeEcShardsMountRequest { + volume_id: 1, + shard_ids: vec![1], + ..Default::default() + })) + .await; + let healthy_holds = service.state.store.read().unwrap().locations[0] + .find_ec_volume(VolumeId(1)) + .is_some_and(|ecv| ecv.has_shard(1)); + assert_eq!( + mounted.is_ok(), + picked == 0, + "mount returned {:?} with disk {picked} picked", + mounted.err() + ); + assert_eq!(mounted.is_ok(), healthy_holds); + } + + /// A shard that fails to mount stops the batch, but the shards mounted + /// before it are still announced. + #[tokio::test] + async fn test_volume_ec_shards_mount_failure_still_announces_earlier_shards() { + use futures::FutureExt; + let (service, tmp) = make_local_service_with_volume("", None); + generate_and_mount_ec_1(&service, vec![0]).await; + let _ = service.state.volume_state_notify.notified().now_or_never(); + std::fs::remove_file(tmp.path().join("1.ec02")).unwrap(); + + service + .volume_ec_shards_mount(Request::new( + volume_server_pb::VolumeEcShardsMountRequest { + volume_id: 1, + shard_ids: vec![1, 2], + ..Default::default() + }, + )) + .await + .unwrap_err(); + assert!( + service + .state + .store + .read() + .unwrap() + .find_ec_volume_with_shard(VolumeId(1), 1) + .is_some() + ); + assert!( + service + .state + .volume_state_notify + .notified() + .now_or_never() + .is_some(), + "shard 1 was mounted but the heartbeat was never told" + ); + } + /// A volume unmounted before the EC read resolves it is "volume not found", /// which fs.verify does not count as a lost needle. #[tokio::test] @@ -13295,7 +13594,7 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 4)] async fn test_batch_delete_peer_reported_ec_deletion_is_not_an_error() { let (service, tmp) = make_local_service_with_volume("", None); - let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, true).await; + let _peer = put_ec_1_needle_shard_on_a_peer(&service, &tmp, vec![1], true).await; assert_eq!( batch_delete_1(&service, 11, 0x3344, true).await, diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index dde4d7106b..07c4209ae1 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -1068,15 +1068,17 @@ fn collect_heartbeat_with_snapshot( master_pb::Heartbeat, Vec, ) { - let (_, expired_ec) = state.store.read().unwrap().find_expired_ec_volumes(); - let mut deleted_ec_shards = Vec::new(); + let (_, expired_ec, mut deleted_ec_shards) = + state.store.read().unwrap().find_expired_ec_volumes(); if !expired_ec.is_empty() { - deleted_ec_shards = state - .store - .write() - .unwrap() - .remove_expired_ec_volumes(expired_ec) - .0; + deleted_ec_shards.extend( + state + .store + .write() + .unwrap() + .remove_expired_ec_volumes(expired_ec) + .0, + ); } #[cfg(test)] { @@ -3083,6 +3085,46 @@ mod tests { assert_eq!(last_ec_shards.len(), 1); } + // The volume heartbeat that quarantines an EC volume tells the master its + // shards are gone, once; neither the next delta nor the next heartbeat + // repeats it. + #[test] + fn test_ec_volume_quarantined_by_a_volume_heartbeat_is_reported_deleted_once() { + let temp_dir = tempfile::tempdir().unwrap(); + let dir = temp_dir.path().to_str().unwrap(); + let state = test_state_with_store(reporting_store(dir, 0)); + mount_test_ec_shard(&state, dir, "ec_delta_kept", 96); + mount_test_ec_shard(&state, dir, "ec_delta_quarantined", 97); + let mut last_ec_shards = collect_ec_shard_delta_messages(&state.store.read().unwrap()); + { + let store = state.store.read().unwrap(); + let ec_vol = store.find_ec_volume(VolumeId(97)).unwrap(); + let eio = crate::storage::volume::media_error_for_test(); + for _ in 0..crate::storage::io_error::IO_ERROR_TOLERANCE { + ec_vol.check_read_write_error(Some(&eio)); + } + } + + let (heartbeat, new_ec_shards, deleted_ec_shards) = + ec_delta_after_volume_tick(&state, &mut last_ec_shards); + assert_eq!( + heartbeat + .deleted_ec_shards + .iter() + .map(|s| s.id) + .collect::>(), + vec![97] + ); + assert!(new_ec_shards.is_empty()); + assert!( + deleted_ec_shards.is_empty(), + "a quarantined EC volume was reported deleted twice" + ); + + let (heartbeat, _, _) = ec_delta_after_volume_tick(&state, &mut last_ec_shards); + assert!(heartbeat.deleted_ec_shards.is_empty()); + } + #[test] fn test_expired_ec_volume_gone_before_removal_is_not_reported_deleted() { let temp_dir = tempfile::tempdir().unwrap(); @@ -3111,7 +3153,7 @@ mod tests { .expire_at_sec = 1; } - let (ec_shards, mut expired) = store.find_expired_ec_volumes(); + let (ec_shards, mut expired, _) = store.find_expired_ec_volumes(); expired.sort(); assert!(ec_shards.is_empty()); assert_eq!(expired, vec![(0, VolumeId(71)), (0, VolumeId(72))]); diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 2860dbab9f..8c2a84ca64 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -40,7 +40,7 @@ use tonic::Request; use crate::pb::master_pb::{self, LookupEcVolumeRequest}; use crate::pb::volume_server_pb::{ - CopyFileRequest, VolumeEcBlobDeleteRequest, VolumeEcShardReadRequest, + CopyFileRequest, VolumeEcBlobDeleteRequest, VolumeEcShardReadRequest, VolumeEcShardReadResponse, }; use crate::server::grpc_client::{ GrpcDialOptions, connect_channel, master_client, parse_grpc_address, volume_server_client, @@ -1035,7 +1035,7 @@ fn scrub_snapshot_under_lock( version: ecv.version, actual_size: get_actual_size(size, ecv.version) as usize, size_for_parse: size, - intervals: read_local_intervals(ecv, &intervals), + intervals: read_local_intervals(ecv, &intervals, true), encode_ts_ns: ecv.encode_ts_ns, }) } @@ -1052,20 +1052,29 @@ struct ScrubSnapshot { } /// Read any locally-held shard intervals, marking the rest `NeedRemote`. Shared -/// by the read-path `build_snapshot` and the scrub snapshot. +/// by the read-path `build_snapshot` and the scrub snapshot. Shards on a +/// quarantined disk count as not local unless `allow_quarantined` is set. fn read_local_intervals( ecv: &crate::storage::erasure_coding::ec_volume::EcVolume, intervals: &[crate::storage::erasure_coding::ec_locate::Interval], + allow_quarantined: bool, ) -> Vec { + let skip_local = !allow_quarantined && ecv.get_io_error_state().2; let mut interval_results = Vec::with_capacity(intervals.len()); for interval in intervals { let (shard_id, shard_offset) = ecv.interval_to_shard_id_and_offset(interval); let buf_size = interval.size as usize; - let local = ecv.shards.get(shard_id as usize).and_then(|s| s.as_ref()); + let local = ecv + .shards + .get(shard_id as usize) + .and_then(|s| s.as_ref()) + .filter(|_| !skip_local); match local { Some(shard) => { let mut buf = vec![0u8; buf_size]; - match shard.read_at(&mut buf, shard_offset as u64) { + let read = shard.read_at(&mut buf, shard_offset as u64); + ecv.check_read_write_error(read.as_ref().err()); + match read { Ok(n) if n == buf_size => interval_results.push(IntervalResult::Local(buf)), _ => interval_results.push(IntervalResult::NeedRemote { shard_id, @@ -1101,7 +1110,7 @@ fn build_snapshot( )); } let actual = get_actual_size(size, ecv.version); - let interval_results = read_local_intervals(ecv, intervals); + let interval_results = read_local_intervals(ecv, intervals, false); let metas = intervals .iter() .map(|iv| { @@ -1359,9 +1368,52 @@ async fn fetch_one_interval( let result = recover_one_remote_ec_shard_interval(state, iv, map, &HashSet::from([iv.shard_id])).await; note_ec_reconstructed_interval(iv, &result); + if result.is_err() + && let Some(buf) = read_quarantined_local_interval(state, iv) + { + tracing::info!( + "read ec shard {}.{} from its quarantined disk: no healthy copy answered", + vid.0, + shard_id + ); + return Ok((buf, false)); + } result } +/// Last resort once peers and reconstruction failed: the interval from a local +/// shard skipped because its disk is quarantined. Mirrors the tail of Go's +/// `readOneEcShardInterval`. +fn read_quarantined_local_interval( + state: &Arc, + iv: EcInterval, +) -> Option> { + let store = state.store.read().unwrap(); + let ecv = store.find_ec_volume_with_shard(iv.vid, iv.shard_id)?; + if !ecv.get_io_error_state().2 + || (iv.expected_encode_ts_ns != 0 && ecv.encode_ts_ns != iv.expected_encode_ts_ns) + { + return None; + } + let shard = ecv.shards.get(iv.shard_id as usize)?.as_ref()?; + let mut buf = vec![0u8; iv.size]; + let read = shard.read_at(&mut buf, iv.shard_offset as u64); + ecv.check_read_write_error(read.as_ref().err()); + match read { + Ok(n) if n == iv.size => Some(buf), + result => { + tracing::info!( + "read quarantined ec shard {}.{} offset {}: {:?}", + iv.vid.0, + iv.shard_id, + iv.shard_offset, + result + ); + None + } + } +} + /// Counts one interval that couldn't be read directly and was reconstructed, /// and at debug logs it with the needle and the encode generation, which the /// counter can't carry. `success` means the reconstruction produced data; the @@ -1461,15 +1513,30 @@ async fn do_read_remote_ec_shard_interval( vid.0, shard_id, source, e )) })?; - let mut stream = resp.into_inner(); + collect_ec_shard_read(resp.into_inner(), source, iv).await +} +/// Reads one `VolumeEcShardRead` reply into the interval's bytes, or the +/// holder's deleted answer, which stands even if the stream fails after it. +async fn collect_ec_shard_read( + mut stream: tonic::Streaming, + source: &str, + iv: EcInterval, +) -> io::Result<(Vec, bool)> { + let EcInterval { + vid, + shard_id, + size, + expected_encode_ts_ns, + .. + } = iv; let mut out = Vec::with_capacity(size); let mut is_deleted = false; - while let Some(msg) = stream - .message() - .await - .map_err(|e| io::Error::other(format!("recv: {}", e)))? - { + // A deleted answer stands even if the stream fails after it. + while let Some(msg) = match stream.message().await { + Err(_) if is_deleted => None, + msg => msg.map_err(|e| io::Error::other(format!("recv: {}", e)))?, + } { // Validate the served shard's identity client-side, so the guard holds even // against a pre-upgrade server that ignored the request field (returns 0). // A mismatch fails the read; the caller recovers from parity. @@ -1548,6 +1615,7 @@ async fn gather_ec_shard_intervals( // shard_locations cache or a failed master lookup, local // survivors still contribute. let mut available = 0usize; + let mut quarantined = Vec::new(); { let store = state.store.read().unwrap(); for (sid, slot) in bufs.iter_mut().enumerate() { @@ -1573,13 +1641,15 @@ async fn gather_ec_shard_intervals( } _ => continue, }; + if owner.get_io_error_state().2 { + quarantined.push(sid_shard); + continue; + } if let Some(Some(shard)) = owner.shards.get(sid) { let mut buf = vec![0u8; size]; - if shard - .read_at(&mut buf, shard_offset as u64) - .map(|n| n == size) - .unwrap_or(false) - { + let read = shard.read_at(&mut buf, shard_offset as u64); + owner.check_read_write_error(read.as_ref().err()); + if read.map(|n| n == size).unwrap_or(false) { *slot = Some(buf); available += 1; } @@ -1655,6 +1725,31 @@ async fn gather_ec_shard_intervals( } } + // Healthy sources fell short: fill in from the quarantined local disk. + for sid in quarantined { + if any_deleted || available >= want { + break; + } + if bufs[sid as usize].is_some() { + continue; + } + if let Some(buf) = read_quarantined_local_interval( + state, + EcInterval { + shard_id: sid, + ..iv + }, + ) { + tracing::debug!( + "recover: read ec shard {}.{} from its quarantined disk", + vid.0, + sid + ); + bufs[sid as usize] = Some(buf); + available += 1; + } + } + (bufs, any_deleted) } @@ -2686,4 +2781,58 @@ mod tests { // An unrefreshed cache always looks up. assert!(needs_refresh(&locations(0), None, false, 10, 14)); } + + struct FailAfterFrame(Option); + + impl http_body::Body for FailAfterFrame { + type Data = bytes::Bytes; + type Error = tonic::Status; + + fn poll_frame( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll, tonic::Status>>> + { + std::task::Poll::Ready(Some(match self.0.take() { + Some(frame) => Ok(http_body::Frame::data(frame)), + None => Err(tonic::Status::unavailable( + "stream failed after the deleted answer", + )), + })) + } + } + + /// A holder's deleted answer stands even when its stream fails afterwards, + /// so a later source or the quarantined shard cannot answer for the needle. + #[tokio::test] + async fn test_collect_ec_shard_read_keeps_deletion_when_the_stream_fails() { + let message = prost::Message::encode_to_vec(&VolumeEcShardReadResponse { + is_deleted: true, + ..Default::default() + }); + let mut frame = vec![0u8]; + frame.extend_from_slice(&(message.len() as u32).to_be_bytes()); + frame.extend_from_slice(&message); + let stream = tonic::Streaming::new_response( + tonic_prost::ProstDecoder::::default(), + FailAfterFrame(Some(frame.into())), + tonic::codegen::http::StatusCode::OK, + None, + None, + ); + let iv = EcInterval { + vid: VolumeId(7), + needle_id: NeedleId(42), + shard_id: 0, + shard_offset: 0, + size: 64, + expected_encode_ts_ns: 0, + }; + + let (buf, is_deleted) = collect_ec_shard_read(stream, "peer:18080", iv) + .await + .unwrap(); + assert!(is_deleted); + assert!(buf.is_empty()); + } } diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index 44f5b3dcd9..3fe1051124 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -1182,13 +1182,14 @@ impl Store { /// Matches Go's Store.MountEcShards which mounts one shard at a time. /// `source_disk_type` is the source volume's disk type from the /// `VolumeEcShardsMount` RPC (#9423); pass `""` for non-RPC paths. + /// Returns the index of the disk the shard mounted on. pub fn mount_ec_shard( &mut self, vid: VolumeId, collection: &str, shard_id: ShardId, source_disk_type: &str, - ) -> Result<(), VolumeError> { + ) -> Result { self.stamp_allocated_vid(vid); // The .ecx may live on a different disk than the shard being mounted // (ec.balance / ec.rebuild can spread shards across sibling disks), so @@ -1201,7 +1202,7 @@ impl Store { // next. Like Go, a NotFound just means "not this disk"; anything else // is collected so an all-disks-fail error names every disk tried. let mut failures: Vec<(String, VolumeError)> = Vec::new(); - for loc in &mut self.locations { + for (disk, loc) in self.locations.iter_mut().enumerate() { // Check if the shard file exists on this location let shard = EcVolumeShard::new(&loc.directory, collection, vid, shard_id); if !std::path::Path::new(&shard.file_name()).exists() { @@ -1226,7 +1227,7 @@ impl Store { &idx_dir, source_disk_type, ) { - Ok(()) => return Ok(()), + Ok(()) => return Ok(disk), Err(VolumeError::Io(e)) if e.kind() == io::ErrorKind::NotFound => {} Err(e) => failures.push((loc.directory.clone(), e)), } @@ -1491,22 +1492,27 @@ impl Store { Vec, Vec, ) { - let (mut ec_shards, expired) = self.find_expired_ec_volumes(); - let (deleted, still_held) = self.remove_expired_ec_volumes(expired); + let (mut ec_shards, expired, quarantined) = self.find_expired_ec_volumes(); + let (mut deleted, still_held) = self.remove_expired_ec_volumes(expired); ec_shards.extend(still_held); + deleted.extend(quarantined); (ec_shards, deleted) } - /// The read half of `delete_expired_ec_volumes`: the shards to report, and - /// the expired EC volumes, by disk index, for `remove_expired_ec_volumes`. + /// The read half of `delete_expired_ec_volumes`: the shards to report, the + /// expired EC volumes, by disk index, for `remove_expired_ec_volumes`, and + /// the shards of volumes this pass quarantined, which go out as deleted + /// once so the master drops them before the next full EC heartbeat. pub fn find_expired_ec_volumes( &self, ) -> ( Vec, Vec<(usize, VolumeId)>, + Vec, ) { let mut ec_shards = Vec::new(); let mut expired = Vec::new(); + let mut quarantined_now = Vec::new(); for (disk_id, loc) in self.locations.iter().enumerate() { for (vid, ec_vol) in loc.ec_volumes() { @@ -1521,6 +1527,8 @@ impl Store { io_count, "ec volume quarantined after consecutive IO errors" ); + quarantined_now + .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)); } } else { ec_shards @@ -1529,7 +1537,7 @@ impl Store { } } - (ec_shards, expired) + (ec_shards, expired, quarantined_now) } /// The write half of `delete_expired_ec_volumes`: destroys each volume that diff --git a/weed/server/volume_grpc_ec_shard_read_test.go b/weed/server/volume_grpc_ec_shard_read_test.go new file mode 100644 index 0000000000..103d9b48b8 --- /dev/null +++ b/weed/server/volume_grpc_ec_shard_read_test.go @@ -0,0 +1,101 @@ +package weed_server + +import ( + "os" + "path/filepath" + "testing" + + "google.golang.org/grpc" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +type ecShardReadRecorder struct { + grpc.ServerStream + responses []*volume_server_pb.VolumeEcShardReadResponse +} + +func (r *ecShardReadRecorder) Send(m *volume_server_pb.VolumeEcShardReadResponse) error { + r.responses = append(r.responses, m) + return nil +} + +// ecVolumeWithLiveNeedle mounts shard 0 of an EC volume whose .ecx holds one +// live needle, and returns the server and the .ecx path. +func ecVolumeWithLiveNeedle(t *testing.T, vid needle.VolumeId, id types.NeedleId) (*VolumeServer, string) { + t.Helper() + dir := t.TempDir() + base := filepath.Join(dir, vid.String()) + + entry := make([]byte, types.NeedleIdSize+types.OffsetSize+types.SizeSize) + types.NeedleIdToBytes(entry[0:types.NeedleIdSize], id) + types.OffsetToBytes(entry[types.NeedleIdSize:types.NeedleIdSize+types.OffsetSize], types.Offset{}) + types.SizeToBytes(entry[types.NeedleIdSize+types.OffsetSize:], types.Size(100)) + + for name, content := range map[string][]byte{ + base + ".ecx": entry, + base + ".ecj": nil, + base + ".ec00": make([]byte, 8), + base + ".vif": {}, + } { + if err := os.WriteFile(name, content, 0o644); err != nil { + t.Fatalf("write %s: %v", name, err) + } + } + + location := storage.NewDiskLocation(dir, 100, util.MinFreeSpace{}, dir, types.HardDriveType, nil, stats.DiskIOProbeConfig{}) + t.Cleanup(location.Close) + if _, err := location.LoadEcShard("", vid, erasure_coding.ShardId(0)); err != nil { + t.Fatalf("load ec shard: %v", err) + } + return &VolumeServer{store: &storage.Store{Locations: []*storage.DiskLocation{location}}}, base + ".ecx" +} + +// Peers move on to the next holder or reconstruct when a shard sits on a +// quarantined disk. +func TestVolumeEcShardRead_RefusesQuarantinedShard(t *testing.T) { + const vid = needle.VolumeId(7) + vs, _ := ecVolumeWithLiveNeedle(t, vid, 1) + req := &volume_server_pb.VolumeEcShardReadRequest{VolumeId: uint32(vid), ShardId: 0, Offset: 0, Size: 8, FileKey: 1} + + if err := vs.VolumeEcShardRead(req, &ecShardReadRecorder{}); err != nil { + t.Fatalf("read from a healthy shard: %v", err) + } + + ecVolume, _ := vs.store.FindEcVolume(vid) + ecVolume.MarkIoQuarantined() + recorder := &ecShardReadRecorder{} + if err := vs.VolumeEcShardRead(req, recorder); err == nil { + t.Error("a quarantined shard was served") + } + if len(recorder.responses) != 0 { + t.Errorf("sent %d responses for a quarantined shard, want 0", len(recorder.responses)) + } +} + +// An .ecx that cannot be read must fail the request: answering "deleted" would +// make the caller drop a live needle. +func TestVolumeEcShardRead_EcxReadErrorIsNotADelete(t *testing.T) { + const vid = needle.VolumeId(7) + vs, ecxPath := ecVolumeWithLiveNeedle(t, vid, 1) + if err := os.Truncate(ecxPath, 0); err != nil { + t.Fatalf("truncate .ecx: %v", err) + } + + recorder := &ecShardReadRecorder{} + err := vs.VolumeEcShardRead(&volume_server_pb.VolumeEcShardReadRequest{VolumeId: uint32(vid), ShardId: 0, Offset: 0, Size: 8, FileKey: 1}, recorder) + for _, resp := range recorder.responses { + if resp.IsDeleted { + t.Fatal("an unreadable .ecx was reported as a deleted needle") + } + } + if err == nil { + t.Error("an unreadable .ecx did not fail the read") + } +} diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 22073ea64b..f83aa2fa13 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -1016,6 +1016,9 @@ func (vs *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_ser vs.recoverMissingEcIndexes(req.VolumeId) } + // A failed shard stops the batch, but the shards mounted before it still + // get the sidecar reload below. + var mountErr error for _, shardId := range req.ShardIds { err := vs.store.MountEcShards(req.Collection, needle.VolumeId(req.VolumeId), erasure_coding.ShardId(shardId), req.SourceDiskType) @@ -1026,7 +1029,8 @@ func (vs *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_ser } if err != nil { - return nil, fmt.Errorf("mount %d.%d: %v", req.VolumeId, shardId, err) + mountErr = fmt.Errorf("mount %d.%d: %v", req.VolumeId, shardId, err) + break } } @@ -1050,6 +1054,9 @@ func (vs *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_ser for _, v := range vs.store.FindAllEcVolumes(needle.VolumeId(req.VolumeId)) { v.ReloadBitrotSidecar(ecMetadataDirs...) } + if mountErr != nil { + return nil, mountErr + } return &volume_server_pb.VolumeEcShardsMountResponse{}, nil } @@ -1095,10 +1102,16 @@ func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardRea if req.EncodeTsNs != 0 && req.EncodeTsNs != ecVolume.EncodeTsNs { return fmt.Errorf("ec shard %d.%d belongs to a different encode run", req.VolumeId, req.ShardId) } + if _, _, quarantined := ecVolume.GetIoErrorState(); quarantined { + return fmt.Errorf("ec shard %d.%d is on a disk quarantined after IO errors", req.VolumeId, req.ShardId) + } if req.FileKey != 0 { - _, size, _ := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey)) - if size.IsDeleted() { + _, size, err := ecVolume.FindNeedleFromEcx(types.Uint64ToNeedleId(req.FileKey)) + if err != nil && !errors.Is(err, erasure_coding.NotFoundError) { + return fmt.Errorf("ec shard %d.%d locate needle %d: %w", req.VolumeId, req.ShardId, req.FileKey, err) + } + if err == nil && size.IsDeleted() { return stream.Send(&volume_server_pb.VolumeEcShardReadResponse{ IsDeleted: true, EncodeTsNs: ecVolume.EncodeTsNs, diff --git a/weed/storage/store.go b/weed/storage/store.go index f77dae808c..05781ef1e4 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -170,6 +170,9 @@ func NewStore( location.ecShardNotifyHandler = func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume) { s.UpdateMaxVolumeIdSeen(vid) + if _, _, quarantined := ecVolume.GetIoErrorState(); quarantined { + return + } si := erasure_coding.NewShardsInfo() si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize()))) fileCount, deleteCount := ecVolume.FileAndDeleteCount() @@ -900,13 +903,16 @@ func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeE // Quarantine EC volumes on faulty media: keep them in memory (so // healthz can observe the quarantine state and an operator can // recover) but stop reporting them so the master re-replicates - // from healthy peers. Mirrors the regular volume quarantine. + // from healthy peers. Mirrors the regular volume quarantine. The + // shards go out as deleted once, so the master drops them now rather + // than at the next full EC heartbeat. for _, ev := range ioQuarantinedEvs { ioErr, ioCount, quarantined := ev.GetIoErrorState() if !quarantined { ev.MarkIoQuarantined() glog.Warningf("ec volume %d quarantined after %d consecutive IO errors: %v", ev.VolumeId, ioCount, ioErr) + deleted = append(deleted, ev.ToVolumeEcShardInformationMessage(uint32(diskId))...) } } } diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index ff8fbf224c..6dd65bd865 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -253,11 +253,22 @@ func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat { } func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, sourceDiskType string) error { + ownersBefore := s.EcShardOwnerDisks(vid, []erasure_coding.ShardId{shardId}) ecVolume, diskId, err := s.LoadEcShardAcrossDisks(collection, vid, shardId) if err != nil { return err } glog.V(0).Infof("MountEcShards %d.%d on disk ID %d", vid, shardId, diskId) + // The master must not route reads to a disk already found faulty, and a + // caller moving the shard here keeps its source copy on this error. A + // shard this call loaded is dropped again, so it does not draw later + // copies to the same disk. + if _, _, quarantined := ecVolume.GetIoErrorState(); quarantined { + if location := s.Locations[diskId]; !slices.Contains(ownersBefore, location) { + location.UnloadEcShard(vid, shardId) + } + return fmt.Errorf("ec volume %d on disk %d is quarantined after IO errors, shard %d not announced", vid, diskId, shardId) + } // Apply the orchestrator-supplied source disk type so the EC // volume reports under it instead of the location's. Empty means @@ -602,8 +613,11 @@ var ( const ecIntervalReadConcurrency = 8 func (s *Store) readEcShardIntervals(needleId types.NeedleId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, is_deleted bool, err error) { - if err = s.cachedLookupEcShardLocations(ecVolume); err != nil { - return nil, false, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.masterAddress(), err) + // Like the Rust server, read on with the cached locations when the lookup + // fails: local shards, a quarantined disk's last-resort copy included, do + // not need it. + if lookupErr := s.cachedLookupEcShardLocations(ecVolume); lookupErr != nil { + glog.V(0).Infof("locate ec volume %d via master grpc %s, using cached locations: %v", ecVolume.VolumeId, s.masterAddress(), lookupErr) } var totalSize int @@ -657,15 +671,15 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur shardId, actualOffset := ecVolume.IntervalToShardIdAndOffset(interval) // try local read - err = s.readLocalEcShardInterval(ecVolume, shardId, data, actualOffset) - if err == nil { + localErr := s.readLocalEcShardInterval(ecVolume, shardId, data, actualOffset, false) + if localErr == nil { return } - if errors.Is(err, errShardNotLocal) { + if errors.Is(localErr, errShardNotLocal) { // expected when shards are spread across servers; fall through to remote read - glog.V(4).Infof("ec shard %d.%d not local, will try remote", ecVolume.VolumeId, shardId) + glog.V(4).Infof("ec shard %d.%d not local, will try remote: %v", ecVolume.VolumeId, shardId, localErr) } else { - glog.V(0).Infof("read local ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, actualOffset, err) + glog.V(0).Infof("read local ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, actualOffset, localErr) } ecVolume.ShardLocationsLock.RLock() @@ -693,6 +707,15 @@ func (s *Store) readOneEcShardInterval(needleId types.NeedleId, ecVolume *erasur } glog.V(0).Infof("recover ec shard %d.%d : %v", ecVolume.VolumeId, shardId, err) + if !is_deleted && errors.Is(localErr, errShardQuarantined) { + lastErr := s.readLocalEcShardInterval(ecVolume, shardId, data, actualOffset, true) + if lastErr == nil { + glog.V(0).Infof("read ec shard %d.%d from its quarantined disk: no healthy copy answered", ecVolume.VolumeId, shardId) + return false, nil + } + glog.V(0).Infof("read quarantined ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, actualOffset, lastErr) + } + return } @@ -1149,7 +1172,14 @@ func (s *Store) cachedLookupEcShardLocations(ecVolume *erasure_coding.EcVolume) return } -func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId, buf []byte, offset int64) error { +// errShardQuarantined marks a local shard skipped because its disk is +// quarantined after repeated IO errors. It reads as not local, so callers go +// to peers or reconstruction first. +var errShardQuarantined = fmt.Errorf("ec shard on a quarantined disk: %w", errShardNotLocal) + +// readLocalEcShardInterval refuses a shard whose disk is quarantined unless +// allowQuarantined is set. +func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shardId erasure_coding.ShardId, buf []byte, offset int64, allowQuarantined bool) error { // Resolve the shard together with the EcVolume on the disk that owns it; the // shard may live on a sibling disk of this server. ownerVolume, shard, found := s.FindEcVolumeWithShard(ecVolume.VolumeId, shardId) @@ -1164,6 +1194,9 @@ func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shar glog.V(1).Infof("skip local ec shard %d.%d from a different encode run: caller EncodeTsNs %d, local %d", ecVolume.VolumeId, shardId, ecVolume.EncodeTsNs, ownerVolume.EncodeTsNs) return fmt.Errorf("shard %d for volume %d: %w", shardId, ecVolume.VolumeId, errShardNotLocal) } + if _, _, quarantined := ownerVolume.GetIoErrorState(); quarantined && !allowQuarantined { + return fmt.Errorf("shard %d for volume %d: %w", shardId, ecVolume.VolumeId, errShardQuarantined) + } readBytes, err := shard.ReadAt(buf, offset) if err != nil { @@ -1219,6 +1252,10 @@ func (s *Store) doReadRemoteEcShardInterval(sourceDataNode pb.ServerAddress, nee break } if receiveErr != nil { + // A deleted answer stands even if the stream fails after it. + if is_deleted { + break + } return fmt.Errorf("receiving ec shard %d.%d from %s: %v", vid, shardId, sourceDataNode, receiveErr) } // Validate the served shard's identity client-side, so the guard holds @@ -1263,6 +1300,7 @@ func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasur // 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 + var quarantined []erasure_coding.ShardId for shardId := erasure_coding.ShardId(0); int(shardId) < ecCtx.Total() && available < want; shardId++ { if skip[shardId] { continue @@ -1271,7 +1309,10 @@ func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasur continue } data := make([]byte, size) - if localErr := s.readLocalEcShardInterval(ecVolume, shardId, data, offset); localErr != nil { + if localErr := s.readLocalEcShardInterval(ecVolume, shardId, data, offset, false); localErr != nil { + if errors.Is(localErr, errShardQuarantined) { + quarantined = append(quarantined, shardId) + } glog.V(3).Infof("recover: read local ec shard %d.%d: %v", ecVolume.VolumeId, shardId, localErr) continue } @@ -1338,7 +1379,26 @@ func (s *Store) gatherEcShardIntervals(needleId types.NeedleId, ecVolume *erasur } } - return shardIntervals, isDeletedFlag.Load() + isDeleted = isDeletedFlag.Load() + // Healthy sources fell short: fill in from the quarantined local disk. + for _, shardId := range quarantined { + if isDeleted || available >= want { + break + } + if shardIntervals[shardId] != nil { + continue + } + data := make([]byte, size) + if readErr := s.readLocalEcShardInterval(ecVolume, shardId, data, offset, true); readErr != nil { + glog.V(0).Infof("recover: read quarantined ec shard %d.%d offset %d: %v", ecVolume.VolumeId, shardId, offset, readErr) + continue + } + glog.V(1).Infof("recover: read ec shard %d.%d from its quarantined disk", ecVolume.VolumeId, shardId) + shardIntervals[shardId] = data + available++ + } + + return shardIntervals, isDeleted } // checkEcShardRebuildable rejects a parity target. ReconstructData rebuilds data shards diff --git a/weed/storage/store_ec_quarantine_test.go b/weed/storage/store_ec_quarantine_test.go new file mode 100644 index 0000000000..c40948888e --- /dev/null +++ b/weed/storage/store_ec_quarantine_test.go @@ -0,0 +1,278 @@ +package storage + +import ( + "bytes" + "fmt" + "net" + "os" + "path/filepath" + "syscall" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func startEcShardPeer(t *testing.T, peer volume_server_pb.VolumeServerServer) pb.ServerAddress { + t.Helper() + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, peer) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + return pb.NewServerAddressWithGrpcPort("127.0.0.1:1", lis.Addr().(*net.TCPAddr).Port) +} + +// mountEcShardRange mounts shards [from, to) of an EC volume written into the +// store's first disk. +func mountEcShardRange(t *testing.T, store *Store, vid needle.VolumeId, from, to int) (baseFileName string, ecVolume *erasure_coding.EcVolume, n *needle.Needle) { + t.Helper() + baseFileName, n = writeEcVolumeFiles(t, store.Locations[0].Directory, vid) + for shardId := from; shardId < to; shardId++ { + if err := store.MountEcShards("", vid, erasure_coding.ShardId(shardId), ""); err != nil { + t.Fatalf("mount shard %d: %v", shardId, err) + } + } + ecVolume, found := store.Locations[0].FindEcVolume(vid) + if !found { + t.Fatal("ec volume not mounted") + } + return baseFileName, ecVolume, n +} + +func readWrittenEcNeedle(t *testing.T, store *Store, vid needle.VolumeId, want *needle.Needle) error { + t.Helper() + got := new(needle.Needle) + got.Id = want.Id + if _, err := store.ReadEcShardNeedle(vid, got, nil); err != nil { + return err + } + if !bytes.Equal(got.Data, want.Data) { + return fmt.Errorf("needle %s data differs from what was written", want.Id) + } + return nil +} + +func quarantineEcVolume(ev *erasure_coding.EcVolume) { + for i := 0; i < IoErrorTolerance; i++ { + ev.CheckReadWriteError(fmt.Errorf("read shard: %w", syscall.EIO)) + } +} + +// A quarantined disk is not read while a peer can serve the shard. +func TestReadEcShardNeedleSkipsQuarantinedLocalShards(t *testing.T) { + store := newTestStore(t, 1) + store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + const vid = needle.VolumeId(7) + baseFileName, ecVolume, n := mountEcShardRange(t, store, vid, 0, erasure_coding.TotalShardsCount) + + peer := &shardServingPeer{baseFileName: baseFileName} + seedShardLocations(ecVolume, startEcShardPeer(t, peer)) + + if err := readWrittenEcNeedle(t, store, vid, n); err != nil { + t.Fatalf("read before quarantine: %v", err) + } + if requests := peer.requests.Load(); requests != 0 { + t.Fatalf("healthy local shards sent %d reads to the peer, want 0", requests) + } + + ecVolume.MarkIoQuarantined() + if err := readWrittenEcNeedle(t, store, vid, n); err != nil { + t.Fatalf("read after quarantine: %v", err) + } + if peer.requests.Load() == 0 { + t.Error("the needle was read from the quarantined disk while a peer held its shards") + } +} + +// When no healthy copy answers, the quarantined shard is still read rather than +// failing the needle. +func TestReadEcShardNeedleFallsBackToQuarantinedShards(t *testing.T) { + store := newTestStore(t, 1) + store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + const vid = needle.VolumeId(7) + _, ecVolume, n := mountEcShardRange(t, store, vid, 0, erasure_coding.TotalShardsCount) + seedShardLocations(ecVolume, startEcShardPeer(t, &volume_server_pb.UnimplementedVolumeServerServer{})) + + ecVolume.MarkIoQuarantined() + if err := readWrittenEcNeedle(t, store, vid, n); err != nil { + t.Fatalf("read with only the quarantined disk holding the shards: %v", err) + } +} + +// deletedThenFailingEcShardPeer reports the needle deleted, then fails the stream. +type deletedThenFailingEcShardPeer struct { + volume_server_pb.UnimplementedVolumeServerServer +} + +func (p *deletedThenFailingEcShardPeer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardReadRequest, stream volume_server_pb.VolumeServer_VolumeEcShardReadServer) error { + if err := stream.Send(&volume_server_pb.VolumeEcShardReadResponse{IsDeleted: true, EncodeTsNs: req.EncodeTsNs}); err != nil { + return err + } + return status.Error(codes.Unavailable, "stream failed after the deleted answer") +} + +// A peer's deleted answer stands even when its stream fails afterwards, so a +// later source or the quarantined shard cannot answer for a deleted needle. +func TestReadRemoteEcShardIntervalKeepsDeletionWhenTheStreamFails(t *testing.T) { + store := newTestStore(t, 1) + store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + sources := []pb.ServerAddress{ + startEcShardPeer(t, &deletedThenFailingEcShardPeer{}), + startEcShardPeer(t, &volume_server_pb.UnimplementedVolumeServerServer{}), + } + + _, isDeleted, err := store.readRemoteEcShardInterval(sources, 42, 7, 0, make([]byte, 64), 0, 0) + if err != nil || !isDeleted { + t.Errorf("read got deleted=%v err=%v, want deleted and no error", isDeleted, err) + } +} + +func ecShardMessagesFor(messages []*master_pb.VolumeEcShardInformationMessage, vid needle.VolumeId) []*master_pb.VolumeEcShardInformationMessage { + var found []*master_pb.VolumeEcShardInformationMessage + for _, m := range messages { + if needle.VolumeId(m.Id) == vid { + found = append(found, m) + } + } + return found +} + +// The heartbeat that quarantines an EC volume tells the master its shards are +// gone, once, instead of leaving them registered until the next full EC +// heartbeat. +func TestCollectHeartbeatReportsQuarantinedEcShardsDeletedOnce(t *testing.T) { + store := newTestStore(t, 1) + const vid = needle.VolumeId(7) + _, ecVolume, _ := mountEcShardRange(t, store, vid, 0, erasure_coding.TotalShardsCount) + + if deleted := ecShardMessagesFor(store.CollectHeartbeat().DeletedEcShards, vid); len(deleted) != 0 { + t.Fatalf("healthy ec volume reported deleted: %v", deleted) + } + + quarantineEcVolume(ecVolume) + deleted := ecShardMessagesFor(store.CollectHeartbeat().DeletedEcShards, vid) + if len(deleted) != 1 { + t.Fatalf("quarantining heartbeat carried %d deleted messages for volume %d, want 1", len(deleted), vid) + } + if got := erasure_coding.ShardsInfoFromVolumeEcShardInformationMessage(deleted[0]).Count(); got != erasure_coding.TotalShardsCount { + t.Errorf("deleted message names %d shards, want %d", got, erasure_coding.TotalShardsCount) + } + if _, _, quarantined := ecVolume.GetIoErrorState(); !quarantined { + t.Fatal("ec volume not quarantined") + } + + if deleted := ecShardMessagesFor(store.CollectHeartbeat().DeletedEcShards, vid); len(deleted) != 0 { + t.Errorf("next heartbeat reported the quarantined shards deleted again: %v", deleted) + } +} + +// A shard mounted onto a quarantined EC volume is not announced, and the mount +// fails so a caller moving the shard here keeps its source copy. +func TestMountEcShardsRefusesQuarantinedEcVolume(t *testing.T) { + store := newTestStore(t, 1) + const vid = needle.VolumeId(7) + last := erasure_coding.ShardId(erasure_coding.TotalShardsCount - 1) + _, ecVolume, _ := mountEcShardRange(t, store, vid, 0, int(last)) + for len(store.NewEcShardsChan) > 0 { + <-store.NewEcShardsChan + } + + ecVolume.MarkIoQuarantined() + if err := store.MountEcShards("", vid, last, ""); err == nil { + t.Error("mount onto a quarantined ec volume succeeded") + } + if _, _, found := store.FindEcShard(vid, last); found { + t.Error("the refused shard stayed loaded") + } + if err := store.MountEcShards("", vid, 0, ""); err == nil { + t.Error("remount of a shard on a quarantined ec volume succeeded") + } + if _, _, found := store.FindEcShard(vid, 0); !found { + t.Error("a refused remount unloaded a shard that was already mounted") + } + store.Locations[0].ecShardNotifyHandler("", vid, last, ecVolume) + if pending := len(store.NewEcShardsChan); pending != 0 { + t.Errorf("%d shard announcements queued for a quarantined ec volume, want 0", pending) + } +} + +// While a quarantined sibling holds the same shard, the result follows the +// disk the store picks for the shard: the healthy disk mounts it, the +// quarantined one refuses and leaves no copy on the healthy disk. +func TestMountEcShardsBesideAQuarantinedSibling(t *testing.T) { + store := newTestStore(t, 2) + const vid = needle.VolumeId(7) + baseFileName, _ := writeEcVolumeFiles(t, store.Locations[0].Directory, vid) + sibling := store.Locations[1] + for _, ext := range []string{".ec01", ".ecx", ".ecj", ".vif"} { + data, err := os.ReadFile(baseFileName + ext) + if err != nil { + t.Fatalf("read %s: %v", ext, err) + } + if err := os.WriteFile(filepath.Join(sibling.Directory, filepath.Base(baseFileName)+ext), data, 0o644); err != nil { + t.Fatalf("copy %s: %v", ext, err) + } + } + quarantined, err := sibling.LoadEcShard("", vid, 1) + if err != nil { + t.Fatalf("load sibling shard: %v", err) + } + quarantined.MarkIoQuarantined() + _, picked, err := store.LoadEcShardAcrossDisks("", vid, 1) + if err != nil { + t.Fatalf("pick a disk for the shard: %v", err) + } + if picked == 0 { + store.Locations[0].UnloadEcShard(vid, 1) + } + + err = store.MountEcShards("", vid, 1, "") + if (err == nil) != (picked == 0) { + t.Errorf("mount returned %v with disk %d picked", err, picked) + } + if _, healthyHolds := store.Locations[0].FindEcShard(vid, 1); (err == nil) != healthyHolds { + t.Errorf("mount returned %v, healthy disk holds the shard: %v", err, healthyHolds) + } +} + +// The last-resort read does not depend on the master: once the quarantined +// shards are dropped from it, the location lookup fails, and the read must +// still reach the local copy. +func TestReadEcShardNeedleFallsBackToQuarantinedShardsWithoutLookup(t *testing.T) { + store := newTestStore(t, 1) + store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + const vid = needle.VolumeId(7) + _, ecVolume, n := mountEcShardRange(t, store, vid, 0, erasure_coding.TotalShardsCount) + + ecVolume.MarkIoQuarantined() + if err := readWrittenEcNeedle(t, store, vid, n); err != nil { + t.Fatalf("read with no master and only the quarantined disk holding the shards: %v", err) + } +} + +// Reconstruction falls back to quarantined local shards when the healthy +// sources cannot supply enough of them. +func TestReadEcShardNeedleReconstructsFromQuarantinedShards(t *testing.T) { + store := newTestStore(t, 1) + store.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + const vid = needle.VolumeId(7) + _, ecVolume, n := mountEcShardRange(t, store, vid, erasure_coding.ParityShardsCount, erasure_coding.TotalShardsCount) + seedShardLocations(ecVolume, startEcShardPeer(t, &volume_server_pb.UnimplementedVolumeServerServer{})) + + ecVolume.MarkIoQuarantined() + if err := readWrittenEcNeedle(t, store, vid, n); err != nil { + t.Fatalf("read with the needle's shards gone and only quarantined shards to rebuild from: %v", err) + } +} diff --git a/weed/storage/store_ec_recover_fanout_test.go b/weed/storage/store_ec_recover_fanout_test.go index 0bd75286e5..848c6cd2db 100644 --- a/weed/storage/store_ec_recover_fanout_test.go +++ b/weed/storage/store_ec_recover_fanout_test.go @@ -182,7 +182,7 @@ func TestRecoverOneRemoteEcShardIntervalUsesLocalShards(t *testing.T) { } want := make([]byte, len(got)) - if err := store.readLocalEcShardInterval(ecVolume, shardIdToRecover, want, actualOffset); err != nil { + if err := store.readLocalEcShardInterval(ecVolume, shardIdToRecover, want, actualOffset, false); err != nil { t.Fatalf("read local ec shard %d: %v", shardIdToRecover, err) } if !bytes.Equal(got, want) { diff --git a/weed/storage/store_ec_scrub.go b/weed/storage/store_ec_scrub.go index 25f2f88427..293827982c 100644 --- a/weed/storage/store_ec_scrub.go +++ b/weed/storage/store_ec_scrub.go @@ -100,7 +100,7 @@ func (s *Store) scrubEcVolumeWalk(ecv *erasure_coding.EcVolume, mode volume_serv shardId, offset := ecv.IntervalToShardIdAndOffset(iv) // try a local shard read first... - if err := s.readLocalEcShardInterval(ecv, shardId, chunk, offset); err == nil { + if err := s.readLocalEcShardInterval(ecv, shardId, chunk, offset, true); err == nil { data = append(data, chunk...) continue }