fix(ec): suppress deleted-needle size mismatch in EC LOCAL scrub (#10147)

* fix(ec): suppress deleted-needle size mismatch in EC LOCAL scrub

EcVolume.ScrubLocal reassembles each fully-local needle and ReadBytes-checks
it, but appended every error unconditionally. A needle the .ecx still reports
live while its reassembled on-disk header carries size 0 (delete state
disagrees between index and header) is not corruption — the LOCAL twin of the
#10130 fix for the FULL path. Suppress the ErrorSizeMismatch in that case;
genuine (non-zero) size mismatches and CRC/tail errors are still reported.

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo

* fix(ec): mirror EC LOCAL scrub deleted-needle suppression into Rust

Same suppression as the Go EcVolume.ScrubLocal change: a NeedleError::SizeMismatch
whose on-disk header size is 0 against a live index entry is a delete-state
disagreement, not corruption.

Claude-Session: https://claude.ai/code/session_015EE9Sc9EvNp8BCVva4RKdo

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"
+        );
+    }
 }

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")
+	}
+}

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))
+				}
 			}
 		}