fix(messagequeue): run queue garbage collection on busy partitions

platform/extension/messagequeue/mysql/subscriber.go

@@ -49,12 +49,6 @@ const (
 	// so it converges over multiple calls even with large backlogs.
 	watermarkAdvancementLimit = 1000
 
-	// gcIdleTickInterval controls how often GC runs during idle poll ticks.
-	// GC runs every Nth idle tick instead of every tick to avoid excessive
-	// queries when many partitions are idle (e.g., 50 idle partitions at 100ms
-	// poll interval = 500 GC queries/sec without throttling).
-	gcIdleTickInterval = 100
-
 	// heartbeatPurgeAfterLeaseDurations sets the age threshold for purging
 	// abandoned heartbeat rows, as a multiple of LeaseDurationMs (10x = 5min
 	// at defaults). Well past every transient window in the protocol — a row
@@ -81,6 +75,13 @@ const (
 	leasePurgeAfterLeaseDurations = 10
 )
 
+// gcTickInterval is the number of poll ticks between garbage collection runs.
+// GC runs every Nth tick regardless of delivery activity; without throttling,
+// many partitions polling in lockstep would flood the store with queries
+// (e.g., 50 partitions at 100ms poll interval = 500 GC queries/sec). A var so
+// tests can shorten it; production always uses the default.
+var gcTickInterval = 100
+
 // HookSignal identifies the type of subscriber lifecycle event.
 // Named after behavioral concerns (what happened) rather than implementation
 // details (which loop ran), so signal names remain stable across refactors.
@@ -167,8 +168,7 @@ type partitionWorker struct {
 	// partition. Set once on the first successful poll, avoiding repeated
 	// initialization calls on every tick.
 	offsetInitialized bool
-	// gcCounter counts idle poll ticks. GC only runs every gcIdleTickInterval
-	// ticks to avoid excessive queries when many partitions are idle.
+	// gcCounter counts poll ticks since the last garbage collection run.
 	gcCounter int
 }
 
@@ -1200,26 +1200,24 @@ func (w *partitionWorker) pollAndDeliver(ctx context.Context) (retErr error) {
 		)
 	}
 
-	// Run GC periodically (throttled to every Nth idle tick)
-	if messageCount == 0 {
-		w.gcCounter++
-		if w.gcCounter >= gcIdleTickInterval {
-			w.gcCounter = 0
-			if err := w.garbageCollect(ctx); err != nil {
-				return fmt.Errorf("garbage collect: %w", err)
-			}
-		}
-	} else {
-		w.gcCounter = 0
-	}
-
 	// Record poll metrics
 	if messageCount > 0 {
 		metrics.NamedCounter(s.scope, "poll", "messages_delivered", int64(messageCount),
 			metrics.NewTag("topic", sub.topic),
 		)
 	}
 
+	// GC runs every Nth tick regardless of delivery activity; an idle-only
+	// gate starved continuously busy partitions of garbage collection.
+	// Metrics are reported above so a GC failure cannot drop the delivery count.
+	w.gcCounter++
+	if w.gcCounter >= gcTickInterval {
+		w.gcCounter = 0
+		if err := w.garbageCollect(ctx); err != nil {
+			return fmt.Errorf("garbage collect: %w", err)
+		}
+	}
+
 	return nil
 }
 

platform/extension/messagequeue/mysql/subscriber_test.go

@@ -710,6 +710,64 @@ func TestSubscriber_PartitionWorkerPollAndDeliver(t *testing.T) {
 	assert.True(t, foundFinish, "expected poll.finish histogram")
 }
 
+// TestSubscriber_PollAndDeliver_GCOnBusyTicks verifies that garbage collection
+// runs on a partition that delivers a message on every poll tick. GC was gated
+// on idle ticks, so a continuously busy partition never reclaimed acked rows.
+func TestSubscriber_PollAndDeliver_GCOnBusyTicks(t *testing.T) {
+	old := gcTickInterval
+	gcTickInterval = 2
+	t.Cleanup(func() { gcTickInterval = old })
+
+	ctrl := gomock.NewController(t)
+
+	mockMessageStore := NewMockmessageStore(ctrl)
+	mockOffsetStore := NewMockoffsetStore(ctrl)
+	mockLeaseStore := NewMockpartitionLeaseStore(ctrl)
+
+	s := setupSubscriberTest(t, mockMessageStore, mockOffsetStore, mockLeaseStore).(*subscriber)
+
+	cfg := testSubscriptionConfig()
+	deliveryCh := make(chan extqueue.Delivery, 10)
+	sub := &subscription{
+		topic:      "test_topic",
+		config:     cfg,
+		deliveryCh: deliveryCh,
+		workers:    make(map[string]*partitionWorker),
+	}
+	w := &partitionWorker{
+		partitionKey: "part-1",
+		sub:          sub,
+		subscriber:   s,
+		done:         make(chan struct{}),
+	}
+
+	row := messageRow{
+		ID:           "msg-1",
+		Offset:       1,
+		PartitionKey: "part-1",
+		Payload:      []byte("payload"),
+		PublishedAt:  time.Now().UnixMilli(),
+	}
+	// Every poll delivers one message, so the partition never idles.
+	mockMessageStore.EXPECT().FetchByOffset(gomock.Any(), "test_topic", "part-1", int64(0), cfg.BatchSize).
+		Return([]messageRow{row}, nil).Times(3)
+	mockOffsetStore.EXPECT().Initialize(gomock.Any(), "test_topic", "part-1", cfg.ConsumerGroup).Return(nil)
+
+	// The counter reaches gcTickInterval on the second busy tick.
+	mockOffsetStore.EXPECT().GetMinAckedOffset(gomock.Any(), "test_topic", "part-1").Return(int64(1), true, nil)
+	mockMessageStore.EXPECT().GarbageCollect(gomock.Any(), "test_topic", "part-1", int64(1)).Return(int64(1), nil)
+
+	ctx := context.Background()
+	for i := 0; i < 3; i++ {
+		require.NoError(t, w.pollAndDeliver(ctx))
+		select {
+		case <-deliveryCh:
+		default:
+			t.Fatal("expected a delivery on every busy tick")
+		}
+	}
+}
+
 // TestSubscriber_PollAndDeliver_PostponedBarrier verifies that a postponed
 // message halts the partition scan (barrier), while a nacked message is
 // skipped and later offsets keep flowing.

