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.

weed/topology/topology.go

@@ -113,9 +113,12 @@ func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
 }
 
 // enforceQuarantine keeps a quarantined copy read-only even when the server's
-// own reports claim it writable again.
+// own reports claim it writable again. The entry also tracks the latest
+// reported flag so a later release restores the server's true state.
 func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo {
-	if t.isQuarantined(dn, vi.Id) {
+	q := quarantinedVolume{dn.Id(), vi.Id}
+	if _, ok := t.quarantinedVolumeIds.Load(q); ok {
+		t.quarantinedVolumeIds.Store(q, vi.ReadOnly)
 		vi.ReadOnly = true
 	}
 	return vi
@@ -611,9 +614,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) {

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

@@ -586,6 +586,15 @@ 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()