diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs index 81dbe2809..d3c4a0179 100644 --- a/seaweed-volume/src/storage/disk_location.rs +++ b/seaweed-volume/src/storage/disk_location.rs @@ -242,27 +242,34 @@ impl DiskLocation { Ok(()) } - /// Validate EC volume shards: all shards must be same size, and if .dat exists, - /// need at least DATA_SHARDS_COUNT shards with size matching expected. + /// Reports whether the EC files for (collection, vid) on this disk may be + /// deleted to reclaim the local .dat. Returns false (delete) only when that + /// provably loses no data; every ambiguity returns true (keep), since the + /// shards may be the only copy. Mirrors Go's validateEcVolume. fn validate_ec_volume(&self, collection: &str, vid: VolumeId) -> bool { let base = volume_file_name(&self.directory, collection, vid); let dat_path = format!("{}.dat", base); + // Custom ratio comes from the volume's own .vif; the server holds no + // cluster EC config in memory. + let data_shards = + ec_data_shards_from_vif(&self.directory, &self.idx_directory, collection, vid); + + // On-disk .dat: an empty <= superblock one is a stub; a transient stat + // error keeps the shards rather than deleting. let mut expected_shard_size: Option = None; - // An empty .dat (<= a superblock, zero needles) cannot be the encode - // source -- it is a leftover stub -- so treat it as absent rather than - // letting it mark a healthy distributed EC volume as an interrupted - // local encode, which would delete its shards. let dat_exists = match fs::metadata(&dat_path) { Ok(meta) if meta.len() > SUPER_BLOCK_SIZE as u64 => { - expected_shard_size = Some(calculate_expected_shard_size(meta.len() as i64)); + expected_shard_size = + Some(calculate_expected_shard_size(meta.len() as i64, data_shards)); true } Ok(_) => false, Err(e) if e.kind() == io::ErrorKind::NotFound => false, - // Unexpected stat error: don't risk classifying local EC as - // distributed; fail validation instead of deleting anything. - Err(_) => return false, + Err(e) => { + warn!(volume_id = vid.0, error = %e, "cannot stat .dat; keeping EC shards"); + return true; + } }; let mut shard_count = 0usize; @@ -276,14 +283,10 @@ impl DiskLocation { let size = meta.len() as i64; if let Some(prev) = actual_shard_size { if size != prev { - warn!( - volume_id = vid.0, - shard = i, - size, - expected = prev, - "EC shard size mismatch" - ); - return false; + // Inconsistent sizes signal corruption or mixed + // generations; not trusted for deletion -> keep. + warn!(volume_id = vid.0, shard = i, size, expected = prev, "EC shard size mismatch; keeping shards"); + return true; } } else { actual_shard_size = Some(size); @@ -291,49 +294,30 @@ impl DiskLocation { shard_count += 1; } Err(e) if e.kind() != io::ErrorKind::NotFound => { - warn!( - volume_id = vid.0, - shard = i, - error = %e, - "failed to stat EC shard" - ); - return false; + warn!(volume_id = vid.0, shard = i, error = %e, "cannot stat EC shard; keeping shards"); + return true; } _ => {} // not found or zero size — skip } } - // If .dat exists, validate shard size matches expected - if dat_exists { - if let (Some(actual), Some(expected)) = (actual_shard_size, expected_shard_size) { - if actual != expected { - warn!( - volume_id = vid.0, - actual_shard_size = actual, - expected_shard_size = expected, - "EC shard size doesn't match .dat file" - ); - return false; - } - } - } - - // Distributed EC (no .dat): any shard count is valid if !dat_exists { - return true; + return true; // distributed EC; any shard count is valid } - // With .dat: need at least DATA_SHARDS_COUNT shards - if shard_count < DATA_SHARDS_COUNT { - warn!( - volume_id = vid.0, - shard_count, - required = DATA_SHARDS_COUNT, - "EC volume has .dat but too few shards" - ); + // Reclaim only when it loses no data. Shards smaller than this .dat's + // full encode are an interrupted encode whose .dat is the complete + // source -> reclaim. Shards >= expected (valid/distributing EC, or a + // stale/partial .dat beside larger real shards) may be the only copy -> keep. + if shard_count == 0 { return false; } - + if let (Some(actual), Some(expected)) = (actual_shard_size, expected_shard_size) { + if actual < expected { + warn!(volume_id = vid.0, actual, expected, "shards smaller than the .dat's full encode; reclaiming the complete .dat"); + return false; + } + } true } @@ -895,28 +879,17 @@ impl DiskLocation { let shard_ids: Vec = shards.iter().map(|(_, sid)| *sid).collect(); if let Err(e) = self.mount_ec_shards(vid, collection, &shard_ids, "") { - // mount_ec_shards adds shards one at a time and increments - // the per-shard metric for each. If it fails halfway, plain - // ec_volumes.remove(vid) would leak metric increments for - // the shards that did mount. Drive cleanup through - // unmount_ec_shards which mirror-decrements the metric, then - // the empty EcVolume drops itself. - if dat_exists { - warn!( - volume_id = vid.0, - "Failed to load EC shards and .dat exists ({}), cleaning up EC files to use .dat", - e, - ); - self.unmount_ec_shards(vid, &shard_ids); - self.remove_ec_volume_files(collection, vid); - } else { - warn!( - volume_id = vid.0, - "Failed to load EC shards: {} (this may be normal for distributed EC volumes)", - e, - ); - self.unmount_ec_shards(vid, &shard_ids); - } + // A mount failure (corrupt/locked .ecx, EMFILE, transient I/O) is + // not proof the shards are disposable -- validate_ec_volume already + // decided they may be the only copy. Release partially-mounted + // shards (mirror-decrements the metric) but keep the files; never + // delete on a load error. + warn!( + volume_id = vid.0, + "Failed to load EC shards: {}; keeping files for retry", + e, + ); + self.unmount_ec_shards(vid, &shard_ids); } } @@ -1003,14 +976,14 @@ pub fn get_disk_stats(path: &str) -> (u64, u64) { /// Calculate expected EC shard size from .dat file size. /// Matches Go's `calculateExpectedShardSize`: large blocks (1GB * data_shards) first, /// then small blocks (1MB * data_shards) for the remainder. -fn calculate_expected_shard_size(dat_file_size: i64) -> i64 { - let large_batch_size = ERASURE_CODING_LARGE_BLOCK_SIZE as i64 * DATA_SHARDS_COUNT as i64; +fn calculate_expected_shard_size(dat_file_size: i64, data_shards: usize) -> i64 { + let large_batch_size = ERASURE_CODING_LARGE_BLOCK_SIZE as i64 * data_shards as i64; let num_large_batches = dat_file_size / large_batch_size; let mut shard_size = num_large_batches * ERASURE_CODING_LARGE_BLOCK_SIZE as i64; let remaining = dat_file_size - (num_large_batches * large_batch_size); if remaining > 0 { - let small_batch_size = ERASURE_CODING_SMALL_BLOCK_SIZE as i64 * DATA_SHARDS_COUNT as i64; + let small_batch_size = ERASURE_CODING_SMALL_BLOCK_SIZE as i64 * data_shards as i64; // Ceiling division let num_small_batches = (remaining + small_batch_size - 1) / small_batch_size; shard_size += num_small_batches * ERASURE_CODING_SMALL_BLOCK_SIZE as i64; @@ -1019,6 +992,29 @@ fn calculate_expected_shard_size(dat_file_size: i64) -> i64 { shard_size } +/// Resolve the EC data-shard count from the volume's own `.vif` (the volume +/// server never holds the cluster EC config in memory), checking the data dir +/// then the idx dir. Falls back to the default ratio when no EC `.vif` is found. +fn ec_data_shards_from_vif(directory: &str, idx_directory: &str, collection: &str, vid: VolumeId) -> usize { + for dir in [directory, idx_directory] { + let vif = format!("{}.vif", volume_file_name(dir, collection, vid)); + if let Some(ds) = fs::read_to_string(&vif) + .ok() + .and_then(|s| serde_json::from_str::(&s).ok()) + .and_then(|vi| vi.ec_shard_config) + .map(|c| c.data_shards as usize) + { + if ds > 0 { + return ds; + } + } + if directory == idx_directory { + break; + } + } + DATA_SHARDS_COUNT +} + /// Parse a volume filename like "collection_42.dat" or "42.dat" into (collection, VolumeId). /// Parse a `_` or `` base name into its parts. /// Mirrors `parseCollectionVolumeId` in @@ -1193,6 +1189,52 @@ mod tests { assert!(!std::path::Path::new(&format!("{}.dat", vbase)).exists()); } + // The headline geometry: a stale/partial .dat (e.g. an interrupted decode) + // smaller than the volume's real source sits next to the full-size shards + // that are the only copy. validate_ec_volume must keep the shards. + #[test] + fn test_validate_ec_volume_partial_dat_next_to_full_shards_keeps() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let loc = DiskLocation::new(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new()).unwrap(); + let base = volume_file_name(dir, "", VolumeId(70)); + let ds = crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT; + let full = calculate_expected_shard_size(30 * 1024 * 1024, ds); + for i in 0..ds { + std::fs::File::create(format!("{}.ec{:02}", base, i)).unwrap().set_len(full as u64).unwrap(); + } + // Partial .dat: bigger than a superblock so it is not swept as a stub, + // but smaller than what these shards encode. + std::fs::File::create(format!("{}.dat", base)).unwrap().set_len(5 * 1024 * 1024).unwrap(); + assert!( + loc.validate_ec_volume("", VolumeId(70)), + "full-size shards beside a smaller (stale/partial) .dat must be kept", + ); + } + + // The legitimate cleanup: a full source .dat next to shards SMALLER than it + // would encode (an interrupted encode). The .dat is the complete source, so + // reclaiming it loses no data. + #[test] + fn test_validate_ec_volume_interrupted_encode_reclaims() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let loc = DiskLocation::new(dir, dir, 10, DiskType::HardDrive, MinFreeSpace::Percent(1.0), Vec::new()).unwrap(); + let base = volume_file_name(dir, "", VolumeId(71)); + let ds = crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT; + let dat_size = 30 * 1024 * 1024i64; + std::fs::File::create(format!("{}.dat", base)).unwrap().set_len(dat_size as u64).unwrap(); + let partial = calculate_expected_shard_size(dat_size, ds) / 3; + assert!(partial > 0); + for i in 0..ds { + std::fs::File::create(format!("{}.ec{:02}", base, i)).unwrap().set_len(partial as u64).unwrap(); + } + assert!( + !loc.validate_ec_volume("", VolumeId(71)), + "shards smaller than the full source .dat should be reclaimable", + ); + } + #[test] fn test_parse_volume_filename() { assert_eq!( diff --git a/seaweed-volume/src/storage/store_ec_reconcile.rs b/seaweed-volume/src/storage/store_ec_reconcile.rs index 0a3edab5e..afeef7547 100644 --- a/seaweed-volume/src/storage/store_ec_reconcile.rs +++ b/seaweed-volume/src/storage/store_ec_reconcile.rs @@ -23,7 +23,6 @@ use tracing::{info, warn}; use crate::storage::disk_location::{is_ec_shard_extension, parse_collection_volume_id_pub}; use crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT; use crate::storage::store::Store; -use crate::storage::super_block::SUPER_BLOCK_SIZE; use crate::storage::types::VolumeId; pub(crate) fn ec_local_ecx_path(dir: &str, collection: &str, vid: VolumeId) -> String { @@ -237,8 +236,15 @@ impl Store { let mut victims: Vec = Vec::new(); for (loc_idx, loc) in self.locations.iter().enumerate() { for (vid, ev) in loc.ec_volumes() { + // Use the volume's own ratio, not the OSS default, so a full + // custom-ratio data set (e.g. 9 of a 9+3) is not mistaken for a leftover. + let data_shards = if ev.data_shards > 0 { + ev.data_shards as usize + } else { + DATA_SHARDS_COUNT + }; let shard_count = ev.shard_count(); - if shard_count >= DATA_SHARDS_COUNT { + if shard_count >= data_shards { continue; } let key = EcKey { @@ -254,15 +260,10 @@ impl Store { // per-disk pass; don't second-guess it here. continue; } - // Credible source size: prefer .vif's encode-time size; when - // unknown (0) require more than a bare superblock so an empty - // 8-byte stub (e.g. a phantom .dat) can't pass. - let required = if ev.dat_file_size > 0 { - ev.dat_file_size as u64 - } else { - SUPER_BLOCK_SIZE as u64 + 1 - }; - if owner.size < required { + // Delete only against a byte-exact committed source: the sibling + // .dat must equal the size .vif recorded at encode time. An + // unknown (0) or mismatched size cannot prove the .dat holds this data. + if ev.dat_file_size <= 0 || owner.size != ev.dat_file_size as u64 { warn!( volume_id = vid.0, collection = %ev.collection, @@ -270,8 +271,30 @@ impl Store { shard_count, sibling_dir = %self.locations[owner.location].directory, sibling_dat_size = owner.size, - required, - "sibling .dat is smaller than the EC source size; leaving partial EC in place so distributed reconstruction is still possible (issue 9478)", + recorded = ev.dat_file_size, + "sibling .dat does not byte-exactly match the recorded EC source size; leaving partial EC in place", + ); + continue; + } + // Never prune when the shards are recoverable node-wide (a set + // split across sibling disks summing to >= data_shards); they + // may be sole copies of a distributed volume. + let mut node_wide_bits = ev.shard_bits().0; + for other in &self.locations { + if let Some(other_ev) = other.find_ec_volume(*vid) { + if other_ev.collection == ev.collection { + node_wide_bits |= other_ev.shard_bits().0; + } + } + } + let node_wide = node_wide_bits.count_ones() as usize; + if node_wide >= data_shards { + warn!( + volume_id = vid.0, + collection = %ev.collection, + node_wide, + data_shards, + "shards present node-wide are independently recoverable; leaving EC in place despite a sibling .dat", ); continue; } @@ -1250,14 +1273,33 @@ mod tests { let collection = "pics"; let vid = 122u32; - // Disk A (sdd): a .dat whose name must be present so index_dat_owners - // records this disk as the .dat owner (content doesn't matter). + // Disk A (sdd): a .dat whose size must byte-exactly match the EC + // source size recorded in the sibling .vif for the prune to treat it + // as the committed source. let dat_path = dat_dir.join(format!("{}_{}.dat", collection, vid)); std::fs::write(&dat_path, vec![0u8; 1024]).unwrap(); // Disk B (sdf): partial EC — one shard, plus .ecx / .ecj / .vif. write_shard(ec_dir.to_str().unwrap(), collection, vid, 1); write_index_files(ec_dir.to_str().unwrap(), collection, vid, 10, 4); + // Record the encode-time source size in the EC .vif so the prune's + // byte-exact credibility gate recognizes the 1024-byte sibling .dat as + // the source (a real encoded volume records this). + std::fs::write( + ec_dir.join(format!("{}_{}.vif", collection, vid)), + serde_json::to_string(&VifVolumeInfo { + version: 3, + dat_file_size: 1024, + ec_shard_config: Some(VifEcShardConfig { + data_shards: 10, + parity_shards: 4, + ..Default::default() + }), + ..Default::default() + }) + .unwrap(), + ) + .unwrap(); let mut store = Store::new(NeedleMapKind::InMemory); store diff --git a/test/volume_server/grpc/ec_multi_disk_lifecycle_test.go b/test/volume_server/grpc/ec_multi_disk_lifecycle_test.go index 418f87dcd..644fb5614 100644 --- a/test/volume_server/grpc/ec_multi_disk_lifecycle_test.go +++ b/test/volume_server/grpc/ec_multi_disk_lifecycle_test.go @@ -14,6 +14,8 @@ import ( "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" ) // TestEcLifecycleAcrossMultipleDisks drives encode, mount, read, drop-dat, @@ -482,7 +484,6 @@ func plantPartialEc(t testing.TB, dir, collection string, volumeID uint32, shard }{ {".ecx", []byte("dummy ecx")}, {".ecj", nil}, - {".vif", nil}, } { p := filepath.Join(dir, sideFileName(collection, volumeID, side.ext)) f, err := os.Create(p) @@ -497,6 +498,19 @@ func plantPartialEc(t testing.TB, dir, collection string, volumeID uint32, shard } f.Close() } + // Record the encode-time source size so the prune's byte-exact gate + // recognizes the sibling .dat as the source, as a real encoded volume does. + vifPath := filepath.Join(dir, sideFileName(collection, volumeID, ".vif")) + if err := volume_info.SaveVolumeInfo(vifPath, &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + DatFileSize: datFileSize, + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: uint32(erasure_coding.DataShardsCount), + ParityShards: uint32(erasure_coding.ParityShardsCount), + }, + }); err != nil { + t.Fatalf("save planted .vif: %v", err) + } } // mirrors calculateExpectedShardSize in weed/storage/disk_location_ec.go diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go index 24d43b337..39b22f0e3 100644 --- a/weed/storage/disk_location_ec.go +++ b/weed/storage/disk_location_ec.go @@ -371,20 +371,12 @@ func (l *DiskLocation) handleFoundEcxFile(shards []string, collection string, vo return } - // Attempt to load the EC shards + // A load failure (corrupt/locked .ecx, EMFILE, transient I/O) is not proof + // the shards are disposable -- validateEcVolume already decided they may be + // the only copy. Release FDs but keep the files for retry; never delete here. if err := l.loadEcShards(shards, collection, volumeId, onShardLoad); err != nil { - // If EC shards failed to load and .dat still exists, clean up EC files to allow .dat file to be used - // If .dat is gone, log error but don't clean up (may be waiting for shards from other servers) - if datExists { - glog.Warningf("Failed to load EC shards for volume %d and .dat exists: %v, cleaning up EC files to use .dat...", volumeId, err) - // Unload first to release FDs, then remove files - l.unloadEcVolume(volumeId) - l.removeEcVolumeFiles(collection, volumeId) - } else { - glog.Warningf("Failed to load EC shards for volume %d: %v (this may be normal for distributed EC volumes)", volumeId, err) - // Clean up any partially loaded in-memory state. This does not delete files. - l.unloadEcVolume(volumeId) - } + glog.Warningf("Failed to load EC shards for volume %d: %v; keeping files for retry", volumeId, err) + l.unloadEcVolume(volumeId) return } } @@ -458,98 +450,70 @@ func calculateExpectedShardSize(datFileSize int64, dataShardCount int) int64 { return shardSize } -// validateEcVolume checks if EC volume has enough shards to be functional -// For distributed EC volumes (where .dat is deleted), any number of shards is valid -// For incomplete EC encoding (where .dat still exists), we need at least DataShardsCount shards -// Also validates that all shards have the same size (required for Reed-Solomon EC) -// If .dat exists, it also validates shards match the expected size based on .dat file size +// validateEcVolume reports whether the EC files for (collection, vid) on this +// disk may be deleted to reclaim the local .dat. It returns false (delete) +// only when that provably loses no data; every ambiguity returns true (keep), +// since the shards may be the only copy of distributed-EC data. func (l *DiskLocation) validateEcVolume(collection string, vid needle.VolumeId) bool { baseFileName := erasure_coding.EcShardFileName(collection, l.Directory, int(vid)) datFileName := baseFileName + ".dat" - var expectedShardSize int64 = -1 - datExists := false - - // Resolve the data-shard count from the volume's own .vif, which is - // written at encode time and travels with the volume. The volume - // server never loads the cluster EC config into memory, so the OSS - // default (10) would size a custom-ratio volume's shards (e.g. 9+3) - // for 10 data shards, fail the size check, and wrongly delete a - // healthy volume on reboot. Fall back to the default only when the - // .vif carries no EC shard config. + // Custom ratio comes from the volume's own .vif; the server holds no + // cluster EC config in memory. dataShards := l.ecDataShardsFromVif(collection, vid) - // If .dat file exists, compute exact expected shard size from it. - // An empty .dat (<= a superblock, zero needles) cannot be the encode - // source -- it is a leftover stub -- so treat it as absent rather than - // letting it mark a healthy distributed EC volume as an interrupted - // local encode, which would delete its shards. + // On-disk .dat size, or -1 when absent (an empty <= superblock .dat is a + // stub). A transient stat error keeps the shards rather than deleting. + var expectedShardSize int64 = -1 + datExists := false if datFileInfo, err := os.Stat(datFileName); err == nil { if datFileInfo.Size() > int64(super_block.SuperBlockSize) { datExists = true expectedShardSize = calculateExpectedShardSize(datFileInfo.Size(), dataShards) } } else if !os.IsNotExist(err) { - // If stat fails with unexpected error (permission, I/O), fail validation - // Don't treat this as "distributed EC" - it could be a temporary error - glog.Warningf("Failed to stat .dat file %s: %v", datFileName, err) - return false + glog.Warningf("EC volume %d: cannot stat .dat %s (%v); keeping EC shards", vid, datFileName, err) + return true } + // Count local shards; a transient stat error or inconsistent sizes -> keep. shardCount := 0 var actualShardSize int64 = -1 - - // Count shards and validate they all have the same size (required for Reed-Solomon EC) - // Check up to MaxShardCount (32) to support custom EC ratios for i := 0; i < erasure_coding.MaxShardCount; i++ { shardFileName := baseFileName + erasure_coding.ToExt(i) fi, err := os.Stat(shardFileName) - if err == nil { - // Check if file has non-zero size if fi.Size() > 0 { - // Validate all shards are the same size (required for Reed-Solomon EC) if actualShardSize == -1 { actualShardSize = fi.Size() } else if fi.Size() != actualShardSize { - glog.Warningf("EC volume %d shard %d has size %d, expected %d (all EC shards must be same size)", - vid, i, fi.Size(), actualShardSize) - return false + glog.Warningf("EC volume %d shard %d size %d != %d; keeping EC shards", vid, i, fi.Size(), actualShardSize) + return true } shardCount++ } } else if !os.IsNotExist(err) { - // If stat fails with unexpected error (permission, I/O), fail validation - // This is consistent with .dat file error handling - glog.Warningf("Failed to stat shard file %s: %v", shardFileName, err) - return false + glog.Warningf("EC volume %d: cannot stat shard %s (%v); keeping EC shards", vid, shardFileName, err) + return true } } - // If .dat file exists, validate shard size matches expected size - if datExists && actualShardSize > 0 && expectedShardSize > 0 { - if actualShardSize != expectedShardSize { - glog.Warningf("EC volume %d: shard size %d doesn't match expected size %d (based on .dat file size)", - vid, actualShardSize, expectedShardSize) - return false - } - } - - // If .dat file is gone, this is a distributed EC volume - any shard count is valid if !datExists { - glog.V(1).Infof("EC volume %d: distributed EC (.dat removed) with %d shards", vid, shardCount) - return true + return true // distributed EC; any shard count is valid } - // If .dat file exists, we need at least DataShards shards locally - // for the volume's configured ratio. Otherwise it's an incomplete - // EC encoding that should be cleaned up. - if shardCount < dataShards { - glog.Warningf("EC volume %d has .dat file but only %d shards (need at least %d for local EC)", - vid, shardCount, dataShards) + // Reclaim only when it loses no data. Shards smaller than this .dat's full + // encode are an interrupted encode whose .dat is the complete source -> + // reclaim. Shards >= expected (valid/distributing EC, or a stale/partial + // .dat beside larger real shards) may be the only copy -> keep. + if shardCount == 0 { + return false + } + if expectedShardSize > 0 && actualShardSize > 0 && actualShardSize < expectedShardSize { + glog.Warningf("EC volume %d: %d shards of %d bytes are smaller than the .dat's full encode (%d bytes); reclaiming the complete .dat", + vid, shardCount, actualShardSize, expectedShardSize) return false } - return true } diff --git a/weed/storage/disk_location_ec_test.go b/weed/storage/disk_location_ec_test.go index d794456d0..f62c9beb6 100644 --- a/weed/storage/disk_location_ec_test.go +++ b/weed/storage/disk_location_ec_test.go @@ -54,14 +54,18 @@ func TestIncompleteEcEncodingCleanup(t *testing.T) { expectLoadSuccess: false, }, { - name: "Incomplete EC: shards with .ecx but < 10 shards, .dat exists - should cleanup", + // Full-size shards beside a .dat are NOT an interrupted local + // encode (which leaves equally-truncated shards smaller than the + // .dat); they may be sole copies of a distributed volume, so the + // safe behavior is to keep them rather than delete on a low count. + name: "Distributed EC: full-size shards with .ecx, < 10 of them, .dat exists - keep", volumeId: 102, collection: "", createDatFile: true, createEcxFile: true, createEcjFile: false, - numShards: 7, // Less than DataShardsCount (10) - expectCleanup: true, + numShards: 7, // Less than DataShardsCount (10), but full size + expectCleanup: false, expectLoadSuccess: false, }, { @@ -275,12 +279,16 @@ func TestValidateEcVolume(t *testing.T) { expectValid: true, }, { - name: "Invalid: .dat exists with < 10 shards", + // Full-size shards smaller in count than dataShards may be sole + // copies of a distributed volume (a real interrupted local encode + // leaves equally-truncated shards, not full-size ones), so they + // are kept rather than deleted in favor of the .dat. + name: "Keep: .dat exists with < 10 full-size shards (possible distributed sole copies)", volumeId: 201, collection: "", createDatFile: true, numShards: 9, - expectValid: false, + expectValid: true, }, { name: "Valid: .dat deleted (distributed EC) with any shards", @@ -307,12 +315,15 @@ func TestValidateEcVolume(t *testing.T) { expectValid: false, }, { - name: "Invalid: .dat exists with different size shards", + // Inconsistent shard sizes signal corruption or mixed generations, + // not a clean interrupted encode; deleting them could destroy the + // only copy, so validation keeps them. + name: "Keep: .dat exists with different size shards (inconsistent, not trusted for deletion)", volumeId: 205, collection: "", createDatFile: true, numShards: 10, // Will create shards with varying sizes - expectValid: false, + expectValid: true, }, } diff --git a/weed/storage/ec_startup_shard_safety_test.go b/weed/storage/ec_startup_shard_safety_test.go new file mode 100644 index 000000000..ef6deeb5b --- /dev/null +++ b/weed/storage/ec_startup_shard_safety_test.go @@ -0,0 +1,103 @@ +package storage + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" +) + +func writeSizedFile(t *testing.T, path string, size int64) { + t.Helper() + f, err := os.Create(path) + if err != nil { + t.Fatalf("create %s: %v", path, err) + } + if size > 0 { + if err := f.Truncate(size); err != nil { + f.Close() + t.Fatalf("truncate %s: %v", path, err) + } + } + if err := f.Close(); err != nil { + t.Fatalf("close %s: %v", path, err) + } +} + +// TestValidateEcVolume_PartialDatNextToFullShardsKeeps reproduces the headline +// data-loss geometry: an interrupted decode (or any stale/partial .dat) leaves +// a .dat SMALLER than the volume's real source next to the full-size EC shards +// that are the only copy. validateEcVolume must keep the shards, not delete +// them in favor of the partial .dat. +func TestValidateEcVolume_PartialDatNextToFullShardsKeeps(t *testing.T) { + l := newTestDiskLocation(t.TempDir()) + base := erasure_coding.EcShardFileName("", l.Directory, 70) + + // Real shards sized for a 30 MB source volume. + fullShardSize := calculateExpectedShardSize(30*1024*1024, erasure_coding.DataShardsCount) + for i := 0; i < erasure_coding.DataShardsCount; i++ { + writeSizedFile(t, base+erasure_coding.ToExt(i), fullShardSize) + } + // A partial .dat (e.g. a decode crashed mid-write) far smaller than the + // shards' real source — bigger than a superblock so it is not swept as a + // stub, but smaller than what these shards encode. + writeSizedFile(t, base+".dat", 5*1024*1024) + + if !l.validateEcVolume("", 70) { + t.Fatal("validateEcVolume deleted full-size shards beside a smaller (stale/partial) .dat; the shards may be the only copy") + } +} + +// TestValidateEcVolume_InterruptedEncodeReclaimsDat is the legitimate cleanup: +// a full source .dat next to shards SMALLER than it would encode (an encode +// interrupted mid-shard-write). The .dat is the complete source, so reclaiming +// it (deleting the partial shards) loses no data. +func TestValidateEcVolume_InterruptedEncodeReclaimsDat(t *testing.T) { + l := newTestDiskLocation(t.TempDir()) + base := erasure_coding.EcShardFileName("", l.Directory, 71) + + datSize := int64(30 * 1024 * 1024) + writeSizedFile(t, base+".dat", datSize) + // Shards truncated well below the .dat's full encode (partial writes). + partialShardSize := calculateExpectedShardSize(datSize, erasure_coding.DataShardsCount) / 3 + if partialShardSize <= 0 { + t.Fatal("test setup: partial shard size must be positive") + } + for i := 0; i < erasure_coding.DataShardsCount; i++ { + writeSizedFile(t, base+erasure_coding.ToExt(i), partialShardSize) + } + + if l.validateEcVolume("", 71) { + t.Fatal("validateEcVolume kept shards smaller than the full source .dat; an interrupted local encode should be reclaimable") + } +} + +// TestValidateEcVolume_TransientStatErrorKeeps confirms a non-ENOENT stat error +// never authorizes deletion: a shard directory that cannot be read (EACCES) +// must keep the data, not delete it. +func TestValidateEcVolume_TransientStatErrorKeeps(t *testing.T) { + if os.Getuid() == 0 { + t.Skip("root bypasses directory permission checks") + } + parent := t.TempDir() + sub := filepath.Join(parent, "locked") + if err := os.Mkdir(sub, 0o755); err != nil { + t.Fatalf("mkdir: %v", err) + } + base := erasure_coding.EcShardFileName("", sub, 73) + writeSizedFile(t, base+".dat", int64(super_block.SuperBlockSize)+1024) + writeSizedFile(t, base+erasure_coding.ToExt(0), 1024) + if err := os.Chmod(sub, 0o000); err != nil { + t.Fatalf("chmod: %v", err) + } + t.Cleanup(func() { os.Chmod(sub, 0o755) }) + + l := newTestDiskLocation(sub) + // stat under the 0000 dir fails with EACCES (not ENOENT); validateEcVolume + // must keep (return true), never delete on a transient error. + if !l.validateEcVolume("", 73) { + t.Fatal("validateEcVolume returned delete on a transient stat error; must keep on ambiguity") + } +} diff --git a/weed/storage/store_ec_hybrid_repro_test.go b/weed/storage/store_ec_hybrid_repro_test.go index 163abd391..c497f9cd8 100644 --- a/weed/storage/store_ec_hybrid_repro_test.go +++ b/weed/storage/store_ec_hybrid_repro_test.go @@ -5,9 +5,11 @@ import ( "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -85,8 +87,17 @@ func TestIssue9478_PartialEcOnSiblingDiskOfHealthyDat(t *testing.T) { if f, err := os.Create(ecBase + ".ecj"); err == nil { f.Close() } - if f, err := os.Create(ecBase + ".vif"); err == nil { - f.Close() + // A credible EC .vif records the encode-time source size and ratio. The + // prune now deletes a partial leftover only when the sibling .dat matches + // that recorded size byte-for-byte (a real encoded volume records it), so + // the test reflects production rather than a loose "any .dat > superblock" + // gate. + if err := volume_info.SaveVolumeInfo(ecBase+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.Version3), + DatFileSize: datFileSize, + EcShardConfig: &volume_server_pb.EcShardConfig{DataShards: 10, ParityShards: 4}, + }); err != nil { + t.Fatalf("save ec .vif: %v", err) } minFreeSpace := util.MinFreeSpace{Type: util.AsPercent, Percent: 1, Raw: "1"} diff --git a/weed/storage/store_ec_reconcile.go b/weed/storage/store_ec_reconcile.go index 6addef419..43cad0164 100644 --- a/weed/storage/store_ec_reconcile.go +++ b/weed/storage/store_ec_reconcile.go @@ -10,7 +10,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" - "github.com/seaweedfs/seaweedfs/weed/storage/super_block" ) // datOwnerInfo records both the disk that holds a .dat for a given @@ -226,6 +225,24 @@ func (s *Store) indexEcxOwners() map[ecKeyForReconcile]ecxOwnerInfo { // to forget the registrations the per-disk pass already emitted on // NewEcShardsChan during startup, instead of waiting for the first // periodic heartbeat to reconcile. +// countEcShardsNodeWide returns the distinct EC shard ids for (collection, vid) +// across every disk on this store. Shards can be split across sibling disks, so +// a per-disk count understates a node-wide-recoverable set. Caller must not hold +// any DiskLocation.ecVolumesLock (this takes them). +func (s *Store) countEcShardsNodeWide(collection string, vid needle.VolumeId) int { + seen := make(map[erasure_coding.ShardId]struct{}) + for _, loc := range s.Locations { + loc.ecVolumesLock.RLock() + if ev, ok := loc.ecVolumes[vid]; ok && ev.Collection == collection { + for _, sh := range ev.Shards { + seen[sh.ShardId] = struct{}{} + } + } + loc.ecVolumesLock.RUnlock() + } + return len(seen) +} + func (s *Store) pruneIncompleteEcWithSiblingDat() { if len(s.Locations) < 2 { return @@ -251,10 +268,8 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() { loc.ecVolumesLock.RLock() for vid, ev := range loc.ecVolumes { shardCount := len(ev.Shards) - // Use the volume's configured data-shard count, not the OSS - // default: a disk holding a full data set for a custom ratio - // (e.g. 9 shards of a 9+3 volume) is independently recoverable - // and must not be mistaken for a partial leftover and wiped. + // Use the volume's own ratio, not the OSS default, so a full + // custom-ratio data set (e.g. 9 of a 9+3) is not mistaken for a leftover. dataShards := erasure_coding.DataShardsCount if ev.ECContext != nil && ev.ECContext.DataShards > 0 { dataShards = ev.ECContext.DataShards @@ -267,16 +282,13 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() { if !hasDat || owner.location == loc { continue } - // Credible source size: prefer .vif's encode-time size; when - // unknown (0) require more than a bare superblock so an empty - // 8-byte stub (e.g. a phantom .dat) can't pass. - requiredDatSize := ev.DatFileSize() - if requiredDatSize <= 0 { - requiredDatSize = int64(super_block.SuperBlockSize) + 1 - } - if owner.size < requiredDatSize { - glog.Warningf("ec volume %d (collection=%q) on %s has only %d shards but sibling .dat on %s is %d bytes (need >= %d); leaving partial EC in place so distributed reconstruction is still possible", - vid, ev.Collection, loc.Directory, shardCount, owner.location.Directory, owner.size, requiredDatSize) + // Delete only against a byte-exact committed source: the sibling + // .dat must equal the size .vif recorded at encode time. An unknown + // (0) or mismatched size cannot prove the .dat holds this data. + datFileSize := ev.DatFileSize() + if datFileSize <= 0 || owner.size != datFileSize { + glog.Warningf("ec volume %d (collection=%q) on %s has only %d shards; sibling .dat on %s is %d bytes but .vif recorded %d (need byte-exact match); leaving partial EC in place", + vid, ev.Collection, loc.Directory, shardCount, owner.location.Directory, owner.size, datFileSize) continue } victims = append(victims, victim{ @@ -291,7 +303,15 @@ func (s *Store) pruneIncompleteEcWithSiblingDat() { loc.ecVolumesLock.RUnlock() for _, v := range victims { - glog.Warningf("ec volume %d (collection=%q) on %s has only %d shards (need %d) while a healthy .dat exists on sibling disk %s; cleaning up leftover EC files (issue 9478)", + // Never prune when the shards are recoverable node-wide (a set + // split across sibling disks summing to >= dataShards); they may + // be sole copies of a distributed volume. + if nodeWide := s.countEcShardsNodeWide(v.collection, v.vid); nodeWide >= v.dataShards { + glog.Warningf("ec volume %d (collection=%q): %d shards present node-wide (>= %d) are independently recoverable; leaving EC in place despite a sibling .dat", + v.vid, v.collection, nodeWide, v.dataShards) + continue + } + glog.Warningf("ec volume %d (collection=%q) on %s has only %d shards (need %d) while a byte-exact source .dat exists on sibling disk %s; cleaning up leftover EC files", v.vid, v.collection, loc.Directory, v.shardCount, v.dataShards, v.datDir) loc.unloadEcVolume(v.vid) loc.removeEcVolumeFiles(v.collection, v.vid)