volume: report a digest of the volumes each heartbeat carries

Digests exactly what goes on the wire: volumes skipped as quarantined, phantom
or expired are absent from both the list and the digest, so the master compares
against the same set the server meant to report.

Runs the master's own hash over the master's own conversion of the message, so
the two ends cannot drift into disagreeing about a field.

weed/storage/store.go

@@ -414,6 +414,10 @@ func (s *Store) GetRack() string {
 
 func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	var volumeMessages []*master_pb.VolumeInformationMessage
+	// 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.
+	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 +491,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 +599,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 +609,18 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 
 }
 
+// reportHashOf digests a volume exactly as the master will digest what it
+// stores for that volume, by running the master's own hash over the same
+// conversion the master applies to the message.
+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,81 @@
+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)
+}
+
+// The digest has to 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.
+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)
+	}
+}
+
+// A server holding nothing reports a digest of 0, which is why the field needs
+// explicit presence: it 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

@@ -4,10 +4,12 @@ import (
 	"testing"
 
 	"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) {
@@ -408,3 +410,43 @@ func TestVolumeIndexDigestFollowsRemovedLookupEntry(t *testing.T) {
 		t.Error("the node that was merely passed in should never have gained the entry")
 	}
 }
+
+// The two ends must agree on real heartbeat data, not just on hand-built
+// messages: the volume server hashes what it is about to send, the master
+// hashes what it stored from it.
+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)
+	}
+}