rust: report the heartbeat volume digest

Mirrors the Go volume server. The master compares this against a digest it
computes itself, so the hash has to agree byte for byte across the two
implementations, not merely be a hash of the same fields: report_hash_vectors
pins it against values generated by the Go side, and the ttl and replica
placement narrowing the master applies when it decodes a message is applied
here too rather than assumed away.

A drift there would not corrupt anything, but every volume server on this
implementation would report a digest the master can never match and fall back
to sending its whole volume list forever, which is the cost the digest exists
to avoid.

seaweed-volume/Cargo.lock

@@ -4556,6 +4556,7 @@ dependencies = [
  "tracing-subscriber",
  "uuid",
  "x509-parser",
+ "xxhash-rust",
 ]
 
 [[package]]
@@ -4989,6 +4990,12 @@ version = "0.13.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4"
 
+[[package]]
+name = "xxhash-rust"
+version = "0.8.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6"
+
 [[package]]
 name = "yoke"
 version = "0.8.2"

seaweed-volume/Cargo.toml

@@ -75,6 +75,9 @@ serde_urlencoded = "0.7"
 crc32c = "0.6"
 crc32fast = "1"
 
+# xxhash64 - must match Go's cespare/xxhash for the heartbeat volume digest
+xxhash-rust = { version = "0.8", features = ["xxh64"] }
+
 # Memory-mapped files
 memmap2 = "0.9"
 

seaweed-volume/src/server/heartbeat.rs

@@ -19,6 +19,7 @@ use crate::pb::master_pb::seaweed_client::SeaweedClient;
 use crate::pb::volume_server_pb;
 use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig};
 use crate::storage::store::Store;
+use crate::storage::volume_report_hash::report_hash;
 use crate::storage::types::NeedleId;
 
 const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection";
@@ -800,6 +801,10 @@ fn build_heartbeat_with_ec_status(
     }
 
     let mut volumes = Vec::new();
+    // Digest of exactly what this heartbeat reports, so the master can tell
+    // whether its copy is current. Volumes skipped above -- quarantined,
+    // phantom, expired -- are absent from both the list and the digest.
+    let mut volume_digest: u64 = 0;
     let mut max_file_key = NeedleId(0);
     let mut max_volume_counts: HashMap<String, u32> = HashMap::new();
     let mut disk_total_bytes: HashMap<String, u64> = HashMap::new();
@@ -875,7 +880,7 @@ fn build_heartbeat_with_ec_status(
                 }
 
                 let (remote_storage_name, remote_storage_key) = vol.remote_storage_name_key();
-                volumes.push(master_pb::VolumeInformationMessage {
+                let volume_message = master_pb::VolumeInformationMessage {
                     id: vol.id.0,
                     size: volume_size,
                     collection: vol.collection.clone(),
@@ -892,8 +897,9 @@ fn build_heartbeat_with_ec_status(
                     disk_id: disk_id as u32,
                     remote_storage_name,
                     remote_storage_key,
-                    ..Default::default()
-                });
+                };
+                volume_digest ^= report_hash(&volume_message);
+                volumes.push(volume_message);
             } else if vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
                 delete_vids.push(vol.id);
                 should_delete_volume = true;
