topology: digest the volumes a master believes each node holds (#10619)

* topology: digest the volumes a master believes each node holds

A volume server resends its whole volume list every heartbeat because that list
is the only way the master can notice a volume that vanished without a delta.
A digest gives the master the same guarantee without the list: the two ends
agree iff the master's copy is current.

VolumeInfo.ReportHash covers every field of VolumeInformationMessage, so a
change the hash misses is a change the master would never hear about. Both ends
run it over the same converted VolumeInfo, so they cannot drift apart.

Disk keeps the xor of its volumes' hashes, which is order-independent and its
own inverse, so add, update and remove each stay O(1) and the running value
needs no per-volume storage.

Nothing reads the digest yet; the heartbeat protocol change comes next.

* topology: test that a changed-volumes-only heartbeat reconciles

The digest is not a change detector -- in a live cluster some volumes always
have changed. It answers whether the master holds what the volume server holds
once the heartbeat's own changes are applied, so reporting three volumes out of
fifty has to reconcile while a volume lost without a delta must not.

* topology: digest the lookup index too, not just the disk maps

The reported digest answers whether the master holds what the volume server
holds. It cannot answer whether the master can serve those volumes: the disk
map and the lookup index are maintained separately, and a disconnect racing a
reconnect drops a volume from the index while leaving it on the node. The
server's report is identical either way, so a digest built from the disk maps
alone matches while the volume answers 'volume id not found'.

Track a second digest over volume ids on both sides of that split, so the
master can see its own indexes disagree without the volume server's help, and
without the O(volumes) scan the full heartbeat currently relies on.

* topology: exclude nodes reporting a duplicate volume id from the digest

A volume id can end up mounted on two disks of one server -- a stale twin
re-attached after a disk repair, which the store handles rather than rejects.
The server reports both copies with different disk ids, but the master keys
volumes by id alone within a disk type and keeps only the last one. Its digest
can then never equal the server's, and no amount of resending the full list
would fix it.

Detect it from the report itself, where deduplicating the ids already tells us
the count, and mark the node. A marked node has to keep sending full lists;
representing both copies is a separate question, and nesting the volume map by
disk id would cost more memory than the digest saves.

* topology: move the lookup digest with the entry, not the node passed in

Two volume servers can hold one address: GetOrCreateDataNode keys on the id a
server reports and refuses to merge a new id onto an address an older node
still claims, while the lookup list keys on address alone. Registering the
second server therefore displaces the first from the entry, and unregistering
through either removes whichever node the entry named.

Crediting the node handed to Set and Remove instead of the one actually
displaced or removed left the digest on the wrong node. A displaced node went
on reporting a consistent index while it could no longer serve the volume,
which is exactly the silent unavailability the digest exists to catch.

Set and Remove now return the node they displaced and removed, so ownership
can be transferred rather than assumed.

go.mod

@@ -11,7 +11,6 @@ require (
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bwmarrin/snowflake v0.3.0
 	github.com/cenkalti/backoff/v4 v4.3.0
-	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/coreos/go-semver v0.3.1 // indirect
 	github.com/coreos/go-systemd/v22 v22.6.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
@@ -127,6 +126,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.32.33
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.32
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.105.2
+	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/cognusion/imaging v1.0.4
 	github.com/fluent/fluent-logger-golang v1.10.1
 	github.com/getsentry/sentry-go v0.44.1

weed/storage/volume_report_hash.go

@@ -0,0 +1,50 @@
+package storage
+
+import (
+	"encoding/binary"
+
+	"github.com/cespare/xxhash/v2"
+)
+
+// ReportHash digests everything a volume server reports about a volume, so the
+// two ends of a heartbeat can agree on whether the master's copy is current
+// without shipping the volume list.
+//
+// It must cover every field of VolumeInformationMessage: a change the hash
+// misses is a change the master would never be told about. Volume servers hash
+// the message they are about to send, masters hash what they already hold, and
+// the two match only when the master is up to date.
+func (vi VolumeInfo) ReportHash() uint64 {
+	var buf [57]byte
+	binary.LittleEndian.PutUint32(buf[0:], uint32(vi.Id))
+	binary.LittleEndian.PutUint64(buf[4:], vi.Size)
+	binary.LittleEndian.PutUint64(buf[12:], uint64(vi.FileCount))
+	binary.LittleEndian.PutUint64(buf[20:], uint64(vi.DeleteCount))
+	binary.LittleEndian.PutUint64(buf[28:], vi.DeletedByteCount)
+	binary.LittleEndian.PutUint32(buf[36:], uint32(vi.ReplicaPlacement.Byte()))
+	binary.LittleEndian.PutUint32(buf[40:], uint32(vi.Version))
+	binary.LittleEndian.PutUint32(buf[44:], vi.Ttl.ToUint32())
+	binary.LittleEndian.PutUint32(buf[48:], vi.CompactRevision)
+	binary.LittleEndian.PutUint32(buf[52:], vi.DiskId)
+	if vi.ReadOnly {
+		buf[56] = 1
+	}
+	h := xxhash.Sum64(buf[:])
+
+	var modified [8]byte
+	binary.LittleEndian.PutUint64(modified[:], uint64(vi.ModifiedAtSecond))
+	h = foldReportHash(h, xxhash.Sum64(modified[:]))
+	h = foldReportHash(h, xxhash.Sum64String(vi.Collection))
+	h = foldReportHash(h, xxhash.Sum64String(vi.DiskType))
+	h = foldReportHash(h, xxhash.Sum64String(vi.RemoteStorageName))
+	h = foldReportHash(h, xxhash.Sum64String(vi.RemoteStorageKey))
+	return h
+}
+
+// foldReportHash combines two hashes order-dependently, so swapping two string
+// fields is not invisible.
+func foldReportHash(h, x uint64) uint64 {
+	h ^= x
+	h *= 0x9E3779B97F4A7C15
+	return h ^ (h >> 29)
+}

weed/topology/data_node.go

@@ -24,6 +24,12 @@ type DataNode struct {
 	IsTerminating bool
 
 	MaintenanceMode bool
+	// lookupDigest covers the volumes reachable through this node in the volume
+	// layouts, for comparison against what its disks actually hold.
+	lookupDigest atomic.Uint64
+	// duplicateVolumeIds records that the node last reported one volume id more
+	// than once, which the master cannot represent.
+	duplicateVolumeIds atomic.Bool
 	// diskMetas holds each physical disk's tags, type, and capacity from the
 	// heartbeat DiskTags, including disks with no volumes or EC shards.
 	diskMetas map[uint32]diskMeta
@@ -82,6 +88,13 @@ func (dn *DataNode) UpdateVolumes(actualVolumes []storage.VolumeInfo) (newVolume
 		actualVolumeIds[v.Id] = struct{}{}
 	}
 
+	// A volume id mounted on two disks of one server -- a stale twin re-attached
+	// after a disk repair -- is reported twice, but the master keys volumes by
+	// id alone and keeps only the last copy. Its digest can then never equal the
+	// server's however often the list is resent, so record it and let the
+	// heartbeat fall back to the full list for this node.
+	dn.duplicateVolumeIds.Store(len(actualVolumeIds) < len(actualVolumes))
+
 	dn.Lock()
 	defer dn.Unlock()
 
@@ -211,6 +224,25 @@ func (dn *DataNode) GetVolumes() (ret []storage.VolumeInfo) {
 	return ret
 }
 
+// HasDuplicateVolumeIds reports whether the node's last full report named one
+// volume id more than once. While it does, the node's digest is not meaningful.
+func (dn *DataNode) HasDuplicateVolumeIds() bool {
+	return dn.duplicateVolumeIds.Load()
+}
+
+// VolumeDigest summarises every volume the master believes this node holds. A
+// volume server that reports a different digest has drifted from the master and
+// needs to resend its volume list.
+func (dn *DataNode) VolumeDigest() uint64 {
+	dn.RLock()
+	defer dn.RUnlock()
+	var digest uint64
+	for _, c := range dn.children {
+		digest ^= c.(*Disk).VolumeDigest()
+	}
+	return digest
+}
+
 func (dn *DataNode) GetVolumesById(id needle.VolumeId) (vInfo storage.VolumeInfo, err error) {
 	dn.RLock()
 	defer dn.RUnlock()

weed/topology/disk.go

@@ -26,6 +26,15 @@ type Disk struct {
 	// outer key is the volume id; the inner key is the physical disk id.
 	ecShards     map[needle.VolumeId]map[types.DiskId]*erasure_coding.EcVolumeInfo
 	ecShardsLock sync.RWMutex
+	// volumeDigest is the xor of every volume's ReportHash. Order-independent
+	// and its own inverse, so it stays current by xoring a volume out before
+	// its old state is dropped and back in after the new one lands.
+	volumeDigest uint64
+	// volumeIdDigest covers which volumes are on the disk, ignoring their
+	// state, so it can be compared against the lookup index the master serves
+	// reads from. The two indexes are maintained separately and have been seen
+	// to drift.
+	volumeIdDigest uint64
 }
 
 // ecShardSlots returns the number of volume slots consumed by the given
@@ -163,6 +172,8 @@ func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool)
 	deltaDiskUsage := &DiskUsageCounts{}
 	if oldV, ok := d.volumes[v.Id]; !ok {
 		d.volumes[v.Id] = v
+		d.volumeDigest ^= v.ReportHash()
+		d.volumeIdDigest ^= VolumeIdDigestHash(v.Id)
 		deltaDiskUsage.volumeCount = 1
 		if v.IsRemote() {
 			deltaDiskUsage.remoteVolumeCount = 1
@@ -183,6 +194,7 @@ func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChanged bool)
 			}
 			d.UpAdjustDiskUsageDelta(types.ToDiskType(v.DiskType), deltaDiskUsage)
 		}
+		d.volumeDigest ^= oldV.ReportHash() ^ v.ReportHash()
 		isChanged = d.volumes[v.Id].ReadOnly != v.ReadOnly
 		if isChanged {
 			// Adjust active volume count when ReadOnly status changes
@@ -233,6 +245,8 @@ func (d *Disk) RemoveVolumesNotIn(keep map[needle.VolumeId]struct{}) (removed []
 		if _, ok := keep[vid]; !ok {
 			removed = append(removed, v)
 			delete(d.volumes, vid)
+			d.volumeDigest ^= v.ReportHash()
+			d.volumeIdDigest ^= VolumeIdDigestHash(vid)
 		}
 	}
 	return removed
@@ -252,7 +266,25 @@ func (d *Disk) GetVolumesById(id needle.VolumeId) (storage.VolumeInfo, error) {
 func (d *Disk) DeleteVolumeById(id needle.VolumeId) {
 	d.Lock()
 	defer d.Unlock()
-	delete(d.volumes, id)
+	if v, ok := d.volumes[id]; ok {
+		d.volumeDigest ^= v.ReportHash()
+		d.volumeIdDigest ^= VolumeIdDigestHash(id)
+		delete(d.volumes, id)
+	}
+}
+
+// VolumeDigest returns the disk's running volume digest.
+func (d *Disk) VolumeDigest() uint64 {
+	d.RLock()
+	defer d.RUnlock()
+	return d.volumeDigest
+}
+
+// VolumeIdDigest returns the digest of which volumes the disk holds.
+func (d *Disk) VolumeIdDigest() uint64 {
+	d.RLock()
+	defer d.RUnlock()
+	return d.volumeIdDigest
 }
 
 func (d *Disk) GetDataCenter() *DataCenter {

weed/topology/volume_digest_test.go

@@ -0,0 +1,410 @@
+package topology
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"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"
+)
+
+func digestTestNode(t *testing.T) (*Topology, *DataNode) {
+	t.Helper()
+	topo := NewTopology("digest", nil, 32*1024*1024*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 1000})
+	return topo, dn
+}
+
+func digestTestVolume(id uint32) *master_pb.VolumeInformationMessage {
+	return &master_pb.VolumeInformationMessage{
+		Id:               id,
+		Size:             1024 * 1024,
+		Collection:       "c",
+		FileCount:        10,
+		DeleteCount:      1,
+		DeletedByteCount: 128,
+		ReplicaPlacement: 0,
+		Version:          3,
+		CompactRevision:  1,
+		ModifiedAtSecond: 1700000000,
+	}
+}
+
+func TestVolumeDigestIsStableAcrossRepeatedHeartbeats(t *testing.T) {
+	topo, dn := digestTestNode(t)
+	volumes := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2), digestTestVolume(3)}
+
+	topo.SyncDataNodeRegistration(volumes, dn)
+	first := dn.VolumeDigest()
+	if first == 0 {
+		t.Fatal("expected a non-zero digest for a node holding volumes")
+	}
+	for i := 0; i < 3; i++ {
+		topo.SyncDataNodeRegistration(volumes, dn)
+		if got := dn.VolumeDigest(); got != first {
+			t.Fatalf("heartbeat %d changed the digest with no change to report: %d != %d", i, got, first)
+		}
+	}
+}
+
+func TestVolumeDigestIsIndependentOfReportOrder(t *testing.T) {
+	topoA, dnA := digestTestNode(t)
+	topoA.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		digestTestVolume(1), digestTestVolume(2), digestTestVolume(3),
+	}, dnA)
+
+	topoB, dnB := digestTestNode(t)
+	topoB.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		digestTestVolume(3), digestTestVolume(1), digestTestVolume(2),
+	}, dnB)
+
+	if dnA.VolumeDigest() != dnB.VolumeDigest() {
+		t.Errorf("digest depends on report order: %d != %d", dnA.VolumeDigest(), dnB.VolumeDigest())
+	}
+}
+
+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)
+			}
+		})
+	}
+}
+
+func TestVolumeDigestFollowsVolumeChanges(t *testing.T) {
+	topo, dn := digestTestNode(t)
+	volumes := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2)}
+	topo.SyncDataNodeRegistration(volumes, dn)
+	original := dn.VolumeDigest()
+
+	grown := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2)}
+	grown[1].Size += 4096
+	topo.SyncDataNodeRegistration(grown, dn)
+	if dn.VolumeDigest() == original {
+		t.Error("a volume that grew left the digest unchanged")
+	}
+
+	topo.SyncDataNodeRegistration(volumes, dn)
+	if dn.VolumeDigest() != original {
+		t.Error("reverting a volume did not restore the digest")
+	}
+
+	topo.SyncDataNodeRegistration(volumes[:1], dn)
+	if dn.VolumeDigest() == original {
+		t.Error("dropping a volume left the digest unchanged")
+	}
+
+	topo.SyncDataNodeRegistration(volumes, dn)
+	if dn.VolumeDigest() != original {
+		t.Error("restoring a dropped volume did not restore the digest")
+	}
+}
+
+func TestVolumeDigestEmptiesWithTheNode(t *testing.T) {
+	topo, dn := digestTestNode(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		digestTestVolume(1), digestTestVolume(2),
+	}, dn)
+
+	topo.SyncDataNodeRegistration(nil, dn)
+	if got := dn.VolumeDigest(); got != 0 {
+		t.Errorf("expected an empty node to digest to 0, got %d", got)
+	}
+}
+
+func TestVolumeDigestFollowsDeltaRegistration(t *testing.T) {
+	topo, dn := digestTestNode(t)
+	full := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2)}
+	topo.SyncDataNodeRegistration(full, dn)
+	both := dn.VolumeDigest()
+
+	topo.IncrementalSyncDataNodeRegistration(nil, []*master_pb.VolumeShortInformationMessage{{Id: 2}}, dn)
+	if dn.VolumeDigest() == both {
+		t.Error("unmounting a volume left the digest unchanged")
+	}
+
+	topo.SyncDataNodeRegistration(full, dn)
+	if dn.VolumeDigest() != both {
+		t.Error("a full heartbeat did not restore the digest after an unmount")
+	}
+}
+
+// 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.
+func TestVolumeDigestMatchesAfterApplyingOnlyChangedVolumes(t *testing.T) {
+	const total = 50
+	full := make([]*master_pb.VolumeInformationMessage, 0, total)
+	for i := 1; i <= total; i++ {
+		v := digestTestVolume(uint32(i))
+		v.ReadOnly = i > 5 // only the first few are writable, as in a tiered cluster
+		full = append(full, v)
+	}
+
+	topo, dn := digestTestNode(t)
+	topo.SyncDataNodeRegistration(full, dn)
+
+	// Three writable volumes take writes between two heartbeats.
+	changed := make([]storage.VolumeInfo, 0, 3)
+	for _, v := range full[:3] {
+		v.Size += 4096
+		v.FileCount++
+		v.ModifiedAtSecond += 5
+		vi, err := storage.NewVolumeInfo(v)
+		if err != nil {
+			t.Fatal(err)
+		}
+		changed = append(changed, vi)
+	}
+
+	// What the volume server would now report for its whole set.
+	reference, referenceNode := digestTestNode(t)
+	reference.SyncDataNodeRegistration(full, referenceNode)
+	want := referenceNode.VolumeDigest()
+
+	if dn.VolumeDigest() == want {
+		t.Fatal("expected the master to be behind before the changes are applied")
+	}
+
+	// The heartbeat carries three volumes, not fifty.
+	dn.DeltaUpdateVolumes(changed, nil)
+
+	if got := dn.VolumeDigest(); got != want {
+		t.Errorf("digest still disagrees after applying the reported changes: %d != %d", got, want)
+	}
+}
+
+// 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.
+func TestVolumeDigestCatchesASilentlyLostVolume(t *testing.T) {
+	full := []*master_pb.VolumeInformationMessage{
+		digestTestVolume(1), digestTestVolume(2), digestTestVolume(3),
+	}
+
+	topo, dn := digestTestNode(t)
+	topo.SyncDataNodeRegistration(full, dn)
+
+	// The volume server no longer has volume 2 and never got to say so.
+	reference, referenceNode := digestTestNode(t)
+	reference.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{full[0], full[2]}, referenceNode)
+
+	if dn.VolumeDigest() == referenceNode.VolumeDigest() {
+		t.Error("a volume lost without a delta went undetected, which is what the full list is for")
+	}
+}
+
+// The disk map and the lookup index are maintained separately, and a disconnect
+// racing a reconnect has been seen to drop a volume from the lookup index while
+// leaving it on the node. The volume server's report is identical either way, so
+// the heartbeat digest cannot see it and the master has to notice on its own.
+func TestVolumeIndexDigestSeesLookupDivergence(t *testing.T) {
+	topo, dn := digestTestNode(t)
+	v := digestTestVolume(1)
+	v.Collection = "drr"
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn)
+
+	if !dn.HasConsistentVolumeIndex() {
+		t.Fatal("a freshly registered node should have a consistent index")
+	}
+	reported := dn.VolumeDigest()
+
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vl := topo.GetVolumeLayout("drr", rp, needle.EMPTY_TTL, types.HardDriveType)
+	vl.SetVolumeUnavailable(dn, needle.VolumeId(1))
+
+	if got := topo.Lookup("drr", needle.VolumeId(1)); got != nil {
+		t.Fatalf("expected the volume to have become unservable, got %v", got)
+	}
+	if _, err := dn.GetVolumesById(needle.VolumeId(1)); err != nil {
+		t.Fatalf("the volume should still be on the node: %v", err)
+	}
+	if dn.VolumeDigest() != reported {
+		t.Error("the reported digest should not move: the volume server sees no change")
+	}
+	if dn.HasConsistentVolumeIndex() {
+		t.Error("a volume held but not servable left the index digests agreeing, so nothing would repair it")
+	}
+
+	// The full heartbeat self-heal puts it back.
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn)
+	if !dn.HasConsistentVolumeIndex() {
+		t.Error("the self-heal did not restore index consistency")
+	}
+}
+
+func TestVolumeIndexDigestFollowsNodeLifecycle(t *testing.T) {
+	topo, dn := digestTestNode(t)
+	full := []*master_pb.VolumeInformationMessage{digestTestVolume(1), digestTestVolume(2), digestTestVolume(3)}
+	topo.SyncDataNodeRegistration(full, dn)
+	if !dn.HasConsistentVolumeIndex() {
+		t.Fatal("registration left the indexes disagreeing")
+	}
+
+	topo.SyncDataNodeRegistration(full[:2], dn)
+	if !dn.HasConsistentVolumeIndex() {
+		t.Error("dropping a volume left the indexes disagreeing")
+	}
+
+	topo.IncrementalSyncDataNodeRegistration(
+		[]*master_pb.VolumeShortInformationMessage{{Id: 9}}, nil, dn)
+	if !dn.HasConsistentVolumeIndex() {
+		t.Error("a mount delta left the indexes disagreeing")
+	}
+
+	topo.IncrementalSyncDataNodeRegistration(
+		nil, []*master_pb.VolumeShortInformationMessage{{Id: 9}}, dn)
+	if !dn.HasConsistentVolumeIndex() {
+		t.Error("an unmount delta left the indexes disagreeing")
+	}
+
+	topo.UnRegisterDataNode(dn)
+	held, servable := dn.VolumeIndexDigests()
+	if held != 0 || servable != 0 {
+		t.Errorf("an unregistered node should hold nothing: held=%d servable=%d", held, servable)
+	}
+}
+
+// A volume id mounted on two disks of one server is reported twice with
+// different disk ids, but the master keys volumes by id alone, so it keeps only
+// one copy and its digest can never equal the server's. Resending the full list
+// cannot fix that, so the node has to be excluded from digest comparison
+// entirely rather than resend forever.
+func TestVolumeDigestRefusesDuplicateVolumeIds(t *testing.T) {
+	topo, dn := digestTestNode(t)
+
+	first := digestTestVolume(1)
+	second := digestTestVolume(1)
+	second.DiskId = 1
+	second.Size = first.Size * 2
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{first, second}, dn)
+	if !dn.HasDuplicateVolumeIds() {
+		t.Fatal("a volume id reported twice went unnoticed, so the digest would be trusted and never reconcile")
+	}
+
+	firstInfo, err := storage.NewVolumeInfo(first)
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondInfo, err := storage.NewVolumeInfo(second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if dn.VolumeDigest() == firstInfo.ReportHash()^secondInfo.ReportHash() {
+		t.Error("expected the master to be unable to represent both copies; if it now can, the guard is no longer needed")
+	}
+
+	// Once the stale twin is gone the node is comparable again.
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{first}, dn)
+	if dn.HasDuplicateVolumeIds() {
+		t.Error("the node stayed marked as duplicated after reporting a clean list")
+	}
+	if dn.VolumeDigest() != firstInfo.ReportHash() {
+		t.Error("digest did not settle on the surviving copy")
+	}
+}
+
+// Two volume servers can hold one address. GetOrCreateDataNode keys on the id a
+// server reports and deliberately refuses to merge a new id onto an address an
+// older node still claims, while the lookup list keys on address alone -- so
+// registering the second server displaces the first from the lookup entry
+// without either node being told. The digest has to follow the entry, not the
+// node that was passed in.
+func addressSharingNodes(t *testing.T) (*Topology, *DataNode, *DataNode) {
+	t.Helper()
+	topo := NewTopology("digest", nil, 32*1024*1024*1024, 5, false)
+	rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
+	counts := map[string]uint32{"": 1000}
+	old := rack.GetOrCreateDataNode("10.1.2.3", 8080, 18080, "", "n1", counts)
+	fresh := rack.GetOrCreateDataNode("10.1.2.3", 8080, 18080, "", "n2", counts)
+	if old == fresh {
+		t.Skip("address reuse no longer produces two nodes")
+	}
+	return topo, old, fresh
+}
+
+func TestVolumeIndexDigestFollowsDisplacedLookupEntry(t *testing.T) {
+	topo, old, fresh := addressSharingNodes(t)
+	v := digestTestVolume(1)
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, old)
+	if !old.HasConsistentVolumeIndex() {
+		t.Fatal("the first node should be consistent before it is displaced")
+	}
+
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, fresh)
+
+	servable := map[*DataNode]bool{}
+	for _, dn := range topo.Lookup("", needle.VolumeId(1)) {
+		servable[dn] = true
+	}
+	if servable[old] || !servable[fresh] {
+		t.Fatalf("expected the lookup entry to move to the new node, got old=%v fresh=%v",
+			servable[old], servable[fresh])
+	}
+
+	if old.HasConsistentVolumeIndex() {
+		t.Error("the displaced node still holds the volume and can no longer serve it, so its index must read as inconsistent")
+	}
+	if !fresh.HasConsistentVolumeIndex() {
+		t.Error("the node the lookup entry now names reads as inconsistent")
+	}
+}
+
+func TestVolumeIndexDigestFollowsRemovedLookupEntry(t *testing.T) {
+	topo, old, fresh := addressSharingNodes(t)
+	v := digestTestVolume(1)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, old)
+
+	// fresh shares old's address, so unregistering through it removes old's
+	// lookup entry. The digest must come off the node that was actually removed.
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
+	vl.SetVolumeUnavailable(fresh, needle.VolumeId(1))
+
+	if got := topo.Lookup("c", needle.VolumeId(1)); got != nil {
+		t.Fatalf("expected the lookup entry to be gone, got %v", got)
+	}
+	if old.HasConsistentVolumeIndex() {
+		t.Error("the removed node still holds the volume and cannot serve it, so its index must read as inconsistent")
+	}
+	if _, servable := fresh.VolumeIndexDigests(); servable != 0 {
+		t.Error("the node that was merely passed in should never have gained the entry")
+	}
+}

