From 118059bfb504a31b430a825648aa3dc3f0d087ca Mon Sep 17 00:00:00 2001 From: Jal Bafana Date: Wed, 19 Aug 2026 22:36:37 +0530 Subject: [PATCH] fix(messagequeue): run queue garbage collection on busy partitions --- .../messagequeue/mysql/subscriber.go | 33 +++++----- .../messagequeue/mysql/subscriber_test.go | 58 ++++++++++++++++++ .../messagequeue/mysql/queue_test.go | 61 +++++++++++++++++++ 3 files changed, 134 insertions(+), 18 deletions(-) diff --git a/platform/extension/messagequeue/mysql/subscriber.go b/platform/extension/messagequeue/mysql/subscriber.go index 8140ca57a..eb8738716 100644 --- a/platform/extension/messagequeue/mysql/subscriber.go +++ b/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,17 +1200,14 @@ 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 { + // GC runs every Nth tick regardless of delivery activity; an idle-only + // gate starved continuously busy partitions of garbage collection. + w.gcCounter++ + if w.gcCounter >= gcTickInterval { w.gcCounter = 0 + if err := w.garbageCollect(ctx); err != nil { + return fmt.Errorf("garbage collect: %w", err) + } } // Record poll metrics diff --git a/platform/extension/messagequeue/mysql/subscriber_test.go b/platform/extension/messagequeue/mysql/subscriber_test.go index b7d08e212..0b5595085 100644 --- a/platform/extension/messagequeue/mysql/subscriber_test.go +++ b/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. diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index 7faf659ff..ce996c886 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -2268,6 +2268,67 @@ 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() + + // 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 + } + + // Establish an acked backlog: 200 acked rows that survive consumption + // because only GC deletes message rows. + 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))) + } + receiveN(t, deliveryChan, initialBatch, func(d extqueue.Delivery, _ int) { + require.NoError(t, d.Ack(s.ctx)) + }) + 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)) + } + + 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.