test/integration/extension/messagequeue/mysql/queue_test.go

@@ -2268,6 +2268,73 @@ func (s *SQLQueueIntegrationSuite) TestIdleLeaseRelease() {
 	t.Logf("Idle lease released and partition resurrected on new traffic")
 }
 
+// TestGCReclaimsAckedRowsUnderContinuousTraffic verifies that garbage
+// collection reclaims acked rows on a partition that never idles. GC was gated
+// on idle poll ticks, so a continuously busy partition grew its message log
+// without bound; it must now run on its own tick cadence regardless of traffic.
+func (s *SQLQueueIntegrationSuite) TestGCReclaimsAckedRowsUnderContinuousTraffic() {
+	t := s.T()
+
+	topic := "gc_busy_topic"
+	partition := "gc-busy-part"
+	consumerGroup := "gc-busy-cg"
+
+	signalCh := make(chan queueMySQL.HookSignal, 100)
+	q, err := queueMySQL.NewQueue(queueMySQL.Params{
+		DB: s.db, Logger: zaptest.NewLogger(t), MetricsScope: tally.NoopScope,
+		OnSignal: signalCh,
+	})
+	require.NoError(t, err)
+	defer q.Close()
+
+	// Seed the backlog before subscribing so the GC counter starts at zero
+	// when polling begins and cannot fire mid-drain on a slow runner.
+	const initialBatch = 200
+	for i := 0; i < initialBatch; i++ {
+		require.NoError(t, q.Publisher().Publish(s.ctx, topic,
+			entityqueue.NewMessage(fmt.Sprintf("gc-%d", i), []byte("x"), partition, nil)))
+	}
+
+	// Fast poll so the 100-tick GC cadence elapses quickly; at the 100ms
+	// default it would take 10s of continuous traffic before reclamation.
+	cfg := testSubConfig("worker-gc-busy", consumerGroup)
+	cfg.PollIntervalMs = 50
+	deliveryChan, err := q.Subscriber().Subscribe(s.ctx, topic, cfg)
+	require.NoError(t, err)
+
+	countMessages := func() int {
+		var n int
+		require.NoError(t, s.db.QueryRowContext(s.ctx,
+			"SELECT COUNT(*) FROM queue_messages WHERE topic = ? AND partition_key = ?",
+			topic, partition).Scan(&n))
+		return n
+	}
+
+	// Drain the acked backlog; the rows survive consumption because only GC
+	// deletes message rows.
+	receiveN(t, deliveryChan, initialBatch, func(d extqueue.Delivery, _ int) {
+		require.NoError(t, d.Ack(s.ctx))
+	})
+	drainSignals(signalCh)
+	require.Equal(t, initialBatch, countMessages())
+
+	// Continuous traffic keeps the partition busy for well over 100 poll ticks;
+	// GC must reclaim the acked backlog without ever observing an idle tick.
+	const continuousTrafficIterations = 150
+	for i := 0; i < continuousTrafficIterations; i++ {
+		require.NoError(t, q.Publisher().Publish(s.ctx, topic,
+			entityqueue.NewMessage(fmt.Sprintf("gc-busy-%d", i), []byte("y"), partition, nil)))
+		delivery := receive(t, deliveryChan)
+		require.NoError(t, delivery.Ack(s.ctx))
+		// Drain signals so the worker's blocking send cannot stall the traffic loop.
+		drainSignals(signalCh)
+	}
+
+	waitForCondition(t, signalCh, func() bool {
+		return countMessages() < initialBatch
+	}, "acked backlog should be garbage collected while the partition stays busy")
+}
+
 // TestNackDoesNotBlockOtherMessages verifies that nacking a message does not
 // block delivery of subsequent messages in the same partition. The nacked
 // message should be skipped (invisible) while later messages are delivered.