topology: list collections without nesting the collection map read lock (#12495)

* topology: list collections without nesting the collection map read lock

Items() already takes the read lock and copies the values. Holding it
around the call made ListCollections take the read lock twice, so a
DeleteCollection waiting for the write lock in between blocked the
second read lock and wedged the master.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* topology: test ListCollections against a concurrent collection delete

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

weed/topology/topology.go

@@ -813,11 +813,9 @@ func (t *Topology) ListCollections(includeNormalVolumes, includeEcVolumes bool)
 	found := make(map[string]bool)
 
 	if includeNormalVolumes {
-		t.collectionMap.RLock()
 		for _, c := range t.collectionMap.Items() {
 			found[c.(*Collection).Name] = true
 		}
-		t.collectionMap.RUnlock()
 	}
 
 	if includeEcVolumes {

weed/topology/topology_test.go

@@ -641,6 +641,40 @@ func TestListCollections(t *testing.T) {
 	}
 }
 
+// A collection delete that queues for the write lock while ListCollections
+// reads the collections must not wedge both.
+func TestListCollectionsWithConcurrentCollectionDelete(t *testing.T) {
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+
+	stop := make(chan struct{})
+	defer close(stop)
+	go func() {
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			topo.GetVolumeLayout("churn", rp, needle.EMPTY_TTL, types.HardDriveType)
+			topo.DeleteCollection("churn")
+		}
+	}()
+
+	listed := make(chan struct{})
+	go func() {
+		defer close(listed)
+		for i := 0; i < 100000; i++ {
+			topo.ListCollections(true, false)
+		}
+	}()
+	select {
+	case <-listed:
+	case <-time.After(10 * time.Second):
+		t.Fatal("ListCollections deadlocked with a concurrent DeleteCollection")
+	}
+}
+
 func TestDataNodeIdBasedIdentification(t *testing.T) {
 	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
 	dc := topo.GetOrCreateDataCenter("dc1")