weed/topology/volume_index_digest.go

@@ -0,0 +1,69 @@
+package topology
+
+import (
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// VolumeIdDigestHash spreads a volume id across the whole 64-bit range so that
+// xoring a set of them together does not collide for the near-consecutive ids a
+// cluster actually allocates.
+func VolumeIdDigestHash(vid needle.VolumeId) uint64 {
+	h := uint64(vid) + 0x9E3779B97F4A7C15
+	h = (h ^ (h >> 30)) * 0xBF58476D1CE4E5B9
+	h = (h ^ (h >> 27)) * 0x94D049BB133111EB
+	return h ^ (h >> 31)
+}
+
+// VolumeIndexDigests returns the digest of the volumes the node's disks hold and
+// the digest of the volumes the lookup index will serve from that node.
+//
+// The master keeps those two indexes separately, and a disconnect racing a
+// reconnect has been seen to drop a volume from the lookup index while leaving
+// it on the node: the volume stays visible in volume.list and the admin UI while
+// LookupVolume answers "volume id not found". A volume server cannot detect that
+// -- its own report is identical either way -- so the master has to notice it
+// on its own rather than rely on the heartbeat digest.
+func (dn *DataNode) VolumeIndexDigests() (held, servable uint64) {
+	dn.RLock()
+	for _, c := range dn.children {
+		held ^= c.(*Disk).VolumeIdDigest()
+	}
+	dn.RUnlock()
+	return held, dn.lookupDigest.Load()
+}
+
+// HasConsistentVolumeIndex reports whether every volume on the node is reachable
+// through the lookup index, and nothing else is.
+func (dn *DataNode) HasConsistentVolumeIndex() bool {
+	held, servable := dn.VolumeIndexDigests()
+	return held == servable
+}
+
+// moveLookupOwnership transfers the digest bit for vid from the node a lookup
+// entry used to name to the node it now names. Either may be nil, for an entry
+// being created or dropped.
+func moveLookupOwnership(vid needle.VolumeId, from, to *DataNode) {
+	if from == to {
+		return
+	}
+	if from != nil {
+		from.trackLookupChange(vid)
+	}
+	if to != nil {
+		to.trackLookupChange(vid)
+	}
+}
+
+// trackLookupChange records that vid became reachable through this node, or
+// stopped being: xor is its own inverse, so both are the same update. Volume
+// layouts for different collections share the node, so it cannot assume the
+// caller's lock.
+func (dn *DataNode) trackLookupChange(vid needle.VolumeId) {
+	delta := VolumeIdDigestHash(vid)
+	for {
+		old := dn.lookupDigest.Load()
+		if dn.lookupDigest.CompareAndSwap(old, old^delta) {
+			return
+		}
+	}
+}

weed/topology/volume_layout.go

@@ -196,7 +196,7 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 
 	defer vl.rememberOversizedVolume(v, dn)
 
-	vl.getOrCreateLocationList(v.Id).Set(dn)
+	moveLookupOwnership(v.Id, vl.getOrCreateLocationList(v.Id).Set(dn), dn)
 	vl.initSizeTracking(v.Id, v.Size, v.CompactRevision)
 	// glog.V(4).Infof("volume %d added to %s len %d copy %d", v.Id, dn.Id(), vl.vid2location[v.Id].Length(), v.ReplicaPlacement.GetCopyCount())
 	for _, dn := range vl.vid2location[v.Id].list {
@@ -315,7 +315,8 @@ func (vl *VolumeLayout) UnRegisterVolume(v *storage.VolumeInfo, dn *DataNode) {
 		return
 	}
 
-	if location.Remove(dn) {
+	if removed := location.Remove(dn); removed != nil {
+		moveLookupOwnership(v.Id, removed, nil)
 
 		vl.readonlyVolumes.Remove(v.Id, dn)
 		vl.oversizedVolumes.Remove(v.Id, dn)
@@ -879,7 +880,8 @@ func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid needle.VolumeId)
 	defer vl.accessLock.Unlock()
 
 	if location, ok := vl.vid2location[vid]; ok {
-		if location.Remove(dn) {
+		if removed := location.Remove(dn); removed != nil {
+			moveLookupOwnership(vid, removed, nil)
 			vl.readonlyVolumes.Remove(vid, dn)
 			vl.oversizedVolumes.Remove(vid, dn)
 			wasWritable := false
@@ -919,7 +921,7 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is
 	// A disconnect during a long vacuum can drop the entry while the volume is
 	// still on the node; re-create it (and seed size tracking) instead of
 	// dereferencing a nil location, so the commit also repairs the split.
-	vl.getOrCreateLocationList(vid).Set(dn)
+	moveLookupOwnership(vid, vl.getOrCreateLocationList(vid).Set(dn), dn)
 	vl.initSizeTracking(vid, vInfo.Size, vInfo.CompactRevision)
 
 	if vInfo.ReadOnly || isReadOnly || isFullCapacity {

weed/topology/volume_location_list.go

@@ -46,24 +46,35 @@ func (dnll *VolumeLocationList) Length() int {
 	return len(dnll.list)
 }
 
-func (dnll *VolumeLocationList) Set(loc *DataNode) {
+// Set adds loc, or replaces the entry at the same address, returning the node it
+// displaced. Two volume servers can share an address -- GetOrCreateDataNode keys
+// on the reported id and refuses to merge a new id onto an address an older node
+// still claims -- so the displaced node is not necessarily loc, and callers
+// tracking which node a volume is reachable through must move their bookkeeping
+// off it rather than assume loc already owned the entry.
+func (dnll *VolumeLocationList) Set(loc *DataNode) (displaced *DataNode) {
 	for i := 0; i < len(dnll.list); i++ {
 		if loc.Ip == dnll.list[i].Ip && loc.Port == dnll.list[i].Port {
+			displaced = dnll.list[i]
 			dnll.list[i] = loc
-			return
+			return displaced
 		}
 	}
 	dnll.list = append(dnll.list, loc)
+	return nil
 }
 
-func (dnll *VolumeLocationList) Remove(loc *DataNode) bool {
+// Remove drops the entry at loc's address and returns the node removed, or nil
+// if the volume was not reachable there. As with Set, the removed node is
+// matched by address and need not be loc.
+func (dnll *VolumeLocationList) Remove(loc *DataNode) (removed *DataNode) {
 	for i, dnl := range dnll.list {
 		if loc.Ip == dnl.Ip && loc.Port == dnl.Port {
 			dnll.list = append(dnll.list[:i], dnll.list[i+1:]...)
-			return true
+			return dnl
 		}
 	}
-	return false
+	return nil
 }
 
 func (dnll *VolumeLocationList) Refresh(freshThreshHold int64) {