From 342bbc64bc2c640ae93ef60448b992ce8c7f7fd6 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 7 Aug 2026 12:53:04 -0700 Subject: [PATCH] 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 | 18 +++++ weed/storage/store_heartbeat_digest_test.go | 81 +++++++++++++++++++++ weed/topology/volume_digest_test.go | 42 +++++++++++ 3 files changed, 141 insertions(+) create mode 100644 weed/storage/store_heartbeat_digest_test.go diff --git a/weed/storage/store.go b/weed/storage/store.go index f898d535c..ab6cc6b99 100644 --- a/weed/storage/store.go +++ b/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() { diff --git a/weed/storage/store_heartbeat_digest_test.go b/weed/storage/store_heartbeat_digest_test.go new file mode 100644 index 000000000..c58e9469e --- /dev/null +++ b/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") + } +} diff --git a/weed/topology/volume_digest_test.go b/weed/topology/volume_digest_test.go index 9d2ab2585..dc085df00 100644 --- a/weed/topology/volume_digest_test.go +++ b/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) + } +}