mq: fix idle-cleanup shard deadlock that permanently wedges the broker's topic map (#11051)

mq: remove emptied topics after the cleanup iteration, not inside it

cleanupIdlePartitions called manager.topics.Remove from inside
manager.topics.IterCb. IterCb holds the shard's read lock while running
the callback, and Remove takes the same shard's write lock, so removing
an emptied topic self-deadlocked the cleanup goroutine. The pending
writer then blocked every later reader of that shard, permanently
hanging ListTopicsInMemory and, for shard-mates, TopicExistsInMemory.

On the Kafka gateway this surfaced as flaky e2e consumer-group tests:
one minute after any earlier topic went idle, the broker's first
'Removing empty topic' wedged the map, every gateway
ListTopics/TopicExists RPC burned its full 5s timeout, Metadata could no
longer finish inside kafka-go's 5s coordinator deadline, and consumer
groups looped in PreparingRebalance until the test timed out.

Collect the emptied topic keys during the iteration and remove them
afterwards via RemoveCb, re-checking emptiness under the shard lock so a
topic that just gained a partition is kept.

Claude-Session: https://claude.ai/code/session_014yA6c8JQcY6MqPXCT13yYA

weed/mq/topic/local_manager.go

@@ -52,6 +52,7 @@ func (manager *LocalTopicManager) StartIdlePartitionCleanup(ctx context.Context,
 // cleanupIdlePartitions removes idle partitions from memory
 func (manager *LocalTopicManager) cleanupIdlePartitions(idleTimeout time.Duration) {
 	cleanedCount := 0
+	var emptyTopics []string
 
 	// Iterate through all topics
 	manager.topics.IterCb(func(topicKey string, localTopic *LocalTopic) {
@@ -78,13 +79,33 @@ func (manager *LocalTopicManager) cleanupIdlePartitions(idleTimeout time.Duratio
 			}
 		}
 
-		// If topic has no partitions left, remove it
 		if len(localTopic.Partitions) == 0 {
-			glog.V(1).Infof("Removing empty topic %s", topicKey)
+			emptyTopics = append(emptyTopics, topicKey)
-			manager.topics.Remove(topicKey)
 		}
 	})
 
+	// Remove emptied topics only after the iteration: IterCb holds the shard's
+	// read lock while running the callback, so calling Remove (which takes the
+	// same shard's write lock) inside it self-deadlocks and permanently wedges
+	// the shard — every later reader of that shard blocks behind the stuck
+	// writer, hanging ListTopicsInMemory/TopicExistsInMemory forever.
+	for _, topicKey := range emptyTopics {
+		manager.topics.RemoveCb(topicKey, func(key string, localTopic *LocalTopic, exists bool) bool {
+			if !exists || localTopic == nil {
+				return false
+			}
+			// Re-check emptiness under the shard's write lock: a publisher or
+			// subscriber may have added a partition since the scan above.
+			localTopic.partitionLock.RLock()
+			defer localTopic.partitionLock.RUnlock()
+			if len(localTopic.Partitions) > 0 {
+				return false
+			}
+			glog.V(1).Infof("Removing empty topic %s", key)
+			return true
+		})
+	}
+
 	if cleanedCount > 0 {
 		glog.V(0).Infof("Cleaned up %d idle partition(s)", cleanedCount)
 	}

weed/mq/topic/local_manager_test.go

@@ -0,0 +1,75 @@
+package topic
+
+import (
+	"testing"
+	"time"
+)
+
+// cleanupIdlePartitions used to call manager.topics.Remove from inside
+// manager.topics.IterCb. IterCb holds the shard's read lock while running the
+// callback, and Remove takes the same shard's write lock, so removing an
+// emptied topic deadlocked the cleanup goroutine and permanently wedged the
+// shard: every later ListTopicsInMemory / TopicExistsInMemory blocked behind
+// the stuck writer. Observed in CI as the Kafka gateway's Metadata requests
+// timing out forever once any topic went idle mid-test.
+func TestCleanupIdlePartitionsRemovesEmptyTopicWithoutDeadlock(t *testing.T) {
+	manager := NewLocalTopicManager()
+
+	tp := NewTopic("test", "idle-topic")
+	localTopic := NewLocalTopic(tp)
+	localPartition := NewLocalPartition(Partition{RingSize: 2520, RangeStart: 0, RangeStop: 2520}, 1, nil, nil)
+	localTopic.Partitions = append(localTopic.Partitions, localPartition)
+	manager.topics.Set(tp.String(), localTopic)
+
+	// Make the partition idle beyond any timeout.
+	localPartition.lastActivityTime.Store(time.Now().Add(-time.Hour).UnixNano())
+
+	done := make(chan struct{})
+	go func() {
+		manager.cleanupIdlePartitions(time.Minute)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("cleanupIdlePartitions deadlocked while removing an emptied topic")
+	}
+
+	if manager.topics.Has(tp.String()) {
+		t.Errorf("emptied topic %s should have been removed", tp.String())
+	}
+
+	// The map must still be fully usable afterwards — with the deadlock the
+	// wedged shard made this call block forever.
+	listed := make(chan []Topic, 1)
+	go func() { listed <- manager.ListTopicsInMemory() }()
+	select {
+	case topics := <-listed:
+		if len(topics) != 0 {
+			t.Errorf("expected no topics in memory, got %v", topics)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("ListTopicsInMemory blocked after cleanup — shard still wedged")
+	}
+}
+
+// A topic that gains a partition between the cleanup scan and the removal must
+// be kept: removal re-checks emptiness under the shard write lock.
+func TestCleanupIdlePartitionsKeepsActiveTopic(t *testing.T) {
+	manager := NewLocalTopicManager()
+
+	tp := NewTopic("test", "active-topic")
+	localPartition := NewLocalPartition(Partition{RingSize: 2520, RangeStart: 0, RangeStop: 2520}, 1, nil, nil)
+	manager.AddLocalPartition(tp, localPartition)
+
+	// Fresh activity: nothing should be cleaned up.
+	manager.cleanupIdlePartitions(time.Minute)
+
+	if !manager.topics.Has(tp.String()) {
+		t.Errorf("active topic %s should not have been removed", tp.String())
+	}
+	if manager.GetLocalPartition(tp, localPartition.Partition) == nil {
+		t.Errorf("active partition should still be registered")
+	}
+}