feat(ec): EC bitrot CHECKSUM scrub on the Rust volume server (#10154)

* proto: add EC bitrot checksum messages + CHECKSUM scrub mode

Mirror weed/pb/volume_server.proto byte-for-byte (field numbers + types) so the
.ecsum sidecar payload is wire-identical across the Go and Rust binaries:
EcBitrotProtection / EcShardChecksums / ChecksumAlgorithm, VolumeScrubMode.CHECKSUM=4,
and VolumeEcShardsCopyRequest.copy_ecsum_file. No code uses them yet — the .ecsum
format, producer, mount-load, copy, and scrub land in following commits.

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

* feat(ec): port the .ecsum bitrot checksum module

ec_bitrot.rs mirrors weed/storage/erasure_coding/ec_bitrot.go: the .ecsum sidecar
format (14-byte big-endian ECSU header + CRC32C over a prost-serialized
EcBitrotProtection payload), the per-shard per-block CRC32C producer
(ShardChecksumBuilder), save/load with payload self-integrity, manifest
validation, status resolution, and verify_shard_file_blocks for the CHECKSUM
scrub. A byte-exact test pins the serialized bytes against the Go reference's
identical constant so a format drift in either binary fails loudly.

Producer wiring (encode/vacuum), mount-load, copy, and the mode-4 dispatch land
in following commits.

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

* test(ec): pin .ecsum sidecar bytes for cross-binary interop

Deterministic EcBitrotProtection -> exact on-disk bytes, asserted against a
canonical constant on BOTH sides (this test and ec_bitrot.rs), so a format drift
in either binary fails its own suite rather than silently desyncing a Go-written
.ecsum from a Rust-written one.

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

* feat(ec): write the .ecsum bitrot sidecar during EC encode

write_ec_files now feeds each shard's bytes through a per-shard
ShardChecksumBuilder as it writes them, then persists the generation-0 sidecar
(<base>.ecsum) alongside the shards — mirroring weed's WriteEcFiles +
SaveBitrotSidecar. Best-effort: a failed sidecar write leaves the generation
unprotected rather than failing the encode. A test confirms the produced sidecar
validates and its per-block CRCs match every on-disk shard.

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

* feat(ec): load the .ecsum at mount + EcVolume::checksum_scrub

EcVolume now loads and validates its generation-0 .ecsum sidecar at mount,
caching the parsed protection + BitrotStatus (Off/On/Invalid), and exposes
bitrot_protection() mirroring Go's EcVolume.BitrotProtection(). checksum_scrub()
verifies every locally-held shard's raw bytes against the sidecar block CRCs —
the only path that exercises cold parity shards — reporting mismatched shards
without mutating anything; a wholesale mismatch beyond parity is flagged as a
suspect sidecar rather than mass shard corruption. Mirrors Go's ChecksumScrub.

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

* feat(scrub): dispatch EC CHECKSUM (mode 4) to checksum_scrub

Accept VolumeScrubMode.CHECKSUM=4 and route it to EcVolume::checksum_scrub,
accumulating blocks scanned + mismatched shards into the scrub response, plus the
CHECKSUM scrub-mode metric label. Read-only bitrot verification over local shards.

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

* feat(ec): copy the .ecsum sidecar during VolumeEcShardsCopy

Honor copy_ecsum_file: when set, copy the generation-0 .ecsum alongside the
shards so protection travels with them, mirroring Go's non-2PC copy path.
Tolerant of a missing source (empty stream) — the 0-byte file is dropped so
mount sees no sidecar (protection off) rather than a truncated/invalid one.

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

* feat(ec): remove the .ecsum sidecars when destroying an EC volume

remove_ec_volume_files now clears <base>.ecsum (and any versioned .ecsum.v<N>)
from the data and idx dirs, so a vid reuse can't load a stale sidecar. Mirrors
Go's removeBitrotSidecars.

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

* style(ec): align bitrot comments and test setup for merge-cleanliness

Match the shared bitrot code (write_ec_files, encode_one_batch, checksum_scrub,
the encode sidecar test) to the canonical wording/layout so the volume-server
Rust port stays line-aligned across trees, keeping periodic merges conflict-free.
No behavior change.

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

seaweed-volume/src/storage/erasure_coding/ec_encoder.rs

@@ -10,6 +10,12 @@ use std::io::{Read, Seek, SeekFrom};
 
 use reed_solomon_erasure::galois_8::ReedSolomon;
 
+use crate::pb::volume_server_pb::{
+    ChecksumAlgorithm, EcBitrotProtection, EcShardChecksums,
+};
+use crate::storage::erasure_coding::ec_bitrot::{
+    self, ShardChecksumBuilder, DEFAULT_BITROT_BLOCK_SIZE,
+};
 use crate::storage::erasure_coding::ec_shard::*;
 use crate::storage::idx;
 use crate::storage::types::*;
@@ -52,12 +58,21 @@ pub fn write_ec_files(
         shard.create()?;
     }
 
+    // Per-shard bitrot checksum builders: accumulate a CRC32C for every
+    // DEFAULT_BITROT_BLOCK_SIZE block of each shard's byte stream as it is
+    // written, so the resulting `.ecsum` sidecar can later detect silent
+    // corruption in any shard (including cold parity).
+    let mut builders: Vec<ShardChecksumBuilder> = (0..total_shards)
+        .map(|_| ShardChecksumBuilder::new(DEFAULT_BITROT_BLOCK_SIZE as i64))
+        .collect();
+
     // Encode in large blocks, then small blocks
     encode_dat_file(
         &dat_file,
         dat_size,
         &rs,
         &mut shards,
+        &mut builders,
         data_shards,
         parity_shards,
     )?;
@@ -67,6 +82,42 @@ pub fn write_ec_files(
         shard.close();
     }
 
+    // Write the generation-0 bitrot sidecar (`<base>.ecsum`). Finalizing each
+    // builder yields covered_size (== total bytes written to that shard) and
+    // the packed little-endian CRC32C array.
+    let mut shard_checksums: Vec<EcShardChecksums> = Vec::with_capacity(total_shards);
+    for (i, builder) in builders.into_iter().enumerate() {
+        let (covered_size, packed) = builder.finalize();
+        shard_checksums.push(EcShardChecksums {
+            shard_id: i as u32,
+            covered_size,
+            block_crc32c: packed,
+        });
+    }
+    let prot = EcBitrotProtection {
+        algorithm: ChecksumAlgorithm::ChecksumCrc32c as i32,
+        block_size: DEFAULT_BITROT_BLOCK_SIZE as u32,
+        generation: 0,
+        ec_shard_config: Some(ec_bitrot::ec_shard_config(
+            data_shards as u32,
+            parity_shards as u32,
+        )),
+        shards: shard_checksums,
+        encode_uuid: ec_bitrot::new_encode_uuid(),
+    };
+    let sidecar_path = ec_bitrot::bitrot_sidecar_path(&base, 0);
+    if let Err(e) = ec_bitrot::save_bitrot_sidecar(&sidecar_path, &prot) {
+        // A failed sidecar must not fail the encode — the shards are already
+        // written and valid. The volume simply runs with bitrot protection
+        // off for this generation until the sidecar is regenerated.
+        tracing::warn!(
+            volume_id = volume_id.0,
+            path = %sidecar_path,
+            error = %e,
+            "ec encode: failed to write bitrot sidecar; protection off for this generation",
+        );
+    }
+
     Ok(())
 }
 
