diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 895a7274c1..3729c9cff8 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -451,6 +451,7 @@ message VolumeEcShardsGenerateResponse { message VolumeEcShardsRebuildRequest { uint32 volume_id = 1; string collection = 2; + bool unsafe_ignore_sidecar = 3; // bypass the bitrot-sidecar fail-closed guard (operator override; distinct from ec.rebuild -force) } message VolumeEcShardsRebuildResponse { repeated uint32 rebuilt_shard_ids = 1; diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index cd86ac1bf4..cc3446770b 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -16,6 +16,7 @@ use crate::pb::filer_pb; use crate::pb::master_pb; use crate::pb::volume_server_pb; use crate::pb::volume_server_pb::volume_server_server::VolumeServer; +use crate::storage::erasure_coding::ec_bitrot; use crate::storage::erasure_coding::ec_shard::{ShardId, shard_id_try_from}; use crate::storage::erasure_coding::ecj_merge::EcjIdDecoder; use crate::storage::needle::needle::{self, Needle}; @@ -61,6 +62,42 @@ fn is_not_leader_status(e: &Status) -> bool { e.message().contains("Not current leader") } +/// A gRPC error that still carries the ids of shards the rebuild wrote before +/// failing. The Go shell pulls VolumeEcShardsRebuildResponse out of the status +/// details so its deferred cleanup keeps the copies the rebuild regenerated +/// in place instead of deleting them as orphans — a plain error drops the +/// response holding them. `details` is the serialized google.rpc.Status that +/// the grpc-status-details-bin trailer carries. +fn status_internal_with_rebuild_ids(message: String, rebuilt_shard_ids: &[u32]) -> Status { + #[derive(Clone, PartialEq, prost::Message)] + struct RpcStatus { + #[prost(int32, tag = "1")] + code: i32, + #[prost(string, tag = "2")] + message: String, + #[prost(message, repeated, tag = "3")] + details: Vec, + } + use prost::Message; + let resp = volume_server_pb::VolumeEcShardsRebuildResponse { + rebuilt_shard_ids: rebuilt_shard_ids.to_vec(), + }; + let status = RpcStatus { + code: tonic::Code::Internal as i32, + message: message.clone(), + details: vec![prost_types::Any { + type_url: "type.googleapis.com/volume_server_pb.VolumeEcShardsRebuildResponse" + .to_string(), + value: resp.encode_to_vec(), + }], + }; + Status::with_details( + tonic::Code::Internal, + message, + status.encode_to_vec().into(), + ) +} + /// Deadline from the caller's `grpc-timeout` header (TimeoutValue + /// H|M|S|m|u|n unit, per the gRPC-over-HTTP2 spec). Tonic's GrpcTimeout layer /// drops an over-deadline handler future anyway; parsing it here lets a queued @@ -3197,63 +3234,215 @@ impl VolumeServer for VolumeGrpcService { })?; let total_shards = data_shards + parity_shards; - // Check which shards are missing (check rebuild dir and all other dirs) - let mut missing: Vec = Vec::new(); - for shard_id in 0..total_shards as u8 { - let shard = crate::storage::erasure_coding::ec_shard::EcVolumeShard::new( - &rebuild_dir, - collection, - vid, - shard_id, - ); - let mut found = std::path::Path::new(&shard.file_name()).exists(); - if !found { - for other_dir in &other_dirs { - let other_shard = crate::storage::erasure_coding::ec_shard::EcVolumeShard::new( - other_dir, collection, vid, shard_id, - ); - if std::path::Path::new(&other_shard.file_name()).exists() { - found = true; - break; + // Discover shard files like Go's findShardFile: the rebuild's own data + // and index directories first, then every sibling directory. Track + // each present shard's path — a shard the sidecar reclassifies as + // corrupt is replaced at ITS discovered path, not at the rebuild dir. + let shard_file_path = |dir: &str, shard_id: ShardId| -> String { + crate::storage::erasure_coding::ec_shard::EcVolumeShard::new( + dir, collection, vid, shard_id, + ) + .file_name() + }; + let mut shard_dirs: Vec = vec![rebuild_dir.clone()]; + if !rebuild_idx_dir.is_empty() && rebuild_idx_dir != rebuild_dir { + shard_dirs.push(rebuild_idx_dir.clone()); + } + shard_dirs.extend(other_dirs.iter().cloned()); + + let mut found: Vec> = vec![None; total_shards as usize]; + for shard_id in 0..total_shards as ShardId { + found[shard_id as usize] = shard_dirs + .iter() + .map(|d| shard_file_path(d, shard_id)) + .find(|p| std::path::Path::new(p).exists()); + } + + // Load the generation-0 checksum sidecar the way Go's + // loadRebuildSidecar does: absent, wrong-generation, or recording no + // geometry => Off; a self-integrity, geometry, or manifest failure => + // Invalid; otherwise On. + let data_base = crate::storage::volume::volume_file_name(&rebuild_dir, collection, vid); + let idx_base = crate::storage::volume::volume_file_name(&rebuild_idx_dir, collection, vid); + let sidecar_dir_refs: Vec<&str> = other_dirs.iter().map(|s| s.as_str()).collect(); + let sidecar_path = + ec_bitrot::find_bitrot_sidecar(0, &data_base, &idx_base, &sidecar_dir_refs); + let (prot, bitrot_status) = if sidecar_path.is_empty() { + (None, ec_bitrot::BitrotStatus::Off) + } else { + match ec_bitrot::load_bitrot_sidecar(&sidecar_path) { + Err(e) => { + tracing::warn!(path = %sidecar_path, error = %e, "bitrot: sidecar self-integrity failed"); + (None, ec_bitrot::BitrotStatus::Invalid) + } + Ok(prot) => { + if prot.generation != 0 || prot.ec_shard_config.is_none() { + (None, ec_bitrot::BitrotStatus::Off) + } else if !ec_bitrot::geometry_matches( + &prot, + data_shards as usize, + parity_shards as usize, + block_size, + ) { + tracing::warn!(path = %sidecar_path, "bitrot: sidecar records a different layout than the rebuild uses"); + (None, ec_bitrot::BitrotStatus::Invalid) + } else if let Err(e) = ec_bitrot::validate_manifest( + &prot, + data_shards as usize, + parity_shards as usize, + ) { + tracing::warn!(path = %sidecar_path, error = %e, "bitrot: sidecar manifest invalid"); + (None, ec_bitrot::BitrotStatus::Invalid) + } else { + (Some(prot), ec_bitrot::BitrotStatus::On) } } } - if !found { - missing.push(shard_id as u32); + }; + let unsafe_ignore_sidecar = req.unsafe_ignore_sidecar; + + // Present/missing/corrupt classification — mirrors pass 1 of Go's + // generateMissingEcFiles plus its bitrot verify-and-exclude. + let mut missing: Vec = Vec::new(); + let mut corrupt_owned: Vec<(u32, String)> = Vec::new(); + let mut present = vec![false; total_shards as usize]; + let mut present_count = 0usize; + for shard_id in 0..total_shards as usize { + match &found[shard_id] { + None => missing.push(shard_id as u32), + Some(p) => { + let size = std::fs::metadata(p).map(|m| m.len()).unwrap_or(0); + if size == 0 { + // Zero-sized residue is a failed operation's leftover, + // not a shard: exclude it from inputs and regenerate + // over it in place, like a reclassified-corrupt shard. + tracing::warn!(shard = shard_id, path = %p, "zero-sized shard; excluding from rebuild inputs and regenerating"); + corrupt_owned.push((shard_id as u32, p.clone())); + } else { + present[shard_id] = true; + present_count += 1; + } + } } } - if missing.is_empty() { - return Ok(Response::new( - volume_server_pb::VolumeEcShardsRebuildResponse { - rebuilt_shard_ids: vec![], - }, - )); + match bitrot_status { + ec_bitrot::BitrotStatus::Invalid => { + if !unsafe_ignore_sidecar { + return Err(Status::internal(format!( + "bitrot sidecar for {} is malformed/unverifiable; refusing to rebuild (pass unsafe_ignore_sidecar to override)", + base_name + ))); + } + tracing::warn!(base = %base_name, "bitrot sidecar is malformed/unverifiable; proceeding because unsafe_ignore_sidecar is set"); + } + ec_bitrot::BitrotStatus::On if !unsafe_ignore_sidecar => { + let prot = prot.as_ref().unwrap(); + let mut corrupt: Vec = Vec::new(); + for shard_id in 0..total_shards as usize { + if !present[shard_id] { + continue; + } + let Some(entry) = ec_bitrot::shard_checksums(prot, shard_id as u32) else { + continue; + }; + let p = found[shard_id].as_ref().unwrap(); + match ec_bitrot::verify_shard_file_blocks(p, entry, prot.block_size as i64) { + Err(e) => { + // A read error means we cannot trust this shard as + // a Reed-Solomon input: exclude it rather than + // silently feeding possibly-corrupt bytes into + // reconstruction. + tracing::warn!(shard = shard_id, error = %e, "bitrot: failed to verify present shard; excluding it"); + corrupt.push(shard_id as u32); + } + Ok(mismatched) if !mismatched.is_empty() => { + corrupt.push(shard_id as u32); + } + _ => {} + } + } + if !corrupt.is_empty() { + // Wholesale-mismatch guard (RS-arbiter conservative form): + // localized bitrot touches a few shards; a stale/wrong + // sidecar mismatches more than parity_shards. + if corrupt.len() > parity_shards as usize { + return Err(Status::internal(format!( + "bitrot sidecar suspect for {}: {}/{} present shards mismatch (> parity {}); refusing to rebuild (pass unsafe_ignore_sidecar to override)", + base_name, + corrupt.len(), + present_count, + parity_shards + ))); + } + if present_count - corrupt.len() < data_shards as usize { + return Err(Status::internal(format!( + "bitrot: only {} verified-good shards for {}, need {} data shards; sidecar may be stale (pass unsafe_ignore_sidecar to override)", + present_count - corrupt.len(), + base_name, + data_shards + ))); + } + for shard_id in corrupt { + tracing::warn!( + shard = shard_id, + "bitrot: present shard fails checksum; excluding from rebuild inputs and regenerating" + ); + let p = found[shard_id as usize].take().unwrap(); + present[shard_id as usize] = false; + present_count -= 1; + corrupt_owned.push((shard_id, p)); + } + } + } + _ => {} } - // Rebuild missing shards, searching all locations for input shards. - // Pass other_dirs so shards on sibling disks are found even when the - // primary rebuild dir doesn't hold them. This one takes a single flat - // list — the shape Go's RebuildEcFiles uses — so unlike the resolvers - // above it cannot be handed the rebuild's own index directory - // separately, and a split -dir/-dir.idx location keeps its .ecx and - // .vif there. Go's additionalDirs carries that directory for the same - // reason. + // Everything the rebuild (re)writes: genuinely missing plus + // reclassified-corrupt. The response — and any failure after this + // point — must report the whole set, since the caller keeps copies + // the rebuild regenerated in place. + let mut rebuilt: Vec = missing.clone(); + rebuilt.extend(corrupt_owned.iter().map(|(id, _)| *id)); + rebuilt.sort_unstable(); + + if !rebuilt.is_empty() && present_count < data_shards as usize { + return Err(Status::internal(format!( + "not enough shards to rebuild {}: found {} shards, need at least {} (data shards), missing shards: {:?}", + base_name, present_count, data_shards, rebuilt + ))); + } + + // Rebuild missing and corrupt shards, searching all locations for + // input shards. This takes a single flat list — the shape Go's + // RebuildEcFiles uses — so unlike the resolvers above it cannot be + // handed the rebuild's own index directory separately, and a split + // -dir/-dir.idx location keeps its .ecx and .vif there. Go's + // additionalDirs carries that directory for the same reason. Nothing + // missing skips only the regeneration: a retry after a failed earlier + // attempt still owes the .ecx rebuild and the mounted-handle repoint + // below. let mut rebuild_search_dirs: Vec = other_dirs.clone(); if !rebuild_idx_dir.is_empty() && rebuild_idx_dir != rebuild_dir { rebuild_search_dirs.push(rebuild_idx_dir.clone()); } let other_dir_refs: Vec<&str> = rebuild_search_dirs.iter().map(|s| s.as_str()).collect(); - crate::storage::erasure_coding::ec_encoder::rebuild_ec_files( - &rebuild_dir, - collection, - vid, - &missing, - data_shards as usize, - parity_shards as usize, - &other_dir_refs, - ) - .map_err(|e| Status::internal(format!("RebuildEcFiles: {}", e)))?; + if !rebuilt.is_empty() { + crate::storage::erasure_coding::ec_encoder::rebuild_ec_files( + &rebuild_dir, + collection, + vid, + &missing, + &corrupt_owned, + data_shards as usize, + parity_shards as usize, + &other_dir_refs, + prot.as_ref().filter(|_| { + bitrot_status == ec_bitrot::BitrotStatus::On && !unsafe_ignore_sidecar + }), + ) + .map_err(|e| Status::internal(format!("RebuildEcFiles: {}", e)))?; + } // Rebuild .ecx; use idx_directory with fallback to data directory let ecx_base = format!("{}/{}", rebuild_idx_dir, base_name); @@ -3276,20 +3465,113 @@ impl VolumeServer for VolumeGrpcService { } ecx_dir_refs.extend(other_dir_refs.iter().copied()); - crate::storage::erasure_coding::ec_encoder::rebuild_ecx_file( - &ecx_rebuild_dir, - collection, - vid, - data_shards as usize, - block_size, - dat_file_size, - &ecx_dir_refs, - ) - .map_err(|e| Status::internal(format!("RebuildEcxFile: {}", e)))?; + // A mounted volume folds its .ecj into the live .ecx in place — + // the same path Go takes under LockDeletionJournal — so readers + // searching the open index never watch it get truncated. Only an + // unmounted index takes the full regen; nothing holds it open to + // interrupt. Either failure still owes the caller the regenerated + // ids — the shards are already written and its deferred cleanup + // would delete the copies as strays without them. + let mut folded = false; + { + let mut store = self.state.store.write().unwrap_or_else(|e| e.into_inner()); + let loc_idx = store + .locations + .iter() + .position(|loc| loc.directory == rebuild_dir); + if let Some(loc_idx) = loc_idx { + if let Some(ev) = store.locations[loc_idx].find_ec_volume_mut(vid) { + let journal_path = ev.ecj_file_name(); + ev.rebuild_ecx_from_journal().map_err(|e| { + status_internal_with_rebuild_ids( + format!("RebuildEcxFile: {}", e), + &rebuilt, + ) + })?; + folded = true; + // The fold replaced a journal other disks may share: + // repoint their handles before dropping the store lock, + // or their next delete lands on the detached inode. + for (i, loc) in store.locations.iter_mut().enumerate() { + if i == loc_idx { + continue; + } + if let Some(other) = loc.find_ec_volume_mut(vid) { + if std::path::Path::new(&other.ecj_file_name()) + == std::path::Path::new(&journal_path) + { + other.adopt_folded_journal().map_err(|e| { + status_internal_with_rebuild_ids( + format!("RebuildEcxFile: {}", e), + &rebuilt, + ) + })?; + } + } + } + } + } + } + if !folded { + crate::storage::erasure_coding::ec_encoder::rebuild_ecx_file( + &ecx_rebuild_dir, + collection, + vid, + data_shards as usize, + block_size, + dat_file_size, + &ecx_dir_refs, + ) + .map_err(|e| { + status_internal_with_rebuild_ids(format!("RebuildEcxFile: {}", e), &rebuilt) + })?; + } + + // Repoint every mounted handle of this volume at the file now on + // disk, not just this run's outputs — mirrors Go's repoint loop in + // VolumeEcShardsRebuild: a shard an earlier rebuild wrote on another + // disk is "present" this run and never re-enters `missing`, yet its + // mounted handle still serves the detached inode. + { + let mut store = self.state.store.write().unwrap_or_else(|e| e.into_inner()); + for loc_idx in 0..store.locations.len() { + let mounted: Vec = match store.locations[loc_idx].find_ec_volume(vid) { + Some(ev) => ev + .shards + .iter() + .enumerate() + .filter_map(|(i, s)| s.as_ref().map(|_| i as ShardId)) + .collect(), + None => continue, + }; + for shard_id in mounted { + if store.locations[loc_idx] + .reopen_ec_shard(vid, shard_id) + .is_ok() + { + continue; + } + // The mounted disk lost the file and the rebuild wrote it + // elsewhere — or it is unreadable in place. Drop the stale + // handle and remount from whichever local disk holds the + // rebuilt file; a shard that cannot be remounted stays + // unmounted rather than serving replaced bytes. + store.locations[loc_idx].unmount_ec_shards(vid, &[shard_id]); + if let Err(e) = store.mount_ec_shard(vid, collection, shard_id, "") { + tracing::warn!( + "rebuilt ec shard {}.{} left unmounted: {}", + vid.0, + shard_id, + e + ); + } + } + } + } Ok(Response::new( volume_server_pb::VolumeEcShardsRebuildResponse { - rebuilt_shard_ids: missing, + rebuilt_shard_ids: rebuilt, }, )) } diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index ab83c246fd..f0484e31e0 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -995,6 +995,32 @@ impl DiskLocation { Ok(()) } + /// Swap the mounted shard's descriptor for a fresh open of its on-disk + /// file, keeping the registration — and therefore the master's view — + /// untouched. Mirrors Go's `DiskLocation.ReopenEcShard`: the rebuild + /// recreates a shard file under a handle that still serves the detached + /// inode, and reopening repoints reads at the repaired file without an + /// unmount/remount heartbeat flip-flop. Readers holding a cloned fd keep + /// the old inode until they finish — the same contract eviction relies on. + pub fn reopen_ec_shard(&mut self, vid: VolumeId, shard_id: ShardId) -> Result { + let dir = self.directory.clone(); + let Some(ec_vol) = self.ec_volumes.get_mut(&vid) else { + return Ok(false); + }; + let Some(existing) = ec_vol + .shards + .get_mut(shard_id as usize) + .and_then(|s| s.as_mut()) + else { + return Ok(false); + }; + let mut new_shard = EcVolumeShard::new(&dir, &existing.collection, vid, shard_id); + new_shard.disk_type = existing.disk_type.clone(); + new_shard.open().map_err(VolumeError::Io)?; + *existing = new_shard; + Ok(true) + } + /// Unmount EC shards for a volume on this location. /// /// Only shards that are actually mounted decrement the per-shard @@ -2359,4 +2385,53 @@ mod tests { ); } } + + /// Mirrors Go's TestReopenEcShardReadsReplacedFile: a shard file the + /// rebuild replaced under its mount keeps serving the old inode through + /// the open descriptor; reopen_ec_shard must repoint the mounted shard at + /// the new file without dropping its registration. + #[test] + fn test_reopen_ec_shard_reads_replaced_file() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut loc = DiskLocation::new( + dir, + dir, + 10, + DiskType::HardDrive, + MinFreeSpace::Percent(0.0), + Vec::new(), + ) + .unwrap(); + + std::fs::write(format!("{}/pics_5.ecx", dir), [0u8; 16]).unwrap(); + let shard_path = format!("{}/pics_5.ec00", dir); + std::fs::write(&shard_path, b"old bytes").unwrap(); + loc.mount_ec_shards(VolumeId(5), "pics", &[0], "").unwrap(); + + let read_shard = |loc: &DiskLocation| -> Vec { + let ec_vol = loc.find_ec_volume(VolumeId(5)).expect("ec volume mounted"); + let shard = ec_vol.shards[0].as_ref().expect("shard 0 mounted"); + let mut buf = vec![0u8; shard.file_size() as usize]; + let n = shard.read_at(&mut buf, 0).unwrap(); + buf.truncate(n); + buf + }; + assert_eq!(read_shard(&loc), b"old bytes"); + + // The rebuild writes a fresh file and renames it over the original: + // the mounted descriptor is now on the detached inode. + let rebuilt = format!("{}.rebuilding", shard_path); + std::fs::write(&rebuilt, b"new bytes").unwrap(); + std::fs::rename(&rebuilt, &shard_path).unwrap(); + assert_eq!(read_shard(&loc), b"old bytes", "mounted shard still reads the replaced inode"); + + assert!(loc.reopen_ec_shard(VolumeId(5), 0).unwrap()); + assert_eq!(read_shard(&loc), b"new bytes", "reopened shard should read the regenerated file"); + + assert!( + !loc.reopen_ec_shard(VolumeId(5), 7).unwrap(), + "a shard that is not mounted should report false" + ); + } } diff --git a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs index d6afaa5e7e..7c7d44549f 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_bitrot.rs @@ -147,6 +147,40 @@ pub fn bitrot_sidecar_path(base: &str, generation: u32) -> String { } } +/// Resolves the sidecar path for a generation, searching the data base, the +/// index base, and any additional directories — mirroring Go's +/// `findBitrotSidecar`, which handles split data/idx layouts and per-disk +/// mirrors the same way shard/.vif lookups do. Returns "" when no candidate +/// exists. +pub fn find_bitrot_sidecar( + generation: u32, + data_base: &str, + index_base: &str, + additional_dirs: &[&str], +) -> String { + let mut candidates = vec![ + bitrot_sidecar_path(data_base, generation), + bitrot_sidecar_path(index_base, generation), + ]; + let base = std::path::Path::new(data_base) + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(); + for dir in additional_dirs { + if dir.is_empty() || base.is_empty() { + continue; + } + candidates.push(bitrot_sidecar_path( + &format!("{}/{}", dir, base), + generation, + )); + } + candidates + .into_iter() + .find(|c| !c.is_empty() && std::path::Path::new(c).exists()) + .unwrap_or_default() +} + /// Remove the bitrot checksum sidecars for a base file name: the legacy /// `.ecsum` (generation 0) and any versioned `.ecsum.v`. /// diff --git a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs index f99f0855d7..834169b4b5 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_encoder.rs @@ -5,6 +5,7 @@ use std::fs::File; use std::io; +use std::io::Write; use reed_solomon_erasure::galois_8::ReedSolomon; @@ -146,16 +147,33 @@ pub fn uniform_block_size(dat_file_size: i64, data_shards: usize) -> i64 { /// `additional_dirs` lists sibling disk locations to search for existing shards when /// they are not found in `dir`. This is required on multi-disk volume servers where /// shards for the same volume may be spread across disks. +/// Rebuild EC shard files from the surviving shards. +/// +/// `missing_shard_ids` are shards absent everywhere: each is written at +/// `dir/.ecNN`. `corrupt_shard_paths` are present shards the caller +/// reclassified as corrupt (a bitrot sidecar mismatch or zero-size residue): +/// each is written to `.rebuilding` and atomically renamed +/// over the corrupt original only after every output verifies, so a failed +/// rebuild leaves the untouched original rather than a half-written file — +/// mirrors Go's `generateMissingEcFiles`. When `protection` is the volume's +/// generation-0 sidecar, every regenerated shard is verified against it +/// before publish: a mismatch means the sidecar is stale, and the rebuild +/// fails closed by removing its outputs instead of blessing wrong bytes. +/// +/// Shards absent from `dir` are read from `additional_dirs` (sibling disks on +/// the same server), mirroring Go's findShardFile. pub fn rebuild_ec_files( dir: &str, collection: &str, volume_id: VolumeId, missing_shard_ids: &[u32], + corrupt_shard_paths: &[(u32, String)], data_shards: usize, parity_shards: usize, additional_dirs: &[&str], + protection: Option<&EcBitrotProtection>, ) -> io::Result<()> { - if missing_shard_ids.is_empty() { + if missing_shard_ids.is_empty() && corrupt_shard_paths.is_empty() { return Ok(()); } @@ -163,6 +181,21 @@ pub fn rebuild_ec_files( .map_err(|e| io::Error::other(format!("reed-solomon init: {:?}", e)))?; let total_shards = data_shards + parity_shards; + + // The rebuild set: genuinely-missing shards plus reclassified-corrupt + // ones. Everything else is a Reed-Solomon input. + let mut is_rebuild = vec![false; total_shards]; + for &i in missing_shard_ids { + if (i as usize) < total_shards { + is_rebuild[i as usize] = true; + } + } + for &(i, _) in corrupt_shard_paths { + if (i as usize) < total_shards { + is_rebuild[i as usize] = true; + } + } + let mut shards: Vec = (0..total_shards as u8) .map(|i| EcVolumeShard::new(dir, collection, volume_id, i)) .collect(); @@ -172,7 +205,7 @@ pub fn rebuild_ec_files( // on the same volume server) before failing — mirrors Go's findShardFile logic. let mut shard_size = 0; for (i, shard) in shards.iter_mut().enumerate() { - if !missing_shard_ids.contains(&(i as u32)) { + if !is_rebuild[i] { if shard.open().is_ok() { let size = shard.file_size(); if size > shard_size { @@ -210,12 +243,72 @@ pub fn rebuild_ec_files( )); } - // Create the missing shards for writing - for i in missing_shard_ids { - if let Some(shard) = shards.get_mut(*i as usize) { - shard.create()?; + // Create output files: a genuinely-absent shard is written at its final + // path; a reclassified-corrupt shard goes to a temp file beside its + // discovered path and is renamed over the original after verification. + // (final_path, write_path, file) — write_path differs only for corrupt. + let mut outputs: Vec)>> = + (0..total_shards).map(|_| None).collect(); + let create_output = |path: &str| -> io::Result { + crate::storage::volume_open::open_volume_file( + std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(true), + path, + ) + }; + for &i in missing_shard_ids { + let idx = i as usize; + if idx >= total_shards { + continue; } + let final_path = format!( + "{}.ec{:02}", + volume_file_name(dir, collection, volume_id), + i + ); + outputs[idx] = Some(( + final_path.clone(), + final_path.clone(), + Some(create_output(&final_path)?), + )); } + for &(i, ref corrupt_path) in corrupt_shard_paths { + let idx = i as usize; + if idx >= total_shards { + continue; + } + let write_path = format!("{}.rebuilding", corrupt_path); + outputs[idx] = Some(( + corrupt_path.clone(), + write_path.clone(), + Some(create_output(&write_path)?), + )); + } + + // Removes every generated output on a failed fail-closed rebuild: temp + // replacements AND genuinely-missing shards written at their final path, + // so no unverified bytes are published. A reclassified-corrupt shard's + // untouched original (at final_path, distinct from its write_path) is left + // in place. Mirrors Go's cleanupRebuildOutputs. + let cleanup_outputs = |mut outputs: Vec)>>| { + let write_paths: Vec = outputs + .iter_mut() + .flatten() + .map(|slot| { + // Take the handle so it closes before the unlink — Windows + // refuses to remove an open file. + let _ = slot.2.take(); + slot.1.clone() + }) + .collect(); + drop(outputs); + for p in write_paths { + let _ = std::fs::remove_file(p); + } + }; let block_size = ERASURE_CODING_SMALL_BLOCK_SIZE; let mut remaining = shard_size; @@ -232,10 +325,11 @@ pub fn rebuild_ec_files( // shard; treat it as an error rather than reconstructing over a // zero-padded tail and publishing the result as restored redundancy. for (i, shard) in shards.iter().enumerate() { - if !missing_shard_ids.contains(&(i as u32)) { + if !is_rebuild[i] { let mut buf = vec![0u8; to_process]; let n = shard.read_at(&mut buf, offset)?; if n != to_process { + cleanup_outputs(std::mem::take(&mut outputs)); return Err(io::Error::new( io::ErrorKind::UnexpectedEof, format!( @@ -249,14 +343,25 @@ pub fn rebuild_ec_files( } // Reconstruct missing shards - rs.reconstruct(&mut buffers) - .map_err(|e| io::Error::other(format!("reed-solomon reconstruct: {:?}", e)))?; + if let Err(e) = rs.reconstruct(&mut buffers) { + cleanup_outputs(std::mem::take(&mut outputs)); + return Err(io::Error::other(format!( + "reed-solomon reconstruct: {:?}", + e + ))); + } - // Write recovered data into the missing shards - for i in missing_shard_ids { - let idx = *i as usize; - if let Some(buf) = buffers[idx].take() { - shards[idx].write_all(&buf)?; + // Write recovered data into the rebuilt shards + for i in 0..total_shards { + if is_rebuild[i] { + if let Some(buf) = buffers[i].take() { + if let Some(f) = outputs[i].as_mut().and_then(|slot| slot.2.as_mut()) { + if let Err(e) = f.write_all(&buf) { + cleanup_outputs(std::mem::take(&mut outputs)); + return Err(e); + } + } + } } } @@ -264,7 +369,80 @@ pub fn rebuild_ec_files( remaining -= to_process as i64; } - // Close all shards + // Verify regenerated shards against the sidecar. Reed-Solomon is + // deterministic, so a regenerated shard that does NOT match the sidecar + // means the sidecar is wrong/stale (not the shard) — fail closed rather + // than publishing bytes we cannot trust. Mirrors Go's post-rebuild + // verification pass. + if let Some(prot) = protection { + for slot in outputs.iter_mut().flatten() { + if let Some(f) = slot.2.as_mut() { + if let Err(e) = f.sync_all() { + cleanup_outputs(std::mem::take(&mut outputs)); + return Err(e); + } + } + } + for i in 0..total_shards { + let Some((_, ref write_path, _)) = outputs[i] else { + continue; + }; + let Some(entry) = ec_bitrot::shard_checksums(prot, i as u32) else { + continue; + }; + match ec_bitrot::verify_shard_file_blocks(write_path, entry, prot.block_size as i64) { + Err(e) => { + cleanup_outputs(std::mem::take(&mut outputs)); + return Err(io::Error::new( + e.kind(), + format!("bitrot: verify regenerated shard {}: {}", i, e), + )); + } + Ok(mismatched) if !mismatched.is_empty() => { + cleanup_outputs(std::mem::take(&mut outputs)); + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "bitrot: regenerated shard {} does not match sidecar ({} blocks differ); sidecar likely stale — aborting", + i, + mismatched.len() + ), + )); + } + _ => {} + } + } + } + + // Flush and close every output, then atomically move reclassified-corrupt + // rebuilds over their originals — the rename lands only after all writes + // and the sidecar verification above have succeeded. + for slot in outputs.iter_mut().flatten() { + if let Some(f) = slot.2.as_mut() { + if let Err(e) = f.sync_all() { + cleanup_outputs(std::mem::take(&mut outputs)); + return Err(e); + } + } + } + drop(outputs); + for &(i, ref corrupt_path) in corrupt_shard_paths { + if (i as usize) >= total_shards { + continue; + } + let write_path = format!("{}.rebuilding", corrupt_path); + if let Err(e) = std::fs::rename(&write_path, corrupt_path) { + return Err(io::Error::new( + e.kind(), + format!( + "bitrot: replace corrupt shard {} ({} -> {}): {}", + i, write_path, corrupt_path, e + ), + )); + } + } + + // Close input shards for shard in &mut shards { shard.close(); } @@ -1016,7 +1194,7 @@ mod tests { // Rebuild a different (genuinely missing) shard. std::fs::remove_file(format!("{}/1.ec07", dir)).unwrap(); - let res = rebuild_ec_files(&dir, "", VolumeId(1), &[7], 10, 4, &[]); + let res = rebuild_ec_files(&dir, "", VolumeId(1), &[7], &[], 10, 4, &[], None); assert!(res.is_err(), "truncated input shard must abort the rebuild"); } @@ -1029,7 +1207,7 @@ mod tests { let dropped = format!("{}/1.ec07", dir); std::fs::remove_file(&dropped).unwrap(); - rebuild_ec_files(&dir, "", VolumeId(1), &[7], 10, 4, &[]).unwrap(); + rebuild_ec_files(&dir, "", VolumeId(1), &[7], &[], 10, 4, &[], None).unwrap(); assert!( std::path::Path::new(&dropped).exists(), "rebuilt shard .ec07 should exist" @@ -1072,7 +1250,7 @@ mod tests { // Without additional_dirs the rebuild must fail: shards 0, 4, 8 are // not in primary and shard 7 cannot be reconstructed without them. - let res = rebuild_ec_files(&primary, "", VolumeId(1), &[7], 10, 4, &[]); + let res = rebuild_ec_files(&primary, "", VolumeId(1), &[7], &[], 10, 4, &[], None); assert!( res.is_err(), "rebuild without additional_dirs must fail when input shards are on another disk" @@ -1088,9 +1266,11 @@ mod tests { "", VolumeId(1), &[7], + &[], 10, 4, &[secondary.as_str()], + None, ) .unwrap(); diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 338d5d935c..da9598f6c0 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -1939,8 +1939,7 @@ impl EcVolume { /// `deleted_needles` instead). The rebuild is atomic with respect to /// the journal: if any individual write fails the .ecj file is left /// in place and the error is propagated so tombstones are not lost. - #[expect(dead_code, reason = "no caller yet; see the doc comment")] - fn rebuild_ecx_from_journal(&mut self) -> io::Result<()> { + pub(crate) fn rebuild_ecx_from_journal(&mut self) -> io::Result<()> { let ecj_path = self.ecj_file_name(); if !std::path::Path::new(&ecj_path).exists() { return Ok(()); @@ -1973,8 +1972,11 @@ impl EcVolume { // Fold successful — drop and recreate the journal, clear the // in-memory deleted set (all of its contents are now materialized - // in .ecx), and reset the cached size. + // in .ecx), and reset the cached size. The old handle goes first: + // if the recreate fails, deletes must error rather than keep + // appending to the removed inode. fs::remove_file(&ecj_path)?; + self.ecj_file = None; self.ecj_file = Some(open_ecj_append(&ecj_path)?); self.ecj_file_size = 0; if let Ok(mut set) = self.deleted_needles.write() { @@ -1984,6 +1986,26 @@ impl EcVolume { Ok(()) } + /// Adopt a journal another runtime sharing this `.ecj` path folded under + /// the store lock: the file was replaced, so swap the detached handle for + /// a fresh one, reset the cached size, and drop the in-memory set — every + /// id it held is now an `.ecx` tombstone. Without this, later appends + /// land on the removed inode and the deletes vanish at restart. + pub(crate) fn adopt_folded_journal(&mut self) -> io::Result<()> { + let ecj_path = self.ecj_file_name(); + // The old handle points at the replaced file: drop it before the + // reopen so a failure leaves deletes erroring, not appending to a + // detached inode. Mirrors Go's ReopenDeletionJournal. + self.ecj_file = None; + let reopened = open_ecj_append(&ecj_path)?; + self.ecj_file_size = reopened.metadata()?.len() as i64; + self.ecj_file = Some(reopened); + if let Ok(mut set) = self.deleted_needles.write() { + set.clear(); + } + Ok(()) + } + // ---- Deletion journal ---- /// Record a needle delete: append the id to the .ecj deletion journal diff --git a/weed/ec/ec_rebuild.go b/weed/ec/ec_rebuild.go index 4afdabace2..ddb234995a 100644 --- a/weed/ec/ec_rebuild.go +++ b/weed/ec/ec_rebuild.go @@ -14,6 +14,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/status" ) type ecRebuilder struct { @@ -265,8 +266,19 @@ func (erb *ecRebuilder) rebuildOneEcVolume(collection string, volumeId needle.Vo if !erb.applyChanges || len(copiedShardIds) == 0 { return } - if derr := SourceServerDeleteEcShards(erb.env.GrpcDialOption, collection, volumeId, pb.NewServerAddressFromDataNode(rebuilder.Info), copiedShardIds); derr != nil { - erb.write("%s delete copied ec shards %s %d.%v: %v\n", rebuilder.Info.Id, collection, volumeId, copiedShardIds, derr) + var orphaned []erasure_coding.ShardId + for _, shardId := range copiedShardIds { + // A copy the rebuild regenerated in place is mounted output, not a + // working file: deleting it strands the only good replica. + if !slices.Contains(generatedShardIds, shardId) { + orphaned = append(orphaned, shardId) + } + } + if len(orphaned) == 0 { + return + } + if derr := SourceServerDeleteEcShards(erb.env.GrpcDialOption, collection, volumeId, pb.NewServerAddressFromDataNode(rebuilder.Info), orphaned); derr != nil { + erb.write("%s delete copied ec shards %s %d.%v: %v\n", rebuilder.Info.Id, collection, volumeId, orphaned, derr) } }() if err != nil { @@ -306,6 +318,18 @@ func (erb *ecRebuilder) generateMissingShards(collection string, volumeId needle }) if rebuildErr == nil { rebuiltShardIds = erasure_coding.Uint32ToShardIds(resp.RebuiltShardIds) + return nil + } + // A rebuild that fails after regenerating shards still reports their + // ids in the status details: without them the deferred cleanup treats + // every copied shard — including ones the rebuild regenerated in + // place — as an orphan and deletes the repaired files. + if st, ok := status.FromError(rebuildErr); ok { + for _, detail := range st.Details() { + if r, ok := detail.(*volume_server_pb.VolumeEcShardsRebuildResponse); ok { + rebuiltShardIds = erasure_coding.Uint32ToShardIds(r.RebuiltShardIds) + } + } } return rebuildErr }) diff --git a/weed/server/volume_grpc_ec_rebuild_reopen_test.go b/weed/server/volume_grpc_ec_rebuild_reopen_test.go new file mode 100644 index 0000000000..be6ea95838 --- /dev/null +++ b/weed/server/volume_grpc_ec_rebuild_reopen_test.go @@ -0,0 +1,185 @@ +package weed_server + +import ( + "bytes" + "context" + "crypto/rand" + "os" + "path/filepath" + "testing" + + "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" + "github.com/stretchr/testify/require" +) + +// A corrupt shard that stays mounted while VolumeEcShardsRebuild regenerates it +// in place must stop serving the pre-repair bytes: the handler reopens the +// mounted shard so reads hit the regenerated file, not the detached inode. +// Regression test for ec.rebuild leaving a replaced shard file open until +// restart. +func TestVolumeEcShardsRebuild_ReopensMountedCorruptShard(t *testing.T) { + if !erasure_coding.BitrotProtectionEnabled { + t.Skip("bitrot protection is compiled out") + } + dataDir := filepath.Join(t.TempDir(), "data") + require.NoError(t, os.MkdirAll(dataDir, 0o755)) + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dataDir}, []int32{100}, []util.MinFreeSpace{{}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, stats.DefaultDiskIOProbeConfig()) + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.NewEcShardsChan: + case <-store.DeletedVolumesChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) + st, err := storage.NewState(t.TempDir()) + require.NoError(t, err) + store.State = st + vs := &VolumeServer{store: store} + + collection := "col" + vid := needle.VolumeId(42) + base := erasure_coding.EcShardFileName(collection, dataDir, int(vid)) + + dat := make([]byte, 7000) + _, err = rand.Read(dat) + require.NoError(t, err) + require.NoError(t, os.WriteFile(base+".dat", dat, 0o644)) + prot, err := erasure_coding.WriteEcFiles(base, nil) + require.NoError(t, err) + require.NoError(t, erasure_coding.SaveBitrotSidecar(erasure_coding.BitrotSidecarPath(base, 0), prot)) + require.NoError(t, os.WriteFile(base+".ecx", make([]byte, 16), 0o644)) + + require.NoError(t, store.MountEcShards(collection, vid, 0, "")) + readShard := func() []byte { + shard, found := store.Locations[0].FindEcShard(vid, 0) + require.True(t, found) + buf := make([]byte, shard.Size()) + _, err := shard.ReadAt(buf, 0) + require.NoError(t, err) + return buf + } + good := readShard() + + // Corrupt the file in place: the mounted descriptor now serves bad bytes. + corrupt := bytes.Repeat([]byte{0xBB}, 1024) + f, err := os.OpenFile(base+erasure_coding.ToExt(0), os.O_WRONLY, 0) + require.NoError(t, err) + _, err = f.WriteAt(corrupt, 0) + require.NoError(t, err) + require.NoError(t, f.Close()) + require.True(t, bytes.Equal(corrupt, readShard()[:len(corrupt)])) + + _, err = vs.VolumeEcShardsRebuild(context.Background(), &volume_server_pb.VolumeEcShardsRebuildRequest{ + VolumeId: uint32(vid), + Collection: collection, + }) + require.NoError(t, err) + + require.True(t, bytes.Equal(good, readShard()), + "mounted shard should serve the regenerated file after rebuild") +} + +// A shard whose file is deleted under the mount is "missing" to the rebuild: +// it regenerates on the rebuild disk while the mounted handle still serves +// the detached inode on another disk. Reopening the mounted disk cannot fix +// that, so the handler moves the registration to the disk holding the rebuilt +// file — and still reports the rebuilt ids so the caller's copy cleanup keeps +// them. +func TestVolumeEcShardsRebuild_RelocatesMountedShardAcrossDisks(t *testing.T) { + if !erasure_coding.BitrotProtectionEnabled { + t.Skip("bitrot protection is compiled out") + } + tempDir := t.TempDir() + dirA := filepath.Join(tempDir, "a") + dirB := filepath.Join(tempDir, "b") + require.NoError(t, os.MkdirAll(dirA, 0o755)) + require.NoError(t, os.MkdirAll(dirB, 0o755)) + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{dirA, dirB}, []int32{100, 100}, []util.MinFreeSpace{{}, {}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType, types.HardDriveType}, nil, 3, stats.DefaultDiskIOProbeConfig()) + done := make(chan struct{}) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.NewEcShardsChan: + case <-store.DeletedVolumesChan: + case <-store.DeletedEcShardsChan: + case <-store.StateUpdateChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) + st, err := storage.NewState(t.TempDir()) + require.NoError(t, err) + store.State = st + vs := &VolumeServer{store: store} + + collection := "col" + vid := needle.VolumeId(42) + baseA := erasure_coding.EcShardFileName(collection, dirA, int(vid)) + baseB := erasure_coding.EcShardFileName(collection, dirB, int(vid)) + shardName := erasure_coding.ToExt(0) + + dat := make([]byte, 7000) + _, err = rand.Read(dat) + require.NoError(t, err) + require.NoError(t, os.WriteFile(baseA+".dat", dat, 0o644)) + prot, err := erasure_coding.WriteEcFiles(baseA, nil) + require.NoError(t, err) + require.NoError(t, erasure_coding.SaveBitrotSidecar(erasure_coding.BitrotSidecarPath(baseA, 0), prot)) + require.NoError(t, os.WriteFile(baseA+".ecx", make([]byte, 16), 0o644)) + + original, err := os.ReadFile(baseA + shardName) + require.NoError(t, err) + + // Mount shard 0 from disk B, then delete the file under the mount: the + // handle keeps serving the detached inode while the disk reports the + // shard missing. + require.NoError(t, os.Rename(baseA+shardName, baseB+shardName)) + require.NoError(t, store.MountEcShards(collection, vid, 0, "")) + diskId, _, found := store.FindEcShard(vid, 0) + require.True(t, found) + require.Equal(t, uint32(1), diskId, "shard 0 should be mounted on disk B") + require.NoError(t, os.Remove(baseB+shardName)) + + resp, err := vs.VolumeEcShardsRebuild(context.Background(), &volume_server_pb.VolumeEcShardsRebuildRequest{ + VolumeId: uint32(vid), + Collection: collection, + }) + require.NoError(t, err) + require.Equal(t, []uint32{0}, resp.RebuiltShardIds, + "a mounted shard whose file was rebuilt on another disk must still be reported rebuilt") + + diskId, shard, found := store.FindEcShard(vid, 0) + require.True(t, found) + require.Equal(t, uint32(0), diskId, "rebuilt shard should be mounted on the disk holding its file") + buf := make([]byte, shard.Size()) + _, err = shard.ReadAt(buf, 0) + require.NoError(t, err) + require.True(t, bytes.Equal(original, buf), "relocated shard should serve the regenerated file") +} diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index 91d4bd9f43..0fb0e32e96 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -186,6 +186,19 @@ func recordEcRebuild(result string, d time.Duration) { stats.VolumeServerECRebuildCounter.WithLabelValues(result).Inc() } +// ecRebuildErrorWithIds reports a rebuild failure that happened after shards +// were regenerated on disk. The error still carries their ids so the caller's +// deferred cleanup keeps the copies it regenerated in place — to the shell an +// unreported regenerated copy is just an orphan to delete, and a plain gRPC +// error would drop the response holding them. +func ecRebuildErrorWithIds(rebuiltShardIds []uint32, err error) error { + st := status.New(codes.Internal, err.Error()) + if withIds, derr := st.WithDetails(&volume_server_pb.VolumeEcShardsRebuildResponse{RebuiltShardIds: rebuiltShardIds}); derr == nil { + st = withIds + } + return st.Err() +} + // VolumeEcShardsRebuild generates the any of the missing .ec00 ~ .ec13 files func (vs *VolumeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_server_pb.VolumeEcShardsRebuildRequest) (*volume_server_pb.VolumeEcShardsRebuildResponse, error) { if err := vs.checkGrpcAdminAuth(ctx); err != nil { @@ -280,13 +293,58 @@ func (vs *VolumeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_s err = ev.ReopenDeletionJournal() } unlockJournal() + if err == nil { + // Mounts on sibling disks can journal into the same .ecj — the + // fold unlinked it, so repoint their handles too or their next + // append lands on the detached inode. + for _, location := range vs.store.Locations { + if location == rebuildLocation { + continue + } + if sibling, ok := location.FindEcVolume(needle.VolumeId(req.VolumeId)); ok && sibling.EcIndexBaseFileName() == indexBaseFileName { + unlockSibling := sibling.LockDeletionJournal() + serr := sibling.ReopenDeletionJournal() + unlockSibling() + if serr != nil { + err = fmt.Errorf("reopen shared ec journal on %s: %v", location.Directory, serr) + break + } + } + } + } if err != nil { recordEcRebuild("failure", time.Since(start)) - return nil, fmt.Errorf("RebuildEcxFile %s: %v", indexBaseFileName, err) + return nil, ecRebuildErrorWithIds(rebuiltShardIds, fmt.Errorf("RebuildEcxFile %s: %v", indexBaseFileName, err)) } } else if err := erasure_coding.RebuildEcxFile(indexBaseFileName); err != nil { recordEcRebuild("failure", time.Since(start)) - return nil, fmt.Errorf("RebuildEcxFile %s: %v", indexBaseFileName, err) + return nil, ecRebuildErrorWithIds(rebuiltShardIds, fmt.Errorf("RebuildEcxFile %s: %v", indexBaseFileName, err)) + } + + // A regenerated shard that was mounted keeps serving its old file through + // the open descriptor — the rename left the mounted handle on the detached + // inode. Repoint every mounted handle of this volume at the file now on + // disk, not just this run's outputs: a shard an earlier rebuild wrote on + // another disk is "present" this run and would never re-enter + // generatedShardIds, yet its mounted handle is still stale. + for _, location := range vs.store.Locations { + for _, shardId := range location.MountedEcShardIds(needle.VolumeId(req.VolumeId)) { + if _, rerr := location.ReopenEcShard(needle.VolumeId(req.VolumeId), shardId); rerr == nil { + continue + } + // The mounted disk lost the file and the rebuild wrote it + // elsewhere — or the file is unreadable in place. Drop the stale + // handle and remount from whichever local disk holds the rebuilt + // file; both steps emit no heartbeat delta, so the master's + // registration never flips. A shard that cannot be remounted is + // left unmounted — the honest state — rather than serving + // replaced bytes. Reporting the rebuilt ids on this path keeps + // the caller's cleanup from deleting them as strays. + location.UnloadEcShard(needle.VolumeId(req.VolumeId), shardId) + if _, _, lerr := vs.store.LoadEcShardAcrossDisks(req.Collection, needle.VolumeId(req.VolumeId), shardId); lerr != nil { + glog.Warningf("rebuilt ec shard %d.%d left unmounted: %v", req.VolumeId, shardId, lerr) + } + } } recordEcRebuild("success", time.Since(start)) diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index 8205ede9c6..5cf6890a78 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -255,6 +255,59 @@ func (l *DiskLocation) loadEcShardWithIdxDir(collection string, vid needle.Volum return ecVolume, nil } +// ReopenEcShard swaps the mounted shard's descriptor for a fresh open of its +// on-disk file while leaving the volume's registration — and therefore the +// master's view — untouched. It exists for the in-place rebuild path: +// RebuildEcFiles renames a regenerated shard over the file a mounted handle is +// still serving, and reopening repoints reads at the repaired file without +// the delete/add heartbeat flip-flop an unmount+remount would send. +func (l *DiskLocation) ReopenEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) (bool, error) { + l.ecVolumesLock.Lock() + defer l.ecVolumesLock.Unlock() + + ecVolume, found := l.ecVolumes[vid] + if !found { + return false, nil + } + var oldShard *erasure_coding.EcVolumeShard + position := -1 + for i, s := range ecVolume.Shards { + if s.ShardId == shardId { + position = i + oldShard = s + break + } + } + if oldShard == nil { + return false, nil + } + + newShard, err := erasure_coding.NewEcVolumeShard(oldShard.DiskType, l.Directory, oldShard.Collection, vid, shardId) + if err != nil { + return true, fmt.Errorf("reopen ec shard %d.%d: %w", vid, shardId, err) + } + ecVolume.Shards[position] = newShard + + // Readers that already resolved the old shard keep their object; its Close + // turns their next ReadAt into a clean error recovered via parity, the same + // contract eviction relies on. Mount/Unmount balance out, so the shard + // gauge does not move. + oldShard.Unmount() + oldShard.Close() + return true, nil +} + +// MountedEcShardIds returns the ids of shards this disk currently serves for +// the volume, or nil when the volume is not mounted here. +func (l *DiskLocation) MountedEcShardIds(vid needle.VolumeId) []erasure_coding.ShardId { + l.ecVolumesLock.RLock() + defer l.ecVolumesLock.RUnlock() + if ecVolume, found := l.ecVolumes[vid]; found { + return ecVolume.ShardIdList() + } + return nil +} + func (l *DiskLocation) UnloadEcShard(vid needle.VolumeId, shardId erasure_coding.ShardId) bool { l.ecVolumesLock.Lock() diff --git a/weed/storage/erasure_coding/ec_volume_delete.go b/weed/storage/erasure_coding/ec_volume_delete.go index abb951d74d..3052674e9e 100644 --- a/weed/storage/erasure_coding/ec_volume_delete.go +++ b/weed/storage/erasure_coding/ec_volume_delete.go @@ -132,6 +132,7 @@ func (ev *EcVolume) ReopenDeletionJournal() error { } ecjFile, err := backend.OpenVolumeFile(ev.FileName(".ecj"), os.O_RDWR|os.O_CREATE) if err != nil { + ev.ecjFileSize = 0 return fmt.Errorf("reopen ec volume journal %s: %w", ev.FileName(".ecj"), err) } ev.ecjFile = ecjFile diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 34142e7252..2da3327e74 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -253,6 +253,41 @@ func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat { } func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, sourceDiskType string) error { + 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) + + // Apply the orchestrator-supplied source disk type so the EC + // volume reports under it instead of the location's. Empty means + // "fall back to location's disk type" (#9423). + if sourceDiskType != "" { + ecVolume.SetDiskType(types.ToDiskType(sourceDiskType)) + } + + si := erasure_coding.NewShardsInfo() + si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize()))) + s.NewEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{ + Id: uint32(vid), + Collection: collection, + EcIndexBits: uint32(si.Bitmap()), + ShardSizes: si.SizesInt64(), + DiskType: string(ecVolume.DiskType()), + ExpireAtSec: ecVolume.ExpireAtSec, + DiskId: uint32(diskId), + EncodeTsNs: ecVolume.EncodeTsNs, + } + s.UpdateMaxVolumeIdSeen(vid) + return nil +} + +// LoadEcShardAcrossDisks mounts shardId on whichever local disk holds its +// file and returns the mounted volume plus the disk it landed on. It emits +// no heartbeat delta — callers that change the master's view do that +// themselves (MountEcShards), while same-server repoints that must not flip +// the master's registration use it directly. +func (s *Store) LoadEcShardAcrossDisks(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId) (*erasure_coding.EcVolume, uint32, error) { // The .ecx index file may live on a different disk than the one // holding the .ec?? shard being mounted: ec.balance / ec.rebuild can // place the .ecx on one local disk while later distributing shards @@ -292,29 +327,7 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er } ecVolume, err := location.loadEcShardWithIdxDir(collection, vid, shardId, idxDir) if err == nil { - glog.V(0).Infof("MountEcShards %d.%d on disk ID %d", vid, shardId, diskId) - - // Apply the orchestrator-supplied source disk type so the EC - // volume reports under it instead of the location's. Empty means - // "fall back to location's disk type" (#9423). - if sourceDiskType != "" { - ecVolume.SetDiskType(types.ToDiskType(sourceDiskType)) - } - - si := erasure_coding.NewShardsInfo() - si.Set(erasure_coding.NewShardInfo(shardId, erasure_coding.ShardSize(ecVolume.ShardSize()))) - s.NewEcShardsChan <- &master_pb.VolumeEcShardInformationMessage{ - Id: uint32(vid), - Collection: collection, - EcIndexBits: uint32(si.Bitmap()), - ShardSizes: si.SizesInt64(), - DiskType: string(ecVolume.DiskType()), - ExpireAtSec: ecVolume.ExpireAtSec, - DiskId: uint32(diskId), - EncodeTsNs: ecVolume.EncodeTsNs, - } - s.UpdateMaxVolumeIdSeen(vid) - return nil + return ecVolume, uint32(diskId), nil } if errors.Is(err, os.ErrNotExist) { // Shard or index not on this disk; another disk may own it. @@ -329,9 +342,9 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er // scanned for so the operator can tell "no .ecx anywhere" apart // from "shard not on this server". if !ecxFound { - return fmt.Errorf("MountEcShards %d.%d: no .ecx index found on any local disk", vid, shardId) + return nil, 0, fmt.Errorf("MountEcShards %d.%d: no .ecx index found on any local disk", vid, shardId) } - return fmt.Errorf("MountEcShards %d.%d not found on disk", vid, shardId) + return nil, 0, fmt.Errorf("MountEcShards %d.%d not found on disk", vid, shardId) } // Some disks returned a real (non-ENOENT) error. Report them all so // the caller can see whether the failures cluster around one disk @@ -343,7 +356,7 @@ func (s *Store) MountEcShards(collection string, vid needle.VolumeId, shardId er } b = append(b, fmt.Sprintf("%s: %v", f.dir, f.err)...) } - return fmt.Errorf("MountEcShards %d.%d load failures: %s", vid, shardId, string(b)) + return nil, 0, fmt.Errorf("MountEcShards %d.%d load failures: %s", vid, shardId, string(b)) } func (s *Store) UnmountEcShards(vid needle.VolumeId, shardId erasure_coding.ShardId, reqEncodeTsNs int64) error { diff --git a/weed/storage/store_ec_rebuild_remount_test.go b/weed/storage/store_ec_rebuild_remount_test.go new file mode 100644 index 0000000000..0d4dc69427 --- /dev/null +++ b/weed/storage/store_ec_rebuild_remount_test.go @@ -0,0 +1,64 @@ +package storage + +import ( + "bytes" + "os" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/stretchr/testify/require" +) + +// A shard file the rebuild replaced under its mount keeps serving the old +// inode through the open descriptor. ReopenEcShard must repoint the mounted +// shard at the new file without dropping its registration. +func TestReopenEcShardReadsReplacedFile(t *testing.T) { + store := newEcTargetTestStore(t, 1) + collection := "col" + vid := needle.VolumeId(42) + dataDir := store.Locations[0].Directory + + writeEcShard0(t, dataDir, collection, vid) + writeEcIndex(t, dataDir, collection, vid) + + shardPath := erasure_coding.EcShardFileName(collection, dataDir, int(vid)) + erasure_coding.ToExt(0) + content := make([]byte, calculateExpectedShardSize(1024*1024, 10)) + for i := range content { + content[i] = 'A' + } + require.NoError(t, os.WriteFile(shardPath, content, 0o644)) + + require.NoError(t, store.MountEcShards(collection, vid, 0, "")) + + readShard := func() []byte { + shard, found := store.Locations[0].FindEcShard(vid, 0) + require.True(t, found) + buf := make([]byte, len(content)) + n, err := shard.ReadAt(buf, 0) + require.NoError(t, err) + require.Equal(t, len(content), n) + return buf + } + require.True(t, bytes.Equal(content, readShard())) + + for i := range content { + content[i] = 'B' + } + rebuilt := shardPath + ".rebuilding" + require.NoError(t, os.WriteFile(rebuilt, content, 0o644)) + require.NoError(t, os.Rename(rebuilt, shardPath)) + + stale := readShard() + require.False(t, bytes.Equal(content, stale), "mounted shard should still read the replaced inode") + + reopened, err := store.Locations[0].ReopenEcShard(vid, 0) + require.NoError(t, err) + require.True(t, reopened, "shard is mounted; reopen should find it") + + require.True(t, bytes.Equal(content, readShard()), "reopened shard should read the regenerated file") + + reopened, err = store.Locations[0].ReopenEcShard(vid, 7) + require.NoError(t, err) + require.False(t, reopened, "a shard that is not mounted should report false") +} diff --git a/weed/topology/data_node_ec.go b/weed/topology/data_node_ec.go index cbda6bfec3..de38fa6cb2 100644 --- a/weed/topology/data_node_ec.go +++ b/weed/topology/data_node_ec.go @@ -140,6 +140,29 @@ func (dn *DataNode) HasEcShards(volumeId needle.VolumeId) (found bool) { return } +// HasEcShard reports whether any disk on this node still holds the shard. +// Registration lookups key on the node alone, so a shard moving between its +// disks must not drop the node from the lookup while a copy remains. +func (dn *DataNode) HasEcShard(volumeId needle.VolumeId, shardId erasure_coding.ShardId) (found bool) { + dn.RLock() + defer dn.RUnlock() + for _, c := range dn.children { + disk := c.(*Disk) + disk.ecShardsLock.RLock() + for _, ev := range disk.ecShards[volumeId] { + if ev.ShardsInfo.Has(shardId) { + found = true + break + } + } + disk.ecShardsLock.RUnlock() + if found { + return true + } + } + return +} + // doUpdateEcShardsLocked rewrites disk.ecShards from actualByKey. Caller // MUST hold dn.Lock; each disk's ecShardsLock is taken internally around // the read-modify-write of its ecShards map. diff --git a/weed/topology/data_node_ec_multi_disk_test.go b/weed/topology/data_node_ec_multi_disk_test.go index cd294ca058..df8dbde19b 100644 --- a/weed/topology/data_node_ec_multi_disk_test.go +++ b/weed/topology/data_node_ec_multi_disk_test.go @@ -161,6 +161,76 @@ func TestEcShardsAfterRestartHeartbeat(t *testing.T) { } } +// TestEcShardRelocationAcrossDisksKeepsLookup covers a rebuild relocating a +// mounted shard between a node's disks: the source disk reports a delete and +// the destination an add, but lookups key on the node — registering first is +// a no-op, so the unregister must not drop a node that still holds the shard +// on another of its disks. +func TestEcShardRelocationAcrossDisksKeepsLookup(t *testing.T) { + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dc := topo.GetOrCreateDataCenter("dc1") + rack := dc.GetOrCreateRack("rack1") + dn := rack.GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 100}) + + const vid = uint32(15) + listed := func(shardId erasure_coding.ShardId) bool { + locs, ok := topo.LookupEcShards(needle.VolumeId(vid)) + if !ok { + return false + } + for _, n := range locs.Locations[shardId] { + if n.Id() == dn.Id() { + return true + } + } + return false + } + + // shard 3 sits on disk 0, then ec.rebuild relocates it to disk 1. + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + buildEcShardMessage(vid, "c", "", 0, []erasure_coding.ShardId{3}), + }, dn) + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + buildEcShardMessage(vid, "c", "", 1, []erasure_coding.ShardId{3}), + }, dn) + if !listed(3) { + t.Fatalf("shard 3 moved disk0->disk1 dropped node %s from lookup", dn.Id()) + } + + // The same move reported as an incremental delta keeps the lookup too. + topo.IncrementalSyncDataNodeEcShards( + []*master_pb.VolumeEcShardInformationMessage{buildEcShardMessage(vid, "c", "", 0, []erasure_coding.ShardId{5})}, + []*master_pb.VolumeEcShardInformationMessage{buildEcShardMessage(vid, "c", "", 1, []erasure_coding.ShardId{5})}, + dn) + if !listed(5) { + t.Fatalf("shard 5 moved incrementally dropped node %s from lookup", dn.Id()) + } + + // A shard duplicated across two disks keeps the node listed when one + // copy goes away. + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + buildEcShardMessage(vid, "c", "", 0, []erasure_coding.ShardId{7}), + buildEcShardMessage(vid, "c", "", 1, []erasure_coding.ShardId{3, 5, 7}), + }, dn) + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + buildEcShardMessage(vid, "c", "", 1, []erasure_coding.ShardId{3, 5, 7}), + }, dn) + if !listed(7) { + t.Fatalf("shard 7 still held on disk 1 dropped node %s from lookup", dn.Id()) + } + + // A shard gone from every disk does leave the lookup. + topo.SyncDataNodeEcShards([]*master_pb.VolumeEcShardInformationMessage{ + buildEcShardMessage(vid, "c", "", 1, []erasure_coding.ShardId{3, 5}), + }, dn) + if listed(7) { + t.Fatalf("shard 7 deleted on all disks still lists node %s", dn.Id()) + } + if !listed(3) || !listed(5) { + t.Fatalf("shards 3 and 5 on disk 1 lost node %s", dn.Id()) + } +} + func buildEcShardMessage(vid uint32, collection, diskType string, diskId uint32, shardIds []erasure_coding.ShardId) *master_pb.VolumeEcShardInformationMessage { si := erasure_coding.NewShardsInfo() for _, sid := range shardIds { diff --git a/weed/topology/topology_ec.go b/weed/topology/topology_ec.go index 58f302515f..61b93f656f 100644 --- a/weed/topology/topology_ec.go +++ b/weed/topology/topology_ec.go @@ -159,6 +159,9 @@ func (t *Topology) UnRegisterEcShards(ecvi *erasure_coding.EcVolumeInfo, dn *Dat return } for _, shardId := range ecvi.ShardsInfo.Ids() { + if dn.HasEcShard(ecvi.VolumeId, shardId) { + continue + } locations.DeleteShard(shardId, dn) } } diff --git a/weed/topology/topology_event_handling.go b/weed/topology/topology_event_handling.go index 1793ff0e93..825ab7c36c 100644 --- a/weed/topology/topology_event_handling.go +++ b/weed/topology/topology_event_handling.go @@ -5,7 +5,6 @@ import ( "time" "github.com/seaweedfs/seaweedfs/weed/stats" - "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/types" "google.golang.org/grpc" @@ -123,9 +122,15 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) { defer t.releaseResolvedQuarantine(v.Id) } - // unregister ec shards when volume server disconnected + // unregister ec shards when volume server disconnected. The node's own + // records clear first so the still-held check cannot skip the drop, and + // each unregister gets a copy — DeleteEcShard drains the shared + // ShardsInfo. for _, s := range dn.GetEcShards() { - t.UnRegisterEcShards(s, dn) + held := *s + held.ShardsInfo = s.ShardsInfo.Copy() + dn.DeleteEcShard(s) + t.UnRegisterEcShards(&held, dn) } negativeUsages := dn.GetDiskUsages().negative() @@ -133,7 +138,6 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) { dn.UpAdjustDiskUsageDelta(dt, du) } dn.DeltaUpdateVolumes([]storage.VolumeInfo{}, dn.GetVolumes()) - dn.DeltaUpdateEcShards([]*erasure_coding.EcVolumeInfo{}, dn.GetEcShards()) t.unregisterDataNodeAddress(dn.ServerAddress(), dn) if dn.Parent() != nil { dn.Parent().UnlinkChildNode(dn.Id())