Detect and quarantine duplicate volume ids (#11763) (#11828)

* master: detect and quarantine duplicate volume ids across layouts (#11763)

* shell: volume.check.duplicate_ids audit command (#11763)

* master: persist duplicate-vid quarantine across heartbeats (#11763)

* master: quarantine all copies of a duplicated volume id

Near-simultaneous registrations let each side quarantine only the other,
so arrival order decided which copy stayed writable. A confirmed
cross-layout collision means either copy could be the impostor, so flag
every colliding copy read-only regardless of registration order.

* topology: return a copy from VolumeLayout.Lookup

Callers iterating the returned slice raced with registrations and
removals replacing the location list in place.

* master: release duplicate-vid quarantine when the conflict resolves

A cross-node volume.move briefly shows the same id on two nodes under
different disk layouts; permanent quarantine would pin a valid target
read-only. Keep the first-reported state per quarantined copy and lift
the marks once the conflicting copy unregisters, so only a persistent
collision stays read-only.

* shell: include ec shards in volume.check.duplicate_ids

Ec-only volumes share the global id space. Compare shard collections
against regular ones while allowing shards and regular copies of the
same encoded volume.

* master: keep quarantine release current and cover all removal paths

Track the latest server-reported read-only flag while quarantined so a
release restores the server s true state, and run the release check on
collection deletion and data-node removal too, not only explicit volume
unregistration.

* shell: do not flag ec-only volumes as duplicates

An ec-only volume has no regular copy, so its shard collections had
nothing to compare against and always read as foreign.

* master: quarantine every conflicting layout at the layout level (#11763)

- collect all conflicting layouts, not just the first, so a vid spanning
  three layouts does not leave one writable
- mark quarantine on the location list only: folding it into the stored
  report made the volume digest permanently disagree with the server and
  forced a full volume list every heartbeat
- isAllWritable now honors the location read-only mark so a quarantined
  copy stays out of writables through resyncs and re-registration
- re-check the conflict before keeping marks, and do not clear a node's
  quarantine while it still holds the vid in another layout
- log the full layout keys on both sides of the conflict

* master: push duplicate-vid quarantine to the volume server (#11763)

A quarantined copy stayed writable on the volume server itself for a
client holding an existing file id, since the master-only mark never
reached the server's local write path. Quarantine now also calls
VolumeMarkReadonly (can_delete so expiry still drains), is re-pushed if
the server comes back reporting it writable, and VolumeMarkWritable
runs when the conflict resolves and the server had reported writable.

* master: retry unenforced quarantine marks and restore server writability on release (#11763)

Dial the advertised grpc port, return mark errors, and keep a
quarantine entry until the writable push succeeds so a released
copy does not stay read-only on the server.

* master: treat rp-only layout differences as an in-flight replication change (#11763)

volume.configure.replication rolls through nodes one at a time, so a
volume legitimately reports the old and new replica placement while it
settles. Layouts differing only in rp within the same collection are a
supported transition, not a reused id.

* shell: group ec shards by collection in the duplicate id audit (#11763)

EC shards of one volume legitimately span disk types when
DiskTypePrefer spills, so grouping by disk type reports healthy
volumes as duplicates.

weed/server/master_server.go

@@ -15,7 +15,9 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster/maintenance"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/telemetry"
 
 	"github.com/seaweedfs/seaweedfs/weed/cluster"
@@ -28,6 +30,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/sequence"
 	"github.com/seaweedfs/seaweedfs/weed/shell"
@@ -162,6 +165,22 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 	ms.Topo.SetGrowthFreezeConfig(
 		v.GetBool("master.volume_growth.freeze_on_topology_mismatch"),
 		v.GetDuration("master.volume_growth.state_loss_recovery_wait"))
+	ms.Topo.QuarantineVolumeFn = func(dn *topology.DataNode, vid needle.VolumeId, readOnly bool) error {
+		err := operation.WithVolumeServerClient(false, dn.ServerAddress(), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			if readOnly {
+				_, err := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid), CanDelete: true})
+				return err
+			}
+			_, err := client.VolumeMarkWritable(ctx, &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)})
+			return err
+		})
+		if err != nil {
+			glog.Warningf("quarantine volume %d on %s readOnly=%v: %v", vid, dn.Id(), readOnly, err)
+		}
+		return err
+	}
 	ms.vg = topology.NewDefaultVolumeGrowth()
 	glog.V(0).Infoln("Volume Size Limit is", ms.option.VolumeSizeLimitMB, "MB")
 

weed/shell/command_volume_check_duplicate_ids.go

@@ -0,0 +1,140 @@
+package shell
+
+import (
+	"fmt"
+	"io"
+	"slices"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+func init() {
+	Commands = append(Commands, &commandVolumeCheckDuplicateIds{})
+}
+
+type commandVolumeCheckDuplicateIds struct {
+}
+
+func (c *commandVolumeCheckDuplicateIds) Name() string {
+	return "volume.check.duplicate_ids"
+}
+
+func (c *commandVolumeCheckDuplicateIds) Help() string {
+	return `find volume ids registered under more than one layout
+
+	volume.check.duplicate_ids
+
+	Volume ids are global. The same id under two collections, replica
+	placements, TTLs, or disk types means the allocator reused an id, for
+	example after master state loss. See volume.growth.resume.
+
+`
+}
+
+func (c *commandVolumeCheckDuplicateIds) HasTag(CommandTag) bool {
+	return false
+}
+
+func (c *commandVolumeCheckDuplicateIds) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+	topoInfo, _, err := collectTopologyInfo(commandEnv, 0)
+	if err != nil {
+		return err
+	}
+
+	// vid -> layout key -> locations; ec shards tracked per collection since
+	// shards plus regular copies of the same encoded volume are legitimate
+	seen := make(map[needle.VolumeId]map[string][]string)
+	ecSeen := make(map[needle.VolumeId]map[string][]string)
+	collections := make(map[needle.VolumeId]map[string]bool)
+	ecCollections := make(map[needle.VolumeId]map[string]bool)
+	add := func(m map[needle.VolumeId]map[string][]string, id uint32, key, location string) {
+		layouts := m[needle.VolumeId(id)]
+		if layouts == nil {
+			layouts = make(map[string][]string)
+			m[needle.VolumeId(id)] = layouts
+		}
+		layouts[key] = append(layouts[key], location)
+	}
+	addCollection := func(m map[needle.VolumeId]map[string]bool, id uint32, collection string) {
+		if m[needle.VolumeId(id)] == nil {
+			m[needle.VolumeId(id)] = make(map[string]bool)
+		}
+		m[needle.VolumeId(id)][collection] = true
+	}
+	for _, dc := range topoInfo.DataCenterInfos {
+		for _, r := range dc.RackInfos {
+			for _, dn := range r.DataNodeInfos {
+				for diskKey, di := range dn.DiskInfos {
+					for _, vi := range di.VolumeInfos {
+						rp, _ := super_block.NewReplicaPlacementFromByte(byte(vi.ReplicaPlacement))
+						add(seen, vi.Id, fmt.Sprintf("collection=%q replication=%s ttl=%s disk=%s",
+							vi.Collection, rp.String(), needle.LoadTTLFromUint32(vi.Ttl), vi.DiskType),
+							fmt.Sprintf("%s %s", dn.Id, diskKey))
+						addCollection(collections, vi.Id, vi.Collection)
+					}
+					for _, si := range di.EcShardInfos {
+						add(ecSeen, si.Id, fmt.Sprintf("collection=%q ec_shards", si.Collection),
+							fmt.Sprintf("%s %s disk=%s", dn.Id, diskKey, si.DiskType))
+						addCollection(ecCollections, si.Id, si.Collection)
+					}
+				}
+			}
+		}
+	}
+
+	vids := make(map[needle.VolumeId]struct{}, len(seen)+len(ecSeen))
+	for vid := range seen {
+		vids[vid] = struct{}{}
+	}
+	for vid := range ecSeen {
+		vids[vid] = struct{}{}
+	}
+
+	duplicates := 0
+	for _, vid := range sortedKeys(vids) {
+		layouts := seen[vid]
+		shardLayouts := ecSeen[vid]
+		ecForeign := false
+		if len(collections[vid]) > 0 {
+			for collection := range ecCollections[vid] {
+				if !collections[vid][collection] {
+					ecForeign = true
+				}
+			}
+		}
+		if len(layouts) < 2 && len(shardLayouts) < 2 && !ecForeign {
+			continue
+		}
+		duplicates++
+		fmt.Fprintf(writer, "volume %d is registered under %d layouts:\n", vid, len(layouts)+len(shardLayouts))
+		for _, key := range sortedKeysString(layouts) {
+			fmt.Fprintf(writer, "  %-60s %v\n", key, layouts[key])
+		}
+		for _, key := range sortedKeysString(shardLayouts) {
+			fmt.Fprintf(writer, "  %-60s %v\n", key, shardLayouts[key])
+		}
+	}
+	if duplicates == 0 {
+		fmt.Fprintf(writer, "no duplicate volume ids found\n")
+	}
+	return nil
+}
+
+func sortedKeys[V any](m map[needle.VolumeId]V) []needle.VolumeId {
+	keys := make([]needle.VolumeId, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	slices.Sort(keys)
+	return keys
+}
+
+func sortedKeysString[V any](m map[string]V) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	slices.Sort(keys)
+	return keys
+}

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)

weed/topology/duplicate_volume_id_test.go

@@ -0,0 +1,187 @@
+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 isWritable(t *testing.T, topo *Topology, v storage.VolumeInfo) bool {
+	t.Helper()
+	vl := topo.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType))
+	for _, id := range vl.CloneWritableVolumes() {
+		if id == v.Id {
+			return true
+		}
+	}
+	return false
+}
+
+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)
+
+	if isWritable(t, topo, v2) {
+		t.Fatal("duplicate newcomer should be quarantined out of the writable list")
+	}
+	if isWritable(t, topo, v1) {
+		t.Fatal("either copy could be the impostor; the earlier copy must be quarantined too")
+	}
+}
+
+func TestSameLayoutReplicaIsNotDuplicate(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+
+	dn1 := NewDataNode("server1")
+	dn1.Ip = "10.0.0.1"
+	dn1.Port = 8080
+	dn2 := NewDataNode("server2")
+	dn2.Ip = "10.0.0.2"
+	dn2.Port = 8080
+
+	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)
+
+	if !isWritable(t, topo, v) {
+		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)
+
+	if !isWritable(t, topo, moved) {
+		t.Fatal("a same-node disk move is not a duplicate")
+	}
+}
+
+func TestQuarantineReleasedWhenConflictResolves(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)
+
+	topo.UnRegisterVolumeLayout(v2, dn2)
+
+	if !isWritable(t, topo, v1) {
+		t.Fatal("the surviving copy must be released once the conflict is gone")
+	}
+}
+
+func TestQuarantineSurvivesResync(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)
+
+	// a resync from the same server re-reports the volume as writable;
+	// the quarantine must keep it out of the writable list
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
+		Id: 7, Collection: "b", ReplicaPlacement: 0, Version: 3,
+	}}, dn2)
+
+	if isWritable(t, topo, v2) {
+		t.Fatal("resync must not clear the quarantine")
+	}
+}
+
+func TestReplicationChangeIsNotDuplicate(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rpOld, _ := super_block.NewReplicaPlacementFromString("000")
+	rpNew, _ := super_block.NewReplicaPlacementFromString("001")
+
+	dn1 := NewDataNode("server1")
+	dn1.Ip = "10.0.0.1"
+	dn1.Port = 8080
+	dn2 := NewDataNode("server2")
+	dn2.Ip = "10.0.0.2"
+	dn2.Port = 8080
+
+	vOld := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rpOld, Ttl: needle.EMPTY_TTL}
+	vNew := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rpNew, Ttl: needle.EMPTY_TTL}
+
+	dn1.UpdateVolumes([]storage.VolumeInfo{vOld})
+	dn2.UpdateVolumes([]storage.VolumeInfo{vNew})
+	topo.RegisterVolumeLayout(vOld, dn1)
+	topo.RegisterVolumeLayout(vNew, dn2)
+	topo.flagDuplicateVolumeId(vNew, dn2)
+
+	if topo.isQuarantined(dn2, vNew.Id) || topo.isQuarantined(dn1, vOld.Id) {
+		t.Fatal("an rp-only layout difference is a replication change in flight, not a duplicate")
+	}
+}
+
+func TestDuplicateVolumeIdAcrossTtls(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	ttl, _ := needle.ReadTTL("3d")
+
+	dn1 := NewDataNode("server1")
+	dn1.Ip = "10.0.0.1"
+	dn1.Port = 8080
+	dn2 := NewDataNode("server2")
+	dn2.Ip = "10.0.0.2"
+	dn2.Port = 8080
+
+	v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	v2 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: ttl}
+
+	dn1.UpdateVolumes([]storage.VolumeInfo{v1})
+	dn2.UpdateVolumes([]storage.VolumeInfo{v2})
+	topo.RegisterVolumeLayout(v1, dn1)
+	topo.RegisterVolumeLayout(v2, dn2)
+	topo.flagDuplicateVolumeId(v2, dn2)
+
+	if !topo.isQuarantined(dn2, v2.Id) || !topo.isQuarantined(dn1, v1.Id) {
+		t.Fatal("same collection with a different ttl is still a reused volume id")
+	}
+}