@@ -536,6 +587,7 @@ fn encode_dat_file(
     dat_size: i64,
     rs: &ReedSolomon,
     shards: &mut [EcVolumeShard],
+    builders: &mut [ShardChecksumBuilder],
     data_shards: usize,
     parity_shards: usize,
 ) -> io::Result<()> {
@@ -553,6 +605,7 @@ fn encode_dat_file(
             large_block_size,
             rs,
             shards,
+            builders,
             data_shards,
             parity_shards,
         )?;
@@ -572,6 +625,7 @@ fn encode_dat_file(
             small_block_size,
             rs,
             shards,
+            builders,
             data_shards,
             parity_shards,
         )?;
@@ -589,6 +643,7 @@ fn encode_one_batch(
     block_size: usize,
     rs: &ReedSolomon,
     shards: &mut [EcVolumeShard],
+    builders: &mut [ShardChecksumBuilder],
     data_shards: usize,
     parity_shards: usize,
 ) -> io::Result<()> {
@@ -642,9 +697,11 @@ fn encode_one_batch(
         )
     })?;
 
-    // Write all shard buffers to files
+    // Write all shard buffers to files and feed the same bytes to each
+    // shard's bitrot checksum builder, keeping covered_size == on-disk length.
     for (i, buf) in buffers.iter().enumerate() {
         shards[i].write_all(buf)?;
+        builders[i].write(buf);
     }
 
     Ok(())
