From 8263b394ad26871e80b2feb2f91338735f18610f Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 7 Aug 2026 13:01:12 -0700 Subject: [PATCH] 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 | 7 ++ seaweed-volume/Cargo.toml | 3 + seaweed-volume/src/server/heartbeat.rs | 62 ++++++++++- seaweed-volume/src/storage/mod.rs | 1 + .../src/storage/volume_report_hash.rs | 104 ++++++++++++++++++ 5 files changed, 174 insertions(+), 3 deletions(-) create mode 100644 seaweed-volume/src/storage/volume_report_hash.rs diff --git a/seaweed-volume/Cargo.lock b/seaweed-volume/Cargo.lock index 150b88856..a4c875258 100644 --- a/seaweed-volume/Cargo.lock +++ b/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" diff --git a/seaweed-volume/Cargo.toml b/seaweed-volume/Cargo.toml index 41bb8fa31..481f7bc12 100644 --- a/seaweed-volume/Cargo.toml +++ b/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" diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index c0fcb3cdc..0a9181220 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/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 = HashMap::new(); let mut disk_total_bytes: HashMap = 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(); diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index 57e195ba5..abe8ba285 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/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; diff --git a/seaweed-volume/src/storage/volume_report_hash.rs b/seaweed-volume/src/storage/volume_report_hash.rs new file mode 100644 index 000000000..c661d4d7f --- /dev/null +++ b/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); + } +}