weed/topology/topology.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"slices"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -99,6 +100,55 @@ type Topology struct {
 
 	reportedTopologyIdsLock    sync.Mutex
 	pendingReportedTopologyIds map[string]struct{}
+
+	// quarantinedVolumeIds are (node, vid) pairs force-read-only because the
+	// vid collided across layouts. Kept separate from the reported flag so
+	// later heartbeats from the same server cannot quietly undo it.
+	quarantinedVolumeIds sync.Map
+
+	// QuarantineVolumeFn pushes the quarantine onto the volume server itself
+	// (VolumeMarkReadonly/VolumeMarkWritable). nil keeps quarantine at the
+	// layout level only.
+	QuarantineVolumeFn func(dn *DataNode, vid needle.VolumeId, readOnly bool) error
+}
+
+type quarantinedVolume struct {
+	dnId NodeId
+	vid  needle.VolumeId
+}
+
+type quarantinedVolumeState struct {
+	reported bool
+	enforced bool
+}
+
+func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
+	_, ok := t.quarantinedVolumeIds.Load(quarantinedVolume{dn.Id(), vid})
+	return ok
+}
+
+// enforceQuarantine keeps the latest reported read-only flag for a quarantined
+// copy so a later release restores the server's true state. The stored report
+// keeps the server's flag: quarantine lives at the layout level, and folding
+// it into the record here would make the master's volume digest permanently
+// disagree with the server's and force a full volume list on every heartbeat.
+func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo {
+	q := quarantinedVolume{dn.Id(), vi.Id}
+	if v, ok := t.quarantinedVolumeIds.Load(q); ok {
+		state := v.(quarantinedVolumeState)
+		if !state.enforced {
+			state.reported = vi.ReadOnly
+		}
+		if t.QuarantineVolumeFn != nil && (!state.enforced || !vi.ReadOnly) {
+			// The mark never landed, or the server lost it (restart, manual
+			// unquarantine); push it while the conflict stands.
+			if t.QuarantineVolumeFn(dn, vi.Id, true) == nil {
+				state.enforced = true
+			}
+		}
+		t.quarantinedVolumeIds.Store(q, state)
+	}
+	return vi
 }
 
 func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
