volume: make volume.scrub report a live needle whose stored id is damaged (#11468)

* storage: scrub live needles' stored id against the index key

scrubVolumeData only compared the needle's stored id for tombstones, so
header damage on a live needle — where the data CRC cannot see it —
passed every scrub mode while reads of that needle kept failing or
serving the wrong key's data. Compare the id for every indexed needle.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* rust volume: scrub live needles' stored id against the index key (parity)

Mirror the Go scrub fix: compare the stored needle id with the index
key for live needles too, not only for deleted ones.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* rust volume: cover damaged live needle id in scrub test

The tombstone test proved the index-key check fires for deleted entries;
add the live-needle mirror of Go's TestScrubVolumeDataChecksLiveNeedleId
so a regression in the live path is caught in Rust too.

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/src/storage/volume.rs

@@ -2974,7 +2974,11 @@ impl Volume {
                         "failed to read needle {} on volume {}: {}",
                         needle_id.0, self.id.0, e
                     ));
-                } else if size.is_deleted() && n.id != needle_id {
+                } else if n.id != needle_id {
+                    // The data CRC does not cover the header: a damaged needle id
+                    // still reads clean while every live-needle lookup on this
+                    // replica keeps finding the index key here. (Go parity:
+                    // the check covers live needles too, not just tombstones.)
                     broken.push(format!(
                         "index key {} does not match needle's Id {} on volume {}",
                         needle_id.0, n.id.0, self.id.0
@@ -6439,6 +6443,54 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_scrub_checks_live_needle_id() {
+        // Mirror of Go's TestScrubVolumeDataChecksLiveNeedleId: a live needle
+        // whose stored id is damaged still reads clean — the data CRC does not
+        // cover the header — so the scrub must catch it via the index key.
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = make_test_volume(dir);
+
+        write_test_needle(&mut v, 1, b"needle data");
+        v.sync_to_disk().unwrap();
+
+        let mut idx_file = File::open(v.file_name(".idx")).unwrap();
+        let mut live_offset = Offset::default();
+        idx::walk_index_file(&mut idx_file, 0, |key, offset, size| {
+            if key == NeedleId(1) && !offset.is_zero() && !size.is_deleted() {
+                live_offset = offset;
+            }
+            Ok(())
+        })
+        .unwrap();
+
+        let (_count, broken) = v.scrub().unwrap();
+        assert!(broken.is_empty(), "healthy needle must pass, got {:?}", broken);
+
+        let mut id_bytes = [0u8; NEEDLE_ID_SIZE];
+        NeedleId(99).to_bytes(&mut id_bytes);
+        let mut dat = OpenOptions::new()
+            .write(true)
+            .open(v.file_name(".dat"))
+            .unwrap();
+        dat.seek(SeekFrom::Start(
+            live_offset.to_actual_offset() as u64 + COOKIE_SIZE as u64,
+        ))
+        .unwrap();
+        dat.write_all(&id_bytes).unwrap();
+        dat.sync_all().unwrap();
+
+        let (_count, broken) = v.scrub().unwrap();
+        assert!(
+            broken
+                .iter()
+                .any(|e| e.contains("does not match needle's Id")),
+            "scrub should report the corrupted live needle's Id, got {:?}",
+            broken
+        );
+    }
+
     #[test]
     fn test_scrub_reports_truncated_local_deletion_tombstone() {
         // Mirror of Go's TestScrubVolumeDataReportsTruncatedLocalDeletionTombstone.

weed/storage/volume_checking.go

@@ -92,7 +92,10 @@ func (v *Volume) scrubVolumeData(idxFile *os.File, idxFileSize int64) (int64, []
 		n := needle.Needle{}
 		if err := n.ReadData(v.DataBackend, offset.ToActualOffset(), physicalSize, version); err != nil {
 			errs = append(errs, fmt.Errorf("failed to read needle %d on volume %d: %v", id, v.Id, err))
-		} else if size.IsDeleted() && n.Id != id {
+		} else if n.Id != id {
+			// The data CRC does not cover the header: a damaged needle id still
+			// reads clean while every live-needle lookup on this replica keeps
+			// finding the index key here.
 			errs = append(errs, fmt.Errorf("index key %v does not match needle's Id %v on volume %d", id, n.Id, v.Id))
 		}
 

weed/storage/volume_checking_test.go

@@ -220,6 +220,42 @@ func TestScrubVolumeDataChecksLocalDeletionTombstone(t *testing.T) {
 	}
 }
 
+// A live needle whose header id is damaged still reads clean — the data CRC
+// does not cover the header — so the scrub must catch it by comparing the
+// stored id with the index key, the same check already done for tombstones.
+func TestScrubVolumeDataChecksLiveNeedleId(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatalf("volume creation: %v", err)
+	}
+	defer v.Close()
+
+	const liveID = uint64(1)
+	offset, _, _, err := v.writeNeedle2(newRandomNeedle(liveID), true, false, false)
+	if err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	syncVolumeFiles(t, v)
+
+	if errs := scrubVolumeErrors(t, v); len(errs) != 0 {
+		t.Fatalf("healthy needle must pass scrub, got %v", errs)
+	}
+
+	corruptedID := make([]byte, types.NeedleIdSize)
+	types.NeedleIdToBytes(corruptedID, types.Uint64ToNeedleId(99))
+	if _, err := v.DataBackend.WriteAt(corruptedID, int64(offset)+types.CookieSize); err != nil {
+		t.Fatalf("corrupt live needle ID: %v", err)
+	}
+	if err := v.DataBackend.Sync(); err != nil {
+		t.Fatalf("sync corrupted .dat: %v", err)
+	}
+
+	if errs := scrubVolumeErrors(t, v); !strings.Contains(fmt.Sprint(errs), "does not match needle's Id") {
+		t.Fatalf("scrub should report the corrupted live needle's Id, got %v", errs)
+	}
+}
+
 func TestScrubVolumeDataReportsTruncatedLocalDeletionTombstone(t *testing.T) {
 	dir := t.TempDir()
 	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)