@@ -967,6 +973,7 @@ fn build_heartbeat_with_ec_status(
         rack: config.rack.clone(),
         admin_port: config.port as u32,
         volumes,
+        volume_digest: Some(volume_digest),
         deleted_ec_shards,
         has_no_volumes,
         has_no_ec_shards,
@@ -1246,6 +1253,55 @@ mod tests {
         assert!(heartbeat.has_no_volumes);
     }
 
+    // The digest must cover exactly the volumes the heartbeat carries. A volume
+    // reported but left out of the digest, or the reverse, makes the master's
+    // comparison disagree forever. An empty store still reports a digest, so
+    // the master can tell it from a server that computes none.
+    #[test]
+    fn test_build_heartbeat_digests_exactly_what_it_reports() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let dir = temp_dir.path().to_str().unwrap();
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                dir,
+                dir,
+                8,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+
+        let empty = build_heartbeat(&test_config(), &mut store);
+        assert_eq!(empty.volume_digest, Some(0));
+
+        for vid in [VolumeId(1), VolumeId(2)] {
+            store
+                .add_volume(
+                    vid,
+                    "pics",
+                    None,
+                    None,
+                    0,
+                    DiskType::HardDrive,
+                    Version::current(),
+                )
+                .unwrap();
+        }
+
+        let heartbeat = build_heartbeat(&test_config(), &mut store);
+        assert_eq!(heartbeat.volumes.len(), 2);
+
+        let expected = heartbeat
+            .volumes
+            .iter()
+            .fold(0u64, |acc, m| acc ^ crate::storage::volume_report_hash::report_hash(m));
+        assert_eq!(heartbeat.volume_digest, Some(expected));
+        assert_ne!(heartbeat.volume_digest, Some(0));
+    }
+
     #[test]
     fn test_build_heartbeat_tracks_go_read_only_labels_and_disk_id() {
         let temp_dir = tempfile::tempdir().unwrap();

seaweed-volume/src/storage/mod.rs

@@ -10,3 +10,4 @@ pub mod super_block;
 pub mod types;
 pub mod volume;
 pub mod volume_idx_repair;
+pub mod volume_report_hash;

seaweed-volume/src/storage/volume_report_hash.rs

@@ -0,0 +1,104 @@
+//! Mirror of `weed/storage/volume_report_hash.go`.
+//!
+//! The master compares the digest a volume server reports against one it
+//! computes itself, so this has to agree with the Go implementation
+//! byte-for-byte. `report_hash_vectors` pins that against values produced by
+//! the Go side; do not change the layout without regenerating them there.
+
+use xxhash_rust::xxh64::xxh64;
+
+use crate::pb::master_pb;
+
+/// Digests everything a volume server reports about a volume.
+///
+/// It must cover every field of `VolumeInformationMessage`: a change the hash
+/// misses is a change the master would never be told about.
+pub fn report_hash(m: &master_pb::VolumeInformationMessage) -> u64 {
+    let mut buf = [0u8; 57];
+    buf[0..4].copy_from_slice(&m.id.to_le_bytes());
+    buf[4..12].copy_from_slice(&m.size.to_le_bytes());
+    buf[12..20].copy_from_slice(&m.file_count.to_le_bytes());
+    buf[20..28].copy_from_slice(&m.delete_count.to_le_bytes());
+    buf[28..36].copy_from_slice(&m.deleted_byte_count.to_le_bytes());
+    // The master stores these narrowed, so hash what it will hold, not what the
+    // wire type could carry.
+    buf[36..40].copy_from_slice(&((m.replica_placement as u8) as u32).to_le_bytes());
+    buf[40..44].copy_from_slice(&((m.version as u8) as u32).to_le_bytes());
+    buf[44..48].copy_from_slice(&normalize_ttl(m.ttl).to_le_bytes());
+    buf[48..52].copy_from_slice(&m.compact_revision.to_le_bytes());
+    buf[52..56].copy_from_slice(&m.disk_id.to_le_bytes());
+    if m.read_only {
+        buf[56] = 1;
+    }
+    let mut h = xxh64(&buf, 0);
+
+    h = fold(h, xxh64(&(m.modified_at_second as u64).to_le_bytes(), 0));
+    h = fold(h, xxh64(m.collection.as_bytes(), 0));
+    h = fold(h, xxh64(m.disk_type.as_bytes(), 0));
+    h = fold(h, xxh64(m.remote_storage_name.as_bytes(), 0));
+    h = fold(h, xxh64(m.remote_storage_key.as_bytes(), 0));
+    h
+}
+
+/// A ttl whose count is zero encodes as zero however the unit is set, matching
+/// what the master stores after decoding it.
+fn normalize_ttl(ttl: u32) -> u32 {
+    let count = (ttl >> 8) & 0xff;
+    if count == 0 {
+        return 0;
+    }
+    (count << 8) | (ttl & 0xff)
+}
+
+/// Combines two hashes order-dependently, so swapping two string fields is not
+/// invisible.
+fn fold(h: u64, x: u64) -> u64 {
+    let h = (h ^ x).wrapping_mul(0x9E37_79B9_7F4A_7C15);
+    h ^ (h >> 29)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Produced by the Go implementation. If these drift, every volume server
+    // running this build reports a digest the master can never match, and falls
+    // back to sending its whole volume list forever.
+    #[test]
+    fn report_hash_vectors() {
+        let empty = master_pb::VolumeInformationMessage::default();
+        assert_eq!(report_hash(&empty), 17122085700329870549);
+
+        let mut one = master_pb::VolumeInformationMessage::default();
+        one.id = 1;
+        assert_eq!(report_hash(&one), 12867601919960834066);
+
+        let full = master_pb::VolumeInformationMessage {
+            id: 42,
+            size: 1 << 30,
+            collection: "c".to_string(),
+            file_count: 7,
+            delete_count: 2,
+            deleted_byte_count: 99,
+            read_only: true,
+            replica_placement: 10,
+            version: 3,
+            ttl: 3 << 8,
+            compact_revision: 5,
+            modified_at_second: 1700000000,
+            remote_storage_name: "s3".to_string(),
+            remote_storage_key: "k/1.dat".to_string(),
+            disk_type: "ssd".to_string(),
+            disk_id: 2,
+        };
+        assert_eq!(report_hash(&full), 12500327696413250175);
+    }
+
+    #[test]
+    fn ttl_with_no_count_is_dropped() {
+        assert_eq!(normalize_ttl(0), 0);
+        assert_eq!(normalize_ttl(3), 0);
+        assert_eq!(normalize_ttl(3 << 8), 3 << 8);
+        assert_eq!(normalize_ttl((3 << 8) | 4), (3 << 8) | 4);
+    }
+}