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.

weed/topology/duplicate_volume_id_test.go

@@ -89,6 +89,31 @@ func TestSameNodeVolumeMoveIsNotDuplicate(t *testing.T) {
 	}
 }
 
+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)
+
+	vi, err := dn1.GetVolumesById(7)
+	if err != nil {
+		t.Fatalf("volume info not stored on dn1: %v", err)
+	}
+	if vi.ReadOnly {
+		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")

weed/topology/topology.go

@@ -656,16 +656,64 @@ 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 all copies read-only", v.Id, dn.Id(), v.Collection, nodes, dupCollection)
-	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)
+	quarantine := func(n *DataNode, vl *VolumeLayout) {
+		reported := false
+		if vi, err := n.GetVolumesById(v.Id); err == nil {
+			reported = vi.ReadOnly
+		}
+		t.quarantinedVolumeIds.LoadOrStore(quarantinedVolume{n.Id(), v.Id}, reported)
+		if vl != nil {
+			vl.SetVolumeReadOnly(n, v.Id)
+		}
 	}
+	quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)))
 	for _, n := range nodes {
-		t.quarantinedVolumeIds.Store(quarantinedVolume{n.Id(), v.Id}, struct{}{})
-		dupLayout.SetVolumeReadOnly(n, v.Id)
+		quarantine(n, dupLayout)
 	}
 }
 
+// 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 _, _, nodes := t.findDuplicateVolumeLayout(vi, dn); len(nodes) > 0 {
+			return true
+		}
+		reported, _ := val.(bool)
+		if reported {
+			dn.SetVolumeReadOnly(vid, true)
+			vl.SetReplicaReadOnlyFlag(dn, vid, true)
+		} else {
+			vl.SetVolumeWritable(dn, vid)
+		}
+		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{}
+}
+
 // findDuplicateVolumeLayout reports the layout 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
@@ -702,6 +750,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})
+	defer t.releaseResolvedQuarantine(v.Id)
 	diskType := types.ToDiskType(v.DiskType)
 	volumeLayout := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 	volumeLayout.UnRegisterVolume(&v, dn)