trim the comments on this change to the parts that are not evident

The list of fields was written out by hand, so a field added to
VolumeInformationMessage later would fall outside the digest while the test
went on passing, and a change to it would never reach the master. Walk the
message descriptor instead.

Some fields are narrowed or normalised on the way into VolumeInfo, so the
smallest change to the wire value can land back on the stored one; the test
offers several values per field and asks only that some change is visible.

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/proto/master.proto

@@ -105,6 +105,11 @@ message Heartbeat {
   // physical disk capacity per disk type, in bytes, from the underlying filesystem
   map<string, uint64> disk_total_bytes = 25;
   map<string, uint64> disk_free_bytes = 26;
+
+  // Lets the master check its copy is current without being sent the whole
+  // list. Optional because a server holding no volumes reports 0, which must
+  // stay distinct from a server that computes none.
+  optional uint64 volume_digest = 27;
 }
 
 message HeartbeatResponse {
@@ -115,6 +120,8 @@ message HeartbeatResponse {
   repeated StorageBackend storage_backends = 5;
   repeated string duplicated_uuids = 6;
   bool preallocate = 7;
+  // The reported digest disagrees, so changes alone are not enough.
+  bool resend_full_volume_list = 8;
 }
 
 message VolumeInformationMessage {
@@ -294,6 +301,11 @@ message StatisticsResponse {
   uint64 total_size = 4;
   uint64 used_size = 5;
   uint64 file_count = 6;
+  // sizes counting one copy of the data: a single replica of a regular volume,
+  // the data shards of an ec volume. logical_total_size scales the free space
+  // by the copies the requested replication makes.
+  uint64 logical_total_size = 7;
+  uint64 logical_used_size = 8;
 }
 
 //

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,9 @@ fn build_heartbeat_with_ec_status(
     }
 
     let mut volumes = Vec::new();
+    // Covers exactly what is reported: volumes skipped below 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 +879,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 +896,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 +972,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 +1252,53 @@ mod tests {
         assert!(heartbeat.has_no_volumes);
     }
 
+    // A volume reported but left out of the digest, or the reverse, makes the
+    // master's comparison disagree forever.
+    #[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,95 @@
+//! Mirror of `weed/storage/volume_report_hash.go`. Has to agree with it byte
+//! for byte, so `report_hash_vectors` pins the layout against values produced
+//! there; regenerate them before changing it.
+
+use xxhash_rust::xxh64::xxh64;
+
+use crate::pb::master_pb;
+
+/// 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.
+    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 zero count encodes as zero however the unit is set, as the master stores it.
+fn normalize_ttl(ttl: u32) -> u32 {
+    let count = (ttl >> 8) & 0xff;
+    if count == 0 {
+        return 0;
+    }
+    (count << 8) | (ttl & 0xff)
+}
+
+/// Order-dependent, 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. Drift here means every volume server on
+    // this build silently 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);
+    }
+}

weed/pb/master.proto

