diff --git a/go.mod b/go.mod
index a680b75d5..bb98743fc 100644
--- a/go.mod
+++ b/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
diff --git a/weed/storage/volume_report_hash.go b/weed/storage/volume_report_hash.go
new file mode 100644
index 000000000..0b1a24543
--- /dev/null
+++ b/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)
+}
diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go
index 8c8a43343..e796fcfb5 100644
--- a/weed/topology/data_node.go
+++ b/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()
diff --git a/weed/topology/disk.go b/weed/topology/disk.go
index 3d42fa7e5..4cc939421 100644
--- a/weed/topology/disk.go
+++ b/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 {
diff --git a/weed/topology/volume_digest_test.go b/weed/topology/volume_digest_test.go
new file mode 100644
index 000000000..9d2ab2585
--- /dev/null
+++ b/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")
+	}
+}
diff --git a/weed/topology/volume_index_digest.go b/weed/topology/volume_index_digest.go
new file mode 100644
index 000000000..2812fd4fa
--- /dev/null
+++ b/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
+		}
+	}
+}
diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go
index 77821c154..32b0ceebf 100644
--- a/weed/topology/volume_layout.go
+++ b/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 {
diff --git a/weed/topology/volume_location_list.go b/weed/topology/volume_location_list.go
index 127ad67eb..be859d3d5 100644
--- a/weed/topology/volume_location_list.go
+++ b/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) {