@@ -712,6 +769,84 @@ mod tests {
         assert!(std::path::Path::new(&ecx_path).exists());
     }
 
+    fn make_volume_with_needles(n: u64) -> TempDir {
+        let tmp = TempDir::new().unwrap();
+        let dir = tmp.path().to_str().unwrap();
+        let mut v = Volume::new(
+            dir,
+            dir,
+            "",
+            VolumeId(1),
+            NeedleMapKind::InMemory,
+            None,
+            None,
+            0,
+            Version::current(),
+        )
+        .unwrap();
+        for i in 1..=n {
+            // Larger payloads so encoded shards span multiple bitrot blocks
+            // would require huge data; small payloads are fine for correctness.
+            let data = format!("test data for needle {} {}", i, "x".repeat(64));
+            let mut needle = Needle {
+                id: NeedleId(i),
+                cookie: Cookie(i as u32),
+                data: data.as_bytes().to_vec(),
+                data_size: data.len() as u32,
+                ..Needle::default()
+            };
+            v.write_needle(&mut needle, true).unwrap();
+        }
+        v.sync_to_disk().unwrap();
+        v.close();
+        tmp
+    }
+
+    /// Encode-time capture writes a valid generation-0 `.ecsum` sidecar whose
+    /// recorded checksums match the actual on-disk shards.
+    #[test]
+    fn test_encode_writes_valid_bitrot_sidecar() {
+        use crate::storage::erasure_coding::ec_bitrot;
+
+        let tmp = make_volume_with_needles(5);
+        let dir = tmp.path().to_str().unwrap();
+        write_ec_files(dir, dir, "", VolumeId(1), 10, 4).unwrap();
+
+        let base = format!("{}/1", dir);
+        let sidecar_path = ec_bitrot::bitrot_sidecar_path(&base, 0);
+        assert!(
+            std::path::Path::new(&sidecar_path).exists(),
+            "generation-0 .ecsum sidecar should exist after encode"
+        );
+
+        let prot = ec_bitrot::load_bitrot_sidecar(&sidecar_path).unwrap();
+        ec_bitrot::validate_manifest(&prot, 10, 4).unwrap();
+        assert_eq!(prot.generation, 0);
+        assert_eq!(prot.shards.len(), 14);
+        assert_eq!(prot.encode_uuid.len(), 16);
+        assert_eq!(prot.block_size, ec_bitrot::DEFAULT_BITROT_BLOCK_SIZE as u32);
+
+        // Every shard's recorded covered_size must equal its on-disk length and
+        // its block CRCs must verify clean.
+        let bs = prot.block_size as i64;
+        for entry in &prot.shards {
+            let path = format!("{}.ec{:02}", base, entry.shard_id);
+            let on_disk = std::fs::metadata(&path).unwrap().len() as i64;
+            assert_eq!(
+                entry.covered_size, on_disk,
+                "covered_size must equal on-disk length for shard {}",
+                entry.shard_id
+            );
+            let mm = ec_bitrot::verify_shard_file_blocks(&path, entry, bs).unwrap();
+            assert!(
+                mm.is_empty(),
+                "shard {} should verify clean, got mismatches {:?}",
+                entry.shard_id,
+                mm
+            );
+        }
+    }
+
     // encode_sample_volume writes a small volume and EC-encodes it, returning
     // the dir path so a test can drop/truncate shards and rebuild.
     fn encode_sample_volume(tmp: &TempDir) -> String {