diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index b6f1d6001..0c2d17c4a 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -12,7 +12,7 @@ use std::time::{SystemTime, UNIX_EPOCH}; use crate::pb::master_pb; use crate::storage::erasure_coding::ec_locate; use crate::storage::erasure_coding::ec_shard::*; -use crate::storage::needle::needle::{get_actual_size, Needle}; +use crate::storage::needle::needle::{get_actual_size, Needle, NeedleError}; use crate::storage::types::*; /// An erasure-coded volume managing its local shards and index. @@ -861,10 +861,18 @@ impl EcVolume { if !has_remote_chunks { let mut n = Needle::default(); if let Err(e) = n.read_bytes(&data, 0, size, self.version) { - errs.push(format!( - "needle {} on volume {}, shards {:?}: {}", - id.0, self.volume_id.0, local_shard_ids, e - )); + // A delete-state disagreement between the .ecx index and the reassembled + // on-disk header (live index vs zero header size) is not corruption. + let delete_state_disagrees = matches!( + &e, + NeedleError::SizeMismatch { found, .. } if size.is_deleted() != (found.0 == 0) + ); + if !delete_state_disagrees { + errs.push(format!( + "needle {} on volume {}, shards {:?}: {}", + id.0, self.volume_id.0, local_shard_ids, e + )); + } } } Ok(()) @@ -1463,4 +1471,82 @@ mod tests { assert!(broken.is_empty(), "{:?}", broken); assert!(errs.is_empty(), "{:?}", errs); } + + #[test] + fn test_scrub_local_suppresses_delete_state_disagreement() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + // Live index entry (size > 0) whose reassembled on-disk header reports size 0 + // (deleted-on-shards but live-in-index) — a delete-state disagreement, not corruption. + let entries = vec![(NeedleId(1), Offset::from_actual_offset(0), Size(100))]; + write_ecx_file(dir, "", VolumeId(1), &entries); + + let vif = crate::storage::volume::VifVolumeInfo { + dat_file_size: 14000, + ..Default::default() + }; + let base = crate::storage::volume::volume_file_name(dir, "", VolumeId(1)); + std::fs::write( + format!("{}.vif", base), + serde_json::to_string_pretty(&vif).unwrap(), + ) + .unwrap(); + + // Shard 0 holds the needle's bytes; all-zero so the parsed header size is 0. + let mut shard0 = EcVolumeShard::new(dir, "", VolumeId(1), 0); + shard0.create().unwrap(); + shard0.write_all(&[0u8; 256]).unwrap(); + shard0.close(); + + let mut vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + vol.add_shard(EcVolumeShard::new(dir, "", VolumeId(1), 0)) + .unwrap(); + + let (count, broken, errs) = vol.scrub_local(); + assert_eq!(count, 1); + assert!(broken.is_empty(), "{:?}", broken); + assert!( + errs.is_empty(), + "delete-state disagreement must be suppressed, got {:?}", + errs + ); + } + + #[test] + fn test_scrub_local_reports_genuine_size_corruption() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let entries = vec![(NeedleId(1), Offset::from_actual_offset(0), Size(100))]; + write_ecx_file(dir, "", VolumeId(1), &entries); + + let vif = crate::storage::volume::VifVolumeInfo { + dat_file_size: 14000, + ..Default::default() + }; + let base = crate::storage::volume::volume_file_name(dir, "", VolumeId(1)); + std::fs::write( + format!("{}.vif", base), + serde_json::to_string_pretty(&vif).unwrap(), + ) + .unwrap(); + + // On-disk header reports a non-zero size (50) that disagrees with the live + // index size (100): genuine corruption, not a delete-state race — must report. + let mut bytes = vec![0u8; 256]; + bytes[15] = 50; // header size field (big-endian u32) = 50 + let mut shard0 = EcVolumeShard::new(dir, "", VolumeId(1), 0); + shard0.create().unwrap(); + shard0.write_all(&bytes).unwrap(); + shard0.close(); + + let mut vol = EcVolume::new(dir, dir, "", VolumeId(1)).unwrap(); + vol.add_shard(EcVolumeShard::new(dir, "", VolumeId(1), 0)) + .unwrap(); + + let (_count, _broken, errs) = vol.scrub_local(); + assert!( + !errs.is_empty(), + "a non-zero size mismatch is genuine corruption and must be reported" + ); + } } diff --git a/weed/storage/erasure_coding/ec_scrub_local_test.go b/weed/storage/erasure_coding/ec_scrub_local_test.go new file mode 100644 index 000000000..af6ccf42c --- /dev/null +++ b/weed/storage/erasure_coding/ec_scrub_local_test.go @@ -0,0 +1,65 @@ +package erasure_coding_test + +import ( + "os" + "path/filepath" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +// setupScrubLocalVolume seeds one live .ecx entry and a shard 0 whose bytes form +// the needle's on-disk header, then returns a loaded EcVolume with shard 0 added. +func setupScrubLocalVolume(t *testing.T, shard0 []byte) *erasure_coding.EcVolume { + t.Helper() + dir := t.TempDir() + collection := "test" + vid := needle.VolumeId(1) + base := filepath.Join(dir, collection+"_1") + + if err := os.WriteFile(base+".ecx", makeNeedleMapEntry(types.NeedleId(1), types.ToOffset(0), types.Size(100)), 0644); err != nil { + t.Fatalf("write ecx: %v", err) + } + if err := os.WriteFile(base+".ecj", []byte{}, 0644); err != nil { + t.Fatalf("write ecj: %v", err) + } + if err := os.WriteFile(base+".vif", []byte{}, 0644); err != nil { + t.Fatalf("write vif: %v", err) + } + if err := os.WriteFile(base+".ec00", shard0, 0644); err != nil { + t.Fatalf("write ec00: %v", err) + } + + ecv, err := erasure_coding.NewEcVolume("hdd", dir, dir, collection, vid) + if err != nil { + t.Fatalf("NewEcVolume: %v", err) + } + shard, err := erasure_coding.NewEcVolumeShard("hdd", dir, collection, vid, 0) + if err != nil { + t.Fatalf("NewEcVolumeShard: %v", err) + } + ecv.AddEcVolumeShard(shard) + return ecv +} + +// A needle the .ecx still reports as live (size 100) but whose reassembled +// on-disk header carries size 0 is a delete-state disagreement, not corruption. +func TestScrubLocal_SuppressesDeleteStateDisagreement(t *testing.T) { + ecv := setupScrubLocalVolume(t, make([]byte, 256)) // all-zero header => size 0 + if _, _, errs := ecv.ScrubLocal(); len(errs) != 0 { + t.Fatalf("delete-state disagreement must be suppressed, got %v", errs) + } +} + +// A non-zero on-disk header size that disagrees with the index is genuine +// corruption and must still be reported. +func TestScrubLocal_ReportsGenuineSizeCorruption(t *testing.T) { + shard0 := make([]byte, 256) + shard0[15] = 50 // header size field (big-endian uint32) = 50, != index 100 and != 0 + ecv := setupScrubLocalVolume(t, shard0) + if _, _, errs := ecv.ScrubLocal(); len(errs) == 0 { + t.Fatal("a non-zero size mismatch is genuine corruption and must be reported") + } +} diff --git a/weed/storage/erasure_coding/ec_volume_scrub.go b/weed/storage/erasure_coding/ec_volume_scrub.go index 208aeb78f..8cb33297e 100644 --- a/weed/storage/erasure_coding/ec_volume_scrub.go +++ b/weed/storage/erasure_coding/ec_volume_scrub.go @@ -2,6 +2,7 @@ package erasure_coding import ( "bytes" + "errors" "fmt" "slices" @@ -280,7 +281,12 @@ func (ecv *EcVolume) ScrubLocal() (int64, []*volume_server_pb.EcShardInfo, []err // needle was fully recovered from local shards \o/ let's check it n := needle.Needle{} if err := n.ReadBytes(data, 0, size, ecv.Version); err != nil { - errs = append(errs, fmt.Errorf("needle %d on volume %d, shards %v: %v", id, ecv.VolumeId, localShardIds, err)) + // A delete-state disagreement between the .ecx index and the reassembled + // on-disk header (live index vs zero header size) is not corruption. + deleteStateDisagrees := errors.Is(err, needle.ErrorSizeMismatch) && size.IsDeleted() != (n.Size == 0) + if !deleteStateDisagrees { + errs = append(errs, fmt.Errorf("needle %d on volume %d, shards %v: %v", id, ecv.VolumeId, localShardIds, err)) + } } }