@@ -591,9 +641,15 @@ func (t *Topology) DeleteCollection(collectionName string) {
 	if !found {
 		return
 	}
+	var vids []needle.VolumeId
 	for _, vl := range collection.(*Collection).GetAllVolumeLayouts() {
+		vids = append(vids, vl.ListVids()...)
 		vl.releaseLookupOwnership()
 	}
+	// a vid colliding with a copy in this collection is no longer a duplicate
+	for _, vid := range vids {
+		t.releaseResolvedQuarantine(vid)
+	}
 }
 
 func (t *Topology) DeleteLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) {
@@ -612,15 +668,172 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 	for {
 		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 		if vl.RegisterVolume(&v, dn) {
+			if t.isQuarantined(dn, v.Id) {
+				vl.SetVolumeQuarantined(dn, v.Id)
+			}
 			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. Every copy across the conflicting layouts is
+// quarantined: either could be the impostor, and arrival order must not
+// decide which copy stays writable.
+func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
+	hits := t.findDuplicateVolumeLayouts(v, dn)
+	if len(hits) == 0 {
+		return
+	}
+	// A conflict could resolve between the scan above and the marks below;
+	// re-run the release so a stale quarantine is cleared instead of kept.
+	defer t.releaseResolvedQuarantine(v.Id)
+	stats.MasterDuplicateVolumeIds.WithLabelValues(v.Collection, v.Id.String()).Inc()
+	var otherKeys []string
+	for _, h := range hits {
+		otherKeys = append(otherKeys, h.layoutKey())
+	}
+	glog.Errorf("duplicate volume id %d: %s reports it under %s but other data nodes also serve it under %s; marking all copies read-only", v.Id, dn.Id(), volumeLayoutKey(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)), strings.Join(otherKeys, ", "))
+	quarantine := func(n *DataNode, vl *VolumeLayout) {
+		q := quarantinedVolume{n.Id(), v.Id}
+		reported := false
+		if vi, err := n.GetVolumesById(v.Id); err == nil {
+			reported = vi.ReadOnly
+		}
+		t.quarantinedVolumeIds.LoadOrStore(q, quarantinedVolumeState{reported: reported})
+		if vl != nil {
+			vl.SetVolumeQuarantined(n, v.Id)
+		}
+		if t.QuarantineVolumeFn != nil && t.QuarantineVolumeFn(n, v.Id, true) == nil {
+			if val, ok := t.quarantinedVolumeIds.Load(q); ok {
+				state := val.(quarantinedVolumeState)
+				state.enforced = true
+				t.quarantinedVolumeIds.Store(q, state)
+			}
+		}
+	}
+	quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)))
+	for _, h := range hits {
+		for _, n := range h.nodes {
+			quarantine(n, h.vl)
+		}
+	}
+}
+
+// releaseResolvedQuarantine lifts read-only marks once the conflicting copy
+// is gone, so a finished move does not leave the surviving copy quarantined.
+func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) {
+	t.quarantinedVolumeIds.Range(func(key, val any) bool {
+		q := key.(quarantinedVolume)
+		if q.vid != vid {
+			return true
+		}
+		dn, vl, vi := t.findQuarantinedVolume(q.dnId, vid)
+		if dn == nil {
+			t.quarantinedVolumeIds.Delete(key)
+			return true
+		}
+		if hits := t.findDuplicateVolumeLayouts(vi, dn); len(hits) > 0 {
+			return true
+		}
+		state, _ := val.(quarantinedVolumeState)
+		if state.reported {
+			dn.SetVolumeReadOnly(vid, true)
+			vl.SetReplicaReadOnlyFlag(dn, vid, true)
+		} else {
+			vl.SetVolumeWritable(dn, vid)
+			if state.enforced && t.QuarantineVolumeFn != nil {
+				if err := t.QuarantineVolumeFn(dn, vid, false); err != nil {
+					// The server still rejects writes; keep the mark so a
+					// later release pass retries the writable push.
+					return true
+				}
+			}
+		}
+		t.quarantinedVolumeIds.Delete(key)
+		return true
+	})
+}
+
+func (t *Topology) findQuarantinedVolume(dnId NodeId, vid needle.VolumeId) (*DataNode, *VolumeLayout, storage.VolumeInfo) {
+	for _, item := range t.collectionMap.Items() {
+		for _, vl := range item.(*Collection).GetAllVolumeLayouts() {
+			for _, n := range vl.Lookup(vid) {
+				if n.Id() == dnId {
+					vi, _ := n.GetVolumesById(vid)
+					return n, vl, vi
+				}
+			}
+		}
+	}
+	return nil, nil, storage.VolumeInfo{}
+}
+
+// duplicateHit is one layout that holds v.Id on data nodes other than the
+// reporter's.
+type duplicateHit struct {
+	collection string
+	vl         *VolumeLayout
+	nodes      []*DataNode
+}
+
+func (h duplicateHit) layoutKey() string {
+	return volumeLayoutKey(h.collection, h.vl.rp, h.vl.ttl, h.vl.diskType)
+}
+
+func volumeLayoutKey(collection string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) string {
+	return fmt.Sprintf("collection=%q rp=%s ttl=%s disk=%s", collection, rp, ttl, diskType)
+}
+
+func ttlEqual(a, b *needle.TTL) bool {
+	if a == nil || b == nil {
+		return a == b
+	}
+	return a.ToUint32() == b.ToUint32()
+}
+
+
+// findDuplicateVolumeLayouts reports every layout that already holds v.Id on
+// data nodes other than dn. Volume ids are global, so the same id held by two
+// nodes under two layouts means the allocator reused an id.
+func (t *Topology) findDuplicateVolumeLayouts(v storage.VolumeInfo, dn *DataNode) []duplicateHit {
+	if v.ReplicaPlacement == nil {
+		return nil
+	}
+	diskType := types.ToDiskType(v.DiskType)
+	var hits []duplicateHit
+	for _, item := range t.collectionMap.Items() {
+		c := item.(*Collection)
+		for _, vl := range c.GetAllVolumeLayouts() {
+			if c.Name == v.Collection && vl.diskType == diskType && ttlEqual(vl.ttl, v.Ttl) {
+				// The reporter's own layout, or an rp-only difference:
+				// a replication change rolls through the nodes one at a
+				// time, so the same volume legitimately reports the old
+				// and new rp while it settles.
+				continue
+			}
+			var others []*DataNode
+			for _, n := range vl.Lookup(v.Id) {
+				if n != dn {
+					others = append(others, n)
+				}
+			}
+			if len(others) > 0 {
+				hits = append(hits, duplicateHit{collection: c.Name, vl: vl, nodes: others})
+			}
+		}
+	}
+	return hits
+}
+
 func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 	glog.Infof("removing volume info: %+v from %v", v, dn.id)
