diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go
index 2b37256f7a..4678c227d5 100644
--- a/weed/stats/metrics.go
+++ b/weed/stats/metrics.go
@@ -109,6 +109,14 @@ var (
 			Help:      "1 while volume growth is frozen after detected master state loss",
 		})
 
+	MasterDuplicateVolumeIds = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: Namespace,
+			Subsystem: subsystemMaster,
+			Name:      "duplicate_volume_ids",
+			Help:      "volume id registered under two different layouts",
+		}, []string{"collection", "volumeId"})
+
 	MasterReceivedHeartbeatCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: Namespace,
@@ -1000,6 +1008,7 @@ func init() {
 	Gather.MustRegister(MasterRaftIsleader)
 	Gather.MustRegister(MasterAdminLock)
 	Gather.MustRegister(MasterVolumeGrowthFrozen)
+	Gather.MustRegister(MasterDuplicateVolumeIds)
 	Gather.MustRegister(MasterReceivedHeartbeatCounter)
 	Gather.MustRegister(MasterLeaderChangeCounter)
 	Gather.MustRegister(MasterReplicaPlacementMismatch)
diff --git a/weed/topology/duplicate_volume_id_test.go b/weed/topology/duplicate_volume_id_test.go
new file mode 100644
index 0000000000..3c0bf5cfc1
--- /dev/null
+++ b/weed/topology/duplicate_volume_id_test.go
@@ -0,0 +1,81 @@
+package topology
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+func TestDuplicateVolumeIdAcrossCollections(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+
+	dn1 := NewDataNode("server1")
+	dn2 := NewDataNode("server2")
+
+	v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+
+	dn1.UpdateVolumes([]storage.VolumeInfo{v1})
+	dn2.UpdateVolumes([]storage.VolumeInfo{v2})
+	topo.RegisterVolumeLayout(v1, dn1)
+	topo.RegisterVolumeLayout(v2, dn2)
+	topo.flagDuplicateVolumeId(v2, dn2)
+
+	vi, err := dn2.GetVolumesById(7)
+	if err != nil {
+		t.Fatalf("volume info not stored on dn2: %v", err)
+	}
+	if !vi.ReadOnly {
+		t.Fatal("duplicate newcomer should be marked read-only")
+	}
+}
+
+func TestSameLayoutReplicaIsNotDuplicate(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+
+	dn1 := NewDataNode("server1")
+	dn2 := NewDataNode("server2")
+
+	v := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	dn1.UpdateVolumes([]storage.VolumeInfo{v})
+	dn2.UpdateVolumes([]storage.VolumeInfo{v})
+	topo.RegisterVolumeLayout(v, dn1)
+	topo.RegisterVolumeLayout(v, dn2)
+	topo.flagDuplicateVolumeId(v, dn2)
+
+	vi, err := dn2.GetVolumesById(7)
+	if err != nil {
+		t.Fatalf("volume info not stored on dn2: %v", err)
+	}
+	if vi.ReadOnly {
+		t.Fatal("a same-layout replica is a legitimate copy, not a duplicate")
+	}
+}
+
+func TestSameNodeVolumeMoveIsNotDuplicate(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	dn := NewDataNode("server1")
+
+	old := storage.VolumeInfo{Id: 7, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	moved := storage.VolumeInfo{Id: 7, Collection: "c", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, DiskType: "ssd"}
+	dn.UpdateVolumes([]storage.VolumeInfo{old})
+	topo.RegisterVolumeLayout(old, dn)
+
+	// the move registers the new layout before the old entry is removed
+	dn.UpdateVolumes([]storage.VolumeInfo{moved})
+	topo.RegisterVolumeLayout(moved, dn)
+	topo.flagDuplicateVolumeId(moved, dn)
+
+	vi, err := dn.GetVolumesById(7)
+	if err != nil {
+		t.Fatalf("volume info not stored on dn: %v", err)
+	}
+	if vi.ReadOnly {
+		t.Fatal("a same-node disk move is not a duplicate")
+	}
+}
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index 0c7c421fbf..6127296727 100644
--- a/weed/topology/topology.go
+++ b/weed/topology/topology.go
@@ -612,12 +612,62 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 		if vl.RegisterVolume(&v, dn) {
 			vl.EnsureCorrectWritables(&v)
+			go t.flagDuplicateVolumeId(v, dn)
 			return
 		}
 		// Dropped with its collection; the next lookup creates a fresh one.
 	}
 }
 
+// flagDuplicateVolumeId reports and quarantines v.Id when another data node
+// already serves it under a different layout key. It runs after registration
+// so a legitimate same-node move (disk type or collection change) settles
+// first and does not trip it.
+func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
+	dupCollection, nodes := t.findDuplicateVolumeLayout(v, dn)
+	if len(nodes) == 0 {
+		return
+	}
+	stats.MasterDuplicateVolumeIds.WithLabelValues(v.Collection, v.Id.String()).Inc()
+	glog.Errorf("duplicate volume id %d: %s reports it under collection %q but %v also serves it under collection %q; marking %s's copy read-only", v.Id, dn.Id(), v.Collection, nodes, dupCollection, dn.Id())
+	if vl := t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)); vl != nil {
+		vl.SetVolumeReadOnly(dn, v.Id)
+	}
+}
+
+// findDuplicateVolumeLayout reports the collection and the other data nodes
+// already holding v.Id under a different layout key. Volume ids are global,
+// so the same id held by two nodes under two layouts means the allocator
+// reused an id.
+func (t *Topology) findDuplicateVolumeLayout(v storage.VolumeInfo, dn *DataNode) (string, []*DataNode) {
+	if v.ReplicaPlacement == nil {
+		return "", nil
+	}
+	diskType := types.ToDiskType(v.DiskType)
+	for _, item := range t.collectionMap.Items() {
+		c := item.(*Collection)
+		var same *VolumeLayout
+		if c.Name == v.Collection {
+			same, _ = c.GetVolumeLayout(v.ReplicaPlacement, v.Ttl, diskType)
+		}
+		for _, vl := range c.GetAllVolumeLayouts() {
+			if vl == same {
+				continue
+			}
+			var others []*DataNode
+			for _, n := range vl.Lookup(v.Id) {
+				if n != dn {
+					others = append(others, n)
+				}
+			}
+			if len(others) > 0 {
+				return c.Name, others
+			}
+		}
+	}
+	return "", nil
+}
+
 func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 	glog.Infof("removing volume info: %+v from %v", v, dn.id)
 	diskType := types.ToDiskType(v.DiskType)