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

weed/topology/duplicate_volume_id_test.go

@@ -3,6 +3,7 @@ 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"
@@ -79,3 +80,32 @@ func TestSameNodeVolumeMoveIsNotDuplicate(t *testing.T) {
 		t.Fatal("a same-node disk move is not a duplicate")
 	}
 }
+
+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 read-only
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
+		Id: 7, Collection: "b", ReplicaPlacement: 0, Version: 3,
+	}}, dn2)
+
+	vi, err := dn2.GetVolumesById(7)
+	if err != nil {
+		t.Fatalf("volume info not stored on dn2: %v", err)
+	}
+	if !vi.ReadOnly {
+		t.Fatal("resync must not clear the quarantine's read-only mark")
+	}
+}

weed/topology/topology.go

@@ -99,6 +99,30 @@ 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
+}
+
+type quarantinedVolume struct {
+	dnId NodeId
+	vid  needle.VolumeId
+}
+
+func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
+	_, ok := t.quarantinedVolumeIds.Load(quarantinedVolume{dn.Id(), vid})
+	return ok
+}
+
+// enforceQuarantine keeps a quarantined copy read-only even when the server's
+// own reports claim it writable again.
+func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo {
+	if t.isQuarantined(dn, vi.Id) {
+		vi.ReadOnly = true
+	}
+	return vi
 }
 
 func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, pulse int, replicationAsMin bool) *Topology {
@@ -631,6 +655,7 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
 	}
 	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())
+	t.quarantinedVolumeIds.Store(quarantinedVolume{dn.Id(), v.Id}, struct{}{})
 	if vl := t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)); vl != nil {
 		vl.SetVolumeReadOnly(dn, v.Id)
 	}
@@ -671,6 +696,7 @@ func (t *Topology) findDuplicateVolumeLayout(v storage.VolumeInfo, dn *DataNode)
 
 func (t *Topology) UnRegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 	glog.Infof("removing volume info: %+v from %v", v, dn.id)
+	t.quarantinedVolumeIds.Delete(quarantinedVolume{dn.Id(), v.Id})
 	diskType := types.ToDiskType(v.DiskType)
 	volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 	volumeLayout.UnRegisterVolume(&v, dn)
@@ -795,6 +821,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 {
@@ -807,7 +836,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
@@ -831,6 +860,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
@@ -854,7 +884,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)
@@ -915,7 +945,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 {
@@ -935,11 +965,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) {