+	defer t.releaseResolvedQuarantine(v.Id)
 	diskType := types.ToDiskType(v.DiskType)
 	volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 	volumeLayout.UnRegisterVolume(&v, dn)
@@ -745,6 +958,9 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 			glog.V(0).Infof("Fail to convert joined volume information: %v", err)
 		}
 	}
+	for i := range volumeInfos {
+		volumeInfos[i] = t.enforceQuarantine(dn, volumeInfos[i])
+	}
 	// find out the delta volumes
 	newVolumes, deletedVolumes, changedVolumes = dn.UpdateVolumes(volumeInfos)
 	for _, v := range newVolumes {
@@ -757,7 +973,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		if v.ReplicaPlacement == nil {
 			continue
 		}
-		t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)).SetReplicaReadOnlyFlag(dn, v.Id, v.ReadOnly)
+		t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)).SetReplicaReadOnlyFlag(dn, v.Id, v.ReadOnly || t.isQuarantined(dn, v.Id))
 	}
 	// Update effective sizes for all reported volumes (decay pending estimates).
 	// If decay brings a volume eagerly removed by RecordAssign back under the
@@ -781,6 +997,7 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 				// one, which the calls below must use too.
 				vl = t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 			}
+			go t.flagDuplicateVolumeId(v, dn)
 			// Volumes new to the disk map were registered above, so reaching
 			// here means only the lookup index had lost it. Clients were told
 			// it went when the node dropped out, so the repair has to tell them
