From 2c510cc30e73d376f098da5bbd9cd7582e80bcd5 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 00:58:30 -0700 Subject: [PATCH] ec.rebuild: remount shards the rebuild replaced, keep regenerated copies (#12193) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * volume: remount ec shards the rebuild replaced under a mounted handle Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec.rebuild: keep copied shards the rebuild regenerated in place Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: reopen regenerated ec shards in place instead of remounting UnmountEcShards and MountEcShards send deltas through separate buffered channels, so a remount can reach the master as add-then-delete and drop the repaired shard from lookups until the next full EC heartbeat. ReopenEcShard swaps the mounted shard's descriptor under the volume lock instead: registration, disk type, and heartbeat state never flip, and readers holding the old shard degrade to a clean error the parity path already recovers from. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: reopen rebuilt ec shards in place, mirroring Go volume_ec_shards_rebuild recreates a shard file under a mounted handle whose descriptor still serves the detached inode (a mounted shard whose file was deleted under it). DiskLocation::reopen_ec_shard swaps the mounted descriptor for a fresh open, matching Go's ReopenEcShard — the registration and heartbeat view never flip. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * volume: fail ec rebuild when a regenerated shard cannot be reopened * rust volume: fail ec shards rebuild when a rebuilt shard cannot be reopened * volume: repoint mounted ec shards after rebuild across disks A mounted shard whose file was deleted under the mount is "missing" to the rebuild: the regenerated file lands on the rebuild location while the handle keeps serving the detached inode on another disk. Reopening the mounted disk then failed the whole RPC, so the caller never saw the rebuilt ids and its deferred cleanup deleted the regenerated copies. Repoint every mounted handle of the rebuilt volume instead of only this run's outputs: reopen in place when the rebuilt file is on the mounted disk, otherwise drop the stale handle and remount the shard from whichever local disk holds the file via a delta-free load split out of MountEcShards — a same-server move, so the master's registration never flips. A shard that can be neither reopened nor remounted is left unmounted rather than serving replaced bytes, and the repoint step no longer fails the RPC, so RebuiltShardIds always reaches the caller. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: repoint mounted ec shards after rebuild across disks Mirror the Go repoint loop: iterate every mounted shard of the rebuilt volume, reopen in place when the rebuilt file is on the mounted disk, and otherwise drop the stale handle and remount the shard from whichever local disk holds it. The repoint no longer fails the RPC, so rebuilt_shard_ids still reaches the caller and its copy cleanup keeps the regenerated shards. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: still rebuild ecx and repoint mounts when no shard is missing * master: keep a node in an ec shard's lookup while a disk still holds it A rebuild that relocates a mounted shard between a node's disks reports the move as a delete on the source disk plus an add on the destination. Registering first is a no-op because shard lookups key on the node, then the unregister dropped the node outright — and steady-state heartbeats never restore it, so readers and later rebuilds lose the repaired copy. Skip the delete while another disk on the node still holds the shard. UnregisterDataNode now clears the node's own ec records before dropping its registrations, so the retained-shard check cannot skip the teardown, and each unregister gets a ShardsInfo copy since DeleteEcShard drains the shared one. * rust volume: fold the journal on a mounted index instead of rewriting it The rebuild handler regenerated .ecx through File::create even while the volume was mounted, truncating the index under readers that search the open descriptor with its old size — a live lookup could hit EOF mid-scan. Go folds the deletion journal in place on a mounted volume; do the same through rebuild_ecx_from_journal and reserve the full regen for the unmounted path, where nothing holds the file open to interrupt. * volume, rust volume: report rebuilt shards on late failures, fix rust corrupt discovery Two review findings the repoint rework left open: - A RebuildEcxFile failure still returns a plain error after shards were regenerated, so the caller discards the response holding the rebuilt ids and the deferred cleanup deletes copies the rebuild regenerated in place. The Go handler now packs the ids into the gRPC status details and the shell pulls them back out on error; the Rust handler carries them in grpc-status-details-bin the same way. - The Rust rebuild still counted a sidecar-corrupt shard as present, so it fed bad bytes into reconstruction and never regenerated the shard. Discovery now loads the generation-0 .ecsum the way Go does, verifies every present shard, and reclassifies mismatches (and zero-size residue) as rebuild targets written to a .rebuilding temp file, verified, and atomically renamed over the original only after every output passes the sidecar check. Honors unsafe_ignore_sidecar, added to VolumeEcShardsRebuildRequest to match the Go proto. * volume: repoint sibling journals sharing a folded .ecj after rebuild RebuildEcxFile unlinks the deletion journal and the mounted path only reopens its own handle — a sibling mount journaling into the same file keeps appending to the detached inode, and those deletes never reach a path-based reader again. Reopen every mounted volume sharing the index base before leaving the rebuild. * rust volume: repoint sibling journals sharing a folded .ecj after rebuild rebuild_ecx_from_journal replaces the journal and reopens only its own handle; mounts on sibling disks sharing that .ecj keep the detached handle, so their next appends are lost at restart. Adopt the folded journal on every mounted runtime sharing the path before releasing the store lock, clearing their in-memory set — every journaled id is now an .ecx tombstone. * ec: swap the deletion journal handle only after the new file opens A failed reopen left ecjFile nil, so every later delete returned "ec volume closed". Open first, then close and swap — mirroring the Rust adopt_folded_journal ordering. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec.rebuild: fail the rebuild when a sibling journal cannot be repointed A warning left the sibling mount journaling to a detached inode while the RPC reported success. Propagate the error like the Rust handler's adopt_folded_journal map_err. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec: keep deletes loud when the journal reopen cannot open the replacement Reverts the swap-on-success ordering: a detached handle kept on failure lets deletes append and sync to the removed inode, then vanish at restart. Closing first also frees the descriptor before the open. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * rust volume: drop the folded journal handle before reopening it adopt_folded_journal and the post-fold recreate now clear ecj_file before opening the replacement, so a failed reopen leaves deletes erroring instead of appending to the detached inode. Mirrors Go's ReopenDeletionJournal. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec: drop the cached journal size when its reopen fails Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- seaweed-volume/proto/volume_server.proto | 1 + seaweed-volume/src/server/grpc_server.rs | 392 +++++++++++++++--- seaweed-volume/src/storage/disk_location.rs | 75 ++++ .../src/storage/erasure_coding/ec_bitrot.rs | 34 ++ .../src/storage/erasure_coding/ec_encoder.rs | 216 +++++++++- .../src/storage/erasure_coding/ec_volume.rs | 28 +- weed/ec/ec_rebuild.go | 28 +- .../volume_grpc_ec_rebuild_reopen_test.go | 185 +++++++++ weed/server/volume_grpc_erasure_coding.go | 62 ++- weed/storage/disk_location_ec.go | 53 +++ .../erasure_coding/ec_volume_delete.go | 1 + weed/storage/store_ec.go | 65 +-- weed/storage/store_ec_rebuild_remount_test.go | 64 +++ weed/topology/data_node_ec.go | 23 + weed/topology/data_node_ec_multi_disk_test.go | 70 ++++ weed/topology/topology_ec.go | 3 + weed/topology/topology_event_handling.go | 12 +- 17 files changed, 1202 insertions(+), 110 deletions(-) create mode 100644 weed/server/volume_grpc_ec_rebuild_reopen_test.go create mode 100644 weed/storage/store_ec_rebuild_remount_test.go 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())