Files
seaweedfs/weed/server/master_grpc_server_digest_test.go
Chris LuandGitHub ce7d388639 heartbeat: send only the volumes that changed (#10640)
* pb: let a heartbeat carry only the volumes that changed

A partial list cannot travel in volumes: a master that did not understand it
would read the absences as deletions. So changes get their own field, used only
once the master has said it compares digests and can tell when it has fallen
behind.

* master: apply the volumes a heartbeat reports as changed

Only the named volumes are touched. A full report says the server holds exactly
these; a changed report says nothing about the ones it leaves out, so absence
must not read as removal.

Also advertises that the master compares digests, which is what lets a server
stop sending its whole list. Advertising it once per connection means a server
reconnecting to a master that does not is back to full lists straight away.

* volume: send only the volumes that changed once the master accepts them

The whole list goes on every heartbeat until the master says it compares
digests, and again whenever it asks, so a master that cannot tell when it has
fallen behind never has to.

has_no_volumes stays derived from a full list alone. Deriving it from what a
heartbeat happens to carry would make a quiet one read as a server that had
lost every volume, and the master would drop them all.

The digest still covers every volume held rather than the ones sent, which is
what lets the master confirm that applying the changes left it current.

Reporting state is per-connection: a server that reconnects, or reaches a
different master, starts again from the full list.

* volume: let the zero reporting state stand for having told no master anything

A Store built as a literal, which tests do, left the reporting state nil and
panicked on the first heartbeat. As a value its zero form already means nothing
has been reported to anyone, which is exactly the state that sends the whole
list.

* rust: send only the volumes that changed once the master accepts them

Mirrors the Go volume server, with one hazard the Go side does not have: mount
and unmount deltas here are derived by diffing successive heartbeats, so a
heartbeat that carries a partial list would report every volume it left out as
unmounted. Collecting now returns the full set alongside the message, and every
site that diffs uses that rather than what went on the wire.

* volume: do not let a full-list request be lost to the heartbeat it raced

The request arrived while a heartbeat was already being built as a delta, and
committing that heartbeat cleared it, so the master waited for another digest
mismatch before asking again. Count the requests and clear only the one the
heartbeat answered.

* rust: stop marking volumes reported by a heartbeat that is thrown away

The state-notify path collected a heartbeat only to diff its volume list, then
sent a delta message of its own and dropped the one it had collected. Once
collecting recorded what the master had been told, every mount or unmount
silently marked the changed volumes as sent, and the master learned of them
only after a digest mismatch.

Snapshotting no longer records anything, and no longer expires ec volumes
whose deletion that path was already discarding.

* master: announce only the volumes a change actually brought

Every changed volume was broadcast as a new location. Volumes grow constantly
and growth moves no location, so on a busy cluster that told every connected
client about volumes it could already reach, filling bounded broadcast queues
and pushing out the topology updates that matter.

* master: ask for the full list when only one can repair the master

Delta heartbeats stop the full report, and with it the only thing that
re-registers a volume the lookup index lost. The volume server cannot see that
divergence and its digest cannot show it, so the master now checks its own two
indexes agree and asks for the list when they do not.

A node reporting one volume id twice is kept on full lists for the same reason
rather than merely skipped: its digest can never be verified, so nothing else
would tell the master what it had stopped holding.

* master: keep the volume options on every heartbeat response

A volume server takes them from whatever response arrives, and preallocate is a
bare bool with no way to tell off from unmentioned. A response sent to ask for
the volume list therefore turned preallocation off until the server reconnected.

Responses sent mid-stream now start from the configured options rather than
being built field by field.

* master: announce a volume the lookup index had lost

Repairing the index makes the volume servable again, but clients were told it
went when the node dropped out and nothing told them otherwise: the disk map
still held it, so it did not count as an arrival.

Reaching the lookup index is what makes a volume servable, so recovering an
entry there is an arrival as far as clients are concerned, on both the full
report and the changed-volume path.
2026-08-07 23:36:28 -07:00

147 lines
6.0 KiB
Go

package weed_server
import (
"testing"
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
"github.com/seaweedfs/seaweedfs/weed/sequence"
"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/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,
}
}
// A volume server that predates the digest keeps sending its whole list and
// must never be asked for anything, whatever the master computes. This is 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")
}
}
// A heartbeat that already carried the whole list has nothing further to give,
// so a mismatch there is a genuine disagreement to report rather than something
// to ask about again.
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")
}
}
// The case the request exists for: a heartbeat carrying no list whose digest
// disagrees means the master has drifted and needs the list back.
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")
}
}
// A node reporting one volume id twice is stored once, so the digests cannot
// agree however often the list is resent.
// A node reporting one volume id twice is stored once, so no digest can ever
// agree. It has to keep sending its whole list: nothing else would tell the
// master what it stopped holding.
func TestDigestCheckKeepsDuplicateNodesOnFullLists(t *testing.T) {
ms, dn := digestTestCluster(t)
duplicated := digestTestVolumeMessage(1)
duplicated.DiskId = 1
ms.Topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
digestTestVolumeMessage(1), duplicated,
}, dn)
digest := dn.VolumeDigest()
if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) {
t.Error("a node whose digest can never be verified was left sending only changes")
}
// And is not asked again for a list it just sent.
if ms.checkVolumeDigest(&master_pb.Heartbeat{
Volumes: []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1), duplicated},
VolumeDigest: &digest,
}, dn) {
t.Error("a node that just sent its whole list was asked for it again")
}
}
// The lookup index can drift from the disks without the volume server seeing
// anything, so its digest still matches. Only a full report re-registers the
// volumes that stopped being servable, and in delta mode nothing else asks for
// one.
func TestDigestCheckAsksForTheListWhenTheLookupIndexDrifts(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{VolumeDigest: &digest}, dn) {
t.Fatal("a healthy node was asked to resend")
}
rp, _ := super_block.NewReplicaPlacementFromString("000")
vl := ms.Topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
vl.SetVolumeUnavailable(dn, needle.VolumeId(1))
if dn.VolumeDigest() != digest {
t.Fatal("expected the reported digest to be unaffected, which is why the index has to be checked")
}
if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) {
t.Error("a volume that stopped being servable left the node sending only changes, so nothing would repair it")
}
}
// A volume server takes the options from every response it receives, and
// preallocate is a bare bool with no way to tell "off" from "not mentioned". A
// response that left it out would turn preallocation off until reconnect.
func TestHeartbeatResponsesCarryTheVolumeOptions(t *testing.T) {
ms := &MasterServer{option: &MasterOption{VolumeSizeLimitMB: 1024}, preallocateSize: 1}
resend := ms.heartbeatResponse()
resend.ResendFullVolumeList = true
if !resend.Preallocate {
t.Error("a resend request would turn off preallocation on the volume server")
}
if resend.VolumeSizeLimit != 1024*1024*1024 {
t.Errorf("a resend request carried volume size limit %d, want the configured one", resend.VolumeSizeLimit)
}
}