@@ -804,7 +1021,7 @@ func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolume
 			glog.V(0).Infof("NewVolumeInfoFromShort %v: %v", v, err)
 			continue
 		}
-		newVis = append(newVis, vi)
+		newVis = append(newVis, t.enforceQuarantine(dn, vi))
 	}
 	for _, v := range deletedVolumes {
 		vi, err := storage.NewVolumeInfoFromShort(v)
@@ -865,7 +1082,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 			glog.V(0).Infof("Fail to convert changed volume information: %v", err)
 			continue
 		}
-		volumeInfos = append(volumeInfos, vi)
+		volumeInfos = append(volumeInfos, t.enforceQuarantine(dn, vi))
 	}
 
 	for _, vi := range volumeInfos {
@@ -885,11 +1102,14 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 			// Dropped with its collection; the next lookup creates a fresh one.
 			vl = t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType))
 		}
+		if becameServable {
+			go t.flagDuplicateVolumeId(vi, dn)
+		}
 		if isNew || becameServable || tierTransition || isChanged {
 			newVolumes = append(newVolumes, vi)
 		}
 		if isChanged {
-			vl.SetReplicaReadOnlyFlag(dn, vi.Id, vi.ReadOnly)
+			vl.SetReplicaReadOnlyFlag(dn, vi.Id, vi.ReadOnly || t.isQuarantined(dn, vi.Id))
 		}
 		vl.UpdateOversizedState(&vi, dn)
 		if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision, true) {

weed/topology/topology_event_handling.go

@@ -118,6 +118,8 @@ func (t *Topology) UnRegisterDataNode(dn *DataNode) {
 		diskType := types.ToDiskType(v.DiskType)
 		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 		vl.SetVolumeUnavailable(dn, v.Id)
+		// this node may have been the conflicting copy pinning a quarantine
+		defer t.releaseResolvedQuarantine(v.Id)
 	}
 
 	// unregister ec shards when volume server disconnected

weed/topology/volume_layout.go

@@ -411,6 +411,9 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {
 // or one on a server in maintenance mode, holds the whole volume out.
 func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool {
 	if location, ok := vl.vid2location[vid]; ok {
+		if location.AnyReadOnly() {
+			return false
+		}
 		for _, dn := range location.list {
 			if dn.InMaintenanceMode() {
 				return false
@@ -586,12 +589,21 @@ func (vl *VolumeLayout) isEmpty() bool {
 	return len(vl.vid2location) == 0
 }
 
+func (vl *VolumeLayout) ListVids() (vids []needle.VolumeId) {
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+	for vid := range vl.vid2location {
+		vids = append(vids, vid)
+	}
+	return
+}
+
 func (vl *VolumeLayout) Lookup(vid needle.VolumeId) []*DataNode {
 	vl.accessLock.RLock()
 	defer vl.accessLock.RUnlock()
 
 	if location := vl.vid2location[vid]; location != nil {
-		return location.list
+		return append([]*DataNode(nil), location.list...)
 	}
 	return nil
 }
@@ -1021,6 +1033,19 @@ func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) boo
 	return false
 }
 
+// SetVolumeQuarantined marks dn's replica read-only at the layout level only.
+// The node's stored report is left untouched so the volume digest still tracks
+// what the server reports and quarantined copies do not force endless full
+// heartbeats.
+func (vl *VolumeLayout) SetVolumeQuarantined(dn *DataNode, vid needle.VolumeId) {
+	vl.accessLock.Lock()
+	defer vl.accessLock.Unlock()
+	if location, ok := vl.vid2location[vid]; ok {
+		location.SetReadOnly(dn, true)
+	}
+	vl.removeFromWritable(vid)
+}
+
 // SetReplicaReadOnlyFlag records what dn's latest heartbeat said about its
 // replica of vid, and nothing more: the writable list is left to
 // EnsureCorrectWritables and its capacity guards. Until now the flag only moved