@@ -105,6 +105,11 @@ message Heartbeat {
   // physical disk capacity per disk type, in bytes, from the underlying filesystem
   map<string, uint64> disk_total_bytes = 25;
   map<string, uint64> disk_free_bytes = 26;
+
+  // Lets the master check its copy is current without being sent the whole
+  // list. Optional because a server holding no volumes reports 0, which must
+  // stay distinct from a server that computes none.
+  optional uint64 volume_digest = 27;
 }
 
 message HeartbeatResponse {
@@ -115,6 +120,8 @@ message HeartbeatResponse {
   repeated StorageBackend storage_backends = 5;
   repeated string duplicated_uuids = 6;
   bool preallocate = 7;
+  // The reported digest disagrees, so changes alone are not enough.
+  bool resend_full_volume_list = 8;
 }
 
 message VolumeInformationMessage {

weed/pb/master_pb/master.pb.go

@@ -121,8 +121,12 @@ type Heartbeat struct {
 	// physical disk capacity per disk type, in bytes, from the underlying filesystem
 	DiskTotalBytes map[string]uint64 `protobuf:"bytes,25,rep,name=disk_total_bytes,json=diskTotalBytes,proto3" json:"disk_total_bytes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
 	DiskFreeBytes  map[string]uint64 `protobuf:"bytes,26,rep,name=disk_free_bytes,json=diskFreeBytes,proto3" json:"disk_free_bytes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
-	unknownFields  protoimpl.UnknownFields
-	sizeCache      protoimpl.SizeCache
+	// Lets the master check its copy is current without being sent the whole
+	// list. Optional because a server holding no volumes reports 0, which must
+	// stay distinct from a server that computes none.
+	VolumeDigest  *uint64 `protobuf:"varint,27,opt,name=volume_digest,json=volumeDigest,proto3,oneof" json:"volume_digest,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
 }
 
 func (x *Heartbeat) Reset() {
@@ -316,6 +320,13 @@ func (x *Heartbeat) GetDiskFreeBytes() map[string]uint64 {
 	return nil
 }
 
+func (x *Heartbeat) GetVolumeDigest() uint64 {
+	if x != nil && x.VolumeDigest != nil {
+		return *x.VolumeDigest
+	}
+	return 0
+}
+
 type HeartbeatResponse struct {
 	state                  protoimpl.MessageState `protogen:"open.v1"`
 	VolumeSizeLimit        uint64                 `protobuf:"varint,1,opt,name=volume_size_limit,json=volumeSizeLimit,proto3" json:"volume_size_limit,omitempty"`
@@ -325,8 +336,10 @@ type HeartbeatResponse struct {
 	StorageBackends        []*StorageBackend      `protobuf:"bytes,5,rep,name=storage_backends,json=storageBackends,proto3" json:"storage_backends,omitempty"`
 	DuplicatedUuids        []string               `protobuf:"bytes,6,rep,name=duplicated_uuids,json=duplicatedUuids,proto3" json:"duplicated_uuids,omitempty"`
 	Preallocate            bool                   `protobuf:"varint,7,opt,name=preallocate,proto3" json:"preallocate,omitempty"`
-	unknownFields          protoimpl.UnknownFields
-	sizeCache              protoimpl.SizeCache
+	// The reported digest disagrees, so changes alone are not enough.
+	ResendFullVolumeList bool `protobuf:"varint,8,opt,name=resend_full_volume_list,json=resendFullVolumeList,proto3" json:"resend_full_volume_list,omitempty"`
+	unknownFields        protoimpl.UnknownFields
+	sizeCache            protoimpl.SizeCache
 }
 
 func (x *HeartbeatResponse) Reset() {
@@ -408,6 +421,13 @@ func (x *HeartbeatResponse) GetPreallocate() bool {
 	return false
 }
 
+func (x *HeartbeatResponse) GetResendFullVolumeList() bool {
+	if x != nil {
+		return x.ResendFullVolumeList
+	}
+	return false
+}
+
 type VolumeInformationMessage struct {
 	state             protoimpl.MessageState `protogen:"open.v1"`
 	Id                uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
@@ -4569,8 +4589,7 @@ const file_master_proto_rawDesc = "" +
 	"\adisk_id\x18\x01 \x01(\rR\x06diskId\x12\x12\n" +
 	"\x04tags\x18\x02 \x03(\tR\x04tags\x12\x12\n" +
 	"\x04type\x18\x03 \x01(\tR\x04type\x12(\n" +
-	"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xe6\n" +
-	"\n" +
+	"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xa2\v\n" +
 	"\tHeartbeat\x12\x0e\n" +
 	"\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" +
 	"\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" +
@@ -4600,7 +4619,8 @@ const file_master_proto_rawDesc = "" +
 	"\x05state\x18\x17 \x01(\v2#.volume_server_pb.VolumeServerStateR\x05state\x12/\n" +
 	"\tdisk_tags\x18\x18 \x03(\v2\x12.master_pb.DiskTagR\bdiskTags\x12R\n" +
 	"\x10disk_total_bytes\x18\x19 \x03(\v2(.master_pb.Heartbeat.DiskTotalBytesEntryR\x0ediskTotalBytes\x12O\n" +
-	"\x0fdisk_free_bytes\x18\x1a \x03(\v2'.master_pb.Heartbeat.DiskFreeBytesEntryR\rdiskFreeBytes\x1aB\n" +
+	"\x0fdisk_free_bytes\x18\x1a \x03(\v2'.master_pb.Heartbeat.DiskFreeBytesEntryR\rdiskFreeBytes\x12(\n" +
+	"\rvolume_digest\x18\x1b \x01(\x04H\x00R\fvolumeDigest\x88\x01\x01\x1aB\n" +
 	"\x14MaxVolumeCountsEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\x1aA\n" +
@@ -4609,7 +4629,8 @@ const file_master_proto_rawDesc = "" +
 	"\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01\x1a@\n" +
 	"\x12DiskFreeBytesEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
-	"\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01\"\xcd\x02\n" +
+	"\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01B\x10\n" +
+	"\x0e_volume_digest\"\x84\x03\n" +
 	"\x11HeartbeatResponse\x12*\n" +
 	"\x11volume_size_limit\x18\x01 \x01(\x04R\x0fvolumeSizeLimit\x12\x16\n" +
 	"\x06leader\x18\x02 \x01(\tR\x06leader\x12'\n" +
@@ -4617,7 +4638,8 @@ const file_master_proto_rawDesc = "" +
 	"\x18metrics_interval_seconds\x18\x04 \x01(\rR\x16metricsIntervalSeconds\x12D\n" +
 	"\x10storage_backends\x18\x05 \x03(\v2\x19.master_pb.StorageBackendR\x0fstorageBackends\x12)\n" +
 	"\x10duplicated_uuids\x18\x06 \x03(\tR\x0fduplicatedUuids\x12 \n" +
-	"\vpreallocate\x18\a \x01(\bR\vpreallocate\"\xb1\x04\n" +
+	"\vpreallocate\x18\a \x01(\bR\vpreallocate\x125\n" +
+	"\x17resend_full_volume_list\x18\b \x01(\bR\x14resendFullVolumeList\"\xb1\x04\n" +
 	"\x18VolumeInformationMessage\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\rR\x02id\x12\x12\n" +
 	"\x04size\x18\x02 \x01(\x04R\x04size\x12\x1e\n" +
@@ -5203,6 +5225,7 @@ func file_master_proto_init() {
 	if File_master_proto != nil {
 		return
 	}
+	file_master_proto_msgTypes[1].OneofWrappers = []any{}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
 		File: protoimpl.DescBuilder{

weed/server/master_grpc_server.go

@@ -275,9 +275,49 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 		if len(message.NewVids) > 0 || len(message.DeletedVids) > 0 || len(message.NewEcVids) > 0 || len(message.DeletedEcVids) > 0 {
 			ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: message})
 		}
+
+		// After everything the heartbeat carried is applied, so a match means
+		// the master is current, not that nothing changed.
+		if resend := ms.checkVolumeDigest(heartbeat, dn); resend {
+			if err := stream.Send(&master_pb.HeartbeatResponse{ResendFullVolumeList: true}); err != nil {
+				glog.Warningf("SendHeartbeat.Send resend request to %s:%d %v", dn.Ip, dn.Port, err)
+				return err
+			}
+		}
 	}
 }
 
+// checkVolumeDigest reports whether the master needs the full volume list to
+// recover. Servers reporting no digest are left alone; they send it anyway.
+func (ms *MasterServer) checkVolumeDigest(heartbeat *master_pb.Heartbeat, dn *topology.DataNode) bool {
+	if heartbeat.VolumeDigest == nil {
+		return false
+	}
+	// Reported twice but stored once, so no amount of resending can agree.
+	if dn.HasDuplicateVolumeIds() {
+		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestNotComparable").Inc()
+		return false
+	}
+
+	reported := heartbeat.GetVolumeDigest()
+	held := dn.VolumeDigest()
+	if held == reported {
+		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMatch").Inc()
+		return false
+	}
+
+	stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMismatch").Inc()
+	// Nothing further to ask for, so say so rather than repeat the request.
+	if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
+		glog.Warningf("volume server %s reported digest %d after a full volume list, master holds %d",
+			dn.Url(), reported, held)
+		return false
+	}
+	glog.V(0).Infof("volume server %s reported digest %d, master holds %d: requesting the full volume list",
+		dn.Url(), reported, held)
+	return true
+}
+
 // KeepConnected keep a stream gRPC call to the master. Used by clients to know the master is up.
 // And clients gets the up-to-date list of volume locations
 func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServer) error {

weed/server/master_grpc_server_digest_test.go

@@ -0,0 +1,81 @@
+package weed_server
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+func digestTestCluster(t *testing.T) (*MasterServer, *topology.DataNode) {
+	t.Helper()
+	topo := topology.NewTopology("test", sequence.NewMemorySequencer(), 32*1024*1024*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+	return &MasterServer{Topo: topo}, dn
+}
+
+func digestTestVolumeMessage(id uint32) *master_pb.VolumeInformationMessage {
+	return &master_pb.VolumeInformationMessage{
+		Id: id, Size: 1024, Collection: "c", Version: 3,
+	}
+}
+
+// What lets the two sides be upgraded in either order.
+func TestDigestCheckIgnoresServersThatReportNone(t *testing.T) {
+	ms, dn := digestTestCluster(t)
+	ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)}, dn)
+
+	if ms.checkVolumeDigest(&master_pb.Heartbeat{
+		Volumes: []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)},
+	}, dn) {
+		t.Error("a server reporting no digest was asked to resend")
+	}
+}
+
+func TestDigestCheckAcceptsAMatchingReport(t *testing.T) {
+	ms, dn := digestTestCluster(t)
+	volumes := []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1), digestTestVolumeMessage(2)}
+	ms.Topo.SyncDataNodeRegistration(volumes, dn)
+
+	digest := dn.VolumeDigest()
+	if ms.checkVolumeDigest(&master_pb.Heartbeat{Volumes: volumes, VolumeDigest: &digest}, dn) {
+		t.Error("a matching digest was asked to resend")
+	}
+}
+
+func TestDigestCheckDoesNotReaskAfterAFullList(t *testing.T) {
+	ms, dn := digestTestCluster(t)
+	volumes := []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)}
+	ms.Topo.SyncDataNodeRegistration(volumes, dn)
+
+	wrong := dn.VolumeDigest() ^ 1
+	if ms.checkVolumeDigest(&master_pb.Heartbeat{Volumes: volumes, VolumeDigest: &wrong}, dn) {
+		t.Error("a full volume list that still disagreed was asked to resend, which would repeat forever")
+	}
+}
+
+func TestDigestCheckAsksForTheListWhenADeltaDisagrees(t *testing.T) {
+	ms, dn := digestTestCluster(t)
+	ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1)}, dn)
+
+	wrong := dn.VolumeDigest() ^ 1
+	if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &wrong}, dn) {
+		t.Error("a disagreeing digest with no list to fall back on was not asked to resend")
+	}
+}
+
+func TestDigestCheckSkipsNodesWithDuplicateVolumeIds(t *testing.T) {
+	ms, dn := digestTestCluster(t)
+	duplicated := digestTestVolumeMessage(1)
+	duplicated.DiskId = 1
+	ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		digestTestVolumeMessage(1), duplicated,
+	}, dn)
+
+	wrong := dn.VolumeDigest() ^ 1
+	if ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &wrong}, dn) {
+		t.Error("a node the master cannot represent was asked to resend, which would repeat forever")
+	}
+}

weed/storage/store.go

@@ -414,6 +414,9 @@ func (s *Store) GetRack() string {
 
 func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	var volumeMessages []*master_pb.VolumeInformationMessage
+	// Covers exactly what is reported: volumes skipped below are absent from
+	// both the list and the digest.
+	var volumeDigest uint64
 	maxVolumeCounts := make(map[string]uint32)
 	// Per-disk effective max for DiskTag, captured alongside the per-type sum.
 	diskMaxByID := make(map[int]int32)
@@ -487,6 +490,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 			shouldDeleteVolume := false
 			if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
 				volumeMessages = append(volumeMessages, volumeMessage)
+				volumeDigest ^= reportHashOf(volumeMessage)
 			} else {
 				if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
 					deleteVids = append(deleteVids, v.Id)
@@ -594,6 +598,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		DataCenter:      s.dataCenter,
 		Rack:            s.rack,
 		Volumes:         volumeMessages,
+		VolumeDigest:    &volumeDigest,
 		DeletedEcShards: deletedEcVolumes,
 		HasNoVolumes:    len(volumeMessages) == 0,
 		HasNoEcShards:   len(ecVolumeMessages) == 0,
@@ -603,6 +608,17 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 
 }
 
+// Runs the master's own hash over the master's own conversion of the message,
+// so the two ends cannot drift.
+func reportHashOf(m *master_pb.VolumeInformationMessage) uint64 {
+	vi, err := NewVolumeInfo(m)
+	if err != nil {
+		glog.Warningf("volume %d: cannot digest heartbeat report: %v", m.Id, err)
+		return 0
+	}
+	return vi.ReportHash()
+}
+
 func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeEcShardInformationMessage) {
 	for diskId, location := range s.Locations {
 		if location.isDiskUnavailable.Load() {

weed/storage/store_heartbeat_digest_test.go

@@ -0,0 +1,78 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+func mountTestVolume(t *testing.T, loc *DiskLocation, vid needle.VolumeId) {
+	t.Helper()
+	v, err := NewVolume(loc.Directory, loc.IdxDirectory, "", vid, NeedleMapInMemory,
+		&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	loc.SetVolume(vid, v)
+}
+
+// A volume reported but left out of the digest, or the reverse, makes the
+// master's comparison disagree forever.
+func TestCollectHeartbeatDigestsExactlyWhatItReports(t *testing.T) {
+	store := newTestStore(t, 2)
+	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 2)
+	mountTestVolume(t, store.Locations[1], 3)
+
+	heartbeat := store.CollectHeartbeat()
+	if heartbeat.VolumeDigest == nil {
+		t.Fatal("heartbeat carried no digest")
+	}
+	if len(heartbeat.Volumes) != 3 {
+		t.Fatalf("expected 3 volumes reported, got %d", len(heartbeat.Volumes))
+	}
+
+	var want uint64
+	for _, m := range heartbeat.Volumes {
+		vi, err := NewVolumeInfo(m)
+		if err != nil {
+			t.Fatal(err)
+		}
+		want ^= vi.ReportHash()
+	}
+	if got := heartbeat.GetVolumeDigest(); got != want {
+		t.Errorf("digest %d does not cover the reported volumes (%d)", got, want)
+	}
+}
+
+// 0 must stay distinguishable from a server that computes no digest at all.
+func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) {
+	store := newTestStore(t, 1)
+
+	heartbeat := store.CollectHeartbeat()
+	if heartbeat.VolumeDigest == nil {
+		t.Fatal("an empty store still has to report a digest, or the master cannot tell it from an old server")
+	}
+	if got := heartbeat.GetVolumeDigest(); got != 0 {
+		t.Errorf("expected an empty store to digest to 0, got %d", got)
+	}
+	if !heartbeat.HasNoVolumes {
+		t.Error("expected has_no_volumes on an empty store")
+	}
+}
+
+func TestCollectHeartbeatDigestFollowsVolumeChanges(t *testing.T) {
+	store := newTestStore(t, 1)
+	mountTestVolume(t, store.Locations[0], 1)
+	first := store.CollectHeartbeat().GetVolumeDigest()
+
+	if second := store.CollectHeartbeat().GetVolumeDigest(); second != first {
+		t.Errorf("an unchanged store reported a different digest: %d then %d", first, second)
+	}
+
+	mountTestVolume(t, store.Locations[0], 2)
+	if grown := store.CollectHeartbeat().GetVolumeDigest(); grown == first {
+		t.Error("mounting a volume left the digest unchanged")
+	}
+}

weed/topology/volume_digest_test.go

@@ -3,11 +3,16 @@ package topology
 import (
 	"testing"
 
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/reflect/protoreflect"
+
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
 func digestTestNode(t *testing.T) (*Topology, *DataNode) {
@@ -66,46 +71,60 @@ func TestVolumeDigestIsIndependentOfReportOrder(t *testing.T) {
 	}
 }
 
+// Enumerated from the message rather than listed here, so a field added later
+// cannot quietly fall outside the digest while this still passes.
 func TestVolumeDigestTracksEveryReportedField(t *testing.T) {
 	base := digestTestVolume(1)
-	mutations := map[string]func(*master_pb.VolumeInformationMessage){
-		"Id":                func(m *master_pb.VolumeInformationMessage) { m.Id = 2 },
-		"Size":              func(m *master_pb.VolumeInformationMessage) { m.Size++ },
-		"Collection":        func(m *master_pb.VolumeInformationMessage) { m.Collection = "other" },
-		"FileCount":         func(m *master_pb.VolumeInformationMessage) { m.FileCount++ },
-		"DeleteCount":       func(m *master_pb.VolumeInformationMessage) { m.DeleteCount++ },
-		"DeletedByteCount":  func(m *master_pb.VolumeInformationMessage) { m.DeletedByteCount++ },
-		"ReadOnly":          func(m *master_pb.VolumeInformationMessage) { m.ReadOnly = true },
-		"ReplicaPlacement":  func(m *master_pb.VolumeInformationMessage) { m.ReplicaPlacement = 10 },
-		"Version":           func(m *master_pb.VolumeInformationMessage) { m.Version = 2 },
-		"Ttl":               func(m *master_pb.VolumeInformationMessage) { m.Ttl = 3 << 8 },
-		"CompactRevision":   func(m *master_pb.VolumeInformationMessage) { m.CompactRevision++ },
-		"ModifiedAtSecond":  func(m *master_pb.VolumeInformationMessage) { m.ModifiedAtSecond++ },
-		"RemoteStorageName": func(m *master_pb.VolumeInformationMessage) { m.RemoteStorageName = "s3" },
-		"RemoteStorageKey":  func(m *master_pb.VolumeInformationMessage) { m.RemoteStorageKey = "k" },
-		"DiskType":          func(m *master_pb.VolumeInformationMessage) { m.DiskType = "ssd" },
-		"DiskId":            func(m *master_pb.VolumeInformationMessage) { m.DiskId = 1 },
-	}
-
 	baseInfo, err := storage.NewVolumeInfo(base)
 	if err != nil {
 		t.Fatal(err)
 	}
-	for name, mutate := range mutations {
-		t.Run(name, func(t *testing.T) {
-			changed := digestTestVolume(1)
-			mutate(changed)
-			changedInfo, err := storage.NewVolumeInfo(changed)
-			if err != nil {
-				t.Fatal(err)
-			}
-			if baseInfo.ReportHash() == changedInfo.ReportHash() {
-				t.Errorf("a change to %s is invisible to the digest, so the master would never be told about it", name)
+
+	fields := base.ProtoReflect().Descriptor().Fields()
+	for i := 0; i < fields.Len(); i++ {
+		fd := fields.Get(i)
+		t.Run(string(fd.Name()), func(t *testing.T) {
+			candidates := distinctValuesFor(t, fd, base.ProtoReflect().Get(fd))
+			for _, candidate := range candidates {
+				changed := proto.Clone(base).(*master_pb.VolumeInformationMessage)
+				changed.ProtoReflect().Set(fd, candidate)
+				changedInfo, err := storage.NewVolumeInfo(changed)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if baseInfo.ReportHash() != changedInfo.ReportHash() {
+					return
+				}
 			}
+			t.Errorf("no change to %s moves the digest, so the master would never be told about one", fd.Name())
 		})
 	}
 }
 
+// Several, because some fields are narrowed on the way into VolumeInfo and the
+// smallest change to the wire value can land back on the stored one.
+func distinctValuesFor(t *testing.T, fd protoreflect.FieldDescriptor, current protoreflect.Value) []protoreflect.Value {
+	t.Helper()
+	switch fd.Kind() {
+	case protoreflect.BoolKind:
+		return []protoreflect.Value{protoreflect.ValueOfBool(!current.Bool())}
+	case protoreflect.Uint32Kind:
+		return []protoreflect.Value{
+			protoreflect.ValueOfUint32(uint32(current.Uint()) + 1),
+			protoreflect.ValueOfUint32(uint32(current.Uint()) + 1<<8),
+			protoreflect.ValueOfUint32(uint32(current.Uint()) + 1<<16),
+		}
+	case protoreflect.Uint64Kind:
+		return []protoreflect.Value{protoreflect.ValueOfUint64(current.Uint() + 1)}
+	case protoreflect.Int64Kind:
+		return []protoreflect.Value{protoreflect.ValueOfInt64(current.Int() + 1)}
+	case protoreflect.StringKind:
+		return []protoreflect.Value{protoreflect.ValueOfString(current.String() + "x")}
+	}
+	t.Fatalf("field %s has kind %s, which this test does not know how to vary", fd.Name(), fd.Kind())
+	return nil
+}
+
 func TestVolumeDigestFollowsVolumeChanges(t *testing.T) {
 	topo, dn := digestTestNode(t)
 	volumes := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2)}
@@ -164,10 +183,9 @@ func TestVolumeDigestFollowsDeltaRegistration(t *testing.T) {
 	}
 }
 
-// The point of the digest is not to detect that volumes changed -- in any live
-// cluster some always have. It is to confirm that after applying the changes a
-// heartbeat did carry, the master holds what the volume server holds. So a
-// heartbeat reporting only the volumes that moved must still reconcile.
+// The digest is not a change detector: in a live cluster some volumes always
+// have changed. It asks whether the master is current once the heartbeat's own
+// changes are applied.
 func TestVolumeDigestMatchesAfterApplyingOnlyChangedVolumes(t *testing.T) {
 	const total = 50
 	full := make([]*master_pb.VolumeInformationMessage, 0, total)
@@ -210,8 +228,7 @@ func TestVolumeDigestMatchesAfterApplyingOnlyChangedVolumes(t *testing.T) {
 	}
 }
 
-// A volume that disappears without a delta is exactly what the full list exists
-// to catch, and is the case the digest has to keep catching.
+// What the full list exists to catch, and the digest has to keep catching.
 func TestVolumeDigestCatchesASilentlyLostVolume(t *testing.T) {
 	full := []*master_pb.VolumeInformationMessage{
 		digestTestVolume(1), digestTestVolume(2), digestTestVolume(3),
@@ -408,3 +425,41 @@ func TestVolumeIndexDigestFollowsRemovedLookupEntry(t *testing.T) {
 		t.Error("the node that was merely passed in should never have gained the entry")
 	}
 }
+
+// Real heartbeat data, not hand-built messages.
+func TestMasterDigestMatchesWhatAVolumeServerReports(t *testing.T) {
+	dir := t.TempDir()
+	loc := storage.NewDiskLocation(dir, 100, util.MinFreeSpace{}, "", types.HardDriveType, nil,
+		stats.DefaultDiskIOProbeConfig())
+	for _, vid := range []needle.VolumeId{1, 2, 3} {
+		v, err := storage.NewVolume(loc.Directory, loc.IdxDirectory, "", vid, storage.NeedleMapInMemory,
+			&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+		loc.SetVolume(vid, v)
+	}
+
+	reported := make([]*master_pb.VolumeInformationMessage, 0, 3)
+	var serverDigest uint64
+	for _, vid := range []needle.VolumeId{1, 2, 3} {
+		v, _ := loc.FindVolume(vid)
+		_, m := v.ToVolumeInformationMessage()
+		if m == nil {
+			t.Fatalf("volume %d reported nothing", vid)
+		}
+		vi, err := storage.NewVolumeInfo(m)
+		if err != nil {
+			t.Fatal(err)
+		}
+		serverDigest ^= vi.ReportHash()
+		reported = append(reported, m)
+	}
+
+	topo, dn := digestTestNode(t)
+	topo.SyncDataNodeRegistration(reported, dn)
+
+	if got := dn.VolumeDigest(); got != serverDigest {
+		t.Errorf("master digest %d does not match the reporting server's %d", got, serverDigest)
+	}
+}